diff --git a/kubernetes/gateway-operator/internal/controller/apigateway_controller.go b/kubernetes/gateway-operator/internal/controller/apigateway_controller.go index 793bdaa3f9..7216a1b869 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,10 +78,15 @@ 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 -// 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" @@ -107,7 +119,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() @@ -125,6 +139,44 @@ 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) +} + +// 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. +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 +260,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 +331,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 +402,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 +411,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 +444,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 +453,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 +465,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 +474,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,27 +498,33 @@ 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) // 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 } } @@ -530,6 +653,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 } @@ -557,17 +686,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 +712,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 +747,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..a779c7ebff --- /dev/null +++ b/kubernetes/gateway-operator/internal/controller/apigateway_recovery_test.go @@ -0,0 +1,709 @@ +/* + * 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) + } +} + +// 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) + + // 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 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) + } + } +} + +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/controller/apigateway_statuspatch_test.go b/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go new file mode 100644 index 0000000000..272be51e9f --- /dev/null +++ b/kubernetes/gateway-operator/internal/controller/apigateway_statuspatch_test.go @@ -0,0 +1,252 @@ +/* + * 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) + + // 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: inputHash, + } + r.gatewayTracker.Set(testTrackKey, entry) + + 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") + } + + // 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.Fatal("expected a bounded next-retry time despite the failed status write") + } + + deployCalls := 0 + r.deployGateway = func(_ context.Context, _ *apiv1.APIGateway, _ string, _ int64, _, _ string) (ctrl.Result, error) { + deployCalls++ + return ctrl.Result{}, nil + } + + // 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) + } +} + +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 1f3e441a45..ad08fbb044 100644 --- a/kubernetes/gateway-operator/internal/helm/client.go +++ b/kubernetes/gateway-operator/internal/helm/client.go @@ -158,19 +158,167 @@ func (c *Client) InstallOrUpgrade(ctx context.Context, opts InstallOrUpgradeOpti // Set registry client in action configuration actionConfig.RegistryClient = c.registryClient - // Check if release exists histClient := action.NewHistory(actionConfig) - histClient.Max = 1 - _, err = histClient.Run(opts.ReleaseName) - releaseExists := err == nil + 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 exec.install(ctx, 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 exec.install(ctx, opts) + + case operationUpgrade: + 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 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 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 := exec.rollback(ctx, opts, plan.rollbackRevision); err != nil { + return err + } + return exec.upgrade(ctx, 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 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) + + client := newPurgeAction(actionConfig, opts) + + 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 +} + +// 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 +} - if releaseExists { - log.Info("Release exists, performing upgrade", "release", opts.ReleaseName) - return c.upgrade(ctx, actionConfig, opts) +// 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 + client.CleanupOnFail = true + 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("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 +375,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 +398,10 @@ func (c *Client) upgrade(ctx context.Context, actionConfig *action.Configuration client.Namespace = opts.Namespace client.Wait = opts.Wait client.Version = opts.Version + // 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 { client.Timeout = time.Duration(opts.Timeout) * time.Second @@ -293,8 +446,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/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 new file mode 100644 index 0000000000..c8d00d35a1 --- /dev/null +++ b/kubernetes/gateway-operator/internal/helm/recovery.go @@ -0,0 +1,204 @@ +/* +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" +) + +// 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. 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. + 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). 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: operationPurgeThenInstall, + reason: fmt.Sprintf("latest revision %d is uninstalled with retained history", 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..bd74f8ee95 --- /dev/null +++ b/kubernetes/gateway-operator/internal/helm/recovery_test.go @@ -0,0 +1,252 @@ +/* + * 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, + }, + { + // 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: operationPurgeThenInstall, + }, + { + // 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) + } + } +}