fix: fix updateRun stuck with reportDiff mode (#81)

Author: jwtty

pkg/controllers/clusterresourceplacementeviction/controller_intergration_test.go

@@ -22,7 +22,6 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"github.com/prometheus/client_golang/prometheus"
 	prometheusclientmodel "github.com/prometheus/client_model/go"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -31,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 
 	placementv1beta1 "github.com/kubefleet-dev/kubefleet/apis/placement/v1beta1"
 	"github.com/kubefleet-dev/kubefleet/pkg/utils/condition"
@@ -55,12 +55,8 @@ const (
 var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 	crpName := fmt.Sprintf(crpNameTemplate, GinkgoParallelProcess())
 	evictionName := fmt.Sprintf(evictionNameTemplate, GinkgoParallelProcess())
-	var customRegistry *prometheus.Registry
 
 	BeforeEach(func() {
-		// Create a test registry
-		customRegistry = prometheus.NewRegistry()
-		Expect(customRegistry.Register(metrics.FleetEvictionStatus)).Should(Succeed())
 		// Reset metrics before each test
 		metrics.FleetEvictionStatus.Reset()
 		// emit incomplete eviction metric to simulate eviction failed once.
@@ -72,7 +68,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 		ensureAllBindingsAreRemoved(crpName)
 		ensureEvictionRemoved(evictionName)
 		ensureCRPRemoved(crpName)
-		Expect(customRegistry.Unregister(metrics.FleetEvictionStatus)).Should(BeTrue())
+		metrics.FleetEvictionStatus.Reset()
 	})
 
 	It("Invalid Eviction Blocked - emit complete metric with isValid=false, isComplete=true", func() {
@@ -90,7 +86,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 		})
 
 		By("Ensure eviction complete metric was emitted", func() {
-			checkEvictionCompleteMetric(customRegistry, "false", "true")
+			checkEvictionCompleteMetric("false", "true")
 		})
 	})
 
@@ -169,7 +165,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 		})
 
 		By("Ensure eviction complete metric was emitted", func() {
-			checkEvictionCompleteMetric(customRegistry, "true", "true")
+			checkEvictionCompleteMetric("true", "true")
 		})
 	})
 
@@ -269,7 +265,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 		})
 
 		By("Ensure eviction complete metric was emitted", func() {
-			checkEvictionCompleteMetric(customRegistry, "true", "true")
+			checkEvictionCompleteMetric("true", "true")
 		})
 	})
 
@@ -348,7 +344,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 		})
 
 		By("Ensure eviction complete metric was emitted", func() {
-			checkEvictionCompleteMetric(customRegistry, "true", "true")
+			checkEvictionCompleteMetric("true", "true")
 		})
 	})
 
@@ -478,7 +474,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 		})
 
 		By("Ensure eviction complete metric was emitted", func() {
-			checkEvictionCompleteMetric(customRegistry, "true", "true")
+			checkEvictionCompleteMetric("true", "true")
 		})
 	})
 
@@ -527,7 +523,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 		})
 
 		By("Ensure eviction complete metric was emitted", func() {
-			checkEvictionCompleteMetric(customRegistry, "false", "true")
+			checkEvictionCompleteMetric("false", "true")
 		})
 	})
 
@@ -623,7 +619,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 			})
 
 			By("Ensure eviction complete metric was emitted", func() {
-				checkEvictionCompleteMetric(customRegistry, "true", "true")
+				checkEvictionCompleteMetric("true", "true")
 			})
 		})
 
@@ -660,7 +656,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 			})
 
 			By("Ensure eviction complete metric was emitted", func() {
-				checkEvictionCompleteMetric(customRegistry, "true", "true")
+				checkEvictionCompleteMetric("true", "true")
 			})
 		})
 
@@ -697,7 +693,7 @@ var _ = Describe("Test ClusterResourcePlacementEviction Controller", func() {
 			})
 
 			By("Ensure eviction complete metric was emitted", func() {
-				checkEvictionCompleteMetric(customRegistry, "true", "true")
+				checkEvictionCompleteMetric("true", "true")
 			})
 		})
 	})
@@ -818,8 +814,8 @@ func ensureAllBindingsAreRemoved(crpName string) {
 	}
 }
 
