🌱 Propagate context from controller.Start to handler

Signed-off-by: Vince Prignano <[email protected]>

Author: vincepri

pkg/internal/controller/controller.go

@@ -130,6 +130,10 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 	c.Queue = c.MakeQueue()
 	defer c.Queue.ShutDown() // needs to be outside the iife so that we shutdown after the stop channel is closed
 
+	// TODO: Propagate context from the Runnable interface, when we're ready to change the signature.
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
 	err := func() error {
 		defer c.mu.Unlock()
 
@@ -170,8 +174,12 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 		// Launch workers to process resources
 		c.Log.Info("Starting workers", "worker count", c.MaxConcurrentReconciles)
 		for i := 0; i < c.MaxConcurrentReconciles; i++ {
-			// Process work items
-			go wait.Until(c.worker, c.JitterPeriod, stop)
+			// Run a worker thread that just dequeues items, processes them, and marks them done.
+			// It enforces that the reconcileHandler is never invoked concurrently with the same object.
+			go wait.Until(func() {
+				for c.processNextWorkItem(ctx) {
+				}
+			}, c.JitterPeriod, stop)
 		}
 
 		c.Started = true
@@ -180,22 +188,16 @@ func (c *Controller) Start(stop <-chan struct{}) error {
 	if err != nil {
 		return err
 	}
+	context.WithCancel(context.Background())
 
 	<-stop
 	c.Log.Info("Stopping workers")
 	return nil
 }
 
-// worker runs a worker thread that just dequeues items, processes them, and marks them done.
-// It enforces that the reconcileHandler is never invoked concurrently with the same object.
-func (c *Controller) worker() {
-	for c.processNextWorkItem() {
-	}
-}
-
 // processNextWorkItem will read a single work item off the workqueue and
 // attempt to process it, by calling the reconcileHandler.
-func (c *Controller) processNextWorkItem() bool {
+func (c *Controller) processNextWorkItem(ctx context.Context) bool {
 	obj, shutdown := c.Queue.Get()
 	if shutdown {
 		// Stop working
@@ -210,10 +212,10 @@ func (c *Controller) processNextWorkItem() bool {
 	// period.
 	defer c.Queue.Done(obj)
 
-	return c.reconcileHandler(obj)
+	return c.reconcileHandler(ctx, obj)
 }
 
-func (c *Controller) reconcileHandler(obj interface{}) bool {
+func (c *Controller) reconcileHandler(ctx context.Context, obj interface{}) bool {
 	// Update metrics after processing each item
 	reconcileStartTS := time.Now()
 	defer func() {
@@ -233,7 +235,7 @@ func (c *Controller) reconcileHandler(obj interface{}) bool {
 	}
 
 	log := c.Log.WithValues("name", req.Name, "namespace", req.Namespace)
-	ctx := logf.IntoContext(context.Background(), log)
+	ctx = logf.IntoContext(ctx, log)
 
 	// RunInformersAndControllers the syncHandler, passing it the namespace/Name string of the
 	// resource to be synced.