diff --git a/pkg/schedule/filter/filters.go b/pkg/schedule/filter/filters.go index 550cadfee88..de077989317 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 ba0f319d499..18acd4e27e1 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/placement/config.go b/pkg/schedule/placement/config.go index 53cb0636536..c85da81f444 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 // TiKV at index 0, TiFlash at index 1. } 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,41 @@ 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 { + 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 6cc2d4ed457..77b8beb9d21 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 49567b8c033..1e38d848373 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 3263f6e593d..474712e576a 100644 --- a/pkg/schedule/placement/rule_manager_test.go +++ b/pkg/schedule/placement/rule_manager_test.go @@ -78,6 +78,49 @@ 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: 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}}, + 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)) + + 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) { re := require.New(t) _, manager := newTestManager(t, false) diff --git a/pkg/schedule/schedulers/hot_region.go b/pkg/schedule/schedulers/hot_region.go index 4ee36834614..ce7ad4d674f 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_rank_v2_test.go b/pkg/schedule/schedulers/hot_region_rank_v2_test.go index d4cbfc092a0..3ee13138236 100644 --- a/pkg/schedule/schedulers/hot_region_rank_v2_test.go +++ b/pkg/schedule/schedulers/hot_region_rank_v2_test.go @@ -20,8 +20,12 @@ import ( "github.com/docker/go-units" "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" @@ -141,6 +145,107 @@ func TestHotWriteRegionScheduleWithRevertRegionsDimFirst(t *testing.T) { clearPendingInfluence(hb) } +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) + 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.setRankFormulaVersion("v2") + 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{ + 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"}}}, + })) + + 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}, + }) + 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.expectedOperatorCount == 2 { + 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 c870ecf6be9..c571030f933 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" @@ -33,6 +34,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 +53,15 @@ type balanceSolver struct { resourceTy resourceType cur *solution + // curFit is the placement-rule fit for the current region. + 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 + placementV2Enabled bool + placementCanRestrict [2]bool best *solution ops []*operator.Operator @@ -73,14 +84,32 @@ type balanceSolver struct { rank } -func (bs *balanceSolver) init() { +type placementScopeKey struct { + rule *placement.Rule + ruleScope string + version uint64 + isTiKV bool +} + +type placementLoadScope struct { + expect statistics.StoreLoad + stddev statistics.StoreLoad +} + +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() 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: @@ -89,6 +118,16 @@ func (bs *balanceSolver) init() { // Init store load detail according to the type. bs.stLoadDetail = bs.sche.stLoadInfos[bs.resourceTy] + 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{} bs.minDst = &statistics.StoreLoad{HotPeerCount: math.MaxFloat64} @@ -99,7 +138,11 @@ 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 checkPlacementRestriction { + 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) @@ -145,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 } @@ -203,10 +256,36 @@ func (bs *balanceSolver) solve() []*operator.Operator { for _, srcStore := range bs.filterSrcStores() { bs.cur.srcStore = srcStore srcStoreID := srcStore.GetID() + 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] { - 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 { + 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) + sourceResultRecorded = true + } if bs.opTy == movePeer && !snapshotFilter.Select(bs.cur.region).IsOK() { hotSchedulerSnapshotSenderLimitCounter.Inc() continue @@ -235,9 +314,18 @@ func (bs *balanceSolver) solve() []*operator.Operator { hotSchedulerSearchRevertRegionsCounter.Inc() dstStoreID := dstStore.GetID() for _, revertPeerStat := range bs.filteredHotPeers[dstStoreID] { - revertRegion := bs.getRegion(revertPeerStat, dstStoreID) + revertRegion, revertFit := bs.getRegion(revertPeerStat, dstStoreID) if revertRegion == nil || revertRegion.GetID() == bs.cur.region.GetID() || - !allowRevertRegion(revertRegion, srcStoreID) { + !allowRevertRegion(revertRegion, srcStoreID) || + bs.placementV2Enabled && !bs.isRevertRegionValid(revertRegion, revertFit) { + continue + } + bs.revertScope = nil + if bs.mayUsePlacementScope(dstStore.IsTiKV()) { + 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 @@ -247,9 +335,16 @@ func (bs *balanceSolver) solve() []*operator.Operator { } bs.cur.revertPeerStat = nil bs.cur.revertRegion = nil + bs.revertScope = nil } } } + if !sourceResultRecorded { + if !checkedRegion { + sourceFailure = globalSourceFailure + } + bs.recordSourceStoreResult(sourceFailureOrSuccess(sourceFailure), srcStoreID) + } } bs.setSearchRevertRegions() @@ -374,14 +469,13 @@ 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. +// 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) - 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,27 +483,55 @@ 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() - continue + if !bs.mayUsePlacementScope(detail.IsTiKV()) { + failure := bs.sourceStoreFailure(detail, nil) + bs.recordSourceStoreResult(sourceFailureOrSuccess(failure), id) + if failure != "" { + continue + } } - ret[id] = detail - hotSchedulerResultCounter.WithLabelValues("src-store-succ-"+bs.resourceTy.String(), strconv.FormatUint(id, 10)).Inc() } return ret } +func (bs *balanceSolver) sourceStoreFailure(detail *statistics.StoreLoadDetail, scope *placementLoadScope) string { + srcToleranceRatio := bs.sche.conf.getSrcToleranceRatio() + if !detail.IsTiKV() { + srcToleranceRatio += tiflashToleranceRatioCorrection + } + expect := bs.expectLoad(detail, scope) + if !bs.checkSrcByPriorityAndTolerance(detail.LoadPred.Min(), expect, srcToleranceRatio) { + return "src-store-failed" + } + 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() +} + 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] @@ -499,31 +621,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 } - if !filter.IsRegionReplicated(bs.SchedulerCluster, region) { + var fit *placement.RegionFit + replicated := false + if bs.GetSchedulerConfig().IsPlacementRulesEnabled() { + 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 { +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 { @@ -533,20 +665,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 @@ -566,7 +698,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) @@ -586,7 +718,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 { @@ -606,7 +738,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() { @@ -622,13 +754,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 @@ -636,16 +780,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.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) { - 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 } @@ -656,6 +795,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] @@ -721,13 +879,150 @@ func (bs *balanceSolver) enableExpectation() bool { return bs.sche.conf.getDstToleranceRatio() > 0 && bs.sche.conf.getSrcToleranceRatio() > 0 } +// 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 := region.GetStorePeer(source.GetID()) + if sourcePeer == nil { + return nil + } + sourceFit := fit.GetRuleFit(sourcePeer.GetId()) + if sourceFit == nil { + return nil + } + + 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 bs.getPlacementLoadScope(rules, source.IsTiKV()) +} + +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}}} + } + 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 bs.placementCanRestrict[1] +} + +// 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 { + 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 + } + } + + 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) + } + } + } + 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 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 9a284c625b8..35bd9e2eefb 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" @@ -34,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() @@ -656,6 +691,129 @@ 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, 7: 200, 8: 300} { + pool := "other" + 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, labels)}, + 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) + 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])) + 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) { + 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 8ddc98c0caa..b02e00d9c5f 100644 --- a/pkg/schedule/schedulers/hot_region_test.go +++ b/pkg/schedule/schedulers/hot_region_test.go @@ -410,6 +410,309 @@ 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 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) + 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) + 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 TestHotWriteRegionScheduleWithPlacementScope(t *testing.T) { + testCases := []struct { + name string + storeLoads map[uint64]float64 + peerLoad float64 + exclusive bool + rankV1 bool + schedule bool + }{ + { + name: "source below global expectation", + storeLoads: map[uint64]float64{ + 1: 20, + 2: 0, + 3: 0, + 4: 0, + 5: 100, + 6: 100, + }, + peerLoad: 5, + schedule: true, + }, + { + name: "target above global expectation", + storeLoads: map[uint64]float64{ + 1: 100, + 2: 80, + 3: 80, + 4: 60, + 5: 0, + 6: 0, + }, + peerLoad: 20, + schedule: true, + }, + { + name: "implicit exclusive labels", + storeLoads: map[uint64]float64{ + 1: 20, + 2: 0, + 3: 0, + 4: 0, + 5: 100, + 6: 100, + }, + 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, + }, + } + + for _, testCase := range testCases { + 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++ { + 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)) + } + if !testCase.exclusive { + setPoolPlacementRule(t, tc) + } + + addRegionInfo(tc, utils.Write, []testRegionInfo{ + {1, []uint64{1, 2, 3}, testCase.peerLoad * units.MiB, 0, 0}, + }) + + 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) + }) + } +} + +func TestHotReadLeaderScheduleWithPlacementScope(t *testing.T) { + re := require.New(t) + tc, hb := preparePlacementHotScheduler(t, readType, readLeader) + hb.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) + setPoolPlacementRule(t, tc) + + 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 TestHotWriteLeaderScheduleWithPlacementScope(t *testing.T) { + 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) + + // 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) + 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) { + 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} + + 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) + setPoolPlacementRule(t, tc) + + 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() diff --git a/pkg/statistics/store_hot_peers_infos.go b/pkg/statistics/store_hot_peers_infos.go index 5d58663c0e0..ed93755ea89 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 037b89e26dc..d61d142c617 100644 --- a/pkg/statistics/store_load.go +++ b/pkg/statistics/store_load.go @@ -80,6 +80,69 @@ 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. 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 + } + 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)