fix(updater): optimize no inplace update (#6533)

liubog2008 · web-flow · commit c0099a5ee31f · 2025-11-10T10:55:50.000Z
diff --git a/pkg/controllers/pdgroup/tasks/updater.go b/pkg/controllers/pdgroup/tasks/updater.go
@@ -100,7 +100,7 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
 				NotLeaderPolicy(),
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
 			).
 			WithUpdatePreferPolicy(
 				NotLeaderPolicy(),
diff --git a/pkg/controllers/replicationworkergroup/tasks/updater.go b/pkg/controllers/replicationworkergroup/tasks/updater.go
@@ -97,7 +97,7 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithDelHooks(topoPolicy).
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
 			).
 			WithMinReadySeconds(coreutil.MinReadySeconds[scope.ReplicationWorkerGroup](obj)).
 			Build().
diff --git a/pkg/controllers/schedulergroup/tasks/updater.go b/pkg/controllers/schedulergroup/tasks/updater.go
@@ -100,7 +100,7 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithDelHooks(topoPolicy).
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
 			).
 			WithMinReadySeconds(coreutil.MinReadySeconds[scope.SchedulerGroup](obj)).
 			Build().
diff --git a/pkg/controllers/schedulinggroup/tasks/updater.go b/pkg/controllers/schedulinggroup/tasks/updater.go
@@ -100,7 +100,7 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithDelHooks(topoPolicy).
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
 			).
 			WithMinReadySeconds(coreutil.MinReadySeconds[scope.SchedulingGroup](obj)).
 			Build().
diff --git a/pkg/controllers/ticdcgroup/tasks/updater.go b/pkg/controllers/ticdcgroup/tasks/updater.go
@@ -113,7 +113,10 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithDelHooks(topoPolicy).
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
+			).
+			WithUpdatePreferPolicy(
+				topoPolicy.PolicyUpdate(),
 			).
 			WithNoInPaceUpdate(noUpdate).
 			WithMinReadySeconds(coreutil.MinReadySeconds[scope.TiCDCGroup](cdcg)).
diff --git a/pkg/controllers/tidbgroup/tasks/updater.go b/pkg/controllers/tidbgroup/tasks/updater.go
@@ -114,7 +114,10 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithDelHooks(topoPolicy).
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
+			).
+			WithUpdatePreferPolicy(
+				topoPolicy.PolicyUpdate(),
 			).
 			WithNoInPaceUpdate(noUpdate).
 			WithMinReadySeconds(coreutil.MinReadySeconds[scope.TiDBGroup](dbg)).
diff --git a/pkg/controllers/tiflashgroup/tasks/updater.go b/pkg/controllers/tiflashgroup/tasks/updater.go
@@ -100,7 +100,7 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithDelHooks(topoPolicy).
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
 			).
 			WithMinReadySeconds(coreutil.MinReadySeconds[scope.TiFlashGroup](fg)).
 			Build().
diff --git a/pkg/controllers/tikvgroup/tasks/updater.go b/pkg/controllers/tikvgroup/tasks/updater.go
@@ -101,7 +101,7 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithDelHooks(topoPolicy).
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
 			).
 			WithMinReadySeconds(coreutil.MinReadySeconds[scope.TiKVGroup](kvg)).
 			Build().
diff --git a/pkg/controllers/tiproxygroup/tasks/updater.go b/pkg/controllers/tiproxygroup/tasks/updater.go
@@ -110,7 +110,12 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			).
 			WithDelHooks(topoPolicy).
 			WithUpdateHooks(topoPolicy).
-			WithScaleInPreferPolicy(topoPolicy).
+			WithScaleInPreferPolicy(
+				topoPolicy.PolicyScaleIn(),
+			).
+			WithUpdatePreferPolicy(
+				topoPolicy.PolicyUpdate(),
+			).
 			WithNoInPaceUpdate(noUpdate).
 			WithMinReadySeconds(coreutil.MinReadySeconds[scope.TiProxyGroup](proxyg)).
 			Build().
