feat: support rolling update by partition in PodDecoration (#168)

* support PodDecoration rolling update by partition

* add ut of poddecoration getter

* ut fix

* Add some comments. Started SharedStrategyController differently

* delete unused code

apis/apps/v1alpha1/poddecoration_types.go

@@ -224,9 +224,6 @@ type PodDecorationStatus struct {
 	// +optional
 	UpdatedAvailablePods int32 `json:"updatedAvailablePods,omitempty"`
 
-	// IsEffective indicates PodDecoration is the only one that takes effect in the same group
-	IsEffective *bool `json:"isEffective,omitempty"`
-
 	// Details record the update information of CollaSets and Pods
 	Details []PodDecorationWorkloadDetail `json:"details,omitempty"`
 }

apis/apps/v1alpha1/well_known_labels.go

@@ -55,6 +55,11 @@ const (
 	CollaSetUpdateIndicateLabelKey = "collaset.kusionstack.io/update-included"
 )
 
+// PodDecoration labels
+const (
+	PodDecorationLabelPrefix = "poddecoration.kusionstack.io/pd-"
+)
+
 var (
 	WellKnownLabelPrefixesWithID = []string{PodOperatingLabelPrefix, PodOperationTypeLabelPrefix, PodPreCheckLabelPrefix, PodPreCheckedLabelPrefix,
 		PodPreparingLabelPrefix, PodDoneOperationTypeLabelPrefix, PodUndoOperationTypeLabelPrefix, PodOperateLabelPrefix, PodOperatedLabelPrefix, PodPostCheckLabelPrefix,

config/crd/bases/apps.kusionstack.io_poddecorations.yaml

@@ -5443,10 +5443,6 @@ spec:
                   injected with this PodDecoration
                 format: int32
                 type: integer
-              isEffective:
-                description: IsEffective indicates PodDecoration is the only one that
-                  takes effect in the same group
-                type: boolean
               matchedPods:
                 description: MatchedPods is the number of Pods whose labels are matched
                   with this PodDecoration's selector

pkg/controllers/collaset/collaset_controller.go

@@ -27,6 +27,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -40,6 +41,8 @@ import (
 	collasetutils "kusionstack.io/operating/pkg/controllers/collaset/utils"
 	controllerutils "kusionstack.io/operating/pkg/controllers/utils"
 	"kusionstack.io/operating/pkg/controllers/utils/expectations"
+	utilspoddecoration "kusionstack.io/operating/pkg/controllers/utils/poddecoration"
+	"kusionstack.io/operating/pkg/controllers/utils/poddecoration/strategy"
 	"kusionstack.io/operating/pkg/controllers/utils/podopslifecycle"
 	"kusionstack.io/operating/pkg/controllers/utils/revision"
 	commonutils "kusionstack.io/operating/pkg/utils"
@@ -91,7 +94,16 @@ func AddToMgr(mgr ctrl.Manager, r reconcile.Reconciler) error {
 		return err
 	}
 
-	err = c.Watch(&source.Kind{Type: &appsv1alpha1.PodDecoration{}}, &podDecorationHandler{})
+	// Only for starting SharedStrategyController
+	err = c.Watch(strategy.SharedStrategyController, &handler.Funcs{})
+	if err != nil {
+		return err
+	}
+
+	ch := make(chan event.GenericEvent, 1<<10)
+	strategy.SharedStrategyController.RegisterGenericEventChannel(ch)
+	// Watch PodDecoration related events
+	err = c.Watch(&source.Channel{Source: ch}, &handler.EnqueueRequestForObject{})
 	if err != nil {
 		return err
 	}
@@ -172,23 +184,17 @@ func (r *CollaSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		UpdatedRevision: updatedRevision.Name,
 	}
 
+	getter, err := utilspoddecoration.NewPodDecorationGetter(r.Client, instance.Namespace)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
 	resources := &collasetutils.RelatedResources{
 		Revisions:       revisions,
 		CurrentRevision: currentRevision,
 		UpdatedRevision: updatedRevision,
 		NewStatus:       newStatus,
+		PDGetter:        getter,
 	}
-	resources.PDGetter, err = utils.NewPodDecorationGetter(ctx, r.Client, instance.Namespace)
-	if err != nil {
-		return ctrl.Result{}, fmt.Errorf("fail to get effective pod decorations by CollaSet %s: %s", key, err)
-	}
-	for _, pd := range resources.PDGetter.GetLatestDecorations() {
-		if pd.Status.ObservedGeneration != pd.Generation {
-			logger.Info("wait for PodDecoration ObservedGeneration", "CollaSet", key, "PodDecoration", commonutils.ObjectKeyString(pd))
-			return ctrl.Result{}, nil
-		}
-	}
-
 	requeueAfter, newStatus, err := r.DoReconcile(ctx, instance, resources)
 	// update status anyway
 	if err := r.updateStatus(ctx, instance, newStatus); err != nil {

pkg/controllers/collaset/collaset_controller_test.go

@@ -50,6 +50,7 @@ import (
 	"kusionstack.io/operating/pkg/controllers/collaset/synccontrol"
 	collasetutils "kusionstack.io/operating/pkg/controllers/collaset/utils"
 	"kusionstack.io/operating/pkg/controllers/poddeletion"
+	"kusionstack.io/operating/pkg/controllers/utils/poddecoration/strategy"
 	"kusionstack.io/operating/pkg/controllers/utils/podopslifecycle"
 	"kusionstack.io/operating/pkg/utils/inject"
 )
@@ -2040,6 +2041,8 @@ var _ = BeforeSuite(func() {
 	Expect(err).NotTo(HaveOccurred())
 	Expect(config).NotTo(BeNil())
 	sch := scheme.Scheme
+	// ignore PodDecoration SharedStrategyController
+	strategy.SharedStrategyController.Synced()
 	Expect(appsv1.SchemeBuilder.AddToScheme(sch)).NotTo(HaveOccurred())
 	Expect(appsv1alpha1.SchemeBuilder.AddToScheme(sch)).NotTo(HaveOccurred())
 	mgr, err = manager.New(config, manager.Options{

pkg/controllers/collaset/revision.go

@@ -40,7 +40,7 @@ func getCollaSetPatch(cls *appsv1alpha1.CollaSet) ([]byte, error) {
 	objCopy := make(map[string]interface{})
 	specCopy := make(map[string]interface{})
 
-	// Create a patch of the DaemonSet that replaces spec.template
+	// Create a patch of the CollaSet that replaces spec.template
 	spec := raw["spec"].(map[string]interface{})
 	template := spec["template"].(map[string]interface{})
 	template["$patch"] = "replace"

pkg/controllers/collaset/synccontrol/sync_control.go

@@ -226,7 +226,7 @@ func (r *RealSyncControl) SyncPods(
 		successCount, err := controllerutils.SlowStartBatch(len(needReplaceOriginPods), controllerutils.SlowStartInitialBatchSize, false, func(i int, _ error) error {
 			originPod := needReplaceOriginPods[i]
 			ownerRef := metav1.NewControllerRef(instance, instance.GroupVersionKind())
-			updatedPDs, err := resources.PDGetter.GetUpdatedDecorationsByOldPod(ctx, originPod)
+			updatedPDs, err := resources.PDGetter.GetEffective(ctx, originPod)
 			replaceRevision := getReplaceRevision(originPod, resources)
 			if err != nil {
 				return err
@@ -439,7 +439,7 @@ func (r *RealSyncControl) Scale(
 					var pds map[string]*appsv1alpha1.PodDecoration
 					if !ok {
 						// get default PodDecorations if no revision in context
-						pds, localErr = resources.PDGetter.GetLatestDecorationsByTargetLabel(ctx, in.Labels)
+						pds, localErr = resources.PDGetter.GetEffective(ctx, in)
 						if localErr != nil {
 							return localErr
 						}
@@ -453,7 +453,7 @@ func (r *RealSyncControl) Scale(
 						for _, info := range infos {
 							revisions = append(revisions, info.Revision)
 						}
-						pds, localErr = resources.PDGetter.GetDecorationByRevisions(ctx, revisions...)
+						pds, localErr = resources.PDGetter.GetByRevisions(ctx, revisions...)
 						if localErr != nil {
 							return localErr
 						}

pkg/controllers/collaset/synccontrol/update.go

@@ -90,11 +90,11 @@ func attachPodUpdateInfo(ctx context.Context, cls *appsv1alpha1.CollaSet, pods [
 		updateInfo := &PodUpdateInfo{
 			PodWrapper: pod,
 		}
-		currentPDs, err := resource.PDGetter.GetCurrentDecorationsOnPod(ctx, pod.Pod)
+		currentPDs, err := resource.PDGetter.GetOnPod(ctx, pod.Pod)
 		if err != nil {
 			return nil, err
 		}
-		updatedPDs, err := resource.PDGetter.GetUpdatedDecorationsByOldPod(ctx, pod.Pod)
+		updatedPDs, err := resource.PDGetter.GetEffective(ctx, pod.Pod)
 		if err != nil {
 			return nil, err
 		}