fix the code that UT catches

Signed-off-by: Ryan Zhang <[email protected]>

Author: Ryan Zhang

pkg/controllers/clusterresourceplacement/controller.go

@@ -279,9 +279,8 @@ func (r *Reconciler) getOrCreateSchedulingPolicySnapshot(ctx context.Context, pl
 		return nil, controller.NewUnexpectedBehaviorError(err)
 	}
 
-	// Use the unified helper function to fetch the latest policy snapshot
-	placementKey := types.NamespacedName{Name: placementObj.GetName(), Namespace: placementObj.GetNamespace()}
-	latestPolicySnapshot, latestPolicySnapshotIndex, err := controller.FetchLatestPolicySnapshot(ctx, r.Client, placementKey)
+	// latestPolicySnapshotIndex should be -1 when there is no snapshot.
+	latestPolicySnapshot, latestPolicySnapshotIndex, err := r.lookupLatestSchedulingPolicySnapshot(ctx, placementObj)
 	if err != nil {
 		return nil, err
 	}
@@ -338,8 +337,102 @@ func (r *Reconciler) getOrCreateSchedulingPolicySnapshot(ctx context.Context, pl
 	return newPolicySnapshot, nil
 }
 
+// lookupLatestSchedulingPolicySnapshot finds the latest snapshots and its policy index.
+// There will be only one active policy snapshot if exists.
+// It first checks whether there is an active policy snapshot.
+// If not, it finds the one whose policyIndex label is the largest.
+// The policy index will always start from 0.
+// Return error when 1) cannot list the snapshots 2) there are more than one active policy snapshots 3) snapshot has the
+// invalid label value.
+// 2 & 3 should never happen.
+func (r *Reconciler) lookupLatestSchedulingPolicySnapshot(ctx context.Context, placement fleetv1beta1.PlacementObj) (fleetv1beta1.PolicySnapshotObj, int, error) {
+	placementKey := types.NamespacedName{Name: placement.GetName(), Namespace: placement.GetNamespace()}
+	snapshotList, err := controller.FetchLatestPolicySnapshot(ctx, r.Client, placementKey)
+	if err != nil {
+		return nil, -1, err
+	}
+	placementKObj := klog.KObj(placement)
+	if err != nil {
+		klog.ErrorS(err, "Failed to list active schedulingPolicySnapshots", "placement", placementKObj)
+		// Placement controller needs a scheduling policy snapshot watcher to enqueue the placement request.
+		// So the snapshots should be read from cache.
+		return nil, -1, controller.NewAPIServerError(true, err)
+	}
+	policySnapshotItems := snapshotList.GetPolicySnapshotObjs()
+	if len(policySnapshotItems) == 1 {
+		policyIndex, err := labels.ParsePolicyIndexFromLabel(policySnapshotItems[0])
+		if err != nil {
+			klog.ErrorS(err, "Failed to parse the policy index label", "placement", placementKObj, "policySnapshot", klog.KObj(policySnapshotItems[0]))
+			return nil, -1, controller.NewUnexpectedBehaviorError(err)
+		}
+		return policySnapshotItems[0], policyIndex, nil
+	} else if len(policySnapshotItems) > 1 {
+		// It means there are multiple active snapshots and should never happen.
+		err := fmt.Errorf("there are %d active schedulingPolicySnapshots owned by placement %v", len(policySnapshotItems), placementKey)
+		klog.ErrorS(err, "Invalid schedulingPolicySnapshots", "placement", placementKObj)
+		return nil, -1, controller.NewUnexpectedBehaviorError(err)
+	}
+	// When there are no active snapshots, find the one who has the largest policy index.
+	// It should be rare only when placement controller is crashed before creating the new active snapshot.
+	sortedList, err := r.listSortedSchedulingPolicySnapshots(ctx, placement)
+	if err != nil {
+		return nil, -1, err
+	}
+
+	if len(sortedList.GetPolicySnapshotObjs()) == 0 {
+		// The policy index of the first snapshot will start from 0.
+		return nil, -1, nil
+	}
+	latestSnapshot := sortedList.GetPolicySnapshotObjs()[len(sortedList.GetPolicySnapshotObjs())-1]
+	policyIndex, err := labels.ParsePolicyIndexFromLabel(latestSnapshot)
+	if err != nil {
+		klog.ErrorS(err, "Failed to parse the policy index label", "placement", placementKObj, "policySnapshot", klog.KObj(latestSnapshot))
+		return nil, -1, controller.NewUnexpectedBehaviorError(err)
+	}
+	return latestSnapshot, policyIndex, nil
+}
+
+// listSortedSchedulingPolicySnapshots returns the policy snapshots sorted by the policy index.
+// Now works with both cluster-scoped and namespaced policy snapshots using interface types.
+func (r *Reconciler) listSortedSchedulingPolicySnapshots(ctx context.Context, placementObj fleetv1beta1.PlacementObj) (fleetv1beta1.PolicySnapshotList, error) {
+	placementKey := types.NamespacedName{
+		Namespace: placementObj.GetNamespace(),
+		Name:      placementObj.GetName(),
+	}
+
+	snapshotList, err := controller.ListPolicySnapshots(ctx, r.Client, placementKey)
+	if err != nil {
+		klog.ErrorS(err, "Failed to list all policySnapshots", "placement", klog.KObj(placementObj))
+		// CRP controller needs a scheduling policy snapshot watcher to enqueue the CRP request.
+		// So the snapshots should be read from cache.
+		return nil, controller.NewAPIServerError(true, err)
+	}
+
+	items := snapshotList.GetPolicySnapshotObjs()
+	var errs []error
+	sort.Slice(items, func(i, j int) bool {
+		ii, err := labels.ParsePolicyIndexFromLabel(items[i])
+		if err != nil {
+			klog.ErrorS(err, "Failed to parse the policy index label", "placement", klog.KObj(placementObj), "policySnapshot", klog.KObj(items[i]))
+			errs = append(errs, err)
+		}
+		ji, err := labels.ParsePolicyIndexFromLabel(items[j])
+		if err != nil {
+			klog.ErrorS(err, "Failed to parse the policy index label", "placement", klog.KObj(placementObj), "policySnapshot", klog.KObj(items[j]))
+			errs = append(errs, err)
+		}
+		return ii < ji
+	})
+
+	if len(errs) > 0 {
+		return nil, controller.NewUnexpectedBehaviorError(utilerrors.NewAggregate(errs))
+	}
+
+	return snapshotList, nil
+}
+
 func (r *Reconciler) deleteRedundantSchedulingPolicySnapshots(ctx context.Context, placementObj fleetv1beta1.PlacementObj, revisionHistoryLimit int) error {
-	sortedList, err := r.listSortedClusterSchedulingPolicySnapshots(ctx, placementObj)
+	sortedList, err := r.listSortedSchedulingPolicySnapshots(ctx, placementObj)
 	if err != nil {
 		return err
 	}
@@ -756,45 +849,6 @@ func (r *Reconciler) ensureLatestResourceSnapshot(ctx context.Context, latest fl
 	return nil
 }
 
-// listSortedClusterSchedulingPolicySnapshots returns the policy snapshots sorted by the policy index.
-// Now works with both cluster-scoped and namespaced policy snapshots using interface types.
-func (r *Reconciler) listSortedClusterSchedulingPolicySnapshots(ctx context.Context, placementObj fleetv1beta1.PlacementObj) (fleetv1beta1.PolicySnapshotList, error) {
-	placementKey := types.NamespacedName{
-		Namespace: placementObj.GetNamespace(),
-		Name:      placementObj.GetName(),
-	}
-
-	snapshotList, err := controller.ListPolicySnapshots(ctx, r.Client, placementKey)
-	if err != nil {
-		klog.ErrorS(err, "Failed to list all policySnapshots", "placement", klog.KObj(placementObj))
-		// CRP controller needs a scheduling policy snapshot watcher to enqueue the CRP request.
-		// So the snapshots should be read from cache.
-		return nil, controller.NewAPIServerError(true, err)
-	}
-
-	items := snapshotList.GetPolicySnapshotObjs()
-	var errs []error
-	sort.Slice(items, func(i, j int) bool {
-		ii, err := labels.ParsePolicyIndexFromLabel(items[i])
-		if err != nil {
-			klog.ErrorS(err, "Failed to parse the policy index label", "placement", klog.KObj(placementObj), "policySnapshot", klog.KObj(items[i]))
-			errs = append(errs, err)
-		}
-		ji, err := labels.ParsePolicyIndexFromLabel(items[j])
-		if err != nil {
-			klog.ErrorS(err, "Failed to parse the policy index label", "placement", klog.KObj(placementObj), "policySnapshot", klog.KObj(items[j]))
-			errs = append(errs, err)
-		}
-		return ii < ji
-	})
-
-	if len(errs) > 0 {
-		return nil, controller.NewUnexpectedBehaviorError(utilerrors.NewAggregate(errs))
-	}
-
-	return snapshotList, nil
-}
-
 // lookupLatestResourceSnapshot finds the latest snapshots and.
 // There will be only one active resource snapshot if exists.
 // It first checks whether there is an active resource snapshot.

pkg/controllers/clusterresourceplacement/controller_test.go

@@ -121,8 +121,9 @@ func clusterResourcePlacementForTest() *fleetv1beta1.ClusterResourcePlacement {
 
 func TestGetOrCreateClusterSchedulingPolicySnapshot(t *testing.T) {
 	testPolicy := placementPolicyForTest()
-	testPolicy.NumberOfClusters = nil
-	jsonBytes, err := json.Marshal(testPolicy)
+	testPolicyHash := testPolicy.DeepCopy()
+	testPolicyHash.NumberOfClusters = nil
+	jsonBytes, err := json.Marshal(testPolicyHash)
 	if err != nil {
 		t.Fatalf("failed to create the policy hash: %v", err)
 	}
@@ -749,10 +750,17 @@ func TestGetOrCreateClusterSchedulingPolicySnapshot(t *testing.T) {
 			if err != nil {
 				t.Fatalf("failed to getOrCreateClusterSchedulingPolicySnapshot: %v", err)
 			}
+
+			// Convert interface to concrete type for comparison
+			gotSnapshot, ok := got.(*fleetv1beta1.ClusterSchedulingPolicySnapshot)
+			if !ok {
+				t.Fatalf("expected *ClusterSchedulingPolicySnapshot, got %T", got)
+			}
+
 			options := []cmp.Option{
 				cmpopts.IgnoreFields(metav1.ObjectMeta{}, "ResourceVersion"),
 			}
-			if diff := cmp.Diff(tc.wantPolicySnapshots[tc.wantLatestSnapshotIndex], got, options...); diff != "" {
+			if diff := cmp.Diff(tc.wantPolicySnapshots[tc.wantLatestSnapshotIndex], *gotSnapshot, options...); diff != "" {
 				t.Errorf("getOrCreateClusterSchedulingPolicySnapshot() mismatch (-want, +got):\n%s", diff)
 			}
 			clusterPolicySnapshotList := &fleetv1beta1.ClusterSchedulingPolicySnapshotList{}
@@ -2785,7 +2793,7 @@ func TestGetOrCreateClusterResourceSnapshot_failure(t *testing.T) {
 		},
 		{
 			// Should never hit this case unless there is a bug in the controller or customers manually modify the clusterResourceSnapshot.
-			name: "existing active policy snapshot does not have hash annotation",
+			name: "existing active resource snapshot does not have hash annotation",
 			resourceSnapshots: []fleetv1beta1.ClusterResourceSnapshot{
 				{
 					ObjectMeta: metav1.ObjectMeta{

pkg/utils/controller/policy_snapshot_resolver.go

@@ -119,41 +119,31 @@ func BuildPolicySnapshot(placementObj fleetv1beta1.PlacementObj, policySnapshotI
 // FetchLatestPolicySnapshot fetches the latest policy snapshot for a given placement.
 // For cluster-scoped placements, it fetches ClusterSchedulingPolicySnapshot.
 // For namespaced placements, it fetches SchedulingPolicySnapshot.
-func FetchLatestPolicySnapshot(ctx context.Context, k8Client client.Reader, placementKey types.NamespacedName) (fleetv1beta1.PolicySnapshotObj, int, error) {
-	policySnapshotList, err := ListPolicySnapshots(ctx, k8Client, placementKey)
-	if err != nil {
-		return nil, -1, err
-	}
+func FetchLatestPolicySnapshot(ctx context.Context, k8Client client.Reader, placementKey types.NamespacedName) (fleetv1beta1.PolicySnapshotList, error) {
+	namespace := placementKey.Namespace
+	name := placementKey.Name
 
-	items := policySnapshotList.GetPolicySnapshotObjs()
-	if len(items) == 0 {
-		klog.V(2).InfoS("No policySnapshots found for the placement", "placement", placementKey)
-		return nil, -1, nil
-	}
+	var policySnapshotList fleetv1beta1.PolicySnapshotList
+	var listOptions []client.ListOption
+	listOptions = append(listOptions, client.MatchingLabels{
+		fleetv1beta1.PlacementTrackingLabel: name,
+		fleetv1beta1.IsLatestSnapshotLabel:  strconv.FormatBool(true),
+	})
 
-	// Find the latest policy snapshot by comparing policy index
-	var latestPolicySnapshot fleetv1beta1.PolicySnapshotObj
-	latestIndex := -1
-
-	for _, policySnapshot := range items {
-		labels := policySnapshot.GetLabels()
-		if indexStr, exists := labels[fleetv1beta1.PolicyIndexLabel]; exists {
-			if index, err := strconv.Atoi(indexStr); err == nil {
-				if index > latestIndex {
-					latestIndex = index
-					latestPolicySnapshot = policySnapshot
-				}
-			}
-		}
+	if namespace != "" {
+		// This is a namespaced SchedulingPolicySnapshotList
+		policySnapshotList = &fleetv1beta1.SchedulingPolicySnapshotList{}
+		listOptions = append(listOptions, client.InNamespace(namespace))
+	} else {
+		// This is a cluster-scoped ClusterSchedulingPolicySnapshotList
+		policySnapshotList = &fleetv1beta1.ClusterSchedulingPolicySnapshotList{}
 	}
 
-	if latestPolicySnapshot == nil {
-		klog.V(2).InfoS("No valid policySnapshot found for the placement", "placement", placementKey)
-		return nil, -1, nil
+	if err := k8Client.List(ctx, policySnapshotList, listOptions...); err != nil {
+		klog.ErrorS(err, "Failed to list the policySnapshots associated with the placement", "placement", placementKey)
+		return nil, err
 	}
-
-	klog.V(2).InfoS("Found the latest policySnapshot", "placement", placementKey, "policySnapshot", klog.KObj(latestPolicySnapshot))
-	return latestPolicySnapshot, latestIndex, nil
+	return policySnapshotList, nil
 }
 
 // ListPolicySnapshots lists all policy snapshots associated with a placement key.

pkg/utils/controller/resource_snapshot_resolver.go

@@ -51,9 +51,9 @@ func FetchLatestMasterResourceSnapshot(ctx context.Context, k8Client client.Read
 			break
 		}
 	}
-	// It is possible that no master resourceSnapshot is found, e.g., when the new resourceSnapshot is created but not yet marked as the latest.
+	// The master is always the first to be created in the resourcegroup so it should be found.
 	if masterResourceSnapshot == nil {
-		return nil, fmt.Errorf("no masterResourceSnapshot found for the placement %v", placementKey)
+		return nil, NewUnexpectedBehaviorError(fmt.Errorf("no masterResourceSnapshot found for the placement %v", placementKey))
 	}
 	klog.V(2).InfoS("Found the latest associated masterResourceSnapshot", "placement", placementKey, "masterResourceSnapshot", klog.KObj(masterResourceSnapshot))
 	return masterResourceSnapshot, nil