test again

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

Author: Ryan Zhang

pkg/controllers/clusterresourceplacement/controller.go

@@ -49,7 +49,7 @@ import (
 )
 
 // The max size of an object in k8s is 1.5MB because of ETCD limit https://etcd.io/docs/v3.3/dev-guide/limit/.
-// We choose 800KB as the soft limit for all the selected resources within one clusterResourceSnapshot object because of this test in k8s which checks
+// We choose 800KB as the soft limit for all the selected resources within one resourceSnapshot object because of this test in k8s which checks
 // if object size is greater than 1MB https://github.com/kubernetes/kubernetes/blob/db1990f48b92d603f469c1c89e2ad36da1b74846/test/integration/master/synthetic_master_test.go#L337
 var resourceSnapshotResourceSizeLimit = 800 * (1 << 10) // 800KB
 
@@ -186,19 +186,19 @@ func (r *Reconciler) handleUpdate(ctx context.Context, placementObj fleetv1beta1
 		// Rebuild the seletedResourceIDs using the latestResourceSnapshot.
 		latestResourceSnapshotIndex, err := labels.ExtractResourceIndexFromResourceSnapshot(latestResourceSnapshot)
 		if err != nil {
-			klog.ErrorS(err, "Failed to extract the resource index from the clusterResourceSnapshot", "placement", placementKObj, "clusterResourceSnapshot", latestResourceSnapshotKObj)
+			klog.ErrorS(err, "Failed to extract the resource index from the resourceSnapshot", "placement", placementKObj, "resourceSnapshot", latestResourceSnapshotKObj)
 			return ctrl.Result{}, controller.NewUnexpectedBehaviorError(err)
 		}
 		selectedResourceIDs, err = controller.CollectResourceIdentifiersUsingMasterResourceSnapshot(ctx, r.Client, placementObj.GetName(), latestResourceSnapshot, strconv.Itoa(latestResourceSnapshotIndex))
 		if err != nil {
-			klog.ErrorS(err, "Failed to collect resource identifiers from the clusterResourceSnapshot", "placement", placementKObj, "clusterResourceSnapshot", latestResourceSnapshotKObj)
+			klog.ErrorS(err, "Failed to collect resource identifiers from the resourceSnapshot", "placement", placementKObj, "resourceSnapshot", latestResourceSnapshotKObj)
 			return ctrl.Result{}, err
 		}
-		klog.V(2).InfoS("Fetched the selected resources from the lastestResourceSnapshot", "placement", placementKObj, "clusterResourceSnapshot", latestResourceSnapshotKObj, "generation", placementObj.GetGeneration())
+		klog.V(2).InfoS("Fetched the selected resources from the lastestResourceSnapshot", "placement", placementKObj, "resourceSnapshot", latestResourceSnapshotKObj, "generation", placementObj.GetGeneration())
 	}
 
-	// isClusterScheduled is to indicate whether we need to requeue the placement request to track the rollout status.
-	isClusterScheduled, err := r.setPlacementStatus(ctx, placementObj, selectedResourceIDs, latestSchedulingPolicySnapshot, latestResourceSnapshot)
+	// isScheduleFullfilled is to indicate whether we need to requeue the placement request to track the rollout status.
+	isScheduleFullfilled, err := r.setPlacementStatus(ctx, placementObj, selectedResourceIDs, latestSchedulingPolicySnapshot, latestResourceSnapshot)
 	if err != nil {
 		return ctrl.Result{}, err
 	}
