improve the canary features

Signed-off-by: Jiaxin Shan <[email protected]>

Author: Jeffwan

pkg/controller/stormservice/canary.go

@@ -139,9 +139,10 @@ func (r *StormServiceReconciler) processCanaryUpdate(ctx context.Context, stormS
 	currentStepIndex := canaryStatus.CurrentStep
 	if currentStepIndex < int32(len(steps)) {
 		currentStep := steps[currentStepIndex]
-		// If current step is a manual pause step but no pause condition exists, advance
+		// Treat as resume ONLY if we've already marked PausedAt (i.e., entered this pause step)
+		// and the CanaryPauseStep condition has been removed by the user.
 		if currentStep.Pause != nil && currentStep.Pause.IsManualPause() {
-			if !r.hasPauseCondition(canaryStatus, orchestrationv1alpha1.PauseReasonCanaryPauseStep) {
+			if canaryStatus.PausedAt != nil && !r.hasPauseCondition(canaryStatus, orchestrationv1alpha1.PauseReasonCanaryPauseStep) {
 				klog.Infof("Manual pause condition removed for StormService %s/%s, advancing to next step",
 					stormService.Namespace, stormService.Name)
 				return r.advanceCanaryStep(ctx, stormService)
@@ -211,7 +212,7 @@ func (r *StormServiceReconciler) processCanaryPauseStep(ctx context.Context, sto
 			addStatusUpdate(func(status *orchestrationv1alpha1.CanaryStatus) {
 				status.PausedAt = &now
 				status.Phase = orchestrationv1alpha1.CanaryPhasePaused
-				
+
 				// Add pause condition to track why it's paused
 				pauseCondition := orchestrationv1alpha1.PauseCondition{
 					Reason:    orchestrationv1alpha1.PauseReasonCanaryPauseStep,
@@ -297,8 +298,12 @@ func (r *StormServiceReconciler) processCanaryPauseStep(ctx context.Context, sto
 			return ctrl.Result{}, fmt.Errorf("failed to add CanaryPauseStep pause condition: %w", err)
 		}
 	} else {
-		r.EventRecorder.Eventf(stormService, "Normal", "CanaryPauseManual",
-			"Canary paused at manual pause step. Remove CanaryPauseStep pause condition to continue")
+		// Emit a consistent CanaryUpdate event even if the pause condition already exists
+		update := newCanaryStatusUpdate().
+			addEvent("Canary paused at manual pause step. Remove CanaryPauseStep pause condition to continue")
+		if err := r.applyCanaryStatusUpdate(ctx, stormService, update); err != nil {
+			return ctrl.Result{}, fmt.Errorf("failed to record manual pause event: %w", err)
+		}
 	}
 
 	// Requeue periodically to check if pause condition was removed
@@ -342,15 +347,15 @@ func (r *StormServiceReconciler) applyCanaryWeight(ctx context.Context, stormSer
 
 	// Check if this is a scaling event during canary (compare with current status replicas)
 	currentStatusReplicas := stormService.Status.Replicas
-	
+
 	// If replicas changed during canary, log and notify but continue with recalculation
 	if currentStatusReplicas > 0 && currentStatusReplicas != totalReplicas {
-		klog.Infof("Scaling detected during canary for StormService %s/%s: %d -> %d replicas, recalculating distribution", 
+		klog.Infof("Scaling detected during canary for StormService %s/%s: %d -> %d replicas, recalculating distribution",
 			stormService.Namespace, stormService.Name, currentStatusReplicas, totalReplicas)
-		
+
 		// Log the recalculation event
 		r.EventRecorder.Eventf(stormService, "Normal", "CanaryScaling",
-			"Recalculated canary distribution due to scaling: %d%% weight with %d total replicas (%d canary, %d stable)", 
+			"Recalculated canary distribution due to scaling: %d%% weight with %d total replicas (%d canary, %d stable)",
 			weight, totalReplicas, canaryReplicas, stableReplicas)
 	}
 
@@ -401,7 +406,7 @@ func (r *StormServiceReconciler) advanceCanaryStep(ctx context.Context, stormSer
 	update := newCanaryStatusUpdate().
 		addStatusUpdate(func(status *orchestrationv1alpha1.CanaryStatus) {
 			status.CurrentStep = nextStep
-			status.PausedAt = nil // Clear pause timestamp
+			status.PausedAt = nil        // Clear pause timestamp
 			status.PauseConditions = nil // Clear pause conditions when resuming
 			status.Phase = orchestrationv1alpha1.CanaryPhaseProgressing
 		})
@@ -474,15 +479,17 @@ func (r *StormServiceReconciler) completeCanary(ctx context.Context, stormServic
 		return ctrl.Result{}, fmt.Errorf("failed to promote canary revision: %w", err)
 	}
 
+	// Use single patch operation for the final status update
+	// Capture the original object BEFORE mutating status to ensure the patch contains the changes
+	original := stormService.DeepCopy()
+
 	// Step 3: Update main status to reflect the promotion
 	stormService.Status.CurrentRevision = promotedRevision
 	stormService.Status.UpdateRevision = promotedRevision
 
 	// Step 4: Clear canary status - this triggers normal rollout logic to take over
 	stormService.Status.CanaryStatus = nil
 
-	// Use single patch operation for the final status update
-	original := stormService.DeepCopy()
 	if err := r.Status().Patch(ctx, stormService, client.MergeFrom(original)); err != nil {
 		return ctrl.Result{}, fmt.Errorf("failed to clear canary status and promote revision: %w", err)
 	}

pkg/controller/stormservice/canary_test.go

@@ -237,29 +237,33 @@ func TestProcessCanaryWeightStep(t *testing.T) {
 
 	// Verify events were recorded (may receive multiple events)
 	var events []string
-	timeout := time.After(time.Second)
-eventLoop:
-	for len(events) < 2 {
+	timeout := time.After(2 * time.Second)
+collect:
+	for len(events) < 3 { // collect a few events or until timeout
 		select {
 		case event := <-eventRecorder.Events:
 			events = append(events, event)
 		case <-timeout:
-			break eventLoop
+			break collect
 		}
 	}
 
-	// Should have received both CanaryWeightApplied and CanaryReplicaMode events
+	// We expect at least the generic CanaryUpdate (weight set) and a mode-specific event
+	// Replica mode is used here since replicas=3
 	require.True(t, len(events) >= 1, "Expected at least one event")
 
-	// Check that we got either CanaryWeightApplied or CanaryReplicaMode (or both)
-	foundWeightEvent := false
-	for _, event := range events {
-		if strings.Contains(event, "33%") && (strings.Contains(event, "CanaryWeightApplied") || strings.Contains(event, "CanaryReplicaMode")) {
-			foundWeightEvent = true
-			break
+	foundUpdate := false
+	foundReplicaMode := false
+	for _, e := range events {
+		if strings.Contains(e, "CanaryUpdate") && strings.Contains(e, "33%") {
+			foundUpdate = true
+		}
+		if strings.Contains(e, "CanaryReplicaMode") && strings.Contains(e, "33%") {
+			foundReplicaMode = true
 		}
 	}
-	assert.True(t, foundWeightEvent, "Expected to find weight-related event with 33%, got events: %v", events)
+	assert.True(t, foundUpdate, "Expected CanaryUpdate event with 33%%, got: %v", events)
+	assert.True(t, foundReplicaMode, "Expected CanaryReplicaMode event with 33%%, got: %v", events)
 }
 
 func TestProcessCanaryPauseStep_AutomaticPause(t *testing.T) {
@@ -459,6 +463,8 @@ func TestAdvanceCanaryStep(t *testing.T) {
 	assert.Equal(t, int32(1), stormService.Status.CanaryStatus.CurrentStep)
 	assert.Nil(t, stormService.Status.CanaryStatus.PausedAt)
 	assert.Equal(t, orchestrationv1alpha1.CanaryPhaseProgressing, stormService.Status.CanaryStatus.Phase)
+	// Pause conditions should also be cleared when advancing
+	assert.Empty(t, stormService.Status.CanaryStatus.PauseConditions)
 }
 
 func TestHasPauseCondition(t *testing.T) {

test/e2e/stormservice_canary_test.go

@@ -727,6 +727,12 @@ func waitForCanaryResume(t *testing.T, client *v1alpha1.Clientset, name string)
 				return false, err
 			}
 
+			if storm.Status.CanaryStatus == nil {
+				// Already completed or cleared; consider as resumed for this check
+				t.Logf("canary status cleared - considered resumed")
+				return true, nil
+			}
+
 			t.Logf("resume check: spec.paused=%v, canary phase=%s, step=%d, pausedAt=%v",
 				storm.Spec.Paused, storm.Status.CanaryStatus.Phase,
 				storm.Status.CanaryStatus.CurrentStep, storm.Status.CanaryStatus.PausedAt)

test/e2e/util.go

@@ -181,12 +181,12 @@ func waitForStormServiceReady(ctx context.Context, t *testing.T, k8sClient *kube
 
 		// Check if all replicas are ready
 		if ss.Status.ReadyReplicas > 0 && ss.Status.ReadyReplicas == ss.Status.Replicas {
-			t.Logf("StormService %s is ready: %d/%d replicas ready", 
+			t.Logf("StormService %s is ready: %d/%d replicas ready",
 				stormService.Name, ss.Status.ReadyReplicas, ss.Status.Replicas)
 			return true, nil
 		}
 
-		t.Logf("Waiting for StormService %s to be ready: %d/%d replicas ready", 
+		t.Logf("Waiting for StormService %s to be ready: %d/%d replicas ready",
 			stormService.Name, ss.Status.ReadyReplicas, ss.Status.Replicas)
 		return false, nil
 	})

test/integration/controller/stormservice_canary_test.go

@@ -317,6 +317,89 @@ var _ = ginkgo.Describe("StormService Canary Controller Integration Test", func(
 				g.Expect(cleared.Status.CanaryStatus.Phase).To(gomega.Equal(orchestrationapi.CanaryPhaseProgressing))
 			}, time.Second*10, time.Millisecond*500).Should(gomega.Succeed())
 		})
+
+		ginkgo.It("should clear canary status on completion", func() {
+			name := "clear-canary-completion"
+			stormService := createCanaryStormService(ns.Name, name)
+
+			// Create StormService
+			gomega.Expect(k8sClient.Create(ctx, stormService)).To(gomega.Succeed())
+
+			// Wait for initial revision to be set by the controller
+			gomega.Eventually(func() bool {
+				got := &orchestrationapi.StormService{}
+				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(stormService), got); err != nil {
+					return false
+				}
+				return got.Status.CurrentRevision != ""
+			}, time.Second*30, time.Millisecond*300).Should(gomega.BeTrue())
+
+			// Trigger a new revision by changing the image to start canary
+			gomega.Eventually(func(g gomega.Gomega) {
+				latest := &orchestrationapi.StormService{}
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(stormService), latest)).To(gomega.Succeed())
+				// Change container image to trigger updateRevision != currentRevision
+				if len(latest.Spec.Template.Spec.Roles) > 0 && len(latest.Spec.Template.Spec.Roles[0].Template.Spec.Containers) > 0 {
+					latest.Spec.Template.Spec.Roles[0].Template.Spec.Containers[0].Image = "nginx:1.21"
+					g.Expect(k8sClient.Update(ctx, latest)).To(gomega.Succeed())
+				}
+			}, time.Second*10, time.Millisecond*200).Should(gomega.Succeed())
+
+			// Wait until controller detects update (either revisions differ or canary status appears)
+			gomega.Eventually(func() bool {
+				updated := &orchestrationapi.StormService{}
+				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(stormService), updated); err != nil {
+					return false
+				}
+				return updated.Status.UpdateRevision != updated.Status.CurrentRevision || updated.Status.CanaryStatus != nil
+			}, time.Second*60, time.Millisecond*300).Should(gomega.BeTrue())
+
+			// Simulate completion by setting CurrentStep >= len(steps) and 100% weight
+			gomega.Eventually(func(g gomega.Gomega) {
+				latest := &orchestrationapi.StormService{}
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(stormService), latest)).To(gomega.Succeed())
+				steps := latest.Spec.UpdateStrategy.Canary.Steps
+				// Ensure canaryStatus exists to edit
+				if latest.Status.CanaryStatus == nil {
+					latest.Status.CanaryStatus = &orchestrationapi.CanaryStatus{}
+				}
+				latest.Status.CanaryStatus.CurrentStep = int32(len(steps)) // mark as completed
+				latest.Status.CanaryStatus.CurrentWeight = 100
+				// Use observed revisions for promotion fields
+				latest.Status.CanaryStatus.StableRevision = latest.Status.CurrentRevision
+				latest.Status.CanaryStatus.CanaryRevision = latest.Status.UpdateRevision
+				latest.Status.CanaryStatus.Phase = orchestrationapi.CanaryPhaseProgressing
+
+				g.Expect(k8sClient.Status().Update(ctx, latest)).To(gomega.Succeed())
+			}, time.Second*5, time.Millisecond*200).Should(gomega.Succeed())
+
+			// Status update should be enough to trigger reconcile; no spec poke required
+
+			// Verify canaryStatus is cleared by controller completion logic and revisions align
+			gomega.Eventually(func(g gomega.Gomega) {
+				final := &orchestrationapi.StormService{}
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(stormService), final)).To(gomega.Succeed())
+
+				// Check if canary is completed first
+				if final.Status.CanaryStatus != nil {
+					// If canary status exists, it should be in completed phase or cleared
+					if final.Status.CanaryStatus.Phase == orchestrationapi.CanaryPhaseCompleted {
+						// Phase is completed, status should be cleared soon - continue waiting
+						g.Expect(final.Status.CanaryStatus).ToNot(gomega.BeNil()) // This will pass, allowing retry
+					}
+					// Still in progress - continue waiting
+					g.Expect(final.Status.CanaryStatus.Phase).To(gomega.BeElementOf(
+						orchestrationapi.CanaryPhaseInitializing,
+						orchestrationapi.CanaryPhaseProgressing,
+						orchestrationapi.CanaryPhasePaused,
+						orchestrationapi.CanaryPhaseCompleted,
+					)) // This will pass for valid phases, allowing retry
+				} else {
+					// Canary status is cleared, verify revisions align
+					g.Expect(final.Status.UpdateRevision).To(gomega.Equal(final.Status.CurrentRevision))
+				}
+			}, time.Second*120, time.Millisecond*500).Should(gomega.Succeed())
+		})
 	})
 
 	ginkgo.Context("Canary Mode Detection", func() {