feat: add controller utils package (#3952)

Signed-off-by: Charles-Edouard Brétéché <charled.breteche@gmail.com>

Co-authored-by: Prateek Pandey <prateek.pandey@nirmata.com>
Co-authored-by: Vyankatesh Kudtarkar <vyankateshkd@gmail.com>

pkg/controllers/config/controller.go

@@ -1,17 +1,11 @@
 package config
 
 import (
-	"time"
-
 	"github.com/kyverno/kyverno/pkg/config"
-	kubeutils "github.com/kyverno/kyverno/pkg/utils/kube"
-	corev1 "k8s.io/api/core/v1"
+	controllerutils "github.com/kyverno/kyverno/pkg/utils/controller"
 	"k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
 	corev1informers "k8s.io/client-go/informers/core/v1"
 	corev1listers "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 )
 
@@ -36,84 +30,16 @@ func NewController(configuration config.Configuration, configmapInformer corev1i
 		configmapLister: configmapInformer.Lister(),
 		queue:           workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "config-controller"),
 	}
-	configmapInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    c.add,
-		UpdateFunc: c.update,
-		DeleteFunc: c.delete,
-	})
+	controllerutils.AddDefaultEventHandlers(logger, configmapInformer.Informer(), c.queue)
 	return &c
 }
 
-func (c *controller) add(obj interface{}) {
-	c.enqueue(obj.(*corev1.ConfigMap))
-}
-
-func (c *controller) update(old, cur interface{}) {
-	c.enqueue(cur.(*corev1.ConfigMap))
-}
-
-func (c *controller) delete(obj interface{}) {
-	cm, ok := kubeutils.GetObjectWithTombstone(obj).(*corev1.ConfigMap)
-	if ok {
-		c.enqueue(cm)
-	} else {
-		logger.Info("Failed to get deleted object", "obj", obj)
-	}
-}
-
-func (c *controller) enqueue(obj *corev1.ConfigMap) {
-	if key, err := cache.MetaNamespaceKeyFunc(obj); err != nil {
-		logger.Error(err, "failed to compute key name")
-	} else {
-		c.queue.Add(key)
-	}
-}
-
-func (c *controller) handleErr(err error, key interface{}) {
-	if err == nil {
-		c.queue.Forget(key)
-	} else if errors.IsNotFound(err) {
-		logger.V(4).Info("Dropping request from the queue", "key", key, "error", err.Error())
-		c.queue.Forget(key)
-	} else if c.queue.NumRequeues(key) < maxRetries {
-		logger.V(3).Info("Retrying request", "key", key, "error", err.Error())
-		c.queue.AddRateLimited(key)
-	} else {
-		logger.Error(err, "Failed to process request", "key", key)
-		c.queue.Forget(key)
-	}
-}
-
-func (c *controller) processNextWorkItem() bool {
-	if key, quit := c.queue.Get(); !quit {
-		defer c.queue.Done(key)
-		c.handleErr(c.reconcile(key.(string)), key)
-		return true
-	}
-	return false
-}
-
-func (c *controller) worker() {
-	for c.processNextWorkItem() {
-	}
-}
-
 func (c *controller) Run(stopCh <-chan struct{}) {
-	defer runtime.HandleCrash()
-	logger.Info("starting ...")
-	defer logger.Info("shutting down")
-	for i := 0; i < workers; i++ {
-		go wait.Until(c.worker, time.Second, stopCh)
-	}
-	<-stopCh
+	controllerutils.Run(logger, c.queue, workers, maxRetries, c.reconcile, stopCh)
 }
 
