Pass stop channel to informer factory instances

kubernetes-sigs · Jan 22, 2022 · 4f41229 · 4f41229
1 parent 98393df
commit 4f41229
Show file tree

Hide file tree

Showing 22 changed files with 79 additions and 61 deletions.
diff --git a/main.go b/main.go
@@ -136,7 +136,7 @@ func main() {
 	}
 
 	// Lookup all the selected sources by names and pass them the desired configuration.
-	sources, err := source.ByNames(&source.SingletonClientGenerator{
+	sources, err := source.ByNames(ctx, &source.SingletonClientGenerator{
 		KubeConfig:   cfg.KubeConfig,
 		APIServerURL: cfg.APIServerURL,
 		// If update events are enabled, disable timeout.

diff --git a/source/ambassador_host.go b/source/ambassador_host.go
@@ -30,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/informers"
@@ -64,6 +63,7 @@ type ambassadorHostSource struct {
 
 // NewAmbassadorHostSource creates a new ambassadorHostSource with the given config.
 func NewAmbassadorHostSource(
+	ctx context.Context,
 	dynamicKubeClient dynamic.Interface,
 	kubeClient kubernetes.Interface,
 	namespace string) (Source, error) {
@@ -82,8 +82,7 @@ func NewAmbassadorHostSource(
 		},
 	)
 
-	// TODO informer is not explicitly stopped since controller is not passing in its channel.
-	informerFactory.Start(wait.NeverStop)
+	informerFactory.Start(ctx.Done())
 
 	if err := waitForDynamicCacheSync(context.Background(), informerFactory); err != nil {
 		return nil, err

diff --git a/source/contour_httpproxy.go b/source/contour_httpproxy.go
@@ -28,7 +28,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/informers"
@@ -53,6 +52,7 @@ type httpProxySource struct {
 
 // NewContourHTTPProxySource creates a new contourHTTPProxySource with the given config.
 func NewContourHTTPProxySource(
+	ctx context.Context,
 	dynamicKubeClient dynamic.Interface,
 	namespace string,
 	annotationFilter string,
@@ -78,8 +78,7 @@ func NewContourHTTPProxySource(
 		},
 	)
 
-	// TODO informer is not explicitly stopped since controller is not passing in its channel.
-	informerFactory.Start(wait.NeverStop)
+	informerFactory.Start(ctx.Done())
 
 	// wait for the local cache to be populated.
 	if err := waitForDynamicCacheSync(context.Background(), informerFactory); err != nil {

diff --git a/source/contour_httpproxy_test.go b/source/contour_httpproxy_test.go
@@ -88,6 +88,7 @@ func (suite *HTTPProxySuite) SetupTest() {
 	var err error
 
 	suite.source, err = NewContourHTTPProxySource(
+		context.TODO(),
 		fakeDynamicClient,
 		"default",
 		"",
@@ -184,6 +185,7 @@ func TestNewContourHTTPProxySource(t *testing.T) {
 			fakeDynamicClient, _ := newDynamicKubernetesClient()
 
 			_, err := NewContourHTTPProxySource(
+				context.TODO(),
 				fakeDynamicClient,
 				"",
 				ti.annotationFilter,
@@ -1033,6 +1035,7 @@ func testHTTPProxyEndpoints(t *testing.T) {
 			}
 
 			httpProxySource, err := NewContourHTTPProxySource(
+				context.TODO(),
 				fakeDynamicClient,
 				ti.targetNamespace,
 				ti.annotationFilter,
@@ -1059,6 +1062,7 @@ func newTestHTTPProxySource() (*httpProxySource, error) {
 	fakeDynamicClient, _ := newDynamicKubernetesClient()
 
 	src, err := NewContourHTTPProxySource(
+		context.TODO(),
 		fakeDynamicClient,
 		"default",
 		"",

diff --git a/source/ingress.go b/source/ingress.go
@@ -26,7 +26,6 @@ import (
 	log "github.com/sirupsen/logrus"
 	networkv1 "k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/wait"
 	kubeinformers "k8s.io/client-go/informers"
 	netinformers "k8s.io/client-go/informers/networking/v1"
 	"k8s.io/client-go/kubernetes"
@@ -64,7 +63,7 @@ type ingressSource struct {
 }
 
 // NewIngressSource creates a new ingressSource with the given config.
-func NewIngressSource(kubeClient kubernetes.Interface, namespace, annotationFilter string, fqdnTemplate string, combineFqdnAnnotation bool, ignoreHostnameAnnotation bool, ignoreIngressTLSSpec bool, ignoreIngressRulesSpec bool, labelSelector labels.Selector) (Source, error) {
+func NewIngressSource(ctx context.Context, kubeClient kubernetes.Interface, namespace, annotationFilter string, fqdnTemplate string, combineFqdnAnnotation bool, ignoreHostnameAnnotation bool, ignoreIngressTLSSpec bool, ignoreIngressRulesSpec bool, labelSelector labels.Selector) (Source, error) {
 	tmpl, err := parseTemplate(fqdnTemplate)
 	if err != nil {
 		return nil, err
@@ -83,8 +82,7 @@ func NewIngressSource(kubeClient kubernetes.Interface, namespace, annotationFilt
 		},
 	)
 
-	// TODO informer is not explicitly stopped since controller is not passing in its channel.
-	informerFactory.Start(wait.NeverStop)
+	informerFactory.Start(ctx.Done())
 
 	// wait for the local cache to be populated.
 	if err := waitForCacheSync(context.Background(), informerFactory); err != nil {

diff --git a/source/ingress_test.go b/source/ingress_test.go
@@ -56,6 +56,7 @@ func (suite *IngressSuite) SetupTest() {
 	suite.NoError(err, "should succeed")
 
 	suite.sc, err = NewIngressSource(
+		context.TODO(),
 		fakeClient,
 		"",
 		"",
@@ -138,6 +139,7 @@ func TestNewIngressSource(t *testing.T) {
 			t.Parallel()
 
 			_, err := NewIngressSource(
+				context.TODO(),
 				fake.NewSimpleClientset(),
 				"",
 				ti.annotationFilter,
@@ -1225,6 +1227,7 @@ func testIngressEndpoints(t *testing.T) {
 			}
 
 			source, _ := NewIngressSource(
+				context.TODO(),
 				fakeClient,
 				ti.targetNamespace,
 				ti.annotationFilter,

diff --git a/source/istio_gateway.go b/source/istio_gateway.go
@@ -30,7 +30,6 @@ import (
 	networkingv1alpha3informer "istio.io/client-go/pkg/informers/externalversions/networking/v1alpha3"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/wait"
 	kubeinformers "k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -56,6 +55,7 @@ type gatewaySource struct {
 
 // NewIstioGatewaySource creates a new gatewaySource with the given config.
 func NewIstioGatewaySource(
+	ctx context.Context,
 	kubeClient kubernetes.Interface,
 	istioClient istioclient.Interface,
 	namespace string,
@@ -93,9 +93,8 @@ func NewIstioGatewaySource(
 		},
 	)
 
-	// TODO informer is not explicitly stopped since controller is not passing in its channel.
-	informerFactory.Start(wait.NeverStop)
-	istioInformerFactory.Start(wait.NeverStop)
+	informerFactory.Start(ctx.Done())
+	istioInformerFactory.Start(ctx.Done())
 
 	// wait for the local cache to be populated.
 	if err := waitForCacheSync(context.Background(), informerFactory); err != nil {

diff --git a/source/istio_gateway_test.go b/source/istio_gateway_test.go
@@ -69,6 +69,7 @@ func (suite *GatewaySuite) SetupTest() {
 	}
 
 	suite.source, err = NewIstioGatewaySource(
+		context.TODO(),
 		fakeKubernetesClient,
 		fakeIstioClient,
 		"",
@@ -142,6 +143,7 @@ func TestNewIstioGatewaySource(t *testing.T) {
 			t.Parallel()
 
 			_, err := NewIstioGatewaySource(
+				context.TODO(),
 				fake.NewSimpleClientset(),
 				istiofake.NewSimpleClientset(),
 				"",
@@ -1165,6 +1167,7 @@ func testGatewayEndpoints(t *testing.T) {
 			}
 
 			gatewaySource, err := NewIstioGatewaySource(
+				context.TODO(),
 				fakeKubernetesClient,
 				fakeIstioClient,
 				ti.targetNamespace,
@@ -1201,6 +1204,7 @@ func newTestGatewaySource(loadBalancerList []fakeIngressGatewayService) (*gatewa
 	}
 
 	src, err := NewIstioGatewaySource(
+		context.TODO(),
 		fakeKubernetesClient,
 		fakeIstioClient,
 		"",

diff --git a/source/istio_virtualservice.go b/source/istio_virtualservice.go
@@ -31,7 +31,6 @@ import (
 	networkingv1alpha3informer "istio.io/client-go/pkg/informers/externalversions/networking/v1alpha3"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/wait"
 	kubeinformers "k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -60,6 +59,7 @@ type virtualServiceSource struct {
 
 // NewIstioVirtualServiceSource creates a new virtualServiceSource with the given config.
 func NewIstioVirtualServiceSource(
+	ctx context.Context,
 	kubeClient kubernetes.Interface,
 	istioClient istioclient.Interface,
 	namespace string,
@@ -97,9 +97,8 @@ func NewIstioVirtualServiceSource(
 		},
 	)
 
-	// TODO informer is not explicitly stopped since controller is not passing in its channel.
-	informerFactory.Start(wait.NeverStop)
-	istioInformerFactory.Start(wait.NeverStop)
+	informerFactory.Start(ctx.Done())
+	istioInformerFactory.Start(ctx.Done())
 
 	// wait for the local cache to be populated.
 	if err := waitForCacheSync(context.Background(), informerFactory); err != nil {

diff --git a/source/istio_virtualservice_test.go b/source/istio_virtualservice_test.go
@@ -89,6 +89,7 @@ func (suite *VirtualServiceSuite) SetupTest() {
 	suite.NoError(err, "should succeed")
 
 	suite.source, err = NewIstioVirtualServiceSource(
+		context.TODO(),
 		fakeKubernetesClient,
 		fakeIstioClient,
 		"",
@@ -165,6 +166,7 @@ func TestNewIstioVirtualServiceSource(t *testing.T) {
 			t.Parallel()
 
 			_, err := NewIstioVirtualServiceSource(
+				context.TODO(),
 				fake.NewSimpleClientset(),
 				istiofake.NewSimpleClientset(),
 				"",
@@ -1482,6 +1484,7 @@ func testVirtualServiceEndpoints(t *testing.T) {
 			}
 
 			virtualServiceSource, err := NewIstioVirtualServiceSource(
+				context.TODO(),
 				fakeKubernetesClient,
 				fakeIstioClient,
 				ti.targetNamespace,
@@ -1557,6 +1560,7 @@ func newTestVirtualServiceSource(loadBalancerList []fakeIngressGatewayService, g
 	}
 
 	src, err := NewIstioVirtualServiceSource(
+		context.TODO(),
 		fakeKubernetesClient,
 		fakeIstioClient,
 		"",

diff --git a/source/kong_tcpingress.go b/source/kong_tcpingress.go
@@ -29,7 +29,6 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/informers"
@@ -57,7 +56,7 @@ type kongTCPIngressSource struct {
 }
 
 // NewKongTCPIngressSource creates a new kongTCPIngressSource with the given config.
-func NewKongTCPIngressSource(dynamicKubeClient dynamic.Interface, kubeClient kubernetes.Interface, namespace string, annotationFilter string) (Source, error) {
+func NewKongTCPIngressSource(ctx context.Context, dynamicKubeClient dynamic.Interface, kubeClient kubernetes.Interface, namespace string, annotationFilter string) (Source, error) {
 	var err error
 
 	// Use shared informer to listen for add/update/delete of Host in the specified namespace.
@@ -73,8 +72,7 @@ func NewKongTCPIngressSource(dynamicKubeClient dynamic.Interface, kubeClient kub
 		},
 	)
 
-	// TODO informer is not explicitly stopped since controller is not passing in its channel.
-	informerFactory.Start(wait.NeverStop)
+	informerFactory.Start(ctx.Done())
 
 	// wait for the local cache to be populated.
 	if err := waitForDynamicCacheSync(context.Background(), informerFactory); err != nil {

diff --git a/source/kong_tcpingress_test.go b/source/kong_tcpingress_test.go
@@ -241,7 +241,7 @@ func TestKongTCPIngressEndpoints(t *testing.T) {
 			_, err = fakeDynamicClient.Resource(kongGroupdVersionResource).Namespace(defaultKongNamespace).Create(context.Background(), &tcpi, metav1.CreateOptions{})
 			assert.NoError(t, err)
 
-			source, err := NewKongTCPIngressSource(fakeDynamicClient, fakeKubernetesClient, defaultKongNamespace, "kubernetes.io/ingress.class=kong")
+			source, err := NewKongTCPIngressSource(context.TODO(), fakeDynamicClient, fakeKubernetesClient, defaultKongNamespace, "kubernetes.io/ingress.class=kong")
 			assert.NoError(t, err)
 			assert.NotNil(t, source)
 

diff --git a/source/node.go b/source/node.go
@@ -25,7 +25,6 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/wait"
 	kubeinformers "k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -42,7 +41,7 @@ type nodeSource struct {
 }
 
 // NewNodeSource creates a new nodeSource with the given config.
-func NewNodeSource(kubeClient kubernetes.Interface, annotationFilter, fqdnTemplate string) (Source, error) {
+func NewNodeSource(ctx context.Context, kubeClient kubernetes.Interface, annotationFilter, fqdnTemplate string) (Source, error) {
 	tmpl, err := parseTemplate(fqdnTemplate)
 	if err != nil {
 		return nil, err
@@ -62,8 +61,7 @@ func NewNodeSource(kubeClient kubernetes.Interface, annotationFilter, fqdnTempla
 		},
 	)
 
-	// TODO informer is not explicitly stopped since controller is not passing in its channel.
-	informerFactory.Start(wait.NeverStop)
+	informerFactory.Start(ctx.Done())
 
 	// wait for the local cache to be populated.
 	if err := waitForCacheSync(context.Background(), informerFactory); err != nil {

diff --git a/source/node_test.go b/source/node_test.go
@@ -71,6 +71,7 @@ func testNodeSourceNewNodeSource(t *testing.T) {
 			t.Parallel()
 
 			_, err := NewNodeSource(
+				context.TODO(),
 				fake.NewSimpleClientset(),
 				ti.annotationFilter,
 				ti.fqdnTemplate,
@@ -353,6 +354,7 @@ func testNodeSourceEndpoints(t *testing.T) {
 
 			// Create our object under test and get the endpoints.
 			client, err := NewNodeSource(
+				context.TODO(),
 				kubernetes,
 				tc.annotationFilter,
 				tc.fqdnTemplate,

diff --git a/source/openshift_route.go b/source/openshift_route.go
@@ -29,7 +29,6 @@ import (
 	log "github.com/sirupsen/logrus"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
 
 	"sigs.k8s.io/external-dns/endpoint"
@@ -54,6 +53,7 @@ type ocpRouteSource struct {
 
 // NewOcpRouteSource creates a new ocpRouteSource with the given config.
 func NewOcpRouteSource(
+	ctx context.Context,
 	ocpClient versioned.Interface,
 	namespace string,
 	annotationFilter string,
@@ -81,8 +81,7 @@ func NewOcpRouteSource(
 		},
 	)
 
-	// TODO informer is not explicitly stopped since controller is not passing in its channel.
-	informerFactory.Start(wait.NeverStop)
+	informerFactory.Start(ctx.Done())
 
 	// wait for the local cache to be populated.
 	if err := waitForCacheSync(context.Background(), informerFactory); err != nil {

diff --git a/source/openshift_route_test.go b/source/openshift_route_test.go
@@ -43,6 +43,7 @@ func (suite *OCPRouteSuite) SetupTest() {
 	var err error
 
 	suite.sc, err = NewOcpRouteSource(
+		context.TODO(),
 		fakeClient,
 		"",
 		"",
@@ -141,6 +142,7 @@ func testOcpRouteSourceNewOcpRouteSource(t *testing.T) {
 			t.Parallel()
 
 			_, err := NewOcpRouteSource(
+				context.TODO(),
 				fake.NewSimpleClientset(),
 				"",
 				ti.annotationFilter,
@@ -439,6 +441,7 @@ func testOcpRouteSourceEndpoints(t *testing.T) {
 			require.NoError(t, err)
 
 			source, err := NewOcpRouteSource(
+				context.TODO(),
 				fakeClient,
 				"",
 				"",