fix, rebase, and fix tests

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"
@@ -312,40 +313,40 @@ func generateWaitingMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *
 	}
 }
 
-func generateStoppingMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
+func generateStuckMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
 	return &prometheusclientmodel.Metric{
 		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionProgressing),
-			string(metav1.ConditionFalse), condition.UpdateRunStoppingReason),
+			string(metav1.ConditionFalse), condition.UpdateRunStuckReason),
 		Gauge: &prometheusclientmodel.Gauge{
 			Value: ptr.To(float64(time.Now().UnixNano()) / 1e9),
 		},
 	}
 }
 
-func generateStoppedMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
+func generateFailedMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
 	return &prometheusclientmodel.Metric{
-		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionProgressing),
-			string(metav1.ConditionFalse), condition.UpdateRunStoppedReason),
+		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionSucceeded),
+			string(metav1.ConditionFalse), condition.UpdateRunFailedReason),
 		Gauge: &prometheusclientmodel.Gauge{
 			Value: ptr.To(float64(time.Now().UnixNano()) / 1e9),
 		},
 	}
 }
 
-func generateStuckMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
+func generateStoppingMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
 	return &prometheusclientmodel.Metric{
 		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionProgressing),
-			string(metav1.ConditionFalse), condition.UpdateRunStuckReason),
+			string(metav1.ConditionUnknown), condition.UpdateRunStoppingReason),
 		Gauge: &prometheusclientmodel.Gauge{
 			Value: ptr.To(float64(time.Now().UnixNano()) / 1e9),
 		},
 	}
 }
 
-func generateFailedMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
+func generateStoppedMetric(updateRun *placementv1beta1.ClusterStagedUpdateRun) *prometheusclientmodel.Metric {
 	return &prometheusclientmodel.Metric{
-		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionSucceeded),
-			string(metav1.ConditionFalse), condition.UpdateRunFailedReason),
+		Label: generateMetricsLabels(updateRun, string(placementv1beta1.StagedUpdateRunConditionProgressing),
+			string(metav1.ConditionFalse), condition.UpdateRunStoppedReason),
 		Gauge: &prometheusclientmodel.Gauge{
 			Value: ptr.To(float64(time.Now().UnixNano()) / 1e9),
 		},
@@ -362,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{
@@ -849,3 +868,12 @@ func generateFalseConditionWithReason(obj client.Object, condType any, reason st
 	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,
+	}
+}