From 0b351c12d396b0db426b48625b22f611a53b506c Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Tue, 28 Jul 2026 12:16:49 +0800 Subject: [PATCH 1/9] schedule: support placement-constrained hot scheduling Skip global load expectation checks for peers governed by placement rules with label constraints. Keep target selection within the matched placement rule and preserve the existing behavior for unconstrained rules. Ref #5545 Signed-off-by: Ryan Leung --- pkg/schedule/schedulers/hot_region_solver.go | 69 +++++++++-- pkg/schedule/schedulers/hot_region_test.go | 119 +++++++++++++++++++ 2 files changed, 175 insertions(+), 13 deletions(-) diff --git a/pkg/schedule/schedulers/hot_region_solver.go b/pkg/schedule/schedulers/hot_region_solver.go index c870ecf6be..c7d2de9b13 100644 --- a/pkg/schedule/schedulers/hot_region_solver.go +++ b/pkg/schedule/schedulers/hot_region_solver.go @@ -33,6 +33,7 @@ import ( sche "github.com/tikv/pd/pkg/schedule/core" "github.com/tikv/pd/pkg/schedule/filter" "github.com/tikv/pd/pkg/schedule/operator" + "github.com/tikv/pd/pkg/schedule/placement" "github.com/tikv/pd/pkg/slice" "github.com/tikv/pd/pkg/statistics" "github.com/tikv/pd/pkg/statistics/buckets" @@ -51,6 +52,11 @@ type balanceSolver struct { resourceTy resourceType cur *solution + // curFit is the placement-rule fit for the current region. + curFit *placement.RegionFit + // skipLoadExpectation indicates whether the current region should skip the + // expectation and variance calculated from all stores. + skipLoadExpectation bool best *solution ops []*operator.Operator @@ -207,6 +213,10 @@ func (bs *balanceSolver) solve() []*operator.Operator { if bs.cur.region = bs.getRegion(mainPeerStat, srcStoreID); bs.cur.region == nil { continue } + bs.prepareForRegion() + if bs.GetSchedulerConfig().IsPlacementRulesEnabled() && !bs.skipLoadExpectation && !bs.isSourceStoreWithinExpectation(srcStore) { + continue + } if bs.opTy == movePeer && !snapshotFilter.Select(bs.cur.region).IsOK() { hotSchedulerSnapshotSenderLimitCounter.Inc() continue @@ -378,10 +388,8 @@ func (bs *balanceSolver) calcMaxZombieDur() time.Duration { // its expectation * ratio, the store would be selected as hot source store func (bs *balanceSolver) filterSrcStores() map[uint64]*statistics.StoreLoadDetail { ret := make(map[uint64]*statistics.StoreLoadDetail) - confSrcToleranceRatio := bs.sche.conf.getSrcToleranceRatio() confEnableForTiFlash := bs.sche.conf.getEnableForTiFlash() for id, detail := range bs.stLoadDetail { - srcToleranceRatio := confSrcToleranceRatio if !detail.IsTiKV() { if !confEnableForTiFlash || detail.IsTiFlashCompute() { continue @@ -389,18 +397,14 @@ func (bs *balanceSolver) filterSrcStores() map[uint64]*statistics.StoreLoadDetai if bs.rwTy != utils.Write || bs.opTy != movePeer { continue } - srcToleranceRatio += tiflashToleranceRatioCorrection } if len(detail.HotPeers) == 0 { continue } - if !bs.checkSrcByPriorityAndTolerance(detail.LoadPred.Min(), &detail.LoadPred.Expect, srcToleranceRatio) { - hotSchedulerResultCounter.WithLabelValues("src-store-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() - continue - } - if !bs.checkSrcHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, &detail.LoadPred.Expect, srcToleranceRatio) { - hotSchedulerResultCounter.WithLabelValues("src-store-history-loads-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() + // With placement rules enabled, source expectation depends on the rule + // matched by each hot peer and is checked after selecting the region. + if !bs.GetSchedulerConfig().IsPlacementRulesEnabled() && !bs.isSourceStoreWithinExpectation(detail) { continue } @@ -410,6 +414,23 @@ func (bs *balanceSolver) filterSrcStores() map[uint64]*statistics.StoreLoadDetai return ret } +func (bs *balanceSolver) isSourceStoreWithinExpectation(detail *statistics.StoreLoadDetail) bool { + id := detail.GetID() + srcToleranceRatio := bs.sche.conf.getSrcToleranceRatio() + if !detail.IsTiKV() { + srcToleranceRatio += tiflashToleranceRatioCorrection + } + if !bs.checkSrcByPriorityAndTolerance(detail.LoadPred.Min(), &detail.LoadPred.Expect, srcToleranceRatio) { + hotSchedulerResultCounter.WithLabelValues("src-store-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() + return false + } + if !bs.checkSrcHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, &detail.LoadPred.Expect, srcToleranceRatio) { + hotSchedulerResultCounter.WithLabelValues("src-store-history-loads-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() + return false + } + return true +} + func (bs *balanceSolver) checkSrcByPriorityAndTolerance(minLoad, expectLoad *statistics.StoreLoad, toleranceRatio float64) bool { return bs.checkByPriorityAndTolerance(minLoad.Loads, func(i int) bool { return minLoad.Loads[i] > toleranceRatio*expectLoad.Loads[i] @@ -566,7 +587,7 @@ func (bs *balanceSolver) filterDstStores() map[uint64]*statistics.StoreLoadDetai 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), + filter.NewPlacementSafeguard(bs.sche.GetName(), bs.GetSchedulerConfig(), bs.GetBasicCluster(), bs.GetRuleManager(), bs.cur.region, srcStore, bs.curFit), } for _, detail := range bs.stLoadDetail { candidates = append(candidates, detail) @@ -640,11 +661,11 @@ func (bs *balanceSolver) pickDstStores(filters []filter.Filter, candidates []*st } if filter.Target(bs.GetSchedulerConfig(), store, filters) { id := store.GetID() - if !bs.checkDstByPriorityAndTolerance(detail.LoadPred.Max(), &detail.LoadPred.Expect, dstToleranceRatio) { + if !bs.skipLoadExpectation && !bs.checkDstByPriorityAndTolerance(detail.LoadPred.Max(), &detail.LoadPred.Expect, dstToleranceRatio) { hotSchedulerResultCounter.WithLabelValues("dst-store-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() continue } - if !bs.checkDstHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, &detail.LoadPred.Expect, dstToleranceRatio) { + if !bs.skipLoadExpectation && !bs.checkDstHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, &detail.LoadPred.Expect, dstToleranceRatio) { hotSchedulerResultCounter.WithLabelValues("dst-store-history-loads-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() continue } @@ -718,7 +739,29 @@ func (bs *balanceSolver) checkHistoryLoadsByPriorityAndToleranceFirstOnly(_ stat } func (bs *balanceSolver) enableExpectation() bool { - return bs.sche.conf.getDstToleranceRatio() > 0 && bs.sche.conf.getSrcToleranceRatio() > 0 + return !bs.skipLoadExpectation && bs.sche.conf.getDstToleranceRatio() > 0 && bs.sche.conf.getSrcToleranceRatio() > 0 +} + +// prepareForRegion determines whether the current region should use the global +// load expectation. A rule with label constraints limits its peer to a subset +// of stores, so valid source-target pairs are compared directly, like +// balance-region scheduling. +func (bs *balanceSolver) prepareForRegion() { + bs.curFit = nil + bs.skipLoadExpectation = false + if !bs.GetSchedulerConfig().IsPlacementRulesEnabled() { + return + } + + bs.curFit = bs.GetRuleManager().FitRegion(bs.GetBasicCluster(), bs.cur.region) + sourcePeer := bs.cur.region.GetStorePeer(bs.cur.srcStore.GetID()) + if sourcePeer == nil { + return + } + ruleFit := bs.curFit.GetRuleFit(sourcePeer.GetId()) + if ruleFit != nil && len(ruleFit.Rule.LabelConstraints) > 0 { + bs.skipLoadExpectation = true + } } func (bs *balanceSolver) isUniformFirstPriority(store *statistics.StoreLoadDetail) bool { diff --git a/pkg/schedule/schedulers/hot_region_test.go b/pkg/schedule/schedulers/hot_region_test.go index 8ddc98c0ca..a48ba0f53c 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -410,6 +410,125 @@ func checkHotWriteRegionPlacement(re *require.Assertions, enablePlacementRules b operatorutil.CheckTransferPeerWithLeaderTransfer(re, ops[0], operator.OpHotRegion, 1, 2) } +func TestHotWriteRegionScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { + testCases := []struct { + name string + storeLoads map[uint64]float64 + peerLoad float64 + }{ + { + name: "source below global expectation", + storeLoads: map[uint64]float64{ + 1: 20, + 2: 0, + 3: 0, + 4: 0, + 5: 100, + 6: 100, + }, + peerLoad: 5, + }, + { + name: "target above global expectation", + storeLoads: map[uint64]float64{ + 1: 100, + 2: 80, + 3: 80, + 4: 60, + 5: 0, + 6: 0, + }, + peerLoad: 20, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + re := require.New(t) + cancel, _, tc, oc := prepareSchedulersTest() + defer cancel() + tc.SetEnablePlacementRules(true) + tc.SetHotRegionCacheHitsThreshold(0) + + hb, err := CreateScheduler(writeType, oc, storage.NewStorageWithMemoryBackend(), nil) + re.NoError(err) + hb.(*hotScheduler).types = []resourceType{writePeer} + hb.(*hotScheduler).conf.setHistorySampleDuration(0) + + for id := uint64(1); id <= 6; id++ { + pool := "other" + if id <= 4 { + pool = "target" + } + tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + tc.UpdateStorageWrittenBytes(id, uint64(testCase.storeLoads[id]*units.MiB*utils.StoreHeartBeatReportInterval)) + } + err = tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, + ID: placement.DefaultRuleID, + Role: placement.Voter, + Count: 3, + LabelConstraints: []placement.LabelConstraint{ + {Key: "pool", Op: placement.In, Values: []string{"target"}}, + }, + }) + re.NoError(err) + + addRegionInfo(tc, utils.Write, []testRegionInfo{ + {1, []uint64{1, 2, 3}, testCase.peerLoad * units.MiB, 0, 0}, + }) + + ops, _ := hb.Schedule(tc, false) + re.Len(ops, 1) + operatorutil.CheckTransferPeerWithLeaderTransfer(re, ops[0], operator.OpHotRegion, 1, 4) + }) + } +} + +func TestHotReadLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { + re := require.New(t) + cancel, _, tc, oc := prepareSchedulersTest() + defer cancel() + tc.SetEnablePlacementRules(true) + tc.SetHotRegionCacheHitsThreshold(0) + + hb, err := CreateScheduler(readType, oc, storage.NewStorageWithMemoryBackend(), nil) + re.NoError(err) + hb.(*hotScheduler).types = []resourceType{readLeader} + hb.(*hotScheduler).conf.setHistorySampleDuration(0) + hb.(*hotScheduler).conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority} + + for id := uint64(1); id <= 6; id++ { + pool := "other" + load := 100 * units.MiB * utils.StoreHeartBeatReportInterval + if id <= 3 { + pool = "target" + load = 0 + } + tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + tc.UpdateStorageReadStats(id, uint64(load), 0) + } + tc.UpdateStorageReadStats(1, 20*units.MiB*utils.StoreHeartBeatReportInterval, 0) + err = tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, + ID: placement.DefaultRuleID, + Role: placement.Voter, + Count: 3, + LabelConstraints: []placement.LabelConstraint{ + {Key: "pool", Op: placement.In, Values: []string{"target"}}, + }, + }) + re.NoError(err) + + addRegionLeaderReadInfo(tc, []testRegionInfo{ + {1, []uint64{1, 2, 3}, 5 * units.MiB, 0, 0}, + }) + + ops, _ := hb.Schedule(tc, false) + re.Len(ops, 1) + operatorutil.CheckTransferLeaderFrom(re, ops[0], operator.OpHotRegion, 1) +} + func checkHotWriteRegionScheduleByteRateOnly(re *require.Assertions, enablePlacementRules bool) { cancel, opt, tc, oc := prepareSchedulersTest() defer cancel() From ea56f194eacaa5dc27e623587f5c578009b7810b Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Tue, 28 Jul 2026 12:33:31 +0800 Subject: [PATCH 2/9] tests: cover all placement-constrained hot paths Add explicit coverage for write-leader and read-peer scheduling when Placement Rules constrain eligible stores and global load expectations would otherwise block valid operations. Ref #5545 Signed-off-by: Ryan Leung --- pkg/schedule/schedulers/hot_region_test.go | 92 ++++++++++++++++++++++ 1 file changed, 92 insertions(+) diff --git a/pkg/schedule/schedulers/hot_region_test.go b/pkg/schedule/schedulers/hot_region_test.go index a48ba0f53c..1f11d65fe6 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -529,6 +529,98 @@ func TestHotReadLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t operatorutil.CheckTransferLeaderFrom(re, ops[0], operator.OpHotRegion, 1) } +func TestHotWriteLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { + re := require.New(t) + cancel, _, tc, oc := prepareSchedulersTest() + defer cancel() + tc.SetEnablePlacementRules(true) + tc.SetHotRegionCacheHitsThreshold(0) + + hb, err := CreateScheduler(writeType, oc, storage.NewStorageWithMemoryBackend(), nil) + re.NoError(err) + hb.(*hotScheduler).types = []resourceType{writeLeader} + hb.(*hotScheduler).conf.setHistorySampleDuration(0) + hb.(*hotScheduler).conf.WriteLeaderPriorities = []string{utils.BytePriority, utils.KeyPriority} + + for id := uint64(1); id <= 6; id++ { + pool := "other" + if id <= 4 { + pool = "target" + } + tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + tc.UpdateStorageWrittenBytes(id, 0) + } + err = tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, + ID: placement.DefaultRuleID, + Role: placement.Voter, + Count: 3, + LabelConstraints: []placement.LabelConstraint{ + {Key: "pool", Op: placement.In, Values: []string{"target"}}, + }, + }) + re.NoError(err) + + addRegionInfo(tc, utils.Write, []testRegionInfo{ + {1, []uint64{1, 2, 3}, 2.5 * units.MiB, 0, 0}, + {2, []uint64{1, 2, 3}, 17.5 * units.MiB, 0, 0}, + {3, []uint64{2, 1, 3}, 10 * units.MiB, 0, 0}, + {4, []uint64{3, 1, 2}, 10 * units.MiB, 0, 0}, + }) + + ops, _ := hb.Schedule(tc, false) + re.Len(ops, 1) + re.Equal(uint64(1), ops[0].RegionID()) + operatorutil.CheckTransferLeaderFrom(re, ops[0], operator.OpHotRegion, 1) +} + +func TestHotReadPeerScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { + re := require.New(t) + cancel, _, tc, oc := prepareSchedulersTest() + defer cancel() + tc.SetEnablePlacementRules(true) + tc.SetHotRegionCacheHitsThreshold(0) + + hb, err := CreateScheduler(readType, oc, storage.NewStorageWithMemoryBackend(), nil) + re.NoError(err) + hb.(*hotScheduler).types = []resourceType{readPeer} + hb.(*hotScheduler).conf.setHistorySampleDuration(0) + hb.(*hotScheduler).conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority} + + for id := uint64(1); id <= 6; id++ { + pool := "other" + load := 100 * units.MiB * utils.StoreHeartBeatReportInterval + if id <= 4 { + pool = "target" + load = 0 + } + tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + tc.UpdateStorageReadStats(id, uint64(load), 0) + } + tc.UpdateStorageReadStats(1, 20*units.MiB*utils.StoreHeartBeatReportInterval, 0) + err = tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, + ID: placement.DefaultRuleID, + Role: placement.Voter, + Count: 3, + LabelConstraints: []placement.LabelConstraint{ + {Key: "pool", Op: placement.In, Values: []string{"target"}}, + }, + }) + re.NoError(err) + + interval := uint64(utils.StoreHeartBeatReportInterval) + tc.AddRegionWithPeerReadInfo( + 1, 3, 1, + 5*units.MiB*interval, 0, interval, + []uint64{1, 2}, + ) + + ops, _ := hb.Schedule(tc, false) + re.Len(ops, 1) + operatorutil.CheckTransferPeer(re, ops[0], operator.OpHotRegion, 1, 4) +} + func checkHotWriteRegionScheduleByteRateOnly(re *require.Assertions, enablePlacementRules bool) { cancel, opt, tc, oc := prepareSchedulersTest() defer cancel() From 2e4d2ac6e50476e2726f9fc227744fa22574246f Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Tue, 28 Jul 2026 14:39:56 +0800 Subject: [PATCH 3/9] schedule: harden placement-scoped hot scheduling Signed-off-by: Ryan Leung --- pkg/schedule/filter/filters.go | 19 ++- pkg/schedule/filter/filters_test.go | 6 +- pkg/schedule/schedulers/hot_region_solver.go | 105 ++++++++++---- pkg/schedule/schedulers/hot_region_test.go | 144 ++++++++----------- 4 files changed, 156 insertions(+), 118 deletions(-) diff --git a/pkg/schedule/filter/filters.go b/pkg/schedule/filter/filters.go index 550cadfee8..de07798931 100644 --- a/pkg/schedule/filter/filters.go +++ b/pkg/schedule/filter/filters.go @@ -703,13 +703,17 @@ type ruleLeaderFitFilter struct { // newRuleLeaderFitFilter creates a filter that ensures after transfer leader with new store, // the isolation level will not decrease. -func newRuleLeaderFitFilter(scope string, cluster *core.BasicCluster, ruleManager *placement.RuleManager, region *core.RegionInfo, srcLeaderStoreID uint64, allowMoveLeader bool) Filter { +func newRuleLeaderFitFilter(scope string, cluster *core.BasicCluster, ruleManager *placement.RuleManager, + region *core.RegionInfo, oldFit *placement.RegionFit, srcLeaderStoreID uint64, allowMoveLeader bool) Filter { + if oldFit == nil { + oldFit = ruleManager.FitRegion(cluster, region) + } return &ruleLeaderFitFilter{ scope: scope, cluster: cluster, ruleManager: ruleManager, region: region, - oldFit: ruleManager.FitRegion(cluster, region), + oldFit: oldFit, srcLeaderStoreID: srcLeaderStoreID, allowMoveLeader: allowMoveLeader, } @@ -816,9 +820,16 @@ func NewPlacementSafeguard(scope string, conf config.SharedConfigProvider, clust // NewPlacementLeaderSafeguard creates a filter that ensures after transfer a leader with // existed peer, the placement restriction will not become worse. // Note that it only worked when PlacementRules enabled otherwise it will always permit the sourceStore. -func NewPlacementLeaderSafeguard(scope string, conf config.SharedConfigProvider, cluster *core.BasicCluster, ruleManager *placement.RuleManager, region *core.RegionInfo, sourceStore *core.StoreInfo, allowMoveLeader bool) Filter { +func NewPlacementLeaderSafeguard(scope string, conf config.SharedConfigProvider, cluster *core.BasicCluster, ruleManager *placement.RuleManager, + region *core.RegionInfo, sourceStore *core.StoreInfo, allowMoveLeader bool) Filter { + return NewPlacementLeaderSafeguardWithFit(scope, conf, cluster, ruleManager, region, sourceStore, nil, allowMoveLeader) +} + +// NewPlacementLeaderSafeguardWithFit creates a placement leader safeguard with a reusable region fit. +func NewPlacementLeaderSafeguardWithFit(scope string, conf config.SharedConfigProvider, cluster *core.BasicCluster, ruleManager *placement.RuleManager, + region *core.RegionInfo, sourceStore *core.StoreInfo, oldFit *placement.RegionFit, allowMoveLeader bool) Filter { if conf.IsPlacementRulesEnabled() { - return newRuleLeaderFitFilter(scope, cluster, ruleManager, region, sourceStore.GetID(), allowMoveLeader) + return newRuleLeaderFitFilter(scope, cluster, ruleManager, region, oldFit, sourceStore.GetID(), allowMoveLeader) } return nil } diff --git a/pkg/schedule/filter/filters_test.go b/pkg/schedule/filter/filters_test.go index ba0f319d49..18acd4e27e 100644 --- a/pkg/schedule/filter/filters_test.go +++ b/pkg/schedule/filter/filters_test.go @@ -144,12 +144,12 @@ func TestRuleFitFilter(t *testing.T) { filter := newRuleFitFilter("", testCluster.GetBasicCluster(), testCluster.GetRuleManager(), region, nil, 1) re.Equal(testCase.sourceRes, filter.Source(testCluster.GetSharedConfig(), testCluster.GetStore(testCase.storeID)).StatusCode) re.Equal(testCase.targetRes, filter.Target(testCluster.GetSharedConfig(), testCluster.GetStore(testCase.storeID)).StatusCode) - leaderFilter := newRuleLeaderFitFilter("", testCluster.GetBasicCluster(), testCluster.GetRuleManager(), region, 1, true) + leaderFilter := newRuleLeaderFitFilter("", testCluster.GetBasicCluster(), testCluster.GetRuleManager(), region, nil, 1, true) re.Equal(testCase.targetRes, leaderFilter.Target(testCluster.GetSharedConfig(), testCluster.GetStore(testCase.storeID)).StatusCode) } // store-6 is not exist in the peers, so it will not allow transferring leader to store 6. - leaderFilter := newRuleLeaderFitFilter("", testCluster.GetBasicCluster(), testCluster.GetRuleManager(), region, 1, false) + leaderFilter := newRuleLeaderFitFilter("", testCluster.GetBasicCluster(), testCluster.GetRuleManager(), region, nil, 1, false) re.False(leaderFilter.Target(testCluster.GetSharedConfig(), testCluster.GetStore(6)).IsOK()) } @@ -223,7 +223,7 @@ func TestRuleFitFilterWithPlacementRule(t *testing.T) { {StoreId: 5, Id: 4}, {StoreId: 6, Id: 5}, }}, &metapb.Peer{StoreId: 1, Id: 1}) - leaderFilter := newRuleLeaderFitFilter("", testCluster.GetBasicCluster(), testCluster.GetRuleManager(), region, 1, true) + leaderFilter := newRuleLeaderFitFilter("", testCluster.GetBasicCluster(), testCluster.GetRuleManager(), region, nil, 1, true) re.Equal(plan.StatusText(plan.StatusStoreNotMatchRule), leaderFilter.Target(testCluster.GetSharedConfig(), testCluster.GetStore(6)).String()) } diff --git a/pkg/schedule/schedulers/hot_region_solver.go b/pkg/schedule/schedulers/hot_region_solver.go index c7d2de9b13..e8dc869fff 100644 --- a/pkg/schedule/schedulers/hot_region_solver.go +++ b/pkg/schedule/schedulers/hot_region_solver.go @@ -206,17 +206,44 @@ func (bs *balanceSolver) solve() []*operator.Operator { snapshotFilter := filter.NewSnapshotSendFilter(bs.GetStores(), constant.Medium) affinityFilter := filter.NewAffinityFilter(bs.SchedulerCluster) splitThresholds := bs.sche.conf.getSplitThresholds() + placementEnabled := bs.GetSchedulerConfig().IsPlacementRulesEnabled() + var placementChecked, placementCanRestrict [2]bool for _, srcStore := range bs.filterSrcStores() { bs.cur.srcStore = srcStore srcStoreID := srcStore.GetID() + sourceFailure := bs.sourceStoreFailure(srcStore) + sourceResultRecorded := sourceFailure == "" + if sourceResultRecorded { + bs.recordSourceStoreResult("src-store-succ", srcStoreID) + } else if !placementEnabled { + bs.recordSourceStoreResult(sourceFailure, srcStoreID) + continue + } else { + engine := 0 + if !srcStore.IsTiKV() { + engine = 1 + } + if !placementChecked[engine] { + placementCanRestrict[engine] = bs.mayUsePlacementScope(srcStore) + placementChecked[engine] = true + } + if !placementCanRestrict[engine] { + bs.recordSourceStoreResult(sourceFailure, srcStoreID) + continue + } + } for _, mainPeerStat := range bs.filteredHotPeers[srcStoreID] { if bs.cur.region = bs.getRegion(mainPeerStat, srcStoreID); bs.cur.region == nil { continue } bs.prepareForRegion() - if bs.GetSchedulerConfig().IsPlacementRulesEnabled() && !bs.skipLoadExpectation && !bs.isSourceStoreWithinExpectation(srcStore) { + if !bs.skipLoadExpectation && sourceFailure != "" { continue } + if !sourceResultRecorded { + bs.recordSourceStoreResult("src-store-succ", srcStoreID) + sourceResultRecorded = true + } if bs.opTy == movePeer && !snapshotFilter.Select(bs.cur.region).IsOK() { hotSchedulerSnapshotSenderLimitCounter.Inc() continue @@ -260,6 +287,9 @@ func (bs *balanceSolver) solve() []*operator.Operator { } } } + if !sourceResultRecorded { + bs.recordSourceStoreResult(sourceFailure, srcStoreID) + } } bs.setSearchRevertRegions() @@ -384,8 +414,7 @@ func (bs *balanceSolver) calcMaxZombieDur() time.Duration { } } -// filterSrcStores compare the min rate and the ratio * expectation rate, if two dim rate is greater than -// its expectation * ratio, the store would be selected as hot source store +// filterSrcStores selects stores with hot peers that support the resource type. func (bs *balanceSolver) filterSrcStores() map[uint64]*statistics.StoreLoadDetail { ret := make(map[uint64]*statistics.StoreLoadDetail) confEnableForTiFlash := bs.sche.conf.getEnableForTiFlash() @@ -401,34 +430,27 @@ func (bs *balanceSolver) filterSrcStores() map[uint64]*statistics.StoreLoadDetai if len(detail.HotPeers) == 0 { continue } - - // With placement rules enabled, source expectation depends on the rule - // matched by each hot peer and is checked after selecting the region. - if !bs.GetSchedulerConfig().IsPlacementRulesEnabled() && !bs.isSourceStoreWithinExpectation(detail) { - continue - } - ret[id] = detail - hotSchedulerResultCounter.WithLabelValues("src-store-succ-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() } return ret } -func (bs *balanceSolver) isSourceStoreWithinExpectation(detail *statistics.StoreLoadDetail) bool { - id := detail.GetID() +func (bs *balanceSolver) sourceStoreFailure(detail *statistics.StoreLoadDetail) string { srcToleranceRatio := bs.sche.conf.getSrcToleranceRatio() if !detail.IsTiKV() { srcToleranceRatio += tiflashToleranceRatioCorrection } if !bs.checkSrcByPriorityAndTolerance(detail.LoadPred.Min(), &detail.LoadPred.Expect, srcToleranceRatio) { - hotSchedulerResultCounter.WithLabelValues("src-store-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() - return false + return "src-store-failed" } if !bs.checkSrcHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, &detail.LoadPred.Expect, srcToleranceRatio) { - hotSchedulerResultCounter.WithLabelValues("src-store-history-loads-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() - return false + return "src-store-history-loads-failed" } - return true + return "" +} + +func (bs *balanceSolver) recordSourceStoreResult(result string, storeID uint64) { + hotSchedulerResultCounter.WithLabelValues(result+"-"+bs.resourceTy.String(), strconv.FormatUint(storeID, 10)).Inc() } func (bs *balanceSolver) checkSrcByPriorityAndTolerance(minLoad, expectLoad *statistics.StoreLoad, toleranceRatio float64) bool { @@ -532,7 +554,14 @@ func (bs *balanceSolver) isRegionAvailable(region *core.RegionInfo) bool { return false } - if !filter.IsRegionReplicated(bs.SchedulerCluster, region) { + replicated := false + if bs.GetSchedulerConfig().IsPlacementRulesEnabled() { + bs.curFit = bs.GetRuleManager().FitRegion(bs.GetBasicCluster(), region) + replicated = bs.curFit.IsSatisfied() + } else { + replicated = filter.IsRegionReplicated(bs.SchedulerCluster, region) + } + if !replicated { log.Debug("region has abnormal replica count", zap.String("scheduler", bs.sche.GetName()), zap.Uint64("region-id", region.GetID())) hotSchedulerAbnormalReplicaCounter.Inc() return false @@ -542,6 +571,7 @@ func (bs *balanceSolver) isRegionAvailable(region *core.RegionInfo) bool { } func (bs *balanceSolver) getRegion(peerStat *statistics.HotPeerStat, storeID uint64) *core.RegionInfo { + bs.curFit = nil region := bs.GetRegion(peerStat.ID()) if !bs.isRegionAvailable(region) { return nil @@ -607,7 +637,7 @@ func (bs *balanceSolver) filterDstStores() map[uint64]*statistics.StoreLoadDetai &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 { + if leaderFilter := filter.NewPlacementLeaderSafeguardWithFit(bs.sche.GetName(), bs.GetSchedulerConfig(), bs.GetBasicCluster(), bs.GetRuleManager(), bs.cur.region, srcStore, bs.curFit, true /*allowMoveLeader*/); leaderFilter != nil { filters = append(filters, leaderFilter) } for storeID, detail := range bs.stLoadDetail { @@ -627,7 +657,7 @@ func (bs *balanceSolver) filterDstStores() map[uint64]*statistics.StoreLoadDetai } } } else { - if leaderFilter := filter.NewPlacementLeaderSafeguard(bs.sche.GetName(), bs.GetSchedulerConfig(), bs.GetBasicCluster(), bs.GetRuleManager(), bs.cur.region, srcStore, false /*allowMoveLeader*/); leaderFilter != nil { + if leaderFilter := filter.NewPlacementLeaderSafeguardWithFit(bs.sche.GetName(), bs.GetSchedulerConfig(), bs.GetBasicCluster(), bs.GetRuleManager(), bs.cur.region, srcStore, bs.curFit, false /*allowMoveLeader*/); leaderFilter != nil { filters = append(filters, leaderFilter) } for _, peer := range bs.cur.region.GetFollowers() { @@ -743,25 +773,42 @@ func (bs *balanceSolver) enableExpectation() bool { } // prepareForRegion determines whether the current region should use the global -// load expectation. A rule with label constraints limits its peer to a subset -// of stores, so valid source-target pairs are compared directly, like -// balance-region scheduling. +// load expectation. MatchLabelConstraints also covers implicit exclusive labels. func (bs *balanceSolver) prepareForRegion() { - bs.curFit = nil bs.skipLoadExpectation = false - if !bs.GetSchedulerConfig().IsPlacementRulesEnabled() { + if bs.curFit == nil { return } - bs.curFit = bs.GetRuleManager().FitRegion(bs.GetBasicCluster(), bs.cur.region) sourcePeer := bs.cur.region.GetStorePeer(bs.cur.srcStore.GetID()) if sourcePeer == nil { return } ruleFit := bs.curFit.GetRuleFit(sourcePeer.GetId()) - if ruleFit != nil && len(ruleFit.Rule.LabelConstraints) > 0 { - bs.skipLoadExpectation = true + if ruleFit != nil { + bs.skipLoadExpectation = bs.ruleRestrictsStoreLoad(ruleFit.Rule, bs.cur.srcStore) + } +} + +func (bs *balanceSolver) mayUsePlacementScope(source *statistics.StoreLoadDetail) bool { + for _, rule := range bs.GetRuleManager().GetAllRules() { + if bs.ruleRestrictsStoreLoad(rule, source) { + return true + } + } + return false +} + +func (bs *balanceSolver) ruleRestrictsStoreLoad(rule *placement.Rule, source *statistics.StoreLoadDetail) bool { + if !placement.MatchLabelConstraints(source.StoreInfo, rule.LabelConstraints) { + return false + } + for _, detail := range bs.stLoadDetail { + if detail.IsTiKV() == source.IsTiKV() && !placement.MatchLabelConstraints(detail.StoreInfo, rule.LabelConstraints) { + return true + } } + return false } func (bs *balanceSolver) isUniformFirstPriority(store *statistics.StoreLoadDetail) bool { diff --git a/pkg/schedule/schedulers/hot_region_test.go b/pkg/schedule/schedulers/hot_region_test.go index 1f11d65fe6..43a86bb948 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -410,11 +410,39 @@ func checkHotWriteRegionPlacement(re *require.Assertions, enablePlacementRules b operatorutil.CheckTransferPeerWithLeaderTransfer(re, ops[0], operator.OpHotRegion, 1, 2) } +func preparePlacementHotScheduler(t *testing.T, typ types.CheckerSchedulerType, resource resourceType) (*mockcluster.Cluster, *hotScheduler) { + t.Helper() + cancel, _, tc, oc := prepareSchedulersTest() + t.Cleanup(cancel) + tc.SetEnablePlacementRules(true) + tc.SetHotRegionCacheHitsThreshold(0) + scheduler, err := CreateScheduler(typ, oc, storage.NewStorageWithMemoryBackend(), nil) + require.NoError(t, err) + hot := scheduler.(*hotScheduler) + hot.types = []resourceType{resource} + hot.conf.setHistorySampleDuration(0) + return tc, hot +} + +func setPoolPlacementRule(t *testing.T, tc *mockcluster.Cluster) { + t.Helper() + require.NoError(t, tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, + ID: placement.DefaultRuleID, + Role: placement.Voter, + Count: 3, + LabelConstraints: []placement.LabelConstraint{ + {Key: "pool", Op: placement.In, Values: []string{"target"}}, + }, + })) +} + func TestHotWriteRegionScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { testCases := []struct { name string storeLoads map[uint64]float64 peerLoad float64 + exclusive bool }{ { name: "source below global expectation", @@ -440,39 +468,42 @@ func TestHotWriteRegionScheduleWithPlacementConstraintsIgnoresGlobalExpectation( }, peerLoad: 20, }, + { + name: "implicit exclusive labels", + storeLoads: map[uint64]float64{ + 1: 20, + 2: 0, + 3: 0, + 4: 0, + 5: 100, + 6: 100, + }, + peerLoad: 5, + exclusive: true, + }, } for _, testCase := range testCases { t.Run(testCase.name, func(t *testing.T) { re := require.New(t) - cancel, _, tc, oc := prepareSchedulersTest() - defer cancel() - tc.SetEnablePlacementRules(true) - tc.SetHotRegionCacheHitsThreshold(0) - - hb, err := CreateScheduler(writeType, oc, storage.NewStorageWithMemoryBackend(), nil) - re.NoError(err) - hb.(*hotScheduler).types = []resourceType{writePeer} - hb.(*hotScheduler).conf.setHistorySampleDuration(0) + tc, hb := preparePlacementHotScheduler(t, writeType, writePeer) for id := uint64(1); id <= 6; id++ { - pool := "other" + labels := map[string]string{"pool": "other"} if id <= 4 { - pool = "target" + labels["pool"] = "target" + } else if testCase.exclusive { + labels = map[string]string{"$group": "other"} + } + if testCase.exclusive && id <= 4 { + labels = nil } - tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + tc.AddLabelsStore(id, 1, labels) tc.UpdateStorageWrittenBytes(id, uint64(testCase.storeLoads[id]*units.MiB*utils.StoreHeartBeatReportInterval)) } - err = tc.SetRule(&placement.Rule{ - GroupID: placement.DefaultGroupID, - ID: placement.DefaultRuleID, - Role: placement.Voter, - Count: 3, - LabelConstraints: []placement.LabelConstraint{ - {Key: "pool", Op: placement.In, Values: []string{"target"}}, - }, - }) - re.NoError(err) + if !testCase.exclusive { + setPoolPlacementRule(t, tc) + } addRegionInfo(tc, utils.Write, []testRegionInfo{ {1, []uint64{1, 2, 3}, testCase.peerLoad * units.MiB, 0, 0}, @@ -487,16 +518,8 @@ func TestHotWriteRegionScheduleWithPlacementConstraintsIgnoresGlobalExpectation( func TestHotReadLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { re := require.New(t) - cancel, _, tc, oc := prepareSchedulersTest() - defer cancel() - tc.SetEnablePlacementRules(true) - tc.SetHotRegionCacheHitsThreshold(0) - - hb, err := CreateScheduler(readType, oc, storage.NewStorageWithMemoryBackend(), nil) - re.NoError(err) - hb.(*hotScheduler).types = []resourceType{readLeader} - hb.(*hotScheduler).conf.setHistorySampleDuration(0) - hb.(*hotScheduler).conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority} + tc, hb := preparePlacementHotScheduler(t, readType, readLeader) + hb.conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority} for id := uint64(1); id <= 6; id++ { pool := "other" @@ -509,16 +532,7 @@ func TestHotReadLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t tc.UpdateStorageReadStats(id, uint64(load), 0) } tc.UpdateStorageReadStats(1, 20*units.MiB*utils.StoreHeartBeatReportInterval, 0) - err = tc.SetRule(&placement.Rule{ - GroupID: placement.DefaultGroupID, - ID: placement.DefaultRuleID, - Role: placement.Voter, - Count: 3, - LabelConstraints: []placement.LabelConstraint{ - {Key: "pool", Op: placement.In, Values: []string{"target"}}, - }, - }) - re.NoError(err) + setPoolPlacementRule(t, tc) addRegionLeaderReadInfo(tc, []testRegionInfo{ {1, []uint64{1, 2, 3}, 5 * units.MiB, 0, 0}, @@ -531,16 +545,8 @@ func TestHotReadLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t func TestHotWriteLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { re := require.New(t) - cancel, _, tc, oc := prepareSchedulersTest() - defer cancel() - tc.SetEnablePlacementRules(true) - tc.SetHotRegionCacheHitsThreshold(0) - - hb, err := CreateScheduler(writeType, oc, storage.NewStorageWithMemoryBackend(), nil) - re.NoError(err) - hb.(*hotScheduler).types = []resourceType{writeLeader} - hb.(*hotScheduler).conf.setHistorySampleDuration(0) - hb.(*hotScheduler).conf.WriteLeaderPriorities = []string{utils.BytePriority, utils.KeyPriority} + tc, hb := preparePlacementHotScheduler(t, writeType, writeLeader) + hb.conf.WriteLeaderPriorities = []string{utils.BytePriority, utils.KeyPriority} for id := uint64(1); id <= 6; id++ { pool := "other" @@ -550,16 +556,7 @@ func TestHotWriteLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation( tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) tc.UpdateStorageWrittenBytes(id, 0) } - err = tc.SetRule(&placement.Rule{ - GroupID: placement.DefaultGroupID, - ID: placement.DefaultRuleID, - Role: placement.Voter, - Count: 3, - LabelConstraints: []placement.LabelConstraint{ - {Key: "pool", Op: placement.In, Values: []string{"target"}}, - }, - }) - re.NoError(err) + setPoolPlacementRule(t, tc) addRegionInfo(tc, utils.Write, []testRegionInfo{ {1, []uint64{1, 2, 3}, 2.5 * units.MiB, 0, 0}, @@ -576,16 +573,8 @@ func TestHotWriteLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation( func TestHotReadPeerScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { re := require.New(t) - cancel, _, tc, oc := prepareSchedulersTest() - defer cancel() - tc.SetEnablePlacementRules(true) - tc.SetHotRegionCacheHitsThreshold(0) - - hb, err := CreateScheduler(readType, oc, storage.NewStorageWithMemoryBackend(), nil) - re.NoError(err) - hb.(*hotScheduler).types = []resourceType{readPeer} - hb.(*hotScheduler).conf.setHistorySampleDuration(0) - hb.(*hotScheduler).conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority} + tc, hb := preparePlacementHotScheduler(t, readType, readPeer) + hb.conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority} for id := uint64(1); id <= 6; id++ { pool := "other" @@ -598,16 +587,7 @@ func TestHotReadPeerScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t * tc.UpdateStorageReadStats(id, uint64(load), 0) } tc.UpdateStorageReadStats(1, 20*units.MiB*utils.StoreHeartBeatReportInterval, 0) - err = tc.SetRule(&placement.Rule{ - GroupID: placement.DefaultGroupID, - ID: placement.DefaultRuleID, - Role: placement.Voter, - Count: 3, - LabelConstraints: []placement.LabelConstraint{ - {Key: "pool", Op: placement.In, Values: []string{"target"}}, - }, - }) - re.NoError(err) + setPoolPlacementRule(t, tc) interval := uint64(utils.StoreHeartBeatReportInterval) tc.AddRegionWithPeerReadInfo( From bb4f54ef876f5930af4a358c58f8d0c4aaff399d Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Tue, 28 Jul 2026 14:57:00 +0800 Subject: [PATCH 4/9] schedule: make placement scope cache source independent Signed-off-by: Ryan Leung --- pkg/schedule/schedulers/hot_region_solver.go | 19 +++++++++++++++---- pkg/schedule/schedulers/hot_region_test.go | 17 +++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/pkg/schedule/schedulers/hot_region_solver.go b/pkg/schedule/schedulers/hot_region_solver.go index e8dc869fff..5c35dc7804 100644 --- a/pkg/schedule/schedulers/hot_region_solver.go +++ b/pkg/schedule/schedulers/hot_region_solver.go @@ -224,7 +224,7 @@ func (bs *balanceSolver) solve() []*operator.Operator { engine = 1 } if !placementChecked[engine] { - placementCanRestrict[engine] = bs.mayUsePlacementScope(srcStore) + placementCanRestrict[engine] = bs.mayUsePlacementScope(srcStore.IsTiKV()) placementChecked[engine] = true } if !placementCanRestrict[engine] { @@ -790,10 +790,21 @@ func (bs *balanceSolver) prepareForRegion() { } } -func (bs *balanceSolver) mayUsePlacementScope(source *statistics.StoreLoadDetail) bool { +func (bs *balanceSolver) mayUsePlacementScope(isTiKV bool) bool { for _, rule := range bs.GetRuleManager().GetAllRules() { - if bs.ruleRestrictsStoreLoad(rule, source) { - return true + var matched, unmatched bool + for _, detail := range bs.stLoadDetail { + if detail.IsTiKV() != isTiKV { + continue + } + if placement.MatchLabelConstraints(detail.StoreInfo, rule.LabelConstraints) { + matched = true + } else { + unmatched = true + } + if matched && unmatched { + return true + } } } return false diff --git a/pkg/schedule/schedulers/hot_region_test.go b/pkg/schedule/schedulers/hot_region_test.go index 43a86bb948..1260b80e32 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -437,6 +437,23 @@ func setPoolPlacementRule(t *testing.T, tc *mockcluster.Cluster) { })) } +func TestMayUsePlacementScopeIsSourceIndependent(t *testing.T) { + cancel, _, tc, _ := prepareSchedulersTest() + defer cancel() + + details := make(map[uint64]*statistics.StoreLoadDetail, 2) + for id, pool := range map[uint64]string{1: "target", 2: "other"} { + tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + details[id] = &statistics.StoreLoadDetail{ + StoreSummaryInfo: &statistics.StoreSummaryInfo{StoreInfo: tc.GetStore(id)}, + } + } + setPoolPlacementRule(t, tc) + + bs := &balanceSolver{SchedulerCluster: tc, stLoadDetail: details} + require.True(t, bs.mayUsePlacementScope(true)) +} + func TestHotWriteRegionScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { testCases := []struct { name string From 0302db8821f63ef1eeba729080b73d7aae6a1d1e Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Wed, 29 Jul 2026 17:28:45 +0800 Subject: [PATCH 5/9] schedule: preserve safeguards in placement-scoped hot scheduling Calculate current, historical, and variance safeguards from each effective placement scope instead of disabling them. Reuse the global fast path, cache single-rule scopes, and avoid scanning or cloning the full rule set in the scheduler loop. Derive leader scopes from all same-role fits and validate revert regions against their own placement fit before creating the reverse operator. Signed-off-by: Ryan Leung --- pkg/schedule/placement/config.go | 45 ++- pkg/schedule/placement/label_constraint.go | 14 +- pkg/schedule/placement/rule_manager.go | 11 + pkg/schedule/placement/rule_manager_test.go | 33 +++ pkg/schedule/schedulers/hot_region_solver.go | 276 +++++++++++++----- .../schedulers/hot_region_solver_test.go | 80 +++++ pkg/schedule/schedulers/hot_region_test.go | 72 ++++- pkg/statistics/store_hot_peers_infos.go | 99 ++----- pkg/statistics/store_load.go | 62 ++++ 9 files changed, 520 insertions(+), 172 deletions(-) diff --git a/pkg/schedule/placement/config.go b/pkg/schedule/placement/config.go index 53cb063653..9b552d2357 100644 --- a/pkg/schedule/placement/config.go +++ b/pkg/schedule/placement/config.go @@ -18,12 +18,15 @@ import ( "bytes" "encoding/json" "time" + + "github.com/tikv/pd/pkg/core" ) // ruleConfig contains rule and rule group configurations. type ruleConfig struct { - rules map[[2]string]*Rule // {group, id} => Rule - groups map[string]*RuleGroup // id => RuleGroup + rules map[[2]string]*Rule // {group, id} => Rule + groups map[string]*RuleGroup // id => RuleGroup + mayRestrictStoreLoad [2]bool } func newRuleConfig() *ruleConfig { @@ -35,6 +38,7 @@ func newRuleConfig() *ruleConfig { // adjust configs for `buildRuleList` and API use. func (c *ruleConfig) adjust() { + c.mayRestrictStoreLoad = [2]bool{} // remove all default group configurations. // if there are rules belong to the group, it will be re-add later. for id, g := range c.groups { @@ -43,6 +47,10 @@ func (c *ruleConfig) adjust() { } } for _, r := range c.rules { + if len(r.LabelConstraints) > 0 { + c.mayRestrictStoreLoad[0] = c.mayRestrictStoreLoad[0] || ruleMayRestrictEngine(r, "", core.EngineTiKV) + c.mayRestrictStoreLoad[1] = c.mayRestrictStoreLoad[1] || ruleMayRestrictEngine(r, core.EngineTiFlash) + } g := c.groups[r.GroupID] if g == nil { // create default group configurations. @@ -54,6 +62,39 @@ func (c *ruleConfig) adjust() { } } +func ruleMayRestrictEngine(rule *Rule, values ...string) bool { + hasOtherConstraint := false + for i := range rule.LabelConstraints { + if rule.LabelConstraints[i].Key != core.EngineKey { + hasOtherConstraint = true + break + } + } + + matched := 0 + for _, value := range values { + hasEngineConstraint, matchesValue := false, true + for i := range rule.LabelConstraints { + constraint := &rule.LabelConstraints[i] + if constraint.Key != core.EngineKey { + continue + } + hasEngineConstraint = true + if !constraint.matchValue(value) { + matchesValue = false + break + } + } + if value != "" && !hasEngineConstraint { + matchesValue = false + } + if matchesValue { + matched++ + } + } + return matched > 0 && (hasOtherConstraint || matched < len(values)) +} + func (c *ruleConfig) getRule(key [2]string) *Rule { return c.rules[key] } diff --git a/pkg/schedule/placement/label_constraint.go b/pkg/schedule/placement/label_constraint.go index 6cc2d4ed45..77b8beb9d2 100644 --- a/pkg/schedule/placement/label_constraint.go +++ b/pkg/schedule/placement/label_constraint.go @@ -51,17 +51,19 @@ type LabelConstraint struct { // MatchStore checks if a store matches the constraint. func (c *LabelConstraint) MatchStore(store *core.StoreInfo) bool { + return c.matchValue(store.GetLabelValue(c.Key)) +} + +func (c *LabelConstraint) matchValue(value string) bool { switch c.Op { case In: - label := store.GetLabelValue(c.Key) - return label != "" && slice.AnyOf(c.Values, func(i int) bool { return c.Values[i] == label }) + return value != "" && slice.AnyOf(c.Values, func(i int) bool { return c.Values[i] == value }) case NotIn: - label := store.GetLabelValue(c.Key) - return label == "" || slice.NoneOf(c.Values, func(i int) bool { return c.Values[i] == label }) + return value == "" || slice.NoneOf(c.Values, func(i int) bool { return c.Values[i] == value }) case Exists: - return store.GetLabelValue(c.Key) != "" + return value != "" case NotExists: - return store.GetLabelValue(c.Key) == "" + return value == "" } return false } diff --git a/pkg/schedule/placement/rule_manager.go b/pkg/schedule/placement/rule_manager.go index 49567b8c03..1e38d84837 100644 --- a/pkg/schedule/placement/rule_manager.go +++ b/pkg/schedule/placement/rule_manager.go @@ -368,6 +368,17 @@ func (m *RuleManager) GetRulesCount() int { return len(m.ruleConfig.rules) } +// MayRestrictStoreLoad returns whether placement rules may restrict the load +// statistics population for the given store engine. +func (m *RuleManager) MayRestrictStoreLoad(isTiKV bool) bool { + m.RLock() + defer m.RUnlock() + if isTiKV { + return m.ruleConfig.mayRestrictStoreLoad[0] + } + return m.ruleConfig.mayRestrictStoreLoad[1] +} + // GetGroupsCount returns the number of rule groups. func (m *RuleManager) GetGroupsCount() int { m.RLock() diff --git a/pkg/schedule/placement/rule_manager_test.go b/pkg/schedule/placement/rule_manager_test.go index 3263f6e593..6f9412fcc1 100644 --- a/pkg/schedule/placement/rule_manager_test.go +++ b/pkg/schedule/placement/rule_manager_test.go @@ -78,6 +78,39 @@ func TestDefault2(t *testing.T) { re.Equal([]string{"zone", "rack", "host"}, rules[1].LocationLabels) } +func TestMayRestrictStoreLoad(t *testing.T) { + re := require.New(t) + _, manager := newTestManager(t, false) + re.False(manager.MayRestrictStoreLoad(true)) + re.False(manager.MayRestrictStoreLoad(false)) + + setConstraints := func(constraints ...LabelConstraint) { + re.NoError(manager.SetRule(&Rule{ + GroupID: "group", ID: "rule", Role: Voter, Count: 1, + LabelConstraints: constraints, + })) + } + + setConstraints(LabelConstraint{Key: "zone", Op: In, Values: []string{"z1"}}) + re.True(manager.MayRestrictStoreLoad(true)) + re.False(manager.MayRestrictStoreLoad(false)) + + setConstraints(LabelConstraint{Key: core.EngineKey, Op: In, Values: []string{core.EngineTiFlash}}) + re.False(manager.MayRestrictStoreLoad(true)) + re.False(manager.MayRestrictStoreLoad(false)) + + setConstraints( + LabelConstraint{Key: core.EngineKey, Op: In, Values: []string{core.EngineTiFlash}}, + LabelConstraint{Key: "zone", Op: In, Values: []string{"z1"}}, + ) + re.False(manager.MayRestrictStoreLoad(true)) + re.True(manager.MayRestrictStoreLoad(false)) + + setConstraints() + re.False(manager.MayRestrictStoreLoad(true)) + re.False(manager.MayRestrictStoreLoad(false)) +} + func TestAdjustRule(t *testing.T) { re := require.New(t) _, manager := newTestManager(t, false) diff --git a/pkg/schedule/schedulers/hot_region_solver.go b/pkg/schedule/schedulers/hot_region_solver.go index 5c35dc7804..6cc291857f 100644 --- a/pkg/schedule/schedulers/hot_region_solver.go +++ b/pkg/schedule/schedulers/hot_region_solver.go @@ -54,9 +54,11 @@ type balanceSolver struct { cur *solution // curFit is the placement-rule fit for the current region. curFit *placement.RegionFit - // skipLoadExpectation indicates whether the current region should skip the - // expectation and variance calculated from all stores. - skipLoadExpectation bool + // curScope and revertScope contain placement-scoped load statistics. A nil + // scope uses the existing engine-wide statistics. + curScope *placementLoadScope + revertScope *placementLoadScope + placementScopeCache map[placementScopeKey]*placementLoadScope best *solution ops []*operator.Operator @@ -79,6 +81,16 @@ type balanceSolver struct { rank } +type placementScopeKey struct { + rule *placement.Rule + isTiKV bool +} + +type placementLoadScope struct { + expect statistics.StoreLoad + stddev statistics.StoreLoad +} + func (bs *balanceSolver) init() { // Load the configuration items of the scheduler. bs.resourceTy = toResourceType(bs.rwTy, bs.opTy) @@ -208,37 +220,50 @@ func (bs *balanceSolver) solve() []*operator.Operator { splitThresholds := bs.sche.conf.getSplitThresholds() placementEnabled := bs.GetSchedulerConfig().IsPlacementRulesEnabled() var placementChecked, placementCanRestrict [2]bool + mayUsePlacementScope := func(store *statistics.StoreLoadDetail) bool { + if !placementEnabled { + return false + } + engine := 0 + if !store.IsTiKV() { + engine = 1 + } + if !placementChecked[engine] { + placementCanRestrict[engine] = bs.mayUsePlacementScope(store.IsTiKV()) + placementChecked[engine] = true + } + return placementCanRestrict[engine] + } for _, srcStore := range bs.filterSrcStores() { bs.cur.srcStore = srcStore srcStoreID := srcStore.GetID() - sourceFailure := bs.sourceStoreFailure(srcStore) - sourceResultRecorded := sourceFailure == "" - if sourceResultRecorded { - bs.recordSourceStoreResult("src-store-succ", srcStoreID) - } else if !placementEnabled { - bs.recordSourceStoreResult(sourceFailure, srcStoreID) - continue - } else { - engine := 0 - if !srcStore.IsTiKV() { - engine = 1 - } - if !placementChecked[engine] { - placementCanRestrict[engine] = bs.mayUsePlacementScope(srcStore.IsTiKV()) - placementChecked[engine] = true - } - if !placementCanRestrict[engine] { - bs.recordSourceStoreResult(sourceFailure, srcStoreID) + usePlacementScope := mayUsePlacementScope(srcStore) + globalSourceFailure := bs.sourceStoreFailure(srcStore, nil) + sourceFailure := globalSourceFailure + sourceResultRecorded := false + if !usePlacementScope { + bs.recordSourceStoreResult(sourceFailureOrSuccess(sourceFailure), srcStoreID) + sourceResultRecorded = true + if sourceFailure != "" { continue } } + checkedRegion := false for _, mainPeerStat := range bs.filteredHotPeers[srcStoreID] { if bs.cur.region = bs.getRegion(mainPeerStat, srcStoreID); bs.cur.region == nil { continue } - bs.prepareForRegion() - if !bs.skipLoadExpectation && sourceFailure != "" { - continue + checkedRegion = true + bs.curScope = nil + if usePlacementScope { + bs.curScope = bs.prepareForRegion(bs.cur.region, bs.curFit, srcStore) + sourceFailure = globalSourceFailure + if bs.curScope != nil { + sourceFailure = bs.sourceStoreFailure(srcStore, bs.curScope) + } + if sourceFailure != "" { + continue + } } if !sourceResultRecorded { bs.recordSourceStoreResult("src-store-succ", srcStoreID) @@ -271,10 +296,22 @@ func (bs *balanceSolver) solve() []*operator.Operator { if bs.needSearchRevertRegions() { hotSchedulerSearchRevertRegionsCounter.Inc() dstStoreID := dstStore.GetID() + mainFit := bs.curFit for _, revertPeerStat := range bs.filteredHotPeers[dstStoreID] { revertRegion := bs.getRegion(revertPeerStat, dstStoreID) + revertFit := bs.curFit + bs.curFit = mainFit if revertRegion == nil || revertRegion.GetID() == bs.cur.region.GetID() || - !allowRevertRegion(revertRegion, srcStoreID) { + !allowRevertRegion(revertRegion, srcStoreID) || + placementEnabled && !bs.isRevertRegionValid(revertRegion, revertFit) { + continue + } + bs.revertScope = nil + if mayUsePlacementScope(dstStore) { + bs.revertScope = bs.prepareForRegion(revertRegion, revertFit, dstStore) + } + if bs.revertScope != nil && (bs.sourceStoreFailure(dstStore, bs.revertScope) != "" || + bs.destinationStoreFailure(srcStore, bs.revertScope, bs.dstToleranceRatio(srcStore)) != "") { continue } bs.cur.revertPeerStat = revertPeerStat @@ -282,13 +319,18 @@ func (bs *balanceSolver) solve() []*operator.Operator { bs.calcProgressiveRank() tryUpdateBestSolution() } + bs.curFit = mainFit bs.cur.revertPeerStat = nil bs.cur.revertRegion = nil + bs.revertScope = nil } } } if !sourceResultRecorded { - bs.recordSourceStoreResult(sourceFailure, srcStoreID) + if !checkedRegion { + sourceFailure = globalSourceFailure + } + bs.recordSourceStoreResult(sourceFailureOrSuccess(sourceFailure), srcStoreID) } } @@ -435,20 +477,35 @@ func (bs *balanceSolver) filterSrcStores() map[uint64]*statistics.StoreLoadDetai return ret } -func (bs *balanceSolver) sourceStoreFailure(detail *statistics.StoreLoadDetail) string { +func (bs *balanceSolver) sourceStoreFailure(detail *statistics.StoreLoadDetail, scope *placementLoadScope) string { srcToleranceRatio := bs.sche.conf.getSrcToleranceRatio() if !detail.IsTiKV() { srcToleranceRatio += tiflashToleranceRatioCorrection } - if !bs.checkSrcByPriorityAndTolerance(detail.LoadPred.Min(), &detail.LoadPred.Expect, srcToleranceRatio) { + expect := bs.expectLoad(detail, scope) + if !bs.checkSrcByPriorityAndTolerance(detail.LoadPred.Min(), expect, srcToleranceRatio) { return "src-store-failed" } - if !bs.checkSrcHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, &detail.LoadPred.Expect, srcToleranceRatio) { + if !bs.checkSrcHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, expect, srcToleranceRatio) { return "src-store-history-loads-failed" } return "" } +func sourceFailureOrSuccess(failure string) string { + if failure == "" { + return "src-store-succ" + } + return failure +} + +func (*balanceSolver) expectLoad(detail *statistics.StoreLoadDetail, scope *placementLoadScope) *statistics.StoreLoad { + if scope != nil { + return &scope.expect + } + return &detail.LoadPred.Expect +} + func (bs *balanceSolver) recordSourceStoreResult(result string, storeID uint64) { hotSchedulerResultCounter.WithLabelValues(result+"-"+bs.resourceTy.String(), strconv.FormatUint(storeID, 10)).Inc() } @@ -673,13 +730,25 @@ func (bs *balanceSolver) filterDstStores() map[uint64]*statistics.StoreLoadDetai return bs.pickDstStores(filters, candidates) } +func (bs *balanceSolver) isRevertRegionValid(region *core.RegionInfo, fit *placement.RegionFit) bool { + source, target := bs.cur.dstStore.StoreInfo, bs.cur.srcStore.StoreInfo + var safeguard filter.Filter + switch bs.opTy { + case movePeer: + safeguard = filter.NewPlacementSafeguard(bs.sche.GetName(), bs.GetSchedulerConfig(), bs.GetBasicCluster(), + bs.GetRuleManager(), region, source, fit) + case transferLeader: + safeguard = filter.NewPlacementLeaderSafeguardWithFit(bs.sche.GetName(), bs.GetSchedulerConfig(), bs.GetBasicCluster(), + bs.GetRuleManager(), region, source, fit, bs.rwTy == utils.Read) + } + return safeguard == nil || safeguard.Target(bs.GetSchedulerConfig(), target).IsOK() +} + func (bs *balanceSolver) pickDstStores(filters []filter.Filter, candidates []*statistics.StoreLoadDetail) map[uint64]*statistics.StoreLoadDetail { ret := make(map[uint64]*statistics.StoreLoadDetail, len(candidates)) - confDstToleranceRatio := bs.sche.conf.getDstToleranceRatio() confEnableForTiFlash := bs.sche.conf.getEnableForTiFlash() for _, detail := range candidates { store := detail.StoreInfo - dstToleranceRatio := confDstToleranceRatio if !detail.IsTiKV() { if !confEnableForTiFlash || detail.IsTiFlashCompute() { continue @@ -687,16 +756,11 @@ func (bs *balanceSolver) pickDstStores(filters []filter.Filter, candidates []*st if bs.rwTy != utils.Write || bs.opTy != movePeer { continue } - dstToleranceRatio += tiflashToleranceRatioCorrection } if filter.Target(bs.GetSchedulerConfig(), store, filters) { id := store.GetID() - if !bs.skipLoadExpectation && !bs.checkDstByPriorityAndTolerance(detail.LoadPred.Max(), &detail.LoadPred.Expect, dstToleranceRatio) { - hotSchedulerResultCounter.WithLabelValues("dst-store-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() - continue - } - if !bs.skipLoadExpectation && !bs.checkDstHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, &detail.LoadPred.Expect, dstToleranceRatio) { - hotSchedulerResultCounter.WithLabelValues("dst-store-history-loads-failed-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() + if failure := bs.destinationStoreFailure(detail, bs.curScope, bs.dstToleranceRatio(detail)); failure != "" { + hotSchedulerResultCounter.WithLabelValues(failure+"-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() continue } @@ -707,6 +771,25 @@ func (bs *balanceSolver) pickDstStores(filters []filter.Filter, candidates []*st return ret } +func (bs *balanceSolver) dstToleranceRatio(detail *statistics.StoreLoadDetail) float64 { + ratio := bs.sche.conf.getDstToleranceRatio() + if !detail.IsTiKV() { + ratio += tiflashToleranceRatioCorrection + } + return ratio +} + +func (bs *balanceSolver) destinationStoreFailure(detail *statistics.StoreLoadDetail, scope *placementLoadScope, toleranceRatio float64) string { + expect := bs.expectLoad(detail, scope) + if !bs.checkDstByPriorityAndTolerance(detail.LoadPred.Max(), expect, toleranceRatio) { + return "dst-store-failed" + } + if !bs.checkDstHistoryLoadsByPriorityAndTolerance(&detail.LoadPred.Current, expect, toleranceRatio) { + return "dst-store-history-loads-failed" + } + return "" +} + func (bs *balanceSolver) checkDstByPriorityAndTolerance(maxLoad, expect *statistics.StoreLoad, toleranceRatio float64) bool { return bs.checkByPriorityAndTolerance(maxLoad.Loads, func(i int) bool { return maxLoad.Loads[i]*toleranceRatio < expect.Loads[i] @@ -769,66 +852,113 @@ func (bs *balanceSolver) checkHistoryLoadsByPriorityAndToleranceFirstOnly(_ stat } func (bs *balanceSolver) enableExpectation() bool { - return !bs.skipLoadExpectation && bs.sche.conf.getDstToleranceRatio() > 0 && bs.sche.conf.getSrcToleranceRatio() > 0 + return bs.sche.conf.getDstToleranceRatio() > 0 && bs.sche.conf.getSrcToleranceRatio() > 0 } -// prepareForRegion determines whether the current region should use the global -// load expectation. MatchLabelConstraints also covers implicit exclusive labels. -func (bs *balanceSolver) prepareForRegion() { - bs.skipLoadExpectation = false - if bs.curFit == nil { - return +// prepareForRegion derives the stores that can legally carry the current load. +// Leaders can move between different RuleFits with the same placement role. +func (bs *balanceSolver) prepareForRegion(region *core.RegionInfo, fit *placement.RegionFit, source *statistics.StoreLoadDetail) *placementLoadScope { + if fit == nil { + return nil } - - sourcePeer := bs.cur.region.GetStorePeer(bs.cur.srcStore.GetID()) + sourcePeer := region.GetStorePeer(source.GetID()) if sourcePeer == nil { - return + return nil } - ruleFit := bs.curFit.GetRuleFit(sourcePeer.GetId()) - if ruleFit != nil { - bs.skipLoadExpectation = bs.ruleRestrictsStoreLoad(ruleFit.Rule, bs.cur.srcStore) + sourceFit := fit.GetRuleFit(sourcePeer.GetId()) + if sourceFit == nil { + return nil } -} -func (bs *balanceSolver) mayUsePlacementScope(isTiKV bool) bool { - for _, rule := range bs.GetRuleManager().GetAllRules() { - var matched, unmatched bool - for _, detail := range bs.stLoadDetail { - if detail.IsTiKV() != isTiKV { - continue - } - if placement.MatchLabelConstraints(detail.StoreInfo, rule.LabelConstraints) { - matched = true - } else { - unmatched = true - } - if matched && unmatched { - return true + rules := []*placement.Rule{sourceFit.Rule} + if bs.opTy == transferLeader { + rules = rules[:0] + for _, ruleFit := range fit.RuleFits { + if ruleFit.Rule.Role == sourceFit.Rule.Role { + rules = append(rules, ruleFit.Rule) } } } - return false + return bs.getPlacementLoadScope(rules, source.IsTiKV()) } -func (bs *balanceSolver) ruleRestrictsStoreLoad(rule *placement.Rule, source *statistics.StoreLoadDetail) bool { - if !placement.MatchLabelConstraints(source.StoreInfo, rule.LabelConstraints) { - return false +func (bs *balanceSolver) mayUsePlacementScope(isTiKV bool) bool { + if bs.GetRuleManager().MayRestrictStoreLoad(isTiKV) { + return true + } + var constraints []placement.LabelConstraint + if !isTiKV { + constraints = []placement.LabelConstraint{ + {Key: core.EngineKey, Op: placement.In, Values: []string{core.EngineTiFlash}}, + } } for _, detail := range bs.stLoadDetail { - if detail.IsTiKV() == source.IsTiKV() && !placement.MatchLabelConstraints(detail.StoreInfo, rule.LabelConstraints) { + if detail.IsTiKV() == isTiKV && !placement.MatchLabelConstraints(detail.StoreInfo, constraints) { return true } } return false } +// getPlacementLoadScope calculates local safeguards only when the placement +// population is narrower than the existing engine-wide population. +func (bs *balanceSolver) getPlacementLoadScope(rules []*placement.Rule, isTiKV bool) *placementLoadScope { + var key placementScopeKey + cacheable := len(rules) == 1 + if cacheable { + key = placementScopeKey{rule: rules[0], isTiKV: isTiKV} + if scope, ok := bs.placementScopeCache[key]; ok { + return scope + } + } + + allCount := 0 + var summary statistics.StoreLoadSummary + details := make([]*statistics.StoreLoadDetail, 0, len(bs.stLoadDetail)) + for _, detail := range bs.stLoadDetail { + if detail.IsTiKV() != isTiKV { + continue + } + allCount++ + if !slice.AnyOf(rules, func(i int) bool { + return placement.MatchLabelConstraints(detail.StoreInfo, rules[i].LabelConstraints) + }) { + continue + } + details = append(details, detail) + summary.Add(&detail.LoadPred.Current) + } + var scope *placementLoadScope + if len(details) < allCount { + expect, stddev := summary.Result(details) + scope = &placementLoadScope{expect: expect, stddev: stddev} + } + if cacheable { + if bs.placementScopeCache == nil { + bs.placementScopeCache = make(map[placementScopeKey]*placementLoadScope) + } + bs.placementScopeCache[key] = scope + } + return scope +} + func (bs *balanceSolver) isUniformFirstPriority(store *statistics.StoreLoadDetail) bool { // first priority should be more uniform than second priority - return store.IsUniform(bs.firstPriority, stddevThreshold*0.5) + return bs.isUniform(store, bs.firstPriority, stddevThreshold*0.5) } func (bs *balanceSolver) isUniformSecondPriority(store *statistics.StoreLoadDetail) bool { - return store.IsUniform(bs.secondPriority, stddevThreshold) + return bs.isUniform(store, bs.secondPriority, stddevThreshold) +} + +func (bs *balanceSolver) isUniform(store *statistics.StoreLoadDetail, dim int, threshold float64) bool { + uniform := func(scope *placementLoadScope) bool { + if scope != nil { + return scope.stddev.Loads[dim] < threshold + } + return store.IsUniform(dim, threshold) + } + return uniform(bs.curScope) || bs.cur.revertRegion != nil && uniform(bs.revertScope) } // isTolerance checks source store and target store by checking the difference value with pendingAmpFactor * pendingPeer. diff --git a/pkg/schedule/schedulers/hot_region_solver_test.go b/pkg/schedule/schedulers/hot_region_solver_test.go index 9a284c625b..761a976089 100644 --- a/pkg/schedule/schedulers/hot_region_solver_test.go +++ b/pkg/schedule/schedulers/hot_region_solver_test.go @@ -15,6 +15,7 @@ package schedulers import ( + "math" "testing" "time" @@ -26,6 +27,7 @@ import ( "github.com/tikv/pd/pkg/core" "github.com/tikv/pd/pkg/schedule/operator" + "github.com/tikv/pd/pkg/schedule/placement" "github.com/tikv/pd/pkg/schedule/types" "github.com/tikv/pd/pkg/statistics" "github.com/tikv/pd/pkg/statistics/buckets" @@ -656,6 +658,84 @@ func TestExpect(t *testing.T) { } } +func TestPlacementLoadScopePreservesExpectationGuards(t *testing.T) { + re := require.New(t) + details := make(map[uint64]*statistics.StoreLoadDetail) + for id, load := range map[uint64]float64{1: 10.5, 2: 9.5, 3: 10, 4: 10, 5: 100, 6: 100} { + pool := "other" + if id <= 4 { + pool = "target" + } + current := statistics.StoreLoad{ + Loads: statistics.Loads{load, 10}, + HotPeerCount: 1, + HistoryLoads: statistics.HistoryLoads{{10}, {10}}, + } + details[id] = &statistics.StoreLoadDetail{ + StoreSummaryInfo: &statistics.StoreSummaryInfo{StoreInfo: core.NewStoreInfoWithLabel(id, map[string]string{"pool": pool})}, + LoadPred: current.ToLoadPred(utils.Write, nil), + } + } + bs := &balanceSolver{ + stLoadDetail: details, firstPriority: utils.ByteDim, secondPriority: utils.KeyDim, + resourceTy: writePeer, + } + bs.rank = initRankV2(bs) + rule := &placement.Rule{LabelConstraints: []placement.LabelConstraint{ + {Key: "pool", Op: placement.In, Values: []string{"target"}}, + }} + scope := bs.getPlacementLoadScope([]*placement.Rule{rule}, true) + re.NotNil(scope) + re.Equal(float64(10), scope.expect.Loads[utils.ByteDim]) + re.Equal([]float64{10}, scope.expect.HistoryLoads[utils.ByteDim]) + re.InDelta(math.Sqrt(0.125)/10, scope.stddev.Loads[utils.ByteDim], 1e-9) + re.False(bs.checkSrcByPriorityAndTolerance(details[1].LoadPred.Min(), &scope.expect, 1.05)) + re.False(bs.checkSrcHistoryLoadsByPriorityAndTolerance(&details[1].LoadPred.Current, &scope.expect, 1.05)) + + bs.cur = &solution{} + bs.curScope = scope + re.True(bs.isUniformFirstPriority(details[1])) + re.Nil(bs.getPlacementLoadScope([]*placement.Rule{{}}, true)) +} + +func TestRevertRegionPlacementSafeguard(t *testing.T) { + re := require.New(t) + cancel, _, tc, oc := prepareSchedulersTest() + defer cancel() + tc.SetEnablePlacementRules(true) + for id, labels := range map[uint64]map[string]string{ + 1: {"a": "yes"}, + 2: {"a": "yes", "b": "yes"}, + 3: {"a": "yes"}, + 4: {"a": "yes"}, + } { + tc.AddLabelsStore(id, 1, labels) + } + re.NoError(tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, ID: placement.DefaultRuleID, + Role: placement.Voter, Count: 2, + LabelConstraints: []placement.LabelConstraint{{Key: "a", Op: placement.In, Values: []string{"yes"}}}, + })) + re.NoError(tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, ID: "voter-b", + Role: placement.Voter, Count: 1, + LabelConstraints: []placement.LabelConstraint{{Key: "b", Op: placement.In, Values: []string{"yes"}}}, + })) + + peers := []*metapb.Peer{{Id: 2, StoreId: 2}, {Id: 3, StoreId: 3}, {Id: 4, StoreId: 4}} + region := core.NewRegionInfo(&metapb.Region{Id: 1, Peers: peers}, peers[1]) + fit := tc.GetRuleManager().FitRegion(tc.GetBasicCluster(), region) + re.True(fit.IsSatisfied()) + scheduler, err := CreateScheduler(writeType, oc, storage.NewStorageWithMemoryBackend(), nil) + re.NoError(err) + bs := newBalanceSolver(scheduler.(*hotScheduler), tc, utils.Write, movePeer) + bs.cur = &solution{ + srcStore: &statistics.StoreLoadDetail{StoreSummaryInfo: &statistics.StoreSummaryInfo{StoreInfo: tc.GetStore(1)}}, + dstStore: &statistics.StoreLoadDetail{StoreSummaryInfo: &statistics.StoreSummaryInfo{StoreInfo: tc.GetStore(2)}}, + } + re.False(bs.isRevertRegionValid(region, fit)) +} + func TestBucketFirstStat(t *testing.T) { re := require.New(t) testdata := []struct { diff --git a/pkg/schedule/schedulers/hot_region_test.go b/pkg/schedule/schedulers/hot_region_test.go index 1260b80e32..c5fe5eb32f 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -437,24 +437,30 @@ func setPoolPlacementRule(t *testing.T, tc *mockcluster.Cluster) { })) } -func TestMayUsePlacementScopeIsSourceIndependent(t *testing.T) { +func TestMayUsePlacementScopeIgnoresEngineSeparation(t *testing.T) { cancel, _, tc, _ := prepareSchedulersTest() defer cancel() - - details := make(map[uint64]*statistics.StoreLoadDetail, 2) - for id, pool := range map[uint64]string{1: "target", 2: "other"} { - tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + details := make(map[uint64]*statistics.StoreLoadDetail) + addStore := func(id uint64, labels map[string]string) { + tc.AddLabelsStore(id, 1, labels) details[id] = &statistics.StoreLoadDetail{ StoreSummaryInfo: &statistics.StoreSummaryInfo{StoreInfo: tc.GetStore(id)}, } } - setPoolPlacementRule(t, tc) + addStore(1, map[string]string{core.EngineKey: core.EngineTiFlash}) + addStore(2, nil) bs := &balanceSolver{SchedulerCluster: tc, stLoadDetail: details} + require.False(t, bs.mayUsePlacementScope(false)) + require.False(t, bs.mayUsePlacementScope(true)) + + addStore(3, map[string]string{core.EngineKey: core.EngineTiKV}) require.True(t, bs.mayUsePlacementScope(true)) + addStore(4, map[string]string{core.EngineKey: core.EngineTiFlash, "$group": "other"}) + require.True(t, bs.mayUsePlacementScope(false)) } -func TestHotWriteRegionScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { +func TestHotWriteRegionScheduleWithPlacementScope(t *testing.T) { testCases := []struct { name string storeLoads map[uint64]float64 @@ -533,7 +539,7 @@ func TestHotWriteRegionScheduleWithPlacementConstraintsIgnoresGlobalExpectation( } } -func TestHotReadLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { +func TestHotReadLeaderScheduleWithPlacementScope(t *testing.T) { re := require.New(t) tc, hb := preparePlacementHotScheduler(t, readType, readLeader) hb.conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority} @@ -560,7 +566,7 @@ func TestHotReadLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t operatorutil.CheckTransferLeaderFrom(re, ops[0], operator.OpHotRegion, 1) } -func TestHotWriteLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { +func TestHotWriteLeaderScheduleWithPlacementScope(t *testing.T) { re := require.New(t) tc, hb := preparePlacementHotScheduler(t, writeType, writeLeader) hb.conf.WriteLeaderPriorities = []string{utils.BytePriority, utils.KeyPriority} @@ -578,8 +584,8 @@ func TestHotWriteLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation( addRegionInfo(tc, utils.Write, []testRegionInfo{ {1, []uint64{1, 2, 3}, 2.5 * units.MiB, 0, 0}, {2, []uint64{1, 2, 3}, 17.5 * units.MiB, 0, 0}, - {3, []uint64{2, 1, 3}, 10 * units.MiB, 0, 0}, - {4, []uint64{3, 1, 2}, 10 * units.MiB, 0, 0}, + {3, []uint64{2, 1, 3}, 0, 0, 0}, + {4, []uint64{3, 1, 2}, 0, 0, 0}, }) ops, _ := hb.Schedule(tc, false) @@ -588,7 +594,49 @@ func TestHotWriteLeaderScheduleWithPlacementConstraintsIgnoresGlobalExpectation( operatorutil.CheckTransferLeaderFrom(re, ops[0], operator.OpHotRegion, 1) } -func TestHotReadPeerScheduleWithPlacementConstraintsIgnoresGlobalExpectation(t *testing.T) { +func TestHotWriteLeaderScheduleAcrossSameRoleRules(t *testing.T) { + re := require.New(t) + tc, hb := preparePlacementHotScheduler(t, writeType, writeLeader) + tc.SetClusterVersion(versioninfo.MinSupportedVersion(versioninfo.HotScheduleWithQuery)) + hb.conf.WriteLeaderPriorities = []string{utils.QueryPriority, utils.BytePriority} + + for id := uint64(1); id <= 6; id++ { + pool := "B" + if id == 1 || id == 4 { + pool = "A" + } + tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + query := uint64(0) + switch id { + case 1: + query = 30 * utils.StoreHeartBeatReportInterval + case 4: + query = 100 * utils.StoreHeartBeatReportInterval + } + tc.UpdateStorageWriteQuery(id, query) + } + re.NoError(tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, ID: placement.DefaultRuleID, + Role: placement.Voter, Count: 1, + LabelConstraints: []placement.LabelConstraint{{Key: "pool", Op: placement.In, Values: []string{"A"}}}, + })) + re.NoError(tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, ID: "voter-b", + Role: placement.Voter, Count: 2, + LabelConstraints: []placement.LabelConstraint{{Key: "pool", Op: placement.In, Values: []string{"B"}}}, + })) + + addRegionInfo(tc, utils.Write, []testRegionInfo{ + {1, []uint64{1, 2, 3}, 12 * units.MiB, 0, 12}, + {2, []uint64{1, 2, 3}, 18 * units.MiB, 0, 18}, + }) + ops, _ := hb.Schedule(tc, false) + re.Len(ops, 1) + re.Equal(uint64(1), ops[0].RegionID()) + operatorutil.CheckTransferLeaderFrom(re, ops[0], operator.OpHotRegion, 1) +} + +func TestHotReadPeerScheduleWithPlacementScope(t *testing.T) { re := require.New(t) tc, hb := preparePlacementHotScheduler(t, readType, readPeer) hb.conf.ReadPriorities = []string{utils.BytePriority, utils.KeyPriority} diff --git a/pkg/statistics/store_hot_peers_infos.go b/pkg/statistics/store_hot_peers_infos.go index 5d58663c0e..ed93755ea8 100644 --- a/pkg/statistics/store_hot_peers_infos.go +++ b/pkg/statistics/store_hot_peers_infos.go @@ -15,8 +15,6 @@ package statistics import ( - "math" - "github.com/tikv/pd/pkg/core" "github.com/tikv/pd/pkg/core/constant" "github.com/tikv/pd/pkg/statistics/utils" @@ -155,13 +153,8 @@ func summaryStoresLoadByEngine( kind constant.ResourceKind, collector storeCollector, ) []*StoreLoadDetail { - var ( - loadDetail = make([]*StoreLoadDetail, 0, len(storeInfos)) - allStoreLoadSum Loads - allStoreHistoryLoadSum HistoryLoads - allStoreCount = 0 - allHotPeersCount = 0 - ) + loadDetail := make([]*StoreLoadDetail, 0, len(storeInfos)) + var summary StoreLoadSummary for _, info := range storeInfos { store := info.StoreInfo @@ -186,30 +179,19 @@ func summaryStoresLoadByEngine( var historyLoads HistoryLoads if storesHistoryLoads != nil { historyLoads = storesHistoryLoads.Get(id, rwTy, kind) - - for i, loads := range historyLoads { - if allStoreHistoryLoadSum[i] == nil || len(allStoreHistoryLoadSum[i]) < len(loads) { - allStoreHistoryLoadSum[i] = make([]float64, len(loads)) - } - for j, historyLoad := range loads { - allStoreHistoryLoadSum[i][j] += historyLoad - } - } - storesHistoryLoads.Add(id, rwTy, kind, currentLoads) } - for i := range allStoreLoadSum { - allStoreLoadSum[i] += currentLoads[i] - } - allStoreCount += 1 - allHotPeersCount += len(hotPeers) - // Build store load prediction from current load and pending influence. - stLoadPred := (&StoreLoad{ + currentLoad := StoreLoad{ Loads: currentLoads, HotPeerCount: float64(len(hotPeers)), HistoryLoads: historyLoads, - }).ToLoadPred(rwTy, info.PendingSum) + } + summary.Add(¤tLoad) + if storesHistoryLoads != nil { + storesHistoryLoads.Add(id, rwTy, kind, currentLoads) + } + stLoadPred := currentLoad.ToLoadPred(rwTy, info.PendingSum) // Construct store load info. loadDetail = append(loadDetail, &StoreLoadDetail{ @@ -219,73 +201,32 @@ func summaryStoresLoadByEngine( }) } - if allStoreCount == 0 { + if len(loadDetail) == 0 { return loadDetail } - - var ( - expectCount = float64(allHotPeersCount) / float64(allStoreCount) - expectLoads Loads - // TODO: remove some the max value or min value to avoid the effect of extreme value. - expectHistoryLoads HistoryLoads - stddevLoads Loads - ) - for i := range utils.DimLen { - expectLoads[i] = allStoreLoadSum[i] / float64(allStoreCount) - } - - for dim := range allStoreHistoryLoadSum { - expectHistoryLoads[dim] = make([]float64, len(allStoreHistoryLoadSum[dim])) - for j := range allStoreHistoryLoadSum[dim] { - expectHistoryLoads[dim][j] = allStoreHistoryLoadSum[dim][j] / float64(allStoreCount) - } - } - if allHotPeersCount != 0 { - for _, detail := range loadDetail { - for i := range expectLoads { - stddevLoads[i] += math.Pow(detail.LoadPred.Current.Loads[i]-expectLoads[i], 2) //nolint:staticcheck - } - } - for i := range stddevLoads { - stddevLoads[i] = math.Sqrt(stddevLoads[i] / float64(allStoreCount)) - if expectLoads[i] == 0 { - stddevLoads[i] = 0 - continue - } - stddevLoads[i] /= expectLoads[i] - } - } + expect, stddev := summary.Result(loadDetail) { // Metric for debug. engine := collector.engine() ty := "exp-byte-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(expectLoads[utils.ByteDim]) + hotPeerSummary.WithLabelValues(ty, engine).Set(expect.Loads[utils.ByteDim]) ty = "exp-key-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(expectLoads[utils.KeyDim]) + hotPeerSummary.WithLabelValues(ty, engine).Set(expect.Loads[utils.KeyDim]) ty = "exp-query-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(expectLoads[utils.QueryDim]) + hotPeerSummary.WithLabelValues(ty, engine).Set(expect.Loads[utils.QueryDim]) ty = "exp-cpu-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(expectLoads[utils.CPUDim]) + hotPeerSummary.WithLabelValues(ty, engine).Set(expect.Loads[utils.CPUDim]) ty = "exp-count-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(expectCount) + hotPeerSummary.WithLabelValues(ty, engine).Set(expect.HotPeerCount) ty = "stddev-byte-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(stddevLoads[utils.ByteDim]) + hotPeerSummary.WithLabelValues(ty, engine).Set(stddev.Loads[utils.ByteDim]) ty = "stddev-key-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(stddevLoads[utils.KeyDim]) + hotPeerSummary.WithLabelValues(ty, engine).Set(stddev.Loads[utils.KeyDim]) ty = "stddev-query-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(stddevLoads[utils.QueryDim]) + hotPeerSummary.WithLabelValues(ty, engine).Set(stddev.Loads[utils.QueryDim]) ty = "stddev-cpu-rate-" + rwTy.String() + "-" + kind.String() - hotPeerSummary.WithLabelValues(ty, engine).Set(stddevLoads[utils.CPUDim]) - } - expect := StoreLoad{ - Loads: expectLoads, - HotPeerCount: expectCount, - HistoryLoads: expectHistoryLoads, - } - stddev := StoreLoad{ - Loads: stddevLoads, - HotPeerCount: expectCount, + hotPeerSummary.WithLabelValues(ty, engine).Set(stddev.Loads[utils.CPUDim]) } for _, detail := range loadDetail { detail.LoadPred.Expect = expect diff --git a/pkg/statistics/store_load.go b/pkg/statistics/store_load.go index 037b89e26d..7da8fce67c 100644 --- a/pkg/statistics/store_load.go +++ b/pkg/statistics/store_load.go @@ -80,6 +80,68 @@ func (li *StoreLoadDetail) IsUniform(dim int, threshold float64) bool { return li.LoadPred.Stddev.Loads[dim] < threshold } +// StoreLoadSummary incrementally summarizes a store population. +type StoreLoadSummary struct { + count int + loadSum Loads + historySum HistoryLoads + hotPeerCount float64 +} + +// Add adds a store load to the summary. +func (s *StoreLoadSummary) Add(load *StoreLoad) { + s.count++ + for dim, value := range load.Loads { + s.loadSum[dim] += value + } + s.hotPeerCount += load.HotPeerCount + for dim, loads := range load.HistoryLoads { + if len(s.historySum[dim]) < len(loads) { + grown := make([]float64, len(loads)) + copy(grown, s.historySum[dim]) + s.historySum[dim] = grown + } + for i, value := range loads { + s.historySum[dim][i] += value + } + } +} + +// Result returns the expectation and normalized standard deviation. +func (s *StoreLoadSummary) Result(details []*StoreLoadDetail) (expect, stddev StoreLoad) { + if s.count == 0 { + return + } + count := float64(s.count) + for dim, load := range s.loadSum { + expect.Loads[dim] = load / count + } + expect.HotPeerCount = s.hotPeerCount / count + stddev.HotPeerCount = expect.HotPeerCount + for dim, loads := range s.historySum { + expect.HistoryLoads[dim] = make([]float64, len(loads)) + for i, load := range loads { + expect.HistoryLoads[dim][i] = load / count + } + } + if expect.HotPeerCount == 0 { + return + } + for _, detail := range details { + for dim, load := range detail.LoadPred.Current.Loads { + stddev.Loads[dim] += math.Pow(load-expect.Loads[dim], 2) //nolint:staticcheck + } + } + for dim, variance := range stddev.Loads { + if expect.Loads[dim] == 0 { + stddev.Loads[dim] = 0 + continue + } + stddev.Loads[dim] = math.Sqrt(variance/count) / expect.Loads[dim] + } + return +} + func toHotPeerStatShow(p *HotPeerStat) HotPeerStatShow { byteRate := p.GetLoad(utils.ByteDim) keyRate := p.GetLoad(utils.KeyDim) From b4b6c13adc8f36e2daa43436fb98c0e5cd9af8b0 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Thu, 30 Jul 2026 10:45:20 +0800 Subject: [PATCH 6/9] schedule: isolate placement-aware hot scheduling to rank v2 Keep rank v1 on the existing global expectation path. Cache multi-rule placement scopes and return revert-region fits explicitly. Add coverage for overlapping rules, v1 isolation, engine constraints, and multi-rule cache reuse. Signed-off-by: Ryan Leung --- pkg/schedule/placement/config.go | 4 +- pkg/schedule/placement/rule_manager_test.go | 10 ++ .../schedulers/hot_region_rank_v2_test.go | 83 +++++++++++ pkg/schedule/schedulers/hot_region_solver.go | 138 +++++++++++------- .../schedulers/hot_region_solver_test.go | 53 ++++++- pkg/schedule/schedulers/hot_region_test.go | 43 +++++- pkg/statistics/store_load.go | 3 +- 7 files changed, 268 insertions(+), 66 deletions(-) diff --git a/pkg/schedule/placement/config.go b/pkg/schedule/placement/config.go index 9b552d2357..c85da81f44 100644 --- a/pkg/schedule/placement/config.go +++ b/pkg/schedule/placement/config.go @@ -26,7 +26,7 @@ import ( type ruleConfig struct { rules map[[2]string]*Rule // {group, id} => Rule groups map[string]*RuleGroup // id => RuleGroup - mayRestrictStoreLoad [2]bool + mayRestrictStoreLoad [2]bool // TiKV at index 0, TiFlash at index 1. } func newRuleConfig() *ruleConfig { @@ -62,6 +62,8 @@ func (c *ruleConfig) adjust() { } } +// ruleMayRestrictEngine treats an empty value as a store without an engine label. +// It reports restriction for non-engine constraints or a partial engine match. func ruleMayRestrictEngine(rule *Rule, values ...string) bool { hasOtherConstraint := false for i := range rule.LabelConstraints { diff --git a/pkg/schedule/placement/rule_manager_test.go b/pkg/schedule/placement/rule_manager_test.go index 6f9412fcc1..474712e576 100644 --- a/pkg/schedule/placement/rule_manager_test.go +++ b/pkg/schedule/placement/rule_manager_test.go @@ -98,6 +98,9 @@ func TestMayRestrictStoreLoad(t *testing.T) { setConstraints(LabelConstraint{Key: core.EngineKey, Op: In, Values: []string{core.EngineTiFlash}}) re.False(manager.MayRestrictStoreLoad(true)) re.False(manager.MayRestrictStoreLoad(false)) + setConstraints(LabelConstraint{Key: core.EngineKey, Op: NotIn, Values: []string{core.EngineTiFlash}}) + re.False(manager.MayRestrictStoreLoad(true)) + re.False(manager.MayRestrictStoreLoad(false)) setConstraints( LabelConstraint{Key: core.EngineKey, Op: In, Values: []string{core.EngineTiFlash}}, @@ -109,6 +112,13 @@ func TestMayRestrictStoreLoad(t *testing.T) { setConstraints() re.False(manager.MayRestrictStoreLoad(true)) re.False(manager.MayRestrictStoreLoad(false)) + + re.NoError(manager.SetRule(&Rule{ + GroupID: "group", ID: "restricted", Role: Voter, Count: 1, + LabelConstraints: []LabelConstraint{{Key: "zone", Op: In, Values: []string{"z1"}}}, + })) + re.True(manager.MayRestrictStoreLoad(true)) + re.False(manager.MayRestrictStoreLoad(false)) } func TestAdjustRule(t *testing.T) { diff --git a/pkg/schedule/schedulers/hot_region_rank_v2_test.go b/pkg/schedule/schedulers/hot_region_rank_v2_test.go index d4cbfc092a..0de4eb1bc2 100644 --- a/pkg/schedule/schedulers/hot_region_rank_v2_test.go +++ b/pkg/schedule/schedulers/hot_region_rank_v2_test.go @@ -20,8 +20,10 @@ import ( "github.com/docker/go-units" "github.com/stretchr/testify/require" + "github.com/tikv/pd/pkg/core" "github.com/tikv/pd/pkg/mock/mockcluster" "github.com/tikv/pd/pkg/schedule/operator" + "github.com/tikv/pd/pkg/schedule/placement" "github.com/tikv/pd/pkg/statistics/utils" "github.com/tikv/pd/pkg/storage" "github.com/tikv/pd/pkg/utils/operatorutil" @@ -141,6 +143,87 @@ func TestHotWriteRegionScheduleWithRevertRegionsDimFirst(t *testing.T) { clearPendingInfluence(hb) } +func TestHotWriteRegionScheduleWithRevertRegionsAndPlacementRulesV2(t *testing.T) { + for _, testCase := range []struct { + name string + revertTargetMatchesB bool + expectedOperatorCount int + }{ + {name: "valid revert", revertTargetMatchesB: true, expectedOperatorCount: 2}, + {name: "invalid revert", expectedOperatorCount: 1}, + } { + t.Run(testCase.name, func(t *testing.T) { + re := require.New(t) + cancel, _, tc, oc := prepareSchedulersTest() + defer cancel() + tc.SetEnablePlacementRules(true) + sche, err := CreateScheduler(writeType, oc, storage.NewStorageWithMemoryBackend(), nil, nil) + re.NoError(err) + hb := sche.(*hotScheduler) + hb.types = []resourceType{writePeer} + hb.conf.setDstToleranceRatio(0.0) + hb.conf.setSrcToleranceRatio(0.0) + hb.conf.setRankFormulaVersion("v2") + hb.conf.setHistorySampleDuration(0) + hb.conf.WritePeerPriorities = []string{utils.BytePriority, utils.KeyPriority} + tc.SetClusterVersion(versioninfo.MinSupportedVersion(versioninfo.Version4_0)) + + for id, labels := range map[uint64]map[string]string{ + 1: {"b": "yes"}, + 2: {"a": "yes"}, + 3: {"b": "yes"}, + 4: {"a": "yes"}, + 5: {"a": "yes", "b": "yes"}, + } { + if id == 2 && testCase.revertTargetMatchesB { + labels["b"] = "yes" + } + tc.AddLabelsStore(id, 20, labels) + } + tc.SetStoreBusy(4, true) + re.NoError(tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, ID: placement.DefaultRuleID, + Role: placement.Voter, Count: 1, + LabelConstraints: []placement.LabelConstraint{{Key: "a", Op: placement.In, Values: []string{"yes"}}}, + })) + re.NoError(tc.SetRule(&placement.Rule{ + GroupID: placement.DefaultGroupID, ID: "voter-b", + Role: placement.Voter, Count: 2, + LabelConstraints: []placement.LabelConstraint{{Key: "b", Op: placement.In, Values: []string{"yes"}}}, + })) + + tc.UpdateStorageWrittenStats(1, 15*units.MiB*utils.StoreHeartBeatReportInterval, 15*units.MiB*utils.StoreHeartBeatReportInterval) + tc.UpdateStorageWrittenStats(2, 20*units.MiB*utils.StoreHeartBeatReportInterval, 14*units.MiB*utils.StoreHeartBeatReportInterval) + tc.UpdateStorageWrittenStats(3, 15*units.MiB*utils.StoreHeartBeatReportInterval, 15*units.MiB*utils.StoreHeartBeatReportInterval) + tc.UpdateStorageWrittenStats(4, 15*units.MiB*utils.StoreHeartBeatReportInterval, 15*units.MiB*utils.StoreHeartBeatReportInterval) + tc.UpdateStorageWrittenStats(5, 10*units.MiB*utils.StoreHeartBeatReportInterval, 16*units.MiB*utils.StoreHeartBeatReportInterval) + addRegionInfo(tc, utils.Write, []testRegionInfo{ + {6, []uint64{3, 2, 1}, 3 * units.MiB, 1.8 * units.MiB, 0}, + {7, []uint64{1, 4, 5}, 0.1 * units.MiB, 2 * units.MiB, 0}, + }) + mainRegion := tc.GetRegion(6) + sourcePeers := append(mainRegion.GetPeers()[:0:0], mainRegion.GetStorePeer(2)) + for _, item := range tc.ExpiredWriteItems(mainRegion.Clone(core.SetPeers(sourcePeers))) { + tc.Update(item, utils.Write) + } + + ops, _ := hb.Schedule(tc, false) + re.Len(ops, 1) + re.True(hb.searchRevertRegions[writePeer]) + clearPendingInfluence(hb) + + ops, _ = hb.Schedule(tc, false) + re.Len(ops, testCase.expectedOperatorCount) + operatorutil.CheckTransferPeer(re, ops[0], operator.OpHotRegion, 2, 5) + if testCase.revertTargetMatchesB { + operatorutil.CheckTransferPeer(re, ops[1], operator.OpHotRegion, 5, 2) + } + re.True(hb.searchRevertRegions[writePeer]) + clearPendingInfluence(hb) + }) + } +} + func TestHotWriteRegionScheduleWithRevertRegionsDimFirstOnly(t *testing.T) { // This is a test that searchRevertRegions finds a solution of rank 2. re := require.New(t) diff --git a/pkg/schedule/schedulers/hot_region_solver.go b/pkg/schedule/schedulers/hot_region_solver.go index 6cc291857f..c3e841abc8 100644 --- a/pkg/schedule/schedulers/hot_region_solver.go +++ b/pkg/schedule/schedulers/hot_region_solver.go @@ -18,6 +18,7 @@ import ( "math" "sort" "strconv" + "strings" "time" "github.com/prometheus/client_golang/prometheus" @@ -82,8 +83,10 @@ type balanceSolver struct { } type placementScopeKey struct { - rule *placement.Rule - isTiKV bool + rule *placement.Rule + ruleScope string + version uint64 + isTiKV bool } type placementLoadScope struct { @@ -218,10 +221,11 @@ func (bs *balanceSolver) solve() []*operator.Operator { snapshotFilter := filter.NewSnapshotSendFilter(bs.GetStores(), constant.Medium) affinityFilter := filter.NewAffinityFilter(bs.SchedulerCluster) splitThresholds := bs.sche.conf.getSplitThresholds() - placementEnabled := bs.GetSchedulerConfig().IsPlacementRulesEnabled() + // Placement-scoped ranking is a rank-v2 feature; rank v1 keeps its existing global behavior. + placementV2Enabled := bs.GetSchedulerConfig().IsPlacementRulesEnabled() && bs.sche.conf.getRankFormulaVersion() == "v2" var placementChecked, placementCanRestrict [2]bool mayUsePlacementScope := func(store *statistics.StoreLoadDetail) bool { - if !placementEnabled { + if !placementV2Enabled { return false } engine := 0 @@ -250,9 +254,11 @@ func (bs *balanceSolver) solve() []*operator.Operator { } checkedRegion := false for _, mainPeerStat := range bs.filteredHotPeers[srcStoreID] { - if bs.cur.region = bs.getRegion(mainPeerStat, srcStoreID); bs.cur.region == nil { + region, fit := bs.getRegion(mainPeerStat, srcStoreID) + if region == nil { continue } + bs.cur.region, bs.curFit = region, fit checkedRegion = true bs.curScope = nil if usePlacementScope { @@ -296,14 +302,11 @@ func (bs *balanceSolver) solve() []*operator.Operator { if bs.needSearchRevertRegions() { hotSchedulerSearchRevertRegionsCounter.Inc() dstStoreID := dstStore.GetID() - mainFit := bs.curFit for _, revertPeerStat := range bs.filteredHotPeers[dstStoreID] { - revertRegion := bs.getRegion(revertPeerStat, dstStoreID) - revertFit := bs.curFit - bs.curFit = mainFit + revertRegion, revertFit := bs.getRegion(revertPeerStat, dstStoreID) if revertRegion == nil || revertRegion.GetID() == bs.cur.region.GetID() || !allowRevertRegion(revertRegion, srcStoreID) || - placementEnabled && !bs.isRevertRegionValid(revertRegion, revertFit) { + placementV2Enabled && !bs.isRevertRegionValid(revertRegion, revertFit) { continue } bs.revertScope = nil @@ -319,7 +322,6 @@ func (bs *balanceSolver) solve() []*operator.Operator { bs.calcProgressiveRank() tryUpdateBestSolution() } - bs.curFit = mainFit bs.cur.revertPeerStat = nil bs.cur.revertRegion = nil bs.revertScope = nil @@ -599,39 +601,41 @@ func sortHotPeers[T any](firstSort, secondSort []*T, maxPeerNum int) map[*T]stru return union } -// isRegionAvailable checks whether the given region is not available to schedule. -func (bs *balanceSolver) isRegionAvailable(region *core.RegionInfo) bool { +// isRegionAvailable checks whether the given region is available to schedule +// and returns its placement fit when placement rules are enabled. +func (bs *balanceSolver) isRegionAvailable(region *core.RegionInfo) (*placement.RegionFit, bool) { if region == nil { hotSchedulerNoRegionCounter.Inc() - return false + return nil, false } if !filter.IsRegionHealthyAllowPending(region) { hotSchedulerUnhealthyReplicaCounter.Inc() - return false + return nil, false } + var fit *placement.RegionFit replicated := false if bs.GetSchedulerConfig().IsPlacementRulesEnabled() { - bs.curFit = bs.GetRuleManager().FitRegion(bs.GetBasicCluster(), region) - replicated = bs.curFit.IsSatisfied() + fit = bs.GetRuleManager().FitRegion(bs.GetBasicCluster(), region) + replicated = fit.IsSatisfied() } else { replicated = filter.IsRegionReplicated(bs.SchedulerCluster, region) } if !replicated { log.Debug("region has abnormal replica count", zap.String("scheduler", bs.sche.GetName()), zap.Uint64("region-id", region.GetID())) hotSchedulerAbnormalReplicaCounter.Inc() - return false + return nil, false } - return true + return fit, true } -func (bs *balanceSolver) getRegion(peerStat *statistics.HotPeerStat, storeID uint64) *core.RegionInfo { - bs.curFit = nil +func (bs *balanceSolver) getRegion(peerStat *statistics.HotPeerStat, storeID uint64) (*core.RegionInfo, *placement.RegionFit) { region := bs.GetRegion(peerStat.ID()) - if !bs.isRegionAvailable(region) { - return nil + fit, available := bs.isRegionAvailable(region) + if !available { + return nil, nil } switch bs.opTy { @@ -641,20 +645,20 @@ func (bs *balanceSolver) getRegion(peerStat *statistics.HotPeerStat, storeID uin log.Debug("region does not have a peer on source store, maybe stat out of date", zap.Uint64("region-id", peerStat.ID()), zap.Uint64("leader-store-id", storeID)) - return nil + return nil, nil } case transferLeader: if region.GetLeader().GetStoreId() != storeID { log.Debug("region leader is not on source store, maybe stat out of date", zap.Uint64("region-id", peerStat.ID()), zap.Uint64("leader-store-id", storeID)) - return nil + return nil, nil } default: - return nil + return nil, nil } - return region + return region, fit } // filterDstStores select the candidate store by filters @@ -903,45 +907,75 @@ func (bs *balanceSolver) mayUsePlacementScope(isTiKV bool) bool { // getPlacementLoadScope calculates local safeguards only when the placement // population is narrower than the existing engine-wide population. func (bs *balanceSolver) getPlacementLoadScope(rules []*placement.Rule, isTiKV bool) *placementLoadScope { - var key placementScopeKey - cacheable := len(rules) == 1 - if cacheable { - key = placementScopeKey{rule: rules[0], isTiKV: isTiKV} - if scope, ok := bs.placementScopeCache[key]; ok { + if len(rules) == 0 { + return nil + } + + var ruleKey placementScopeKey + if len(rules) == 1 { + ruleKey = placementScopeKey{rule: rules[0], version: rules[0].Version, isTiKV: isTiKV} + if scope, ok := bs.placementScopeCache[ruleKey]; ok { return scope } } - allCount := 0 - var summary statistics.StoreLoadSummary - details := make([]*statistics.StoreLoadDetail, 0, len(bs.stLoadDetail)) - for _, detail := range bs.stLoadDetail { - if detail.IsTiKV() != isTiKV { - continue - } - allCount++ - if !slice.AnyOf(rules, func(i int) bool { - return placement.MatchLabelConstraints(detail.StoreInfo, rules[i].LabelConstraints) - }) { - continue + var builder strings.Builder + builder.WriteString(strconv.Itoa(len(rules))) + builder.WriteByte(':') + for _, rule := range rules { + builder.WriteString(strconv.Itoa(len(rule.LabelConstraints))) + builder.WriteByte(':') + for _, constraint := range rule.LabelConstraints { + appendPlacementScopeKeyPart(&builder, constraint.Key) + appendPlacementScopeKeyPart(&builder, string(constraint.Op)) + builder.WriteString(strconv.Itoa(len(constraint.Values))) + builder.WriteByte(':') + for _, value := range constraint.Values { + appendPlacementScopeKeyPart(&builder, value) + } } - details = append(details, detail) - summary.Add(&detail.LoadPred.Current) } - var scope *placementLoadScope - if len(details) < allCount { - expect, stddev := summary.Result(details) - scope = &placementLoadScope{expect: expect, stddev: stddev} - } - if cacheable { + key := placementScopeKey{ruleScope: builder.String(), isTiKV: isTiKV} + scope, ok := bs.placementScopeCache[key] + + if !ok { + allCount := 0 + var summary statistics.StoreLoadSummary + var details []*statistics.StoreLoadDetail + for _, detail := range bs.stLoadDetail { + if detail.IsTiKV() != isTiKV { + continue + } + allCount++ + if !slice.AnyOf(rules, func(i int) bool { + return placement.MatchLabelConstraints(detail.StoreInfo, rules[i].LabelConstraints) + }) { + continue + } + details = append(details, detail) + summary.Add(&detail.LoadPred.Current) + } + if len(details) < allCount { + expect, stddev := summary.Result(details) + scope = &placementLoadScope{expect: expect, stddev: stddev} + } if bs.placementScopeCache == nil { bs.placementScopeCache = make(map[placementScopeKey]*placementLoadScope) } bs.placementScopeCache[key] = scope } + if len(rules) == 1 { + bs.placementScopeCache[ruleKey] = scope + } return scope } +func appendPlacementScopeKeyPart(builder *strings.Builder, value string) { + builder.WriteString(strconv.Itoa(len(value))) + builder.WriteByte(':') + builder.WriteString(value) +} + func (bs *balanceSolver) isUniformFirstPriority(store *statistics.StoreLoadDetail) bool { // first priority should be more uniform than second priority return bs.isUniform(store, bs.firstPriority, stddevThreshold*0.5) diff --git a/pkg/schedule/schedulers/hot_region_solver_test.go b/pkg/schedule/schedulers/hot_region_solver_test.go index 761a976089..4897f270ba 100644 --- a/pkg/schedule/schedulers/hot_region_solver_test.go +++ b/pkg/schedule/schedulers/hot_region_solver_test.go @@ -661,18 +661,22 @@ func TestExpect(t *testing.T) { func TestPlacementLoadScopePreservesExpectationGuards(t *testing.T) { re := require.New(t) details := make(map[uint64]*statistics.StoreLoadDetail) - for id, load := range map[uint64]float64{1: 10.5, 2: 9.5, 3: 10, 4: 10, 5: 100, 6: 100} { + for id, load := range map[uint64]float64{1: 10.5, 2: 9.5, 3: 10, 4: 10, 5: 100, 6: 100, 7: 200, 8: 300} { pool := "other" - if id <= 4 { + if id <= 4 || id == 7 { pool = "target" } + labels := map[string]string{"pool": pool} + if id >= 7 { + labels[core.EngineKey] = core.EngineTiFlash + } current := statistics.StoreLoad{ Loads: statistics.Loads{load, 10}, HotPeerCount: 1, HistoryLoads: statistics.HistoryLoads{{10}, {10}}, } details[id] = &statistics.StoreLoadDetail{ - StoreSummaryInfo: &statistics.StoreSummaryInfo{StoreInfo: core.NewStoreInfoWithLabel(id, map[string]string{"pool": pool})}, + StoreSummaryInfo: &statistics.StoreSummaryInfo{StoreInfo: core.NewStoreInfoWithLabel(id, labels)}, LoadPred: current.ToLoadPred(utils.Write, nil), } } @@ -689,13 +693,54 @@ func TestPlacementLoadScopePreservesExpectationGuards(t *testing.T) { re.Equal(float64(10), scope.expect.Loads[utils.ByteDim]) re.Equal([]float64{10}, scope.expect.HistoryLoads[utils.ByteDim]) re.InDelta(math.Sqrt(0.125)/10, scope.stddev.Loads[utils.ByteDim], 1e-9) + crossEngineRule := &placement.Rule{LabelConstraints: append(rule.LabelConstraints, + placement.LabelConstraint{Key: core.EngineKey, Op: placement.NotIn, Values: []string{"other"}})} + re.Equal(float64(10), bs.getPlacementLoadScope([]*placement.Rule{crossEngineRule}, true).expect.Loads[utils.ByteDim]) + re.Equal(float64(200), bs.getPlacementLoadScope([]*placement.Rule{crossEngineRule}, false).expect.Loads[utils.ByteDim]) re.False(bs.checkSrcByPriorityAndTolerance(details[1].LoadPred.Min(), &scope.expect, 1.05)) re.False(bs.checkSrcHistoryLoadsByPriorityAndTolerance(&details[1].LoadPred.Current, &scope.expect, 1.05)) bs.cur = &solution{} bs.curScope = scope re.True(bs.isUniformFirstPriority(details[1])) - re.Nil(bs.getPlacementLoadScope([]*placement.Rule{{}}, true)) + unconstrainedRule := &placement.Rule{} + re.Nil(bs.getPlacementLoadScope([]*placement.Rule{unconstrainedRule}, true)) + cacheSize := len(bs.placementScopeCache) + re.Nil(bs.getPlacementLoadScope([]*placement.Rule{unconstrainedRule}, true)) + re.Len(bs.placementScopeCache, cacheSize) + + equivalentRule := &placement.Rule{GroupID: "other", ID: "same-scope", LabelConstraints: []placement.LabelConstraint{ + {Key: "pool", Op: placement.In, Values: []string{"target"}}, + }} + re.Same(scope, bs.getPlacementLoadScope([]*placement.Rule{equivalentRule}, true)) + + otherRule := &placement.Rule{GroupID: "pd", ID: "other", LabelConstraints: []placement.LabelConstraint{ + {Key: "pool", Op: placement.In, Values: []string{"other"}}, + }} + updatedRule := &placement.Rule{GroupID: "pd", ID: "updated", Version: 1, LabelConstraints: rule.LabelConstraints} + re.Same(scope, bs.getPlacementLoadScope([]*placement.Rule{updatedRule}, true)) + updatedRule.LabelConstraints = otherRule.LabelConstraints + updatedRule.Version++ + re.Equal(float64(100), bs.getPlacementLoadScope([]*placement.Rule{updatedRule}, true).expect.Loads[utils.ByteDim]) + + rules := []*placement.Rule{rule, otherRule} + re.Nil(bs.getPlacementLoadScope(rules, true)) + cacheSize = len(bs.placementScopeCache) + re.Nil(bs.getPlacementLoadScope(rules, true)) + re.Len(bs.placementScopeCache, cacheSize) + + targetRules := []*placement.Rule{ + {GroupID: "pd", ID: "a", LabelConstraints: rule.LabelConstraints}, + {GroupID: "pd", ID: "b", LabelConstraints: rule.LabelConstraints}, + } + otherRules := []*placement.Rule{ + {GroupID: "pd", ID: "a", LabelConstraints: otherRule.LabelConstraints}, + {GroupID: "pd", ID: "b", LabelConstraints: otherRule.LabelConstraints}, + } + targetScope := bs.getPlacementLoadScope(targetRules, true) + otherScope := bs.getPlacementLoadScope(otherRules, true) + re.Equal(float64(10), targetScope.expect.Loads[utils.ByteDim]) + re.Equal(float64(100), otherScope.expect.Loads[utils.ByteDim]) } func TestRevertRegionPlacementSafeguard(t *testing.T) { diff --git a/pkg/schedule/schedulers/hot_region_test.go b/pkg/schedule/schedulers/hot_region_test.go index c5fe5eb32f..dd850a5a5b 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -466,6 +466,8 @@ func TestHotWriteRegionScheduleWithPlacementScope(t *testing.T) { storeLoads map[uint64]float64 peerLoad float64 exclusive bool + rankV1 bool + schedule bool }{ { name: "source below global expectation", @@ -478,6 +480,7 @@ func TestHotWriteRegionScheduleWithPlacementScope(t *testing.T) { 6: 100, }, peerLoad: 5, + schedule: true, }, { name: "target above global expectation", @@ -490,6 +493,7 @@ func TestHotWriteRegionScheduleWithPlacementScope(t *testing.T) { 6: 0, }, peerLoad: 20, + schedule: true, }, { name: "implicit exclusive labels", @@ -503,6 +507,20 @@ func TestHotWriteRegionScheduleWithPlacementScope(t *testing.T) { }, peerLoad: 5, exclusive: true, + schedule: true, + }, + { + name: "rank v1 keeps global expectation", + storeLoads: map[uint64]float64{ + 1: 20, + 2: 0, + 3: 0, + 4: 0, + 5: 100, + 6: 100, + }, + peerLoad: 5, + rankV1: true, }, } @@ -510,16 +528,21 @@ func TestHotWriteRegionScheduleWithPlacementScope(t *testing.T) { t.Run(testCase.name, func(t *testing.T) { re := require.New(t) tc, hb := preparePlacementHotScheduler(t, writeType, writePeer) + if testCase.rankV1 { + hb.conf.setRankFormulaVersion("v1") + } for id := uint64(1); id <= 6; id++ { - labels := map[string]string{"pool": "other"} - if id <= 4 { - labels["pool"] = "target" - } else if testCase.exclusive { - labels = map[string]string{"$group": "other"} - } - if testCase.exclusive && id <= 4 { - labels = nil + var labels map[string]string + if testCase.exclusive { + if id > 4 { + labels = map[string]string{"$group": "other"} + } + } else { + labels = map[string]string{"pool": "other"} + if id <= 4 { + labels["pool"] = "target" + } } tc.AddLabelsStore(id, 1, labels) tc.UpdateStorageWrittenBytes(id, uint64(testCase.storeLoads[id]*units.MiB*utils.StoreHeartBeatReportInterval)) @@ -533,6 +556,10 @@ func TestHotWriteRegionScheduleWithPlacementScope(t *testing.T) { }) ops, _ := hb.Schedule(tc, false) + if !testCase.schedule { + re.Empty(ops) + return + } re.Len(ops, 1) operatorutil.CheckTransferPeerWithLeaderTransfer(re, ops[0], operator.OpHotRegion, 1, 4) }) diff --git a/pkg/statistics/store_load.go b/pkg/statistics/store_load.go index 7da8fce67c..d61d142c61 100644 --- a/pkg/statistics/store_load.go +++ b/pkg/statistics/store_load.go @@ -107,7 +107,8 @@ func (s *StoreLoadSummary) Add(load *StoreLoad) { } } -// Result returns the expectation and normalized standard deviation. +// Result returns the expectation and normalized standard deviation. Details +// must be the same store population whose current loads were passed to Add. func (s *StoreLoadSummary) Result(details []*StoreLoadDetail) (expect, stddev StoreLoad) { if s.count == 0 { return From 2eabcf42774540e1852a87d483108c26d5ef1c98 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Thu, 30 Jul 2026 15:41:31 +0800 Subject: [PATCH 7/9] schedule: avoid placement hot-path overhead Signed-off-by: Ryan Leung --- pkg/schedule/schedulers/hot_region_solver.go | 97 +++++++++++--------- pkg/schedule/schedulers/hot_region_test.go | 6 +- 2 files changed, 57 insertions(+), 46 deletions(-) diff --git a/pkg/schedule/schedulers/hot_region_solver.go b/pkg/schedule/schedulers/hot_region_solver.go index c3e841abc8..66d401860c 100644 --- a/pkg/schedule/schedulers/hot_region_solver.go +++ b/pkg/schedule/schedulers/hot_region_solver.go @@ -57,9 +57,11 @@ type balanceSolver struct { curFit *placement.RegionFit // curScope and revertScope contain placement-scoped load statistics. A nil // scope uses the existing engine-wide statistics. - curScope *placementLoadScope - revertScope *placementLoadScope - placementScopeCache map[placementScopeKey]*placementLoadScope + curScope *placementLoadScope + revertScope *placementLoadScope + placementScopeCache map[placementScopeKey]*placementLoadScope + placementV2Enabled bool + placementCanRestrict [2]bool best *solution ops []*operator.Operator @@ -101,7 +103,8 @@ func (bs *balanceSolver) init() { bs.minHotDegree = bs.GetSchedulerConfig().GetHotRegionCacheHitsThreshold() bs.firstPriority, bs.secondPriority = prioritiesToDim(bs.getPriorities()) bs.greatDecRatio, bs.minorDecRatio = bs.sche.conf.getGreatDecRatio(), bs.sche.conf.getMinorDecRatio() - switch bs.sche.conf.getRankFormulaVersion() { + rankFormulaVersion := bs.sche.conf.getRankFormulaVersion() + switch rankFormulaVersion { case "v1": bs.rank = initRankV1(bs) default: @@ -110,6 +113,11 @@ func (bs *balanceSolver) init() { // Init store load detail according to the type. bs.stLoadDetail = bs.sche.stLoadInfos[bs.resourceTy] + bs.placementV2Enabled = bs.GetSchedulerConfig().IsPlacementRulesEnabled() && rankFormulaVersion == "v2" + if bs.placementV2Enabled { + bs.placementCanRestrict[0] = bs.GetRuleManager().MayRestrictStoreLoad(true) + bs.placementCanRestrict[1] = bs.GetRuleManager().MayRestrictStoreLoad(false) + } bs.maxSrc = &statistics.StoreLoad{} bs.minDst = &statistics.StoreLoad{HotPeerCount: math.MaxFloat64} @@ -121,6 +129,9 @@ func (bs *balanceSolver) init() { bs.filteredHotPeers = make(map[uint64][]*statistics.HotPeerStat) bs.nthHotPeer = make(map[uint64][]*statistics.HotPeerStat) for _, detail := range bs.stLoadDetail { + if bs.placementV2Enabled { + bs.recordPlacementRestriction(detail) + } bs.maxSrc = statistics.MaxLoad(bs.maxSrc, detail.LoadPred.Min()) bs.minDst = statistics.MinLoad(bs.minDst, detail.LoadPred.Max()) maxCur = statistics.MaxLoad(maxCur, &detail.LoadPred.Current) @@ -221,36 +232,15 @@ func (bs *balanceSolver) solve() []*operator.Operator { snapshotFilter := filter.NewSnapshotSendFilter(bs.GetStores(), constant.Medium) affinityFilter := filter.NewAffinityFilter(bs.SchedulerCluster) splitThresholds := bs.sche.conf.getSplitThresholds() - // Placement-scoped ranking is a rank-v2 feature; rank v1 keeps its existing global behavior. - placementV2Enabled := bs.GetSchedulerConfig().IsPlacementRulesEnabled() && bs.sche.conf.getRankFormulaVersion() == "v2" - var placementChecked, placementCanRestrict [2]bool - mayUsePlacementScope := func(store *statistics.StoreLoadDetail) bool { - if !placementV2Enabled { - return false - } - engine := 0 - if !store.IsTiKV() { - engine = 1 - } - if !placementChecked[engine] { - placementCanRestrict[engine] = bs.mayUsePlacementScope(store.IsTiKV()) - placementChecked[engine] = true - } - return placementCanRestrict[engine] - } for _, srcStore := range bs.filterSrcStores() { bs.cur.srcStore = srcStore srcStoreID := srcStore.GetID() - usePlacementScope := mayUsePlacementScope(srcStore) - globalSourceFailure := bs.sourceStoreFailure(srcStore, nil) - sourceFailure := globalSourceFailure - sourceResultRecorded := false - if !usePlacementScope { - bs.recordSourceStoreResult(sourceFailureOrSuccess(sourceFailure), srcStoreID) - sourceResultRecorded = true - if sourceFailure != "" { - continue - } + usePlacementScope := bs.mayUsePlacementScope(srcStore.IsTiKV()) + sourceResultRecorded := !usePlacementScope + var globalSourceFailure, sourceFailure string + if usePlacementScope { + globalSourceFailure = bs.sourceStoreFailure(srcStore, nil) + sourceFailure = globalSourceFailure } checkedRegion := false for _, mainPeerStat := range bs.filteredHotPeers[srcStoreID] { @@ -306,11 +296,11 @@ func (bs *balanceSolver) solve() []*operator.Operator { revertRegion, revertFit := bs.getRegion(revertPeerStat, dstStoreID) if revertRegion == nil || revertRegion.GetID() == bs.cur.region.GetID() || !allowRevertRegion(revertRegion, srcStoreID) || - placementV2Enabled && !bs.isRevertRegionValid(revertRegion, revertFit) { + bs.placementV2Enabled && !bs.isRevertRegionValid(revertRegion, revertFit) { continue } bs.revertScope = nil - if mayUsePlacementScope(dstStore) { + if bs.mayUsePlacementScope(dstStore.IsTiKV()) { bs.revertScope = bs.prepareForRegion(revertRegion, revertFit, dstStore) } if bs.revertScope != nil && (bs.sourceStoreFailure(dstStore, bs.revertScope) != "" || @@ -459,6 +449,8 @@ func (bs *balanceSolver) calcMaxZombieDur() time.Duration { } // filterSrcStores selects stores with hot peers that support the resource type. +// The global load safeguards are deferred only when placement may narrow the +// store population for the engine. func (bs *balanceSolver) filterSrcStores() map[uint64]*statistics.StoreLoadDetail { ret := make(map[uint64]*statistics.StoreLoadDetail) confEnableForTiFlash := bs.sche.conf.getEnableForTiFlash() @@ -474,6 +466,13 @@ func (bs *balanceSolver) filterSrcStores() map[uint64]*statistics.StoreLoadDetai if len(detail.HotPeers) == 0 { continue } + if !bs.mayUsePlacementScope(detail.IsTiKV()) { + failure := bs.sourceStoreFailure(detail, nil) + bs.recordSourceStoreResult(sourceFailureOrSuccess(failure), id) + if failure != "" { + continue + } + } ret[id] = detail } return ret @@ -886,22 +885,32 @@ func (bs *balanceSolver) prepareForRegion(region *core.RegionInfo, fit *placemen return bs.getPlacementLoadScope(rules, source.IsTiKV()) } -func (bs *balanceSolver) mayUsePlacementScope(isTiKV bool) bool { - if bs.GetRuleManager().MayRestrictStoreLoad(isTiKV) { - return true +func (bs *balanceSolver) recordPlacementRestriction(detail *statistics.StoreLoadDetail) { + isTiKV := detail.IsTiKV() + engine := 0 + if !isTiKV { + engine = 1 + } + if bs.placementCanRestrict[engine] { + return } var constraints []placement.LabelConstraint if !isTiKV { - constraints = []placement.LabelConstraint{ - {Key: core.EngineKey, Op: placement.In, Values: []string{core.EngineTiFlash}}, - } + constraints = []placement.LabelConstraint{{Key: core.EngineKey, Op: placement.In, Values: []string{core.EngineTiFlash}}} } - for _, detail := range bs.stLoadDetail { - if detail.IsTiKV() == isTiKV && !placement.MatchLabelConstraints(detail.StoreInfo, constraints) { - return true - } + if !placement.MatchLabelConstraints(detail.StoreInfo, constraints) { + bs.placementCanRestrict[engine] = true + } +} + +func (bs *balanceSolver) mayUsePlacementScope(isTiKV bool) bool { + if !bs.placementV2Enabled { + return false + } + if isTiKV { + return bs.placementCanRestrict[0] } - return false + return bs.placementCanRestrict[1] } // getPlacementLoadScope calculates local safeguards only when the placement diff --git a/pkg/schedule/schedulers/hot_region_test.go b/pkg/schedule/schedulers/hot_region_test.go index dd850a5a5b..1764152084 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -441,16 +441,18 @@ func TestMayUsePlacementScopeIgnoresEngineSeparation(t *testing.T) { cancel, _, tc, _ := prepareSchedulersTest() defer cancel() details := make(map[uint64]*statistics.StoreLoadDetail) + bs := &balanceSolver{SchedulerCluster: tc, stLoadDetail: details, placementV2Enabled: true} addStore := func(id uint64, labels map[string]string) { tc.AddLabelsStore(id, 1, labels) - details[id] = &statistics.StoreLoadDetail{ + detail := &statistics.StoreLoadDetail{ StoreSummaryInfo: &statistics.StoreSummaryInfo{StoreInfo: tc.GetStore(id)}, } + details[id] = detail + bs.recordPlacementRestriction(detail) } addStore(1, map[string]string{core.EngineKey: core.EngineTiFlash}) addStore(2, nil) - bs := &balanceSolver{SchedulerCluster: tc, stLoadDetail: details} require.False(t, bs.mayUsePlacementScope(false)) require.False(t, bs.mayUsePlacementScope(true)) From c19e61acd59043c3f431dd39f60ae642036ab9f2 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Thu, 30 Jul 2026 18:03:47 +0800 Subject: [PATCH 8/9] schedule: reuse placement precheck for read solvers Signed-off-by: Ryan Leung --- pkg/schedule/schedulers/hot_region.go | 15 ++++++-- pkg/schedule/schedulers/hot_region_solver.go | 35 +++++++++++++++---- .../schedulers/hot_region_solver_test.go | 33 +++++++++++++++++ 3 files changed, 74 insertions(+), 9 deletions(-) diff --git a/pkg/schedule/schedulers/hot_region.go b/pkg/schedule/schedulers/hot_region.go index 4ee3683461..ce7ad4d674 100644 --- a/pkg/schedule/schedulers/hot_region.go +++ b/pkg/schedule/schedulers/hot_region.go @@ -347,10 +347,21 @@ func (s *hotScheduler) tryAddPendingInfluence(op *operator.Operator, srcStore [] return true } +func newBalanceReadSolvers(s *hotScheduler, cluster sche.SchedulerCluster) (leaderSolver, peerSolver *balanceSolver) { + // The read-peer population contains every TiKV store considered by the + // read-leader solver, so both solvers can share the placement precheck. + peerSolver = newBalanceSolver(s, cluster, utils.Read, movePeer) + placementState := &placementLoadState{ + enabled: peerSolver.placementV2Enabled, + canRestrict: peerSolver.placementCanRestrict, + } + leaderSolver = newBalanceSolverWithPlacementState(s, cluster, utils.Read, transferLeader, placementState) + return leaderSolver, peerSolver +} + func (s *hotScheduler) balanceHotReadRegions(cluster sche.SchedulerCluster) []*operator.Operator { - leaderSolver := newBalanceSolver(s, cluster, utils.Read, transferLeader) + leaderSolver, peerSolver := newBalanceReadSolvers(s, cluster) leaderOps := leaderSolver.solve() - peerSolver := newBalanceSolver(s, cluster, utils.Read, movePeer) peerOps := peerSolver.solve() if len(leaderOps) == 0 && len(peerOps) == 0 { return nil diff --git a/pkg/schedule/schedulers/hot_region_solver.go b/pkg/schedule/schedulers/hot_region_solver.go index 66d401860c..c571030f93 100644 --- a/pkg/schedule/schedulers/hot_region_solver.go +++ b/pkg/schedule/schedulers/hot_region_solver.go @@ -96,7 +96,12 @@ type placementLoadScope struct { stddev statistics.StoreLoad } -func (bs *balanceSolver) init() { +type placementLoadState struct { + enabled bool + canRestrict [2]bool +} + +func (bs *balanceSolver) init(placementState *placementLoadState) { // Load the configuration items of the scheduler. bs.resourceTy = toResourceType(bs.rwTy, bs.opTy) bs.maxPeerNum = bs.sche.conf.getMaxPeerNumber() @@ -113,10 +118,15 @@ func (bs *balanceSolver) init() { // Init store load detail according to the type. bs.stLoadDetail = bs.sche.stLoadInfos[bs.resourceTy] - bs.placementV2Enabled = bs.GetSchedulerConfig().IsPlacementRulesEnabled() && rankFormulaVersion == "v2" - if bs.placementV2Enabled { - bs.placementCanRestrict[0] = bs.GetRuleManager().MayRestrictStoreLoad(true) - bs.placementCanRestrict[1] = bs.GetRuleManager().MayRestrictStoreLoad(false) + if placementState == nil { + bs.placementV2Enabled = bs.GetSchedulerConfig().IsPlacementRulesEnabled() && rankFormulaVersion == "v2" + if bs.placementV2Enabled { + bs.placementCanRestrict[0] = bs.GetRuleManager().MayRestrictStoreLoad(true) + bs.placementCanRestrict[1] = bs.GetRuleManager().MayRestrictStoreLoad(false) + } + } else { + bs.placementV2Enabled = placementState.enabled + bs.placementCanRestrict = placementState.canRestrict } bs.maxSrc = &statistics.StoreLoad{} @@ -128,8 +138,9 @@ func (bs *balanceSolver) init() { bs.filteredHotPeers = make(map[uint64][]*statistics.HotPeerStat) bs.nthHotPeer = make(map[uint64][]*statistics.HotPeerStat) + checkPlacementRestriction := placementState == nil && bs.placementV2Enabled for _, detail := range bs.stLoadDetail { - if bs.placementV2Enabled { + if checkPlacementRestriction { bs.recordPlacementRestriction(detail) } bs.maxSrc = statistics.MaxLoad(bs.maxSrc, detail.LoadPred.Min()) @@ -177,13 +188,23 @@ func (bs *balanceSolver) getPriorities() []string { } func newBalanceSolver(sche *hotScheduler, cluster sche.SchedulerCluster, rwTy utils.RWType, opTy opType) *balanceSolver { + return newBalanceSolverWithPlacementState(sche, cluster, rwTy, opTy, nil) +} + +func newBalanceSolverWithPlacementState( + sche *hotScheduler, + cluster sche.SchedulerCluster, + rwTy utils.RWType, + opTy opType, + placementState *placementLoadState, +) *balanceSolver { bs := &balanceSolver{ SchedulerCluster: cluster, sche: sche, rwTy: rwTy, opTy: opTy, } - bs.init() + bs.init(placementState) return bs } diff --git a/pkg/schedule/schedulers/hot_region_solver_test.go b/pkg/schedule/schedulers/hot_region_solver_test.go index 4897f270ba..35bd9e2eef 100644 --- a/pkg/schedule/schedulers/hot_region_solver_test.go +++ b/pkg/schedule/schedulers/hot_region_solver_test.go @@ -36,6 +36,39 @@ import ( "github.com/tikv/pd/pkg/versioninfo" ) +var benchmarkBalanceSolvers [2]*balanceSolver + +func BenchmarkNewBalanceReadSolvers(b *testing.B) { + cancel, _, tc, oc := prepareSchedulersTest() + b.Cleanup(cancel) + scheduler, err := CreateScheduler(types.BalanceHotRegionScheduler, oc, storage.NewStorageWithMemoryBackend(), + ConfigSliceDecoder(types.BalanceHotRegionScheduler, nil)) + if err != nil { + b.Fatal(err) + } + hot := scheduler.(*hotScheduler) + for id := uint64(1); id <= 1000; id++ { + detail := &statistics.StoreLoadDetail{ + StoreSummaryInfo: &statistics.StoreSummaryInfo{ + StoreInfo: core.NewStoreInfoWithLabel(id, map[string]string{ + "zone": "z1", + "rack": "r1", + "host": "h1", + }), + }, + LoadPred: (statistics.StoreLoad{}).ToLoadPred(utils.Read, nil), + } + hot.stLoadInfos[readLeader][id] = detail + hot.stLoadInfos[readPeer][id] = detail + } + + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + benchmarkBalanceSolvers[0], benchmarkBalanceSolvers[1] = newBalanceReadSolvers(hot, tc) + } +} + func TestSplitBucketsBySize(t *testing.T) { re := require.New(t) cancel, _, tc, oc := prepareSchedulersTest() From c47ec162152e3d010f5b3d1bba2aeaf775e00306 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Fri, 31 Jul 2026 11:11:59 +0800 Subject: [PATCH 9/9] schedule: strengthen placement scope tests Signed-off-by: Ryan Leung --- .../schedulers/hot_region_rank_v2_test.go | 40 +++++++++--- pkg/schedule/schedulers/hot_region_test.go | 62 ++++++++++++------- 2 files changed, 71 insertions(+), 31 deletions(-) diff --git a/pkg/schedule/schedulers/hot_region_rank_v2_test.go b/pkg/schedule/schedulers/hot_region_rank_v2_test.go index 0de4eb1bc2..3ee1313823 100644 --- a/pkg/schedule/schedulers/hot_region_rank_v2_test.go +++ b/pkg/schedule/schedulers/hot_region_rank_v2_test.go @@ -21,9 +21,11 @@ import ( "github.com/stretchr/testify/require" "github.com/tikv/pd/pkg/core" + "github.com/tikv/pd/pkg/core/constant" "github.com/tikv/pd/pkg/mock/mockcluster" "github.com/tikv/pd/pkg/schedule/operator" "github.com/tikv/pd/pkg/schedule/placement" + "github.com/tikv/pd/pkg/statistics" "github.com/tikv/pd/pkg/statistics/utils" "github.com/tikv/pd/pkg/storage" "github.com/tikv/pd/pkg/utils/operatorutil" @@ -147,10 +149,12 @@ func TestHotWriteRegionScheduleWithRevertRegionsAndPlacementRulesV2(t *testing.T for _, testCase := range []struct { name string revertTargetMatchesB bool + historyRejectsRevert bool expectedOperatorCount int }{ {name: "valid revert", revertTargetMatchesB: true, expectedOperatorCount: 2}, {name: "invalid revert", expectedOperatorCount: 1}, + {name: "revert rejected by scoped history", revertTargetMatchesB: true, historyRejectsRevert: true, expectedOperatorCount: 1}, } { t.Run(testCase.name, func(t *testing.T) { re := require.New(t) @@ -161,11 +165,13 @@ func TestHotWriteRegionScheduleWithRevertRegionsAndPlacementRulesV2(t *testing.T re.NoError(err) hb := sche.(*hotScheduler) hb.types = []resourceType{writePeer} - hb.conf.setDstToleranceRatio(0.0) - hb.conf.setSrcToleranceRatio(0.0) hb.conf.setRankFormulaVersion("v2") - hb.conf.setHistorySampleDuration(0) hb.conf.WritePeerPriorities = []string{utils.BytePriority, utils.KeyPriority} + // Retain one complete history sample so both current and historical + // placement-scoped safeguards are active without waiting in the test. + historyInterval := hb.conf.getHistorySampleInterval() + hb.conf.setHistorySampleDuration(historyInterval) + hb.updateHistoryLoadConfig(historyInterval, historyInterval) tc.SetClusterVersion(versioninfo.MinSupportedVersion(versioninfo.Version4_0)) for id, labels := range map[uint64]map[string]string{ @@ -192,11 +198,27 @@ func TestHotWriteRegionScheduleWithRevertRegionsAndPlacementRulesV2(t *testing.T LabelConstraints: []placement.LabelConstraint{{Key: "b", Op: placement.In, Values: []string{"yes"}}}, })) - tc.UpdateStorageWrittenStats(1, 15*units.MiB*utils.StoreHeartBeatReportInterval, 15*units.MiB*utils.StoreHeartBeatReportInterval) - tc.UpdateStorageWrittenStats(2, 20*units.MiB*utils.StoreHeartBeatReportInterval, 14*units.MiB*utils.StoreHeartBeatReportInterval) - tc.UpdateStorageWrittenStats(3, 15*units.MiB*utils.StoreHeartBeatReportInterval, 15*units.MiB*utils.StoreHeartBeatReportInterval) - tc.UpdateStorageWrittenStats(4, 15*units.MiB*utils.StoreHeartBeatReportInterval, 15*units.MiB*utils.StoreHeartBeatReportInterval) - tc.UpdateStorageWrittenStats(5, 10*units.MiB*utils.StoreHeartBeatReportInterval, 16*units.MiB*utils.StoreHeartBeatReportInterval) + storeLoads := map[uint64]statistics.Loads{ + 1: {utils.ByteDim: 15 * units.MiB, utils.KeyDim: 15 * units.MiB}, + 2: {utils.ByteDim: 20 * units.MiB, utils.KeyDim: 14 * units.MiB}, + 3: {utils.ByteDim: 15 * units.MiB, utils.KeyDim: 15 * units.MiB}, + 4: {utils.ByteDim: 15 * units.MiB, utils.KeyDim: 15 * units.MiB}, + 5: {utils.ByteDim: 10 * units.MiB, utils.KeyDim: 16 * units.MiB}, + } + for id, loads := range storeLoads { + tc.UpdateStorageWrittenStats( + id, + uint64(loads[utils.ByteDim]*utils.StoreHeartBeatReportInterval), + uint64(loads[utils.KeyDim]*utils.StoreHeartBeatReportInterval), + ) + historyLoads := loads + if testCase.historyRejectsRevert { + // The main move still passes by byte load, while the revert + // source and target no longer straddle their scoped key history. + historyLoads[utils.KeyDim] = 15 * units.MiB + } + hb.stHistoryLoads.Add(id, utils.Write, constant.RegionKind, historyLoads) + } addRegionInfo(tc, utils.Write, []testRegionInfo{ {6, []uint64{3, 2, 1}, 3 * units.MiB, 1.8 * units.MiB, 0}, {7, []uint64{1, 4, 5}, 0.1 * units.MiB, 2 * units.MiB, 0}, @@ -215,7 +237,7 @@ func TestHotWriteRegionScheduleWithRevertRegionsAndPlacementRulesV2(t *testing.T ops, _ = hb.Schedule(tc, false) re.Len(ops, testCase.expectedOperatorCount) operatorutil.CheckTransferPeer(re, ops[0], operator.OpHotRegion, 2, 5) - if testCase.revertTargetMatchesB { + if testCase.expectedOperatorCount == 2 { operatorutil.CheckTransferPeer(re, ops[1], operator.OpHotRegion, 5, 2) } re.True(hb.searchRevertRegions[writePeer]) diff --git a/pkg/schedule/schedulers/hot_region_test.go b/pkg/schedule/schedulers/hot_region_test.go index 1764152084..b02e00d9c5 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -596,31 +596,49 @@ func TestHotReadLeaderScheduleWithPlacementScope(t *testing.T) { } func TestHotWriteLeaderScheduleWithPlacementScope(t *testing.T) { - re := require.New(t) - tc, hb := preparePlacementHotScheduler(t, writeType, writeLeader) - hb.conf.WriteLeaderPriorities = []string{utils.BytePriority, utils.KeyPriority} + for _, testCase := range []struct { + name string + rankVersion string + schedule bool + }{ + {name: "v2 uses scoped expectation", rankVersion: "v2", schedule: true}, + {name: "v1 keeps global expectation", rankVersion: "v1"}, + } { + t.Run(testCase.name, func(t *testing.T) { + re := require.New(t) + tc, hb := preparePlacementHotScheduler(t, writeType, writeLeader) + hb.conf.setRankFormulaVersion(testCase.rankVersion) + hb.conf.WriteLeaderPriorities = []string{utils.BytePriority, utils.KeyPriority} - for id := uint64(1); id <= 6; id++ { - pool := "other" - if id <= 4 { - pool = "target" - } - tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) - tc.UpdateStorageWrittenBytes(id, 0) - } - setPoolPlacementRule(t, tc) + for id := uint64(1); id <= 6; id++ { + pool := "other" + if id <= 4 { + pool = "target" + } + tc.AddLabelsStore(id, 1, map[string]string{"pool": pool}) + tc.UpdateStorageWrittenBytes(id, 0) + } + setPoolPlacementRule(t, tc) - addRegionInfo(tc, utils.Write, []testRegionInfo{ - {1, []uint64{1, 2, 3}, 2.5 * units.MiB, 0, 0}, - {2, []uint64{1, 2, 3}, 17.5 * units.MiB, 0, 0}, - {3, []uint64{2, 1, 3}, 0, 0, 0}, - {4, []uint64{3, 1, 2}, 0, 0, 0}, - }) + // The target stores are above the global expectation (31/6) but + // below the placement-scoped expectation (31/4). + addRegionInfo(tc, utils.Write, []testRegionInfo{ + {1, []uint64{1, 2, 3}, 2.5 * units.MiB, 0, 0}, + {2, []uint64{1, 2, 3}, 17.5 * units.MiB, 0, 0}, + {3, []uint64{2, 1, 3}, 5.5 * units.MiB, 0, 0}, + {4, []uint64{3, 1, 2}, 5.5 * units.MiB, 0, 0}, + }) - ops, _ := hb.Schedule(tc, false) - re.Len(ops, 1) - re.Equal(uint64(1), ops[0].RegionID()) - operatorutil.CheckTransferLeaderFrom(re, ops[0], operator.OpHotRegion, 1) + ops, _ := hb.Schedule(tc, false) + if !testCase.schedule { + re.Empty(ops) + return + } + re.Len(ops, 1) + re.Equal(uint64(1), ops[0].RegionID()) + operatorutil.CheckTransferLeaderFrom(re, ops[0], operator.OpHotRegion, 1) + }) + } } func TestHotWriteLeaderScheduleAcrossSameRoleRules(t *testing.T) {