metrics,internal/client: add metrics for batch client stream receiving duration (#279)

Signed-off-by: tiancaiamao <tiancaiamao@gmail.com>

Co-authored-by: disksing <i@disksing.com>

internal/client/client_batch.go

@@ -425,6 +425,7 @@ type batchCommandsStream struct {
 }
 
 func (s *batchCommandsStream) recv() (resp *tikvpb.BatchCommandsResponse, err error) {
+	now := time.Now()
 	defer func() {
 		if r := recover(); r != nil {
 			metrics.TiKVPanicCounter.WithLabelValues(metrics.LabelBatchRecvLoop).Inc()
@@ -433,6 +434,11 @@ func (s *batchCommandsStream) recv() (resp *tikvpb.BatchCommandsResponse, err er
 				zap.Stack("stack"))
 			err = errors.SuspendStack(errors.New("batch conn recv paniced"))
 		}
+		if err == nil {
+			metrics.BatchRecvHistogramOK.Observe(float64(time.Since(now)))
+		} else {
+			metrics.BatchRecvHistogramError.Observe(float64(time.Since(now)))
+		}
 	}()
 	if _, err := util.EvalFailpoint("gotErrorInRecvLoop"); err == nil {
 		return nil, errors.New("injected error in batchRecvLoop")

metrics/metrics.go

@@ -66,6 +66,7 @@ var (
 	TiKVBatchClientUnavailable               prometheus.Histogram
 	TiKVBatchClientWaitEstablish             prometheus.Histogram
 	TiKVBatchClientRecycle                   prometheus.Histogram
+	TiKVBatchRecvLatency                     *prometheus.HistogramVec
 	TiKVRangeTaskStats                       *prometheus.GaugeVec
 	TiKVRangeTaskPushDuration                *prometheus.HistogramVec
 	TiKVTokenWaitDuration                    prometheus.Histogram
@@ -275,6 +276,15 @@ func initMetrics(namespace, subsystem string) {
 			Help:      "batch send latency",
 		})
 
+	TiKVBatchRecvLatency = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "batch_recv_latency",
+			Buckets:   prometheus.ExponentialBuckets(1000, 2, 34), // 1us ~ 8000s
+			Help:      "batch recv latency",
+		}, []string{LblResult})
+
 	TiKVBatchWaitOverLoad = prometheus.NewCounter(
 		prometheus.CounterOpts{
 			Namespace: namespace,
@@ -554,6 +564,7 @@ func RegisterMetrics() {
 	prometheus.MustRegister(TiKVStatusCounter)
 	prometheus.MustRegister(TiKVBatchWaitDuration)
 	prometheus.MustRegister(TiKVBatchSendLatency)
+	prometheus.MustRegister(TiKVBatchRecvLatency)
 	prometheus.MustRegister(TiKVBatchWaitOverLoad)
 	prometheus.MustRegister(TiKVBatchPendingRequests)
 	prometheus.MustRegister(TiKVBatchRequests)

metrics/shortcuts.go

@@ -119,6 +119,9 @@ var (
 	OnePCTxnCounterOk       prometheus.Counter
 	OnePCTxnCounterError    prometheus.Counter
 	OnePCTxnCounterFallback prometheus.Counter
+
+	BatchRecvHistogramOK    prometheus.Observer
+	BatchRecvHistogramError prometheus.Observer
 )
 
 func initShortcuts() {
@@ -203,4 +206,7 @@ func initShortcuts() {
 	OnePCTxnCounterOk = TiKVOnePCTxnCounter.WithLabelValues("ok")
 	OnePCTxnCounterError = TiKVOnePCTxnCounter.WithLabelValues("err")
 	OnePCTxnCounterFallback = TiKVOnePCTxnCounter.WithLabelValues("fallback")
+
+	BatchRecvHistogramOK = TiKVBatchRecvLatency.WithLabelValues("ok")
+	BatchRecvHistogramError = TiKVBatchRecvLatency.WithLabelValues("err")
 }