core, schedule: use size as a factor of score

conf/config.toml

@@ -51,6 +51,7 @@ max-store-down-time = "1h"
 leader-schedule-limit = 64
 region-schedule-limit = 16
 replica-schedule-limit = 24
+tolerant-size-ratio = 2.5
 
 # customized schedulers, the format is as below
 # if empty, it will use balance-leader, balance-region, hot-region as default

server/cache.go

@@ -170,6 +170,13 @@ func (c *clusterInfo) GetStores() []*core.StoreInfo {
 	return c.BasicCluster.GetStores()
 }
 
+// GetStoresAverageScore returns the total resource score of all StoreInfo
+func (c *clusterInfo) GetStoresAverageScore(kind core.ResourceKind) float64 {
+	c.RLock()
+	defer c.RUnlock()
+	return c.BasicCluster.GetStoresAverageScore(kind)
+}
+
 func (c *clusterInfo) getMetaStores() []*metapb.Store {
 	c.RLock()
 	defer c.RUnlock()

server/config.go

@@ -316,6 +316,8 @@ type ScheduleConfig struct {
 	RegionScheduleLimit uint64 `toml:"region-schedule-limit,omitempty" json:"region-schedule-limit"`
 	// ReplicaScheduleLimit is the max coexist replica schedules.
 	ReplicaScheduleLimit uint64 `toml:"replica-schedule-limit,omitempty" json:"replica-schedule-limit"`
+	// TolerantSizeRatio is the ratio of buffer size for balance scheduler.
+	TolerantSizeRatio float64 `toml:"tolerant-size-ratio,omitempty" json:"tolerant-size-ratio"`
 	// Schedulers support for loding customized schedulers
 	Schedulers SchedulerConfigs `toml:"schedulers,omitempty" json:"schedulers-v2"` // json v2 is for the sake of compatible upgrade
 }
@@ -329,6 +331,7 @@ func (c *ScheduleConfig) clone() *ScheduleConfig {
 		LeaderScheduleLimit:  c.LeaderScheduleLimit,
 		RegionScheduleLimit:  c.RegionScheduleLimit,
 		ReplicaScheduleLimit: c.ReplicaScheduleLimit,
+		TolerantSizeRatio:    c.TolerantSizeRatio,
 		Schedulers:           schedulers,
 	}
 }
@@ -350,6 +353,7 @@ const (
 	defaultLeaderScheduleLimit  = 64
 	defaultRegionScheduleLimit  = 12
 	defaultReplicaScheduleLimit = 16
+	defaultTolerantSizeRatio    = 2.5
 )
 
 var defaultSchedulers = SchedulerConfigs{
@@ -365,6 +369,7 @@ func (c *ScheduleConfig) adjust() {
 	adjustUint64(&c.LeaderScheduleLimit, defaultLeaderScheduleLimit)
 	adjustUint64(&c.RegionScheduleLimit, defaultRegionScheduleLimit)
 	adjustUint64(&c.ReplicaScheduleLimit, defaultReplicaScheduleLimit)
+	adjustFloat64(&c.TolerantSizeRatio, defaultTolerantSizeRatio)
 	adjustSchedulers(&c.Schedulers, defaultSchedulers)
 }
 
@@ -454,6 +459,10 @@ func (o *scheduleOption) GetReplicaScheduleLimit() uint64 {
 	return o.load().ReplicaScheduleLimit
 }
 
+func (o *scheduleOption) GetTolerantSizeRatio() float64 {
+	return o.load().TolerantSizeRatio
+}
+
 func (o *scheduleOption) GetSchedulers() SchedulerConfigs {
 	return o.load().Schedulers
 }

server/core/store.go

@@ -106,17 +106,17 @@ const minWeight = 1e-6
 // LeaderScore returns the store's leader score: leaderCount / leaderWeight.
 func (s *StoreInfo) LeaderScore() float64 {
 	if s.LeaderWeight <= 0 {
-		return float64(s.LeaderCount) / minWeight
+		return float64(s.LeaderSize) / minWeight
 	}
-	return float64(s.LeaderCount) / s.LeaderWeight
+	return float64(s.LeaderSize) / s.LeaderWeight
 }
 
 // RegionScore returns the store's region score: regionSize / regionWeight.
 func (s *StoreInfo) RegionScore() float64 {
 	if s.RegionWeight <= 0 {
-		return float64(s.RegionCount) / minWeight
+		return float64(s.RegionSize) / minWeight
 	}
-	return float64(s.RegionCount) / s.RegionWeight
+	return float64(s.RegionSize) / s.RegionWeight
 }
 
 // StorageSize returns store's used storage size reported from tikv.
@@ -175,6 +175,24 @@ func (s *StoreInfo) ResourceScore(kind ResourceKind) float64 {
 	}
 }
 
