improve the canary features

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

Author: Jeffwan

pkg/controller/stormservice/canary.go

@@ -135,19 +135,20 @@ func (r *StormServiceReconciler) processCanaryUpdate(ctx context.Context, stormS
 		return r.handleCanaryPause(ctx, stormService)
 	}
 
-	// Check if we're waiting for pause condition removal on a manual pause step
-	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
-		if currentStep.Pause != nil && currentStep.Pause.IsManualPause() {
-			if !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)
-			}
-		}
-	}
+    // Check if we're waiting for pause condition removal on a manual pause step
+    currentStepIndex := canaryStatus.CurrentStep
+    if currentStepIndex < int32(len(steps)) {
+        currentStep := steps[currentStepIndex]
+        // 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 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)
+            }
+        }
+    }
 
 	// Process current step
 	if currentStepIndex >= int32(len(steps)) {
@@ -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
 		})

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
 	})