-func checkEvictionCompleteMetric(registry *prometheus.Registry, isValid, isComplete string) {
-	metricFamilies, err := registry.Gather()
+func checkEvictionCompleteMetric(isValid, isComplete string) {
+	metricFamilies, err := ctrlmetrics.Registry.Gather()
 	Expect(err).Should(Succeed())
 	var evictionCompleteMetrics []*prometheusclientmodel.Metric
 	for _, mf := range metricFamilies {

pkg/controllers/clusterresourceplacementeviction/controller_test.go

@@ -26,7 +26,6 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
-	"github.com/prometheus/client_golang/prometheus"
 	prometheusclientmodel "github.com/prometheus/client_model/go"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -36,6 +35,7 @@ import (
 	controllerruntime "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 
 	placementv1beta1 "github.com/kubefleet-dev/kubefleet/apis/placement/v1beta1"
 	"github.com/kubefleet-dev/kubefleet/pkg/utils/condition"
@@ -1492,12 +1492,6 @@ func TestReconcileForIncompleteEvictionMetric(t *testing.T) {
 	isValid := "unknown"
 	isComplete := "false"
 
-	// Create a test registry
-	customRegistry := prometheus.NewRegistry()
-	if err := customRegistry.Register(metrics.FleetEvictionStatus); err != nil {
-		t.Errorf("Failed to register metric: %v", err)
-	}
-
 	// Reset metrics before each test
 	metrics.FleetEvictionStatus.Reset()
 
@@ -1513,7 +1507,7 @@ func TestReconcileForIncompleteEvictionMetric(t *testing.T) {
 		t.Errorf("reconcile should have failed")
 	}
 
-	metricFamilies, err := customRegistry.Gather()
+	metricFamilies, err := ctrlmetrics.Registry.Gather()
 	if err != nil {
 		t.Errorf("error gathering metrics: %v", err)
 	}

pkg/controllers/updaterun/controller_integration_test.go

@@ -18,16 +18,17 @@ package updaterun
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strconv"
 	"time"
 
 	"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"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -38,6 +39,7 @@ import (
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 
 	clusterv1beta1 "github.com/kubefleet-dev/kubefleet/apis/cluster/v1beta1"
 	placementv1alpha1 "github.com/kubefleet-dev/kubefleet/apis/placement/v1alpha1"
@@ -54,13 +56,6 @@ const (
 	// duration is the time to duration to check for Consistently
 	duration = time.Second * 20
 
-	// numTargetClusters is the number of scheduled clusters
-	numTargetClusters = 10
-	// numUnscheduledClusters is the number of unscheduled clusters
-	numUnscheduledClusters = 3
-	// numberOfClustersAnnotation is the number of clusters in the test latest policy snapshot
-	numberOfClustersAnnotation = numTargetClusters
-
 	// testResourceSnapshotIndex is the index of the test resource snapshot
 	testResourceSnapshotIndex = "0"
 )
@@ -77,8 +72,6 @@ var (
 	testUpdateStrategyName   string
 	testCROName              string
 	updateRunNamespacedName  types.NamespacedName
-	testNamespace            []byte
-	customRegistry           *prometheus.Registry
 )
 
 var _ = Describe("Test the clusterStagedUpdateRun controller", func() {
@@ -90,15 +83,13 @@ var _ = Describe("Test the clusterStagedUpdateRun controller", func() {
 		testUpdateStrategyName = "updatestrategy-" + utils.RandStr()
 		testCROName = "cro-" + utils.RandStr()
 		updateRunNamespacedName = types.NamespacedName{Name: testUpdateRunName}
-
-		customRegistry = initializeUpdateRunMetricsRegistry()
 	})
 
 	AfterEach(func() {
 		By("Checking the update run status metrics are removed")
 		// No metrics are emitted as all are removed after updateRun is deleted.
-		validateUpdateRunMetricsEmitted(customRegistry)
-		unregisterUpdateRunMetrics(customRegistry)
+		validateUpdateRunMetricsEmitted()
+		resetUpdateRunMetrics()
 	})
 
 	Context("Test reconciling a clusterStagedUpdateRun", func() {
@@ -242,23 +233,14 @@ var _ = Describe("Test the clusterStagedUpdateRun controller", func() {
 	})
 })
 
-func initializeUpdateRunMetricsRegistry() *prometheus.Registry {
-	// Create a test registry
-	customRegistry := prometheus.NewRegistry()
-	Expect(customRegistry.Register(metrics.FleetUpdateRunStatusLastTimestampSeconds)).Should(Succeed())
-	// Reset metrics before each test
+func resetUpdateRunMetrics() {
 	metrics.FleetUpdateRunStatusLastTimestampSeconds.Reset()
-	return customRegistry
-}
-
-func unregisterUpdateRunMetrics(registry *prometheus.Registry) {
-	Expect(registry.Unregister(metrics.FleetUpdateRunStatusLastTimestampSeconds)).Should(BeTrue())
 }
 
 // validateUpdateRunMetricsEmitted validates the update run status metrics are emitted and are emitted in the correct order.
-func validateUpdateRunMetricsEmitted(registry *prometheus.Registry, wantMetrics ...*prometheusclientmodel.Metric) {
+func validateUpdateRunMetricsEmitted(wantMetrics ...*prometheusclientmodel.Metric) {
 	Eventually(func() error {
-		metricFamilies, err := registry.Gather()
+		metricFamilies, err := ctrlmetrics.Registry.Gather()
 		if err != nil {
 			return fmt.Errorf("failed to gather metrics: %w", err)
 		}
@@ -388,7 +370,7 @@ func generateTestClusterResourcePlacement() *placementv1beta1.ClusterResourcePla
 	}
 }
 
-func generateTestClusterSchedulingPolicySnapshot(idx int) *placementv1beta1.ClusterSchedulingPolicySnapshot {
+func generateTestClusterSchedulingPolicySnapshot(idx, numberOfClustersAnnotation int) *placementv1beta1.ClusterSchedulingPolicySnapshot {
 	return &placementv1beta1.ClusterSchedulingPolicySnapshot{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: fmt.Sprintf(placementv1beta1.PolicySnapshotNameFmt, testCRPName, idx),
@@ -410,6 +392,51 @@ func generateTestClusterSchedulingPolicySnapshot(idx int) *placementv1beta1.Clus
 	}
 }
 
+func generateTestClusterResourceBindingsAndClusters(policySnapshotIndex int) ([]*placementv1beta1.ClusterResourceBinding, []*clusterv1beta1.MemberCluster, []*clusterv1beta1.MemberCluster) {
+	numTargetClusters := 10
+	numUnscheduledClusters := 3
+	policySnapshotName := fmt.Sprintf(placementv1beta1.PolicySnapshotNameFmt, testCRPName, policySnapshotIndex)
+	resourceBindings := make([]*placementv1beta1.ClusterResourceBinding, numTargetClusters+numUnscheduledClusters)
+	targetClusters := make([]*clusterv1beta1.MemberCluster, numTargetClusters)
+	for i := range targetClusters {
+		// split the clusters into 2 regions
+		region := regionEastus
+		if i%2 == 0 {
+			region = regionWestus
+		}
+		// reserse the order of the clusters by index
+		targetClusters[i] = generateTestMemberCluster(numTargetClusters-1-i, "cluster-"+strconv.Itoa(i), map[string]string{"group": "prod", "region": region})
+		resourceBindings[i] = generateTestClusterResourceBinding(policySnapshotName, targetClusters[i].Name, placementv1beta1.BindingStateScheduled)
+	}
+
+	unscheduledClusters := make([]*clusterv1beta1.MemberCluster, numUnscheduledClusters)
+	for i := range unscheduledClusters {
+		unscheduledClusters[i] = generateTestMemberCluster(i, "unscheduled-cluster-"+strconv.Itoa(i), map[string]string{"group": "staging"})
+		// update the policySnapshot name so that these clusters are considered to-be-deleted
+		resourceBindings[numTargetClusters+i] = generateTestClusterResourceBinding(policySnapshotName+"a", unscheduledClusters[i].Name, placementv1beta1.BindingStateUnscheduled)
+	}
+	return resourceBindings, targetClusters, unscheduledClusters
+}
+
+func generateSmallTestClusterResourceBindingsAndClusters(policySnapshotIndex int) ([]*placementv1beta1.ClusterResourceBinding, []*clusterv1beta1.MemberCluster, []*clusterv1beta1.MemberCluster) {
+	numTargetClusters := 3
+	policySnapshotName := fmt.Sprintf(placementv1beta1.PolicySnapshotNameFmt, testCRPName, policySnapshotIndex)
+	resourceBindings := make([]*placementv1beta1.ClusterResourceBinding, numTargetClusters)
+	targetClusters := make([]*clusterv1beta1.MemberCluster, numTargetClusters)
+	for i := range targetClusters {
+		// split the clusters into 2 regions
+		region := regionEastus
+		if i%2 == 0 {
+			region = regionWestus
+		}
+		// reserse the order of the clusters by index
+		targetClusters[i] = generateTestMemberCluster(numTargetClusters-1-i, "cluster-"+strconv.Itoa(i), map[string]string{"group": "prod", "region": region})
+		resourceBindings[i] = generateTestClusterResourceBinding(policySnapshotName, targetClusters[i].Name, placementv1beta1.BindingStateScheduled)
+	}
+	unscheduledClusters := make([]*clusterv1beta1.MemberCluster, 0)
+	return resourceBindings, targetClusters, unscheduledClusters
+}
+
 func generateTestClusterResourceBinding(policySnapshotName, targetCluster string, state placementv1beta1.BindingState) *placementv1beta1.ClusterResourceBinding {
 	binding := &placementv1beta1.ClusterResourceBinding{
 		ObjectMeta: metav1.ObjectMeta{
@@ -504,7 +531,36 @@ func generateTestClusterStagedUpdateStrategy() *placementv1beta1.ClusterStagedUp
 	}
 }
 
+func generateTestClusterStagedUpdateStrategyWithSingleStage(afterStageTasks []placementv1beta1.AfterStageTask) *placementv1beta1.ClusterStagedUpdateStrategy {
+	return &placementv1beta1.ClusterStagedUpdateStrategy{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: testUpdateStrategyName,
+		},
+		Spec: placementv1beta1.StagedUpdateStrategySpec{
+			Stages: []placementv1beta1.StageConfig{
+				{
+					Name:            "stage1",
+					LabelSelector:   &metav1.LabelSelector{}, // Select all clusters.
+					AfterStageTasks: afterStageTasks,
+				},
+			},
+		},
+	}
+}
+
 func generateTestClusterResourceSnapshot() *placementv1beta1.ClusterResourceSnapshot {
+	testNamespace, _ := json.Marshal(corev1.Namespace{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Namespace",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-namespace",
+			Labels: map[string]string{
+				"fleet.azure.com/name": "test-namespace",
+			},
+		},
+	})
 	clusterResourceSnapshot := &placementv1beta1.ClusterResourceSnapshot{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: testResourceSnapshotName,
@@ -671,6 +727,8 @@ func generateTrueCondition(obj client.Object, condType any) metav1.Condition {
 		switch cond {
 		case placementv1beta1.ResourceBindingAvailable:
 			reason = condition.AvailableReason
+		case placementv1beta1.ResourceBindingDiffReported:
+			reason = condition.DiffReportedStatusTrueReason
 		}
 		typeStr = string(cond)
 	}
@@ -713,6 +771,8 @@ func generateFalseCondition(obj client.Object, condType any) metav1.Condition {
 		switch cond {
 		case placementv1beta1.ResourceBindingApplied:
 			reason = condition.ApplyFailedReason
+		case placementv1beta1.ResourceBindingDiffReported:
+			reason = condition.DiffReportedStatusFalseReason
 		}
 		typeStr = string(cond)
 	}

pkg/controllers/updaterun/execution.go

@@ -460,13 +460,15 @@ func checkClusterUpdateResult(
 	updateRun *placementv1beta1.ClusterStagedUpdateRun,
 ) (bool, error) {
 	availCond := binding.GetCondition(string(placementv1beta1.ResourceBindingAvailable))
-	if condition.IsConditionStatusTrue(availCond, binding.Generation) {
-		// The resource updated on the cluster is available.
+	diffReportCondition := binding.GetCondition(string(placementv1beta1.ResourceBindingDiffReported))
+	if condition.IsConditionStatusTrue(availCond, binding.Generation) ||
+		condition.IsConditionStatusTrue(diffReportCondition, binding.Generation) {
+		// The resource updated on the cluster is available or diff is successfully reported.
 		klog.InfoS("The cluster has been updated", "cluster", clusterStatus.ClusterName, "stage", updatingStage.StageName, "clusterStagedUpdateRun", klog.KObj(updateRun))
 		markClusterUpdatingSucceeded(clusterStatus, updateRun.Generation)
 		return true, nil
 	}
-	if bindingutils.HasBindingFailed(binding) {
+	if bindingutils.HasBindingFailed(binding) || condition.IsConditionStatusFalse(diffReportCondition, binding.Generation) {
 		// We have no way to know if the failed condition is recoverable or not so we just let it run
 		klog.InfoS("The cluster updating encountered an error", "cluster", clusterStatus.ClusterName, "stage", updatingStage.StageName, "clusterStagedUpdateRun", klog.KObj(updateRun))
 		// TODO(wantjian): identify some non-recoverable error and mark the cluster updating as failed