stopping update run implementation

Signed-off-by: Britania Rodriguez Reyes <britaniar@microsoft.com>

Author: britaniar

apis/placement/v1beta1/stageupdate_types.go

@@ -489,7 +489,8 @@ const (
 	// StageUpdatingConditionProgressing indicates whether the stage updating is making progress.
 	// Its condition status can be one of the following:
 	// - "True": The stage updating is making progress.
-	// - "False": The stage updating is waiting/pausing.
+	// - "False": The stage updating is waiting.
+	// - "Unknown" means it is unknown.
 	StageUpdatingConditionProgressing StageUpdatingConditionType = "Progressing"
 
 	// StageUpdatingConditionSucceeded indicates whether the stage updating is completed successfully.

pkg/controllers/updaterun/controller.go

@@ -77,6 +77,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req runtime.Request) (runtim
 		klog.ErrorS(err, "Failed to get updateRun object", "updateRun", req.NamespacedName)
 		return runtime.Result{}, client.IgnoreNotFound(err)
 	}
+
+	// Update all existing conditions' ObservedGeneration to the current generation
+	updateAllStatusConditionsGeneration(updateRun.GetUpdateRunStatus(), updateRun.GetGeneration())
+
 	runObjRef := klog.KObj(updateRun)
 
 	// Remove waitTime from the updateRun status for BeforeStageTask and AfterStageTask for type Approval.
