grafana add store infomation

Signed-off-by: bufferflies <1045931706@qq.com>

server/cluster/cluster.go

@@ -894,6 +894,11 @@ func (c *RaftCluster) GetStores() []*core.StoreInfo {
 	return c.core.GetStores()
 }
 
+// GetLeaderStoreByRegionID returns the leader store of the given region.
+func (c *RaftCluster) GetLeaderStoreByRegionID(regionID uint64) *core.StoreInfo {
+	return c.core.GetLeaderStoreByRegionID(regionID)
+}
+
 // GetStore gets store from cluster.
 func (c *RaftCluster) GetStore(storeID uint64) *core.StoreInfo {
 	return c.core.GetStore(storeID)

server/core/basic_cluster.go

@@ -116,6 +116,17 @@ func (bc *BasicCluster) GetFollowerStores(region *RegionInfo) []*StoreInfo {
 	return Stores
 }
 
+// GetLeaderStoreByRegionID returns the leader store of the given region.
+func (bc *BasicCluster) GetLeaderStoreByRegionID(regionID uint64) *StoreInfo {
+	bc.RLock()
+	defer bc.RUnlock()
+	region := bc.Regions.GetRegion(regionID)
+	if region == nil || region.GetLeader() == nil {
+		return nil
+	}
+	return bc.Stores.GetStore(region.GetLeader().GetStoreId())
+}
+
 // GetLeaderStore returns all Stores that contains the region's leader peer.
 func (bc *BasicCluster) GetLeaderStore(region *RegionInfo) *StoreInfo {
 	bc.RLock()

server/grpc_service.go

@@ -788,19 +788,26 @@ func (s *GrpcServer) ReportBuckets(stream pdpb.PD_ReportBucketsServer) error {
 		if err := s.validateRequest(request.GetHeader()); err != nil {
 			return err
 		}
-		start := time.Now()
-		bucketReportCounter.WithLabelValues("report", "recv").Inc()
 		buckets := request.GetBuckets()
 		if buckets == nil || len(buckets.Keys) == 0 {
 			continue
 		}
-		err = rc.HandleBucketHeartbeat(request.Buckets)
+		store := rc.GetLeaderStoreByRegionID(buckets.GetRegionId())
+		if store == nil {
+			return errors.Errorf("region's store %d not found", buckets.GetRegionId())
+		}
+		storeLabel := strconv.FormatUint(store.GetID(), 10)
+		storeAddress := store.GetAddress()
+		bucketReportCounter.WithLabelValues(storeAddress, storeLabel, "report", "recv").Inc()
+
+		start := time.Now()
+		err = rc.HandleBucketHeartbeat(buckets)
 		if err != nil {
-			bucketReportCounter.WithLabelValues("report", "err").Inc()
+			bucketReportCounter.WithLabelValues(storeAddress, storeLabel, "report", "err").Inc()
 			continue
 		}
-		bucketReportHandleDuration.WithLabelValues().Observe(time.Since(start).Seconds())
-		bucketReportCounter.WithLabelValues("report", "ok").Inc()
+		bucketReportLatency.WithLabelValues(storeAddress, storeLabel).Observe(time.Since(start).Seconds())
+		bucketReportCounter.WithLabelValues(storeAddress, storeLabel, "report", "ok").Inc()
 	}
 }
 

server/metrics.go

@@ -30,7 +30,7 @@ var (
 			Subsystem: "server",
 			Name:      "bucket_report",
 			Help:      "Counter of bucket report.",
-		}, []string{"type", "status"})
+		}, []string{"address", "store", "type", "status"})
 	regionHeartbeatCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: "pd",
@@ -91,14 +91,14 @@ var (
 			Buckets:   prometheus.ExponentialBuckets(0.0005, 2, 13),
 		})
 
-	bucketReportHandleDuration = prometheus.NewHistogramVec(
+	bucketReportLatency = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Namespace: "pd",
 			Subsystem: "server",
 			Name:      "handle_bucket_report_duration_seconds",
 			Help:      "Bucketed histogram of processing time (s) of handled bucket report requests.",
 			Buckets:   prometheus.ExponentialBuckets(0.0001, 2, 29), // 0.1ms ~ 7hours
-		}, []string{})
+		}, []string{"address", "store"})
 
 	regionHeartbeatHandleDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -149,6 +149,6 @@ func init() {
 	prometheus.MustRegister(storeHeartbeatHandleDuration)
 	prometheus.MustRegister(serverInfo)
 	prometheus.MustRegister(bucketReportCounter)
-	prometheus.MustRegister(bucketReportHandleDuration)
+	prometheus.MustRegister(bucketReportLatency)
 	prometheus.MustRegister(serviceAuditHistogram)
 }