Skip to content

NO-JIRA: certregenerationcontroller: Improve goroutine mgmt #1981

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tchap wants to merge 2 commits into
base: main
Choose a base branch
from
Open

NO-JIRA: certregenerationcontroller: Improve goroutine mgmt #1981

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b79ae03
certregenerationcontroller: Improve goroutine mgmt
tchap Apr 22, 2026
e6bd2b1
certregenerationcontroller: Use factory.New
tchap Apr 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 14 additions & 81 deletions pkg/cmd/certregenerationcontroller/cabundlesyncer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,127 +2,60 @@ package certregenerationcontroller

import (
"context"
"fmt"
"time"

utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
corev1listers "k8s.io/client-go/listers/core/v1"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/util/workqueue"
"k8s.io/klog/v2"

"github.com/openshift/library-go/pkg/controller/factory"
"github.com/openshift/library-go/pkg/operator/events"
"github.com/openshift/library-go/pkg/operator/v1helpers"

"github.com/openshift/cluster-kube-apiserver-operator/pkg/operator/operatorclient"
"github.com/openshift/cluster-kube-apiserver-operator/pkg/operator/targetconfigcontroller"
)

const workQueueKey = "key"

// CABundleController composes individual certs into CA bundle that is used
// caBundleController composes individual certs into CA bundle that is used
// by kube-apiserver to validate clients.
// Cert recovery refreshes "kube-control-plane-signer-ca" and needs the containing
// bundle regenerated so kube-controller-manager and kube-scheduler can connect
// using client certs.
type CABundleController struct {
type caBundleController struct {
configMapGetter corev1client.ConfigMapsGetter
configMapLister corev1listers.ConfigMapLister

eventRecorder events.Recorder

cachesToSync []cache.InformerSynced

// queue only ever has one item, but it has nice error handling backoff/retry semantics
queue workqueue.RateLimitingInterface
eventRecorder events.Recorder
}

func NewCABundleController(
configMapGetter corev1client.ConfigMapsGetter,
kubeInformersForNamespaces v1helpers.KubeInformersForNamespaces,
eventRecorder events.Recorder,
) (*CABundleController, error) {
c := &CABundleController{
) factory.Controller {
c := &caBundleController{
configMapGetter: configMapGetter,
configMapLister: kubeInformersForNamespaces.ConfigMapLister(),
eventRecorder: eventRecorder.WithComponentSuffix("manage-client-ca-bundle-recovery-controller"),
queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "CABundleRecoveryController"),
}

handler := cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) { c.queue.Add(workQueueKey) },
UpdateFunc: func(old, new interface{}) { c.queue.Add(workQueueKey) },
DeleteFunc: func(obj interface{}) { c.queue.Add(workQueueKey) },
}

// we react to some config changes
namespaces := []string{
operatorclient.GlobalUserSpecifiedConfigNamespace,
operatorclient.GlobalMachineSpecifiedConfigNamespace,
operatorclient.OperatorNamespace,
operatorclient.TargetNamespace,
}
for _, namespace := range namespaces {
informers := kubeInformersForNamespaces.InformersFor(namespace)
informers.Core().V1().ConfigMaps().Informer().AddEventHandler(handler)
c.cachesToSync = append(c.cachesToSync, informers.Core().V1().ConfigMaps().Informer().HasSynced)
}

return c, nil
}