@@ -110,12 +114,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req runtime.Request) (runtim
 	var toBeUpdatedBindings, toBeDeletedBindings []placementv1beta1.BindingObj
 	updateRunStatus := updateRun.GetUpdateRunStatus()
 	initCond := meta.FindStatusCondition(updateRunStatus.Conditions, string(placementv1beta1.StagedUpdateRunConditionInitialized))
-	// Check if initialized regardless of generation.
-	// The updateRun spec fields are immutable except for the state field. When the state changes,
-	// the update run generation increments, but we don't need to reinitialize since initialization is a one-time setup.
-	if !(initCond != nil && initCond.Status == metav1.ConditionTrue) {
+	if !condition.IsConditionStatusTrue(initCond, updateRun.GetGeneration()) {
 		// Check if initialization failed for the current generation.
-		if initCond != nil && initCond.Status == metav1.ConditionFalse {
+		if condition.IsConditionStatusFalse(initCond, updateRun.GetGeneration()) {
 			klog.V(2).InfoS("The updateRun has failed to initialize", "errorMsg", initCond.Message, "updateRun", runObjRef)
 			return runtime.Result{}, nil
 		}
@@ -158,9 +159,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, req runtime.Request) (runtim
 		return runtime.Result{}, r.recordUpdateRunSucceeded(ctx, updateRun)
 	}
 
-	// Execute the updateRun.
-	if state == placementv1beta1.StateRun {
-		klog.V(2).InfoS("Continue to execute the updateRun", "state", state, "updatingStageIndex", updatingStageIndex, "updateRun", runObjRef)
+	switch state {
+	case placementv1beta1.StateInitialize:
+		klog.V(2).InfoS("The updateRun is initialized but not executed, waiting to execute", "state", state, "updateRun", runObjRef)
+	case placementv1beta1.StateRun:
+		// Execute the updateRun.
+		klog.InfoS("Continue to execute the updateRun", "updatingStageIndex", updatingStageIndex, "updateRun", runObjRef)
 		finished, waitTime, execErr := r.execute(ctx, updateRun, updatingStageIndex, toBeUpdatedBindings, toBeDeletedBindings)
 		if errors.Is(execErr, errStagedUpdatedAborted) {
 			// errStagedUpdatedAborted cannot be retried.
@@ -172,21 +176,41 @@ func (r *Reconciler) Reconcile(ctx context.Context, req runtime.Request) (runtim
 			return runtime.Result{}, r.recordUpdateRunSucceeded(ctx, updateRun)
 		}
 
-		// The execution is not finished yet or it encounters a retriable error.
-		// We need to record the status and requeue.
-		if updateErr := r.recordUpdateRunStatus(ctx, updateRun); updateErr != nil {
-			return runtime.Result{}, updateErr
+		return r.handleIncompleteUpdateRun(ctx, updateRun, waitTime, execErr, state, runObjRef)
+	case placementv1beta1.StateStop:
+		// Stop the updateRun.
+		klog.V(2).InfoS("Stopping the updateRun", "state", state, "updatingStageIndex", updatingStageIndex, "updateRun", runObjRef)
+		finished, waitTime, stopErr := r.stop(updateRun, updatingStageIndex, toBeUpdatedBindings, toBeDeletedBindings)
+		if errors.Is(stopErr, errStagedUpdatedAborted) {
+			// errStagedUpdatedAborted cannot be retried.
+			return runtime.Result{}, r.recordUpdateRunFailed(ctx, updateRun, stopErr.Error())
 		}
-		klog.V(2).InfoS("The updateRun is not finished yet", "requeueWaitTime", waitTime, "execErr", execErr, "updateRun", runObjRef)
-		if execErr != nil {
-			return runtime.Result{}, execErr
+
+		if finished {
+			klog.V(2).InfoS("The updateRun is stopped", "updateRun", runObjRef)
+			return runtime.Result{}, r.recordUpdateRunStopped(ctx, updateRun)
 		}
-		return runtime.Result{Requeue: true, RequeueAfter: waitTime}, nil
+
+		return r.handleIncompleteUpdateRun(ctx, updateRun, waitTime, stopErr, state, runObjRef)
 	}
-	klog.V(2).InfoS("The updateRun is initialized but not executed, waiting to execute", "state", state, "updateRun", runObjRef)
 	return runtime.Result{}, nil
 }
 
+func (r *Reconciler) handleIncompleteUpdateRun(ctx context.Context, updateRun placementv1beta1.UpdateRunObj, waitTime time.Duration, err error, state placementv1beta1.State, runObjRef klog.ObjectRef) (runtime.Result, error) {
+	// The execution or stopping is not finished yet or it encounters a retriable error.
+	// We need to record the status and requeue.
+	if updateErr := r.recordUpdateRunStatus(ctx, updateRun); updateErr != nil {
+		return runtime.Result{}, updateErr
+	}
+
+	klog.V(2).InfoS("The updateRun is not finished yet", "state", state, "requeueWaitTime", waitTime, "err", err, "updateRun", runObjRef)
+
+	if err != nil {
+		return runtime.Result{}, err
+	}
+	return runtime.Result{Requeue: true, RequeueAfter: waitTime}, nil
+}
+
 // handleDelete handles the deletion of the updateRun object.
 // We delete all the dependent resources, including approvalRequest objects, of the updateRun object.
 func (r *Reconciler) handleDelete(ctx context.Context, updateRun placementv1beta1.UpdateRunObj) (bool, time.Duration, error) {
@@ -277,6 +301,25 @@ func (r *Reconciler) recordUpdateRunFailed(ctx context.Context, updateRun placem
 	return nil
 }
 
+// recordUpdateRunStopped records the progressing condition as stopped in the updateRun status.
+func (r *Reconciler) recordUpdateRunStopped(ctx context.Context, updateRun placementv1beta1.UpdateRunObj) error {
+	updateRunStatus := updateRun.GetUpdateRunStatus()
+	meta.SetStatusCondition(&updateRunStatus.Conditions, metav1.Condition{
+		Type:               string(placementv1beta1.StagedUpdateRunConditionProgressing),
+		Status:             metav1.ConditionFalse,
+		ObservedGeneration: updateRun.GetGeneration(),
+		Reason:             condition.UpdateRunStoppedReason,
+		Message:            "The update run has been stopped",
+	})
+
+	if updateErr := r.Client.Status().Update(ctx, updateRun); updateErr != nil {
+		klog.ErrorS(updateErr, "Failed to update the updateRun status as stopped", "updateRun", klog.KObj(updateRun))
+		// updateErr can be retried.
+		return controller.NewUpdateIgnoreConflictError(updateErr)
+	}
+	return nil
+}
+
 // recordUpdateRunStatus records the updateRun status.
 func (r *Reconciler) recordUpdateRunStatus(ctx context.Context, updateRun placementv1beta1.UpdateRunObj) error {
 	if updateErr := r.Client.Status().Update(ctx, updateRun); updateErr != nil {
@@ -484,3 +527,57 @@ func removeWaitTimeFromUpdateRunStatus(updateRun placementv1beta1.UpdateRunObj)
 		}
 	}
 }
+
+// updateAllStatusConditionsGeneration iterates through all existing conditions in the UpdateRun status
+// and updates their ObservedGeneration field to the current UpdateRun generation.
+func updateAllStatusConditionsGeneration(updateRunStatus *placementv1beta1.UpdateRunStatus, generation int64) {
+	// Update main UpdateRun conditions
+	for i := range updateRunStatus.Conditions {
+		updateRunStatus.Conditions[i].ObservedGeneration = generation
+	}
+
+	// Update stage-level conditions and nested task conditions if it exists
+	for i := range updateRunStatus.StagesStatus {
+		stageStatus := &updateRunStatus.StagesStatus[i]
+
+		// Update stage conditions
+		updateAllStageStatusConditionsGeneration(stageStatus, generation)
+	}
+
+	// Update deletion stage conditions and nested tasks if it exists
+	if updateRunStatus.DeletionStageStatus != nil {
+		deletionStageStatus := updateRunStatus.DeletionStageStatus
+
+		// Update deletion stage conditions
+		updateAllStageStatusConditionsGeneration(deletionStageStatus, generation)
+	}
+}
+
+// updateAllStageStatusConditionsGeneration updates all conditions' ObservedGeneration in the given stage status.
+func updateAllStageStatusConditionsGeneration(stageStatus *placementv1beta1.StageUpdatingStatus, generation int64) {
+	// Update stage conditions
+	for j := range stageStatus.Conditions {
+		stageStatus.Conditions[j].ObservedGeneration = generation
+	}
+
+	// Update before stage task conditions
+	for j := range stageStatus.BeforeStageTaskStatus {
+		for k := range stageStatus.BeforeStageTaskStatus[j].Conditions {
+			stageStatus.BeforeStageTaskStatus[j].Conditions[k].ObservedGeneration = generation
+		}
+	}
+
+	// Update after stage task conditions
+	for j := range stageStatus.AfterStageTaskStatus {
+		for k := range stageStatus.AfterStageTaskStatus[j].Conditions {
+			stageStatus.AfterStageTaskStatus[j].Conditions[k].ObservedGeneration = generation
+		}
+	}
+
+	// Update cluster-level conditions
+	for j := range stageStatus.Clusters {
+		for k := range stageStatus.Clusters[j].Conditions {
+			stageStatus.Clusters[j].Conditions[k].ObservedGeneration = generation
+		}
+	}
+}

pkg/controllers/updaterun/controller_integration_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/prometheus/client_golang/prometheus"
 	prometheusclientmodel "github.com/prometheus/client_model/go"
 
 	corev1 "k8s.io/api/core/v1"
@@ -332,6 +333,26 @@ func generateFailedMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *p
 	}
 }
 
+func generateStoppingMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
+	return &prometheusclientmodel.Metric{
+		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionProgressing),
+			string(metav1.ConditionUnknown), condition.UpdateRunStoppingReason),
+		Gauge: &prometheusclientmodel.Gauge{
+			Value: ptr.To(float64(time.Now().UnixNano()) / 1e9),
+		},
+	}
+}
+
+func generateStoppedMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
+	return &prometheusclientmodel.Metric{
+		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionProgressing),
+			string(metav1.ConditionFalse), condition.UpdateRunStoppedReason),
+		Gauge: &prometheusclientmodel.Gauge{
+			Value: ptr.To(float64(time.Now().UnixNano()) / 1e9),
+		},
+	}
+}
+
 func generateSucceededMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
 	return &prometheusclientmodel.Metric{
 		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionSucceeded),
