From 39d798678e07ab6b75f07b1f182bd31047b5e1f3 Mon Sep 17 00:00:00 2001 From: grokspawn Date: Thu, 15 Oct 2026 14:22:44 -0500 Subject: [PATCH] add csv-based healthcheck retry backoff Signed-off-by: grokspawn --- .../operators/olm/health_check_backoff.go | 48 ++++++ .../olm/health_check_backoff_test.go | 140 ++++++++++++++++++ pkg/controller/operators/olm/operator.go | 27 +++- 3 files changed, 207 insertions(+), 8 deletions(-) create mode 100644 pkg/controller/operators/olm/health_check_backoff.go create mode 100644 pkg/controller/operators/olm/health_check_backoff_test.go diff --git a/pkg/controller/operators/olm/health_check_backoff.go b/pkg/controller/operators/olm/health_check_backoff.go new file mode 100644 index 0000000000..b7fbe58a42 --- /dev/null +++ b/pkg/controller/operators/olm/health_check_backoff.go @@ -0,0 +1,48 @@ +package olm + +import ( + "time" + + "github.com/operator-framework/api/pkg/operators/v1alpha1" + "github.com/operator-framework/operator-lifecycle-manager/pkg/controller/install" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/workqueue" +) + +const ( + healthCheckBackoffBaseDelay = 10 * time.Second + healthCheckBackoffMaxDelay = 5 * time.Minute +) + +// csvHealthCheckBackoff tracks repeated transient Deployment readiness failures +// independently for each CSV. +type csvHealthCheckBackoff struct { + rateLimiter workqueue.TypedRateLimiter[types.NamespacedName] +} + +func newCSVHealthCheckBackoff() *csvHealthCheckBackoff { + return &csvHealthCheckBackoff{ + rateLimiter: workqueue.NewTypedItemExponentialFailureRateLimiter[types.NamespacedName]( + healthCheckBackoffBaseDelay, + healthCheckBackoffMaxDelay, + ), + } +} + +func (b *csvHealthCheckBackoff) next(key types.NamespacedName) time.Duration { + return b.rateLimiter.When(key) +} + +func (b *csvHealthCheckBackoff) forget(key types.NamespacedName) { + b.rateLimiter.Forget(key) +} + +func (a *Operator) requeueAfterStrategyError(csv *v1alpha1.ClusterServiceVersion, strategyErr error) error { + key := types.NamespacedName{Namespace: csv.GetNamespace(), Name: csv.GetName()} + if install.ReasonForError(strategyErr) == install.StrategyErrReasonWaiting { + return a.csvQueueSet.RequeueAfter(csv.GetNamespace(), csv.GetName(), a.healthCheckBackoff.next(key)) + } + + a.healthCheckBackoff.forget(key) + return a.csvQueueSet.Requeue(csv.GetNamespace(), csv.GetName()) +} diff --git a/pkg/controller/operators/olm/health_check_backoff_test.go b/pkg/controller/operators/olm/health_check_backoff_test.go new file mode 100644 index 0000000000..f293d84ae6 --- /dev/null +++ b/pkg/controller/operators/olm/health_check_backoff_test.go @@ -0,0 +1,140 @@ +package olm + +import ( + "context" + "testing" + "time" + + "github.com/operator-framework/api/pkg/operators/v1alpha1" + "github.com/operator-framework/operator-lifecycle-manager/pkg/controller/install" + "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/queueinformer" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/workqueue" +) + +type delayedQueueItem struct { + key types.NamespacedName + delay time.Duration +} + +type recordingCSVQueue struct { + workqueue.TypedRateLimitingInterface[types.NamespacedName] + added []types.NamespacedName + delayed []delayedQueueItem +} + +func (q *recordingCSVQueue) Add(key types.NamespacedName) { + q.added = append(q.added, key) +} + +func (q *recordingCSVQueue) AddAfter(key types.NamespacedName, delay time.Duration) { + q.delayed = append(q.delayed, delayedQueueItem{key: key, delay: delay}) +} + +func TestCSVHealthCheckBackoffIsExponentialAndCapped(t *testing.T) { + backoff := newCSVHealthCheckBackoff() + key := types.NamespacedName{Namespace: "operators", Name: "sample"} + otherKey := types.NamespacedName{Namespace: "operators", Name: "other"} + want := []time.Duration{ + 10 * time.Second, + 20 * time.Second, + 40 * time.Second, + 80 * time.Second, + 160 * time.Second, + 5 * time.Minute, + 5 * time.Minute, + } + + for _, expected := range want { + require.Equal(t, expected, backoff.next(key)) + } + require.Equal(t, 10*time.Second, backoff.next(otherKey), "backoff should be isolated per CSV") + + backoff.forget(key) + require.Equal(t, 10*time.Second, backoff.next(key), "forget should reset the retry sequence") +} + +func TestRequeueAfterStrategyErrorDelaysOnlyDeploymentWaits(t *testing.T) { + queue := &recordingCSVQueue{} + queueSet := queueinformer.NewEmptyResourceQueueSet() + queueSet.Set("operators", queue) + op := &Operator{ + csvQueueSet: queueSet, + healthCheckBackoff: newCSVHealthCheckBackoff(), + } + csv := &v1alpha1.ClusterServiceVersion{ObjectMeta: metav1.ObjectMeta{ + Namespace: "operators", + Name: "sample", + }} + waiting := install.StrategyError{Reason: install.StrategyErrReasonWaiting, Message: "deployment is not ready"} + + require.NoError(t, op.requeueAfterStrategyError(csv, waiting)) + require.Empty(t, queue.added, "health waits must not use immediate queue.Add") + require.Equal(t, []delayedQueueItem{{ + key: types.NamespacedName{Namespace: "operators", Name: "sample"}, + delay: 10 * time.Second, + }}, queue.delayed) + + require.NoError(t, op.requeueAfterStrategyError(csv, waiting)) + require.Equal(t, 20*time.Second, queue.delayed[1].delay) + + other := install.StrategyError{Reason: install.StrategyErrDeploymentUpdated, Message: "deployment spec changed"} + require.NoError(t, op.requeueAfterStrategyError(csv, other)) + require.Equal(t, []types.NamespacedName{{Namespace: "operators", Name: "sample"}}, queue.added) + + require.NoError(t, op.requeueAfterStrategyError(csv, waiting)) + require.Equal(t, 10*time.Second, queue.delayed[2].delay, "non-waiting errors should reset health backoff") +} + +func TestUpdateInstallStatusResetsHealthBackoffWhenDeploymentIsHealthy(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + op, err := NewFakeOperator(ctx, withNamespaces("health-backoff-test")) + require.NoError(t, err) + op.recorder = &record.FakeRecorder{Events: make(chan string, 10)} + + csv := &v1alpha1.ClusterServiceVersion{ + ObjectMeta: metav1.ObjectMeta{Name: "sample", Namespace: "health-backoff-test"}, + Status: v1alpha1.ClusterServiceVersionStatus{ + Phase: v1alpha1.CSVPhaseFailed, + Reason: v1alpha1.CSVReasonComponentUnhealthy, + Message: "deployment is not ready", + }, + } + key := types.NamespacedName{Namespace: csv.GetNamespace(), Name: csv.GetName()} + require.Equal(t, 10*time.Second, op.healthCheckBackoff.next(key)) + require.Equal(t, 20*time.Second, op.healthCheckBackoff.next(key)) + + err = op.updateInstallStatus( + csv, + NewTestInstaller(nil, nil), + &v1alpha1.StrategyDetailsDeployment{}, + v1alpha1.CSVPhaseFailed, + v1alpha1.CSVReasonComponentUnhealthy, + ) + require.NoError(t, err) + require.Equal(t, 10*time.Second, op.healthCheckBackoff.next(key), "healthy checks should reset the retry sequence") +} + +func TestCSVDeletionForgetsHealthBackoff(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + op, err := NewFakeOperator(ctx, withNamespaces("health-backoff-test")) + require.NoError(t, err) + csv := &v1alpha1.ClusterServiceVersion{ObjectMeta: metav1.ObjectMeta{ + Name: "sample", + Namespace: "health-backoff-test", + }} + key := types.NamespacedName{Namespace: csv.GetNamespace(), Name: csv.GetName()} + op.healthCheckBackoff.next(key) + require.Equal(t, 20*time.Second, op.healthCheckBackoff.next(key)) + + op.handleClusterServiceVersionDeletion(csv) + + require.Equal(t, 10*time.Second, op.healthCheckBackoff.next(key), "deleted CSVs should not retain retry state") +} diff --git a/pkg/controller/operators/olm/operator.go b/pkg/controller/operators/olm/operator.go index 7507b97aaf..141b3feaa2 100644 --- a/pkg/controller/operators/olm/operator.go +++ b/pkg/controller/operators/olm/operator.go @@ -101,6 +101,7 @@ type Operator struct { copiedCSVLister operatorsv1alpha1listers.ClusterServiceVersionLister ogQueueSet *queueinformer.ResourceQueueSet csvQueueSet *queueinformer.ResourceQueueSet + healthCheckBackoff *csvHealthCheckBackoff olmConfigQueue workqueue.TypedRateLimitingInterface[types.NamespacedName] csvCopyQueueSet *queueinformer.ResourceQueueSet copiedCSVQueueSet *queueinformer.ResourceQueueSet @@ -209,13 +210,14 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat } op := &Operator{ - Operator: queueOperator, - clock: config.clock, - logger: config.logger, - opClient: config.operatorClient, - client: config.externalClient, - ogQueueSet: queueinformer.NewEmptyResourceQueueSet(), - csvQueueSet: queueinformer.NewEmptyResourceQueueSet(), + Operator: queueOperator, + clock: config.clock, + logger: config.logger, + opClient: config.operatorClient, + client: config.externalClient, + ogQueueSet: queueinformer.NewEmptyResourceQueueSet(), + csvQueueSet: queueinformer.NewEmptyResourceQueueSet(), + healthCheckBackoff: newCSVHealthCheckBackoff(), olmConfigQueue: workqueue.NewTypedRateLimitingQueueWithConfig[types.NamespacedName]( workqueue.DefaultTypedControllerRateLimiter[types.NamespacedName](), workqueue.TypedRateLimitingQueueConfig[types.NamespacedName]{ @@ -1188,6 +1190,11 @@ func (a *Operator) handleClusterServiceVersionDeletion(obj interface{}) { } } + a.healthCheckBackoff.forget(types.NamespacedName{ + Namespace: clusterServiceVersion.GetNamespace(), + Name: clusterServiceVersion.GetName(), + }) + if a.csvNotification != nil { a.csvNotification.OnDelete(clusterServiceVersion) } @@ -2611,6 +2618,10 @@ func (a *Operator) checkReplacementsAndUpdateStatus(csv *v1alpha1.ClusterService func (a *Operator) updateInstallStatus(csv *v1alpha1.ClusterServiceVersion, installer install.StrategyInstaller, strategy install.Strategy, requeuePhase v1alpha1.ClusterServiceVersionPhase, requeueConditionReason v1alpha1.ConditionReason) error { strategyInstalled, strategyErr := installer.CheckInstalled(strategy) now := a.now() + key := types.NamespacedName{Namespace: csv.GetNamespace(), Name: csv.GetName()} + if strategyErr == nil { + a.healthCheckBackoff.forget(key) + } if strategyErr != nil { a.logger.WithError(strategyErr).Debug("operator not installed") @@ -2686,7 +2697,7 @@ func (a *Operator) updateInstallStatus(csv *v1alpha1.ClusterServiceVersion, inst } else { csv.SetPhaseWithEventIfChanged(requeuePhase, requeueConditionReason, fmt.Sprintf("installing: %s", strategyErr), now, a.recorder) } - if err := a.csvQueueSet.Requeue(csv.GetNamespace(), csv.GetName()); err != nil { + if err := a.requeueAfterStrategyError(csv, strategyErr); err != nil { a.logger.Warn(err.Error()) }