From 6d23bc4b65b9830d67b2d32185e262b41b4fab33 Mon Sep 17 00:00:00 2001 From: Aditi Gupta Date: Wed, 22 Oct 2025 12:08:40 -0700 Subject: [PATCH] kube-aggregator: Refactor OpenAPI controllers to be context-aware --- .../pkg/controllers/openapi/controller.go | 41 ++++++++++------ .../pkg/controllers/openapiv3/controller.go | 48 +++++++++++-------- 2 files changed, 55 insertions(+), 34 deletions(-) diff --git a/staging/src/k8s.io/kube-aggregator/pkg/controllers/openapi/controller.go b/staging/src/k8s.io/kube-aggregator/pkg/controllers/openapi/controller.go index 3643df8003e..6665ead8806 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/controllers/openapi/controller.go +++ b/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) diff --git a/staging/src/k8s.io/kube-aggregator/pkg/controllers/openapiv3/controller.go b/staging/src/k8s.io/kube-aggregator/pkg/controllers/openapiv3/controller.go index d91222824c1..68673b1b80f 100644 --- a/staging/src/k8s.io/kube-aggregator/pkg/controllers/openapiv3/controller.go +++ b/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