feat: support clean old lifecycle before new update

pkg/controllers/collaset/collaset_controller_test.go

@@ -614,6 +614,9 @@ var _ = Describe("collaset controller", func() {
 		}, 5*time.Second, 1*time.Second).Should(BeTrue())
 
 		Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+		for i := range podList.Items {
+			Expect(cleanPodLifecycleLabels(c, podList.Items[i].Namespace, podList.Items[i].Name)).Should(BeNil())
+		}
 		for _, number := range []int32{1, 2, 3, 4} {
 			pod := podList.Items[number-1]
 			// label pod to trigger update
@@ -625,10 +628,24 @@ var _ = Describe("collaset controller", func() {
 				return true
 			})).Should(BeNil())
 
+			Eventually(func() bool {
+				Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, &pod)).Should(BeNil())
+				if !podopslifecycle.IsDuringOps(collasetutils.UpdateOpsLifecycleAdapter, &pod) {
+					return false
+				}
+				// allow Pod to do update
+				Expect(updatePodWithRetry(c, pod.Namespace, pod.Name, func(pod *corev1.Pod) bool {
+					labelOperate := fmt.Sprintf("%s/%s", appsv1alpha1.PodOperateLabelPrefix, collasetutils.UpdateOpsLifecycleAdapter.GetID())
+					pod.Labels[labelOperate] = fmt.Sprintf("%d", time.Now().UnixNano())
+					return true
+				})).Should(BeNil())
+				return true
+			}, 10*time.Second, 1*time.Second).Should(BeTrue())
+
 			Eventually(func() error {
 				// check updated pod replicas by CollaSet status
 				return expectedStatusReplicas(c, cs, 0, 0, 0, 4, number, 1, 0, 0)
-			}, 5*time.Second, 1*time.Second).Should(BeNil())
+			}, 10*time.Second, 1*time.Second).Should(BeNil())
 
 			Expect(updatePodStatusWithRetry(c, pod.Namespace, pod.Name, func(pod *corev1.Pod) bool {
 				pod.Status.ContainerStatuses = []corev1.ContainerStatus{
@@ -957,6 +974,16 @@ var _ = Describe("collaset controller", func() {
 			return true
 		})).Should(BeNil())
 
+		Eventually(func() bool {
+			Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+			for k := range podList.Items[0].Labels {
+				if strings.HasPrefix(k, appsv1alpha1.PodOperatingLabelPrefix) {
+					return true
+				}
+			}
+			return false
+		}, 5*time.Second, 1*time.Second).Should(BeTrue())
+
 		// allow Pod to update
 		Expect(updatePodWithRetry(c, pod.Namespace, pod.Name, func(pod *corev1.Pod) bool {
 			labelOperate := fmt.Sprintf("%s/%s", appsv1alpha1.PodOperateLabelPrefix, collasetutils.UpdateOpsLifecycleAdapter.GetID())
@@ -1069,15 +1096,27 @@ var _ = Describe("collaset controller", func() {
 			return true
 		}))
 
-		for i := range podList.Items {
-			pod := &podList.Items[i]
-			// allow Pod to update
-			Expect(updatePodWithRetry(c, pod.Namespace, pod.Name, func(pod *corev1.Pod) bool {
-				labelOperate := fmt.Sprintf("%s/%s", appsv1alpha1.PodOperateLabelPrefix, collasetutils.UpdateOpsLifecycleAdapter.GetID())
-				pod.Labels[labelOperate] = "true"
-				return true
-			})).Should(BeNil())
-		}
+		Eventually(func() int {
+			count := 0
+			for i := range podList.Items {
+				pod := &podList.Items[i]
+				Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, pod)).Should(BeNil())
+				for k := range pod.Labels {
+					if strings.HasPrefix(k, appsv1alpha1.PodOperatingLabelPrefix) {
+						count++
+						// allow Pod to update
+						Expect(updatePodWithRetry(c, pod.Namespace, pod.Name, func(pod *corev1.Pod) bool {
+							labelOperate := fmt.Sprintf("%s/%s", appsv1alpha1.PodOperateLabelPrefix, collasetutils.UpdateOpsLifecycleAdapter.GetID())
+							pod.Labels[labelOperate] = "true"
+							return true
+						})).Should(BeNil())
+					}
+				}
+
+			}
+			return count
+		}, 5*time.Second, 1*time.Second).Should(BeEquivalentTo(1))
+
 		Eventually(func() error {
 			// check updated pod replicas by CollaSet status
 			return expectedStatusReplicas(c, cs, 0, 0, 0, 2, 1, 1, 0, 0)
@@ -2082,8 +2121,7 @@ var _ = Describe("collaset controller", func() {
 
 		Eventually(func() bool {
 			pod := &corev1.Pod{}
-			error := c.Get(context.TODO(), types.NamespacedName{Namespace: originPod1.Namespace, Name: originPod1.Name}, pod)
-			Expect(error).Should(BeNil())
+			Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: originPod1.Namespace, Name: originPod1.Name}, pod)).Should(BeNil())
 			Expect(pod.Labels).ShouldNot(BeNil())
 			_, replaceIndicate := pod.Labels[appsv1alpha1.PodReplaceIndicationLabelKey]
 			_, replaceByUpdate := pod.Labels[appsv1alpha1.PodReplaceByReplaceUpdateLabelKey]
@@ -2101,14 +2139,21 @@ var _ = Describe("collaset controller", func() {
 		})).Should(BeNil())
 
 		// allow originPod2 to do inPlace update
+		Eventually(func() bool {
+			Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: originPod2.Namespace, Name: originPod2.Name}, &originPod2)).Should(BeNil())
+			for k := range originPod2.Labels {
+				if strings.HasPrefix(k, appsv1alpha1.PodOperatingLabelPrefix) {
+					return true
+				}
+			}
+			return false
+		}, time.Second*10, time.Second).Should(BeTrue())
 		Expect(updatePodWithRetry(c, originPod2.Namespace, originPod2.Name, func(pod *corev1.Pod) bool {
 			labelOperate := fmt.Sprintf("%s/%s", appsv1alpha1.PodOperateLabelPrefix, collasetutils.UpdateOpsLifecycleAdapter.GetID())
 			pod.Labels[labelOperate] = fmt.Sprintf("%d", time.Now().UnixNano())
 			return true
 		})).Should(BeNil())
 
-		time.Sleep(3 * time.Second)
-
 		Eventually(func() error {
 			// check updated pod replicas by CollaSet status
 			return expectedStatusReplicas(c, cs, 0, 0, 0, 3, 2, 2, 0, 0)
@@ -3686,7 +3731,7 @@ var _ = Describe("collaset controller", func() {
 			return true
 		})).Should(BeNil())
 
-		// included pod should not have ownerReference
+		// included pod should have ownerReference
 		Eventually(func() bool {
 			Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
 			podIncluded := false
@@ -3703,7 +3748,7 @@ var _ = Describe("collaset controller", func() {
 			return podIncluded
 		}, 10*time.Second, 1*time.Second).Should(BeTrue())
 
-		// included pvc should not have ownerReference
+		// included pvc should have ownerReference
 		Eventually(func() bool {
 			pvcIncluded := false
 			Expect(c.List(context.TODO(), pvcList, client.InNamespace(cs.Namespace))).Should(BeNil())
@@ -3738,6 +3783,16 @@ var _ = Describe("collaset controller", func() {
 			return true
 		})).Should(BeNil())
 
+		Eventually(func() bool {
+			Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+			for k := range podList.Items[0].Labels {
+				if strings.HasPrefix(k, appsv1alpha1.PodOperatingLabelPrefix) {
+					return true
+				}
+			}
+			return false
+		}, 5*time.Second, 1*time.Second).Should(BeTrue())
+
 		// allow pod to update
 		Expect(c.List(context.TODO(), pvcList, client.InNamespace(cs.Namespace))).Should(BeNil())
 		for i := range podList.Items {
@@ -3912,17 +3967,27 @@ var _ = Describe("collaset controller", func() {
 			},
 		}
 
+		Expect(c.Create(context.TODO(), podDecoration)).Should(BeNil())
+
 		// allow Pod to do update
 		for i := range podList.Items {
 			pod := podList.Items[i]
+			Eventually(func() bool {
+				Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, &pod)).Should(BeNil())
+				for k := range pod.Labels {
+					if strings.HasPrefix(k, appsv1alpha1.PodOperatingLabelPrefix) {
+						return true
+					}
+				}
+				return false
+			}, time.Second*10, time.Second).Should(BeTrue())
 			Expect(updatePodWithRetry(c, pod.Namespace, pod.Name, func(pod *corev1.Pod) bool {
 				labelOperate := fmt.Sprintf("%s/%s", appsv1alpha1.PodOperateLabelPrefix, collasetutils.UpdateOpsLifecycleAdapter.GetID())
 				pod.Labels[labelOperate] = fmt.Sprintf("%d", time.Now().UnixNano())
 				return true
 			})).Should(BeNil())
 		}
 
-		Expect(c.Create(context.TODO(), podDecoration)).Should(BeNil())
 		Eventually(func() int32 {
 			err := c.Get(context.TODO(), types.NamespacedName{Namespace: podDecoration.Namespace, Name: podDecoration.Name}, podDecoration)
 			if !errors.IsNotFound(err) {
@@ -4165,6 +4230,33 @@ func updatePodStatusWithRetry(c client.Client, namespace, name string, updateFn
 	})
 }
 
+func cleanPodLifecycleLabels(c client.Client, namespace, name string) error {
+	pod := &corev1.Pod{}
+	if err := c.Get(context.TODO(), types.NamespacedName{Namespace: namespace, Name: name}, pod); err != nil {
+		return err
+	}
+	newLabels := make(map[string]string)
+	for k, v := range pod.Labels {
+		has := false
+		for _, prefix := range appsv1alpha1.WellKnownLabelPrefixesWithID {
+			if strings.HasPrefix(k, prefix) {
+				has = true
+				break
+			}
+		}
+		if !has {
+			newLabels[k] = v
+		}
+	}
+	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		if err := c.Get(context.TODO(), types.NamespacedName{Namespace: namespace, Name: name}, pod); err != nil {
+			return err
+		}
+		pod.Labels = newLabels
+		return c.Update(context.TODO(), pod)
+	})
+}
+
 func testReconcile(inner reconcile.Reconciler) (reconcile.Reconciler, chan reconcile.Request) {
 	requests := make(chan reconcile.Request, 5)
 	fn := reconcile.Func(func(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {

pkg/controllers/collaset/synccontrol/sync_control.go

@@ -762,7 +762,7 @@ func (r *RealSyncControl) Update(
 					"UpdatePodCanceled",
 					"pod %s/%s with revision %s update is canceled due to not started and not included by partition",
 					podInfo.Namespace, podInfo.Name, podInfo.CurrentRevision.Name)
-				return ojutils.CancelOpsLifecycle(ctx, r.client, collasetutils.UpdateOpsLifecycleAdapter, podInfo.Pod)
+				return ojutils.CancelOpsLifecycle(r.client, collasetutils.UpdateOpsLifecycleAdapter, podInfo.Pod)
 			}
 			// not allowedOps, skip GetPodUpdateFinishStatus
 			return nil

pkg/controllers/collaset/synccontrol/update.go

@@ -378,7 +378,8 @@ func (u *GenericPodUpdater) BeginUpdatePod(_ context.Context, resources *collase
 	succCount, err := controllerutils.SlowStartBatch(len(podCh), controllerutils.SlowStartInitialBatchSize, false, func(int, error) error {
 		podInfo := <-podCh
 		u.Recorder.Eventf(podInfo.Pod, corev1.EventTypeNormal, "PodUpdateLifecycle", "try to begin PodOpsLifecycle for updating Pod of CollaSet")
-		if updated, err := podopslifecycle.Begin(u.Client, collasetutils.UpdateOpsLifecycleAdapter, podInfo.Pod, func(obj client.Object) (bool, error) {
+
+		if updated, err := podopslifecycle.BeginWithCleaningOld(u.Client, collasetutils.UpdateOpsLifecycleAdapter, podInfo.Pod, func(obj client.Object) (bool, error) {
 			if !podInfo.OnlyMetadataChanged && !podInfo.InPlaceUpdateSupport {
 				return podopslifecycle.WhenBeginDelete(obj)
 			}

pkg/controllers/operationjob/operationjob_manager.go

@@ -323,7 +323,7 @@
 	lifecycleAdapter := NewLifecycleAdapter(operationJob.Name, operationJob.Spec.Action)
 
 	if forced {
-		err := ojutils.CancelOpsLifecycle(ctx, r.Client, lifecycleAdapter, candidate.Pod)
+		err := ojutils.CancelOpsLifecycle(r.Client, lifecycleAdapter, candidate.Pod)
 		if err != nil {
 			return err
 		}

pkg/controllers/operationjob/utils/lifecycle.go

@@ -17,11 +17,9 @@ limitations under the License.
 package utils
 
 import (
-	"context"
 	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
-	appsv1alpha1 "kusionstack.io/kube-api/apps/v1alpha1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	podopslifecycleutil "kusionstack.io/kuperator/pkg/controllers/podopslifecycle"
@@ -66,22 +64,17 @@ func FinishOperateLifecycle(client client.Client, adapter podopslifecycle.Lifecy
 	return nil
 }
 
-func CancelOpsLifecycle(ctx context.Context, client client.Client, adapter podopslifecycle.LifecycleAdapter, pod *corev1.Pod) error {
+func CancelOpsLifecycle(client client.Client, adapter podopslifecycle.LifecycleAdapter, pod *corev1.Pod) error {
 	if pod == nil {
 		return nil
 	}
 
 	// only cancel when lifecycle exist on pod
 	if exist, err := podopslifecycleutil.IsLifecycleOnPod(adapter.GetID(), pod); err != nil {
-		return err
+		return fmt.Errorf("fail to check %s PodOpsLifecycle on Pod %s/%s: %s", adapter.GetID(), pod.Namespace, pod.Name, err)
 	} else if !exist {
 		return nil
 	}
 
-	labelUndo := fmt.Sprintf("%s/%s", appsv1alpha1.PodUndoOperationTypeLabelPrefix, adapter.GetID())
-	if pod.Labels == nil {
-		pod.Labels = make(map[string]string)
-	}
-	pod.Labels[labelUndo] = string(adapter.GetType())
-	return client.Update(ctx, pod)
+	return podopslifecycle.Undo(client, adapter, pod)
 }

pkg/controllers/podopslifecycle/utils.go

@@ -22,7 +22,6 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
-
 	"kusionstack.io/kube-api/apps/v1alpha1"
 )
 

pkg/controllers/utils/podopslifecycle/utils.go

@@ -23,10 +23,13 @@
 	"strings"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"kusionstack.io/kube-api/apps/v1alpha1"
+
+	podopslifecycleutil "kusionstack.io/kuperator/pkg/controllers/podopslifecycle"
 )
 
 // IsDuringOps decides whether the Pod is during ops or not
@@ -85,6 +88,18 @@
 	return false, nil
 }
 
+// BeginWithCleaningOld is used for an CRD Operator to begin a lifecycle with cleaning the old lifecycle
+func BeginWithCleaningOld(c client.Client, adapter LifecycleAdapter, obj client.Object, updateFunc ...UpdateFunc) (updated bool, err error) {
+	if podInUpdateLifecycle, err := podopslifecycleutil.IsLifecycleOnPod(adapter.GetID(), obj.(*corev1.Pod)); err != nil {
+		return false, fmt.Errorf("fail to check %s PodOpsLifecycle on Pod %s/%s: %s", adapter.GetID(), obj.GetNamespace(), obj.GetName(), err)
+	} else if podInUpdateLifecycle {
+		if err := Undo(c, adapter, obj); err != nil {
+			return false, err
+		}
+	}
+	return Begin(c, adapter, obj, updateFunc...)
+}
+
 // AllowOps is used to check whether the PodOpsLifecycle phase is in UPGRADE to do following operations.
 func AllowOps(adapter LifecycleAdapter, operationDelaySeconds int32, obj client.Object) (requeueAfter *time.Duration, allow bool) {
 	if !IsDuringOps(adapter, obj) {
@@ -139,6 +154,12 @@
 	return false, err
 }
 
+// Undo is used for an CRD Operator to undo a lifecycle
+func Undo(c client.Client, adapter LifecycleAdapter, obj client.Object) error {
+	setUndo(adapter, obj)
+	return c.Update(context.Background(), obj)
+}
+
 func checkOperatingID(adapter LifecycleAdapter, obj client.Object) (val string, ok bool) {
 	labelID := fmt.Sprintf("%s/%s", v1alpha1.PodOperatingLabelPrefix, adapter.GetID())
 	_, ok = obj.GetLabels()[labelID]
@@ -177,6 +198,11 @@
 	return
 }
 
+func setUndo(adapter LifecycleAdapter, obj client.Object) {
+	labelUndo := fmt.Sprintf("%s/%s", v1alpha1.PodUndoOperationTypeLabelPrefix, adapter.GetID())
+	obj.GetLabels()[labelUndo] = string(adapter.GetType())
+}
+
 func deleteOperatingID(adapter LifecycleAdapter, obj client.Object) (val string, ok bool) {
 	labelID := fmt.Sprintf("%s/%s", v1alpha1.PodOperatingLabelPrefix, adapter.GetID())
 	delete(obj.GetLabels(), labelID)