Implement rollout strategy

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

pkg/work/helper/helpers.go

@@ -19,12 +19,17 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/dynamic"
 	"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"
 )
 
@@ -467,3 +472,27 @@ func BuildResourceMeta(
 	resourceMeta.Resource = mapping.Resource.Resource
 	return resourceMeta, mapping.Resource, err
 }
+
+type PlacementDecisionGetter struct {
+	Client clusterlister.PlacementDecisionLister
+}
+
+func (pdl PlacementDecisionGetter) List(selector labels.Selector, namespace string) ([]*clusterv1beta1.PlacementDecision, error) {
+	return pdl.Client.PlacementDecisions(namespace).List(selector)
+}
+
+// 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) {
+	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_test.go

@@ -20,7 +20,7 @@ import (
 
 func TestDeployReconcileAsExpected(t *testing.T) {
 	mwrSet := helpertest.CreateTestManifestWorkReplicaSet("mwrSet-test", "default", "place-test")
-	mw, _ := CreateManifestWork(mwrSet, "cls1")
+	mw, _ := CreateManifestWork(mwrSet, "cls1", "plc1")
 	fWorkClient := fakeworkclient.NewSimpleClientset(mwrSet, mw)
 	workInformerFactory := workinformers.NewSharedInformerFactoryWithOptions(fWorkClient, 1*time.Second)
 

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

@@ -18,7 +18,7 @@ import (
 // Test finalize reconcile
 func TestFinalizeReconcile(t *testing.T) {
 	mwrSetTest := helpertest.CreateTestManifestWorkReplicaSet("mwrSet-test", "default", "place-test")
-	mw, _ := CreateManifestWork(mwrSetTest, "cluster1")
+	mw, _ := CreateManifestWork(mwrSetTest, "cluster1", "plc1")
 	fakeClient := fakeclient.NewSimpleClientset(mwrSetTest, mw)
 	manifestWorkInformerFactory := workinformers.NewSharedInformerFactoryWithOptions(fakeClient, 1*time.Second)
 	mwLister := manifestWorkInformerFactory.Work().V1().ManifestWorks().Lister()

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

@@ -18,7 +18,7 @@ import (
 func TestPlaceMWControllerIndex(t *testing.T) {
 	mwrSetTest := helpertest.CreateTestManifestWorkReplicaSet("mwrSet-test", "default", "place-test")
 	mwrSetTest.Status.Summary.Total = 1
-	mw, _ := CreateManifestWork(mwrSetTest, "cls1")
+	mw, _ := CreateManifestWork(mwrSetTest, "cls1", "plc1")
 	fWorkClient := fakeworkclient.NewSimpleClientset(mwrSetTest, mw)
 	workInformerFactory := workinformers.NewSharedInformerFactoryWithOptions(fWorkClient, 1*time.Second)
 

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

@@ -29,7 +29,7 @@ func TestStatusReconcileAsExpected(t *testing.T) {
 	}
 
 	for _, cls := range clusters {
-		mw, _ := CreateManifestWork(mwrSetTest, cls)
+		mw, _ := CreateManifestWork(mwrSetTest, cls, "plc1")
 		cond := getCondition(workv1.WorkApplied, "", "", metav1.ConditionTrue)
 		apimeta.SetStatusCondition(&mw.Status.Conditions, cond)
 
@@ -94,7 +94,7 @@ func TestStatusReconcileAsProcessing(t *testing.T) {
 	}
 
 	for id, cls := range clusters {
-		mw, _ := CreateManifestWork(mwrSetTest, cls)
+		mw, _ := CreateManifestWork(mwrSetTest, cls, "plc1")
 		cond := getCondition(workv1.WorkApplied, "", "", metav1.ConditionTrue)
 		apimeta.SetStatusCondition(&mw.Status.Conditions, cond)
 
@@ -166,7 +166,7 @@ func TestStatusReconcileNotAsExpected(t *testing.T) {
 
 	avaCount, processingCount, degradCount := 0, 0, 0
 	for id, cls := range clusters {
-		mw, _ := CreateManifestWork(mwrSetTest, cls)
+		mw, _ := CreateManifestWork(mwrSetTest, cls, "plc1")
 		cond := getCondition(workv1.WorkApplied, "", "", metav1.ConditionTrue)
 		apimeta.SetStatusCondition(&mw.Status.Conditions, cond)