From 6325e7b5655b811eadf9cb7cea9f433f65f8530c Mon Sep 17 00:00:00 2001 From: Sanjula Herath Date: Wed, 5 Aug 2026 01:10:11 +0530 Subject: [PATCH 1/4] fix(gateway-operator): recover failed gateway deployments Two independent defects each left an APIGateway permanently wedged. Helm: InstallOrUpgrade chose install versus upgrade purely on whether release history existed, and treated any history error as "absent". A release interrupted mid-operation keeps history in a pending state, so the operator kept calling upgrade and Helm kept refusing it. Read the full history, decide from the latest revision's status, and recover: purge and reinstall a revision-1 pending-install (nothing successful to lose), roll back a pending-upgrade or pending-rollback to the newest successful revision first, and report a clear error when no safe revision exists rather than deleting anything. Revisions are compared numerically, only a genuine not-found means absent, and install/upgrade now run context-aware with cleanup on failure. Reconciliation: exhausting the retry budget recorded the gateway as Deployed while persisting Programmed=False with observedGeneration equal to the CR generation. Neither the already-programmed branch nor the newer-generation branch then applied, so reconciliation fell through to "nothing to do" and only a generation bump could revive it. Record an explicit Failed state, keep reporting the failure truthfully, and give the gateway a bounded retry after the configured sync period. A change to the referenced ConfigMap or to deployment-affecting annotations retries immediately and resets the retry budget, while an unchanged failure requeues without deploying or writing status so a status-driven reconcile cannot hot-loop. Also removes an equal-generation check that was unreachable inside the newer-generation branch. Fixes #3105 --- .../controller/apigateway_controller.go | 235 ++++++- .../controller/apigateway_recovery_test.go | 612 ++++++++++++++++++ .../gateway-operator/internal/helm/client.go | 113 +++- .../internal/helm/recovery.go | 205 ++++++ .../internal/helm/recovery_test.go | 258 ++++++++ 5 files changed, 1378 insertions(+), 45 deletions(-) create mode 100644 kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go create mode 100644 kubernetes/gateway-operator/internal/helm/recovery.go create mode 100644 kubernetes/gateway-operator/internal/helm/recovery_test.go diff --git a/kubernetes/gateway-operator/internal/controller/apigateway_controller.go b/kubernetes/gateway-operator/internal/controller/apigateway_controller.go index 793bdaa3f9..3dddbd516f 100644 --- a/kubernetes/gateway-operator/internal/controller/apigateway_controller.go +++ b/kubernetes/gateway-operator/internal/controller/apigateway_controller.go @@ -18,6 +18,8 @@ package controller import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" "math" "sync" @@ -62,6 +64,11 @@ const ( GatewayTrackingStatusRetrying GatewayTrackingStatus = "Retrying" GatewayTrackingStatusDeployed GatewayTrackingStatus = "Deployed" GatewayTrackingStatusConfigChanged GatewayTrackingStatus = "ConfigChanged" + // GatewayTrackingStatusFailed marks a deployment that exhausted its retry budget. + // It is deliberately distinct from Deployed: recording an exhausted deployment as + // deployed made the failure terminal, because every recovery path then treated the + // gateway as finished. + GatewayTrackingStatusFailed GatewayTrackingStatus = "Failed" ) // GatewayTrackingEntry tracks the state of an APIGateway deployment @@ -71,6 +78,10 @@ type GatewayTrackingEntry struct { RetryCount int LastRetryTime time.Time NextRetryTime time.Time + // InputHash fingerprints the inputs that decide what would be deployed. A change means + // the previous failure may no longer apply, so the retry budget is reset and the next + // reconcile deploys immediately instead of waiting out the retry window. + InputHash string } // GatewayTracker manages in-memory tracking of APIGateway deployment states @@ -125,6 +136,29 @@ type GatewayReconciler struct { Config *config.OperatorConfig gatewayTracker *GatewayTracker Logger *slog.Logger + + // deployGateway performs a deployment. It defaults to processGatewayDeployment and is + // overridable so reconciliation decisions can be asserted without a cluster. + deployGateway func(ctx context.Context, gatewayConfig *apiv1.APIGateway, trackingKey string, generation int64, configHash, inputHash string) (ctrl.Result, error) +} + +// deploy dispatches to the configured deployment function, falling back to the real one so a +// directly constructed reconciler behaves normally. +func (r *GatewayReconciler) deploy(ctx context.Context, gatewayConfig *apiv1.APIGateway, trackingKey string, generation int64, configHash, inputHash string) (ctrl.Result, error) { + if r.deployGateway != nil { + return r.deployGateway(ctx, gatewayConfig, trackingKey, generation, configHash, inputHash) + } + return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, generation, configHash, inputHash) +} + +// retryCountFor returns the retry count a new attempt starts from. The budget carries over +// only while the same generation is retried against unchanged inputs, so a corrected input +// does not inherit an already-exhausted count. +func retryCountFor(existing *GatewayTrackingEntry, hasExisting bool, generation int64, inputHash string) int { + if hasExisting && existing != nil && existing.Generation == generation && existing.InputHash == inputHash { + return existing.RetryCount + } + return 0 } // NewGatewayReconciler creates a new GatewayReconciler @@ -208,17 +242,19 @@ func (r *GatewayReconciler) decideAndProcess( ) (ctrl.Result, error) { log := r.Logger.With(slog.String("controller", "APIGateway"), slog.String("name", gatewayConfig.Name)) + // Read the deployment inputs once: the ConfigMap values feed the persisted config hash, + // and their fingerprint (together with the CR infrastructure overlay) decides whether a + // previously failed deployment is worth retrying immediately. + values, inputHash, err := r.deploymentInputs(ctx, gatewayConfig) + if err != nil { + // If we can't read the config map, it might be transient or deleted. + // Treat it as an error so the reconcile is retried. + return ctrl.Result{}, fmt.Errorf("failed to get deployment inputs: %w", err) + } + // Calculate current config hash currentConfigHash := "" if gatewayConfig.Spec.ConfigRef != nil { - values, err := configMapValuesYAML(ctx, r.Client, gatewayConfig.Spec.ConfigRef.Name, gatewayConfig.Namespace) - if err != nil { - // If we can't read config map, it might be transient or deleted - // We should probably fail to deploy/reconcile - // But if we are already deployed, maybe we just log error? - // For now, let's treat it as error so we can retry - return ctrl.Result{}, fmt.Errorf("failed to get config map values: %w", err) - } currentConfigHash = auth.CalculateConfigHash(values) } @@ -277,6 +313,56 @@ func (r *GatewayReconciler) decideAndProcess( return ctrl.Result{}, nil } + // Case 1b: the CR generation has already been observed but the gateway is not + // programmed. An exhausted deployment lands here, and this is where it used to become + // terminal: the branch above requires Programmed=True and the branch below requires a + // newer generation, so an equal-generation failure fell through to "nothing to do" and + // could only be revived by bumping the CR generation. + if crGeneration == statusObservedGen && (programmedCond == nil || programmedCond.Status != metav1.ConditionTrue) { + if hasTrackingEntry && trackingEntry.Generation == crGeneration { + switch trackingEntry.Status { + case GatewayTrackingStatusProcessing: + // Another reconcile is already deploying this generation. + log.Debug("Already processing this generation, skipping", + slog.String("name", gatewayConfig.Name), + slog.Int64("generation", crGeneration)) + return ctrl.Result{}, nil + + case GatewayTrackingStatusDeployed: + // The deployment finished; only the success status has not been observed yet. + log.Debug("Deployment completed but status not yet propagated, skipping", + slog.String("name", gatewayConfig.Name), + slog.Int64("generation", crGeneration)) + return ctrl.Result{}, nil + + case GatewayTrackingStatusFailed: + deployNow, requeueAfter, reason := r.decideFailedRecovery(trackingEntry, inputHash, time.Now()) + if !deployNow { + // Requeue without deploying or writing status, so a status-driven + // reconcile cannot spin on a permanently failed gateway. + log.Debug("Failed APIGateway awaiting its retry window", + slog.String("name", gatewayConfig.Name), + slog.Int64("generation", crGeneration), + slog.Duration("retryIn", requeueAfter)) + return ctrl.Result{RequeueAfter: requeueAfter}, nil + } + log.Info("Recovering failed APIGateway deployment", + slog.String("name", gatewayConfig.Name), + slog.Int64("generation", crGeneration), + slog.String("reason", reason)) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) + } + } + + // Either the operator restarted and holds no entry for this generation, or the entry + // is mid-retry. Deploy so a persisted failure does not outlive the process that + // recorded it. + log.Info("Processing APIGateway that is not programmed at the observed generation", + slog.String("name", gatewayConfig.Name), + slog.Int64("generation", crGeneration)) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) + } + // Case 2: CR generation > status observed generation // Need to deploy/update if crGeneration > statusObservedGen { @@ -298,7 +384,7 @@ func (r *GatewayReconciler) decideAndProcess( slog.String("name", gatewayConfig.Name), slog.Int64("generation", crGeneration), slog.Int("retryCount", trackingEntry.RetryCount)) - return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) } // If Deployed but status not updated yet, wait for status propagation if trackingEntry.Status == GatewayTrackingStatusDeployed { @@ -307,12 +393,30 @@ func (r *GatewayReconciler) decideAndProcess( slog.Int64("generation", crGeneration)) return ctrl.Result{}, nil } + // Retry budget exhausted for this generation. Reached when the failure was + // recorded but the status write has not been observed, so the same bounded + // recovery applies as when the failed condition is visible. + if trackingEntry.Status == GatewayTrackingStatusFailed { + deployNow, requeueAfter, reason := r.decideFailedRecovery(trackingEntry, inputHash, time.Now()) + if !deployNow { + log.Debug("Failed APIGateway awaiting its retry window", + slog.String("name", gatewayConfig.Name), + slog.Int64("generation", crGeneration), + slog.Duration("retryIn", requeueAfter)) + return ctrl.Result{RequeueAfter: requeueAfter}, nil + } + log.Info("Recovering failed APIGateway deployment", + slog.String("name", gatewayConfig.Name), + slog.Int64("generation", crGeneration), + slog.String("reason", reason)) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) + } // If ConfigChanged, proceed to redeploy if trackingEntry.Status == GatewayTrackingStatusConfigChanged { log.Info("Processing APIGateway config change redeployment", slog.String("name", gatewayConfig.Name), slog.Int64("generation", crGeneration)) - return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) } } @@ -322,7 +426,7 @@ func (r *GatewayReconciler) decideAndProcess( slog.String("name", gatewayConfig.Name), slog.Int64("oldGeneration", trackingEntry.Generation), slog.Int64("newGeneration", crGeneration)) - return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) } } else { // No tracking entry @@ -331,7 +435,7 @@ func (r *GatewayReconciler) decideAndProcess( log.Info("Processing new Gateway", slog.String("name", gatewayConfig.Name), slog.Int64("generation", crGeneration)) - return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) } // Controller restart scenario: @@ -343,7 +447,7 @@ func (r *GatewayReconciler) decideAndProcess( slog.String("name", gatewayConfig.Name), slog.Int64("statusObservedGen", statusObservedGen), slog.Int64("crGeneration", crGeneration)) - return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) } if statusObservedGen == 0 && crGeneration > 1 { @@ -352,17 +456,12 @@ func (r *GatewayReconciler) decideAndProcess( log.Info("Controller restart detected - retrying incomplete initial deployment", slog.String("name", gatewayConfig.Name), slog.Int64("crGeneration", crGeneration)) - return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash) + return r.deploy(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash, inputHash) } - // statusObservedGen == crGeneration but condition is not True - // Something failed before, retry - if statusObservedGen == crGeneration { - log.Info("Retrying previously failed deployment", - slog.String("name", gatewayConfig.Name), - slog.Int64("generation", crGeneration)) - return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, crGeneration, currentConfigHash) - } + // An equal-generation check used to live here, unreachable inside this + // crGeneration > statusObservedGen branch. Equal generations are now handled by + // case 1b above, which is where a persisted failure is actually observed. } } @@ -381,21 +480,22 @@ func (r *GatewayReconciler) processGatewayDeployment( trackingKey string, generation int64, configHash string, + inputHash string, ) (ctrl.Result, error) { log := r.Logger.With(slog.String("controller", "APIGateway"), slog.String("name", gatewayConfig.Name)) - // Get existing entry to preserve retry count if retrying same generation + // Preserve the retry count only while the same generation is being retried against the + // same inputs. Changed inputs are a new attempt, so the retry budget starts again + // instead of the gateway inheriting an already-exhausted count. existingEntry, hasExisting := r.gatewayTracker.Get(trackingKey) - retryCount := 0 - if hasExisting && existingEntry.Generation == generation { - retryCount = existingEntry.RetryCount - } + retryCount := retryCountFor(existingEntry, hasExisting, generation, inputHash) // Update tracker to Processing entry := &GatewayTrackingEntry{ Generation: generation, Status: GatewayTrackingStatusProcessing, RetryCount: retryCount, + InputHash: inputHash, } r.gatewayTracker.Set(trackingKey, entry) @@ -557,17 +657,24 @@ func (r *GatewayReconciler) handleGatewayDeploymentError( } if entry.RetryCount >= maxRetries { + retryWindow := r.retryWindow() + log.Error("Max retries exceeded", slog.Any("error", err), slog.String("gateway", gatewayConfig.Name), slog.Int("retryCount", entry.RetryCount), - slog.Int("maxRetries", maxRetries)) - - // Mark as deployed (failed) - keeps tracking but won't retry - entry.Status = GatewayTrackingStatusDeployed + slog.Int("maxRetries", maxRetries), + slog.Duration("nextAttemptIn", retryWindow)) + + // Record the exhaustion as a failure, not as a deployment. The entry keeps the + // input fingerprint so corrected inputs retry immediately, and a next-retry time so + // an unchanged gateway still gets a bounded attempt later rather than never. + entry.Status = GatewayTrackingStatusFailed + entry.NextRetryTime = time.Now().Add(retryWindow) r.gatewayTracker.Set(trackingKey, entry) - // Update status with final failure + // Report the failure truthfully; the condition remains Programmed=False with + // DeploymentFailed against this generation. if updateErr := r.updateGatewayProgrammedCondition(ctx, gatewayConfig, metav1.Condition{ Type: apiv1.GatewayConditionProgrammed, Status: metav1.ConditionFalse, @@ -576,10 +683,12 @@ func (r *GatewayReconciler) handleGatewayDeploymentError( Message: fmt.Sprintf("Max retries (%d) exceeded. Last error: %s", maxRetries, err.Error()), LastTransitionTime: metav1.Now(), }, &selectedCount, ""); updateErr != nil { + // The tracker already holds the bounded retry, so recovery still happens even + // though this status write failed. return ctrl.Result{}, updateErr } - return ctrl.Result{}, nil + return ctrl.Result{RequeueAfter: retryWindow}, nil } // Calculate backoff @@ -609,6 +718,66 @@ func (r *GatewayReconciler) handleGatewayDeploymentError( return ctrl.Result{RequeueAfter: backoff}, nil } +// deploymentInputHash fingerprints everything that decides what would be deployed: the +// referenced ConfigMap's values and the overlay derived from the CR's infrastructure +// labels/annotations. Both arguments are already-marshalled YAML with sorted keys, so the +// result is stable across processes and does not depend on Go map iteration order. +func deploymentInputHash(configMapValues, crOverlay string) string { + sum := sha256.Sum256([]byte(configMapValues + "\x00" + crOverlay)) + return hex.EncodeToString(sum[:]) +} + +// deploymentInputs reads the inputs that feed a deployment and returns the ConfigMap values +// (used for the persisted config hash) together with a fingerprint of every deployment +// input (used for the retry decision). +func (r *GatewayReconciler) deploymentInputs(ctx context.Context, gatewayConfig *apiv1.APIGateway) (values string, inputHash string, err error) { + if gatewayConfig.Spec.ConfigRef != nil { + values, err = configMapValuesYAML(ctx, r.Client, gatewayConfig.Spec.ConfigRef.Name, gatewayConfig.Namespace) + if err != nil { + return "", "", err + } + } + crOverlay, err := buildCRValuesOverlay(gatewayConfig) + if err != nil { + return "", "", err + } + return values, deploymentInputHash(values, crOverlay), nil +} + +// retryWindow is how long an exhausted deployment waits before it is given another bounded +// attempt. It follows the configured sync period so recovery happens on the same cadence as +// the operator's periodic resync. +func (r *GatewayReconciler) retryWindow() time.Duration { + if r.Config != nil && r.Config.Reconciliation.SyncPeriod > 0 { + return r.Config.Reconciliation.SyncPeriod + } + return 10 * time.Minute +} + +// decideFailedRecovery decides what to do with a deployment that already exhausted its retry +// budget. It returns whether to deploy now, and otherwise how long to wait. +// +// Without this, an exhausted deployment was unreachable: the persisted condition carried +// Programmed=False with observedGeneration equal to the CR generation, so neither the +// "already programmed" nor the "newer generation" path applied and reconciliation fell +// through to "nothing to do". Recovery had to wait for the CR generation to change, which +// no amount of fixing the referenced ConfigMap or restarting the operator achieves. +func (r *GatewayReconciler) decideFailedRecovery( + entry *GatewayTrackingEntry, + currentInputHash string, + now time.Time, +) (deployNow bool, requeueAfter time.Duration, reason string) { + if entry.InputHash != currentInputHash { + return true, 0, "deployment inputs changed since the failure" + } + if entry.NextRetryTime.IsZero() || !now.Before(entry.NextRetryTime) { + return true, 0, "retry window elapsed" + } + // Still inside the window. Requeue rather than deploy so a status-driven reconcile + // cannot spin: nothing is written and no deployment is attempted. + return false, entry.NextRetryTime.Sub(now), "waiting for the retry window" +} + // calculateBackoff calculates exponential backoff duration using operator configuration func (r *GatewayReconciler) calculateBackoff(retryCount int) time.Duration { cfg := r.Config.Reconciliation diff --git a/kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go b/kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go new file mode 100644 index 0000000000..ae361570d7 --- /dev/null +++ b/kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go @@ -0,0 +1,612 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package controller + +import ( + "context" + "errors" + "io" + "log/slog" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + apiv1 "github.com/wso2/api-platform/kubernetes/gateway-operator/api/v1" + "github.com/wso2/api-platform/kubernetes/gateway-operator/internal/config" +) + +const ( + testSyncPeriod = 10 * time.Minute + testNamespace = "gw-ns" + testGateway = "gw" + testTrackKey = testNamespace + "/" + testGateway +) + +func recoveryScheme(t *testing.T) *runtime.Scheme { + t.Helper() + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("add corev1 to scheme: %v", err) + } + if err := apiv1.AddToScheme(scheme); err != nil { + t.Fatalf("add apiv1 to scheme: %v", err) + } + return scheme +} + +func recoveryConfig() *config.OperatorConfig { + cfg := &config.OperatorConfig{} + cfg.Reconciliation.MaxRetryAttempts = 3 + cfg.Reconciliation.InitialBackoff = time.Second + cfg.Reconciliation.MaxBackoffDuration = time.Minute + cfg.Reconciliation.SyncPeriod = testSyncPeriod + return cfg +} + +// failedGateway builds an APIGateway whose persisted status reports the exhausted failure: +// Programmed=False with DeploymentFailed at the current generation. This is the exact shape +// that used to be terminal. +func failedGateway(generation int64) *apiv1.APIGateway { + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{ + Name: testGateway, + Namespace: testNamespace, + Generation: generation, + Finalizers: []string{apigatewayFinalizerName}, + }, + } + meta.SetStatusCondition(&gw.Status.Conditions, metav1.Condition{ + Type: apiv1.GatewayConditionProgrammed, + Status: metav1.ConditionFalse, + ObservedGeneration: generation, + Reason: apiv1.GatewayProgrammedReasonDeploymentFailed, + Message: "Max retries (3) exceeded. Last error: boom", + LastTransitionTime: metav1.Now(), + }) + return gw +} + +// newRecoveryReconciler wires a reconciler whose deployments are recorded instead of +// executed, so a reconciliation decision can be asserted without a cluster or Helm. +func newRecoveryReconciler(t *testing.T, objs ...runtime.Object) (*GatewayReconciler, *int) { + t.Helper() + scheme := recoveryScheme(t) + builder := fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(&apiv1.APIGateway{}) + for _, o := range objs { + builder = builder.WithRuntimeObjects(o) + } + deployCalls := 0 + r := &GatewayReconciler{ + Client: builder.Build(), + Scheme: scheme, + Config: recoveryConfig(), + gatewayTracker: NewGatewayTracker(), + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + } + r.deployGateway = func(_ context.Context, _ *apiv1.APIGateway, trackingKey string, generation int64, _, inputHash string) (ctrl.Result, error) { + deployCalls++ + r.gatewayTracker.Set(trackingKey, &GatewayTrackingEntry{ + Generation: generation, + Status: GatewayTrackingStatusProcessing, + RetryCount: retryCountFor(mustEntry(r, trackingKey), true, generation, inputHash), + InputHash: inputHash, + }) + return ctrl.Result{}, nil + } + return r, &deployCalls +} + +func mustEntry(r *GatewayReconciler, key string) *GatewayTrackingEntry { + entry, _ := r.gatewayTracker.Get(key) + return entry +} + +// typesName is the namespaced name of the gateway under test. +func typesName() types.NamespacedName { + return types.NamespacedName{Namespace: testNamespace, Name: testGateway} +} + +// callDecide runs the reconciliation decision for a gateway using its persisted status. +func callDecide(t *testing.T, r *GatewayReconciler, gw *apiv1.APIGateway) (ctrl.Result, error) { + t.Helper() + cond := meta.FindStatusCondition(gw.Status.Conditions, apiv1.GatewayConditionProgrammed) + observed := int64(0) + if cond != nil { + observed = cond.ObservedGeneration + } + entry, has := r.gatewayTracker.Get(testTrackKey) + return r.decideAndProcess(context.Background(), gw, testTrackKey, gw.Generation, observed, cond, entry, has) +} + +// --- exhaustion bookkeeping ------------------------------------------------------------- + +func TestHandleGatewayDeploymentError_exhaustionRecordsFailedNotDeployed(t *testing.T) { + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{Name: testGateway, Namespace: testNamespace, Generation: 1}, + } + r, _ := newRecoveryReconciler(t, gw) + + // One attempt short of the limit, so this call exhausts the budget. + entry := &GatewayTrackingEntry{Generation: 1, Status: GatewayTrackingStatusProcessing, RetryCount: 2, InputHash: "h1"} + r.gatewayTracker.Set(testTrackKey, entry) + + before := time.Now() + res, err := r.handleGatewayDeploymentError(context.Background(), gw, testTrackKey, entry, errors.New("boom"), 0) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + stored, ok := r.gatewayTracker.Get(testTrackKey) + if !ok { + t.Fatal("expected a tracking entry after exhaustion") + } + // The regression: exhaustion used to be recorded as Deployed, which made every recovery + // path treat the gateway as finished. + if stored.Status != GatewayTrackingStatusFailed { + t.Errorf("status = %q, want %q", stored.Status, GatewayTrackingStatusFailed) + } + if stored.Status == GatewayTrackingStatusDeployed { + t.Error("an exhausted deployment must never be recorded as deployed") + } + if !stored.NextRetryTime.After(before) { + t.Errorf("NextRetryTime = %v, want a bounded future retry", stored.NextRetryTime) + } + if res.RequeueAfter != testSyncPeriod { + t.Errorf("RequeueAfter = %v, want %v", res.RequeueAfter, testSyncPeriod) + } +} + +func TestHandleGatewayDeploymentError_exhaustionWritesDeploymentFailed(t *testing.T) { + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{Name: testGateway, Namespace: testNamespace, Generation: 1}, + } + r, _ := newRecoveryReconciler(t, gw) + entry := &GatewayTrackingEntry{Generation: 1, Status: GatewayTrackingStatusProcessing, RetryCount: 2} + r.gatewayTracker.Set(testTrackKey, entry) + + if _, err := r.handleGatewayDeploymentError(context.Background(), gw, testTrackKey, entry, errors.New("boom"), 0); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + stored := &apiv1.APIGateway{} + if err := r.Get(context.Background(), typesName(), stored); err != nil { + t.Fatalf("get gateway: %v", err) + } + cond := meta.FindStatusCondition(stored.Status.Conditions, apiv1.GatewayConditionProgrammed) + if cond == nil { + t.Fatal("expected a Programmed condition") + } + if cond.Status != metav1.ConditionFalse { + t.Errorf("condition status = %v, want False", cond.Status) + } + if cond.Reason != apiv1.GatewayProgrammedReasonDeploymentFailed { + t.Errorf("reason = %q, want %q", cond.Reason, apiv1.GatewayProgrammedReasonDeploymentFailed) + } + if cond.ObservedGeneration != 1 { + t.Errorf("observedGeneration = %d, want 1", cond.ObservedGeneration) + } +} + +func TestHandleGatewayDeploymentError_belowLimitStillRetries(t *testing.T) { + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{Name: testGateway, Namespace: testNamespace, Generation: 1}, + } + r, _ := newRecoveryReconciler(t, gw) + entry := &GatewayTrackingEntry{Generation: 1, Status: GatewayTrackingStatusProcessing, RetryCount: 0} + r.gatewayTracker.Set(testTrackKey, entry) + + res, err := r.handleGatewayDeploymentError(context.Background(), gw, testTrackKey, entry, errors.New("boom"), 0) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + stored, _ := r.gatewayTracker.Get(testTrackKey) + if stored.Status != GatewayTrackingStatusRetrying { + t.Errorf("status = %q, want %q (unchanged behaviour below the limit)", stored.Status, GatewayTrackingStatusRetrying) + } + if res.RequeueAfter <= 0 { + t.Error("expected a backoff requeue while retries remain") + } +} + +// --- recovery decisions ------------------------------------------------------------------ + +func TestDecideAndProcess_restartWithPersistedFailureRecovers(t *testing.T) { + // Controller restarted: the failure is persisted but no tracker entry exists. This used + // to fall through to "nothing to do" and wedge the gateway permanently. + gw := failedGateway(1) + r, calls := newRecoveryReconciler(t, gw) + + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 1 { + t.Fatalf("deploy attempts = %d, want 1 after a restart with a persisted failure", *calls) + } +} + +func TestDecideAndProcess_unchangedFailureBeforeWindowDoesNotDeploy(t *testing.T) { + gw := failedGateway(1) + r, calls := newRecoveryReconciler(t, gw) + + _, inputHash, err := r.deploymentInputs(context.Background(), gw) + if err != nil { + t.Fatalf("deploymentInputs: %v", err) + } + r.gatewayTracker.Set(testTrackKey, &GatewayTrackingEntry{ + Generation: 1, + Status: GatewayTrackingStatusFailed, + RetryCount: 3, + InputHash: inputHash, + NextRetryTime: time.Now().Add(testSyncPeriod), + }) + + res, err := callDecide(t, r, gw) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 0 { + t.Errorf("deploy attempts = %d, want 0 while inside the retry window", *calls) + } + if res.RequeueAfter <= 0 || res.RequeueAfter > testSyncPeriod { + t.Errorf("RequeueAfter = %v, want a bounded wait within the retry window", res.RequeueAfter) + } +} + +func TestDecideAndProcess_statusOnlyReconcilesDoNotHotLoop(t *testing.T) { + // Repeated status-driven reconciles of an unchanged failure must not deploy and must not + // write status, otherwise each write triggers the next reconcile. + gw := failedGateway(1) + r, calls := newRecoveryReconciler(t, gw) + + _, inputHash, err := r.deploymentInputs(context.Background(), gw) + if err != nil { + t.Fatalf("deploymentInputs: %v", err) + } + r.gatewayTracker.Set(testTrackKey, &GatewayTrackingEntry{ + Generation: 1, + Status: GatewayTrackingStatusFailed, + RetryCount: 3, + InputHash: inputHash, + NextRetryTime: time.Now().Add(testSyncPeriod), + }) + + for i := 0; i < 5; i++ { + res, err := callDecide(t, r, gw) + if err != nil { + t.Fatalf("iteration %d: unexpected error: %v", i, err) + } + if res.RequeueAfter <= 0 { + t.Fatalf("iteration %d: expected a timed requeue, got %v", i, res) + } + } + if *calls != 0 { + t.Errorf("deploy attempts = %d, want 0 across repeated status reconciles", *calls) + } + + // The condition must be untouched, so no status write re-triggers reconciliation. + stored := &apiv1.APIGateway{} + if err := r.Get(context.Background(), typesName(), stored); err != nil { + t.Fatalf("get gateway: %v", err) + } + cond := meta.FindStatusCondition(stored.Status.Conditions, apiv1.GatewayConditionProgrammed) + if cond == nil || cond.Reason != apiv1.GatewayProgrammedReasonDeploymentFailed { + t.Errorf("condition = %v, want the failure left untouched", cond) + } +} + +func TestDecideAndProcess_retryWindowElapsedDeploysOnce(t *testing.T) { + gw := failedGateway(1) + r, calls := newRecoveryReconciler(t, gw) + + _, inputHash, err := r.deploymentInputs(context.Background(), gw) + if err != nil { + t.Fatalf("deploymentInputs: %v", err) + } + r.gatewayTracker.Set(testTrackKey, &GatewayTrackingEntry{ + Generation: 1, + Status: GatewayTrackingStatusFailed, + RetryCount: 3, + InputHash: inputHash, + NextRetryTime: time.Now().Add(-time.Second), // window already elapsed + }) + + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 1 { + t.Errorf("deploy attempts = %d, want exactly 1 once the retry window elapsed", *calls) + } +} + +func TestDecideAndProcess_configMapChangeRecoversImmediately(t *testing.T) { + gw := failedGateway(1) + gw.Spec.ConfigRef = &corev1.LocalObjectReference{Name: "gw-values"} + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "gw-values", Namespace: testNamespace}, + Data: map[string]string{"values.yaml": "replicaCount: 2\n"}, + } + r, calls := newRecoveryReconciler(t, gw, cm) + + // The failure was recorded against different ConfigMap content. + r.gatewayTracker.Set(testTrackKey, &GatewayTrackingEntry{ + Generation: 1, + Status: GatewayTrackingStatusFailed, + RetryCount: 3, + InputHash: "stale-hash", + NextRetryTime: time.Now().Add(testSyncPeriod), // window has NOT elapsed + }) + + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 1 { + t.Errorf("deploy attempts = %d, want 1 immediately after a ConfigMap change", *calls) + } +} + +func TestDecideAndProcess_annotationChangeRecoversImmediately(t *testing.T) { + gw := failedGateway(1) + r, calls := newRecoveryReconciler(t, gw) + + // Record the failure against the gateway with no infrastructure annotations. + _, baseHash, err := r.deploymentInputs(context.Background(), gw) + if err != nil { + t.Fatalf("deploymentInputs: %v", err) + } + r.gatewayTracker.Set(testTrackKey, &GatewayTrackingEntry{ + Generation: 1, + Status: GatewayTrackingStatusFailed, + RetryCount: 3, + InputHash: baseHash, + NextRetryTime: time.Now().Add(testSyncPeriod), + }) + + // A deployment-affecting annotation changes what would be rendered. + gw.Spec.Infrastructure = &apiv1.GatewayInfrastructure{ + Annotations: map[string]string{"example.com/rollout": "2"}, + } + + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 1 { + t.Errorf("deploy attempts = %d, want 1 immediately after an annotation change", *calls) + } +} + +func TestDecideAndProcess_failedStatusNotYetObservedStillRecovers(t *testing.T) { + // The failure was recorded in the tracker but the status write never became visible, so + // the condition still shows the earlier retry with observedGeneration 0. + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{ + Name: testGateway, Namespace: testNamespace, Generation: 1, + Finalizers: []string{apigatewayFinalizerName}, + }, + } + meta.SetStatusCondition(&gw.Status.Conditions, metav1.Condition{ + Type: apiv1.GatewayConditionProgrammed, + Status: metav1.ConditionFalse, + ObservedGeneration: 0, + Reason: apiv1.GatewayProgrammedReasonRetrying, + LastTransitionTime: metav1.Now(), + }) + r, calls := newRecoveryReconciler(t, gw) + + r.gatewayTracker.Set(testTrackKey, &GatewayTrackingEntry{ + Generation: 1, + Status: GatewayTrackingStatusFailed, + RetryCount: 3, + InputHash: "stale-hash", // inputs since corrected + NextRetryTime: time.Now().Add(testSyncPeriod), + }) + + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 1 { + t.Errorf("deploy attempts = %d, want 1 even when the failure status is not yet observable", *calls) + } +} + +func TestDecideAndProcess_programmedGatewayIsUntouched(t *testing.T) { + // Existing success behaviour must not change. + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{ + Name: testGateway, Namespace: testNamespace, Generation: 2, + Finalizers: []string{apigatewayFinalizerName}, + }, + } + meta.SetStatusCondition(&gw.Status.Conditions, metav1.Condition{ + Type: apiv1.GatewayConditionProgrammed, + Status: metav1.ConditionTrue, + ObservedGeneration: 2, + Reason: apiv1.GatewayProgrammedReasonProgrammed, + LastTransitionTime: metav1.Now(), + }) + r, calls := newRecoveryReconciler(t, gw) + + // The already-programmed path also re-registers the gateway, which needs the gateway's + // Services to exist. That registration is unrelated to recovery, so its error is ignored + // here; what matters is that no deployment is triggered and the tracker records Deployed. + _, _ = callDecide(t, r, gw) + + if *calls != 0 { + t.Errorf("deploy attempts = %d, want 0 for an already programmed gateway", *calls) + } + stored, ok := r.gatewayTracker.Get(testTrackKey) + if !ok || stored.Status != GatewayTrackingStatusDeployed { + t.Errorf("tracker = %v, want a Deployed entry", stored) + } +} + +func TestDecideAndProcess_newerGenerationStillDeploys(t *testing.T) { + // A newer generation over a persisted failure must deploy, as before. + gw := failedGateway(1) + gw.Generation = 2 + r, calls := newRecoveryReconciler(t, gw) + + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 1 { + t.Errorf("deploy attempts = %d, want 1 for a newer generation", *calls) + } +} + +func TestDecideAndProcess_processingGenerationIsNotRedeployed(t *testing.T) { + gw := failedGateway(1) + r, calls := newRecoveryReconciler(t, gw) + r.gatewayTracker.Set(testTrackKey, &GatewayTrackingEntry{ + Generation: 1, + Status: GatewayTrackingStatusProcessing, + }) + + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 0 { + t.Errorf("deploy attempts = %d, want 0 while a deployment is in progress", *calls) + } +} + +// --- pure helpers ------------------------------------------------------------------------ + +func TestDecideFailedRecovery(t *testing.T) { + now := time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC) + + tests := []struct { + name string + entry *GatewayTrackingEntry + inputHash string + wantDeploy bool + wantRequeueGT time.Duration + }{ + { + name: "changed inputs deploy immediately", + entry: &GatewayTrackingEntry{InputHash: "old", NextRetryTime: now.Add(time.Hour)}, + inputHash: "new", + wantDeploy: true, + }, + { + name: "window elapsed deploys", + entry: &GatewayTrackingEntry{InputHash: "same", NextRetryTime: now.Add(-time.Second)}, + inputHash: "same", + wantDeploy: true, + }, + { + name: "exact boundary deploys", + entry: &GatewayTrackingEntry{InputHash: "same", NextRetryTime: now}, + inputHash: "same", + wantDeploy: true, + }, + { + name: "zero next retry time deploys rather than waiting forever", + entry: &GatewayTrackingEntry{InputHash: "same"}, + inputHash: "same", + wantDeploy: true, + }, + { + name: "inside the window waits", + entry: &GatewayTrackingEntry{InputHash: "same", NextRetryTime: now.Add(5 * time.Minute)}, + inputHash: "same", + wantDeploy: false, + wantRequeueGT: 4 * time.Minute, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + r := &GatewayReconciler{Config: recoveryConfig()} + deployNow, requeueAfter, reason := r.decideFailedRecovery(tc.entry, tc.inputHash, now) + if deployNow != tc.wantDeploy { + t.Errorf("deployNow = %v, want %v", deployNow, tc.wantDeploy) + } + if !tc.wantDeploy && requeueAfter <= tc.wantRequeueGT { + t.Errorf("requeueAfter = %v, want > %v", requeueAfter, tc.wantRequeueGT) + } + if reason == "" { + t.Error("expected a reason for the decision") + } + }) + } +} + +func TestRetryCountFor_resetsWhenInputsChange(t *testing.T) { + exhausted := &GatewayTrackingEntry{Generation: 1, RetryCount: 3, InputHash: "old"} + + if got := retryCountFor(exhausted, true, 1, "old"); got != 3 { + t.Errorf("same generation and inputs: retryCount = %d, want 3 (budget preserved)", got) + } + if got := retryCountFor(exhausted, true, 1, "new"); got != 0 { + t.Errorf("changed inputs: retryCount = %d, want 0 (budget reset)", got) + } + if got := retryCountFor(exhausted, true, 2, "old"); got != 0 { + t.Errorf("new generation: retryCount = %d, want 0", got) + } + if got := retryCountFor(nil, false, 1, "old"); got != 0 { + t.Errorf("no entry: retryCount = %d, want 0", got) + } + // Nil entry with hasExisting set must not panic. + if got := retryCountFor(nil, true, 1, "old"); got != 0 { + t.Errorf("nil entry: retryCount = %d, want 0", got) + } +} + +func TestDeploymentInputHash(t *testing.T) { + base := deploymentInputHash("a: 1\n", "commonAnnotations:\n x: y\n") + + if base != deploymentInputHash("a: 1\n", "commonAnnotations:\n x: y\n") { + t.Error("hash must be stable for identical inputs") + } + if base == deploymentInputHash("a: 2\n", "commonAnnotations:\n x: y\n") { + t.Error("ConfigMap content must affect the hash") + } + if base == deploymentInputHash("a: 1\n", "commonAnnotations:\n x: z\n") { + t.Error("deployment-affecting annotations must affect the hash") + } + // The separator must stop different splits from colliding. + if deploymentInputHash("ab", "c") == deploymentInputHash("a", "bc") { + t.Error("field boundaries must be unambiguous") + } +} + +func TestRetryWindow_fallsBackWhenUnset(t *testing.T) { + r := &GatewayReconciler{Config: &config.OperatorConfig{}} + if got := r.retryWindow(); got <= 0 { + t.Errorf("retryWindow = %v, want a positive fallback", got) + } + // A nil config must not panic; recovery still needs a bounded window. + rNil := &GatewayReconciler{} + if got := rNil.retryWindow(); got <= 0 { + t.Errorf("retryWindow with nil config = %v, want a positive fallback", got) + } + if got := r.retryWindow(); got != 10*time.Minute { + t.Errorf("retryWindow fallback = %v, want 10m", got) + } + rSet := &GatewayReconciler{Config: recoveryConfig()} + if got := rSet.retryWindow(); got != testSyncPeriod { + t.Errorf("retryWindow = %v, want the configured sync period %v", got, testSyncPeriod) + } +} diff --git a/kubernetes/gateway-operator/internal/helm/client.go b/kubernetes/gateway-operator/internal/helm/client.go index 1f3e441a45..fc069cd9c3 100644 --- a/kubernetes/gateway-operator/internal/helm/client.go +++ b/kubernetes/gateway-operator/internal/helm/client.go @@ -158,19 +158,103 @@ func (c *Client) InstallOrUpgrade(ctx context.Context, opts InstallOrUpgradeOpti // Set registry client in action configuration actionConfig.RegistryClient = c.registryClient - // Check if release exists + // Read the full history so a pending release can be recovered. Only a genuine + // "not found" means the release is absent; any other failure must not be mistaken for + // one, or a transient storage error would trigger a fresh install over a live release. histClient := action.NewHistory(actionConfig) - histClient.Max = 1 - _, err = histClient.Run(opts.ReleaseName) - releaseExists := err == nil + histClient.Max = maxHistoryRevisions + history, err := histClient.Run(opts.ReleaseName) + switch { + case errors.Is(err, driver.ErrReleaseNotFound): + log.Info("Release does not exist, performing install", "release", opts.ReleaseName) + return c.install(ctx, actionConfig, opts) + case err != nil: + return fmt.Errorf("failed to read history for release %q: %w", opts.ReleaseName, err) + } + + plan, err := planRelease(history) + if err != nil { + return fmt.Errorf("cannot deploy release %q: %w", opts.ReleaseName, err) + } + + log.Info("Planned Helm operation", + "release", opts.ReleaseName, + "operation", plan.operation.String(), + "reason", plan.reason) + + switch plan.operation { + case operationInstall: + return c.install(ctx, actionConfig, opts) - if releaseExists { - log.Info("Release exists, performing upgrade", "release", opts.ReleaseName) + case operationUpgrade: return c.upgrade(ctx, actionConfig, opts) + + case operationPurgeThenInstall: + // Drop the stuck release and its history so the install below is not itself + // rejected as an in-progress operation. Safe only because planRelease proved there + // is no successful revision to preserve. + if err := c.purge(ctx, actionConfig, opts); err != nil { + return err + } + return c.install(ctx, actionConfig, opts) + + case operationRollbackThenUpgrade: + // Roll back first so the release leaves the pending state; the upgrade then starts + // from a known-good revision. If the rollback succeeds and the process dies before + // the upgrade, the next reconcile sees a deployed release and simply upgrades. + if err := c.rollback(ctx, actionConfig, opts, plan.rollbackRevision); err != nil { + return err + } + return c.upgrade(ctx, actionConfig, opts) + + default: + return fmt.Errorf("unsupported Helm operation %q for release %q", plan.operation, opts.ReleaseName) + } +} + +// purge removes a release together with its history so that a subsequent install is not +// rejected because an operation is still recorded as in progress. +func (c *Client) purge(ctx context.Context, actionConfig *action.Configuration, opts InstallOrUpgradeOptions) error { + log := log.FromContext(ctx) + + client := action.NewUninstall(actionConfig) + client.KeepHistory = false + client.Wait = opts.Wait + if opts.Timeout > 0 { + client.Timeout = time.Duration(opts.Timeout) * time.Second + } + + log.Info("Purging unrecoverable release before reinstall", "release", opts.ReleaseName) + if _, err := client.Run(opts.ReleaseName); err != nil { + // An already-absent release is the desired end state for this step. + if errors.Is(err, driver.ErrReleaseNotFound) { + log.Info("Release already absent, continuing with install", "release", opts.ReleaseName) + return nil + } + return fmt.Errorf("failed to purge release %q: %w", opts.ReleaseName, err) + } + return nil +} + +// rollback returns a release stuck in a pending state to the given known-good revision. +func (c *Client) rollback(ctx context.Context, actionConfig *action.Configuration, opts InstallOrUpgradeOptions, revision int) error { + log := log.FromContext(ctx) + + client := action.NewRollback(actionConfig) + client.Version = revision + client.Wait = opts.Wait + client.CleanupOnFail = true + if opts.Timeout > 0 { + client.Timeout = time.Duration(opts.Timeout) * time.Second } - log.Info("Release does not exist, performing install", "release", opts.ReleaseName) - return c.install(ctx, actionConfig, opts) + log.Info("Rolling back release to recover from pending state", + "release", opts.ReleaseName, + "revision", revision) + if err := client.Run(opts.ReleaseName); err != nil { + return fmt.Errorf("failed to roll back release %q to revision %d: %w", opts.ReleaseName, revision, err) + } + return nil } // install performs a Helm install @@ -227,8 +311,9 @@ func (c *Client) install(ctx context.Context, actionConfig *action.Configuration return fmt.Errorf("failed to parse values: %w", err) } - // Install the chart - release, err := client.Run(chart, values) + // Install the chart. RunWithContext lets an operator shutdown or a cancelled reconcile + // abort the wait instead of blocking for the full timeout. + release, err := client.RunWithContext(ctx, chart, values) if err != nil { return fmt.Errorf("failed to install chart: %w", err) } @@ -249,6 +334,9 @@ func (c *Client) upgrade(ctx context.Context, actionConfig *action.Configuration client.Namespace = opts.Namespace client.Wait = opts.Wait client.Version = opts.Version + // Roll partially-applied resources back when an upgrade fails, so a failed attempt + // leaves a usable revision rather than a half-applied one for the next reconcile. + client.CleanupOnFail = true if opts.Timeout > 0 { client.Timeout = time.Duration(opts.Timeout) * time.Second @@ -293,8 +381,9 @@ func (c *Client) upgrade(ctx context.Context, actionConfig *action.Configuration return fmt.Errorf("failed to parse values: %w", err) } - // Upgrade the chart - release, err := client.Run(opts.ReleaseName, chart, values) + // Upgrade the chart. RunWithContext lets an operator shutdown or a cancelled reconcile + // abort the wait instead of blocking for the full timeout. + release, err := client.RunWithContext(ctx, opts.ReleaseName, chart, values) if err != nil { return fmt.Errorf("failed to upgrade chart: %w", err) } diff --git a/kubernetes/gateway-operator/internal/helm/recovery.go b/kubernetes/gateway-operator/internal/helm/recovery.go new file mode 100644 index 0000000000..fc66bafcf9 --- /dev/null +++ b/kubernetes/gateway-operator/internal/helm/recovery.go @@ -0,0 +1,205 @@ +/* +Copyright 2025. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package helm + +import ( + "fmt" + + "helm.sh/helm/v3/pkg/release" +) + +// maxHistoryRevisions bounds how many stored revisions are read when planning. Helm's +// action.History truncates to Max, and treats a Max of 0 as "return nothing", so this must +// be a positive number large enough to cover any retained history (Helm keeps 10 by +// default) rather than left unset. +const maxHistoryRevisions = 1024 + +// releaseOperation is the operation needed to move a release towards the desired chart. +type releaseOperation int + +const ( + // operationInstall installs the release; there is nothing to preserve. + operationInstall releaseOperation = iota + // operationUpgrade upgrades the release in place. + operationUpgrade + // operationPurgeThenInstall removes an unusable release (history included) and installs + // it again. Only planned when no successful revision would be discarded. + operationPurgeThenInstall + // operationRollbackThenUpgrade rolls the release back to a known-good revision so the + // storage leaves the pending state, then upgrades from there. + operationRollbackThenUpgrade +) + +func (o releaseOperation) String() string { + switch o { + case operationInstall: + return "install" + case operationUpgrade: + return "upgrade" + case operationPurgeThenInstall: + return "purge-then-install" + case operationRollbackThenUpgrade: + return "rollback-then-upgrade" + default: + return fmt.Sprintf("unknown(%d)", int(o)) + } +} + +// recoveryPlan is the decision taken from a release's stored history. It is produced by +// planRelease, which is pure so every recovery rule is unit-testable without a cluster. +type recoveryPlan struct { + // operation is what the caller must perform. + operation releaseOperation + // rollbackRevision is the revision to roll back to, set only for + // operationRollbackThenUpgrade. + rollbackRevision int + // reason explains the decision and is logged before acting, so an operator can see why + // a destructive step was taken. + reason string +} + +// planRelease decides how to reach the desired chart from the release's stored history. +// +// A Helm operation that is interrupted rather than failing cleanly leaves the release in a +// pending state. Helm refuses to upgrade such a release ("another operation is in +// progress"), so choosing install-versus-upgrade purely on whether history exists wedges +// the release permanently. Each pending state is recovered here instead: +// +// - pending-install at revision 1 has no earlier successful revision to preserve, so the +// release is purged and installed again; +// - pending-upgrade and pending-rollback are rolled back to the newest successful +// revision, which returns the storage to a usable state before the upgrade; +// - a pending state with no successful revision to recover to is reported as an error +// rather than resolved destructively. +// +// The caller must pass the full history; revisions are compared numerically so no ordering +// is assumed. +func planRelease(history []*release.Release) (recoveryPlan, error) { + latest := latestRevision(history) + if latest == nil { + // No usable history: nothing to preserve and nothing to upgrade from. + return recoveryPlan{operation: operationInstall, reason: "no release history found"}, nil + } + + status := release.StatusUnknown + if latest.Info != nil { + status = latest.Info.Status + } + + switch status { + case release.StatusDeployed: + return recoveryPlan{ + operation: operationUpgrade, + reason: fmt.Sprintf("latest revision %d is deployed", latest.Version), + }, nil + + case release.StatusUninstalled: + // History was kept after an uninstall (--keep-history). There is no deployed + // revision to upgrade from, so install again. + return recoveryPlan{ + operation: operationInstall, + reason: fmt.Sprintf("latest revision %d is uninstalled", latest.Version), + }, nil + + case release.StatusUninstalling: + // An interrupted uninstall is ambiguous: the recorded intent was removal, so + // neither upgrading nor purging can be shown to be what the operator wanted. + return recoveryPlan{}, fmt.Errorf( + "release is stuck in %s at revision %d; complete or roll back the uninstall manually before redeploying", + status, latest.Version) + + case release.StatusPendingInstall: + // Purging is only safe while there is no successful revision to lose. A + // pending-install normally only exists at revision 1, so anything else is treated + // as a pending state needing rollback rather than a purge. + successful := latestSuccessfulRevision(history) + if latest.Version == 1 && successful == 0 { + return recoveryPlan{ + operation: operationPurgeThenInstall, + reason: "revision 1 is stuck in pending-install and no successful revision exists", + }, nil + } + if successful > 0 { + return recoveryPlan{ + operation: operationRollbackThenUpgrade, + rollbackRevision: successful, + reason: fmt.Sprintf("latest revision %d is stuck in %s; recovering to revision %d", + latest.Version, status, successful), + }, nil + } + return recoveryPlan{}, fmt.Errorf( + "release is stuck in %s at revision %d with no successful revision to recover to; resolve it manually before redeploying", + status, latest.Version) + + case release.StatusPendingUpgrade, release.StatusPendingRollback: + successful := latestSuccessfulRevision(history) + if successful == 0 { + return recoveryPlan{}, fmt.Errorf( + "release is stuck in %s at revision %d with no successful revision to recover to; resolve it manually before redeploying", + status, latest.Version) + } + return recoveryPlan{ + operation: operationRollbackThenUpgrade, + rollbackRevision: successful, + reason: fmt.Sprintf("latest revision %d is stuck in %s; recovering to revision %d", + latest.Version, status, successful), + }, nil + + default: + // failed, superseded and unknown are all states Helm can upgrade from. Preserving + // the upgrade here keeps the existing behaviour for an ordinary failed release. + return recoveryPlan{ + operation: operationUpgrade, + reason: fmt.Sprintf("latest revision %d has status %s", latest.Version, status), + }, nil + } +} + +// latestRevision returns the entry with the highest revision number. Helm sorts history +// before returning it, but the order is not part of the contract this package relies on. +func latestRevision(history []*release.Release) *release.Release { + var latest *release.Release + for _, rel := range history { + if rel == nil { + continue + } + if latest == nil || rel.Version > latest.Version { + latest = rel + } + } + return latest +} + +// latestSuccessfulRevision returns the highest revision that reached a usable state, or 0 +// when there is none. Deployed is the current successful revision; superseded revisions +// were deployed successfully before a later revision replaced them, so both are valid +// rollback targets. +func latestSuccessfulRevision(history []*release.Release) int { + best := 0 + for _, rel := range history { + if rel == nil || rel.Info == nil { + continue + } + switch rel.Info.Status { + case release.StatusDeployed, release.StatusSuperseded: + if rel.Version > best { + best = rel.Version + } + } + } + return best +} diff --git a/kubernetes/gateway-operator/internal/helm/recovery_test.go b/kubernetes/gateway-operator/internal/helm/recovery_test.go new file mode 100644 index 0000000000..cf29000775 --- /dev/null +++ b/kubernetes/gateway-operator/internal/helm/recovery_test.go @@ -0,0 +1,258 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package helm + +import ( + "strings" + "testing" + + "helm.sh/helm/v3/pkg/release" +) + +// rel builds a history entry at the given revision and status. +func rel(revision int, status release.Status) *release.Release { + return &release.Release{ + Name: "test-gw", + Version: revision, + Info: &release.Info{Status: status}, + } +} + +func TestPlanRelease(t *testing.T) { + tests := []struct { + name string + history []*release.Release + wantOperation releaseOperation + wantRollbackRev int + wantErrSubstring string + }{ + { + // Nothing stored: there is nothing to preserve or upgrade from. + name: "empty history installs", + history: nil, + wantOperation: operationInstall, + }, + { + name: "deployed upgrades", + history: []*release.Release{rel(1, release.StatusDeployed)}, + wantOperation: operationUpgrade, + }, + { + // Existing behaviour: Helm can upgrade a failed release in place. + name: "failed upgrades", + history: []*release.Release{rel(1, release.StatusFailed)}, + wantOperation: operationUpgrade, + }, + { + name: "superseded upgrades", + history: []*release.Release{rel(1, release.StatusSuperseded)}, + wantOperation: operationUpgrade, + }, + { + // History kept after an uninstall leaves no revision to upgrade from. + name: "uninstalled installs", + history: []*release.Release{rel(1, release.StatusUninstalled)}, + wantOperation: operationInstall, + }, + { + // The interrupted first install: purging discards nothing successful. + name: "revision 1 pending install is purged and reinstalled", + history: []*release.Release{rel(1, release.StatusPendingInstall)}, + wantOperation: operationPurgeThenInstall, + }, + { + name: "pending upgrade rolls back to latest successful revision", + history: []*release.Release{ + rel(1, release.StatusSuperseded), + rel(2, release.StatusSuperseded), + rel(3, release.StatusPendingUpgrade), + }, + wantOperation: operationRollbackThenUpgrade, + wantRollbackRev: 2, + }, + { + name: "pending rollback rolls back to latest successful revision", + history: []*release.Release{ + rel(1, release.StatusSuperseded), + rel(2, release.StatusPendingRollback), + }, + wantOperation: operationRollbackThenUpgrade, + wantRollbackRev: 1, + }, + { + // Helm sorts history before returning it, but the planner must not depend on it. + name: "unordered history selects the highest revision", + history: []*release.Release{ + rel(3, release.StatusPendingUpgrade), + rel(1, release.StatusSuperseded), + rel(2, release.StatusDeployed), + }, + wantOperation: operationRollbackThenUpgrade, + wantRollbackRev: 2, + }, + { + // A failed revision is not a rollback target, so nothing safe remains. + name: "pending upgrade with no successful revision fails", + history: []*release.Release{ + rel(1, release.StatusFailed), + rel(2, release.StatusPendingUpgrade), + }, + wantErrSubstring: "no successful revision to recover to", + }, + { + // Never purge here: revision 2 means an earlier revision might have succeeded. + name: "pending install beyond revision 1 with no successful revision fails", + history: []*release.Release{ + rel(1, release.StatusFailed), + rel(2, release.StatusPendingInstall), + }, + wantErrSubstring: "no successful revision to recover to", + }, + { + name: "pending install beyond revision 1 rolls back when a successful revision exists", + history: []*release.Release{ + rel(1, release.StatusSuperseded), + rel(2, release.StatusPendingInstall), + }, + wantOperation: operationRollbackThenUpgrade, + wantRollbackRev: 1, + }, + { + // Recorded intent was removal, so neither upgrading nor purging is inferable. + name: "uninstalling fails rather than guessing", + history: []*release.Release{rel(1, release.StatusUninstalling)}, + wantErrSubstring: "stuck in uninstalling", + }, + { + // Defensive: a nil Info must not panic and must not be read as successful. + name: "missing info is treated as upgradeable", + history: []*release.Release{{Name: "test-gw", Version: 1}}, + wantOperation: operationUpgrade, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + plan, err := planRelease(tc.history) + + if tc.wantErrSubstring != "" { + if err == nil { + t.Fatalf("expected an error containing %q, got plan %v", tc.wantErrSubstring, plan.operation) + } + if !strings.Contains(err.Error(), tc.wantErrSubstring) { + t.Fatalf("expected error containing %q, got %q", tc.wantErrSubstring, err.Error()) + } + return + } + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if plan.operation != tc.wantOperation { + t.Errorf("operation = %v, want %v", plan.operation, tc.wantOperation) + } + if plan.rollbackRevision != tc.wantRollbackRev { + t.Errorf("rollbackRevision = %d, want %d", plan.rollbackRevision, tc.wantRollbackRev) + } + if plan.reason == "" { + t.Error("expected a non-empty reason so the decision is auditable in logs") + } + }) + } +} + +func TestPlanRelease_nilEntriesIgnored(t *testing.T) { + // A nil entry must not panic or mask the real latest revision. + plan, err := planRelease([]*release.Release{nil, rel(1, release.StatusDeployed), nil}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if plan.operation != operationUpgrade { + t.Errorf("operation = %v, want %v", plan.operation, operationUpgrade) + } +} + +func TestLatestRevision_numericNotPositional(t *testing.T) { + // Revision 10 must beat revision 9 ; a string comparison would not. + history := []*release.Release{rel(9, release.StatusSuperseded), rel(10, release.StatusDeployed)} + latest := latestRevision(history) + if latest == nil || latest.Version != 10 { + t.Fatalf("latestRevision = %v, want revision 10", latest) + } +} + +func TestLatestSuccessfulRevision(t *testing.T) { + tests := []struct { + name string + history []*release.Release + want int + }{ + {name: "none", history: []*release.Release{rel(1, release.StatusFailed)}, want: 0}, + {name: "deployed", history: []*release.Release{rel(4, release.StatusDeployed)}, want: 4}, + {name: "superseded counts", history: []*release.Release{rel(2, release.StatusSuperseded)}, want: 2}, + { + name: "highest successful wins over later failure", + history: []*release.Release{ + rel(1, release.StatusSuperseded), + rel(2, release.StatusDeployed), + rel(3, release.StatusFailed), + }, + want: 2, + }, + { + name: "pending revisions are not rollback targets", + history: []*release.Release{ + rel(1, release.StatusDeployed), + rel(2, release.StatusPendingUpgrade), + }, + want: 1, + }, + {name: "nil safe", history: []*release.Release{nil, {Name: "x", Version: 1}}, want: 0}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := latestSuccessfulRevision(tc.history); got != tc.want { + t.Errorf("latestSuccessfulRevision = %d, want %d", got, tc.want) + } + }) + } +} + +func TestReleaseOperation_String(t *testing.T) { + // The reason strings are logged before a destructive step, so the names must be stable. + cases := map[releaseOperation]string{ + operationInstall: "install", + operationUpgrade: "upgrade", + operationPurgeThenInstall: "purge-then-install", + operationRollbackThenUpgrade: "rollback-then-upgrade", + } + for op, want := range cases { + if got := op.String(); got != want { + t.Errorf("String() = %q, want %q", got, want) + } + } +} + +func TestMaxHistoryRevisions_isPositive(t *testing.T) { + // action.History treats Max == 0 as "return nothing", which would make every release + // look absent and turn an upgrade into a fresh install. + if maxHistoryRevisions <= 0 { + t.Fatalf("maxHistoryRevisions = %d, want a positive limit", maxHistoryRevisions) + } +} From 5724fe9c3bb946c1db704a78482819f9a2369bc4 Mon Sep 17 00:00:00 2001 From: Sanjula Herath Date: Wed, 5 Aug 2026 01:33:28 +0530 Subject: [PATCH 2/4] fix(gateway-operator): purge retained history before reinstalling An uninstalled release keeps its name in Helm's storage when history was retained, so Helm's availableName check rejects a plain install with "cannot re-use a name that is still in use" unless Install.Replace is set, which this package does not set. Plan that state as purge-then-install instead: the uninstall already removed the resources, so dropping the retained history discards nothing live, and an already-uninstalled release is purgeable because KeepHistory is false. Also corrects the upgrade comment: CleanupOnFail deletes resources the failed upgrade created, it does not roll the release back. Related to #3105 --- .../gateway-operator/internal/helm/client.go | 18 +++++++++++------- .../gateway-operator/internal/helm/recovery.go | 11 +++++++---- .../internal/helm/recovery_test.go | 8 +++++--- 3 files changed, 23 insertions(+), 14 deletions(-) diff --git a/kubernetes/gateway-operator/internal/helm/client.go b/kubernetes/gateway-operator/internal/helm/client.go index fc069cd9c3..dba9acaf20 100644 --- a/kubernetes/gateway-operator/internal/helm/client.go +++ b/kubernetes/gateway-operator/internal/helm/client.go @@ -190,9 +190,9 @@ func (c *Client) InstallOrUpgrade(ctx context.Context, opts InstallOrUpgradeOpti return c.upgrade(ctx, actionConfig, opts) case operationPurgeThenInstall: - // Drop the stuck release and its history so the install below is not itself - // rejected as an in-progress operation. Safe only because planRelease proved there - // is no successful revision to preserve. + // Drop the release and its history so the install below is not rejected, either as + // an in-progress operation or because the retained name is still in use. planRelease + // only plans this where nothing live is discarded. if err := c.purge(ctx, actionConfig, opts); err != nil { return err } @@ -212,8 +212,11 @@ func (c *Client) InstallOrUpgrade(ctx context.Context, opts InstallOrUpgradeOpti } } -// purge removes a release together with its history so that a subsequent install is not -// rejected because an operation is still recorded as in progress. +// purge removes a release together with its history so a subsequent install is not rejected +// — either because an operation is still recorded as in progress, or because retained +// history from an earlier uninstall still holds the release name. KeepHistory is false, which +// is also what lets an already-uninstalled release be purged rather than reported as +// "already deleted". func (c *Client) purge(ctx context.Context, actionConfig *action.Configuration, opts InstallOrUpgradeOptions) error { log := log.FromContext(ctx) @@ -334,8 +337,9 @@ func (c *Client) upgrade(ctx context.Context, actionConfig *action.Configuration client.Namespace = opts.Namespace client.Wait = opts.Wait client.Version = opts.Version - // Roll partially-applied resources back when an upgrade fails, so a failed attempt - // leaves a usable revision rather than a half-applied one for the next reconcile. + // Delete resources the failed upgrade created, so a partway attempt does not leave + // orphaned objects behind. This does not roll the release back: it still records a + // failed revision, which planRelease upgrades from on the next reconcile. client.CleanupOnFail = true if opts.Timeout > 0 { diff --git a/kubernetes/gateway-operator/internal/helm/recovery.go b/kubernetes/gateway-operator/internal/helm/recovery.go index fc66bafcf9..d62622971d 100644 --- a/kubernetes/gateway-operator/internal/helm/recovery.go +++ b/kubernetes/gateway-operator/internal/helm/recovery.go @@ -108,11 +108,14 @@ func planRelease(history []*release.Release) (recoveryPlan, error) { }, nil case release.StatusUninstalled: - // History was kept after an uninstall (--keep-history). There is no deployed - // revision to upgrade from, so install again. + // History was kept after an uninstall (--keep-history). A plain install would be + // rejected by Helm's availableName check ("cannot re-use a name that is still in + // use") because the retained history still holds the name, and this package does not + // set Install.Replace. Purging first drops that history; the resources were already + // removed by the uninstall, so nothing live is discarded. return recoveryPlan{ - operation: operationInstall, - reason: fmt.Sprintf("latest revision %d is uninstalled", latest.Version), + operation: operationPurgeThenInstall, + reason: fmt.Sprintf("latest revision %d is uninstalled with retained history", latest.Version), }, nil case release.StatusUninstalling: diff --git a/kubernetes/gateway-operator/internal/helm/recovery_test.go b/kubernetes/gateway-operator/internal/helm/recovery_test.go index cf29000775..c8d4bfe7b0 100644 --- a/kubernetes/gateway-operator/internal/helm/recovery_test.go +++ b/kubernetes/gateway-operator/internal/helm/recovery_test.go @@ -64,10 +64,12 @@ func TestPlanRelease(t *testing.T) { wantOperation: operationUpgrade, }, { - // History kept after an uninstall leaves no revision to upgrade from. - name: "uninstalled installs", + // Retained history still holds the name, so Helm would reject a plain install + // with "cannot re-use a name that is still in use". The uninstall already + // removed the resources, so purging the history discards nothing live. + name: "uninstalled purges retained history then installs", history: []*release.Release{rel(1, release.StatusUninstalled)}, - wantOperation: operationInstall, + wantOperation: operationPurgeThenInstall, }, { // The interrupted first install: purging discards nothing successful. From c6c86bfc692f81a90e820cb7628878d22156cbbf Mon Sep 17 00:00:00 2001 From: Sanjula Herath Date: Wed, 5 Aug 2026 09:45:03 +0530 Subject: [PATCH 3/4] fix(gateway-operator): harden failed deployment recovery Remove the History.Max handling. In helm.sh/helm/v3@v3.21.3 action.History.Run returns cfg.Releases.History(name) directly and never reads Max, so setting it was a no-op and the comments describing it as a truncation control were wrong. The real improvements stay: the returned history is now actually used, revisions are compared numerically, and a history error that is not "release not found" is propagated instead of being read as an absent release. Add a release executor seam so InstallOrUpgrade's dispatch is testable: the planner alone cannot catch a correct plan that is dispatched wrongly, which is how the uninstalled-release bug reached review. Purge and rollback action construction is split out so KeepHistory=false and the planned rollback revision are assertable without a cluster. Drop the tracking entry when a status write fails. A Processing or Deployed entry tells the next reconcile that another operation owns the generation, but the reconcile that owned it has already returned; the next one then skipped and returned nil, ending the requeue chain the returned error had started. A failed initial-condition patch and a failed Programmed=True patch both wedged the gateway this way. Rename the spec.infrastructure annotation test to say what it covers, and add the metadata-annotation case from the issue. Changing metadata.annotations is not a deployment input and is deliberately not an immediate retry trigger; recovery there is the bounded retry window. Related to #3105 --- .../controller/apigateway_controller.go | 30 +- .../controller/apigateway_recovery_test.go | 101 +++++- .../controller/apigateway_statuspatch_test.go | 229 +++++++++++++ .../gateway-operator/internal/helm/client.go | 107 ++++-- .../internal/helm/dispatch_test.go | 314 ++++++++++++++++++ .../internal/helm/recovery.go | 10 +- .../internal/helm/recovery_test.go | 8 - 7 files changed, 758 insertions(+), 41 deletions(-) create mode 100644 kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go create mode 100644 kubernetes/gateway-operator/internal/helm/dispatch_test.go diff --git a/kubernetes/gateway-operator/internal/controller/apigateway_controller.go b/kubernetes/gateway-operator/internal/controller/apigateway_controller.go index 3dddbd516f..52d5f13a14 100644 --- a/kubernetes/gateway-operator/internal/controller/apigateway_controller.go +++ b/kubernetes/gateway-operator/internal/controller/apigateway_controller.go @@ -118,7 +118,9 @@ func (t *GatewayTracker) Set(key string, entry *GatewayTrackingEntry) { t.entries[key] = entry } -// Delete removes a tracking entry (only called when APIGateway CR is deleted) +// Delete removes a tracking entry. Called when the APIGateway CR is deleted, and when a +// status write fails so the next reconcile re-derives state from the CR instead of trusting +// a tracker state whose owning reconcile has already returned. func (t *GatewayTracker) Delete(key string) { t.mu.Lock() defer t.mu.Unlock() @@ -151,6 +153,21 @@ func (r *GatewayReconciler) deploy(ctx context.Context, gatewayConfig *apiv1.API return r.processGatewayDeployment(ctx, gatewayConfig, trackingKey, generation, configHash, inputHash) } +// forgetTracking drops the in-memory entry so the next reconcile re-derives state from the CR. +// +// It is used when a status write fails. No tracker state may mean "another operation will +// complete this" once the reconcile that owned that operation has returned: states like +// Processing and Deployed cause the next reconcile to skip and return nil, which ends the +// requeue chain that the returned error would otherwise have driven. +func (r *GatewayReconciler) forgetTracking(trackingKey, reason string) { + r.gatewayTracker.Delete(trackingKey) + if r.Logger != nil { + r.Logger.Info("Cleared APIGateway tracking so the next reconcile can retry", + slog.String("key", trackingKey), + slog.String("reason", reason)) + } +} + // retryCountFor returns the retry count a new attempt starts from. The budget carries over // only while the same generation is retried against unchanged inputs, so a corrected input // does not inherit an already-exhausted count. @@ -502,6 +519,11 @@ func (r *GatewayReconciler) processGatewayDeployment( // Set initial conditions (Accepted=True, Programmed=False/Pending) if retryCount == 0 { if err := r.setGatewayInitialConditions(ctx, gatewayConfig, generation); err != nil { + // The tracker still says Processing, which tells the next reconcile that another + // operation owns this generation — but this reconcile is returning, so nothing + // will. That next reconcile would skip and return nil, ending the requeue chain + // the returned error starts. Forget the entry so it re-derives from the CR. + r.forgetTracking(trackingKey, "initial status patch failed") return ctrl.Result{}, err } } @@ -630,6 +652,12 @@ func (r *GatewayReconciler) handleGatewayDeploymentSuccess( Message: readinessMsg, LastTransitionTime: metav1.Now(), }, &selectedCount, configHash); err != nil { + // Programmed=True never landed, so the CR still reports failure while the tracker + // says Deployed. Every reconcile would then skip on "status not yet propagated" and + // the gateway would stay false forever. Forget the entry so the next reconcile + // re-drives the deployment and rewrites the status; the Helm operation is safe to + // repeat because it upgrades an already-deployed release. + r.forgetTracking(trackingKey, "success status patch failed") return ctrl.Result{}, err } diff --git a/kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go b/kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go index ae361570d7..a779c7ebff 100644 --- a/kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go +++ b/kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go @@ -366,7 +366,12 @@ func TestDecideAndProcess_configMapChangeRecoversImmediately(t *testing.T) { } } -func TestDecideAndProcess_annotationChangeRecoversImmediately(t *testing.T) { +// TestDecideAndProcess_specInfrastructureAnnotationChangeRecoversImmediately covers a change +// to spec.infrastructure.annotations, which is a deployment input: it becomes commonAnnotations +// in the rendered values. Note this is a *spec* change, so on a real cluster it also bumps +// metadata.generation; it is not the `kubectl annotate` route from the issue, which is covered +// by TestDecideAndProcess_metadataAnnotationOnlyChangeWaitsForRetryWindow below. +func TestDecideAndProcess_specInfrastructureAnnotationChangeRecoversImmediately(t *testing.T) { gw := failedGateway(1) r, calls := newRecoveryReconciler(t, gw) @@ -392,7 +397,99 @@ func TestDecideAndProcess_annotationChangeRecoversImmediately(t *testing.T) { t.Fatalf("unexpected error: %v", err) } if *calls != 1 { - t.Errorf("deploy attempts = %d, want 1 immediately after an annotation change", *calls) + t.Errorf("deploy attempts = %d, want 1 immediately after a spec.infrastructure annotation change", *calls) + } +} + +// TestDecideAndProcess_metadataAnnotationOnlyChangeWaitsForRetryWindow pins down the exact +// route the issue tried: `kubectl annotate apigateway ...`, which changes ObjectMeta.Annotations +// without bumping metadata.generation. +// +// This is NOT an immediate retry trigger, and that is deliberate. Metadata annotations are not +// deployment inputs — only spec.infrastructure labels/annotations reach the rendered values via +// commonAnnotations — so fingerprinting them would mean retrying on changes that cannot alter +// the outcome. It would also be a hot-loop risk: this codebase already writes operator-managed +// annotations onto resources it reconciles (see httproute_controller.go), so any such annotation +// added to APIGateway later would retrigger deployments on its own writes. Making +// re-annotation a supported reset would mean introducing a documented reset annotation, which +// is a public API decision for the maintainers. +// +// The gateway is still not wedged: it recovers on the bounded retry window, on an operator +// restart, and on a ConfigMap correction. This test asserts the honest behaviour so the +// limitation cannot be mistaken for a fix. +func TestDecideAndProcess_metadataAnnotationOnlyChangeWaitsForRetryWindow(t *testing.T) { + gw := failedGateway(1) + r, calls := newRecoveryReconciler(t, gw) + + _, baseHash, err := r.deploymentInputs(context.Background(), gw) + if err != nil { + t.Fatalf("deploymentInputs: %v", err) + } + r.gatewayTracker.Set(testTrackKey, &GatewayTrackingEntry{ + Generation: 1, + Status: GatewayTrackingStatusFailed, + RetryCount: 3, + InputHash: baseHash, + NextRetryTime: time.Now().Add(testSyncPeriod), + }) + + // `kubectl annotate` — metadata only, generation unchanged. + generationBefore := gw.Generation + gw.ObjectMeta.Annotations = map[string]string{"ops.example.com/nudge": "1"} + if gw.Generation != generationBefore { + t.Fatalf("test setup changed the generation (%d -> %d); it must stay equal to observedGeneration", + generationBefore, gw.Generation) + } + + res, err := callDecide(t, r, gw) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if *calls != 0 { + t.Errorf("deploy attempts = %d, want 0: a metadata annotation is not a deployment input", *calls) + } + // Crucially it is not wedged either — a bounded retry is still scheduled. + if res.RequeueAfter <= 0 { + t.Errorf("RequeueAfter = %v, want a bounded retry so the gateway still recovers", res.RequeueAfter) + } + + // And the fingerprint must be unchanged by metadata annotations, so the retry budget is + // not silently reset by an unrelated edit. + _, afterHash, err := r.deploymentInputs(context.Background(), gw) + if err != nil { + t.Fatalf("deploymentInputs: %v", err) + } + if afterHash != baseHash { + t.Error("metadata annotations must not affect the deployment input fingerprint") + } +} + +func TestDeploymentInputs_infrastructureAnnotationOrderIsStable(t *testing.T) { + // The overlay is marshalled YAML with sorted keys, so a differently-ordered literal must + // produce the same fingerprint; otherwise map iteration order would reset retry budgets. + gwA := failedGateway(1) + gwA.Spec.Infrastructure = &apiv1.GatewayInfrastructure{ + Annotations: map[string]string{"a": "1", "b": "2", "c": "3"}, + } + gwB := failedGateway(1) + gwB.Spec.Infrastructure = &apiv1.GatewayInfrastructure{ + Annotations: map[string]string{"c": "3", "b": "2", "a": "1"}, + } + + r, _ := newRecoveryReconciler(t, gwA) + _, hashA, err := r.deploymentInputs(context.Background(), gwA) + if err != nil { + t.Fatalf("deploymentInputs A: %v", err) + } + // Repeat many times: a single comparison could pass by luck with random map ordering. + for i := 0; i < 50; i++ { + _, hashB, err := r.deploymentInputs(context.Background(), gwB) + if err != nil { + t.Fatalf("deploymentInputs B: %v", err) + } + if hashA != hashB { + t.Fatalf("iteration %d: fingerprint depends on map ordering (%s vs %s)", i, hashA, hashB) + } } } diff --git a/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go b/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go new file mode 100644 index 0000000000..a8995ccfed --- /dev/null +++ b/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go @@ -0,0 +1,229 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package controller + +import ( + "context" + "errors" + "io" + "log/slog" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + apiv1 "github.com/wso2/api-platform/kubernetes/gateway-operator/api/v1" +) + +// failingStatusWriter fails the first n status patches, then delegates. It reproduces a +// transient status-subresource write failure without touching production code. +type failingStatusWriter struct { + client.SubResourceWriter + failures *int +} + +func (w *failingStatusWriter) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if *w.failures > 0 { + *w.failures-- + return errors.New("simulated status patch failure") + } + return w.SubResourceWriter.Patch(ctx, obj, patch, opts...) +} + +func (w *failingStatusWriter) Update(ctx context.Context, obj client.Object, opts ...client.SubResourceUpdateOption) error { + if *w.failures > 0 { + *w.failures-- + return errors.New("simulated status update failure") + } + return w.SubResourceWriter.Update(ctx, obj, opts...) +} + +type failingStatusClient struct { + client.Client + failures *int +} + +func (c *failingStatusClient) Status() client.SubResourceWriter { + return &failingStatusWriter{SubResourceWriter: c.Client.Status(), failures: c.failures} +} + +// newStatusFailureReconciler builds a reconciler whose first statusFailures status writes fail. +// deployGateway is left as the real processGatewayDeployment so the actual state transitions +// run; the Helm step is replaced by deployResult. +func newStatusFailureReconciler(t *testing.T, gw *apiv1.APIGateway, statusFailures int) (*GatewayReconciler, *int) { + t.Helper() + scheme := recoveryScheme(t) + base := fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(&apiv1.APIGateway{}). + WithRuntimeObjects(gw).Build() + remaining := statusFailures + r := &GatewayReconciler{ + Client: &failingStatusClient{Client: base, failures: &remaining}, + Scheme: scheme, + Config: recoveryConfig(), + gatewayTracker: NewGatewayTracker(), + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + } + return r, &remaining +} + +// TestProcessGatewayDeployment_initialStatusPatchFailureStaysRecoverable covers the first +// wedge: processGatewayDeployment records Processing, then the initial-condition patch fails +// and the reconcile returns an error. Because controller-runtime requeues that error, the +// next reconcile must be able to retry — if the tracker still said Processing it would skip +// and return nil, ending the requeue chain and stranding the gateway. +func TestProcessGatewayDeployment_initialStatusPatchFailureStaysRecoverable(t *testing.T) { + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{ + Name: testGateway, Namespace: testNamespace, Generation: 1, + Finalizers: []string{apigatewayFinalizerName}, + }, + } + r, _ := newStatusFailureReconciler(t, gw, 1) + + // Attempt 1: the real code path, with the initial status patch failing. + _, err := r.processGatewayDeployment(context.Background(), gw, testTrackKey, 1, "", "hash-1") + if err == nil { + t.Fatal("expected the initial status patch failure to be returned") + } + + if entry, ok := r.gatewayTracker.Get(testTrackKey); ok { + t.Fatalf("tracker still holds %q after a failed initial status patch; the next reconcile would skip", entry.Status) + } + + // Attempt 2 is the requeue. Drive the real decision path and prove it deploys rather + // than skipping. + deployCalls := 0 + r.deployGateway = func(_ context.Context, _ *apiv1.APIGateway, _ string, _ int64, _, _ string) (ctrl.Result, error) { + deployCalls++ + return ctrl.Result{}, nil + } + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("second reconcile returned an error: %v", err) + } + if deployCalls != 1 { + t.Errorf("deploy attempts on the requeued reconcile = %d, want 1", deployCalls) + } +} + +// TestHandleGatewayDeploymentSuccess_statusPatchFailureStaysRecoverable covers the second +// wedge: the deployment succeeded, the tracker moved to Deployed, then Programmed=True failed +// to persist. The CR still reports failure, and a Deployed tracker made every later reconcile +// skip on "status not yet propagated", so the gateway stayed false forever. +func TestHandleGatewayDeploymentSuccess_statusPatchFailureStaysRecoverable(t *testing.T) { + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{ + Name: testGateway, Namespace: testNamespace, Generation: 1, + Finalizers: []string{apigatewayFinalizerName}, + }, + } + r, _ := newStatusFailureReconciler(t, gw, 1) + + entry := &GatewayTrackingEntry{Generation: 1, Status: GatewayTrackingStatusProcessing, InputHash: "hash-1"} + r.gatewayTracker.Set(testTrackKey, entry) + + _, err := r.handleGatewayDeploymentSuccess(context.Background(), gw, testTrackKey, entry, 0, "ready", "") + if err == nil { + t.Fatal("expected the success status patch failure to be returned") + } + + if stored, ok := r.gatewayTracker.Get(testTrackKey); ok { + t.Fatalf("tracker still holds %q after a failed success status patch; every later reconcile would skip", stored.Status) + } + + // Confirm the persisted condition really is still not programmed, so the wedge would + // have been permanent rather than cosmetic. + stored := &apiv1.APIGateway{} + if err := r.Get(context.Background(), typesName(), stored); err != nil { + t.Fatalf("get gateway: %v", err) + } + if cond := findProgrammed(stored); cond != nil && cond.Status == metav1.ConditionTrue { + t.Fatal("expected Programmed to remain not-True after the failed patch") + } + + // The requeued reconcile must re-drive the deployment. + deployCalls := 0 + r.deployGateway = func(_ context.Context, _ *apiv1.APIGateway, _ string, _ int64, _, _ string) (ctrl.Result, error) { + deployCalls++ + return ctrl.Result{}, nil + } + if _, err := callDecide(t, r, stored); err != nil { + t.Fatalf("second reconcile returned an error: %v", err) + } + if deployCalls != 1 { + t.Errorf("deploy attempts on the requeued reconcile = %d, want 1", deployCalls) + } +} + +// TestHandleGatewayDeploymentError_exhaustionStatusPatchFailureStaysRecoverable drives the real +// exhaustion path with the status patch failing, rather than hand-building a Failed entry. +func TestHandleGatewayDeploymentError_exhaustionStatusPatchFailureStaysRecoverable(t *testing.T) { + gw := &apiv1.APIGateway{ + ObjectMeta: metav1.ObjectMeta{ + Name: testGateway, Namespace: testNamespace, Generation: 1, + Finalizers: []string{apigatewayFinalizerName}, + }, + } + r, _ := newStatusFailureReconciler(t, gw, 1) + + entry := &GatewayTrackingEntry{ + Generation: 1, Status: GatewayTrackingStatusProcessing, RetryCount: 2, InputHash: "hash-1", + } + r.gatewayTracker.Set(testTrackKey, entry) + + _, err := r.handleGatewayDeploymentError(context.Background(), gw, testTrackKey, entry, errors.New("boom"), 0) + if err == nil { + t.Fatal("expected the failure status patch error to be returned") + } + + // Unlike Processing/Deployed, a Failed entry carries its own bounded retry, so keeping it + // is what makes the gateway recoverable even though the failure was never persisted. + stored, ok := r.gatewayTracker.Get(testTrackKey) + if !ok { + t.Fatal("expected the Failed entry to be retained so the bounded retry survives") + } + if stored.Status != GatewayTrackingStatusFailed { + t.Errorf("status = %q, want %q", stored.Status, GatewayTrackingStatusFailed) + } + if stored.NextRetryTime.IsZero() { + t.Error("expected a bounded next-retry time despite the failed status write") + } + + // Corrected inputs must still recover immediately, without waiting for the window. + deployCalls := 0 + r.deployGateway = func(_ context.Context, _ *apiv1.APIGateway, _ string, _ int64, _, _ string) (ctrl.Result, error) { + deployCalls++ + return ctrl.Result{}, nil + } + deployNow, _, _ := r.decideFailedRecovery(stored, "hash-2", stored.NextRetryTime.Add(-time.Hour)) + if !deployNow { + t.Error("changed inputs must recover immediately even after a failed status write") + } + _ = deployCalls +} + +func findProgrammed(gw *apiv1.APIGateway) *metav1.Condition { + for i := range gw.Status.Conditions { + if gw.Status.Conditions[i].Type == apiv1.GatewayConditionProgrammed { + return &gw.Status.Conditions[i] + } + } + return nil +} diff --git a/kubernetes/gateway-operator/internal/helm/client.go b/kubernetes/gateway-operator/internal/helm/client.go index dba9acaf20..ad08fbb044 100644 --- a/kubernetes/gateway-operator/internal/helm/client.go +++ b/kubernetes/gateway-operator/internal/helm/client.go @@ -158,16 +158,61 @@ func (c *Client) InstallOrUpgrade(ctx context.Context, opts InstallOrUpgradeOpti // Set registry client in action configuration actionConfig.RegistryClient = c.registryClient - // Read the full history so a pending release can be recovered. Only a genuine - // "not found" means the release is absent; any other failure must not be mistaken for - // one, or a transient storage error would trigger a fresh install over a live release. histClient := action.NewHistory(actionConfig) - histClient.Max = maxHistoryRevisions - history, err := histClient.Run(opts.ReleaseName) + return planAndExecute(ctx, histClient.Run, &clientExecutor{client: c, actionConfig: actionConfig}, opts) +} + +// releaseExecutor performs the individual Helm operations a plan is made of. It exists so the +// dispatch order in planAndExecute can be tested without a cluster; the planner alone cannot +// catch a plan that is valid but dispatched wrongly. +type releaseExecutor interface { + install(ctx context.Context, opts InstallOrUpgradeOptions) error + upgrade(ctx context.Context, opts InstallOrUpgradeOptions) error + purge(ctx context.Context, opts InstallOrUpgradeOptions) error + rollback(ctx context.Context, opts InstallOrUpgradeOptions, revision int) error +} + +// clientExecutor runs the operations against a real Helm action configuration. +type clientExecutor struct { + client *Client + actionConfig *action.Configuration +} + +func (e *clientExecutor) install(ctx context.Context, opts InstallOrUpgradeOptions) error { + return e.client.install(ctx, e.actionConfig, opts) +} + +func (e *clientExecutor) upgrade(ctx context.Context, opts InstallOrUpgradeOptions) error { + return e.client.upgrade(ctx, e.actionConfig, opts) +} + +func (e *clientExecutor) purge(ctx context.Context, opts InstallOrUpgradeOptions) error { + return e.client.purge(ctx, e.actionConfig, opts) +} + +func (e *clientExecutor) rollback(ctx context.Context, opts InstallOrUpgradeOptions, revision int) error { + return e.client.rollback(ctx, e.actionConfig, opts, revision) +} + +// planAndExecute reads the release history, plans the recovery, and dispatches it. +// +// historyFn returns every stored revision for the release; the planner compares revisions +// numerically rather than relying on their order. Only a genuine "not found" means the release +// is absent — any other error is propagated, because treating a transient storage or +// unreachable-cluster failure as "absent" would start a fresh install over a live release. +func planAndExecute( + ctx context.Context, + historyFn func(string) ([]*release.Release, error), + exec releaseExecutor, + opts InstallOrUpgradeOptions, +) error { + log := log.FromContext(ctx) + + history, err := historyFn(opts.ReleaseName) switch { case errors.Is(err, driver.ErrReleaseNotFound): log.Info("Release does not exist, performing install", "release", opts.ReleaseName) - return c.install(ctx, actionConfig, opts) + return exec.install(ctx, opts) case err != nil: return fmt.Errorf("failed to read history for release %q: %w", opts.ReleaseName, err) } @@ -184,28 +229,28 @@ func (c *Client) InstallOrUpgrade(ctx context.Context, opts InstallOrUpgradeOpti switch plan.operation { case operationInstall: - return c.install(ctx, actionConfig, opts) + return exec.install(ctx, opts) case operationUpgrade: - return c.upgrade(ctx, actionConfig, opts) + return exec.upgrade(ctx, opts) case operationPurgeThenInstall: // Drop the release and its history so the install below is not rejected, either as - // an in-progress operation or because the retained name is still in use. planRelease - // only plans this where nothing live is discarded. - if err := c.purge(ctx, actionConfig, opts); err != nil { + // an in-progress operation or because a retained name is still in use. planRelease + // only plans this where no live resources are discarded. + if err := exec.purge(ctx, opts); err != nil { return err } - return c.install(ctx, actionConfig, opts) + return exec.install(ctx, opts) case operationRollbackThenUpgrade: // Roll back first so the release leaves the pending state; the upgrade then starts // from a known-good revision. If the rollback succeeds and the process dies before // the upgrade, the next reconcile sees a deployed release and simply upgrades. - if err := c.rollback(ctx, actionConfig, opts, plan.rollbackRevision); err != nil { + if err := exec.rollback(ctx, opts, plan.rollbackRevision); err != nil { return err } - return c.upgrade(ctx, actionConfig, opts) + return exec.upgrade(ctx, opts) default: return fmt.Errorf("unsupported Helm operation %q for release %q", plan.operation, opts.ReleaseName) @@ -220,12 +265,7 @@ func (c *Client) InstallOrUpgrade(ctx context.Context, opts InstallOrUpgradeOpti func (c *Client) purge(ctx context.Context, actionConfig *action.Configuration, opts InstallOrUpgradeOptions) error { log := log.FromContext(ctx) - client := action.NewUninstall(actionConfig) - client.KeepHistory = false - client.Wait = opts.Wait - if opts.Timeout > 0 { - client.Timeout = time.Duration(opts.Timeout) * time.Second - } + client := newPurgeAction(actionConfig, opts) log.Info("Purging unrecoverable release before reinstall", "release", opts.ReleaseName) if _, err := client.Run(opts.ReleaseName); err != nil { @@ -239,10 +279,23 @@ func (c *Client) purge(ctx context.Context, actionConfig *action.Configuration, return nil } -// rollback returns a release stuck in a pending state to the given known-good revision. -func (c *Client) rollback(ctx context.Context, actionConfig *action.Configuration, opts InstallOrUpgradeOptions, revision int) error { - log := log.FromContext(ctx) +// newPurgeAction configures the uninstall used to drop a release before reinstalling it. +// KeepHistory must stay false: it is what removes the retained history holding the release +// name, and what lets an already-uninstalled release be purged instead of being reported as +// "already deleted". Split out so those settings are assertable without a cluster. +func newPurgeAction(actionConfig *action.Configuration, opts InstallOrUpgradeOptions) *action.Uninstall { + client := action.NewUninstall(actionConfig) + client.KeepHistory = false + client.Wait = opts.Wait + if opts.Timeout > 0 { + client.Timeout = time.Duration(opts.Timeout) * time.Second + } + return client +} +// newRollbackAction configures the rollback used to leave a pending state, targeting the +// revision the planner selected. Split out so the target revision is assertable. +func newRollbackAction(actionConfig *action.Configuration, opts InstallOrUpgradeOptions, revision int) *action.Rollback { client := action.NewRollback(actionConfig) client.Version = revision client.Wait = opts.Wait @@ -250,6 +303,14 @@ func (c *Client) rollback(ctx context.Context, actionConfig *action.Configuratio if opts.Timeout > 0 { client.Timeout = time.Duration(opts.Timeout) * time.Second } + return client +} + +// rollback returns a release stuck in a pending state to the given known-good revision. +func (c *Client) rollback(ctx context.Context, actionConfig *action.Configuration, opts InstallOrUpgradeOptions, revision int) error { + log := log.FromContext(ctx) + + client := newRollbackAction(actionConfig, opts, revision) log.Info("Rolling back release to recover from pending state", "release", opts.ReleaseName, diff --git a/kubernetes/gateway-operator/internal/helm/dispatch_test.go b/kubernetes/gateway-operator/internal/helm/dispatch_test.go new file mode 100644 index 0000000000..504dc8162c --- /dev/null +++ b/kubernetes/gateway-operator/internal/helm/dispatch_test.go @@ -0,0 +1,314 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package helm + +import ( + "context" + "errors" + "strings" + "testing" + + "helm.sh/helm/v3/pkg/action" + "helm.sh/helm/v3/pkg/release" + "helm.sh/helm/v3/pkg/storage/driver" +) + +// recordingExecutor records the operations planAndExecute dispatches, in order, so the +// sequence and arguments can be asserted. Any operation can be made to fail. +type recordingExecutor struct { + calls []string + rollbackRevisions []int + contexts []context.Context + + installErr error + upgradeErr error + purgeErr error + rollbackErr error +} + +func (e *recordingExecutor) install(ctx context.Context, _ InstallOrUpgradeOptions) error { + e.calls = append(e.calls, "install") + e.contexts = append(e.contexts, ctx) + return e.installErr +} + +func (e *recordingExecutor) upgrade(ctx context.Context, _ InstallOrUpgradeOptions) error { + e.calls = append(e.calls, "upgrade") + e.contexts = append(e.contexts, ctx) + return e.upgradeErr +} + +func (e *recordingExecutor) purge(ctx context.Context, _ InstallOrUpgradeOptions) error { + e.calls = append(e.calls, "purge") + e.contexts = append(e.contexts, ctx) + return e.purgeErr +} + +func (e *recordingExecutor) rollback(ctx context.Context, _ InstallOrUpgradeOptions, revision int) error { + e.calls = append(e.calls, "rollback") + e.rollbackRevisions = append(e.rollbackRevisions, revision) + e.contexts = append(e.contexts, ctx) + return e.rollbackErr +} + +func historyReturning(history []*release.Release, err error) func(string) ([]*release.Release, error) { + return func(string) ([]*release.Release, error) { return history, err } +} + +func joined(calls []string) string { return strings.Join(calls, ",") } + +var dispatchOpts = InstallOrUpgradeOptions{ReleaseName: "test-gw", Namespace: "gw-ns"} + +func TestPlanAndExecute_dispatchOrder(t *testing.T) { + tests := []struct { + name string + history []*release.Release + historyErr error + wantCalls string + wantRollbacks []int + wantErrSubstr string + }{ + { + // Absent release: install, and never touch a destructive operation. + name: "missing history installs", + historyErr: driver.ErrReleaseNotFound, + wantCalls: "install", + }, + { + name: "deployed upgrades", + history: []*release.Release{rel(1, release.StatusDeployed)}, + wantCalls: "upgrade", + }, + { + // The interrupted first install: purge must precede install, in that order. + name: "revision 1 pending install purges then installs", + history: []*release.Release{rel(1, release.StatusPendingInstall)}, + wantCalls: "purge,install", + }, + { + // The bug the planner suite missed: a plain install here is rejected by Helm. + name: "uninstalled retained history purges then installs", + history: []*release.Release{rel(1, release.StatusUninstalled)}, + wantCalls: "purge,install", + }, + { + name: "pending upgrade rolls back to the planned revision then upgrades", + history: []*release.Release{ + rel(1, release.StatusSuperseded), + rel(2, release.StatusDeployed), + rel(3, release.StatusPendingUpgrade), + }, + wantCalls: "rollback,upgrade", + wantRollbacks: []int{2}, + }, + { + name: "pending rollback rolls back then upgrades", + history: []*release.Release{ + rel(1, release.StatusSuperseded), + rel(2, release.StatusPendingRollback), + }, + wantCalls: "rollback,upgrade", + wantRollbacks: []int{1}, + }, + { + // No safe revision: nothing destructive may run. + name: "unsafe pending state dispatches nothing", + history: []*release.Release{ + rel(1, release.StatusFailed), + rel(2, release.StatusPendingUpgrade), + }, + wantCalls: "", + wantErrSubstr: "no successful revision to recover to", + }, + { + name: "uninstalling dispatches nothing", + history: []*release.Release{rel(1, release.StatusUninstalling)}, + wantCalls: "", + wantErrSubstr: "stuck in uninstalling", + }, + { + // A transient storage or unreachable-cluster error must never look like "absent", + // or a fresh install would run over a live release. + name: "non not-found history error dispatches nothing", + historyErr: errors.New("etcdserver: request timed out"), + wantCalls: "", + wantErrSubstr: "failed to read history", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + exec := &recordingExecutor{} + err := planAndExecute(context.Background(), historyReturning(tc.history, tc.historyErr), exec, dispatchOpts) + + if tc.wantErrSubstr != "" { + if err == nil { + t.Fatalf("expected an error containing %q, got nil (calls: %q)", tc.wantErrSubstr, joined(exec.calls)) + } + if !strings.Contains(err.Error(), tc.wantErrSubstr) { + t.Fatalf("error = %q, want it to contain %q", err.Error(), tc.wantErrSubstr) + } + } else if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if got := joined(exec.calls); got != tc.wantCalls { + t.Errorf("dispatched %q, want %q", got, tc.wantCalls) + } + if tc.wantRollbacks != nil { + if len(exec.rollbackRevisions) != len(tc.wantRollbacks) { + t.Fatalf("rollback revisions = %v, want %v", exec.rollbackRevisions, tc.wantRollbacks) + } + for i, want := range tc.wantRollbacks { + if exec.rollbackRevisions[i] != want { + t.Errorf("rollback revision[%d] = %d, want %d", i, exec.rollbackRevisions[i], want) + } + } + } + }) + } +} + +func TestPlanAndExecute_noInstallWhenPurgeFails(t *testing.T) { + // Installing after a failed purge would hit the very rejection the purge exists to avoid. + exec := &recordingExecutor{purgeErr: errors.New("purge boom")} + err := planAndExecute(context.Background(), + historyReturning([]*release.Release{rel(1, release.StatusPendingInstall)}, nil), exec, dispatchOpts) + + if err == nil { + t.Fatal("expected the purge error to be returned") + } + if got := joined(exec.calls); got != "purge" { + t.Errorf("dispatched %q, want only %q", got, "purge") + } +} + +func TestPlanAndExecute_noUpgradeWhenRollbackFails(t *testing.T) { + // The release is still pending, so upgrading would be rejected by Helm. + exec := &recordingExecutor{rollbackErr: errors.New("rollback boom")} + err := planAndExecute(context.Background(), + historyReturning([]*release.Release{ + rel(1, release.StatusDeployed), + rel(2, release.StatusPendingUpgrade), + }, nil), exec, dispatchOpts) + + if err == nil { + t.Fatal("expected the rollback error to be returned") + } + if got := joined(exec.calls); got != "rollback" { + t.Errorf("dispatched %q, want only %q", got, "rollback") + } +} + +func TestPlanAndExecute_operationErrorsPropagate(t *testing.T) { + installBoom := errors.New("install boom") + exec := &recordingExecutor{installErr: installBoom} + err := planAndExecute(context.Background(), historyReturning(nil, driver.ErrReleaseNotFound), exec, dispatchOpts) + if !errors.Is(err, installBoom) { + t.Errorf("error = %v, want it to wrap %v", err, installBoom) + } +} + +func TestPlanAndExecute_threadsCallerContext(t *testing.T) { + // The caller's context must reach the operations so a cancelled reconcile or an operator + // shutdown can abort the Helm wait rather than blocking for the whole timeout. + type ctxKey string + ctx := context.WithValue(context.Background(), ctxKey("marker"), "present") + + exec := &recordingExecutor{} + if err := planAndExecute(ctx, + historyReturning([]*release.Release{rel(1, release.StatusPendingUpgrade), rel(2, release.StatusDeployed)}, nil), + exec, dispatchOpts); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(exec.contexts) == 0 { + t.Fatal("expected at least one dispatched operation") + } + for i, got := range exec.contexts { + if got.Value(ctxKey("marker")) != "present" { + t.Errorf("operation %d did not receive the caller's context", i) + } + } +} + +func TestPlanAndExecute_cancelledContextStillPropagates(t *testing.T) { + // planAndExecute does not itself check cancellation; it must hand the cancelled context + // to the operation, which is where the Helm SDK observes it via RunWithContext. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + exec := &recordingExecutor{} + if err := planAndExecute(ctx, historyReturning(nil, driver.ErrReleaseNotFound), exec, dispatchOpts); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(exec.contexts) != 1 { + t.Fatalf("dispatched %d operations, want 1", len(exec.contexts)) + } + if !errors.Is(exec.contexts[0].Err(), context.Canceled) { + t.Errorf("operation context error = %v, want context.Canceled", exec.contexts[0].Err()) + } +} + +// --- action configuration ----------------------------------------------------------------- + +func TestNewPurgeAction_dropsHistory(t *testing.T) { + // KeepHistory=false is what removes the retained history holding the release name, and + // what lets an already-uninstalled release be purged instead of erroring "already + // deleted". A true value here would silently break both purge paths. + client := newPurgeAction(&action.Configuration{}, InstallOrUpgradeOptions{Wait: true, Timeout: 90}) + if client.KeepHistory { + t.Error("KeepHistory = true, want false so the retained history is removed") + } + if !client.Wait { + t.Error("Wait was not carried over from the options") + } + if client.Timeout.Seconds() != 90 { + t.Errorf("Timeout = %v, want 90s", client.Timeout) + } +} + +func TestNewRollbackAction_targetsPlannedRevision(t *testing.T) { + client := newRollbackAction(&action.Configuration{}, InstallOrUpgradeOptions{Wait: true, Timeout: 30}, 7) + if client.Version != 7 { + t.Errorf("Version = %d, want the planned revision 7", client.Version) + } + if !client.CleanupOnFail { + t.Error("CleanupOnFail = false, want true so a failed rollback does not leave new resources") + } + if !client.Wait { + t.Error("Wait was not carried over from the options") + } + if client.Timeout.Seconds() != 30 { + t.Errorf("Timeout = %v, want 30s", client.Timeout) + } +} + +func TestNewRollbackAction_zeroRevisionIsNotPlanned(t *testing.T) { + // Guard on the planner contract rather than the action: revision 0 means "previous" to + // Helm, which would be an unintended target, so planRelease must never emit it. + plan, err := planRelease([]*release.Release{ + rel(1, release.StatusSuperseded), + rel(2, release.StatusPendingUpgrade), + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if plan.rollbackRevision <= 0 { + t.Errorf("rollbackRevision = %d, want a concrete revision", plan.rollbackRevision) + } +} diff --git a/kubernetes/gateway-operator/internal/helm/recovery.go b/kubernetes/gateway-operator/internal/helm/recovery.go index d62622971d..c8d00d35a1 100644 --- a/kubernetes/gateway-operator/internal/helm/recovery.go +++ b/kubernetes/gateway-operator/internal/helm/recovery.go @@ -22,12 +22,6 @@ import ( "helm.sh/helm/v3/pkg/release" ) -// maxHistoryRevisions bounds how many stored revisions are read when planning. Helm's -// action.History truncates to Max, and treats a Max of 0 as "return nothing", so this must -// be a positive number large enough to cover any retained history (Helm keeps 10 by -// default) rather than left unset. -const maxHistoryRevisions = 1024 - // releaseOperation is the operation needed to move a release towards the desired chart. type releaseOperation int @@ -37,7 +31,9 @@ const ( // operationUpgrade upgrades the release in place. operationUpgrade // operationPurgeThenInstall removes an unusable release (history included) and installs - // it again. Only planned when no successful revision would be discarded. + // it again. The safety property is that no *live resources* are discarded: either the + // release never deployed (revision-1 pending-install), or it was already uninstalled and + // only its history remains — which may still contain earlier successful revisions. operationPurgeThenInstall // operationRollbackThenUpgrade rolls the release back to a known-good revision so the // storage leaves the pending state, then upgrades from there. diff --git a/kubernetes/gateway-operator/internal/helm/recovery_test.go b/kubernetes/gateway-operator/internal/helm/recovery_test.go index c8d4bfe7b0..bd74f8ee95 100644 --- a/kubernetes/gateway-operator/internal/helm/recovery_test.go +++ b/kubernetes/gateway-operator/internal/helm/recovery_test.go @@ -250,11 +250,3 @@ func TestReleaseOperation_String(t *testing.T) { } } } - -func TestMaxHistoryRevisions_isPositive(t *testing.T) { - // action.History treats Max == 0 as "return nothing", which would make every release - // look absent and turn an upgrade into a fresh install. - if maxHistoryRevisions <= 0 { - t.Fatalf("maxHistoryRevisions = %d, want a positive limit", maxHistoryRevisions) - } -} From 9cc3d7af7cd269afbf849c6784dcc5de4cb1bec4 Mon Sep 17 00:00:00 2001 From: Sanjula Herath Date: Wed, 5 Aug 2026 10:44:49 +0530 Subject: [PATCH 4/4] test(gateway-operator): drive exhaustion recovery through reconcile The exhaustion status-patch test configured a deployment spy but never exercised it: it called decideFailedRecovery directly and silenced the unused variable, so it did not prove the reconcile path recovers. Run the real decision path twice after the failed status write instead. Inside the retry window it must not deploy and must still return a bounded RequeueAfter; once the window has elapsed it must deploy exactly once. The tracker entry now carries the gateway's real input fingerprint so the unchanged-input path is the one under test. Also correct the GatewayTracker comment, which still claimed entries persist until the CR is deleted. Related to #3105 --- .../controller/apigateway_controller.go | 5 ++- .../controller/apigateway_statuspatch_test.go | 41 +++++++++++++++---- 2 files changed, 35 insertions(+), 11 deletions(-) diff --git a/kubernetes/gateway-operator/internal/controller/apigateway_controller.go b/kubernetes/gateway-operator/internal/controller/apigateway_controller.go index 52d5f13a14..7216a1b869 100644 --- a/kubernetes/gateway-operator/internal/controller/apigateway_controller.go +++ b/kubernetes/gateway-operator/internal/controller/apigateway_controller.go @@ -84,8 +84,9 @@ type GatewayTrackingEntry struct { InputHash string } -// GatewayTracker manages in-memory tracking of APIGateway deployment states -// Entries persist until the APIGateway CR is deleted +// GatewayTracker manages in-memory tracking of APIGateway deployment states. +// Entries persist until the APIGateway CR is deleted, or until a status write fails and the +// entry is dropped so the next reconcile re-derives state from the CR (see forgetTracking). type GatewayTracker struct { mu sync.RWMutex entries map[string]*GatewayTrackingEntry // key: "namespace/name" diff --git a/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go b/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go index a8995ccfed..272be51e9f 100644 --- a/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go +++ b/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go @@ -183,13 +183,18 @@ func TestHandleGatewayDeploymentError_exhaustionStatusPatchFailureStaysRecoverab } r, _ := newStatusFailureReconciler(t, gw, 1) + // Record the failure against the gateway's real deployment inputs, so the follow-up + // reconciles exercise the unchanged-input path rather than an accidental hash mismatch. + _, inputHash, err := r.deploymentInputs(context.Background(), gw) + if err != nil { + t.Fatalf("deploymentInputs: %v", err) + } entry := &GatewayTrackingEntry{ - Generation: 1, Status: GatewayTrackingStatusProcessing, RetryCount: 2, InputHash: "hash-1", + Generation: 1, Status: GatewayTrackingStatusProcessing, RetryCount: 2, InputHash: inputHash, } r.gatewayTracker.Set(testTrackKey, entry) - _, err := r.handleGatewayDeploymentError(context.Background(), gw, testTrackKey, entry, errors.New("boom"), 0) - if err == nil { + if _, err := r.handleGatewayDeploymentError(context.Background(), gw, testTrackKey, entry, errors.New("boom"), 0); err == nil { t.Fatal("expected the failure status patch error to be returned") } @@ -203,20 +208,38 @@ func TestHandleGatewayDeploymentError_exhaustionStatusPatchFailureStaysRecoverab t.Errorf("status = %q, want %q", stored.Status, GatewayTrackingStatusFailed) } if stored.NextRetryTime.IsZero() { - t.Error("expected a bounded next-retry time despite the failed status write") + t.Fatal("expected a bounded next-retry time despite the failed status write") } - // Corrected inputs must still recover immediately, without waiting for the window. deployCalls := 0 r.deployGateway = func(_ context.Context, _ *apiv1.APIGateway, _ string, _ int64, _, _ string) (ctrl.Result, error) { deployCalls++ return ctrl.Result{}, nil } - deployNow, _, _ := r.decideFailedRecovery(stored, "hash-2", stored.NextRetryTime.Add(-time.Hour)) - if !deployNow { - t.Error("changed inputs must recover immediately even after a failed status write") + + // Second reconcile, inside the retry window: the failure was never persisted, so the CR + // still carries no Programmed condition and this runs the real decision path. + res, err := callDecide(t, r, gw) + if err != nil { + t.Fatalf("reconcile inside the retry window returned an error: %v", err) + } + if deployCalls != 0 { + t.Errorf("deploy attempts inside the retry window = %d, want 0", deployCalls) + } + if res.RequeueAfter <= 0 { + t.Errorf("RequeueAfter = %v, want a bounded retry despite the failed status write", res.RequeueAfter) + } + + // Third reconcile, once the window has elapsed: exactly one bounded attempt. + stored.NextRetryTime = time.Now().Add(-time.Second) + r.gatewayTracker.Set(testTrackKey, stored) + + if _, err := callDecide(t, r, gw); err != nil { + t.Fatalf("reconcile after the retry window returned an error: %v", err) + } + if deployCalls != 1 { + t.Errorf("deploy attempts after the retry window = %d, want exactly 1", deployCalls) } - _ = deployCalls } func findProgrammed(gw *apiv1.APIGateway) *metav1.Condition {