func (c *CABundleController) Run(ctx context.Context) {
defer utilruntime.HandleCrash()

// FIXME: These are missing a wait group to track goroutines and handle graceful termination
Copy link
Copy Markdown
Contributor

@p0lyn0mial p0lyn0mial Jan 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what does graceful termination mean in this context ?

Copy link
Copy Markdown
Contributor Author

@tchap tchap Jan 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pretty much nothing, I don't think the command is handling signals really.

Always nice to handle signals, but not really important here, I guess. It's a bit orthogonal to handling goroutines. It's always good to have all goroutines terminated when Run returns.

// (@deads2k wants time to think it through)

klog.Info("Starting CA bundle controller")
defer func() {
klog.Info("Shutting down CA bundle controller")
c.queue.ShutDown()
klog.Info("CA bundle controller shut down")
}()

if !cache.WaitForNamedCacheSync("CABundleController", ctx.Done(), c.cachesToSync...) {
return
var informers []factory.Informer
for _, ns := range namespaces {
informers = append(informers, kubeInformersForNamespaces.InformersFor(ns).Core().V1().ConfigMaps().Informer())
}

go func() {
wait.UntilWithContext(ctx, c.runWorker, time.Second)
}()

<-ctx.Done()
}

func (c *CABundleController) runWorker(ctx context.Context) {
for c.processNextItem(ctx) {
}
}

func (c *CABundleController) processNextItem(ctx context.Context) bool {
key, quit := c.queue.Get()
if quit {
return false
}
defer c.queue.Done(key)

err := c.sync(ctx)

if err == nil {
c.queue.Forget(key)
return true
}

utilruntime.HandleError(fmt.Errorf("%v failed with : %w", key, err))
c.queue.AddRateLimited(key)

return true
return factory.New().
WithInformers(informers...).
WithSync(c.sync).
ToController("CABundleRecoveryController", c.eventRecorder)
}

func (c *CABundleController) sync(ctx context.Context) error {
func (c *caBundleController) sync(ctx context.Context, _ factory.SyncContext) error {
// Always start 10 seconds later after a change occurred. Makes us less likely to steal work and logs from the operator.
timer := time.NewTimer(10 * time.Second)
defer timer.Stop()
Expand Down
35 changes: 20 additions & 15 deletions pkg/cmd/certregenerationcontroller/cmd.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package certregenerationcontroller
import (
"context"
"fmt"
"sync"
"time"

"github.com/spf13/cobra"
Expand Down Expand Up @@ -117,8 +118,17 @@ func (o *Options) Run(ctx context.Context, clock clock.Clock) error {
o.controllerContext.EventRecorder,
)

go configInformers.Start(ctx.Done())
go featureGateAccessor.Run(ctx)
var wg sync.WaitGroup
defer wg.Wait()
// cancel must happen before wg.Wait (so in a later defer), otherwise we can get stuck on early return.
ctx, cancel := context.WithCancel(ctx)
defer cancel()

configInformers.Start(ctx.Done())

wg.Go(func() {
featureGateAccessor.Run(ctx)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
})

var featureGates featuregates.FeatureGate
select {
Expand All @@ -128,6 +138,8 @@ func (o *Options) Run(ctx context.Context, clock clock.Clock) error {
case <-time.After(1 * time.Minute):
klog.Errorf("timed out waiting for FeatureGate detection")
return fmt.Errorf("timed out waiting for FeatureGate detection")
case <-ctx.Done():
return ctx.Err()
}

kubeAPIServerCertRotationController, err := certrotationcontroller.NewCertRotationControllerOnlyWhenExpired(
Expand All @@ -142,30 +154,23 @@ func (o *Options) Run(ctx context.Context, clock clock.Clock) error {
return err
}

caBundleController, err := NewCABundleController(
caBundleController := NewCABundleController(
kubeClient.CoreV1(),
kubeAPIServerInformersForNamespaces,
o.controllerContext.EventRecorder,
)
if err != nil {
return err
}

// We can't start informers until after the resources have been requested. Now is the time.
kubeAPIServerInformersForNamespaces.Start(ctx.Done())
dynamicInformers.Start(ctx.Done())
configInformers.Start(ctx.Done())

// FIXME: These are missing a wait group to track goroutines and handle graceful termination
// (@deads2k wants time to think it through)

go func() {
wg.Go(func() {
kubeAPIServerCertRotationController.Run(ctx, 1)
}()

go func() {
caBundleController.Run(ctx)
}()
})
wg.Go(func() {
caBundleController.Run(ctx, 1)
})

<-ctx.Done()

Expand Down