feat: refactor rollout controller for RP (#158)

---------

Signed-off-by: Wantong Jiang <[email protected]>

Author: jwtty

CLAUDE.md

@@ -178,6 +178,9 @@ cmd/memberagent/        # Member agent main and setup
 - Use shared test manifests in `test/integration/manifests/`
 - Run with `make e2e-tests` against 3 Kind clusters
 
+### Test Coding Style
+- Use `want` or `wanted` instead of `expect` or `expected` when creating the desired state
+
 ## Key Patterns
 
 ### Controller Pattern

apis/placement/v1beta1/commons.go

@@ -112,7 +112,7 @@ const (
 	// ParentNamespaceLabel is the label applied to work that contains the namespace of the binding that generates the work.
 	ParentNamespaceLabel = fleetPrefix + "parent-placement-namespace"
 
-	// CRPGenerationAnnotation indicates the generation of the CRP from which an object is derived or last updated.
+	// CRPGenerationAnnotation indicates the generation of the placement from which an object is derived or last updated.
 	// TODO: rename this variable
 	CRPGenerationAnnotation = fleetPrefix + "CRP-generation"
 

cmd/hubagent/workload/setup.go

@@ -227,7 +227,7 @@ func SetupControllers(ctx context.Context, wg *sync.WaitGroup, mgr ctrl.Manager,
 			UncachedReader:          mgr.GetAPIReader(),
 			MaxConcurrentReconciles: int(math.Ceil(float64(opts.MaxFleetSizeSupported)/30) * math.Ceil(float64(opts.MaxConcurrentClusterPlacement)/10)),
 			InformerManager:         dynamicInformerManager,
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManagerForClusterResourcePlacement(mgr); err != nil {
 			klog.ErrorS(err, "Unable to set up rollout controller")
 			return err
 		}

pkg/controllers/clusterresourceplacement/controller.go

@@ -135,7 +135,7 @@ func (r *Reconciler) handleUpdate(ctx context.Context, placementObj fleetv1beta1
 
 	if placementSpec.RevisionHistoryLimit != nil {
 		if revisionLimit <= 0 {
-			err := fmt.Errorf("invalid placement %s: invalid revisionHistoryLimit %d", placementObj.GetName(), revisionLimit)
+			err := fmt.Errorf("invalid placement %s/%s: invalid revisionHistoryLimit %d", placementObj.GetNamespace(), placementObj.GetName(), revisionLimit)
 			klog.ErrorS(controller.NewUnexpectedBehaviorError(err), "Invalid revisionHistoryLimit value and using default value instead", "placement", placementKObj)
 		} else {
 			revisionLimit = *placementSpec.RevisionHistoryLimit
@@ -189,7 +189,8 @@ func (r *Reconciler) handleUpdate(ctx context.Context, placementObj fleetv1beta1
 			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))
+		placementKey := controller.GetObjectKeyFromNamespaceName(placementObj.GetNamespace(), placementObj.GetName())
+		selectedResourceIDs, err = controller.CollectResourceIdentifiersUsingMasterResourceSnapshot(ctx, r.Client, placementKey, latestResourceSnapshot, strconv.Itoa(latestResourceSnapshotIndex))
 		if err != nil {
 			klog.ErrorS(err, "Failed to collect resource identifiers from the resourceSnapshot", "placement", placementKObj, "resourceSnapshot", latestResourceSnapshotKObj)
 			return ctrl.Result{}, err
@@ -459,7 +460,7 @@ func (r *Reconciler) getOrCreateResourceSnapshot(ctx context.Context, placement
 		// check to see all that the master cluster resource snapshot and sub-indexed snapshots belonging to the same group index exists.
 		resourceSnapshotList, err := controller.ListAllResourceSnapshotWithAnIndex(ctx, r.Client, latestResourceSnapshot.GetLabels()[fleetv1beta1.ResourceIndexLabel], placement.GetName(), placement.GetNamespace())
 		if err != nil {
-			klog.ErrorS(err, "Failed to list the latest group resourceSnapshots associated with the placement", "placement", placement.GetName())
+			klog.ErrorS(err, "Failed to list the latest group resourceSnapshots associated with the placement", "placement", placementKObj)
 			return ctrl.Result{}, nil, controller.NewAPIServerError(true, err)
 		}
 		if len(resourceSnapshotList.GetResourceSnapshotObjs()) == numberOfSnapshots {
@@ -1054,7 +1055,7 @@ func (r *Reconciler) determineRolloutStateForPlacementWithExternalRolloutStrateg
 ) (bool, error) {
 	if len(selected) == 0 {
 		// This should not happen as we already checked in setPlacementStatus.
-		err := controller.NewUnexpectedBehaviorError(fmt.Errorf("selected cluster list is empty for placement %s when checking per-cluster rollout state", placementObj.GetName()))
+		err := controller.NewUnexpectedBehaviorError(fmt.Errorf("selected cluster list is empty for placement %s/%s when checking per-cluster rollout state", placementObj.GetNamespace(), placementObj.GetName()))
 		klog.ErrorS(err, "Should not happen: selected cluster list is empty in determineRolloutStateForPlacementWithExternalRolloutStrategy()")
 		return false, err
 	}
@@ -1120,10 +1121,7 @@ func (r *Reconciler) determineRolloutStateForPlacementWithExternalRolloutStrateg
 	} else {
 		placementStatus.ObservedResourceIndex = observedResourceIndex
 		// Construct placement key for the resource collection function
-		placementKey := placementObj.GetName()
-		if placementObj.GetNamespace() != "" {
-			placementKey = placementObj.GetNamespace() + "/" + placementObj.GetName()
-		}
+		placementKey := controller.GetObjectKeyFromNamespaceName(placementObj.GetNamespace(), placementObj.GetName())
 		selectedResources, err := controller.CollectResourceIdentifiersFromResourceSnapshot(ctx, r.Client, placementKey, observedResourceIndex)
 		if err != nil {
 			klog.ErrorS(err, "Failed to collect resource identifiers from resourceSnapshot", "placement", klog.KObj(placementObj), "resourceSnapshotIndex", observedResourceIndex)

pkg/controllers/clusterschedulingpolicysnapshot/controller.go

@@ -42,8 +42,6 @@ type Reconciler struct {
 
 // Reconcile triggers a single placement reconcile round when scheduling policy has changed.
 func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	var snapshot fleetv1beta1.PolicySnapshotObj
-	var err error
 	snapshotKRef := klog.KRef(req.Namespace, req.Name)
 
 	startTime := time.Now()
@@ -53,6 +51,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		klog.V(2).InfoS("Reconciliation ends", "policySnapshot", snapshotKRef, "latency", latency)
 	}()
 
+	var snapshot fleetv1beta1.PolicySnapshotObj
+	var err error
 	if req.Namespace == "" {
 		// ClusterSchedulingPolicySnapshot (cluster-scoped)
 		var clusterSchedulingPolicySnapshot fleetv1beta1.ClusterSchedulingPolicySnapshot
@@ -71,8 +71,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		snapshot = &schedulingPolicySnapshot
 	}
 
-	placementName, exist := snapshot.GetLabels()[fleetv1beta1.PlacementTrackingLabel]
-	if !exist || len(placementName) == 0 {
+	placementName := snapshot.GetLabels()[fleetv1beta1.PlacementTrackingLabel]
+	if len(placementName) == 0 {
 		err := fmt.Errorf("label %s is missing or has empty value", fleetv1beta1.PlacementTrackingLabel)
 		klog.ErrorS(err, "Failed to enqueue placement due to invalid schedulingPolicySnapshot", "policySnapshot", snapshotKRef)
 		return ctrl.Result{}, controller.NewUnexpectedBehaviorError(err)