Add metrics for etcd (#18962)

Signed-off-by: yun.zhang <yun.zhang@zilliz.com>

Signed-off-by: yun.zhang <yun.zhang@zilliz.com>

cmd/roles/roles.go

@@ -63,6 +63,7 @@ func init() {
 	Registry = prometheus.NewRegistry()
 	Registry.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
 	Registry.MustRegister(prometheus.NewGoCollector())
+	metrics.RegisterEtcdMetrics(Registry)
 }
 
 func stopRocksmq() {

internal/kv/etcd/etcd_kv.go

@@ -23,6 +23,8 @@ import (
 	"path"
 	"time"
 
+	"github.com/milvus-io/milvus/internal/metrics"
+
 	"github.com/milvus-io/milvus/internal/common"
 
 	"github.com/milvus-io/milvus/internal/log"
@@ -687,6 +689,7 @@ func (kv *EtcdKV) CompareVersionAndSwapBytes(key string, source int64, target []
 // CheckElapseAndWarn checks the elapsed time and warns if it is too long.
 func CheckElapseAndWarn(start time.Time, message string, fields ...zap.Field) bool {
 	elapsed := time.Since(start)
+	metrics.EtcdRequestLatency.Observe(float64(elapsed))
 	if elapsed.Milliseconds() > 2000 {
 		log.Warn(message, append([]zap.Field{zap.String("time spent", elapsed.String())}, fields...)...)
 		return true
@@ -696,6 +699,7 @@ func CheckElapseAndWarn(start time.Time, message string, fields ...zap.Field) bo
 
 func CheckValueSizeAndWarn(key string, value interface{}) bool {
 	size := binary.Size(value)
+	metrics.EtcdPutKvSize.Observe(float64(size))
 	if size > 102400 {
 		log.Warn("value size large than 100kb", zap.String("key", key), zap.Int("value_size(kb)", size/1024))
 		return true

internal/kv/etcd/etcd_kv_test.go

@@ -21,6 +21,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/milvus-io/milvus/internal/util/funcutil"
+
 	etcdkv "github.com/milvus-io/milvus/internal/kv/etcd"
 	"github.com/milvus-io/milvus/internal/util/etcd"
 	"github.com/milvus-io/milvus/internal/util/paramtable"
@@ -936,11 +938,7 @@ func TestCheckTnxStringValueSizeAndWarn(t *testing.T) {
 	ret := etcdkv.CheckTnxStringValueSizeAndWarn(kvs)
 	assert.False(t, ret)
 
-	bytes := make([]byte, 1024000)
-	for i := 0; i < 1024000; i++ {
-		bytes[i] = byte('a')
-	}
-	kvs["k"] = string(bytes)
+	kvs["k1"] = funcutil.RandomString(1024000)
 	ret = etcdkv.CheckTnxStringValueSizeAndWarn(kvs)
 	assert.True(t, ret)
 }

internal/metrics/etcd_metrics.go

@@ -0,0 +1,31 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	EtcdPutKvSize = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: milvusNamespace,
+			Subsystem: "etcd",
+			Name:      "etcd_kv_size",
+			Help:      "kv size stats",
+			Buckets:   buckets,
+		})
+
+	EtcdRequestLatency = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: milvusNamespace,
+			Subsystem: "etcd",
+			Name:      "client_request_latency",
+			Help:      "request latency on the client side ",
+			Buckets:   buckets,
+		})
+)
+
+//RegisterEtcdMetrics registers etcd metrics
+func RegisterEtcdMetrics(registry *prometheus.Registry) {
+	registry.MustRegister(EtcdPutKvSize)
+	registry.MustRegister(EtcdRequestLatency)
+}

internal/metrics/metrics_test.go

@@ -35,6 +35,7 @@ func TestRegisterMetrics(t *testing.T) {
 	RegisterProxy(r)
 	RegisterQueryNode(r)
 	RegisterQueryCoord(r)
+	RegisterEtcdMetrics(r)
 	ServeHTTP(r)
 }