scheduler: hotspot: solution-based key-rate-aware balance solver (#2141)

tikv · Feb 26, 2020 · c8589fc · c8589fc
1 parent 904e833
commit c8589fc
Show file tree

Hide file tree

Showing 7 changed files with 1,120 additions and 501 deletions.
diff --git a/pkg/mock/mockcluster/mockcluster.go b/pkg/mock/mockcluster/mockcluster.go
@@ -270,9 +270,14 @@ func (mc *Cluster) AddLeaderRegionWithRange(regionID uint64, startKey string, en
 }
 
 // AddLeaderRegionWithReadInfo adds region with specified leader, followers and read info.
-func (mc *Cluster) AddLeaderRegionWithReadInfo(regionID uint64, leaderID uint64, readBytes uint64, reportInterval uint64, followerIds ...uint64) {
+func (mc *Cluster) AddLeaderRegionWithReadInfo(
+	regionID uint64, leaderID uint64,
+	readBytes, readKeys uint64,
+	reportInterval uint64,
+	followerIds []uint64) {
 	r := mc.newMockRegionInfo(regionID, leaderID, followerIds...)
 	r = r.Clone(core.SetReadBytes(readBytes))
+	r = r.Clone(core.SetReadKeys(readKeys))
 	r = r.Clone(core.SetReportInterval(reportInterval))
 	items := mc.HotCache.CheckRead(r, mc.StoresStats)
 	for _, item := range items {
@@ -282,9 +287,14 @@ func (mc *Cluster) AddLeaderRegionWithReadInfo(regionID uint64, leaderID uint64,
 }
 
 // AddLeaderRegionWithWriteInfo adds region with specified leader, followers and write info.
-func (mc *Cluster) AddLeaderRegionWithWriteInfo(regionID uint64, leaderID uint64, writtenBytes uint64, reportInterval uint64, followerIds ...uint64) {
+func (mc *Cluster) AddLeaderRegionWithWriteInfo(
+	regionID uint64, leaderID uint64,
+	writtenBytes, writtenKeys uint64,
+	reportInterval uint64,
+	followerIds []uint64) {
 	r := mc.newMockRegionInfo(regionID, leaderID, followerIds...)
 	r = r.Clone(core.SetWrittenBytes(writtenBytes))
+	r = r.Clone(core.SetWrittenKeys(writtenKeys))
 	r = r.Clone(core.SetReportInterval(reportInterval))
 	items := mc.HotCache.CheckWrite(r, mc.StoresStats)
 	for _, item := range items {
@@ -404,6 +414,32 @@ func (mc *Cluster) UpdateStorageReadBytes(storeID uint64, bytesRead uint64) {
 	mc.PutStore(newStore)
 }
 
+// UpdateStorageWrittenKeys updates store written keys.
+func (mc *Cluster) UpdateStorageWrittenKeys(storeID uint64, keysWritten uint64) {
+	store := mc.GetStore(storeID)
+	newStats := proto.Clone(store.GetStoreStats()).(*pdpb.StoreStats)
+	newStats.KeysWritten = keysWritten
+	now := time.Now().Second()
+	interval := &pdpb.TimeInterval{StartTimestamp: uint64(now - statistics.StoreHeartBeatReportInterval), EndTimestamp: uint64(now)}
+	newStats.Interval = interval
+	newStore := store.Clone(core.SetStoreStats(newStats))
+	mc.Set(storeID, newStats)
+	mc.PutStore(newStore)
+}
+
+// UpdateStorageReadKeys updates store read bytes.
+func (mc *Cluster) UpdateStorageReadKeys(storeID uint64, keysRead uint64) {
+	store := mc.GetStore(storeID)
+	newStats := proto.Clone(store.GetStoreStats()).(*pdpb.StoreStats)
+	newStats.KeysRead = keysRead
+	now := time.Now().Second()
+	interval := &pdpb.TimeInterval{StartTimestamp: uint64(now - statistics.StoreHeartBeatReportInterval), EndTimestamp: uint64(now)}
+	newStats.Interval = interval
+	newStore := store.Clone(core.SetStoreStats(newStats))
+	mc.Set(storeID, newStats)
+	mc.PutStore(newStore)
+}
+
 // UpdateStoreStatus updates store status.
 func (mc *Cluster) UpdateStoreStatus(id uint64) {
 	leaderCount := mc.Regions.GetStoreLeaderCount(id)

diff --git a/server/cluster/coordinator.go b/server/cluster/coordinator.go
@@ -420,6 +420,8 @@ func (c *coordinator) collectHotSpotMetrics() {
 		infl := pendings[storeID]
 		// TODO: add to tidb-ansible after merging pending influence into operator influence.
 		hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "write_pending_influence_byte_rate").Set(infl.ByteRate)
+		hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "write_pending_influence_key_rate").Set(infl.KeyRate)
+		hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "write_pending_influence_count").Set(infl.Count)
 	}
 
 	// Collects hot read region metrics.
@@ -440,6 +442,8 @@ func (c *coordinator) collectHotSpotMetrics() {
 
 		infl := pendings[storeID]
 		hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "read_pending_influence_byte_rate").Set(infl.ByteRate)
+		hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "read_pending_influence_key_rate").Set(infl.KeyRate)
+		hotSpotStatusGauge.WithLabelValues(storeAddress, storeLabel, "read_pending_influence_count").Set(infl.Count)
 	}
 }