Add receivedNQ metric and change receivedBytes metric to collection level (#24199)

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>

internal/proxy/impl.go

@@ -2115,7 +2115,9 @@ func (node *Proxy) Insert(ctx context.Context, request *milvuspb.InsertRequest)
 	}
 	method := "Insert"
 	tr := timerecord.NewTimeRecorder(method)
-	metrics.ProxyReceiveBytes.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10), metrics.InsertLabel).Add(float64(proto.Size(request)))
+	metrics.ProxyReceiveBytes.WithLabelValues(
+		strconv.FormatInt(paramtable.GetNodeID(), 10),
+		metrics.InsertLabel, request.GetCollectionName()).Add(float64(proto.Size(request)))
 	metrics.ProxyFunctionCall.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10), method, metrics.TotalLabel).Inc()
 
 	it := &insertTask{
@@ -2232,7 +2234,9 @@ func (node *Proxy) Delete(ctx context.Context, request *milvuspb.DeleteRequest)
 	log.Debug("Start processing delete request in Proxy")
 	defer log.Debug("Finish processing delete request in Proxy")
 
-	metrics.ProxyReceiveBytes.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10), metrics.DeleteLabel).Add(float64(proto.Size(request)))
+	metrics.ProxyReceiveBytes.WithLabelValues(
+		strconv.FormatInt(paramtable.GetNodeID(), 10),
+		metrics.DeleteLabel, request.GetCollectionName()).Add(float64(proto.Size(request)))
 
 	if !node.checkHealthy() {
 		return &milvuspb.MutationResult{
@@ -2342,7 +2346,9 @@ func (node *Proxy) Upsert(ctx context.Context, request *milvuspb.UpsertRequest)
 	method := "Upsert"
 	tr := timerecord.NewTimeRecorder(method)
 
-	metrics.ProxyReceiveBytes.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10), metrics.UpsertLabel).Add(float64(proto.Size(request)))
+	metrics.ProxyReceiveBytes.WithLabelValues(
+		strconv.FormatInt(paramtable.GetNodeID(), 10),
+		metrics.UpsertLabel, request.GetCollectionName()).Add(float64(proto.Size(request)))
 	metrics.ProxyFunctionCall.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10), method, metrics.TotalLabel).Inc()
 
 	it := &upsertTask{
@@ -2462,7 +2468,13 @@ func (node *Proxy) Upsert(ctx context.Context, request *milvuspb.UpsertRequest)
 // Search search the most similar records of requests.
 func (node *Proxy) Search(ctx context.Context, request *milvuspb.SearchRequest) (*milvuspb.SearchResults, error) {
 	receiveSize := proto.Size(request)
-	metrics.ProxyReceiveBytes.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10), metrics.SearchLabel).Add(float64(receiveSize))
+	metrics.ProxyReceiveBytes.WithLabelValues(
+		strconv.FormatInt(paramtable.GetNodeID(), 10),
+		metrics.SearchLabel, request.GetCollectionName()).Add(float64(receiveSize))
+
+	metrics.ProxyReceivedNQ.WithLabelValues(
+		strconv.FormatInt(paramtable.GetNodeID(), 10),
+		metrics.SearchLabel, request.GetCollectionName()).Add(float64(request.GetNq()))
 
 	rateCol.Add(internalpb.RateType_DQLSearch.String(), float64(request.GetNq()))
 
@@ -2657,7 +2669,13 @@ func (node *Proxy) Flush(ctx context.Context, request *milvuspb.FlushRequest) (*
 // Query get the records by primary keys.
 func (node *Proxy) Query(ctx context.Context, request *milvuspb.QueryRequest) (*milvuspb.QueryResults, error) {
 	receiveSize := proto.Size(request)
-	metrics.ProxyReceiveBytes.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10), metrics.QueryLabel).Add(float64(receiveSize))
+	metrics.ProxyReceiveBytes.WithLabelValues(
+		strconv.FormatInt(paramtable.GetNodeID(), 10),
+		metrics.QueryLabel, request.GetCollectionName()).Add(float64(receiveSize))
+
+	metrics.ProxyReceivedNQ.WithLabelValues(
+		strconv.FormatInt(paramtable.GetNodeID(), 10),
+		metrics.SearchLabel, request.GetCollectionName()).Add(float64(1))
 
 	rateCol.Add(internalpb.RateType_DQLQuery.String(), 1)
 

pkg/metrics/proxy_metrics.go

@@ -25,6 +25,14 @@ import (
 )
 
 var (
+	ProxyReceivedNQ = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: milvusNamespace,
+			Subsystem: typeutil.ProxyRole,
+			Name:      "received_nq",
+			Help:      "counter of nq of received search and query requests",
+		}, []string{nodeIDLabelName, queryTypeLabelName, collectionName})
+
 	// ProxySearchVectors record the number of vectors search successfully.
 	ProxySearchVectors = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
@@ -204,7 +212,7 @@ var (
 			Subsystem: typeutil.ProxyRole,
 			Name:      "receive_bytes_count",
 			Help:      "count of bytes received  from sdk",
-		}, []string{nodeIDLabelName, msgTypeLabelName})
+		}, []string{nodeIDLabelName, msgTypeLabelName, collectionName})
 
 	// ProxyReadReqSendBytes record the bytes sent back to client by Proxy
 	ProxyReadReqSendBytes = prometheus.NewCounterVec(
@@ -243,6 +251,7 @@ var (
 
 // RegisterProxy registers Proxy metrics
 func RegisterProxy(registry *prometheus.Registry) {
+	registry.MustRegister(ProxyReceivedNQ)
 	registry.MustRegister(ProxySearchVectors)
 	registry.MustRegister(ProxyInsertVectors)
 
@@ -286,4 +295,18 @@ func CleanupCollectionMetrics(nodeID int64, collection string) {
 		msgTypeLabelName: InsertLabel, collectionName: collection})
 	ProxyCollectionMutationLatency.Delete(prometheus.Labels{nodeIDLabelName: strconv.FormatInt(nodeID, 10),
 		msgTypeLabelName: DeleteLabel, collectionName: collection})
+	ProxyReceivedNQ.Delete(prometheus.Labels{nodeIDLabelName: strconv.FormatInt(nodeID, 10),
+		queryTypeLabelName: SearchLabel, collectionName: collection})
+	ProxyReceivedNQ.Delete(prometheus.Labels{nodeIDLabelName: strconv.FormatInt(nodeID, 10),
+		queryTypeLabelName: QueryLabel, collectionName: collection})
+	ProxyReceiveBytes.Delete(prometheus.Labels{nodeIDLabelName: strconv.FormatInt(nodeID, 10),
+		msgTypeLabelName: SearchLabel, collectionName: collection})
+	ProxyReceiveBytes.Delete(prometheus.Labels{nodeIDLabelName: strconv.FormatInt(nodeID, 10),
+		msgTypeLabelName: QueryLabel, collectionName: collection})
+	ProxyReceiveBytes.Delete(prometheus.Labels{nodeIDLabelName: strconv.FormatInt(nodeID, 10),
+		msgTypeLabelName: InsertLabel, collectionName: collection})
+	ProxyReceiveBytes.Delete(prometheus.Labels{nodeIDLabelName: strconv.FormatInt(nodeID, 10),
+		msgTypeLabelName: DeleteLabel, collectionName: collection})
+	ProxyReceiveBytes.Delete(prometheus.Labels{nodeIDLabelName: strconv.FormatInt(nodeID, 10),
+		msgTypeLabelName: UpsertLabel, collectionName: collection})
 }