kube-aggregator: Refactor OpenAPI controllers to be context-aware

staging/src/k8s.io/kube-aggregator/pkg/controllers/openapi/controller.go

@@ -17,6 +17,7 @@ limitations under the License.
 package openapi
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"time"
@@ -70,32 +71,42 @@ func NewAggregationController(downloader *aggregator.Downloader, openAPIAggregat
 	return c
 }
 
-// Run starts OpenAPI AggregationController
+// Run is a legacy wrapper that starts the controller.
+//
+//logcheck:context // RunWithContext should be used instead of Run in code which supports contextual logging.
 func (c *AggregationController) Run(stopCh <-chan struct{}) {
-	defer utilruntime.HandleCrash()
-	defer c.queue.ShutDown()
-
-	klog.Info("Starting OpenAPI AggregationController")
-	defer klog.Info("Shutting down OpenAPI AggregationController")
-
-	go wait.Until(c.runWorker, time.Second, stopCh)
-
-	<-stopCh
+	c.RunWithContext(wait.ContextForChannel(stopCh))
 }
 
-func (c *AggregationController) runWorker() {
-	for c.processNextWorkItem() {
+// RunWithContext starts OpenAPI AggregationController and blocks until the context is cancelled.
+func (c *AggregationController) RunWithContext(ctx context.Context) {
+	defer utilruntime.HandleCrashWithContext(ctx)
+	defer c.queue.ShutDown()
+
+	logger := klog.FromContext(ctx)
+	logger.Info("Starting OpenAPI AggregationController")
+	defer logger.Info("Shutting down OpenAPI AggregationController")
+
+	go wait.UntilWithContext(ctx, c.runWorker, time.Second)
+
+	<-ctx.Done()
+}
+
+func (c *AggregationController) runWorker(ctx context.Context) {
+	logger := klog.FromContext(ctx)
+	for c.processNextWorkItem(logger) {
 	}
 }
 
 // processNextWorkItem deals with one key off the queue.  It returns false when it's time to quit.
-func (c *AggregationController) processNextWorkItem() bool {
+func (c *AggregationController) processNextWorkItem(logger klog.Logger) bool {
 	key, quit := c.queue.Get()
 	defer c.queue.Done(key)
 	if quit {
 		return false
 	}
-	klog.V(4).Infof("OpenAPI AggregationController: Processing item %s", key)
+
+	logger.V(4).Info("OpenAPI AggregationController: Processing item", "key", key)
 
 	action, err := c.syncHandler(key)
 	if err != nil {
@@ -106,7 +117,7 @@ func (c *AggregationController) processNextWorkItem() bool {
 	case syncRequeue:
 		c.queue.AddAfter(key, successfulUpdateDelay)
 	case syncRequeueRateLimited:
-		klog.Infof("OpenAPI AggregationController: action for item %s: Rate Limited Requeue.", key)
+		logger.Info("OpenAPI AggregationController: action for item: Rate Limited Requeue", "key", key)
 		c.queue.AddRateLimited(key)
 	case syncNothing:
 		c.queue.Forget(key)

staging/src/k8s.io/kube-aggregator/pkg/controllers/openapiv3/controller.go

@@ -17,6 +17,7 @@ limitations under the License.
 package openapiv3
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"time"
@@ -72,26 +73,35 @@ func NewAggregationController(openAPIAggregationManager aggregator.SpecProxier)
 	return c
 }
 
-// Run starts OpenAPI AggregationController
+// Run is a legacy wrapper that starts the controller.
+//
+//logcheck:context // RunWithContext should be used instead of Run in code which supports contextual logging.
 func (c *AggregationController) Run(stopCh <-chan struct{}) {
-	defer utilruntime.HandleCrash()
-	defer c.queue.ShutDown()
-
-	klog.Info("Starting OpenAPI V3 AggregationController")
-	defer klog.Info("Shutting down OpenAPI V3 AggregationController")
-
-	go wait.Until(c.runWorker, time.Second, stopCh)
-
-	<-stopCh
+	c.RunWithContext(wait.ContextForChannel(stopCh))
 }
 
-func (c *AggregationController) runWorker() {
-	for c.processNextWorkItem() {
+// RunWithContext starts OpenAPI V3 AggregationController and blocks until the context is cancelled.
+func (c *AggregationController) RunWithContext(ctx context.Context) {
+	defer utilruntime.HandleCrashWithContext(ctx)
+	defer c.queue.ShutDown()
+
+	logger := klog.FromContext(ctx)
+	logger.Info("Starting OpenAPI V3 AggregationController")
+	defer logger.Info("Shutting down OpenAPI V3 AggregationController")
+
+	go wait.UntilWithContext(ctx, c.runWorker, time.Second)
+
+	<-ctx.Done()
+}
+
+func (c *AggregationController) runWorker(ctx context.Context) {
+	logger := klog.FromContext(ctx)
+	for c.processNextWorkItem(logger) {
 	}
 }
 
 // processNextWorkItem deals with one key off the queue.  It returns false when it's time to quit.
-func (c *AggregationController) processNextWorkItem() bool {
+func (c *AggregationController) processNextWorkItem(logger klog.Logger) bool {
 	key, quit := c.queue.Get()
 	defer c.queue.Done(key)
 	if quit {
@@ -101,9 +111,9 @@ func (c *AggregationController) processNextWorkItem() bool {
 	if aggregator.IsLocalAPIService(key) {
 		// for local delegation targets that are aggregated once per second, log at
 		// higher level to avoid flooding the log
-		klog.V(6).Infof("OpenAPI AggregationController: Processing item %s", key)
+		logger.V(6).Info("OpenAPI AggregationController: Processing item", "key", key)
 	} else {
-		klog.V(4).Infof("OpenAPI AggregationController: Processing item %s", key)
+		logger.V(4).Info("OpenAPI AggregationController: Processing item", "key", key)
 	}
 
 	action, err := c.syncHandler(key)
@@ -116,17 +126,17 @@ func (c *AggregationController) processNextWorkItem() bool {
 	switch action {
 	case syncRequeue:
 		if aggregator.IsLocalAPIService(key) {
-			klog.V(7).Infof("OpenAPI AggregationController: action for local item %s: Requeue after %s.", key, successfulUpdateDelayLocal)
+			logger.V(7).Info("OpenAPI AggregationController: action for local item: Requeue", "key", key, "duration", successfulUpdateDelayLocal)
 			c.queue.AddAfter(key, successfulUpdateDelayLocal)
 		} else {
-			klog.V(7).Infof("OpenAPI AggregationController: action for item %s: Requeue.", key)
+			logger.V(7).Info("OpenAPI AggregationController: action for item: Requeue", "key", key)
 			c.queue.AddAfter(key, successfulUpdateDelay)
 		}
 	case syncRequeueRateLimited:
-		klog.Infof("OpenAPI AggregationController: action for item %s: Rate Limited Requeue.", key)
+		logger.Info("OpenAPI AggregationController: action for item: Rate Limited Requeue", "key", key)
 		c.queue.AddRateLimited(key)
 	case syncNothing:
-		klog.Infof("OpenAPI AggregationController: action for item %s: Nothing (removed from the queue).", key)
+		logger.Info("OpenAPI AggregationController: action for item: Nothing (removed from the queue)", "key", key)
 	}
 
 	return true