Refactor watchers - Create separate type for namespaced informers (#3238)

* Create separate type for namespaced watchers

* Create separate type for namespaced watchers in certmanager controller

* Create separate type for namespaced watchers in extdns controller

* Fix case where namespace is not watching secrets

Author: ciarams87

internal/certmanager/cm_controller.go

@@ -56,16 +56,15 @@ const (
 // and creates/ updates certificates for VS resources as required,
 // and VS resources when certificate objects are created/ updated
 type CmController struct {
-	vsLister                  []listers_v1.VirtualServerLister
-	sync                      SyncFn
-	ctx                       context.Context
-	mustSync                  []cache.InformerSynced
-	queue                     workqueue.RateLimitingInterface
-	vsSharedInformerFactory   []vsinformers.SharedInformerFactory
-	cmSharedInformerFactory   []cm_informers.SharedInformerFactory
-	kubeSharedInformerFactory []kubeinformers.SharedInformerFactory
-	recorder                  record.EventRecorder
-	cmClient                  *cm_clientset.Clientset
+	sync          SyncFn
+	ctx           context.Context
+	mustSync      []cache.InformerSynced
+	queue         workqueue.RateLimitingInterface
+	informerGroup map[string]*namespacedInformer
+	recorder      record.EventRecorder
+	cmClient      *cm_clientset.Clientset
+	kubeClient    kubernetes.Interface
+	vsClient      k8s_nginx.Interface
 }
 
 // CmOpts is the options required for building the CmController
@@ -78,27 +77,42 @@ type CmOpts struct {
 	vsClient      k8s_nginx.Interface
 }
 
+type namespacedInformer struct {
+	vsSharedInformerFactory   vsinformers.SharedInformerFactory
+	cmSharedInformerFactory   cm_informers.SharedInformerFactory
+	kubeSharedInformerFactory kubeinformers.SharedInformerFactory
+	vsLister                  listers_v1.VirtualServerLister
+	cmLister                  cmlisters.CertificateLister
+}
+
 func (c *CmController) register() workqueue.RateLimitingInterface {
-	var cmLister []cmlisters.CertificateLister
-	for _, sif := range c.vsSharedInformerFactory {
-		c.vsLister = append(c.vsLister, sif.K8s().V1().VirtualServers().Lister())
-		sif.K8s().V1().VirtualServers().Informer().AddEventHandler(&controllerpkg.QueuingEventHandler{
-			Queue: c.queue,
-		})
-		c.mustSync = append(c.mustSync, sif.K8s().V1().VirtualServers().Informer().HasSynced)
-	}
+	c.sync = SyncFnFor(c.recorder, c.cmClient, c.informerGroup)
+	return c.queue
+}
 
-	for _, cif := range c.cmSharedInformerFactory {
-		cif.Certmanager().V1().Certificates().Informer().AddEventHandler(&controllerpkg.BlockingEventHandler{
-			WorkFunc: certificateHandler(c.queue),
-		})
-		cmLister = append(cmLister, cif.Certmanager().V1().Certificates().Lister())
-		c.mustSync = append(c.mustSync, cif.Certmanager().V1().Certificates().Informer().HasSynced)
-	}
+func (c *CmController) newNamespacedInformer(ns string) {
+	nsi := &namespacedInformer{}
+	nsi.cmSharedInformerFactory = cm_informers.NewSharedInformerFactoryWithOptions(c.cmClient, resyncPeriod, cm_informers.WithNamespace(ns))
+	nsi.kubeSharedInformerFactory = kubeinformers.NewSharedInformerFactoryWithOptions(c.kubeClient, resyncPeriod, kubeinformers.WithNamespace(ns))
+	nsi.vsSharedInformerFactory = vsinformers.NewSharedInformerFactoryWithOptions(c.vsClient, resyncPeriod, vsinformers.WithNamespace(ns))
 
-	c.sync = SyncFnFor(c.recorder, c.cmClient, cmLister)
+	c.addHandlers(nsi)
 
-	return c.queue
+	c.informerGroup[ns] = nsi
+}
+
+func (c *CmController) addHandlers(nsi *namespacedInformer) {
+	nsi.vsLister = nsi.vsSharedInformerFactory.K8s().V1().VirtualServers().Lister()
+	nsi.vsSharedInformerFactory.K8s().V1().VirtualServers().Informer().AddEventHandler(&controllerpkg.QueuingEventHandler{
+		Queue: c.queue,
+	})
+	c.mustSync = append(c.mustSync, nsi.vsSharedInformerFactory.K8s().V1().VirtualServers().Informer().HasSynced)
+
+	nsi.cmSharedInformerFactory.Certmanager().V1().Certificates().Informer().AddEventHandler(&controllerpkg.BlockingEventHandler{
+		WorkFunc: certificateHandler(c.queue),
+	})
+	nsi.cmLister = nsi.cmSharedInformerFactory.Certmanager().V1().Certificates().Lister()
+	c.mustSync = append(c.mustSync, nsi.cmSharedInformerFactory.Certmanager().V1().Certificates().Informer().HasSynced)
 }
 
 func (c *CmController) processItem(ctx context.Context, key string) error {
@@ -108,14 +122,11 @@ func (c *CmController) processItem(ctx context.Context, key string) error {
 		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
 		return err
 	}
+	nsi := getNamespacedInformer(namespace, c.informerGroup)
 
 	var vs *conf_v1.VirtualServer
-	for _, vl := range c.vsLister {
-		vs, err = vl.VirtualServers(namespace).Get(name)
-		if err == nil {
-			break
-		}
-	}
+	vs, err = nsi.vsLister.VirtualServers(namespace).Get(name)
+
 	if err != nil {
 		return err
 	}
@@ -168,25 +179,22 @@ func NewCmController(opts *CmOpts) *CmController {
 	// Create a cert-manager api client
 	intcl, _ := cm_clientset.NewForConfig(opts.kubeConfig)
 
-	var vsSharedInformerFactory []vsinformers.SharedInformerFactory
-	var cmSharedInformerFactory []cm_informers.SharedInformerFactory
-	var kubeSharedInformerFactory []kubeinformers.SharedInformerFactory
+	ig := make(map[string]*namespacedInformer)
 
-	for _, ns := range opts.namespace {
-		cmSharedInformerFactory = append(cmSharedInformerFactory, cm_informers.NewSharedInformerFactoryWithOptions(intcl, resyncPeriod, cm_informers.WithNamespace(ns)))
-		kubeSharedInformerFactory = append(kubeSharedInformerFactory, kubeinformers.NewSharedInformerFactoryWithOptions(opts.kubeClient, resyncPeriod, kubeinformers.WithNamespace(ns)))
-		vsSharedInformerFactory = append(vsSharedInformerFactory, vsinformers.NewSharedInformerFactoryWithOptions(opts.vsClient, resyncPeriod, vsinformers.WithNamespace(ns)))
+	cm := &CmController{
+		ctx:           opts.context,
+		queue:         workqueue.NewNamedRateLimitingQueue(controllerpkg.DefaultItemBasedRateLimiter(), ControllerName),
+		informerGroup: ig,
+		recorder:      opts.eventRecorder,
+		cmClient:      intcl,
+		kubeClient:    opts.kubeClient,
+		vsClient:      opts.vsClient,
 	}
 
-	cm := &CmController{
-		ctx:                       opts.context,
-		queue:                     workqueue.NewNamedRateLimitingQueue(controllerpkg.DefaultItemBasedRateLimiter(), ControllerName),
-		cmSharedInformerFactory:   cmSharedInformerFactory,
-		kubeSharedInformerFactory: kubeSharedInformerFactory,
-		recorder:                  opts.eventRecorder,
-		cmClient:                  intcl,
-		vsSharedInformerFactory:   vsSharedInformerFactory,
+	for _, ns := range opts.namespace {
+		cm.newNamespacedInformer(ns)
 	}
+
 	cm.register()
 	return cm
 }
@@ -201,14 +209,10 @@ func (c *CmController) Run(stopCh <-chan struct{}) {
 
 	glog.Infof("Starting cert-manager control loop")
 
-	for _, vif := range c.vsSharedInformerFactory {
-		go vif.Start(c.ctx.Done())
-	}
-	for _, cif := range c.cmSharedInformerFactory {
-		go cif.Start(c.ctx.Done())
-	}
-	for _, kif := range c.kubeSharedInformerFactory {
-		go kif.Start(c.ctx.Done())
+	for _, ig := range c.informerGroup {
+		go ig.vsSharedInformerFactory.Start(c.ctx.Done())
+		go ig.cmSharedInformerFactory.Start(c.ctx.Done())
+		go ig.kubeSharedInformerFactory.Start(c.ctx.Done())
 	}
 	// // wait for all the informer caches we depend on are synced
 	glog.V(3).Infof("Waiting for %d caches to sync", len(c.mustSync))

internal/certmanager/cm_controller_test.go

@@ -23,19 +23,16 @@ import (
 
 	cmapi "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
 	cmclient "github.com/cert-manager/cert-manager/pkg/client/clientset/versioned"
-	cm_informers "github.com/cert-manager/cert-manager/pkg/client/informers/externalversions"
 	controllerpkg "github.com/cert-manager/cert-manager/pkg/controller"
 	testpkg "github.com/nginxinc/kubernetes-ingress/internal/certmanager/test_files"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/util/workqueue"
 
 	vsapi "github.com/nginxinc/kubernetes-ingress/pkg/apis/configuration/v1"
 	k8s_nginx "github.com/nginxinc/kubernetes-ingress/pkg/client/clientset/versioned"
-	vsinformers "github.com/nginxinc/kubernetes-ingress/pkg/client/informers/externalversions"
 )
 
 func Test_controller_Register(t *testing.T) {
@@ -138,15 +135,27 @@ func Test_controller_Register(t *testing.T) {
 			// Certificate event is received then HasSynced has not been setup
 			// properly.
 
+			ig := make(map[string]*namespacedInformer)
+
+			nsi := &namespacedInformer{
+				cmSharedInformerFactory:   b.Context.SharedInformerFactory,
+				kubeSharedInformerFactory: b.Context.KubeSharedInformerFactory,
+				vsSharedInformerFactory:   b.VsSharedInformerFactory,
+			}
+
+			ig[""] = nsi
+
 			cm := &CmController{
-				ctx:                       b.RootContext,
-				queue:                     workqueue.NewNamedRateLimitingQueue(controllerpkg.DefaultItemBasedRateLimiter(), ControllerName),
-				cmSharedInformerFactory:   []cm_informers.SharedInformerFactory{b.FakeCMInformerFactory()},
-				kubeSharedInformerFactory: []kubeinformers.SharedInformerFactory{b.FakeKubeInformerFactory()},
-				recorder:                  b.Recorder,
-				vsSharedInformerFactory:   []vsinformers.SharedInformerFactory{b.VsSharedInformerFactory},
+				ctx:           b.RootContext,
+				queue:         workqueue.NewNamedRateLimitingQueue(controllerpkg.DefaultItemBasedRateLimiter(), ControllerName),
+				informerGroup: ig,
+				recorder:      b.Recorder,
+				kubeClient:    b.Client,
+				vsClient:      b.VSClient,
 			}
 
+			cm.addHandlers(nsi)
+
 			queue := cm.register()
 
 			b.Start()

internal/certmanager/helper.go

@@ -120,3 +120,21 @@ func translateVsSpec(crt *cmapi.Certificate, vsCmSpec *vsapi.CertManager) error
 	}
 	return nil
 }
+
+func getNamespacedInformer(ns string, ig map[string]*namespacedInformer) *namespacedInformer {
+	var nsi *namespacedInformer
+	var isGlobalNs bool
+	var exists bool
+
+	nsi, isGlobalNs = ig[""]
+
+	if !isGlobalNs {
+		// get the correct namespaced informers
+		nsi, exists = ig[ns]
+		if !exists {
+			// we are not watching this namespace
+			return nil
+		}
+	}
+	return nsi
+}

internal/certmanager/sync.go

@@ -60,7 +60,7 @@ type SyncFn func(context.Context, *vsapi.VirtualServer) error
 func SyncFnFor(
 	rec record.EventRecorder,
 	cmClient clientset.Interface,
-	cmLister []cmlisters.CertificateLister,
+	ig map[string]*namespacedInformer,
 ) SyncFn {
 	return func(ctx context.Context, vs *vsapi.VirtualServer) error {
 		var err error
@@ -75,7 +75,9 @@ func SyncFnFor(
 			return err
 		}
 
-		newCrts, updateCrts, err := buildCertificates(cmLister, vs, issuerName, issuerKind, issuerGroup)
+		nsi := getNamespacedInformer(vs.GetNamespace(), ig)
+
+		newCrts, updateCrts, err := buildCertificates(nsi.cmLister, vs, issuerName, issuerKind, issuerGroup)
 		if err != nil {
 			glog.Errorf("Incorrect cert-manager configuration for VirtualServer resource: %v", err)
 			rec.Eventf(vs, corev1.EventTypeWarning, reasonBadConfig, "Incorrect cert-manager configuration for VirtualServer resource: %s",
@@ -106,12 +108,8 @@ func SyncFnFor(
 		}
 		var certs []*cmapi.Certificate
 
-		for _, cl := range cmLister {
-			certs, err = cl.Certificates(vs.GetNamespace()).List(labels.Everything())
-			if len(certs) > 0 {
-				break
-			}
-		}
+		certs, err = nsi.cmLister.Certificates(vs.GetNamespace()).List(labels.Everything())
+
 		if err != nil {
 			return err
 		}
@@ -131,7 +129,7 @@ func SyncFnFor(
 }
 
 func buildCertificates(
-	cmLister []cmlisters.CertificateLister,
+	cmLister cmlisters.CertificateLister,
 	vs *vsapi.VirtualServer,
 	issuerName, issuerKind, issuerGroup string,
 ) (newCert, update []*cmapi.Certificate, _ error) {
@@ -140,12 +138,8 @@ func buildCertificates(
 	var existingCrt *cmapi.Certificate
 	var err error
 
-	for _, cl := range cmLister {
-		existingCrt, err = cl.Certificates(vs.Namespace).Get(vs.Spec.TLS.Secret)
-		if err == nil {
-			break
-		}
-	}
+	existingCrt, err = cmLister.Certificates(vs.Namespace).Get(vs.Spec.TLS.Secret)
+
 	if !apierrors.IsNotFound(err) && err != nil {
 		return nil, nil, err
 	}

internal/certmanager/sync_test.go

@@ -23,7 +23,6 @@ import (
 
 	cmapi "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
 	cmmeta "github.com/cert-manager/cert-manager/pkg/apis/meta/v1"
-	cmlisters "github.com/cert-manager/cert-manager/pkg/client/listers/certmanager/v1"
 	"github.com/cert-manager/cert-manager/test/unit/gen"
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -491,7 +490,19 @@ func TestSync(t *testing.T) {
 			}
 			b.Init()
 			defer b.Stop()
-			sync := SyncFnFor(b.Recorder, b.CMClient, []cmlisters.CertificateLister{b.SharedInformerFactory.Certmanager().V1().Certificates().Lister()})
+
+			ig := make(map[string]*namespacedInformer)
+
+			nsi := &namespacedInformer{
+				cmSharedInformerFactory:   b.FakeCMInformerFactory(),
+				kubeSharedInformerFactory: b.FakeKubeInformerFactory(),
+				vsSharedInformerFactory:   b.VsSharedInformerFactory,
+				cmLister:                  b.SharedInformerFactory.Certmanager().V1().Certificates().Lister(),
+			}
+
+			ig[""] = nsi
+
+			sync := SyncFnFor(b.Recorder, b.CMClient, ig)
 			b.Start()
 
 			err := sync(context.Background(), &test.VirtualServer)