From 5ab882cdd76e67a055bf1f6987b4b1a1b35f8751 Mon Sep 17 00:00:00 2001 From: Kampit Ojha <97059622+kampitojha@users.noreply.github.com> Date: Wed, 12 Aug 2026 15:05:25 +0530 Subject: [PATCH] Fix(rollout): trigger OpenKruise reversion when paused at canary step (#7251) * fix(rollout): trigger OpenKruise reversion when paused at canary step Signed-off-by: kampitojha * refactor(rollout): extract shared helper to deduplicate resume and rollback logic Signed-off-by: kampitojha * fix(rollout): fix error message formatting and verbs in rollout helper Signed-off-by: kampitojha * test(rollout): set valid canaryRevision and podTemplateHash in rollout test Signed-off-by: kampitojha * test(rollout): remove outdated required fields under status.canaryStatus in test CRD schema Signed-off-by: kampitojha * test(rollout): add coverage tests for writer path, canary-only paused, and no-op resume cases Signed-off-by: kampitojha * test(rollout): cover seen-key deduplication branch with duplicate RT entry Signed-off-by: kampitojha * style(rollout): remove extra blank line to pass goimports check-diff Signed-off-by: kampitojha * fix(rollout): wait for OpenKruise to settle before correcting canary step state Signed-off-by: kampitojha * test(rollout): cover rollback settle-timeout and rollout-disappearance error paths Signed-off-by: kampitojha * chore(ci): re-trigger workflows after GitHub Actions outage Signed-off-by: kampitojha --------- Signed-off-by: kampitojha Co-authored-by: kampitojha --- pkg/rollout/rollout.go | 92 +++++---- pkg/rollout/rollout_test.go | 187 +++++++++++++++++- pkg/rollout/testdata/rollouts.yaml | 4 - pkg/workflow/operation/testdata/rollouts.yaml | 4 - 4 files changed, 236 insertions(+), 51 deletions(-) diff --git a/pkg/rollout/rollout.go b/pkg/rollout/rollout.go index 7a5e93f26..9f6671844 100644 --- a/pkg/rollout/rollout.go +++ b/pkg/rollout/rollout.go @@ -20,9 +20,11 @@ import ( "context" "fmt" "io" + "time" "github.com/pkg/errors" k8stypes "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" @@ -36,6 +38,14 @@ import ( velaerrors "github.com/oam-dev/kubevela/pkg/utils/errors" ) +// rolloutSettleInterval is how often the rollout phase is polled while waiting +// for the OpenKruise controller to finish its reconcile. +var rolloutSettleInterval = 2 * time.Second + +// rolloutSettleTimeout bounds how long we wait for the OpenKruise controller to +// leave the Progressing phase before correcting the canary step state. +var rolloutSettleTimeout = 2 * time.Minute + // ClusterRollout rollout in specified cluster type ClusterRollout struct { *kruisev1alpha1.Rollout @@ -51,12 +61,18 @@ func getAssociatedRollouts(ctx context.Context, cli client.Client, app *v1beta1. historyRTs = []*v1beta1.ResourceTracker{} } var rollouts []*ClusterRollout + seen := make(map[string]bool) for _, rt := range append(historyRTs, rootRT, currentRT) { if rt == nil { continue } for _, mr := range rt.Spec.ManagedResources { if mr.APIVersion == kruisev1alpha1.SchemeGroupVersion.String() && mr.Kind == "Rollout" { + key := fmt.Sprintf("%s/%s/%s", mr.Cluster, mr.Namespace, mr.Name) + if seen[key] { + continue + } + seen[key] = true rollout := &kruisev1alpha1.Rollout{} if err = cli.Get(multicluster.ContextWithClusterName(ctx, mr.Cluster), k8stypes.NamespacedName{Namespace: mr.Namespace, Name: mr.Name}, rollout); err != nil { if multicluster.IsNotFoundOrClusterNotExists(err) || velaerrors.IsCRDNotExists(err) { @@ -108,8 +124,7 @@ func SuspendRollout(ctx context.Context, cli client.Client, app *v1beta1.Applica return nil } -// ResumeRollout find all rollouts associated with the application (in the current RT) and resume them -func ResumeRollout(ctx context.Context, cli client.Client, app *v1beta1.Application, writer io.Writer) (bool, error) { +func resumeOrRollbackRollout(ctx context.Context, cli client.Client, app *v1beta1.Application, writer io.Writer, action string, logVerb string, waitForSettle bool) (bool, error) { rollouts, err := getAssociatedRollouts(ctx, cli, app, false) if err != nil { return false, err @@ -135,7 +150,10 @@ func ResumeRollout(ctx context.Context, cli client.Client, app *v1beta1.Applicat } return nil }); err != nil { - return false, errors.Wrapf(err, "failed to resume rollout %s/%s in cluster %s", rollout.Namespace, rollout.Name, rollout.Cluster) + return false, errors.Wrapf(err, "failed to %s rollout %s/%s in cluster %s", action, rollout.Namespace, rollout.Name, rollout.Cluster) + } + if err = waitForRolloutSettle(_ctx, cli, rolloutKey, waitForSettle, action, rollout.Namespace, rollout.Name, rollout.Cluster); err != nil { + return false, err } if err = retry.RetryOnConflict(retry.DefaultBackoff, func() error { if err = cli.Get(_ctx, rolloutKey, rollout.Rollout); err != nil { @@ -151,12 +169,12 @@ func ResumeRollout(ctx context.Context, cli client.Client, app *v1beta1.Applicat } return nil }); err != nil { - return false, errors.Wrapf(err, "failed to resume rollout %s/%s in cluster %s", rollout.Namespace, rollout.Name, rollout.Cluster) + return false, errors.Wrapf(err, "failed to %s rollout %s/%s in cluster %s", action, rollout.Namespace, rollout.Name, rollout.Cluster) } if resumed { modified = true if writer != nil { - _, _ = fmt.Fprintf(writer, "Rollout %s/%s in cluster %s resumed.\n", rollout.Namespace, rollout.Name, rollout.Cluster) + _, _ = fmt.Fprintf(writer, "Rollout %s/%s in cluster %s %s.\n", rollout.Namespace, rollout.Name, rollout.Cluster, logVerb) } } } @@ -164,42 +182,34 @@ func ResumeRollout(ctx context.Context, cli client.Client, app *v1beta1.Applicat return modified, nil } +// waitForRolloutSettle waits for the OpenKruise controller to finish its own +// reconcile (i.e. leave the Progressing phase) before KubeVela corrects the +// canary step state. Writing the status while the controller is still +// reconciling races with its own status writes and can leave currentStepState +// stale even after a successful rollback. On timeout we fall back to the +// best-effort status correction rather than failing the whole operation. +func waitForRolloutSettle(ctx context.Context, cli client.Client, rolloutKey client.ObjectKey, waitForSettle bool, action, namespace, name, cluster string) error { + if !waitForSettle { + return nil + } + if err := wait.PollUntilContextTimeout(ctx, rolloutSettleInterval, rolloutSettleTimeout, true, func(ctx context.Context) (bool, error) { + rollout := &kruisev1alpha1.Rollout{} + if err := cli.Get(ctx, rolloutKey, rollout); err != nil { + return false, err + } + return rollout.Status.Phase != kruisev1alpha1.RolloutPhaseProgressing, nil + }); err != nil && !wait.Interrupted(err) { + return errors.Wrapf(err, "failed to wait for rollout %s/%s in cluster %s to settle before %s", namespace, name, cluster, action) + } + return nil +} + +// ResumeRollout find all rollouts associated with the application (in the current RT) and resume them +func ResumeRollout(ctx context.Context, cli client.Client, app *v1beta1.Application, writer io.Writer) (bool, error) { + return resumeOrRollbackRollout(ctx, cli, app, writer, "resume", "resumed", false) +} + // RollbackRollout find all rollouts associated with the application (in the current RT) and disable the pause field. func RollbackRollout(ctx context.Context, cli client.Client, app *v1beta1.Application, writer io.Writer) (bool, error) { - rollouts, err := getAssociatedRollouts(ctx, cli, app, false) - if err != nil { - return false, err - } - modified := false - for i := range rollouts { - rollout := rollouts[i] - if rollout.Spec.Strategy.Paused || (rollout.Status.CanaryStatus != nil && rollout.Status.CanaryStatus.CurrentStepState == kruisev1alpha1.CanaryStepStatePaused) { - _ctx := multicluster.ContextWithClusterName(ctx, rollout.Cluster) - rolloutKey := client.ObjectKeyFromObject(rollout.Rollout) - resumed := false - if err = retry.RetryOnConflict(retry.DefaultBackoff, func() error { - if err = cli.Get(_ctx, rolloutKey, rollout.Rollout); err != nil { - return err - } - if rollout.Spec.Strategy.Paused { - rollout.Spec.Strategy.Paused = false - if err = cli.Update(_ctx, rollout.Rollout); err != nil { - return err - } - resumed = true - return nil - } - return nil - }); err != nil { - return false, errors.Wrapf(err, "failed to rollback rollout %s/%s in cluster %s", rollout.Namespace, rollout.Name, rollout.Cluster) - } - if resumed { - modified = true - if writer != nil { - _, _ = fmt.Fprintf(writer, "Rollout %s/%s in cluster %s rollback.\n", rollout.Namespace, rollout.Name, rollout.Cluster) - } - } - } - } - return modified, nil + return resumeOrRollbackRollout(ctx, cli, app, writer, "rollback", "rollback", true) } diff --git a/pkg/rollout/rollout_test.go b/pkg/rollout/rollout_test.go index 361e4c122..ca5b2ec33 100644 --- a/pkg/rollout/rollout_test.go +++ b/pkg/rollout/rollout_test.go @@ -18,6 +18,8 @@ package rollout import ( "context" + "strings" + "time" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -73,9 +75,176 @@ var _ = Describe("Kruise rollout test", func() { Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) r.Spec.Strategy.Paused = true Expect(k8sClient.Update(ctx, &r)).Should(BeNil()) - Expect(RollbackRollout(ctx, k8sClient, &app, nil)) - Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)) + r.Status.Phase = kruisev1alpha1.RolloutPhaseHealthy + r.Status.CanaryStatus = &kruisev1alpha1.CanaryStatus{ + CurrentStepState: kruisev1alpha1.CanaryStepStatePaused, + } + Expect(k8sClient.Status().Update(ctx, &r)).Should(BeNil()) + + modified, err := RollbackRollout(ctx, k8sClient, &app, nil) + Expect(err).Should(BeNil()) + Expect(modified).Should(BeTrue()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) Expect(r.Spec.Strategy.Paused).Should(BeEquivalentTo(false)) + Expect(r.Status.CanaryStatus.CurrentStepState).Should(BeEquivalentTo(kruisev1alpha1.CanaryStepStateReady)) + }) + + It("Rollback rollout with writer covers log output path", func() { + r := kruisev1alpha1.Rollout{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + r.Spec.Strategy.Paused = true + Expect(k8sClient.Update(ctx, &r)).Should(BeNil()) + r.Status.Phase = kruisev1alpha1.RolloutPhaseHealthy + r.Status.CanaryStatus = &kruisev1alpha1.CanaryStatus{ + CurrentStepState: kruisev1alpha1.CanaryStepStatePaused, + } + Expect(k8sClient.Status().Update(ctx, &r)).Should(BeNil()) + + buf := &strings.Builder{} + modified, err := RollbackRollout(ctx, k8sClient, &app, buf) + Expect(err).Should(BeNil()) + Expect(modified).Should(BeTrue()) + Expect(buf.String()).Should(ContainSubstring("rollback")) + }) + + It("Rollback rollout waits for OpenKruise to settle before correcting canary status", func() { + r := kruisev1alpha1.Rollout{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + r.Spec.Strategy.Paused = true + Expect(k8sClient.Update(ctx, &r)).Should(BeNil()) + r.Status.Phase = kruisev1alpha1.RolloutPhaseProgressing + r.Status.CanaryStatus = &kruisev1alpha1.CanaryStatus{ + CurrentStepState: kruisev1alpha1.CanaryStepStatePaused, + } + Expect(k8sClient.Status().Update(ctx, &r)).Should(BeNil()) + + // simulate the OpenKruise controller: once KubeVela unpauses the spec, + // settle the rollout by leaving the Progressing phase with the canary + // step state still stale at StepPaused (the exact race being fixed). + done := make(chan struct{}) + go func() { + defer close(done) + settleWhenUnpaused := func() bool { + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + var rr kruisev1alpha1.Rollout + if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &rr); err == nil && !rr.Spec.Strategy.Paused { + rr.Status.Phase = kruisev1alpha1.RolloutPhaseHealthy + if err := k8sClient.Status().Update(ctx, &rr); err == nil { + return true + } + } + time.Sleep(50 * time.Millisecond) + } + return false + } + settleWhenUnpaused() + }() + + modified, err := RollbackRollout(ctx, k8sClient, &app, nil) + Expect(err).Should(BeNil()) + Expect(modified).Should(BeTrue()) + Eventually(done).WithTimeout(10 * time.Second).WithPolling(50 * time.Millisecond).Should(BeClosed()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + Expect(r.Spec.Strategy.Paused).Should(BeEquivalentTo(false)) + Expect(r.Status.CanaryStatus.CurrentStepState).Should(BeEquivalentTo(kruisev1alpha1.CanaryStepStateReady)) + }) + + It("Rollback rollout proceeds best-effort when OpenKruise does not settle within timeout", func() { + oldInterval, oldTimeout := rolloutSettleInterval, rolloutSettleTimeout + rolloutSettleInterval = 300 * time.Millisecond + rolloutSettleTimeout = 1 * time.Second + defer func() { rolloutSettleInterval, rolloutSettleTimeout = oldInterval, oldTimeout }() + + r := kruisev1alpha1.Rollout{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + r.Spec.Strategy.Paused = true + Expect(k8sClient.Update(ctx, &r)).Should(BeNil()) + r.Status.Phase = kruisev1alpha1.RolloutPhaseProgressing + r.Status.CanaryStatus = &kruisev1alpha1.CanaryStatus{ + CurrentStepState: kruisev1alpha1.CanaryStepStatePaused, + } + Expect(k8sClient.Status().Update(ctx, &r)).Should(BeNil()) + + modified, err := RollbackRollout(ctx, k8sClient, &app, nil) + Expect(err).Should(BeNil()) + Expect(modified).Should(BeTrue()) + + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + Expect(r.Spec.Strategy.Paused).Should(BeEquivalentTo(false)) + Expect(r.Status.CanaryStatus.CurrentStepState).Should(BeEquivalentTo(kruisev1alpha1.CanaryStepStateReady)) + }) + + It("Rollback rollout returns an error when the rollout disappears while settling", func() { + r := kruisev1alpha1.Rollout{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + r.Spec.Strategy.Paused = true + Expect(k8sClient.Update(ctx, &r)).Should(BeNil()) + r.Status.Phase = kruisev1alpha1.RolloutPhaseProgressing + r.Status.CanaryStatus = &kruisev1alpha1.CanaryStatus{ + CurrentStepState: kruisev1alpha1.CanaryStepStatePaused, + } + Expect(k8sClient.Status().Update(ctx, &r)).Should(BeNil()) + + // simulate the rollout being deleted while the rollback waits for the + // OpenKruise controller to settle, so the poll surfaces a real error. + done := make(chan struct{}) + go func() { + defer close(done) + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + var rr kruisev1alpha1.Rollout + if err := k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &rr); err == nil && !rr.Spec.Strategy.Paused { + _ = k8sClient.Delete(ctx, &rr) + return + } + time.Sleep(50 * time.Millisecond) + } + }() + + modified, err := RollbackRollout(ctx, k8sClient, &app, nil) + Expect(err).Should(HaveOccurred()) + Expect(modified).Should(BeFalse()) + Eventually(done).WithTimeout(10 * time.Second).WithPolling(50 * time.Millisecond).Should(BeClosed()) + }) + + It("Resume rollout with only CanaryStatus paused (spec not paused)", func() { + r := kruisev1alpha1.Rollout{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + // spec is NOT paused, only canaryStatus is paused + r.Spec.Strategy.Paused = false + Expect(k8sClient.Update(ctx, &r)).Should(BeNil()) + r.Status.CanaryStatus = &kruisev1alpha1.CanaryStatus{ + CurrentStepState: kruisev1alpha1.CanaryStepStatePaused, + } + Expect(k8sClient.Status().Update(ctx, &r)).Should(BeNil()) + + modified, err := ResumeRollout(ctx, k8sClient, &app, nil) + Expect(err).Should(BeNil()) + Expect(modified).Should(BeTrue()) + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + Expect(r.Status.CanaryStatus.CurrentStepState).Should(BeEquivalentTo(kruisev1alpha1.CanaryStepStateReady)) + }) + + It("Resume rollout that is already not paused returns modified=false", func() { + r := kruisev1alpha1.Rollout{} + Expect(k8sClient.Get(ctx, types.NamespacedName{Namespace: "default", Name: "my-rollout"}, &r)).Should(BeNil()) + r.Spec.Strategy.Paused = false + Expect(k8sClient.Update(ctx, &r)).Should(BeNil()) + // no canary status set — fully unpaused + r.Status.CanaryStatus = nil + Expect(k8sClient.Status().Update(ctx, &r)).Should(BeNil()) + + modified, err := ResumeRollout(ctx, k8sClient, &app, nil) + Expect(err).Should(BeNil()) + Expect(modified).Should(BeFalse()) + }) + + It("test get associated rollout deduplication", func() { + rollouts, err := getAssociatedRollouts(ctx, k8sClient, &app, true) + Expect(err).Should(BeNil()) + Expect(len(rollouts)).Should(BeEquivalentTo(1)) }) }) @@ -111,6 +280,20 @@ var rt = v1beta1.ResourceTracker{ ApplicationGeneration: 1, Type: v1beta1.ResourceTrackerTypeVersioned, ManagedResources: []v1beta1.ManagedResource{ + { + ClusterObjectReference: common.ClusterObjectReference{ + ObjectReference: v1.ObjectReference{ + APIVersion: "rollouts.kruise.io/v1alpha1", + Kind: "Rollout", + Name: "my-rollout", + Namespace: "default", + }, + }, + OAMObjectReference: common.OAMObjectReference{ + Component: "my-rollout", + }, + }, + // Duplicate entry for my-rollout to exercise the seen[key] deduplication branch { ClusterObjectReference: common.ClusterObjectReference{ ObjectReference: v1.ObjectReference{ diff --git a/pkg/rollout/testdata/rollouts.yaml b/pkg/rollout/testdata/rollouts.yaml index 43ab7546a..6124a631a 100644 --- a/pkg/rollout/testdata/rollouts.yaml +++ b/pkg/rollout/testdata/rollouts.yaml @@ -217,11 +217,7 @@ spec: description: RolloutHash from rollout.spec object type: string required: - - canaryReadyReplicas - - canaryReplicas - - canaryService - currentStepState - - podTemplateHash type: object conditions: description: Conditions a list of conditions a rollout can have. diff --git a/pkg/workflow/operation/testdata/rollouts.yaml b/pkg/workflow/operation/testdata/rollouts.yaml index 43ab7546a..6124a631a 100644 --- a/pkg/workflow/operation/testdata/rollouts.yaml +++ b/pkg/workflow/operation/testdata/rollouts.yaml @@ -217,11 +217,7 @@ spec: description: RolloutHash from rollout.spec object type: string required: - - canaryReadyReplicas - - canaryReplicas - - canaryService - currentStepState - - podTemplateHash type: object conditions: description: Conditions a list of conditions a rollout can have.