-func (c *controller) reconcile(key string) error {
-	logger.Info("reconciling ...", "key", key)
-	namespace, name, err := cache.SplitMetaNamespaceKey(key)
-	if err != nil {
-		return err
-	}
+func (c *controller) reconcile(key, namespace, name string) error {
+	logger.Info("reconciling ...", "key", key, "namespace", namespace, "name", name)
 	if namespace != config.KyvernoNamespace() || name != config.KyvernoConfigMapName() {
 		return nil
 	}

pkg/controllers/policycache/controller.go

@@ -1,18 +1,14 @@
 package policycache
 
 import (
-	"time"
-
 	kyvernov1 "github.com/kyverno/kyverno/api/kyverno/v1"
 	kyvernov1informers "github.com/kyverno/kyverno/pkg/client/informers/externalversions/kyverno/v1"
 	kyvernov1listers "github.com/kyverno/kyverno/pkg/client/listers/kyverno/v1"
 	pcache "github.com/kyverno/kyverno/pkg/policycache"
-	kubeutils "github.com/kyverno/kyverno/pkg/utils/kube"
+	controllerutils "github.com/kyverno/kyverno/pkg/utils/controller"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 )
@@ -40,68 +36,11 @@ func NewController(pcache pcache.Cache, cpolInformer kyvernov1informers.ClusterP
 		polLister:  polInformer.Lister(),
 		queue:      workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "policycache-controller"),
 	}
-	cpolInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    c.add,
-		UpdateFunc: c.update,
-		DeleteFunc: c.delete,
-	})
-	polInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    c.add,
-		UpdateFunc: c.update,
-		DeleteFunc: c.delete,
-	})
+	controllerutils.AddDefaultEventHandlers(logger, cpolInformer.Informer(), c.queue)
+	controllerutils.AddDefaultEventHandlers(logger, polInformer.Informer(), c.queue)
 	return &c
 }
 