@@ -342,6 +363,24 @@ func generateSucceededMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun)
 	}
 }
 
+func labelPairsToMap(pairs []*prometheusclientmodel.LabelPair) prometheus.Labels {
+	m := prometheus.Labels{}
+	for _, p := range pairs {
+		m[p.GetName()] = p.GetValue()
+	}
+	return m
+}
+
+func removeMetricFromMetricList(metricList []*prometheusclientmodel.Metric, metricToRemove *prometheusclientmodel.Metric) []*prometheusclientmodel.Metric {
+	var result []*prometheusclientmodel.Metric
+	for _, metric := range metricList {
+		if !cmp.Equal(labelPairsToMap(metric.Label), labelPairsToMap(metricToRemove.Label)) {
+			result = append(result, metric)
+		}
+	}
+	return result
+}
+
 func generateTestClusterStagedUpdateRun() *placementv1beta1.ClusterStagedUpdateRun {
 	return &placementv1beta1.ClusterStagedUpdateRun{
 		ObjectMeta: metav1.ObjectMeta{
@@ -823,3 +862,18 @@ func generateFalseProgressingCondition(obj client.Object, condType any, reason s
 	falseCond.Reason = reason
 	return falseCond
 }
+
+func generateFalseConditionWithReason(obj client.Object, condType any, reason string) metav1.Condition {
+	falseCond := generateFalseCondition(obj, condType)
+	falseCond.Reason = reason
+	return falseCond
+}
+
+func generateProgressingUnknownConditionWithReason(obj client.Object, reason string) metav1.Condition {
+	return metav1.Condition{
+		Status:             metav1.ConditionUnknown,
+		Type:               string(placementv1beta1.StageUpdatingConditionProgressing),
+		ObservedGeneration: obj.GetGeneration(),
+		Reason:             reason,
+	}
+}

pkg/controllers/updaterun/execution.go

@@ -18,7 +18,6 @@ package updaterun
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"reflect"
 	"strconv"
@@ -62,20 +61,13 @@ func (r *Reconciler) execute(
 	updateRun placementv1beta1.UpdateRunObj,
 	updatingStageIndex int,
 	toBeUpdatedBindings, toBeDeletedBindings []placementv1beta1.BindingObj,
-) (finished bool, waitTime time.Duration, err error) {
+) (finished bool, waitTime time.Duration, execErr error) {
 	updateRunStatus := updateRun.GetUpdateRunStatus()
 	var updatingStageStatus *placementv1beta1.StageUpdatingStatus
 
 	// Set up defer function to handle errStagedUpdatedAborted.
 	defer func() {
-		if errors.Is(err, errStagedUpdatedAborted) {
-			if updatingStageStatus != nil {
-				markStageUpdatingFailed(updatingStageStatus, updateRun.GetGeneration(), err.Error())
-			} else {
-				// Handle deletion stage case.
-				markStageUpdatingFailed(updateRunStatus.DeletionStageStatus, updateRun.GetGeneration(), err.Error())
-			}
-		}
+		checkIfErrorStagedUpdateAborted(execErr, updateRun, updatingStageStatus)
 	}()
 
 	// Mark updateRun as progressing if it's not already marked as waiting or stuck.
@@ -84,9 +76,9 @@ func (r *Reconciler) execute(
 	markUpdateRunProgressingIfNotWaitingOrStuck(updateRun)
 	if updatingStageIndex < len(updateRunStatus.StagesStatus) {
 		updatingStageStatus = &updateRunStatus.StagesStatus[updatingStageIndex]
-		approved, err := r.checkBeforeStageTasksStatus(ctx, updatingStageIndex, updateRun)
-		if err != nil {
-			return false, 0, err
+		approved, execErr := r.checkBeforeStageTasksStatus(ctx, updatingStageIndex, updateRun)
+		if execErr != nil {
+			return false, 0, execErr
 		}
 		if !approved {
 			markStageUpdatingWaiting(updatingStageStatus, updateRun.GetGeneration(), "Not all before-stage tasks are completed, waiting for approval")
@@ -97,13 +89,13 @@ func (r *Reconciler) execute(
 		if err != nil {
 			return false, 0, err
 		}
-		waitTime, err = r.executeUpdatingStage(ctx, updateRun, updatingStageIndex, toBeUpdatedBindings, maxConcurrency)
+		waitTime, execErr = r.executeUpdatingStage(ctx, updateRun, updatingStageIndex, toBeUpdatedBindings, maxConcurrency)
 		// The execution has not finished yet.
-		return false, waitTime, err
+		return false, waitTime, execErr
 	}
 	// All the stages have finished, now start the delete stage.
-	finished, err = r.executeDeleteStage(ctx, updateRun, toBeDeletedBindings)
-	return finished, clusterUpdatingWaitTime, err
+	finished, execErr = r.executeDeleteStage(ctx, updateRun, toBeDeletedBindings)
+	return finished, clusterUpdatingWaitTime, execErr
 }
 
 // checkBeforeStageTasksStatus checks if the before stage tasks have finished.
@@ -232,9 +224,7 @@ func (r *Reconciler) executeUpdatingStage(
 				}
 			}
 			markClusterUpdatingStarted(clusterStatus, updateRun.GetGeneration())
-			if finishedClusterCount == 0 {
-				markStageUpdatingStarted(updatingStageStatus, updateRun.GetGeneration())
-			}
+			markStageUpdatingStarted(updatingStageStatus, updateRun.GetGeneration())
 			// Need to continue as we need to process at most maxConcurrency number of clusters in parallel.
 			continue
 		}
@@ -564,6 +554,9 @@ func calculateMaxConcurrencyValue(status *placementv1beta1.UpdateRunStatus, stag
 func aggregateUpdateRunStatus(updateRun placementv1beta1.UpdateRunObj, stageName string, stuckClusterNames []string) {
 	if len(stuckClusterNames) > 0 {
 		markUpdateRunStuck(updateRun, stageName, strings.Join(stuckClusterNames, ", "))
+	} else if updateRun.GetUpdateRunSpec().State == placementv1beta1.StateStop {
+		// If there is no stuck cluster and the update run is stopping, mark the update run as stopped.
+		markUpdateRunStopping(updateRun)
 	} else {
 		// If there is no stuck cluster but some progress has been made, mark the update run as progressing.
 		markUpdateRunProgressing(updateRun)
@@ -672,7 +665,7 @@ func markUpdateRunProgressing(updateRun placementv1beta1.UpdateRunObj) {
 	})
 }
 
-// markUpdateRunProgressingIfNotWaitingOrStuck marks the update run as proegressing in memory if it's not marked as waiting or stuck already.
+// markUpdateRunProgressingIfNotWaitingOrStuck marks the update run as progressing in memory if it's not marked as waiting or stuck already.
 func markUpdateRunProgressingIfNotWaitingOrStuck(updateRun placementv1beta1.UpdateRunObj) {
 	updateRunStatus := updateRun.GetUpdateRunStatus()
 	progressingCond := meta.FindStatusCondition(updateRunStatus.Conditions, string(placementv1beta1.StagedUpdateRunConditionProgressing))