fix(instanceset): stabilize rolling update participants

controllers/workloads/instanceset_controller_2_test.go

@@ -371,7 +371,7 @@ var _ = Describe("InstanceSet Controller 2", func() {
 			}).Should(Succeed())
 		})
 
-		It("uses status role order for a roleful rolling-update window", func() {
+		It("keeps a roleful rolling-update window stable when roles change", func() {
 			createITSObj(itsName, func(f *testapps.MockInstanceSetFactory) {
 				f.SetRoles([]workloads.ReplicaRole{
 					{
@@ -405,6 +405,8 @@ var _ = Describe("InstanceSet Controller 2", func() {
 			})).Should(Succeed())
 
 			By("update its spec")
+			beforeUpdate := time.Now()
+			time.Sleep(time.Second)
 			Expect(testapps.GetAndChangeObj(&testCtx, itsKey, func(its *workloads.InstanceSet) {
 				its.Spec.Template.Spec.DNSPolicy = corev1.DNSClusterFirstWithHostNet
 			})()).ShouldNot(HaveOccurred())
@@ -415,7 +417,24 @@ var _ = Describe("InstanceSet Controller 2", func() {
 				g.Expect(inst.Spec.Template.Spec.DNSPolicy).Should(Equal(corev1.DNSClusterFirstWithHostNet))
 			})).Should(Succeed())
 
-			By("keep leaders outside the rolling-update window")
+			By("make the updated member ready and switch the follower role")
+			Eventually(testapps.CheckObj(&testCtx, followerKey, func(g Gomega, pod *corev1.Pod) {
+				g.Expect(pod.CreationTimestamp.After(beforeUpdate)).Should(BeTrue())
+			})).Should(Succeed())
+			mockPodReadyNAvailableWithRole(itsObj.Namespace, podName(0), "leader", 0)
+			mockPodReadyNAvailableWithRole(itsObj.Namespace, podName(1), "follower", 0)
+			Eventually(func(g Gomega) {
+				leader := &workloads.Instance{}
+				g.Expect(testCtx.Cli.Get(testCtx.Ctx, followerKey, leader)).Should(Succeed())
+				g.Expect(leader.Status.Role).Should(Equal("leader"))
+
+				newFollower := &workloads.Instance{}
+				newFollowerKey := types.NamespacedName{Namespace: itsObj.Namespace, Name: podName(1)}
+				g.Expect(testCtx.Cli.Get(testCtx.Ctx, newFollowerKey, newFollower)).Should(Succeed())
+				g.Expect(newFollower.Status.Role).Should(Equal("follower"))
+			}).Should(Succeed())
+
+			By("keep the original participant and leave the new follower outside the window")
 			Consistently(func(g Gomega) {
 				for i := int32(1); i < replicas; i++ {
 					inst := &workloads.Instance{}

pkg/controller/instanceset/instance_util.go

@@ -43,6 +43,7 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/builder"
 	"github.com/apecloud/kubeblocks/pkg/controller/instancetemplate"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/apecloud/kubeblocks/pkg/controller/rollingupdate"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 )
 
@@ -555,9 +556,15 @@ func buildInstanceTemplateRevision(template *corev1.PodTemplateSpec, parent *wor
 		mutateTemplateFn(templateCopy)
 	}
 	podTemplate := filterInPlaceFields(templateCopy)
+	annotations := make(map[string]string, len(parent.Annotations))
+	for key, value := range parent.Annotations {
+		if key != rollingupdate.WindowAnnotationKey {
+			annotations[key] = value
+		}
+	}
 	its := builder.NewInstanceSetBuilder(parent.Namespace, parent.Name).
 		SetUID(parent.UID).
-		AddAnnotationsInMap(parent.Annotations).
+		AddAnnotationsInMap(annotations).
 		SetSelectorMatchLabel(parent.Labels).
 		SetTemplate(*podTemplate).
 		GetObject()

pkg/controller/instanceset/instance_util_test.go

@@ -38,6 +38,7 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/builder"
 	"github.com/apecloud/kubeblocks/pkg/controller/instancetemplate"
+	"github.com/apecloud/kubeblocks/pkg/controller/rollingupdate"
 )
 
 var _ = Describe("instance util test", func() {
@@ -84,6 +85,21 @@ var _ = Describe("instance util test", func() {
 		})
 	})
 
+	Context("buildInstanceTemplateRevision", func() {
+		It("ignores the internal rolling-update window annotation", func() {
+			before, err := buildInstanceTemplateRevision(&its.Spec.Template, its, nil)
+			Expect(err).ShouldNot(HaveOccurred())
+
+			if its.Annotations == nil {
+				its.Annotations = make(map[string]string)
+			}
+			its.Annotations[rollingupdate.WindowAnnotationKey] = `{"rolloutID":"2","replicas":1,"participants":["pod-0"]}`
+			after, err := buildInstanceTemplateRevision(&its.Spec.Template, its, nil)
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(after).Should(Equal(before))
+		})
+	})
+
 	Context("configsToUpdate", func() {
 		It("treats nil and empty config hash as equal", func() {
 			its := builder.NewInstanceSetBuilder(namespace, name).

pkg/controller/instanceset/reconciler_update.go

@@ -37,6 +37,7 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx"
 	"github.com/apecloud/kubeblocks/pkg/controller/lifecycle"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/apecloud/kubeblocks/pkg/controller/rollingupdate"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 )
 
@@ -117,12 +118,25 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 
 	priorities := ComposeRolePriorityMap(its.Spec.Roles)
 	sortObjects(oldPodList, priorities, false)
+	orderedNames := make([]string, len(oldPodList))
+	for i, pod := range oldPodList {
+		orderedNames[i] = pod.Name
+	}
+	updateRevisions, err := GetRevisions(its.Status.UpdateRevisions)
+	if err != nil {
+		return kubebuilderx.Continue, err
+	}
+	participants, windowChanged := rollingupdate.Participants(its,
+		rollingupdate.RolloutID(updateRevisions), rollingUpdateQuota, orderedNames)
+	if windowChanged {
+		return kubebuilderx.Commit, nil
+	}
 
 	// treat old and Pending pod as a special case, as they can be updated without a consequence
 	// PodUpdatePolicy is ignored here since in-place update for a pending pod doesn't make much sense.
-	for i, pod := range oldPodList {
-		if i >= rollingUpdateQuota {
-			break
+	for _, pod := range oldPodList {
+		if !participants.Has(pod.Name) {
+			continue
 		}
 		updatePolicy, _, _, err := getPodUpdatePolicy(its, pod)
 		if err != nil {
@@ -135,15 +149,12 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 		}
 	}
 
-	// updatedPods tracks the positions already covered by the rolling-update
-	// window, while updatingPods tracks actual updates admitted in this round.
-	updatedPods := 0
 	updatingPods := 0
 	isBlocked := false
 	needRetry := false
 	for _, pod := range oldPodList {
-		if updatedPods >= rollingUpdateQuota {
-			break
+		if !participants.Has(pod.Name) {
+			continue
 		}
 		if updatingPods >= unavailableQuota {
 			break
@@ -224,7 +235,6 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 			}
 			updatingPods++
 		}
-		updatedPods++
 	}
 
 	if !isBlocked {

pkg/controller/instanceset/reconciler_update_test.go

@@ -21,19 +21,26 @@ package instanceset
 
 import (
 	"context"
+	"fmt"
 	"slices"
 	"time"
 
+	"github.com/go-logr/logr"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/ptr"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	kbappsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
@@ -201,6 +208,11 @@ var _ = Describe("update reconciler test", func() {
 			}
 			// order: bar-hello-0, bar-foo-1, bar-foo-0, bar-3, bar-2, bar-1, bar-0
 			// expected: bar-hello-0, bar-foo-1 being deleted
+			res, err = reconciler.Reconcile(partitionTree)
+			Expect(err).Should(BeNil())
+			Expect(res).Should(Equal(kubebuilderx.Commit))
+			expectUpdatedPods(partitionTree, []string{})
+
 			res, err = reconciler.Reconcile(partitionTree)
 			Expect(err).Should(BeNil())
 			Expect(res).Should(Equal(kubebuilderx.Continue))
@@ -225,6 +237,10 @@ var _ = Describe("update reconciler test", func() {
 				Expect(ok).Should(BeTrue())
 				makePodLatestRevision(pod)
 			}
+			res, err = reconciler.Reconcile(partitionTree)
+			Expect(err).Should(BeNil())
+			Expect(res).Should(Equal(kubebuilderx.Commit))
+
 			res, err = reconciler.Reconcile(partitionTree)
 			Expect(err).Should(BeNil())
 			Expect(res).Should(Equal(kubebuilderx.Continue))
@@ -315,9 +331,13 @@ var _ = Describe("update reconciler test", func() {
 			expectUpdatedPods(tree, []string{lastPod.GetName()})
 		})
 
-		It("keeps a pending pod outside the rolling-update window untouched", func() {
+		It("keeps a newly prioritized pending pod outside the persisted window untouched", func() {
 			tree := kubebuilderx.NewObjectTree()
 			its.Spec.PodManagementPolicy = appsv1.ParallelPodManagement
+			its.Spec.Roles = []workloads.ReplicaRole{
+				{Name: "follower", UpdatePriority: 1},
+				{Name: "leader", UpdatePriority: 2},
+			}
 			its.Spec.InstanceUpdateStrategy = &workloads.InstanceUpdateStrategy{
 				RollingUpdate: &workloads.RollingUpdate{
 					Replicas:       ptr.To(intstr.FromInt32(1)),
@@ -332,27 +352,75 @@ var _ = Describe("update reconciler test", func() {
 				pod, ok := object.(*corev1.Pod)
 				Expect(ok).Should(BeTrue())
 				pod.Labels[appsv1.ControllerRevisionHashLabelKey] = "old-revision"
-				if pod.Name == "bar-0" {
-					pod.Status.Phase = corev1.PodPending
-					continue
-				}
+				pod.Labels[RoleLabelKey] = "leader"
 				pod.Status.Phase = corev1.PodRunning
 				pod.Status.Conditions = append(pod.Status.Conditions, getPodReadyCondition())
 			}
+			participant := builder.NewPodBuilder(namespace, "bar-2").GetObject()
+			object, err := tree.Get(participant)
+			Expect(err).Should(BeNil())
+			object.(*corev1.Pod).Labels[RoleLabelKey] = "follower"
 
 			reconciler = NewUpdateReconciler()
 			res, err := reconciler.Reconcile(tree)
 			Expect(err).Should(BeNil())
+			Expect(res).Should(Equal(kubebuilderx.Commit))
+			expectUpdatedPods(tree, []string{})
+
+			// Role drift moves bar-1 into the current first position, but bar-2
+			// remains the sole persisted participant.
+			object.(*corev1.Pod).Labels[RoleLabelKey] = "leader"
+			pending := builder.NewPodBuilder(namespace, "bar-1").GetObject()
+			object, err = tree.Get(pending)
+			Expect(err).Should(BeNil())
+			pending = object.(*corev1.Pod)
+			pending.Labels[RoleLabelKey] = "follower"
+			pending.Status.Phase = corev1.PodPending
+
+			res, err = reconciler.Reconcile(tree)
+			Expect(err).Should(BeNil())
 			Expect(res).Should(Equal(kubebuilderx.Continue))
 			expectUpdatedPods(tree, []string{"bar-2"})
 
-			pending := builder.NewPodBuilder(namespace, "bar-0").GetObject()
-			object, err := tree.Get(pending)
+			object, err = tree.Get(pending)
 			Expect(err).Should(BeNil())
 			Expect(object).ShouldNot(BeNil())
 			Expect(object.(*corev1.Pod).Status.Phase).Should(Equal(corev1.PodPending))
 		})
 
+		It("does not update a child when the participant window patch conflicts", func() {
+			tree := kubebuilderx.NewObjectTree()
+			its.Spec.PodManagementPolicy = appsv1.ParallelPodManagement
+			its.Spec.InstanceUpdateStrategy = &workloads.InstanceUpdateStrategy{
+				RollingUpdate: &workloads.RollingUpdate{
+					Replicas:       ptr.To(intstr.FromInt32(1)),
+					MaxUnavailable: ptr.To(intstr.FromInt32(1)),
+				},
+			}
+			tree.SetRoot(its)
+			prepareForUpdate(tree)
+
+			for _, object := range tree.List(&corev1.Pod{}) {
+				pod := object.(*corev1.Pod)
+				pod.Labels[appsv1.ControllerRevisionHashLabelKey] = "old-revision"
+				pod.Status.Phase = corev1.PodRunning
+				pod.Status.Conditions = append(pod.Status.Conditions, getPodReadyCondition())
+			}
+			its.Spec.Template.Spec.DNSPolicy = corev1.DNSClusterFirstWithHostNet
+
+			cli := &rootPatchFailureClient{Client: fake.NewClientBuilder().Build()}
+			result, err := kubebuilderx.NewController(context.Background(), cli, ctrl.Request{},
+				record.NewFakeRecorder(10), logr.Discard()).
+				Prepare(staticTreeLoader{tree: tree}).
+				Do(NewRevisionUpdateReconciler()).
+				Do(NewUpdateReconciler()).
+				Commit()
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(result.Requeue).Should(BeTrue())
+			Expect(cli.rootPatches).Should(Equal(1))
+			Expect(cli.childWrites).Should(BeZero())
+		})
+
 		It("respects maxUnavailable with pending pods", func() {
 			// update order: bar-2, bar-1, bar-0
 			tree := kubebuilderx.NewObjectTree()
@@ -616,6 +684,43 @@ var _ = Describe("update reconciler test", func() {
 	})
 })
 
+type staticTreeLoader struct {
+	tree *kubebuilderx.ObjectTree
+}
+
+func (l staticTreeLoader) Load(_ context.Context, _ client.Reader, _ ctrl.Request, _ record.EventRecorder,
+	_ logr.Logger) (*kubebuilderx.ObjectTree, error) {
+	return l.tree, nil
+}
+
+type rootPatchFailureClient struct {
+	client.Client
+	rootPatches int
+	childWrites int
+}
+
+func (c *rootPatchFailureClient) Patch(_ context.Context, obj client.Object, _ client.Patch,
+	_ ...client.PatchOption) error {
+	if _, ok := obj.(*workloads.InstanceSet); ok {
+		c.rootPatches++
+		return apierrors.NewConflict(schema.GroupResource{
+			Group: "workloads.kubeblocks.io", Resource: "instancesets",
+		}, obj.GetName(), fmt.Errorf("injected participant window conflict"))
+	}
+	c.childWrites++
+	return fmt.Errorf("unexpected child patch for %s", obj.GetName())
+}
+
+func (c *rootPatchFailureClient) Update(_ context.Context, obj client.Object, _ ...client.UpdateOption) error {
+	c.childWrites++
+	return fmt.Errorf("unexpected child update for %s", obj.GetName())
+}
+
+func (c *rootPatchFailureClient) Delete(_ context.Context, obj client.Object, _ ...client.DeleteOption) error {
+	c.childWrites++
+	return fmt.Errorf("unexpected child delete for %s", obj.GetName())
+}
+
 // lifecycleCallSpy is a test double for lifecycle.Lifecycle used to assert
 // that switchover is or is not invoked during reconciliation. Methods that
 // the call-site tests do not exercise return nil to satisfy the interface.