schedule: skip slow/evicted stores in hot region scheduling (#10946)

pkg/schedule/filter/counter.go

@@ -52,6 +52,7 @@ const (
 	engine
 	specialUse
 	isolation
+	leaderEvicted
 
 	storeStateOK
 	storeStateTombstone
@@ -82,6 +83,7 @@ var filters = [filtersLen]string{
 	"engine-filter",
 	"special-use-filter",
 	"isolation-filter",
+	"leader-evicted-filter",
 
 	"store-state-ok-filter",
 	"store-state-tombstone-filter",

pkg/schedule/filter/filters.go

@@ -558,6 +558,37 @@ func (f *StoreStateFilter) Target(conf config.SharedConfigProvider, store *core.
 	return statusOK
 }
 
+type hotRegionEvictedTargetFilter struct{ scope string }
+
+// NewHotRegionEvictedTargetFilter creates a filter that rejects stores already
+// marked as unsuitable hot-region targets.
+func NewHotRegionEvictedTargetFilter(scope string) Filter {
+	return &hotRegionEvictedTargetFilter{scope: scope}
+}
+
+// Scope returns the scheduler or the checker which the filter acts on.
+func (f *hotRegionEvictedTargetFilter) Scope() string {
+	return f.scope
+}
+
+// Type returns the type of the filter.
+func (*hotRegionEvictedTargetFilter) Type() filterType {
+	return leaderEvicted
+}
+
+// Source filters stores when select them as schedule source.
+func (*hotRegionEvictedTargetFilter) Source(config.SharedConfigProvider, *core.StoreInfo) *plan.Status {
+	return statusOK
+}
+
+// Target filters stores when select them as schedule target.
+func (*hotRegionEvictedTargetFilter) Target(_ config.SharedConfigProvider, store *core.StoreInfo) *plan.Status {
+	if store.EvictedAsSlowStore() || store.EvictedAsStoppingStore() || store.IsEvictedAsSlowTrend() {
+		return statusStoreRejectLeader
+	}
+	return statusOK
+}
+
 // labelConstraintFilter is a filter that selects stores satisfy the constraints.
 type labelConstraintFilter struct {
 	scope       string

pkg/schedule/filter/filters_test.go

@@ -287,6 +287,28 @@ func TestStoreStateFilter(t *testing.T) {
 	check(store, testCases)
 }
 
+func TestHotRegionEvictedTargetFilter(t *testing.T) {
+	re := require.New(t)
+	filter := NewHotRegionEvictedTargetFilter("")
+	opt := mockconfig.NewTestOptions()
+
+	testCases := []struct {
+		name      string
+		store     *core.StoreInfo
+		targetRes plan.StatusCode
+	}{
+		{name: "normal", store: core.NewStoreInfoWithLabel(1, map[string]string{}), targetRes: plan.StatusOK},
+		{name: "slow-store", store: core.NewStoreInfoWithLabel(1, map[string]string{}).Clone(core.SlowStoreEvicted()), targetRes: plan.StatusStoreRejectLeader},
+		{name: "stopping-store", store: core.NewStoreInfoWithLabel(1, map[string]string{}).Clone(core.StoppingStoreEvicted()), targetRes: plan.StatusStoreRejectLeader},
+		{name: "slow-trend", store: core.NewStoreInfoWithLabel(1, map[string]string{}).Clone(core.SlowTrendEvicted()), targetRes: plan.StatusStoreRejectLeader},
+	}
+
+	for _, testCase := range testCases {
+		re.Equal(plan.StatusOK, filter.Source(opt, testCase.store).StatusCode, testCase.name)
+		re.Equal(testCase.targetRes, filter.Target(opt, testCase.store).StatusCode, testCase.name)
+	}
+}
+
 func TestStoreStateFilterReason(t *testing.T) {
 	re := require.New(t)
 	filters := []Filter{

pkg/schedule/schedulers/grant_hot_region.go

@@ -292,6 +292,7 @@ func (s *grantHotRegionScheduler) transfer(cluster sche.SchedulerCluster, region
 		candidate = []uint64{s.conf.getStoreLeaderID()}
 	} else {
 		filters = append(filters, &filter.StoreStateFilter{ActionScope: s.GetName(), MoveRegion: true, OperatorLevel: constant.High},
+			filter.NewHotRegionEvictedTargetFilter(s.GetName()),
 			filter.NewExcludedFilter(s.GetName(), srcRegion.GetStoreIDs(), srcRegion.GetStoreIDs()))
 		candidate = storeIDs
 	}

pkg/schedule/schedulers/hot_region.go

@@ -1001,6 +1001,7 @@ func (bs *balanceSolver) filterDstStores() map[uint64]*statistics.StoreLoadDetai
 		}
 		filters = []filter.Filter{
 			&filter.StoreStateFilter{ActionScope: bs.sche.GetName(), MoveRegion: true, OperatorLevel: constant.High},
+			filter.NewHotRegionEvictedTargetFilter(bs.sche.GetName()),
 			filter.NewExcludedFilter(bs.sche.GetName(), bs.cur.region.GetStoreIDs(), bs.cur.region.GetStoreIDs()),
 			filter.NewSpecialUseFilter(bs.sche.GetName(), filter.SpecialUseHotRegion),
 			filter.NewPlacementSafeguard(bs.sche.GetName(), bs.GetSchedulerConfig(), bs.GetBasicCluster(), bs.GetRuleManager(), bs.cur.region, srcStore, nil),
@@ -1019,7 +1020,10 @@ func (bs *balanceSolver) filterDstStores() map[uint64]*statistics.StoreLoadDetai
 		}
 		if bs.rwTy == utils.Read {
 			peers := bs.cur.region.GetPeers()
-			moveLeaderFilters := []filter.Filter{&filter.StoreStateFilter{ActionScope: bs.sche.GetName(), MoveRegion: true, OperatorLevel: constant.High}}
+			moveLeaderFilters := []filter.Filter{
+				&filter.StoreStateFilter{ActionScope: bs.sche.GetName(), MoveRegion: true, OperatorLevel: constant.High},
+				filter.NewHotRegionEvictedTargetFilter(bs.sche.GetName()),
+			}
 			if leaderFilter := filter.NewPlacementLeaderSafeguard(bs.sche.GetName(), bs.GetSchedulerConfig(), bs.GetBasicCluster(), bs.GetRuleManager(), bs.cur.region, srcStore, true /*allowMoveLeader*/); leaderFilter != nil {
 				filters = append(filters, leaderFilter)
 			}

pkg/schedule/schedulers/hot_region_test.go

@@ -1467,6 +1467,97 @@ func TestHotReadRegionScheduleByteRateOnly(t *testing.T) {
 	clearPendingInfluence(hb)
 }
 
+func TestHotReadRegionScheduleSkipsLeaderEvictedMoveLeaderTarget(t *testing.T) {
+	testCases := []struct {
+		name string
+		opt  core.StoreCreateOption
+	}{
+		{name: "slow-store", opt: core.SlowStoreEvicted()},
+		{name: "stopping-store", opt: core.StoppingStoreEvicted()},
+		{name: "slow-trend", opt: core.SlowTrendEvicted()},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			re := require.New(t)
+			cancel, _, tc, oc := prepareSchedulersTest()
+			defer cancel()
+			tc.SetClusterVersion(versioninfo.MinSupportedVersion(versioninfo.Version4_0))
+			scheduler, err := CreateScheduler(readType, oc, storage.NewStorageWithMemoryBackend(), nil)
+			re.NoError(err)
+			hb := scheduler.(*hotScheduler)
+			hb.conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority}
+			hb.conf.setHistorySampleDuration(0)
+
+			tc.AddRegionStore(1, 3)
+			tc.AddRegionStore(2, 2)
+			tc.AddRegionStore(3, 2)
+			tc.AddRegionStore(4, 2)
+			tc.AddRegionStore(5, 0)
+
+			tc.UpdateStorageReadBytes(1, 7.5*units.MiB*utils.StoreHeartBeatReportInterval)
+			tc.UpdateStorageReadBytes(2, 4.9*units.MiB*utils.StoreHeartBeatReportInterval)
+			tc.UpdateStorageReadBytes(3, 3.7*units.MiB*utils.StoreHeartBeatReportInterval)
+			tc.UpdateStorageReadBytes(4, 6*units.MiB*utils.StoreHeartBeatReportInterval)
+			tc.UpdateStorageReadBytes(5, 0)
+			tc.PutStore(tc.GetStore(5).Clone(testCase.opt))
+
+			addRegionInfo(tc, utils.Read, []testRegionInfo{
+				{1, []uint64{1, 2, 3}, 512 * units.KiB, 0, 0},
+				{2, []uint64{2, 1, 3}, 511 * units.KiB, 0, 0},
+				{3, []uint64{1, 2, 3}, 510 * units.KiB, 0, 0},
+				{11, []uint64{1, 2, 3}, 7 * units.KiB, 0, 0},
+			})
+
+			testutil.Eventually(re, func() bool {
+				return tc.IsRegionHot(tc.GetRegion(1)) && !tc.IsRegionHot(tc.GetRegion(11))
+			})
+
+			ops, _ := hb.Schedule(tc, false)
+			re.Len(ops, 1)
+			operatorutil.CheckTransferLeader(re, ops[0], operator.OpHotRegion, 1, 3)
+		})
+	}
+}
+
+func TestHotReadPeerScheduleSkipsLeaderEvictedTarget(t *testing.T) {
+	testCases := []struct {
+		name string
+		opt  core.StoreCreateOption
+	}{
+		{name: "slow-store", opt: core.SlowStoreEvicted()},
+		{name: "stopping-store", opt: core.StoppingStoreEvicted()},
+		{name: "slow-trend", opt: core.SlowTrendEvicted()},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			re := require.New(t)
+			cancel, _, tc, oc := prepareSchedulersTest()
+			defer cancel()
+			tc.SetClusterVersion(versioninfo.MinSupportedVersion(versioninfo.Version4_0))
+			for id := uint64(1); id <= 6; id++ {
+				tc.PutStoreWithLabels(id)
+			}
+
+			sche, err := CreateScheduler(readType, oc, storage.NewStorageWithMemoryBackend(), ConfigJSONDecoder([]byte("null")))
+			re.NoError(err)
+			hb := sche.(*hotScheduler)
+			hb.conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority}
+			tc.PutStore(tc.GetStore(4).Clone(testCase.opt))
+
+			tc.UpdateStorageReadStats(1, 20*units.MiB, 20*units.MiB)
+			tc.UpdateStorageReadStats(2, 19*units.MiB, 19*units.MiB)
+			tc.UpdateStorageReadStats(3, 19*units.MiB, 19*units.MiB)
+			tc.UpdateStorageReadStats(4, 0*units.MiB, 0*units.MiB)
+			tc.AddRegionWithPeerReadInfo(1, 3, 1, uint64(0.9*units.KiB*float64(10)), uint64(0.9*units.KiB*float64(10)), 10, []uint64{1, 2}, 3)
+
+			ops, _ := hb.Schedule(tc, false)
+			re.Empty(ops)
+		})
+	}
+}
+
 func TestHotReadRegionScheduleWithQuery(t *testing.T) {
 	re := require.New(t)
 

pkg/schedule/schedulers/shuffle_hot_region.go

@@ -152,6 +152,7 @@ func (s *shuffleHotRegionScheduler) randomSchedule(cluster sche.SchedulerCluster
 
 		filters := []filter.Filter{
 			&filter.StoreStateFilter{ActionScope: s.GetName(), MoveRegion: true, OperatorLevel: constant.Low},
+			filter.NewHotRegionEvictedTargetFilter(s.GetName()),
 			filter.NewExcludedFilter(s.GetName(), srcRegion.GetStoreIDs(), srcRegion.GetStoreIDs()),
 			filter.NewPlacementSafeguard(s.GetName(), cluster.GetSchedulerConfig(), cluster.GetBasicCluster(), cluster.GetRuleManager(), srcRegion, srcStore, nil),
 		}