Implement rollout strategy

Signed-off-by: melserngawy <melserng@redhat.com>

Author: serngawy

pkg/work/helper/helpers.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/klog/v2"
 
 	clusterlister "open-cluster-management.io/api/client/cluster/listers/cluster/v1beta1"
+	clusterv1alpha1 "open-cluster-management.io/api/cluster/v1alpha1"
 	clusterv1beta1 "open-cluster-management.io/api/cluster/v1beta1"
 	workapiv1 "open-cluster-management.io/api/work/v1"
 )
@@ -483,7 +484,15 @@ func (pdl PlacementDecisionGetter) List(selector labels.Selector, namespace stri
 // Get added and deleted clusters names
 func GetClusters(client clusterlister.PlacementDecisionLister, placement *clusterv1beta1.Placement,
 	existingClusters sets.Set[string]) (sets.Set[string], sets.Set[string], error) {
-	pdtracker := clusterv1beta1.NewPlacementDecisionClustersTracker(placement, PlacementDecisionGetter{Client: client}, existingClusters)
+	existingClusterGroups := make(map[clusterv1beta1.GroupKey]sets.Set[string])
+	pdtracker := clusterv1beta1.NewPlacementDecisionClustersTracker(placement, PlacementDecisionGetter{Client: client}, existingClusters, existingClusterGroups)
 
 	return pdtracker.Get()
 }
+
+func GetRollOutHandler(client clusterlister.PlacementDecisionLister, placement *clusterv1beta1.Placement, existingClusters sets.Set[string]) (*clusterv1alpha1.RolloutHandler, error) {
+	existingClusterGroups := make(map[clusterv1beta1.GroupKey]sets.Set[string])
+	pdtracker := clusterv1beta1.NewPlacementDecisionClustersTracker(placement, PlacementDecisionGetter{Client: client}, existingClusters, existingClusterGroups)
+
+	return clusterv1alpha1.NewRolloutHandler(pdtracker)
+}

pkg/work/hub/controllers/manifestworkreplicasetcontroller/manifestworkreplicaset_controller.go

@@ -37,6 +37,10 @@ const (
 	// TODO move this to the api repo
 	ManifestWorkReplicaSetControllerNameLabelKey = "work.open-cluster-management.io/manifestworkreplicaset"
 
+	// ManifestWorkReplicaSetPlacementNameLabelKey is the label key on manifestwork to ref to the Placement that select
+	// the managedCluster on the manifestWorkReplicaSet's PlacementRef.
+	ManifestWorkReplicaSetPlacementNameLabelKey = "work.open-cluster-management.io/PlacementName"
+
 	// ManifestWorkReplicaSetFinalizer is the name of the finalizer added to ManifestWorkReplicaSet. It is used to ensure
 	// related manifestworks is deleted
 	ManifestWorkReplicaSetFinalizer = "work.open-cluster-management.io/manifest-work-cleanup"
@@ -180,3 +184,19 @@ func listManifestWorksByManifestWorkReplicaSet(mwrs *workapiv1alpha1.ManifestWor
 	selector := labels.NewSelector().Add(*req)
 	return manifestWorkLister.List(selector)
 }
+
+func listManifestWorksByMWRSetPlacementRef(mwrs *workapiv1alpha1.ManifestWorkReplicaSet, placementName string,
+	manifestWorkLister worklisterv1.ManifestWorkLister) ([]*workapiv1.ManifestWork, error) {
+	reqMWRSet, err := labels.NewRequirement(ManifestWorkReplicaSetControllerNameLabelKey, selection.Equals, []string{manifestWorkReplicaSetKey(mwrs)})
+	if err != nil {
+		return nil, err
+	}
+
+	reqPlacementRef, err := labels.NewRequirement(ManifestWorkReplicaSetPlacementNameLabelKey, selection.Equals, []string{placementName})
+	if err != nil {
+		return nil, err
+	}
+
+	selector := labels.NewSelector().Add(*reqMWRSet, *reqPlacementRef)
+	return manifestWorkLister.List(selector)
+}

pkg/work/hub/controllers/manifestworkreplicasetcontroller/manifestworkreplicaset_deploy_reconcile.go

@@ -12,7 +12,8 @@ import (
 
 	clusterlister "open-cluster-management.io/api/client/cluster/listers/cluster/v1beta1"
 	worklisterv1 "open-cluster-management.io/api/client/work/listers/work/v1"
-	clusterv1beta1 "open-cluster-management.io/api/cluster/v1beta1"
+	//clusterv1beta1 "open-cluster-management.io/api/cluster/v1beta1"
+	clusterv1alpha1 "open-cluster-management.io/api/cluster/v1alpha1"
 	"open-cluster-management.io/api/utils/work/v1/workapplier"
 	workv1 "open-cluster-management.io/api/work/v1"
 	workapiv1alpha1 "open-cluster-management.io/api/work/v1alpha1"
@@ -31,8 +32,17 @@ type deployReconciler struct {
 func (d *deployReconciler) reconcile(ctx context.Context, mwrSet *workapiv1alpha1.ManifestWorkReplicaSet,
 ) (*workapiv1alpha1.ManifestWorkReplicaSet, reconcileState, error) {
 	// Manifestwork create/update/delete logic.
-	var placements []*clusterv1beta1.Placement
+
+	//var placements []*clusterv1beta1.Placement
+	//existingClusters := make(map[string]sets.String)
+	// placementName as key and RolloutResult is the value
+	//placementsRolloutResults := make(map[string]clusterv1alpha1.RolloutResult)
+	var errs []error
+	count := 0
+
+	// Getting the placements and the created ManifestWorks related to each placement
 	for _, placementRef := range mwrSet.Spec.PlacementRefs {
+		existingClusters, addedClusters, deletedClusters := sets.New[string](), sets.New[string](), sets.New[string]()
 		placement, err := d.placementLister.Placements(mwrSet.Namespace).Get(placementRef.Name)
 		if errors.IsNotFound(err) {
 			apimeta.SetStatusCondition(&mwrSet.Status.Conditions, GetPlacementDecisionVerified(workapiv1alpha1.ReasonPlacementDecisionNotFound, ""))
@@ -41,74 +51,101 @@ func (d *deployReconciler) reconcile(ctx context.Context, mwrSet *workapiv1alpha
 		if err != nil {
 			return mwrSet, reconcileContinue, fmt.Errorf("failed get placement %w", err)
 		}
-		placements = append(placements, placement)
-	}
 
-	manifestWorks, err := listManifestWorksByManifestWorkReplicaSet(mwrSet, d.manifestWorkLister)
-	if err != nil {
-		return mwrSet, reconcileContinue, err
-	}
+		//placements = append(placements, placement)
+		manifestWorks, err := listManifestWorksByMWRSetPlacementRef(mwrSet, placementRef.Name, d.manifestWorkLister)
+		if err != nil {
+			return mwrSet, reconcileContinue, err
+		}
 
-	var errs []error
-	addedClusters, deletedClusters, existingClusters := sets.New[string](), sets.New[string](), sets.New[string]()
-	for _, mw := range manifestWorks {
-		existingClusters.Insert(mw.Namespace)
-	}
+		for _, mw := range manifestWorks {
+			existingClusters.Insert(mw.Namespace)
+		}
 
-	for _, placement := range placements {
-		added, deleted, err := helper.GetClusters(d.placeDecisionLister, placement, existingClusters)
+		rolloutHandler, err := helper.GetRollOutHandler(d.placeDecisionLister, placement, existingClusters)
 		if err != nil {
 			apimeta.SetStatusCondition(&mwrSet.Status.Conditions, GetPlacementDecisionVerified(workapiv1alpha1.ReasonNotAsExpected, ""))
 
 			return mwrSet, reconcileContinue, utilerrors.NewAggregate(errs)
 		}
 
-		addedClusters = addedClusters.Union(added)
-		deletedClusters = deletedClusters.Union(deleted)
-	}
-
-	// Create manifestWork for added clusters
-	for cls := range addedClusters {
-		mw, err := CreateManifestWork(mwrSet, cls)
-		if err != nil {
-			errs = append(errs, err)
-			continue
+		// The clusterRolloutStatusFunc need different signature only clusterName as input param not helping
+		_, rolloutResult, err := rolloutHandler.GetRolloutCluster(placementRef.RolloutStrategy, d.clusterRolloutStatusFunc)
+		//placementsRolloutResults[placementRef.Name] = rolloutResult
+
+		// Creating ManifestWork that has the ToApply status
+		for cls, rolloutStatue := range rolloutResult.ClustersToRollout {
+			if rolloutStatue.Status == clusterv1alpha1.ToApply {
+				mw, err := CreateManifestWork(mwrSet, cls, placementRef.Name)
+				if err != nil {
+					errs = append(errs, err)
+					continue
+				}
+
+				_, err = d.workApplier.Apply(ctx, mw)
+				if err != nil {
+					errs = append(errs, err)
+				}
+			}
 		}
 
-		_, err = d.workApplier.Apply(ctx, mw)
+		// checking the Added and deleted clusters.
+		// How to check if the added cluster is part of the rolloutResult ?
+		// Does the RolloutResult already consider the added clusters ?
+		// Better to have the deleted/added clusters within the rolloutResult
+		addedClusters, deletedClusters, err = helper.GetClusters(d.placeDecisionLister, placement, existingClusters)
 		if err != nil {
-			errs = append(errs, err)
+			apimeta.SetStatusCondition(&mwrSet.Status.Conditions, GetPlacementDecisionVerified(workapiv1alpha1.ReasonNotAsExpected, ""))
+
+			return mwrSet, reconcileContinue, utilerrors.NewAggregate(errs)
 		}
-	}
 
-	// Update manifestWorks in case there are changes at ManifestWork or ManifestWorkReplicaSet
-	for cls := range existingClusters {
-		// Delete manifestWork for deleted clusters
-		if deletedClusters.Has(cls) {
-			err = d.workApplier.Delete(ctx, cls, mwrSet.Name)
+		// Create manifestWork for added clusters
+		for cls := range addedClusters {
+			mw, err := CreateManifestWork(mwrSet, cls, placementRef.Name)
 			if err != nil {
 				errs = append(errs, err)
+				continue
 			}
-			continue
-		}
 
-		mw, err := CreateManifestWork(mwrSet, cls)
-		if err != nil {
-			errs = append(errs, err)
-			continue
+			_, err = d.workApplier.Apply(ctx, mw)
+			if err != nil {
+				errs = append(errs, err)
+			}
 		}
 
-		_, err = d.workApplier.Apply(ctx, mw)
-		if err != nil {
-			errs = append(errs, err)
+		// How to rollOut updated manifestWork ? that can be handled in the clusterRolloutStatusFunc
+		// Update manifestWorks in case there are changes at ManifestWork or ManifestWorkReplicaSet
+		for cls := range existingClusters {
+			// Delete manifestWork for deleted clusters
+			if deletedClusters.Has(cls) {
+				err = d.workApplier.Delete(ctx, cls, mwrSet.Name)
+				if err != nil {
+					errs = append(errs, err)
+				}
+				continue
+			}
+
+			mw, err := CreateManifestWork(mwrSet, cls, placementRef.Name)
+			if err != nil {
+				errs = append(errs, err)
+				continue
+			}
+
+			_, err = d.workApplier.Apply(ctx, mw)
+			if err != nil {
+				errs = append(errs, err)
+			}
 		}
+		count = count + len(rolloutResult.ClustersToRollout)
 	}
 
 	// Set the Summary
 	if mwrSet.Status.Summary == (workapiv1alpha1.ManifestWorkReplicaSetSummary{}) {
 		mwrSet.Status.Summary = workapiv1alpha1.ManifestWorkReplicaSetSummary{}
 	}
-	total := len(existingClusters) - len(deletedClusters) + len(addedClusters)
+	//total := len(existingClusters) - len(deletedClusters) + len(addedClusters)
+	total := count
 	if total < 0 {
 		total = 0
 	}
@@ -127,6 +164,13 @@ func (d *deployReconciler) reconcile(ctx context.Context, mwrSet *workapiv1alpha
 	return mwrSet, reconcileContinue, utilerrors.NewAggregate(errs)
 }
 
+// TODO: Implement clusterRolloutStatusFunc logic.
+func (d *deployReconciler) clusterRolloutStatusFunc(clusterName string) clusterv1alpha1.ClusterRolloutStatus {
+	clsRolloutStatus := clusterv1alpha1.ClusterRolloutStatus{}
+
+	return clsRolloutStatus
+}
+
 // GetManifestworkApplied return only True status if there all clusters have manifests applied as expected
 func GetManifestworkApplied(reason string, message string) metav1.Condition {
 	if reason == workapiv1alpha1.ReasonAsExpected {
@@ -156,7 +200,7 @@ func getCondition(conditionType string, reason string, message string, status me
 	}
 }
 
-func CreateManifestWork(mwrSet *workapiv1alpha1.ManifestWorkReplicaSet, clusterNS string) (*workv1.ManifestWork, error) {
+func CreateManifestWork(mwrSet *workapiv1alpha1.ManifestWorkReplicaSet, clusterNS string, placementRefName string) (*workv1.ManifestWork, error) {
 	if clusterNS == "" {
 		return nil, fmt.Errorf("invalid cluster namespace")
 	}
@@ -165,7 +209,8 @@ func CreateManifestWork(mwrSet *workapiv1alpha1.ManifestWorkReplicaSet, clusterN
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      mwrSet.Name,
 			Namespace: clusterNS,
-			Labels:    map[string]string{ManifestWorkReplicaSetControllerNameLabelKey: manifestWorkReplicaSetKey(mwrSet)},
+			Labels: map[string]string{ManifestWorkReplicaSetControllerNameLabelKey: manifestWorkReplicaSetKey(mwrSet),
+				ManifestWorkReplicaSetPlacementNameLabelKey: placementRefName},
 		},
 		Spec: mwrSet.Spec.ManifestWorkTemplate}, nil
 }