Scale and Update Parall with revert partition

pkg/controllers/collaset/collaset_controller.go

@@ -228,16 +228,13 @@ func (r *CollaSetReconciler) doSync(
 		return podWrappers, nil, err
 	}
 
-	scaling, scaleRequeueAfter, err := r.syncControl.Scale(ctx, instance, resources, podWrappers, ownedIDs)
-	if err != nil || scaling {
-		return podWrappers, scaleRequeueAfter, err
-	}
+	_, scaleRequeueAfter, scaleErr := r.syncControl.Scale(ctx, instance, resources, podWrappers, ownedIDs)
+	_, updateRequeueAfter, updateErr := r.syncControl.Update(ctx, instance, resources, podWrappers, ownedIDs)
 
-	_, updateRequeueAfter, err := r.syncControl.Update(ctx, instance, resources, podWrappers, ownedIDs)
+	err = controllerutils.AggregateErrors([]error{scaleErr, updateErr})
 	if updateRequeueAfter != nil && (scaleRequeueAfter == nil || *updateRequeueAfter < *scaleRequeueAfter) {
 		return podWrappers, updateRequeueAfter, err
 	}
-
 	return podWrappers, scaleRequeueAfter, err
 }
 
@@ -254,7 +251,8 @@ func calculateStatus(
 	var scheduledReplicas, readyReplicas, availableReplicas, replicas, updatedReplicas, operatingReplicas,
 		updatedReadyReplicas, updatedAvailableReplicas int32
 
-	for _, podWrapper := range podWrappers {
+	activePods := synccontrol.FilterOutPlaceHolderPodWrappers(podWrappers)
+	for _, podWrapper := range activePods {
 		replicas++
 
 		isUpdated := false

pkg/controllers/collaset/collaset_controller_test.go

@@ -393,6 +393,233 @@ var _ = Describe("collaset controller", func() {
 		}
 	})
 
+	It("scale succeed and update parallel", func() {
+		testcase := "test-scale-and-update"
+		Expect(createNamespace(c, testcase)).Should(BeNil())
+
+		cs := &appsv1alpha1.CollaSet{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: testcase,
+				Name:      "foo",
+			},
+			Spec: appsv1alpha1.CollaSetSpec{
+				Replicas: int32Pointer(2),
+				UpdateStrategy: appsv1alpha1.UpdateStrategy{
+					RollingUpdate: &appsv1alpha1.RollingUpdateCollaSetStrategy{
+						ByPartition: &appsv1alpha1.ByPartition{
+							Partition: int32Pointer(1),
+						},
+					},
+				},
+				Selector: &metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						"app": "foo",
+					},
+				},
+				Template: corev1.PodTemplateSpec{
+					ObjectMeta: metav1.ObjectMeta{
+						Labels: map[string]string{
+							"app": "foo",
+						},
+					},
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:  "foo",
+								Image: "nginx:v1",
+							},
+						},
+					},
+				},
+			},
+		}
+
+		Expect(c.Create(context.TODO(), cs)).Should(BeNil())
+
+		podList := &corev1.PodList{}
+		Eventually(func() bool {
+			Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+			return len(podList.Items) == 2
+		}, 5*time.Second, 1*time.Second).Should(BeTrue())
+		Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: cs.Namespace, Name: cs.Name}, cs)).Should(BeNil())
+		Expect(expectedStatusReplicas(c, cs, 0, 0, 0, 2, 2, 0, 0, 0)).Should(BeNil())
+
+		// update and scale out to 3 replicas
+		Expect(updateCollaSetWithRetry(c, cs.Namespace, cs.Name, func(cls *appsv1alpha1.CollaSet) bool {
+			cls.Spec.Replicas = int32Pointer(3)
+			cls.Spec.Template.Spec.Containers[0].Image = "nginx:v2"
+			return true
+		})).Should(BeNil())
+
+		Eventually(func() int {
+			Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+			return len(podList.Items)
+		}, 5*time.Second, 1*time.Second).Should(BeEquivalentTo(3))
+
+		// allow Pod to do update
+		for i := range podList.Items {
+			pod := &podList.Items[i]
+			if !podopslifecycle.IsDuringOps(collasetutils.UpdateOpsLifecycleAdapter, pod) {
+				continue
+			}
+			if _, allowed := podopslifecycle.AllowOps(collasetutils.UpdateOpsLifecycleAdapter, 0, pod); allowed {
+				continue
+			}
+			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())
+		}
+		Eventually(func() error {
+			return expectedStatusReplicas(c, cs, 0, 0, 0, 3, 2, 1, 0, 0)
+		}, 10*time.Second, 1*time.Second).Should(BeNil())
+
+		// delete updated pods to trigger scale out
+		Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+		for _, pod := range podList.Items {
+			if pod.Labels[appsv1.ControllerRevisionHashLabelKey] == cs.Status.UpdatedRevision {
+				Expect(c.Delete(ctx, &pod)).Should(BeNil())
+			}
+		}
+
+		Eventually(func() error {
+			return expectedStatusReplicas(c, cs, 0, 0, 0, 3, 2, 0, 0, 0)
+		}, 10*time.Second, 1*time.Second).Should(BeNil())
+
+	})
+
+	It("scale failed and update parallel", func() {
+		testcase := "test-scale-failed-and-update"
+		Expect(createNamespace(c, testcase)).Should(BeNil())
+
+		cs := &appsv1alpha1.CollaSet{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: testcase,
+				Name:      "foo",
+			},
+			Spec: appsv1alpha1.CollaSetSpec{
+				Replicas: int32Pointer(3),
+				Selector: &metav1.LabelSelector{
+					MatchLabels: map[string]string{
+						"app": "foo",
+					},
+				},
+				Template: corev1.PodTemplateSpec{
+					ObjectMeta: metav1.ObjectMeta{
+						Labels: map[string]string{
+							"app": "foo",
+						},
+					},
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:  "foo",
+								Image: "nginx:v1",
+							},
+						},
+					},
+				},
+			},
+		}
+
+		Expect(c.Create(context.TODO(), cs)).Should(BeNil())
+
+		podList := &corev1.PodList{}
+		Eventually(func() int {
+			Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+			return len(podList.Items)
+		}, 5*time.Second, 1*time.Second).Should(BeEquivalentTo(3))
+		Expect(expectedStatusReplicas(c, cs, 0, 0, 0, 3, 3, 0, 0, 0)).Should(BeNil())
+
+		// update collaset with non-exist pvc by partition=1
+		Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: cs.Namespace, Name: cs.Name}, cs)).Should(BeNil())
+		Expect(updateCollaSetWithRetry(c, cs.Namespace, cs.Name, func(cls *appsv1alpha1.CollaSet) bool {
+			cls.Spec.UpdateStrategy.RollingUpdate = &appsv1alpha1.RollingUpdateCollaSetStrategy{
+				ByPartition: &appsv1alpha1.ByPartition{
+					Partition: int32Pointer(1),
+				},
+			}
+			cls.Spec.Template.Spec.Containers = []corev1.Container{
+				{
+					Name:  "foo",
+					Image: "nginx:v1",
+					VolumeMounts: []corev1.VolumeMount{
+						{
+							Name:      "not-exist",
+							MountPath: "/tmp/not-exist",
+						},
+					},
+				},
+			}
+			return true
+		})).Should(BeNil())
+		// allow Pod to do update
+		Eventually(func() int32 {
+			Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: cs.Namespace, Name: cs.Name}, cs)).Should(BeNil())
+			return cs.Status.OperatingReplicas
+		}, 10*time.Second, 1*time.Second).Should(BeEquivalentTo(2))
+		Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+		for i := range podList.Items {
+			pod := &podList.Items[i]
+			if !podopslifecycle.IsDuringOps(collasetutils.UpdateOpsLifecycleAdapter, pod) {
+				continue
+			}
+			if _, allowed := podopslifecycle.AllowOps(collasetutils.UpdateOpsLifecycleAdapter, 0, pod); allowed {
+				continue
+			}
+			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())
+		}
+		// scaling out failed with non-exist pvc
+		Eventually(func() int {
+			Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+			return len(podList.Items)
+		}, 10*time.Second, 1*time.Second).Should(BeEquivalentTo(1))
+		Expect(expectedStatusReplicas(c, cs, 0, 0, 0, 1, 0, 0, 0, 0)).Should(BeNil())
+
+		// update cls with good pod template
+		Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: cs.Namespace, Name: cs.Name}, cs)).Should(BeNil())
+		// allow Pod to do update
+		Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+		for i := range podList.Items {
+			pod := &podList.Items[i]
+			if !podopslifecycle.IsDuringOps(collasetutils.UpdateOpsLifecycleAdapter, pod) {
+				continue
+			}
+			if _, allowed := podopslifecycle.AllowOps(collasetutils.UpdateOpsLifecycleAdapter, 0, pod); allowed {
+				continue
+			}
+			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(updateCollaSetWithRetry(c, cs.Namespace, cs.Name, func(cls *appsv1alpha1.CollaSet) bool {
+			cls.Spec.UpdateStrategy.RollingUpdate = &appsv1alpha1.RollingUpdateCollaSetStrategy{
+				ByPartition: &appsv1alpha1.ByPartition{
+					Partition: int32Pointer(1),
+				},
+			}
+			cls.Spec.Template.Spec.Containers = []corev1.Container{
+				{
+					Name:  "foo",
+					Image: "nginx:v1",
+				},
+			}
+			return true
+		})).Should(BeNil())
+		Eventually(func() int {
+			Expect(c.List(context.TODO(), podList, client.InNamespace(cs.Namespace))).Should(BeNil())
+			return len(podList.Items)
+		}, 5*time.Second, 1*time.Second).Should(BeEquivalentTo(3))
+		Expect(expectedStatusReplicas(c, cs, 0, 0, 0, 3, 3, 0, 0, 0)).Should(BeNil())
+	})
+
 	It("update pod policy", func() {
 		testcase := "test-update-pod-policy"
 		Expect(createNamespace(c, testcase)).Should(BeNil())
@@ -1442,7 +1669,7 @@ var _ = Describe("collaset controller", func() {
 		}, 5*time.Second, 1*time.Second).Should(BeTrue())
 		Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: cs.Namespace, Name: cs.Name}, cs)).Should(BeNil())
 		Expect(expectedStatusReplicas(c, cs, 0, 0, 0, 3, 3, 0, 0, 0)).Should(BeNil())
