fix(store): cannot scale in pending pod of store (#6531)

Author: liubog2008

pkg/controllers/common/store_offline.go

@@ -39,12 +39,16 @@ import (
 // - B1: state == Preparing || state == Serving
 // - B2: state == Removing
 // - B3: state == Removed
+// - B4: store does not exist
 // 1. (A1, B1): do nothing, waiting for changes of spec.offline
 // 2. (A2, B1): call delete store api
 // 3. (A1, B2): call cancel api
 // 4. (A2, B2): do nothing, waiting for changes of state
 // 5. (A1, B3): do nothing, waiting for instance being removed
 // 6. (A2, B3): do nothing, waiting for instance being removed
+// 7. (A1, B4): do nothing, waiting for pod being created
+// 8. (A2, B4): do nothing, waiting for instance being removed
+// nolint:gocyclo // refactor later if necessary
 func TaskOfflineStore[
 	S scope.Instance[F, T],
 	F client.Object,
@@ -57,31 +61,39 @@ func TaskOfflineStore[
 	storeID string,
 	state pdv1.NodeState,
 ) error {
-	if storeID == "" {
-		// can do nothing until store id is found
-		return fmt.Errorf("%w: store does not exists", task.ErrWait)
-	}
 	isOffline := coreutil.IsOffline[S](obj)
 
 	switch {
+	case !isOffline && storeID == "":
+		// do nothing because store does not exist
+		return nil
+
+	case isOffline && storeID == "":
+		return fmt.Errorf("%w: store does not exist and is offline", task.ErrWait)
+
 	case isOffline && (state == pdv1.NodeStatePreparing || state == pdv1.NodeStateServing):
 		if err := c.DeleteStore(ctx, storeID); err != nil {
 			return err
 		}
 
 		return fmt.Errorf("%w: node state should be changed to Removing", task.ErrWait)
+
 	case isOffline && state == pdv1.NodeStateRemoving:
 		return fmt.Errorf("%w: node state should be changed to Removed", task.ErrWait)
+
 	case isOffline && state == pdv1.NodeStateRemoved:
 		return nil
+
 	case !isOffline && (state == pdv1.NodeStatePreparing || state == pdv1.NodeStateServing):
 		// do nothing
 		return nil
+
 	case !isOffline && state == pdv1.NodeStateRemoving:
 		if err := c.CancelDeleteStore(ctx, storeID); err != nil {
 			return err
 		}
 		return fmt.Errorf("%w: node state should be changed to Preparing or Serving", task.ErrWait)
+
 	case !isOffline && state == pdv1.NodeStateRemoved:
 		// store has been removed, cannot cancel the deletion
 		return nil
@@ -108,19 +120,18 @@ func TaskInstanceConditionOffline[
 		isOffline := coreutil.IsOffline[S](instance)
 		state := s.GetStoreState()
 
-		if state == "" {
-			return task.Wait().With("wait for state is synced")
-		}
-
 		var needUpdate, isCompleted bool
 		var reason string
 		switch {
+		case isOffline && state == "":
+			reason = v1alpha1.ReasonOfflineCompleted
+			isCompleted = true
 		case state == pdv1.NodeStateRemoved:
 			reason = v1alpha1.ReasonOfflineCompleted
 			isCompleted = true
 		case isOffline:
 			reason = v1alpha1.ReasonOfflineProcessing
-		case !isOffline && (state == pdv1.NodeStatePreparing || state == pdv1.NodeStateServing):
+		case !isOffline && (state == "" || state == pdv1.NodeStatePreparing || state == pdv1.NodeStateServing):
 			reason = ""
 		case !isOffline && state == pdv1.NodeStateRemoving:
 			reason = v1alpha1.ReasonOfflineCanceling

pkg/controllers/common/store_offline_test.go

@@ -56,7 +56,13 @@ func TestTaskOfflineStore(t *testing.T) {
 		isWaitErr     bool
 	}{
 		{
-			desc:      "empty store id",
+			desc:   "spec.offline == false, empty store id",
+			obj:    newFakeStore(false, nil),
+			hasErr: false,
+		},
+		{
+			desc:      "spec.offline == true, empty store id",
+			obj:       newFakeStore(true, nil),
 			hasErr:    true,
 			isWaitErr: true,
 		},
@@ -169,11 +175,18 @@ func TestTaskInstanceConditionOffline(t *testing.T) {
 		expectedObj           *v1alpha1.TiKV
 	}{
 		{
-			desc:           "empty state",
+			desc:           "spec.offline == false, empty state",
 			obj:            newFakeStore(false, nil),
-			expectedStatus: task.SWait,
+			expectedStatus: task.SComplete,
 			expectedObj:    newFakeStore(false, nil),
 		},
+		{
+			desc:                  "spec.offline == true, empty state",
+			obj:                   newFakeStore(true, nil),
+			expectedStatusChanged: true,
+			expectedStatus:        task.SComplete,
+			expectedObj:           newFakeStore(true, coreutil.Offlined()),
+		},
 		{
 			desc:       "spec.offline == false, state == Preparing",
 			obj:        newFakeStore(false, nil),

pkg/controllers/tiflash/builder.go

@@ -76,9 +76,9 @@ func (r *Reconciler) NewRunner(state *tasks.ReconcileContext, reporter task.Task
 		// only set ready if pd is synced
 		task.If(PDIsSynced(state),
 			common.TaskInstanceConditionReady[scope.TiFlash](state),
+			common.TaskInstanceConditionOffline[scope.TiFlash](state),
 		),
 		common.TaskInstanceConditionRunning[scope.TiFlash](state),
-		common.TaskInstanceConditionOffline[scope.TiFlash](state),
 		tasks.TaskStatus(state, r.Client),
 	)
 

pkg/controllers/tikv/builder.go

@@ -43,7 +43,7 @@ func (r *Reconciler) NewRunner(state *tasks.ReconcileContext, reporter task.Task
 			tasks.TaskFinalizerDel(state, r.Client),
 		),
 
-		// if instance is deleting and store is removed
+		// if instance is not deleting but store is offlined
 		task.IfBreak(common.CondObjectIsNotDeletingButOfflined[scope.TiKV](state),
 			common.TaskDeleteOfflinedStore[scope.TiKV](state, r.Client),
 		),
@@ -86,9 +86,9 @@ func (r *Reconciler) NewRunner(state *tasks.ReconcileContext, reporter task.Task
 		// only set ready if pd is synced
 		task.If(PDIsSynced(state),
 			common.TaskInstanceConditionReady[scope.TiKV](state),
+			common.TaskInstanceConditionOffline[scope.TiKV](state),
 		),
 		common.TaskInstanceConditionRunning[scope.TiKV](state),
-		common.TaskInstanceConditionOffline[scope.TiKV](state),
 		tasks.TaskStatus(state, r.Client),
 	)
 

pkg/controllers/tikv/tasks/evict_leader.go

@@ -46,6 +46,7 @@ func TaskEvictLeader(state *ReconcileContext) task.Task {
 }
 
 // TaskEndEvictLeader only be called when object is deleting and store has been removed
+// TODO(liubo02): it's not stable because status.ID may be lost
 func TaskEndEvictLeader(state *ReconcileContext) task.Task {
 	return task.NameTaskFunc("EndEvictLeader", func(ctx context.Context) task.Result {
 		msg := "ensure evict leader scheduler doesn't exist"

pkg/controllers/tikv/tasks/pod.go

@@ -92,9 +92,6 @@ func TaskPod(state *ReconcileContext, c client.Client) task.Task {
 
 		if !reloadable.CheckTiKVPod(state.TiKV(), pod) {
 			logger.Info("will recreate the pod")
-			if err := k8s.RemoveFinalizer(ctx, c, pod); err != nil {
-				return task.Fail().With("cannot remove finalizer: %w", err)
-			}
 			regionCount := 0
 			if state.Store != nil {
 				regionCount = state.Store.RegionCount

pkg/updater/builder.go

@@ -71,6 +71,8 @@ func (b *builder[T, O, R]) Build() Executor {
 
 	updatePolicies := b.updatePreferPolicies
 	updatePolicies = append(updatePolicies, PreferNotRunning[R](), PreferUnready[R]())
+	scaleInPolicies := b.scaleInPreferPolicies
+	scaleInPolicies = append(scaleInPolicies, PreferNotRunning[R](), PreferUnready[R]())
 	actor := &actor[T, O, R]{
 		c: b.c,
 		f: b.f,
@@ -86,7 +88,7 @@ func (b *builder[T, O, R]) Build() Executor {
 		updateHooks: append(b.updateHooks, KeepName[R](), KeepTopology[R](), KeepResourceVersion[R]()),
 		delHooks:    b.delHooks,
 
-		scaleInSelector: NewSelector(b.scaleInPreferPolicies...),
+		scaleInSelector: NewSelector(scaleInPolicies...),
 		updateSelector:  NewSelector(updatePolicies...),
 	}
 	return NewExecutor(

tests/e2e/data/tikv.go

@@ -15,6 +15,7 @@
 package data
 
 import (
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
@@ -108,3 +109,34 @@ s3-region = "local"
 `
 	})
 }
+
+func WithTiKVPodAntiAffinity() GroupPatch[*v1alpha1.TiKVGroup] {
+	return GroupPatchFunc[*v1alpha1.TiKVGroup](func(obj *v1alpha1.TiKVGroup) {
+		if obj.Spec.Template.Spec.Overlay == nil {
+			obj.Spec.Template.Spec.Overlay = &v1alpha1.Overlay{}
+		}
+		o := obj.Spec.Template.Spec.Overlay
+		if o.Pod == nil {
+			o.Pod = &v1alpha1.PodOverlay{}
+		}
+		if o.Pod.Spec == nil {
+			o.Pod.Spec = &corev1.PodSpec{}
+		}
+
+		o.Pod.Spec.Affinity = &corev1.Affinity{
+			PodAntiAffinity: &corev1.PodAntiAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+					{
+						LabelSelector: &metav1.LabelSelector{
+							MatchLabels: map[string]string{
+								"pingcap.com/component": "tikv",
+								"pingcap.com/group":     obj.GetName(),
+							},
+						},
+						TopologyKey: "kubernetes.io/hostname",
+					},
+				},
+			},
+		}
+	})
+}

tests/e2e/e2e_test.go

@@ -37,6 +37,7 @@ import (
 	// TODO: move all tests into suite
 	_ "github.com/pingcap/tidb-operator/tests/e2e/suite/availability"
 	_ "github.com/pingcap/tidb-operator/tests/e2e/suite/cluster"
+	_ "github.com/pingcap/tidb-operator/tests/e2e/suite/scale"
 )
 
 func TestE2E(t *testing.T) {

tests/e2e/suite/scale/tikv.go

@@ -0,0 +1,142 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package scale
+
+import (
+	"context"
+	"time"
+
+	"github.com/onsi/ginkgo/v2"
+	"k8s.io/utils/ptr"
+
+	"github.com/pingcap/tidb-operator/pkg/client"
+	"github.com/pingcap/tidb-operator/pkg/runtime"
+	"github.com/pingcap/tidb-operator/pkg/runtime/scope"
+	"github.com/pingcap/tidb-operator/tests/e2e/data"
+	"github.com/pingcap/tidb-operator/tests/e2e/framework"
+	"github.com/pingcap/tidb-operator/tests/e2e/label"
+	"github.com/pingcap/tidb-operator/tests/e2e/utils/waiter"
+)
+
+var _ = ginkgo.Describe("Scale TiKV", label.TiKV, label.MultipleAZ, label.P0, label.Scale, func() {
+	f := framework.New()
+	f.Setup()
+
+	ginkgo.It("support scale out from 3 to 4", func(ctx context.Context) {
+		pdg := f.MustCreatePD(ctx)
+		kvg := f.MustCreateTiKV(ctx,
+			data.WithReplicas[scope.TiKVGroup](3),
+			data.WithTiKVEvenlySpreadPolicy(),
+		)
+
+		f.WaitForPDGroupReady(ctx, pdg)
+		f.WaitForTiKVGroupReady(ctx, kvg)
+
+		f.MustEvenlySpreadTiKV(ctx, kvg)
+
+		patch := client.MergeFrom(kvg.DeepCopy())
+		kvg.Spec.Replicas = ptr.To[int32](4)
+
+		f.Must(f.Client.Patch(ctx, kvg, patch))
+		f.WaitForTiKVGroupReady(ctx, kvg)
+		f.MustEvenlySpreadTiKV(ctx, kvg)
+	})
+
+	ginkgo.It("support scale in from 4 to 3", func(ctx context.Context) {
+		pdg := f.MustCreatePD(ctx)
+		kvg := f.MustCreateTiKV(ctx,
+			data.WithReplicas[scope.TiKVGroup](4),
+			data.WithTiKVEvenlySpreadPolicy(),
+		)
+
+		f.WaitForPDGroupReady(ctx, pdg)
+		f.WaitForTiKVGroupReady(ctx, kvg)
+
+		f.MustEvenlySpreadTiKV(ctx, kvg)
+
+		patch := client.MergeFrom(kvg.DeepCopy())
+		kvg.Spec.Replicas = ptr.To[int32](3)
+
+		f.Must(f.Client.Patch(ctx, kvg, patch))
+		f.WaitForTiKVGroupReady(ctx, kvg)
+		f.MustEvenlySpreadTiKV(ctx, kvg)
+	})
+
+	ginkgo.It("support scale in from 4 to 3 with a pending pod", func(ctx context.Context) {
+		pdg := f.MustCreatePD(ctx)
+		kvg := f.MustCreateTiKV(ctx,
+			data.WithReplicas[scope.TiKVGroup](3),
+			data.WithTiKVEvenlySpreadPolicy(),
+			data.WithTiKVPodAntiAffinity(),
+		)
+
+		f.WaitForPDGroupReady(ctx, pdg)
+		f.WaitForTiKVGroupReady(ctx, kvg)
+
+		f.MustEvenlySpreadTiKV(ctx, kvg)
+
+		patch := client.MergeFrom(kvg.DeepCopy())
+		kvg.Spec.Replicas = ptr.To[int32](4)
+		f.Must(f.Client.Patch(ctx, kvg, patch))
+
+		// just wait a few seconds
+		time.Sleep(time.Second * 15)
+
+		patch2 := client.MergeFrom(kvg.DeepCopy())
+		kvg.Spec.Replicas = ptr.To[int32](3)
+
+		f.Must(f.Client.Patch(ctx, kvg, patch2))
+		f.WaitForTiKVGroupReady(ctx, kvg)
+		f.MustEvenlySpreadTiKV(ctx, kvg)
+	})
+
+	ginkgo.It("support scale from 3 to 6 and rolling update at same time", ginkgo.Serial, label.Update, func(ctx context.Context) {
+		pdg := f.MustCreatePD(ctx)
+		kvg := f.MustCreateTiKV(ctx,
+			data.WithReplicas[scope.TiKVGroup](3),
+			data.WithTiKVEvenlySpreadPolicy(),
+		)
+
+		f.WaitForPDGroupReady(ctx, pdg)
+		f.WaitForTiKVGroupReady(ctx, kvg)
+		f.MustEvenlySpreadTiKV(ctx, kvg)
+
+		patch := client.MergeFrom(kvg.DeepCopy())
+		kvg.Spec.Replicas = ptr.To[int32](6)
+		kvg.Spec.Template.Annotations = map[string]string{
+			"test": "test",
+		}
+
+		nctx, cancel := context.WithCancel(ctx)
+		ch := make(chan struct{})
+		go func() {
+			defer close(ch)
+			defer ginkgo.GinkgoRecover()
+			f.Must(waiter.WaitPodsRollingUpdateOnce(nctx, f.Client, runtime.FromTiKVGroup(kvg), 3, 0, waiter.LongTaskTimeout))
+		}()
+
+		changeTime, err := waiter.MaxPodsCreateTimestamp(ctx, f.Client, runtime.FromTiKVGroup(kvg))
+		f.Must(err)
+
+		ginkgo.By("Change config and replicas of the TiKVGroup")
+		f.Must(f.Client.Patch(ctx, kvg, patch))
+		f.Must(waiter.WaitForPodsRecreated(ctx, f.Client, runtime.FromTiKVGroup(kvg), *changeTime, waiter.LongTaskTimeout))
+		f.WaitForTiKVGroupReady(ctx, kvg)
+		cancel()
+		<-ch
+
+		f.MustEvenlySpreadTiKV(ctx, kvg)
+	})
+})