Skip to content
Merged
4 changes: 3 additions & 1 deletion pkg/schedule/schedulers/balance_region.go
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,6 @@ func (s *balanceRegionScheduler) Schedule(cluster sche.SchedulerCluster, dryRun
// sourcesStore is sorted by region score desc, so we pick the first store as source store.
for sourceIndex, solver.Source = range sourceStores {
retryLimit := s.getLimit(solver.Source)
solver.sourceScore = solver.sourceStoreScore(s.GetName())
if sourceIndex == len(sourceStores)-1 {
break
}
Expand Down Expand Up @@ -196,6 +195,9 @@ func (s *balanceRegionScheduler) Schedule(cluster sche.SchedulerCluster, dryRun
continue
}
solver.Step++
// Now that the candidate region is known, its size can be folded
// into the source score the same way targetStoreScore folds it in.
solver.sourceScore = solver.sourceStoreScore(s.GetName())
// the replica filter will cache the last region fit and the select one will only pict the first one region that
// satisfy all the filters, so the region fit must belong the scheduled region.
solver.fit = replicaFilter.(*filter.RegionReplicatedFilter).GetFit()
Expand Down
182 changes: 182 additions & 0 deletions pkg/schedule/schedulers/balance_region_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,13 @@ package schedulers
import (
"fmt"
"testing"
"time"

"github.com/docker/go-units"
"github.com/stretchr/testify/require"

"github.com/pingcap/kvproto/pkg/metapb"
"github.com/pingcap/kvproto/pkg/pdpb"

"github.com/tikv/pd/pkg/core"
"github.com/tikv/pd/pkg/core/constant"
Expand Down Expand Up @@ -77,6 +80,185 @@ func TestInfluenceAmp(t *testing.T) {
re.Less(solver.sourceScore-solver.targetScore, float64(1))
}

// TestBalanceRegionOrdinaryMoveNotBlockedByCandidateSize guards against
// double-counting the candidate region's size on top of tolerantResource in
// targetStoreScore. tolerantResource already represents roughly one average
// region's worth of margin, so a candidate whose size sits at or below that
// margin must not additionally raise the bar the target has to clear. Here
// the source holds three ordinary, equal-sized regions and the target is
// empty: moving one region is a clear improvement (192 vs 96 after the move)
// and must still be scheduled.
func TestBalanceRegionOrdinaryMoveNotBlockedByCandidateSize(t *testing.T) {
cancel, _, tc, oc := prepareSchedulersTest()
defer cancel()
re := require.New(t)

tc.SetTolerantSizeRatio(1)
tc.SetRegionScoreFormulaVersion("v1")

tc.AddRegionStore(1, 3, 288)
tc.AddRegionStore(2, 0, 0)
tc.AddLeaderRegion(1, 1)
region := tc.GetRegion(1).Clone(core.SetApproximateSize(96))
tc.PutRegion(region)

kind := constant.NewScheduleKind(constant.RegionKind, constant.BySize)
influence := oc.GetOpInfluence(tc.GetBasicCluster())
basePlan := plan.NewBalanceSchedulerPlan()
solver := newSolver(basePlan, kind, tc, influence)
solver.Source, solver.Target, solver.Region = tc.GetStore(1), tc.GetStore(2), tc.GetRegion(1)
solver.sourceScore, solver.targetScore = solver.sourceStoreScore(""), solver.targetStoreScore("")
re.True(solver.shouldBalance(""))
}

// TestBalanceRegionLargeCandidateDoesNotOvershoot guards against overshoot
// when the candidate region is much larger than the tolerant margin.
// sourceStoreScore and targetStoreScore both apply the same
// max(tolerantResource, candidateSize) delta, so a candidate that would leave
// the target heavier than the source after the move is correctly rejected:
// tolerantResource=10, candidate=100, source=150, target=0 project to
// source=50, target=100 after the move, which must not be scheduled even
// though comparing raw pre-move sizes alone would allow it.
func TestBalanceRegionLargeCandidateDoesNotOvershoot(t *testing.T) {
cancel, _, tc, oc := prepareSchedulersTest()
defer cancel()
re := require.New(t)

tc.SetTolerantSizeRatio(0.1)
tc.SetRegionScoreFormulaVersion("v1")

tc.AddRegionStore(1, 1, 150)
tc.AddRegionStore(2, 0, 0)
tc.AddLeaderRegion(1, 1)
region := tc.GetRegion(1).Clone(core.SetApproximateSize(100))
tc.PutRegion(region)

kind := constant.NewScheduleKind(constant.RegionKind, constant.BySize)
influence := oc.GetOpInfluence(tc.GetBasicCluster())
basePlan := plan.NewBalanceSchedulerPlan()
solver := newSolver(basePlan, kind, tc, influence)
solver.Source, solver.Target, solver.Region = tc.GetStore(1), tc.GetStore(2), tc.GetRegion(1)
solver.sourceScore, solver.targetScore = solver.sourceStoreScore(""), solver.targetStoreScore("")
re.False(solver.shouldBalance(""))
}

// TestSingleRegionOnLargeEmptyDiskDoesNotMigrate verifies that when a store's
// disk is mostly empty except for a single small region (e.g. 6TiB capacity
// with only a 10MiB region), balance-region does NOT move that region to an
// otherwise-identical, entirely empty peer store. Moving it would not fix any
// real imbalance (10MiB vs 6TiB capacity is a negligible utilization
// difference either way) and would just relocate the same "which store holds
// the only real data" state onto a different empty store — inviting the kind
// of pointless churn reported in #11135, since every other empty store looks
// equally attractive as a target on the next scheduling pass.
func TestSingleRegionOnLargeEmptyDiskDoesNotMigrate(t *testing.T) {
cancel, _, tc, oc := prepareSchedulersTest()
defer cancel()
re := require.New(t)

const (
capacity = 6 * units.TiB
regionSizeMB = 10 // MiB, matches core.StoreInfo.GetRegionSize()'s unit
)

mkStore := func(id uint64, usedMB int64) *core.StoreInfo {
usedBytes := uint64(usedMB) * units.MiB
stats := &pdpb.StoreStats{
Capacity: capacity,
UsedSize: usedBytes,
Available: capacity - usedBytes,
}
return core.NewStoreInfo(
&metapb.Store{Id: id, State: metapb.StoreState_Up},
core.SetStoreStats(stats),
core.SetRegionCount(10),
core.SetRegionSize(usedMB),
core.SetLastHeartbeatTS(time.Now()),
)
}

// storeA holds the cluster's only non-empty region; storeB is otherwise
// identical (same capacity, same region count) but has no data at all,
// and does not already hold a peer of region 1 — a real Schedule() run
// would still consider it a legitimate, unfiltered candidate target.
tc.PutStore(mkStore(1, regionSizeMB))
tc.PutStore(mkStore(2, 0))

tc.AddLeaderRegion(1, 1)
region := tc.GetRegion(1).Clone(core.SetApproximateSize(regionSizeMB))
tc.PutRegion(region)

// Mirror the real-world setup that motivated this test: each store already
// holds 10 regions (region *count* is balanced), but region 1 above is the
// only one with any data — the other 19 are freshly-split, empty regions.
var nextID uint64 = 2
for range 9 {
tc.AddLeaderRegion(nextID, 1)
empty := tc.GetRegion(nextID).Clone(core.SetApproximateSize(0))
tc.PutRegion(empty)
nextID++
}
for range 10 {
tc.AddLeaderRegion(nextID, 2)
empty := tc.GetRegion(nextID).Clone(core.SetApproximateSize(0))
tc.PutRegion(empty)
nextID++
}

kind := constant.NewScheduleKind(constant.RegionKind, constant.BySize)
influence := oc.GetOpInfluence(tc.GetBasicCluster())
basePlan := plan.NewBalanceSchedulerPlan()
solver := newSolver(basePlan, kind, tc, influence)
solver.Source, solver.Target, solver.Region = tc.GetStore(1), tc.GetStore(2), tc.GetRegion(1)
solver.sourceScore, solver.targetScore = solver.sourceStoreScore(""), solver.targetStoreScore("")
re.False(solver.shouldBalance(""))
}

// TestBalanceRegionRealScheduleDoesNotMoveSoleRegion drives the actual
// Schedule() entry point (not the solver methods directly) through the
// #11135 churn scenario: one store holds only the cluster's one real region,
// the other holds enough empty regions to dilute GetAverageRegionSize() well
// below the candidate's size. On pre-PR code this deterministically produces
// an operator — the real region is its store's only candidate, so region
// selection can't dodge it by picking an empty region instead — while the
// symmetric getRegionScoreDelta() must refuse it here. Checked in both
// directions so the result isn't an artifact of store iteration/sort order.
func TestBalanceRegionRealScheduleDoesNotMoveSoleRegion(t *testing.T) {
for _, realOnStore1 := range []bool{true, false} {
cancel, _, tc, oc := prepareSchedulersTest(false)
re := require.New(t)

tc.SetTolerantSizeRatio(1)
tc.SetRegionScoreFormulaVersion("v1")
tc.SetClusterVersion(versioninfo.MinSupportedVersion(versioninfo.Version4_0))
tc.SetEnablePlacementRules(false)
tc.SetMaxReplicasWithLabel(false, 1)
sb, err := CreateScheduler(types.BalanceRegionScheduler, oc, storage.NewStorageWithMemoryBackend(), ConfigSliceDecoder(types.BalanceRegionScheduler, []string{"", ""}))
re.NoError(err)

realStore, emptyStore := uint64(1), uint64(2)
if !realOnStore1 {
realStore, emptyStore = 2, 1
}
tc.AddRegionStore(realStore, 1, 96)
tc.AddRegionStore(emptyStore, 9, 0)
tc.AddLeaderRegion(1, realStore)
tc.PutRegion(tc.GetRegion(1).Clone(core.SetApproximateSize(96)))
var nextID uint64 = 2
for range 9 {
tc.AddLeaderRegion(nextID, emptyStore)
tc.PutRegion(tc.GetRegion(nextID).Clone(core.SetApproximateSize(0)))
nextID++
}

for range 10 {
ops, _ := sb.Schedule(tc, false)
re.Empty(ops)
}
cancel()
}
}

func TestShouldBalance(t *testing.T) {
// store size = 100GiB
// region size = 96MiB
Expand Down
27 changes: 24 additions & 3 deletions pkg/schedule/schedulers/utils.go
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ func (p *solver) sourceStoreScore(scheduleName string) float64 {
sourceDelta := influence - tolerantResource
score = p.Source.LeaderScore(p.kind.Policy, sourceDelta)
case constant.RegionKind:
sourceDelta := influence*influenceAmp - tolerantResource
sourceDelta := influence*influenceAmp - p.getRegionScoreDelta()
score = p.Source.RegionScore(p.GetSchedulerConfig().GetRegionScoreFormulaVersion(), p.GetSchedulerConfig().GetHighSpaceRatio(), p.GetSchedulerConfig().GetLowSpaceRatio(), sourceDelta)
case constant.WitnessKind:
sourceDelta := influence - tolerantResource
Expand Down Expand Up @@ -139,7 +139,7 @@ func (p *solver) targetStoreScore(scheduleName string) float64 {
targetDelta := influence + tolerantResource
score = p.Target.LeaderScore(p.kind.Policy, targetDelta)
case constant.RegionKind:
targetDelta := influence*influenceAmp + tolerantResource
targetDelta := influence*influenceAmp + p.getRegionScoreDelta()
score = p.Target.RegionScore(p.GetSchedulerConfig().GetRegionScoreFormulaVersion(), p.GetSchedulerConfig().GetHighSpaceRatio(), p.GetSchedulerConfig().GetLowSpaceRatio(), targetDelta)
case constant.WitnessKind:
targetDelta := influence + tolerantResource
Expand All @@ -160,12 +160,20 @@ func (p *solver) shouldBalance(scheduleName string) bool {
shouldBalance := p.sourceScore > p.targetScore

if !shouldBalance && log.GetLevel() <= zap.DebugLevel {
// For RegionKind, the delta actually applied to sourceScore/targetScore
// is getRegionScoreDelta(), not the bare tolerant margin, whenever the
// candidate region is larger than that margin — log the value that
// actually drove the decision instead of the one that may not have.
tolerantResource := p.getTolerantResource()
if p.kind.Resource == constant.RegionKind {
tolerantResource = p.getRegionScoreDelta()
}
log.Debug("skip balance "+p.kind.Resource.String(),
zap.String("scheduler", scheduleName), zap.Uint64("region-id", p.Region.GetID()), zap.Uint64("source-store", sourceID), zap.Uint64("target-store", targetID),
zap.Int64("source-size", p.Source.GetRegionSize()), zap.Float64("source-score", p.sourceScore),
zap.Int64("target-size", p.Target.GetRegionSize()), zap.Float64("target-score", p.targetScore),
zap.Int64("average-region-size", p.GetAverageRegionSize()),
zap.Int64("tolerant-resource", p.getTolerantResource()))
zap.Int64("tolerant-resource", tolerantResource))
}
return shouldBalance
}
Expand All @@ -184,6 +192,19 @@ func (p *solver) getTolerantResource() int64 {
return p.tolerantSource
}

// getRegionScoreDelta returns the delta used to score a candidate move for
// RegionKind balancing. It is the larger of the general tolerant margin and
// the candidate region's own size, applied symmetrically to both the source
// and target side, so a projected post-move comparison isn't skewed by one
// side knowing the candidate's size and the other not.
func (p *solver) getRegionScoreDelta() int64 {
delta := p.getTolerantResource()
if p.Region != nil {
delta = max(delta, p.Region.GetApproximateSize())
Comment thread
bufferflies marked this conversation as resolved.
}
return delta
}

func adjustTolerantRatio(cluster sche.SchedulerCluster, kind constant.ScheduleKind) float64 {
var tolerantSizeRatio float64
switch c := cluster.(type) {
Expand Down
Loading