-func (c *controller) add(obj interface{}) {
-	c.enqueue(obj)
-}
-
-func (c *controller) update(_, cur interface{}) {
-	c.enqueue(cur)
-}
-
-func (c *controller) delete(obj interface{}) {
-	c.enqueue(kubeutils.GetObjectWithTombstone(obj))
-}
-
-func (c *controller) enqueue(obj interface{}) {
-	if key, err := cache.MetaNamespaceKeyFunc(obj); err != nil {
-		logger.Error(err, "failed to compute key name")
-	} else {
-		c.queue.Add(key)
-	}
-}
-
-func (c *controller) handleErr(err error, key interface{}) {
-	if err == nil {
-		c.queue.Forget(key)
-	} else if errors.IsNotFound(err) {
-		logger.V(4).Info("Dropping request from the queue", "key", key, "error", err.Error())
-		c.queue.Forget(key)
-	} else if c.queue.NumRequeues(key) < maxRetries {
-		logger.V(3).Info("Retrying request", "key", key, "error", err.Error())
-		c.queue.AddRateLimited(key)
-	} else {
-		logger.Error(err, "Failed to process request", "key", key)
-		c.queue.Forget(key)
-	}
-}
-
-func (c *controller) processNextWorkItem() bool {
-	if key, quit := c.queue.Get(); !quit {
-		defer c.queue.Done(key)
-		c.handleErr(c.reconcile(key.(string)), key)
-		return true
-	}
-	return false
-}
-
-func (c *controller) worker() {
-	for c.processNextWorkItem() {
-	}
-}
-
 func (c *controller) WarmUp() error {
 	logger.Info("warming up ...")
 	defer logger.Info("warm up done")
@@ -131,21 +70,11 @@ func (c *controller) WarmUp() error {
 }
 
 func (c *controller) Run(stopCh <-chan struct{}) {
-	defer runtime.HandleCrash()
-	logger.Info("starting ...")
-	defer logger.Info("shutting down")
-	for i := 0; i < workers; i++ {
-		go wait.Until(c.worker, time.Second, stopCh)
-	}
-	<-stopCh
+	controllerutils.Run(logger, c.queue, workers, maxRetries, c.reconcile, stopCh)
 }
 
-func (c *controller) reconcile(key string) error {
-	logger.Info("reconciling ...", "key", key)
-	namespace, name, err := cache.SplitMetaNamespaceKey(key)
-	if err != nil {
-		return err
-	}
+func (c *controller) reconcile(key, namespace, name string) error {
+	logger.Info("reconciling ...", "key", key, "namespace", namespace, "name", name)
 	policy, err := c.loadPolicy(namespace, name)
 	if err != nil {
 		if errors.IsNotFound(err) {

pkg/utils/controller/handlers.go

@@ -0,0 +1,52 @@
+package controller
+
+import (
+	"github.com/go-logr/logr"
+	kubeutils "github.com/kyverno/kyverno/pkg/utils/kube"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+)
+
+type (
+	addFunc    func(interface{})
+	updateFunc func(interface{}, interface{})
+	deleteFunc func(interface{})
+)
+
+func AddEventHandlers(informer cache.SharedInformer, a addFunc, u updateFunc, d deleteFunc) {
+	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    a,
+		UpdateFunc: u,
+		DeleteFunc: d,
+	})
+}
+
+func AddDefaultEventHandlers(logger logr.Logger, informer cache.SharedInformer, queue workqueue.RateLimitingInterface) {
+	AddEventHandlers(informer, Add(logger, queue), Update(logger, queue), Delete(logger, queue))
+}
+
+func Enqueue(logger logr.Logger, queue workqueue.RateLimitingInterface, obj interface{}) {
+	if key, err := cache.MetaNamespaceKeyFunc(obj); err != nil {
+		logger.Error(err, "failed to compute key name")
+	} else {
+		queue.Add(key)
+	}
+}
+
+func Add(logger logr.Logger, queue workqueue.RateLimitingInterface) addFunc {
+	return func(obj interface{}) {
+		Enqueue(logger, queue, obj)
+	}
+}
+
+func Update(logger logr.Logger, queue workqueue.RateLimitingInterface) updateFunc {
+	return func(_, obj interface{}) {
+		Enqueue(logger, queue, obj)
+	}
+}
+
+func Delete(logger logr.Logger, queue workqueue.RateLimitingInterface) deleteFunc {
+	return func(obj interface{}) {
+		Enqueue(logger, queue, kubeutils.GetObjectWithTombstone(obj))
+	}
+}

pkg/utils/controller/run.go

@@ -0,0 +1,61 @@
+package controller
+
+import (
+	"time"
+
+	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+)
+
+type reconcileFunc func(string, string, string) error
+
+func Run(logger logr.Logger, queue workqueue.RateLimitingInterface, n, maxRetries int, r reconcileFunc, stopCh <-chan struct{}) {
+	defer runtime.HandleCrash()
+	logger.Info("starting ...")
+	defer logger.Info("shutting down")
+	for i := 0; i < n; i++ {
+		go wait.Until(func() { worker(logger, queue, maxRetries, r) }, time.Second, stopCh)
+	}
+	<-stopCh
+}
+
+func worker(logger logr.Logger, queue workqueue.RateLimitingInterface, maxRetries int, r reconcileFunc) {
+	for processNextWorkItem(logger, queue, maxRetries, r) {
+	}
+}
+
+func processNextWorkItem(logger logr.Logger, queue workqueue.RateLimitingInterface, maxRetries int, r reconcileFunc) bool {
+	if key, quit := queue.Get(); !quit {
+		defer queue.Done(key)
+		handleErr(logger, queue, maxRetries, reconcile(key.(string), r), key)
+		return true
+	}
+	return false
+}
+
+func handleErr(logger logr.Logger, queue workqueue.RateLimitingInterface, maxRetries int, err error, key interface{}) {
+	if err == nil {
+		queue.Forget(key)
+	} else if errors.IsNotFound(err) {
+		logger.V(4).Info("Dropping request from the queue", "key", key, "error", err.Error())
+		queue.Forget(key)
+	} else if queue.NumRequeues(key) < maxRetries {
+		logger.V(3).Info("Retrying request", "key", key, "error", err.Error())
+		queue.AddRateLimited(key)
+	} else {
+		logger.Error(err, "Failed to process request", "key", key)
+		queue.Forget(key)
+	}
+}
+
+func reconcile(key string, r reconcileFunc) error {
+	if namespace, name, err := cache.SplitMetaNamespaceKey(key); err != nil {
+		return err
+	} else {
+		return r(key, namespace, name)
+	}
+}