From 84c8e12cf1a06a9249e7f80486811f6ba71010d1 Mon Sep 17 00:00:00 2001 From: Wei Cao Date: Thu, 11 Jun 2026 15:23:17 +0800 Subject: [PATCH 1/4] fix: persist member join status with conflict retry --- .../transformer_component_workload_ops.go | 9 +++ pkg/controller/component/replicas.go | 12 ++++ pkg/controller/component/replicas_test.go | 56 +++++++++++++++++++ 3 files changed, 77 insertions(+) diff --git a/controllers/apps/component/transformer_component_workload_ops.go b/controllers/apps/component/transformer_component_workload_ops.go index bdfd805b6a1..783ccfd269c 100644 --- a/controllers/apps/component/transformer_component_workload_ops.go +++ b/controllers/apps/component/transformer_component_workload_ops.go @@ -26,6 +26,7 @@ import ( "time" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" @@ -353,6 +354,14 @@ func (r *componentWorkloadOps) joinMember4ScaleOut() error { if err := r.joinMemberForPod(pod, pods); err != nil { joinErrors = append(joinErrors, fmt.Errorf("pod %s: %w", pod.Name, err)) } else { + key := types.NamespacedName{Namespace: r.protoITS.Namespace, Name: r.protoITS.Name} + if err := component.UpdateReplicaStatusWithRetry(r.transCtx.Context, r.cli, key, pod.Name, func(status *component.ReplicaStatus) error { + status.MemberJoined = ptr.To(true) + return nil + }); err != nil { + joinErrors = append(joinErrors, fmt.Errorf("pod %s: persist member joined status: %w", pod.Name, err)) + continue + } replicas.Status[i].MemberJoined = ptr.To(true) } } diff --git a/pkg/controller/component/replicas.go b/pkg/controller/component/replicas.go index 1a5c536849b..5c319ed847d 100644 --- a/pkg/controller/component/replicas.go +++ b/pkg/controller/component/replicas.go @@ -30,6 +30,7 @@ import ( "github.com/go-logr/logr" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/retry" "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" @@ -397,3 +398,14 @@ func updateReplicaStatusFunc(ctx context.Context, cli client.Client, } return cli.Update(ctx, its) } + +func UpdateReplicaStatusWithRetry(ctx context.Context, cli client.Client, + key types.NamespacedName, replicaName string, f func(*ReplicaStatus) error) error { + return retry.RetryOnConflict(retry.DefaultBackoff, func() error { + its := &workloads.InstanceSet{} + if err := cli.Get(ctx, key, its); err != nil { + return err + } + return updateReplicaStatusFunc(ctx, cli, its, replicaName, f) + }) +} diff --git a/pkg/controller/component/replicas_test.go b/pkg/controller/component/replicas_test.go index c45fd4144e8..80af0d7dabb 100644 --- a/pkg/controller/component/replicas_test.go +++ b/pkg/controller/component/replicas_test.go @@ -20,18 +20,37 @@ along with this program. If not, see . package component import ( + "context" "slices" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" workloads "github.com/apecloud/kubeblocks/apis/workloads/v1" "github.com/apecloud/kubeblocks/pkg/constant" ) +type conflictOnceClient struct { + client.Client + conflict bool +} + +func (c *conflictOnceClient) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error { + if c.conflict { + c.conflict = false + return apierrors.NewConflict(schema.GroupResource{Group: "workloads.kubeblocks.io", Resource: "instancesets"}, obj.GetName(), nil) + } + return c.Client.Update(ctx, obj, opts...) +} + var _ = Describe("replicas", func() { var ( its *workloads.InstanceSet @@ -181,6 +200,43 @@ var _ = Describe("replicas", func() { })).Should(Succeed()) }) + It("retries member-joined status update on conflict", func() { + key := types.NamespacedName{Namespace: testCtx.DefaultNamespace, Name: "test-cluster-its-member-join-retry"} + _ = k8sClient.Delete(ctx, &workloads.InstanceSet{ObjectMeta: metav1.ObjectMeta{Namespace: key.Namespace, Name: key.Name}}) + + its.Name = key.Name + its.Namespace = key.Namespace + its.Spec.Selector = &metav1.LabelSelector{MatchLabels: map[string]string{"app": key.Name}} + its.Spec.Template = corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": key.Name}}, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "test", Image: "test"}}, + }, + } + Expect(StatusReplicasStatus(its, replicas, true, true)).Should(Succeed()) + its.Spec.Replicas = ptr.To[int32](4) + newReplica := "test-cluster-its-3" + Expect(NewReplicasStatus(its, []string{newReplica}, true, false)).Should(Succeed()) + Expect(k8sClient.Create(ctx, its)).Should(Succeed()) + + cli := &conflictOnceClient{Client: k8sClient, conflict: true} + Expect(UpdateReplicaStatusWithRetry(ctx, cli, key, newReplica, func(status *ReplicaStatus) error { + status.MemberJoined = ptr.To(true) + return nil + })).Should(Succeed()) + + got := &workloads.InstanceSet{} + Expect(k8sClient.Get(ctx, key, got)).Should(Succeed()) + status, err := getReplicasStatus(got) + Expect(err).Should(BeNil()) + i := slices.IndexFunc(status.Status, func(s ReplicaStatus) bool { + return s.Name == newReplica + }) + Expect(i).Should(BeNumerically(">=", 0)) + Expect(status.Status[i].MemberJoined).ShouldNot(BeNil()) + Expect(*status.Status[i].MemberJoined).Should(BeTrue()) + }) + // It("task event for new replicas - succeed", func() { // Expect(StatusReplicasStatus(its, replicas, true, true)).Should(Succeed()) // From dacdc2913de252862e4599f0eb4310560fb6b637 Mon Sep 17 00:00:00 2001 From: Wei Cao Date: Mon, 15 Jun 2026 11:19:55 +0800 Subject: [PATCH 2/4] fix: gate scale-in on pending member lifecycle --- .../transformer_component_workload_ops.go | 11 +++++++++- .../transformer_component_workload_test.go | 21 +++++++++++++++++++ pkg/controller/component/replicas.go | 10 +++++++++ 3 files changed, 41 insertions(+), 1 deletion(-) diff --git a/controllers/apps/component/transformer_component_workload_ops.go b/controllers/apps/component/transformer_component_workload_ops.go index 783ccfd269c..9a985c1e946 100644 --- a/controllers/apps/component/transformer_component_workload_ops.go +++ b/controllers/apps/component/transformer_component_workload_ops.go @@ -121,8 +121,17 @@ func (r *componentWorkloadOps) scaleIn() error { } deleteReplicas := r.runningItsPodNameSet.Difference(r.desiredCompPodNameSet).UnsortedList() + pendingReplicas, err := component.PendingLifecycleReplicas(r.protoITS, deleteReplicas) + if err != nil { + return err + } + if len(pendingReplicas) > 0 { + return intctrlutil.NewRequeueError(time.Second, + fmt.Sprintf("replicas still have pending lifecycle before scale-in: %v", pendingReplicas)) + } + joinedReplicas := make([]string, 0) - err := component.DeleteReplicasStatus(r.protoITS, deleteReplicas, func(s component.ReplicaStatus) { + err = component.DeleteReplicasStatus(r.protoITS, deleteReplicas, func(s component.ReplicaStatus) { // has no member join defined or has joined successfully if s.Provisioned && (s.MemberJoined == nil || *s.MemberJoined) { joinedReplicas = append(joinedReplicas, s.Name) diff --git a/controllers/apps/component/transformer_component_workload_test.go b/controllers/apps/component/transformer_component_workload_test.go index c7121e50478..38b4a6e0a06 100644 --- a/controllers/apps/component/transformer_component_workload_test.go +++ b/controllers/apps/component/transformer_component_workload_test.go @@ -24,6 +24,8 @@ import ( "github.com/golang/mock/gomock" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/utils/ptr" appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1" workloads "github.com/apecloud/kubeblocks/apis/workloads/v1" @@ -198,6 +200,25 @@ var _ = Describe("Component Workload Operations Test", func() { Expect(ops.leaveMemberForPod(pod1, pods)).Should(Succeed()) }) + It("should wait before scale in when target replica lifecycle is pending", func() { + ops.runningItsPodNameSet = sets.New(pod0.Name, pod1.Name) + ops.desiredCompPodNameSet = sets.New(pod0.Name) + + Expect(component.StatusReplicasStatus(ops.protoITS, []string{pod0.Name}, true, true)).Should(Succeed()) + Expect(component.NewReplicasStatus(ops.protoITS, []string{pod1.Name}, true, true)).Should(Succeed()) + ops.protoITS.Spec.Replicas = ptr.To[int32](1) + + err := ops.scaleIn() + Expect(err).Should(HaveOccurred()) + Expect(intctrlutil.IsRequeueError(err)).Should(BeTrue()) + Expect(err.Error()).Should(ContainSubstring("pending lifecycle")) + Expect(err.Error()).Should(ContainSubstring(pod1.Name)) + + pending, err := component.PendingLifecycleReplicas(ops.protoITS, []string{pod1.Name}) + Expect(err).ShouldNot(HaveOccurred()) + Expect(pending).Should(ConsistOf(pod1.Name)) + }) + It("should eliminate upgrade-only diff by preserving legacy config-manager", func() { oldITS := testapps.NewInstanceSetFactory(testCtx.DefaultNamespace, "old-its", clusterName, compName). diff --git a/pkg/controller/component/replicas.go b/pkg/controller/component/replicas.go index 5c319ed847d..aa69e4b8000 100644 --- a/pkg/controller/component/replicas.go +++ b/pkg/controller/component/replicas.go @@ -137,6 +137,16 @@ func DeleteReplicasStatus(its *workloads.InstanceSet, replicas []string, f func( }) } +func PendingLifecycleReplicas(its *workloads.InstanceSet, replicas []string) ([]string, error) { + return GetReplicasStatusFunc(its, func(s ReplicaStatus) bool { + if !slices.Contains(replicas, s.Name) { + return false + } + return (s.DataLoaded != nil && !*s.DataLoaded) || + (s.MemberJoined != nil && !*s.MemberJoined) + }) +} + func StatusReplicasStatus(its *workloads.InstanceSet, replicas []string, hasMemberJoin, hasDataAction bool) error { loaded := func() *bool { if hasDataAction { From 37d4229c85713be056f6c7f6c6a718b077586568 Mon Sep 17 00:00:00 2001 From: Wei Cao Date: Mon, 15 Jun 2026 12:00:53 +0800 Subject: [PATCH 3/4] fix: narrow member join status persistence --- .../transformer_component_workload_ops.go | 11 +--------- .../transformer_component_workload_test.go | 21 ------------------- pkg/controller/component/replicas.go | 10 --------- 3 files changed, 1 insertion(+), 41 deletions(-) diff --git a/controllers/apps/component/transformer_component_workload_ops.go b/controllers/apps/component/transformer_component_workload_ops.go index 9a985c1e946..783ccfd269c 100644 --- a/controllers/apps/component/transformer_component_workload_ops.go +++ b/controllers/apps/component/transformer_component_workload_ops.go @@ -121,17 +121,8 @@ func (r *componentWorkloadOps) scaleIn() error { } deleteReplicas := r.runningItsPodNameSet.Difference(r.desiredCompPodNameSet).UnsortedList() - pendingReplicas, err := component.PendingLifecycleReplicas(r.protoITS, deleteReplicas) - if err != nil { - return err - } - if len(pendingReplicas) > 0 { - return intctrlutil.NewRequeueError(time.Second, - fmt.Sprintf("replicas still have pending lifecycle before scale-in: %v", pendingReplicas)) - } - joinedReplicas := make([]string, 0) - err = component.DeleteReplicasStatus(r.protoITS, deleteReplicas, func(s component.ReplicaStatus) { + err := component.DeleteReplicasStatus(r.protoITS, deleteReplicas, func(s component.ReplicaStatus) { // has no member join defined or has joined successfully if s.Provisioned && (s.MemberJoined == nil || *s.MemberJoined) { joinedReplicas = append(joinedReplicas, s.Name) diff --git a/controllers/apps/component/transformer_component_workload_test.go b/controllers/apps/component/transformer_component_workload_test.go index 38b4a6e0a06..c7121e50478 100644 --- a/controllers/apps/component/transformer_component_workload_test.go +++ b/controllers/apps/component/transformer_component_workload_test.go @@ -24,8 +24,6 @@ import ( "github.com/golang/mock/gomock" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/util/sets" - "k8s.io/utils/ptr" appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1" workloads "github.com/apecloud/kubeblocks/apis/workloads/v1" @@ -200,25 +198,6 @@ var _ = Describe("Component Workload Operations Test", func() { Expect(ops.leaveMemberForPod(pod1, pods)).Should(Succeed()) }) - It("should wait before scale in when target replica lifecycle is pending", func() { - ops.runningItsPodNameSet = sets.New(pod0.Name, pod1.Name) - ops.desiredCompPodNameSet = sets.New(pod0.Name) - - Expect(component.StatusReplicasStatus(ops.protoITS, []string{pod0.Name}, true, true)).Should(Succeed()) - Expect(component.NewReplicasStatus(ops.protoITS, []string{pod1.Name}, true, true)).Should(Succeed()) - ops.protoITS.Spec.Replicas = ptr.To[int32](1) - - err := ops.scaleIn() - Expect(err).Should(HaveOccurred()) - Expect(intctrlutil.IsRequeueError(err)).Should(BeTrue()) - Expect(err.Error()).Should(ContainSubstring("pending lifecycle")) - Expect(err.Error()).Should(ContainSubstring(pod1.Name)) - - pending, err := component.PendingLifecycleReplicas(ops.protoITS, []string{pod1.Name}) - Expect(err).ShouldNot(HaveOccurred()) - Expect(pending).Should(ConsistOf(pod1.Name)) - }) - It("should eliminate upgrade-only diff by preserving legacy config-manager", func() { oldITS := testapps.NewInstanceSetFactory(testCtx.DefaultNamespace, "old-its", clusterName, compName). diff --git a/pkg/controller/component/replicas.go b/pkg/controller/component/replicas.go index aa69e4b8000..5c319ed847d 100644 --- a/pkg/controller/component/replicas.go +++ b/pkg/controller/component/replicas.go @@ -137,16 +137,6 @@ func DeleteReplicasStatus(its *workloads.InstanceSet, replicas []string, f func( }) } -func PendingLifecycleReplicas(its *workloads.InstanceSet, replicas []string) ([]string, error) { - return GetReplicasStatusFunc(its, func(s ReplicaStatus) bool { - if !slices.Contains(replicas, s.Name) { - return false - } - return (s.DataLoaded != nil && !*s.DataLoaded) || - (s.MemberJoined != nil && !*s.MemberJoined) - }) -} - func StatusReplicasStatus(its *workloads.InstanceSet, replicas []string, hasMemberJoin, hasDataAction bool) error { loaded := func() *bool { if hasDataAction { From 07045ea6437184a5e1faa39109bc8178229d5ef6 Mon Sep 17 00:00:00 2001 From: Wei Cao Date: Wed, 24 Jun 2026 05:00:55 +0800 Subject: [PATCH 4/4] fix: remove direct InstanceSet write, rely on DAG/reconcile retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address Leon11 P1: UpdateReplicaStatusWithRetry wrote the live InstanceSet from inside the transformer, bypassing the DAG/plan execution model. ResourceVersion conflict is a normal reconcile failure — the controller retries the whole reconcile and the memberJoin action is idempotent (addon contract requirement), so the status converges naturally through the DAG update path. Co-Authored-By: Claude Opus 4.6 --- .../transformer_component_workload_ops.go | 9 --- pkg/controller/component/replicas.go | 12 ---- pkg/controller/component/replicas_test.go | 56 ------------------- 3 files changed, 77 deletions(-) diff --git a/controllers/apps/component/transformer_component_workload_ops.go b/controllers/apps/component/transformer_component_workload_ops.go index 783ccfd269c..bdfd805b6a1 100644 --- a/controllers/apps/component/transformer_component_workload_ops.go +++ b/controllers/apps/component/transformer_component_workload_ops.go @@ -26,7 +26,6 @@ import ( "time" corev1 "k8s.io/api/core/v1" - "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" @@ -354,14 +353,6 @@ func (r *componentWorkloadOps) joinMember4ScaleOut() error { if err := r.joinMemberForPod(pod, pods); err != nil { joinErrors = append(joinErrors, fmt.Errorf("pod %s: %w", pod.Name, err)) } else { - key := types.NamespacedName{Namespace: r.protoITS.Namespace, Name: r.protoITS.Name} - if err := component.UpdateReplicaStatusWithRetry(r.transCtx.Context, r.cli, key, pod.Name, func(status *component.ReplicaStatus) error { - status.MemberJoined = ptr.To(true) - return nil - }); err != nil { - joinErrors = append(joinErrors, fmt.Errorf("pod %s: persist member joined status: %w", pod.Name, err)) - continue - } replicas.Status[i].MemberJoined = ptr.To(true) } } diff --git a/pkg/controller/component/replicas.go b/pkg/controller/component/replicas.go index 5c319ed847d..1a5c536849b 100644 --- a/pkg/controller/component/replicas.go +++ b/pkg/controller/component/replicas.go @@ -30,7 +30,6 @@ import ( "github.com/go-logr/logr" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/util/retry" "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" @@ -398,14 +397,3 @@ func updateReplicaStatusFunc(ctx context.Context, cli client.Client, } return cli.Update(ctx, its) } - -func UpdateReplicaStatusWithRetry(ctx context.Context, cli client.Client, - key types.NamespacedName, replicaName string, f func(*ReplicaStatus) error) error { - return retry.RetryOnConflict(retry.DefaultBackoff, func() error { - its := &workloads.InstanceSet{} - if err := cli.Get(ctx, key, its); err != nil { - return err - } - return updateReplicaStatusFunc(ctx, cli, its, replicaName, f) - }) -} diff --git a/pkg/controller/component/replicas_test.go b/pkg/controller/component/replicas_test.go index 80af0d7dabb..c45fd4144e8 100644 --- a/pkg/controller/component/replicas_test.go +++ b/pkg/controller/component/replicas_test.go @@ -20,37 +20,18 @@ along with this program. If not, see . package component import ( - "context" "slices" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" - corev1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime/schema" - "k8s.io/apimachinery/pkg/types" "k8s.io/utils/ptr" - "sigs.k8s.io/controller-runtime/pkg/client" workloads "github.com/apecloud/kubeblocks/apis/workloads/v1" "github.com/apecloud/kubeblocks/pkg/constant" ) -type conflictOnceClient struct { - client.Client - conflict bool -} - -func (c *conflictOnceClient) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error { - if c.conflict { - c.conflict = false - return apierrors.NewConflict(schema.GroupResource{Group: "workloads.kubeblocks.io", Resource: "instancesets"}, obj.GetName(), nil) - } - return c.Client.Update(ctx, obj, opts...) -} - var _ = Describe("replicas", func() { var ( its *workloads.InstanceSet @@ -200,43 +181,6 @@ var _ = Describe("replicas", func() { })).Should(Succeed()) }) - It("retries member-joined status update on conflict", func() { - key := types.NamespacedName{Namespace: testCtx.DefaultNamespace, Name: "test-cluster-its-member-join-retry"} - _ = k8sClient.Delete(ctx, &workloads.InstanceSet{ObjectMeta: metav1.ObjectMeta{Namespace: key.Namespace, Name: key.Name}}) - - its.Name = key.Name - its.Namespace = key.Namespace - its.Spec.Selector = &metav1.LabelSelector{MatchLabels: map[string]string{"app": key.Name}} - its.Spec.Template = corev1.PodTemplateSpec{ - ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": key.Name}}, - Spec: corev1.PodSpec{ - Containers: []corev1.Container{{Name: "test", Image: "test"}}, - }, - } - Expect(StatusReplicasStatus(its, replicas, true, true)).Should(Succeed()) - its.Spec.Replicas = ptr.To[int32](4) - newReplica := "test-cluster-its-3" - Expect(NewReplicasStatus(its, []string{newReplica}, true, false)).Should(Succeed()) - Expect(k8sClient.Create(ctx, its)).Should(Succeed()) - - cli := &conflictOnceClient{Client: k8sClient, conflict: true} - Expect(UpdateReplicaStatusWithRetry(ctx, cli, key, newReplica, func(status *ReplicaStatus) error { - status.MemberJoined = ptr.To(true) - return nil - })).Should(Succeed()) - - got := &workloads.InstanceSet{} - Expect(k8sClient.Get(ctx, key, got)).Should(Succeed()) - status, err := getReplicasStatus(got) - Expect(err).Should(BeNil()) - i := slices.IndexFunc(status.Status, func(s ReplicaStatus) bool { - return s.Name == newReplica - }) - Expect(i).Should(BeNumerically(">=", 0)) - Expect(status.Status[i].MemberJoined).ShouldNot(BeNil()) - Expect(*status.Status[i].MemberJoined).Should(BeTrue()) - }) - // It("task event for new replicas - succeed", func() { // Expect(StatusReplicasStatus(its, replicas, true, true)).Should(Succeed()) //