feat: add metrics (zeromicro#3624)

core/metric/histogram.go

@@ -8,18 +8,21 @@ import (
 type (
 	// A HistogramVecOpts is a histogram vector options.
 	HistogramVecOpts struct {
-		Namespace string
-		Subsystem string
-		Name      string
-		Help      string
-		Labels    []string
-		Buckets   []float64
+		Namespace   string
+		Subsystem   string
+		Name        string
+		Help        string
+		Labels      []string
+		Buckets     []float64
+		ConstLabels map[string]string
 	}
 
 	// A HistogramVec interface represents a histogram vector.
 	HistogramVec interface {
 		// Observe adds observation v to labels.
 		Observe(v int64, labels ...string)
+		// ObserveFloat allow to observe float64 values.
+		ObserveFloat(v float64, labels ...string)
 		close() bool
 	}
 
@@ -35,11 +38,12 @@ func NewHistogramVec(cfg *HistogramVecOpts) HistogramVec {
 	}
 
 	vec := prom.NewHistogramVec(prom.HistogramOpts{
-		Namespace: cfg.Namespace,
-		Subsystem: cfg.Subsystem,
-		Name:      cfg.Name,
-		Help:      cfg.Help,
-		Buckets:   cfg.Buckets,
+		Namespace:   cfg.Namespace,
+		Subsystem:   cfg.Subsystem,
+		Name:        cfg.Name,
+		Help:        cfg.Help,
+		Buckets:     cfg.Buckets,
+		ConstLabels: cfg.ConstLabels,
 	}, cfg.Labels)
 	prom.MustRegister(vec)
 	hv := &promHistogramVec{
@@ -58,6 +62,12 @@ func (hv *promHistogramVec) Observe(v int64, labels ...string) {
 	})
 }
 
+func (hv *promHistogramVec) ObserveFloat(v float64, labels ...string) {
+	update(func() {
+		hv.histogram.WithLabelValues(labels...).Observe(v)
+	})
+}
+
 func (hv *promHistogramVec) close() bool {
 	return prom.Unregister(hv.histogram)
 }

core/metric/histogram_test.go

@@ -14,7 +14,7 @@ func TestNewHistogramVec(t *testing.T) {
 		Help:    "rpc server requests duration(ms).",
 		Buckets: []float64{1, 2, 3},
 	})
-	defer histogramVec.close()
+	defer histogramVec.(*promHistogramVec).close()
 	histogramVecNil := NewHistogramVec(nil)
 	assert.NotNil(t, histogramVec)
 	assert.Nil(t, histogramVecNil)
@@ -28,21 +28,22 @@ func TestHistogramObserve(t *testing.T) {
 		Buckets: []float64{1, 2, 3},
 		Labels:  []string{"method"},
 	})
-	defer histogramVec.close()
+	defer histogramVec.(*promHistogramVec).close()
 	hv, _ := histogramVec.(*promHistogramVec)
 	hv.Observe(2, "/Users")
+	hv.ObserveFloat(1.1, "/Users")
 
 	metadata := `
 		# HELP counts rpc server requests duration(ms).
         # TYPE counts histogram
 `
 	val := `
 		counts_bucket{method="/Users",le="1"} 0
-		counts_bucket{method="/Users",le="2"} 1
-		counts_bucket{method="/Users",le="3"} 1
-		counts_bucket{method="/Users",le="+Inf"} 1
-		counts_sum{method="/Users"} 2
-        counts_count{method="/Users"} 1
+		counts_bucket{method="/Users",le="2"} 2
+		counts_bucket{method="/Users",le="3"} 2
+		counts_bucket{method="/Users",le="+Inf"} 2
+		counts_sum{method="/Users"} 3.1
+        counts_count{method="/Users"} 2
 `
 
 	err := testutil.CollectAndCompare(hv.histogram, strings.NewReader(metadata+val))

core/stores/redis/hook.go