-		for _, partition := range []int32{0, 1, 2, 3} {
+		for _, partition := range []int32{3, 2, 1, 0} {
 			Expect(c.Get(context.TODO(), types.NamespacedName{Namespace: cs.Namespace, Name: cs.Name}, cs)).Should(BeNil())
 			Expect(updateCollaSetWithRetry(c, cs.Namespace, cs.Name, func(cls *appsv1alpha1.CollaSet) bool {
 				cls.Spec.VolumeClaimTemplates = []corev1.PersistentVolumeClaim{

pkg/controllers/collaset/podcontext/podcontext.go

@@ -32,9 +32,11 @@ import (
 )
 
 const (
-	OwnerContextKey          = "Owner"
-	RevisionContextDataKey   = "Revision"
-	PodDecorationRevisionKey = "PodDecorationRevisions"
+	OwnerContextKey              = "Owner"
+	RevisionContextDataKey       = "Revision"
+	PodDecorationRevisionKey     = "PodDecorationRevisions"
+	JustCreateContextDataKey     = "PodJustCreate"
+	RecreateUpdateContextDataKey = "PodRecreateUpdate"
 )
 
 func AllocateID(c client.Client, instance *appsv1alpha1.CollaSet, defaultRevision string, replicas int) (map[int]*appsv1alpha1.ContextDetail, error) {
@@ -84,9 +86,11 @@ func AllocateID(c client.Client, instance *appsv1alpha1.CollaSet, defaultRevisio
 
 		detail := &appsv1alpha1.ContextDetail{
 			ID: candidateID,
+			// TODO choose just create pods' revision according to scaleStrategy
 			Data: map[string]string{
-				OwnerContextKey:        instance.Name,
-				RevisionContextDataKey: defaultRevision,
+				OwnerContextKey:          instance.Name,
+				RevisionContextDataKey:   defaultRevision,
+				JustCreateContextDataKey: "true",
 			},
 		}
 		existingIDs[candidateID] = detail