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"
 )
 
@@ -556,9 +557,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,6 +118,11 @@ 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
+	}
+	participants := rollingupdate.Participants(its, fmt.Sprint(its.Generation), rollingUpdateQuota, orderedNames)
 
 	// 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.
@@ -132,15 +138,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
@@ -221,7 +224,6 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 			}
 			updatingPods++
 		}
-		updatedPods++
 	}
 
 	if !isBlocked {

pkg/controller/instanceset2/reconciler_update.go

@@ -30,6 +30,7 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/instancetemplate"
 	"github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/apecloud/kubeblocks/pkg/controller/rollingupdate"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 )
 
@@ -118,12 +119,14 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 		updateCount = len(instancesToBeUpdated)
 	}
 
-	// updatedInstances tracks the positions already covered by the rolling-update
-	// window, while updatingInstances tracks actual updates admitted in this round.
-	updatedInstances := 0
 	updatingInstances := 0
 	priorities := composeRolePriorityMap(its.Spec.Roles)
 	sortInstanceObjects(oldInstanceList, priorities, false)
+	orderedNames := make([]string, len(oldInstanceList))
+	for i, inst := range oldInstanceList {
+		orderedNames[i] = inst.Name
+	}
+	participants := rollingupdate.Participants(its, fmt.Sprint(its.Generation), replicas, orderedNames)
 
 	canBeUpdated := func(inst *workloads.Instance) bool {
 		if !intctrlutil.IsInstanceReady(inst) {
@@ -142,8 +145,8 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 	}
 
 	for _, inst := range oldInstanceList {
-		if updatedInstances >= replicas {
-			break
+		if !participants.Has(inst.Name) {
+			continue
 		}
 		if updatingInstances >= min(unavailable, updateCount) {
 			break
@@ -165,7 +168,6 @@ func (r *updateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 			}
 			updatingInstances++
 		}
-		updatedInstances++
 	}
 	return kubebuilderx.Continue, nil
 }

pkg/controller/rollingupdate/window.go

@@ -0,0 +1,111 @@
+/*
+Copyright (C) 2022-2026 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package rollingupdate
+
+import (
+	"encoding/json"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+const WindowAnnotationKey = "workloads.kubeblocks.io/rolling-update-window"
+
+type window struct {
+	RolloutID    string   `json:"rolloutID"`
+	Replicas     int      `json:"replicas"`
+	Participants []string `json:"participants"`
+}
+
+// Participants returns the stable set of instance names admitted to a rolling
+// update. orderedNames must already be sorted according to the update policy.
+func Participants(owner metav1.Object, rolloutID string, replicas int, orderedNames []string) sets.Set[string] {
+	if replicas < 0 {
+		replicas = 0
+	}
+	if replicas >= len(orderedNames) {
+		removeWindow(owner)
+		return sets.New(orderedNames...)
+	}
+
+	if saved, ok := loadWindow(owner, rolloutID, replicas, orderedNames); ok {
+		return sets.New(saved.Participants...)
+	}
+
+	participants := append([]string(nil), orderedNames[:replicas]...)
+	saveWindow(owner, window{
+		RolloutID:    rolloutID,
+		Replicas:     replicas,
+		Participants: participants,
+	})
+	return sets.New(participants...)
+}
+
+func loadWindow(owner metav1.Object, rolloutID string, replicas int, orderedNames []string) (window, bool) {
+	annotations := owner.GetAnnotations()
+	if annotations == nil {
+		return window{}, false
+	}
+	raw, ok := annotations[WindowAnnotationKey]
+	if !ok {
+		return window{}, false
+	}
+
+	var saved window
+	if json.Unmarshal([]byte(raw), &saved) != nil || saved.RolloutID != rolloutID || saved.Replicas != replicas ||
+		len(saved.Participants) != replicas {
+		return window{}, false
+	}
+
+	validNames := sets.New(orderedNames...)
+	participants := sets.New[string]()
+	for _, name := range saved.Participants {
+		if !validNames.Has(name) || participants.Has(name) {
+			return window{}, false
+		}
+		participants.Insert(name)
+	}
+	return saved, true
+}
+
+func saveWindow(owner metav1.Object, state window) {
+	data, err := json.Marshal(state)
+	if err != nil {
+		return
+	}
+	annotations := owner.GetAnnotations()
+	if annotations == nil {
+		annotations = make(map[string]string)
+	}
+	annotations[WindowAnnotationKey] = string(data)
+	owner.SetAnnotations(annotations)
+}
+
+func removeWindow(owner metav1.Object) {
+	annotations := owner.GetAnnotations()
+	if annotations == nil {
+		return
+	}
+	if _, ok := annotations[WindowAnnotationKey]; !ok {
+		return
+	}
+	delete(annotations, WindowAnnotationKey)
+	owner.SetAnnotations(annotations)
+}

pkg/controller/rollingupdate/window_test.go

@@ -0,0 +1,69 @@
+/*
+Copyright (C) 2022-2026 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package rollingupdate
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+func TestParticipantsRemainStable(t *testing.T) {
+	owner := &metav1.PartialObjectMetadata{}
+	got := Participants(owner, "2", 1, []string{"a", "b", "c"})
+	if !got.Equal(sets.New("a")) {
+		t.Fatalf("expected initial participant a, got %v", got)
+	}
+
+	got = Participants(owner, "2", 1, []string{"b", "a", "c"})
+	if !got.Equal(sets.New("a")) {
+		t.Fatalf("expected participant a after reorder, got %v", got)
+	}
+
+	got = Participants(owner, "3", 1, []string{"b", "a", "c"})
+	if !got.Equal(sets.New("b")) {
+		t.Fatalf("expected new rollout to select b, got %v", got)
+	}
+}
+
+func TestParticipantsResetForReplicaChangeAndInvalidState(t *testing.T) {
+	owner := &metav1.PartialObjectMetadata{}
+	Participants(owner, "2", 1, []string{"a", "b", "c"})
+
+	got := Participants(owner, "2", 2, []string{"b", "a", "c"})
+	if !got.Equal(sets.New("b", "a")) {
+		t.Fatalf("expected replica change to rebuild window, got %v", got)
+	}
+
+	owner.Annotations[WindowAnnotationKey] = "invalid"
+	got = Participants(owner, "2", 1, []string{"c", "b", "a"})
+	if !got.Equal(sets.New("c")) {
+		t.Fatalf("expected invalid state to be rebuilt, got %v", got)
+	}
+
+	got = Participants(owner, "2", 3, []string{"c", "b", "a"})
+	if !got.Equal(sets.New("a", "b", "c")) {
+		t.Fatalf("expected all participants, got %v", got)
+	}
+	if _, ok := owner.Annotations[WindowAnnotationKey]; ok {
+		t.Fatal("expected a full window not to retain the annotation")
+	}
+}