+// ResourceWeight returns weight of leader/region in the score
+func (s *StoreInfo) ResourceWeight(kind ResourceKind) float64 {
+	switch kind {
+	case LeaderKind:
+		if s.LeaderWeight <= 0 {
+			return minWeight
+		}
+		return s.LeaderWeight
+	case RegionKind:
+		if s.RegionWeight <= 0 {
+			return minWeight
+		}
+		return s.RegionWeight
+	default:
+		return 0
+	}
+}
+
 // GetStartTS returns the start timestamp.
 func (s *StoreInfo) GetStartTS() time.Time {
 	return time.Unix(int64(s.Stats.GetStartTime()), 0)
@@ -366,6 +384,23 @@ func (s *StoresInfo) SetRegionSize(storeID uint64, regionSize int64) {
 	}
 }
 
+// AverageResourceScore return the total resource score of all StoreInfo
+func (s *StoresInfo) AverageResourceScore(kind ResourceKind) float64 {
+	var totalResourceSize int64
+	var totalResourceWeight float64
+	for _, s := range s.stores {
+		if s.IsUp() {
+			totalResourceWeight += s.ResourceWeight(kind)
+			totalResourceSize += s.ResourceSize(kind)
+		}
+	}
+
+	if totalResourceWeight == 0 {
+		return 0
+	}
+	return float64(totalResourceSize) / totalResourceWeight
+}
+
 // TotalWrittenBytes return the total written bytes of all StoreInfo
 func (s *StoresInfo) TotalWrittenBytes() uint64 {
 	var totalWrittenBytes uint64

server/schedule/basic_cluster.go

@@ -135,6 +135,11 @@ func (bc *BasicCluster) GetLeaderStore(region *core.RegionInfo) *core.StoreInfo
 	return bc.Stores.GetStore(region.Leader.GetStoreId())
 }
 
+// GetStoresAverageScore returns the total resource score of all StoreInfo
+func (bc *BasicCluster) GetStoresAverageScore(kind core.ResourceKind) float64 {
+	return bc.Stores.AverageResourceScore(kind)
+}
+
 // BlockStore stops balancer from selecting the store.
 func (bc *BasicCluster) BlockStore(storeID uint64) error {
 	return errors.Trace(bc.Stores.BlockStore(storeID))

server/schedule/opts.go

@@ -21,6 +21,7 @@ import (
 type Options interface {
 	GetLeaderScheduleLimit() uint64
 	GetRegionScheduleLimit() uint64
+	GetTolerantSizeRatio() float64
 
 	GetMaxSnapshotCount() uint64
 	GetMaxPendingPeerCount() uint64

server/schedule/scheduler.go

@@ -34,6 +34,7 @@ type Cluster interface {
 	GetRegionStores(region *core.RegionInfo) []*core.StoreInfo
 	GetFollowerStores(region *core.RegionInfo) []*core.StoreInfo
 	GetLeaderStore(region *core.RegionInfo) *core.StoreInfo
+	GetStoresAverageScore(kind core.ResourceKind) float64
 	ScanRegions(startKey []byte, limit int) []*core.RegionInfo
 
 	BlockStore(id uint64) error

server/schedulers/balance_leader.go

@@ -76,7 +76,8 @@ func (l *balanceLeaderScheduler) Schedule(cluster schedule.Cluster, opInfluence
 
 	source := cluster.GetStore(region.Leader.GetStoreId())
 	target := cluster.GetStore(newLeader.GetStoreId())
-	if !shouldBalance(source, target, core.LeaderKind, opInfluence) {
+	avgScore := cluster.GetStoresAverageScore(core.LeaderKind)
+	if !shouldBalance(source, target, avgScore, core.LeaderKind, region, opInfluence, l.opt.GetTolerantSizeRatio()) {
 		schedulerCounter.WithLabelValues(l.GetName(), "skip").Inc()
 		return nil
 	}

server/schedulers/balance_region.go

@@ -118,7 +118,8 @@ func (s *balanceRegionScheduler) transferPeer(cluster schedule.Cluster, region *
 	}
 
 	target := cluster.GetStore(newPeer.GetStoreId())
-	if !shouldBalance(source, target, core.RegionKind, opInfluence) {
+	avgScore := cluster.GetStoresAverageScore(core.RegionKind)
+	if !shouldBalance(source, target, avgScore, core.RegionKind, region, opInfluence, s.opt.GetTolerantSizeRatio()) {
 		schedulerCounter.WithLabelValues(s.GetName(), "skip").Inc()
 		return nil
 	}