enhance: add db label for deletion and bulkinsert operation (#32611)

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

internal/datacoord/import_scheduler.go

@@ -275,6 +275,13 @@ func (s *importScheduler) processInProgressImport(task ImportTask) {
 		log.Warn("import failed", WrapTaskLog(task, zap.String("reason", resp.GetReason()))...)
 		return
 	}
+
+	collInfo := s.meta.GetCollection(task.GetCollectionID())
+	dbName := ""
+	if collInfo != nil {
+		dbName = collInfo.DatabaseName
+	}
+
 	for _, info := range resp.GetImportSegmentsInfo() {
 		segment := s.meta.GetSegment(info.GetSegmentID())
 		if info.GetImportedRows() <= segment.GetNumOfRows() {
@@ -287,7 +294,9 @@ func (s *importScheduler) processInProgressImport(task ImportTask) {
 			log.Warn("update import segment rows failed", WrapTaskLog(task, zap.Error(err))...)
 			return
 		}
+
 		metrics.DataCoordBulkVectors.WithLabelValues(
+			dbName,
 			strconv.FormatInt(task.GetCollectionID(), 10),
 		).Add(float64(diff))
 	}

internal/proxy/impl.go

@@ -2538,11 +2538,12 @@ func (node *Proxy) Delete(ctx context.Context, request *milvuspb.DeleteRequest)
 	rateCol.Add(internalpb.RateType_DMLDelete.String(), float64(receiveSize))
 
 	successCnt := dr.result.GetDeleteCnt()
-	metrics.ProxyDeleteVectors.WithLabelValues(strconv.FormatInt(paramtable.GetNodeID(), 10)).Add(float64(successCnt))
+
+	dbName := request.DbName
+	nodeID := paramtable.GetStringNodeID()
+	metrics.ProxyDeleteVectors.WithLabelValues(nodeID, dbName).Add(float64(successCnt))
 
 	username := GetCurUserFromContextOrDefault(ctx)
-	nodeID := paramtable.GetStringNodeID()
-	dbName := request.DbName
 	collectionName := request.CollectionName
 	v := Extension.Report(map[string]any{
 		hookutil.OpTypeKey:     hookutil.OpTypeDelete,

pkg/metrics/datacoord_metrics.go

@@ -103,6 +103,7 @@ var (
 			Name:      "bulk_insert_vectors_count",
 			Help:      "counter of vectors successfully bulk inserted",
 		}, []string{
+			databaseLabelName,
 			collectionIDLabelName,
 		})
 
@@ -332,7 +333,7 @@ func CleanupDataCoordNumStoredRows(collectionID int64) {
 }
 
 func CleanupDataCoordBulkInsertVectors(collectionID int64) {
-	DataCoordBulkVectors.Delete(prometheus.Labels{
+	DataCoordBulkVectors.DeletePartialMatch(prometheus.Labels{
 		collectionIDLabelName: fmt.Sprint(collectionID),
 	})
 }

pkg/metrics/proxy_metrics.go

@@ -66,7 +66,7 @@ var (
 			Subsystem: typeutil.ProxyRole,
 			Name:      "delete_vectors_count",
 			Help:      "counter of vectors successfully deleted",
-		}, []string{nodeIDLabelName})
+		}, []string{nodeIDLabelName, databaseLabelName})
 
 	// ProxySQLatency record the latency of search successfully.
 	ProxySQLatency = prometheus.NewHistogramVec(
@@ -398,6 +398,10 @@ func CleanupProxyDBMetrics(nodeID int64, dbName string) {
 		nodeIDLabelName:   strconv.FormatInt(nodeID, 10),
 		databaseLabelName: dbName,
 	})
+	ProxyDeleteVectors.DeletePartialMatch(prometheus.Labels{
+		nodeIDLabelName:   strconv.FormatInt(nodeID, 10),
+		databaseLabelName: dbName,
+	})
 	ProxySQLatency.DeletePartialMatch(prometheus.Labels{
 		nodeIDLabelName:   strconv.FormatInt(nodeID, 10),
 		databaseLabelName: dbName,