merg

Signed-off-by: bufferflies <[email protected]>

Author: bufferflies

server/statistics/buckets/bucket_stat_informer.go

@@ -207,3 +207,10 @@ func (b *BucketTreeItem) calculateHotDegree() {
 		}
 	}
 }
+
+// collectBucketsMetrics collects the metrics of the hot stats.
+func (stats *BucketTreeItem) collectBucketsMetrics() {
+	for _, bucket := range stats.stats {
+		bucketsHotDegreeHist.Observe(float64(bucket.HotDegree))
+	}
+}

server/statistics/buckets/hot_bucket_cache.go

@@ -17,6 +17,7 @@ package buckets
 import (
 	"bytes"
 	"context"
+	"time"
 
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/log"
@@ -52,6 +53,23 @@ type HotBucketCache struct {
 	ctx             context.Context
 }
 
+// GetHotBucketStats returns the hot stats of the regions that great than degree.
+func (h *HotBucketCache) GetHotBucketStats(degree int) map[uint64][]*BucketStat {
+	rst := make(map[uint64][]*BucketStat)
+	for _, item := range h.bucketsOfRegion {
+		stats := make([]*BucketStat, 0)
+		for _, b := range item.stats {
+			if b.HotDegree >= degree {
+				stats = append(stats, b)
+			}
+		}
+		if len(stats) > 0 {
+			rst[item.regionID] = stats
+		}
+	}
+	return rst
+}
+
 // bucketDebrisFactory returns the debris if the key range of the item is bigger than the given key range.
 // start and end key:    | 001------------------------200|
 // the split key range:              |050---150|
@@ -137,7 +155,9 @@ func (h *HotBucketCache) schedule() {
 		case <-h.ctx.Done():
 			return
 		case task := <-h.taskQueue:
+			start := time.Now()
 			task.runTask(h)
+			bucketsTaskDuration.WithLabelValues(task.taskType().String()).Observe(time.Since(start).Seconds())
 		}
 	}
 }
@@ -156,6 +176,7 @@ func (h *HotBucketCache) checkBucketsFlow(buckets *metapb.Buckets) (newItem *Buc
 	}
 	newItem.inherit(overlaps)
 	newItem.calculateHotDegree()
+	newItem.collectBucketsMetrics()
 	return newItem, overlaps
 }
 

server/statistics/buckets/hot_bucket_task.go

@@ -15,18 +15,24 @@
 package buckets
 
 import (
+	"context"
+
 	"github.com/pingcap/kvproto/pkg/metapb"
 )
 
 type flowItemTaskKind uint32
 
 const (
 	checkBucketsTaskType flowItemTaskKind = iota
+	collectBucketStatsTaskType
 )
 
 func (kind flowItemTaskKind) String() string {
-	if kind == checkBucketsTaskType {
+	switch kind {
+	case checkBucketsTaskType:
 		return "check_buckets"
+	case collectBucketStatsTaskType:
+		return "collect_bucket_stats"
 	}
 	return "unknown"
 }
@@ -57,3 +63,34 @@ func (t *checkBucketsTask) runTask(cache *HotBucketCache) {
 	newItems, overlaps := cache.checkBucketsFlow(t.Buckets)
 	cache.putItem(newItems, overlaps)
 }
+
+type collectBucketStatsTask struct {
+	minDegree int
+	ret       chan map[uint64][]*BucketStat // RegionID ==>Buckets
+}
+
+// NewCollectBucketStatsTask creates task to collect bucket stats.
+func NewCollectBucketStatsTask(minDegree int) *collectBucketStatsTask {
+	return &collectBucketStatsTask{
+		minDegree: minDegree,
+		ret:       make(chan map[uint64][]*BucketStat, 1),
+	}
+}
+
+func (t *collectBucketStatsTask) taskType() flowItemTaskKind {
+	return collectBucketStatsTaskType
+}
+
+func (t *collectBucketStatsTask) runTask(cache *HotBucketCache) {
+	t.ret <- cache.GetHotBucketStats(t.minDegree)
+}
+
+// WaitRet returns the result of the task.
+func (t *collectBucketStatsTask) WaitRet(ctx context.Context) map[uint64][]*BucketStat {
+	select {
+	case <-ctx.Done():
+		return nil
+	case ret := <-t.ret:
+		return ret
+	}
+}

server/statistics/buckets/hot_bucket_task_test.go

@@ -29,14 +29,8 @@ var _ = Suite(&testHotBucketTaskCache{})
 type testHotBucketTaskCache struct {
 }
 
-func (s *testHotBucketTaskCache) SetUpSuite(_ *C) {
-}
-
-func (s *testHotBucketTaskCache) TearDownTest(_ *C) {
-}
-
 func getAllBucketStats(ctx context.Context, hotCache *HotBucketCache) map[uint64][]*BucketStat {
-	task := NewCollectBucketStatsTask(-100)
+	task := NewCollectBucketStatsTask(minHotDegree)
 	hotCache.CheckAsync(task)
 	return task.WaitRet(ctx)
 }
@@ -56,7 +50,7 @@ func (s *testHotBucketTaskCache) TestColdHot(c *C) {
 		isHot:   true,
 	}}
 	for _, v := range testdata {
-		for i := 0; i < 100; i++ {
+		for i := 0; i < 20; i++ {
 			task := NewCheckPeerTask(v.buckets)
 			c.Assert(hotCache.CheckAsync(task), IsTrue)
 			hotBuckets := getAllBucketStats(ctx, hotCache)