@@ -222,8 +222,8 @@ func (r *Reconciler) handleUpdate(ctx context.Context, placementObj fleetv1beta1
 
 	// Rollout is considered to be completed when all the expected condition types are set to the
 	// True status.
-	if isRolloutCompleted(placementObj) {
-		if !isRolloutCompleted(oldPlacement) {
+	if isPlacementRolloutCompleted(placementObj) {
+		if !isPlacementRolloutCompleted(oldPlacement) {
 			klog.V(2).InfoS("Placement has finished the rollout process and reached the desired status", "placement", placementKObj, "generation", placementObj.GetGeneration())
 			r.Recorder.Event(placementObj, corev1.EventTypeNormal, "PlacementRolloutCompleted", "Placement has finished the rollout process and reached the desired status")
 		}
@@ -236,7 +236,7 @@ func (r *Reconciler) handleUpdate(ctx context.Context, placementObj fleetv1beta1
 		return ctrl.Result{}, nil
 	}
 
-	if !isClusterScheduled {
+	if !isScheduleFullfilled {
 		// Note:
 		// If the scheduledCondition is failed, it means the placement requirement cannot be satisfied fully. For example,
 		// pickN deployment requires 5 clusters and scheduler schedules the resources on 3 clusters. And the appliedCondition
@@ -352,12 +352,6 @@ func (r *Reconciler) lookupLatestSchedulingPolicySnapshot(ctx context.Context, p
 		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])
@@ -533,20 +527,20 @@ func (r *Reconciler) getOrCreateResourceSnapshot(ctx context.Context, placement
 	if latestResourceSnapshot != nil {
 		latestResourceSnapshotHash, err = annotations.ParseResourceGroupHashFromAnnotation(latestResourceSnapshot)
 		if err != nil {
-			klog.ErrorS(err, "Failed to get the ResourceGroupHashAnnotation", "clusterResourceSnapshot", klog.KObj(latestResourceSnapshot))
+			klog.ErrorS(err, "Failed to get the ResourceGroupHashAnnotation", "resourceSnapshot", klog.KObj(latestResourceSnapshot))
 			return ctrl.Result{}, nil, controller.NewUnexpectedBehaviorError(err)
 		}
 		numberOfSnapshots, err = annotations.ExtractNumberOfResourceSnapshotsFromResourceSnapshot(latestResourceSnapshot)
 		if err != nil {
-			klog.ErrorS(err, "Failed to get the NumberOfResourceSnapshotsAnnotation", "clusterResourceSnapshot", klog.KObj(latestResourceSnapshot))
+			klog.ErrorS(err, "Failed to get the NumberOfResourceSnapshotsAnnotation", "resourceSnapshot", klog.KObj(latestResourceSnapshot))
 			return ctrl.Result{}, nil, controller.NewUnexpectedBehaviorError(err)
 		}
 	}
 
 	shouldCreateNewMasterResourceSnapshot := true
 	// This index indicates the selected resource in the split selectedResourceList, if this index is zero we start
-	// from creating the master clusterResourceSnapshot if it's greater than zero it means that the master clusterResourceSnapshot
-	// got created but not all sub-indexed clusterResourceSnapshots have been created yet. It covers the corner case where the
+	// from creating the master resourceSnapshot if it's greater than zero it means that the master resourceSnapshot
+	// got created but not all sub-indexed resourceSnapshots have been created yet. It covers the corner case where the
 	// controller crashes in the middle.
 	resourceSnapshotStartIndex := 0
 	if latestResourceSnapshot != nil && latestResourceSnapshotHash == resourceHash {
@@ -560,7 +554,7 @@ func (r *Reconciler) getOrCreateResourceSnapshot(ctx context.Context, placement
 			return ctrl.Result{}, nil, controller.NewAPIServerError(true, err)
 		}
 		if len(resourceSnapshotList.GetResourceSnapshotObjs()) == numberOfSnapshots {
-			klog.V(2).InfoS("ClusterResourceSnapshots have not changed", "placement", placementKObj, "clusterResourceSnapshot", klog.KObj(latestResourceSnapshot))
+			klog.V(2).InfoS("resourceSnapshots have not changed", "placement", placementKObj, "resourceSnapshot", klog.KObj(latestResourceSnapshot))
 			return ctrl.Result{}, latestResourceSnapshot, nil
 		}
 		// we should not create a new master cluster resource snapshot.
@@ -593,10 +587,10 @@ func (r *Reconciler) getOrCreateResourceSnapshot(ctx context.Context, placement
 		labels[fleetv1beta1.IsLatestSnapshotLabel] = strconv.FormatBool(false)
 		latestResourceSnapshot.SetLabels(labels)
 		if err := r.Client.Update(ctx, latestResourceSnapshot); err != nil {
-			klog.ErrorS(err, "Failed to set the isLatestSnapshot label to false", "clusterResourceSnapshot", klog.KObj(latestResourceSnapshot))
+			klog.ErrorS(err, "Failed to set the isLatestSnapshot label to false", "resourceSnapshot", klog.KObj(latestResourceSnapshot))
 			return ctrl.Result{}, nil, controller.NewUpdateIgnoreConflictError(err)
 		}
-		klog.V(2).InfoS("Marked the existing clusterResourceSnapshot as inactive", "placement", placementKObj, "clusterResourceSnapshot", klog.KObj(latestResourceSnapshot))
+		klog.V(2).InfoS("Marked the existing resourceSnapshot as inactive", "placement", placementKObj, "resourceSnapshot", klog.KObj(latestResourceSnapshot))
 	}
 
 	// only delete redundant resource snapshots and increment the latest resource snapshot index if new master resource snapshot is to be created.
@@ -645,32 +639,33 @@ func (r *Reconciler) shouldCreateNewResourceSnapshotNow(ctx context.Context, lat
 	nextResourceSnapshotCandidateDetectionTime, err := annotations.ExtractNextResourceSnapshotCandidateDetectionTimeFromResourceSnapshot(latestResourceSnapshot)
 	if nextResourceSnapshotCandidateDetectionTime.IsZero() || err != nil {
 		if err != nil {
-			klog.ErrorS(controller.NewUnexpectedBehaviorError(err), "Failed to get the NextResourceSnapshotCandidateDetectionTimeAnnotation", "clusterResourceSnapshot", snapshotKObj)
+			klog.ErrorS(controller.NewUnexpectedBehaviorError(err), "Failed to get the NextResourceSnapshotCandidateDetectionTimeAnnotation", "resourceSnapshot", snapshotKObj)
 		}
 		// If the annotation is not set, set next resource snapshot candidate detection time is now.
 		if latestResourceSnapshot.GetAnnotations() == nil {
 			latestResourceSnapshot.SetAnnotations(make(map[string]string))
 		}
 		latestResourceSnapshot.GetAnnotations()[fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation] = now.Format(time.RFC3339)
 		if err := r.Client.Update(ctx, latestResourceSnapshot); err != nil {
-			klog.ErrorS(err, "Failed to update the NextResourceSnapshotCandidateDetectionTime annotation", "clusterResourceSnapshot", snapshotKObj)
+			klog.ErrorS(err, "Failed to update the NextResourceSnapshotCandidateDetectionTime annotation", "resourceSnapshot", snapshotKObj)
 			return ctrl.Result{}, controller.NewUpdateIgnoreConflictError(err)
 		}
 		nextResourceSnapshotCandidateDetectionTime = now
-		klog.V(2).InfoS("Updated the NextResourceSnapshotCandidateDetectionTime annotation", "clusterResourceSnapshot", snapshotKObj, "nextResourceSnapshotCandidateDetectionTimeAnnotation", now.Format(time.RFC3339))
+		klog.V(2).InfoS("Updated the NextResourceSnapshotCandidateDetectionTime annotation", "resourceSnapshot", snapshotKObj, "nextResourceSnapshotCandidateDetectionTimeAnnotation", now.Format(time.RFC3339))
 	}
 	nextCreationTime := fleettime.MaxTime(nextResourceSnapshotCandidateDetectionTime.Add(r.ResourceChangesCollectionDuration), latestResourceSnapshot.GetCreationTimestamp().Add(r.ResourceSnapshotCreationMinimumInterval))
 	if now.Before(nextCreationTime) {
 		// If the next resource snapshot creation time is not reached, we requeue the request to avoid too frequent update.
 		klog.V(2).InfoS("Delaying the new resourceSnapshot creation",
-			"clusterResourceSnapshot", snapshotKObj, "nextCreationTime", nextCreationTime, "latestResourceSnapshotCreationTime", latestResourceSnapshot.GetCreationTimestamp(),
+			"resourceSnapshot", snapshotKObj, "nextCreationTime", nextCreationTime, "latestResourceSnapshotCreationTime", latestResourceSnapshot.GetCreationTimestamp(),
 			"resourceSnapshotCreationMinimumInterval", r.ResourceSnapshotCreationMinimumInterval, "resourceChangesCollectionDuration", r.ResourceChangesCollectionDuration,
 			"afterDuration", nextCreationTime.Sub(now))
 		return ctrl.Result{Requeue: true, RequeueAfter: nextCreationTime.Sub(now)}, nil
 	}
 	return ctrl.Result{}, nil
 }
 
+// TODO: move this to library package
 // buildMasterResourceSnapshot builds and returns the master resource snapshot for the latest resource snapshot index and selected resources.
 func BuildMasterResourceSnapshot(latestResourceSnapshotIndex, resourceSnapshotCount, envelopeObjCount int, placementName, placementNamespace, resourceHash string, selectedResources []fleetv1beta1.ResourceContent) fleetv1beta1.ResourceSnapshotObj {
 	labels := map[string]string{
@@ -710,6 +705,7 @@ func BuildMasterResourceSnapshot(latestResourceSnapshotIndex, resourceSnapshotCo
 	}
 }
 
+// TODO: move this to library package
 // BuildSubIndexResourceSnapshot builds and returns the sub index resource snapshot for both cluster-scoped and namespace-scoped placements.
 // Returns a ClusterResourceSnapshot for cluster-scoped placements (empty namespace) or ResourceSnapshot for namespace-scoped placements.
 func BuildSubIndexResourceSnapshot(latestResourceSnapshotIndex, resourceSnapshotSubIndex int, placementName, placementNamespace string, selectedResources []fleetv1beta1.ResourceContent) fleetv1beta1.ResourceSnapshotObj {
@@ -1037,7 +1033,8 @@ func (r *Reconciler) setPlacementStatus(
 		// conditions (any condition type other than Scheduled).
 
 		// Note that the scheduled condition has been set earlier in this method.
-		placementStatus.Conditions = []metav1.Condition{scheduledCondition}
+		placementStatus.Conditions = []metav1.Condition{}
+		placementObj.SetConditions(scheduledCondition)
 		return isPolicySelectingNoClusters(placementObj.GetPlacementSpec().Policy), nil
 	}
 
@@ -1250,7 +1247,11 @@ func buildResourcePlacementStatusMap(placementObj fleetv1beta1.PlacementObj) map
 	return m
 }
 
-func isRolloutCompleted(placementObj fleetv1beta1.PlacementObj) bool {
+// TODO: make this work for RP too
+// isPlacementRolloutCompleted checks if the placement rollout is completed which means:
+// 1. Placement Scheduled condition is true.
+// 2. All expected placement conditions are true depends on what type of policy placementObj has.
+func isPlacementRolloutCompleted(placementObj fleetv1beta1.PlacementObj) bool {
 	if !condition.IsConditionStatusTrue(placementObj.GetCondition(string(fleetv1beta1.ClusterResourcePlacementScheduledConditionType)), placementObj.GetGeneration()) {
 		return false
 	}

pkg/controllers/clusterresourceplacement/controller_test.go

@@ -3569,7 +3569,7 @@ func TestIsRolloutComplete(t *testing.T) {
 					Conditions: tc.conditions,
 				},
 			}
-			got := isRolloutCompleted(crp)
+			got := isPlacementRolloutCompleted(crp)
 			if got != tc.want {
 				t.Errorf("isRolloutCompleted() got %v, want %v", got, tc.want)
 			}

pkg/controllers/clusterresourceplacement/placement_status.go

@@ -113,6 +113,7 @@ func determineExpectedPlacementAndResourcePlacementStatusCondType(placementObj f
 
 // appendScheduledResourcePlacementStatuses appends the resource placement statuses for the
 // scheduled clusters to the list of all resource placement statuses.
+// it returns the updated list of resource placement statuses.
 func (r *Reconciler) appendScheduledResourcePlacementStatuses(
 	ctx context.Context,
 	allRPS []fleetv1beta1.ResourcePlacementStatus,

pkg/utils/controller/policy_snapshot_resolver.go

@@ -63,22 +63,31 @@ func DeletePolicySnapshots(ctx context.Context, k8Client client.Client, placemen
 // For cluster-scoped placements, it returns a ClusterSchedulingPolicySnapshot.
 // For namespaced placements, it returns a SchedulingPolicySnapshot.
 func BuildPolicySnapshot(placementObj fleetv1beta1.PlacementObj, policySnapshotIndex int, policyHash string) fleetv1beta1.PolicySnapshotObj {
+	var snapshot fleetv1beta1.PolicySnapshotObj
 	labels := map[string]string{
 		fleetv1beta1.PlacementTrackingLabel: placementObj.GetName(),
 		fleetv1beta1.IsLatestSnapshotLabel:  strconv.FormatBool(true),
 		fleetv1beta1.PolicyIndexLabel:       strconv.Itoa(policySnapshotIndex),
 	}
-
 	annotations := map[string]string{
 		fleetv1beta1.CRPGenerationAnnotation: strconv.FormatInt(placementObj.GetGeneration(), 10),
 	}
+	// Add NumberOfClusters annotation if placement is selectN type
+	if spec := placementObj.GetPlacementSpec(); spec.Policy != nil &&
+		spec.Policy.PlacementType == fleetv1beta1.PickNPlacementType &&
+		spec.Policy.NumberOfClusters != nil {
+		annotations[fleetv1beta1.NumberOfClustersAnnotation] = strconv.Itoa(int(*spec.Policy.NumberOfClusters))
+	}
 
+	spec := fleetv1beta1.SchedulingPolicySnapshotSpec{
+		Policy:     placementObj.GetPlacementSpec().Policy,
+		PolicyHash: []byte(policyHash),
+	},
 	// Set the name following the convention: {PlacementName}-{index}
 	name := fmt.Sprintf(fleetv1beta1.PolicySnapshotNameFmt, placementObj.GetName(), policySnapshotIndex)
-
 	if placementObj.GetNamespace() != "" {
 		// This is a namespaced ResourcePlacement - create SchedulingPolicySnapshot
-		snapshot := &fleetv1beta1.SchedulingPolicySnapshot{
+		snapshot = &fleetv1beta1.SchedulingPolicySnapshot{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:        name,
 				Namespace:   placementObj.GetNamespace(),
@@ -90,10 +99,9 @@ func BuildPolicySnapshot(placementObj fleetv1beta1.PlacementObj, policySnapshotI
 				PolicyHash: []byte(policyHash),
 			},
 		}
-		return snapshot
 	} else {
 		// This is a cluster-scoped ClusterResourcePlacement - create ClusterSchedulingPolicySnapshot
-		snapshot := &fleetv1beta1.ClusterSchedulingPolicySnapshot{
+		snapshot = &fleetv1beta1.ClusterSchedulingPolicySnapshot{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:        name,
 				Labels:      labels,
@@ -104,16 +112,8 @@ func BuildPolicySnapshot(placementObj fleetv1beta1.PlacementObj, policySnapshotI
 				PolicyHash: []byte(policyHash),
 			},
 		}
-
-		// Add NumberOfClusters annotation if placement is selectN type
-		if spec := placementObj.GetPlacementSpec(); spec.Policy != nil &&
-			spec.Policy.PlacementType == fleetv1beta1.PickNPlacementType &&
-			spec.Policy.NumberOfClusters != nil {
-			snapshot.Annotations[fleetv1beta1.NumberOfClustersAnnotation] = strconv.Itoa(int(*spec.Policy.NumberOfClusters))
-		}
-
-		return snapshot
 	}
+	return snapshot
 }
 
 // FetchLatestPolicySnapshot fetches the latest policy snapshot for a given placement.