@@ -54,9 +54,10 @@ func (h hook) AfterProcess(ctx context.Context, cmd red.Cmder) error {
 	duration := timex.Since(start)
 	if duration > slowThreshold.Load() {
 		logDuration(ctx, []red.Cmder{cmd}, duration)
+		metricSlowCount.Inc(cmd.Name())
 	}
 
-	metricReqDur.Observe(duration.Milliseconds(), cmd.Name())
+	metricReqDur.ObserveFloat(float64(duration)/float64(time.Millisecond), cmd.Name())
 	if msg := formatError(err); len(msg) > 0 {
 		metricReqErr.Inc(cmd.Name(), msg)
 	}

core/stores/redis/metrics.go

@@ -1,6 +1,12 @@
 package redis
 
-import "github.com/zeromicro/go-zero/core/metric"
+import (
+	"sync"
+
+	red "github.com/go-redis/redis/v8"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/zeromicro/go-zero/core/metric"
+)
 
 const namespace = "redis_client"
 
@@ -11,7 +17,7 @@ var (
 		Name:      "duration_ms",
 		Help:      "redis client requests duration(ms).",
 		Labels:    []string{"command"},
-		Buckets:   []float64{5, 10, 25, 50, 100, 250, 500, 1000, 2500},
+		Buckets:   []float64{0.25, 0.5, 1, 1.5, 2, 3, 5, 10, 25, 50, 100, 250, 500, 1000, 2000, 5000, 10000, 15000},
 	})
 	metricReqErr = metric.NewCounterVec(&metric.CounterVecOpts{
 		Namespace: namespace,
@@ -20,4 +26,164 @@ var (
 		Help:      "redis client requests error count.",
 		Labels:    []string{"command", "error"},
 	})
+	metricSlowCount = metric.NewCounterVec(&metric.CounterVecOpts{
+		Namespace: namespace,
+		Subsystem: "requests",
+		Name:      "slow_total",
+		Help:      "redis client requests slow count.",
+		Labels:    []string{"command"},
+	})
+
+	connLabels = []string{"key", "client_type"}
+
+	connCollector = newCollector()
+
+	_ prometheus.Collector = (*collector)(nil)
+)
+
+type (
+	statGetter struct {
+		clientType string
+		key        string
+		poolSize   int
+		poolStats  func() *red.PoolStats
+	}
+
+	// collector collects statistics from a redis client.
+	// It implements the prometheus.Collector interface.
+	collector struct {
+		hitDesc     *prometheus.Desc
+		missDesc    *prometheus.Desc
+		timeoutDesc *prometheus.Desc
+		totalDesc   *prometheus.Desc
+		idleDesc    *prometheus.Desc
+		staleDesc   *prometheus.Desc
+		maxDesc     *prometheus.Desc
+
+		clients []*statGetter
+		lock    sync.Mutex
+	}
 )
+
+func newCollector() *collector {
+	c := &collector{
+		hitDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_hit_total"),
+			"Number of times a connection was found in the pool",
+			connLabels, nil,
+		),
+		missDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_miss_total"),
+			"Number of times a connection was not found in the pool",
+			connLabels, nil,
+		),
+		timeoutDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_timeout_total"),
+			"Number of times a timeout occurred when looking for a connection in the pool",
+			connLabels, nil,
+		),
+		totalDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_conn_total_current"),
+			"Current number of connections in the pool",
+			connLabels, nil,
+		),
+		idleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_conn_idle_current"),
+			"Current number of idle connections in the pool",
+			connLabels, nil,
+		),
+		staleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_conn_stale_total"),
+			"Number of times a connection was removed from the pool because it was stale",
+			connLabels, nil,
+		),
+		maxDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_conn_max"),
+			"Max number of connections in the pool",
+			connLabels, nil,
+		),
+	}
+
+	prometheus.MustRegister(c)
+
+	return c
+}
+
+// Describe implements the prometheus.Collector interface.
+func (s *collector) Describe(descs chan<- *prometheus.Desc) {
+	descs <- s.hitDesc
+	descs <- s.missDesc
+	descs <- s.timeoutDesc
+	descs <- s.totalDesc
+	descs <- s.idleDesc
+	descs <- s.staleDesc
+	descs <- s.maxDesc
+}
+
+// Collect implements the prometheus.Collector interface.
+func (s *collector) Collect(metrics chan<- prometheus.Metric) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	for _, client := range s.clients {
+		key, clientType := client.key, client.clientType
+		stats := client.poolStats()
+
+		metrics <- prometheus.MustNewConstMetric(
+			s.hitDesc,
+			prometheus.CounterValue,
+			float64(stats.Hits),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.missDesc,
+			prometheus.CounterValue,
+			float64(stats.Misses),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.timeoutDesc,
+			prometheus.CounterValue,
+			float64(stats.Timeouts),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.totalDesc,
+			prometheus.GaugeValue,
+			float64(stats.TotalConns),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.idleDesc,
+			prometheus.GaugeValue,
+			float64(stats.IdleConns),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.staleDesc,
+			prometheus.CounterValue,
+			float64(stats.StaleConns),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.maxDesc,
+			prometheus.CounterValue,
+			float64(client.poolSize),
+			key,
+			clientType,
+		)
+	}
+}
+
+func (s *collector) registerClient(client *statGetter) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	s.clients = append(s.clients, client)
+}