diff --git a/pkg/controllers/tsogroup/tasks/updater.go b/pkg/controllers/tsogroup/tasks/updater.go
@@ -101,7 +101,7 @@ func TaskUpdater(state *ReconcileContext, c client.Client, af tracker.AllocateFa
 			WithUpdateHooks(topoPolicy).
 			WithScaleInPreferPolicy(
 				NotLeaderPolicy(),
-				topoPolicy,
+				topoPolicy.PolicyScaleIn(),
 			).
 			WithUpdatePreferPolicy(
 				NotLeaderPolicy(),
diff --git a/pkg/updater/actor.go b/pkg/updater/actor.go
@@ -222,12 +222,12 @@ func (act *actor[T, O, R]) ScaleInOutdated(ctx context.Context) (bool, error) {
 func (act *actor[T, O, R]) scaleInOutdated(ctx context.Context, name string, deferDel bool) (bool, error) {
 	logger := logr.FromContextOrDiscard(ctx)
 	if name == "" {
-		choosed, err := act.chooseToScaleIn(act.outdated.List())
+		chosen, err := act.chooseToScaleIn(act.outdated.List())
 		if err != nil {
 			return false, err
 		}
 
-		name = choosed
+		name = chosen
 	}
 
 	obj := act.outdated.Del(name)
@@ -306,23 +306,29 @@ func (act *actor[T, O, R]) deferDelete(ctx context.Context, obj R) error {
 // Recreate update strategy (noInPlaceUpdate=true):
 // - Scales in an outdated instance (immediate deletion)
 // - Scales out a new instance to replace it
-func (act *actor[T, O, R]) Update(ctx context.Context) error {
+func (act *actor[T, O, R]) Update(ctx context.Context, unavailable bool) error {
 	logger := logr.FromContextOrDiscard(ctx)
 
 	name, err := act.chooseToUpdate(act.outdated.List())
 	if err != nil {
 		return err
 	}
-
-	if act.noInPlaceUpdate {
-		if _, err := act.scaleInOutdated(ctx, name, false); err != nil {
-			return err
-		}
-		if err := act.ScaleOut(ctx); err != nil {
-			return err
+	chosen := act.outdated.Get(name)
+	// 1. need an unavailable outdated
+	// 2. the chosen one is not unavailable outdated
+	if unavailable && !chosen.IsNotRunning() {
+		if act.noInPlaceUpdate {
+			if _, err := act.scaleInOutdated(ctx, name, false); err != nil {
+				return err
+			}
+			if err := act.ScaleOut(ctx); err != nil {
+				return err
+			}
+
+			return nil
 		}
 
-		return nil
+		return fmt.Errorf("want to update an unavailable outdated, but the chosen one %s is not", name)
 	}
 
 	outdated := act.outdated.Del(name)
diff --git a/pkg/updater/builder.go b/pkg/updater/builder.go
@@ -69,10 +69,18 @@ type builder[T runtime.Tuple[O, R], O client.Object, R runtime.Instance] struct
 func (b *builder[T, O, R]) Build() Executor {
 	update, outdated, beingOffline, deleted := split(b.instances, b.rev)
 
-	updatePolicies := b.updatePreferPolicies
-	updatePolicies = append(updatePolicies, PreferNotRunning[R](), PreferUnready[R]())
-	scaleInPolicies := b.scaleInPreferPolicies
-	scaleInPolicies = append(scaleInPolicies, PreferNotRunning[R](), PreferUnready[R]())
+	updatePolicies := []PreferPolicy[R]{
+		PreferUnready[R](),
+		PreferNotRunning[R](),
+	}
+	updatePolicies = append(updatePolicies, b.updatePreferPolicies...)
+
+	scaleInPolicies := []PreferPolicy[R]{
+		PreferUnready[R](),
+		PreferNotRunning[R](),
+	}
+	scaleInPolicies = append(scaleInPolicies, b.scaleInPreferPolicies...)
+
 	actor := &actor[T, O, R]{
 		c: b.c,
 		f: b.f,
diff --git a/pkg/updater/executor.go b/pkg/updater/executor.go
@@ -23,9 +23,19 @@ import (
 const verbose = 10
 
 type Actor interface {
+	// ScaleOut creates a new update instance
 	ScaleOut(ctx context.Context) error
-	Update(ctx context.Context) error
+
+	// Update will update an instance CR
+	// unavailable means update must choose an unavailable instance to update
+	Update(ctx context.Context, unavailable bool) error
+
+	// ScaleInOutdated will scale in an outdated instance
+	// If unavailable is true, the chosen instance is unavailable
 	ScaleInUpdate(ctx context.Context) (unavailable bool, _ error)
+
+	// ScaleInUpdate will scale in an updated instance
+	// If unavailable is true, the chosen instance is unavailable
 	ScaleInOutdated(ctx context.Context) (unavailable bool, _ error)
 
 	// Cleanup deletes all instances marked as defer deletion
@@ -106,7 +116,7 @@ func (ex *executor) Do(ctx context.Context) (bool, error) {
 				// unavailable and outdated instances
 				if ex.unavailableOutdated > 0 {
 					logger.V(verbose).Info("update unavailable outdated")
-					if err := ex.act.Update(ctx); err != nil {
+					if err := ex.act.Update(ctx, true); err != nil {
 						return false, err
 					}
 					ex.outdated -= 1
@@ -122,7 +132,7 @@ func (ex *executor) Do(ctx context.Context) (bool, error) {
 					}
 
 					logger.V(verbose).Info("update available outdated")
-					if err := ex.act.Update(ctx); err != nil {
+					if err := ex.act.Update(ctx, false); err != nil {
 						return false, err
 					}
 					ex.outdated -= 1
diff --git a/pkg/updater/executor_test.go b/pkg/updater/executor_test.go
@@ -40,27 +40,27 @@ func (a *FakeActor) ScaleOut(_ context.Context) error {
 
 func (a *FakeActor) ScaleInOutdated(_ context.Context) (unavailable bool, err error) {
 	a.Actions = append(a.Actions, actionScaleInOutdated)
-	a.outdated -= 1
+	a.outdated--
 	if a.preferAvailable || a.unavailableOutdated == 0 {
 		return false, nil
 	}
 
-	a.unavailableOutdated -= 1
+	a.unavailableOutdated--
 	return true, nil
 }
 
 func (a *FakeActor) ScaleInUpdate(_ context.Context) (unavailable bool, err error) {
 	a.Actions = append(a.Actions, actionScaleInUpdate)
-	a.update -= 1
+	a.update--
 	if a.preferAvailable || a.unavailableUpdate == 0 {
 		return false, nil
 	}
 
-	a.unavailableUpdate -= 1
+	a.unavailableUpdate--
 	return true, nil
 }
 
-func (a *FakeActor) Update(_ context.Context) error {
+func (a *FakeActor) Update(_ context.Context, _ bool) error {
 	a.Actions = append(a.Actions, actionUpdate)
 	return nil
 }
@@ -70,10 +70,6 @@ func (a *FakeActor) Cleanup(_ context.Context) error {
 	return nil
 }
 
-func (a *FakeActor) RecordedActions() []action {
-	return a.Actions
-}
-
 func TestExecutor(t *testing.T) {
 	cases := []struct {
 		desc                string
diff --git a/pkg/updater/policy/topology.go b/pkg/updater/policy/topology.go
@@ -34,7 +34,8 @@ type TopologyPolicy[R runtime.Instance] interface {
 	updater.AddHook[R]
 	updater.DelHook[R]
 	updater.UpdateHook[R]
-	updater.PreferPolicy[R]
+	PolicyScaleIn() updater.PreferPolicy[R]
+	PolicyUpdate() updater.PreferPolicy[R]
 }
 
 func NewTopologyPolicy[R runtime.Instance](ts []v1alpha1.ScheduleTopology, rev string, rs ...R) (TopologyPolicy[R], error) {
@@ -94,12 +95,28 @@ func (p *topologyPolicy[R]) Delete(name string) {
 	p.updated.Del(name)
 }
 
-func (p *topologyPolicy[R]) Prefer(allowed []R) []R {
+func (p *topologyPolicy[R]) PolicyScaleIn() updater.PreferPolicy[R] {
+	return &deletePreferPolicy[R]{
+		p: p,
+	}
+}
+
+func (p *topologyPolicy[R]) PolicyUpdate() updater.PreferPolicy[R] {
+	return &updatePreferPolicy[R]{
+		p: p,
+	}
+}
+
+type deletePreferPolicy[R runtime.Instance] struct {
+	p *topologyPolicy[R]
+}
+
+func (p *deletePreferPolicy[R]) Prefer(allowed []R) []R {
 	if len(allowed) == 0 {
 		return nil
 	}
-	names := p.all.NextDel()
-	preferred := make([]R, 0, len(allowed))
+	names := p.p.all.NextDel()
+	var preferred []R
 	for _, item := range allowed {
 		for _, name := range names {
 			if item.GetName() == name {
@@ -111,9 +128,34 @@ func (p *topologyPolicy[R]) Prefer(allowed []R) []R {
 	return preferred
 }
 
+type updatePreferPolicy[R runtime.Instance] struct {
+	p *topologyPolicy[R]
+}
+
+// Choose a preferred item to update
+// Update will not change topology spreads of "all" set.
+// However, spreads of "update" set will be changed.
+func (p *updatePreferPolicy[R]) Prefer(allowed []R) []R {
+	if len(allowed) == 0 {
+		return nil
+	}
+	topos := p.p.updated.NextAdd()
+
+	var preferred []R
+	for _, item := range allowed {
+		for _, topo := range topos {
+			if maps.Equal(topo, item.GetTopology()) {
+				preferred = append(preferred, item)
+			}
+		}
+	}
+
+	return preferred
+}
+
 // choose a preferred topology
-// - prefer all instances are well spread
-// - if no
+// - prefer "all" and "update" set are well spread, choose the intersection
+// - if no intersection, just return the first one of "all" set
 func choose(all, update []v1alpha1.Topology) v1alpha1.Topology {
 	// No topology is preferred
 	// Normally because of no topology policy is specified
diff --git a/pkg/updater/policy/topology_test.go b/pkg/updater/policy/topology_test.go
diff --git a/tests/e2e/framework/action/test.go b/tests/e2e/framework/action/test.go
diff --git a/tests/e2e/tidb/tidb.go b/tests/e2e/tidb/tidb.go
diff --git a/tests/e2e/tidb/topology.go b/tests/e2e/tidb/topology.go
diff --git a/tests/e2e/tiproxy/tiproxy.go b/tests/e2e/tiproxy/tiproxy.go
diff --git a/tests/e2e/utils/waiter/pod.go b/tests/e2e/utils/waiter/pod.go
