Add binlogSize and ttDelay metrics monitor (milvus-io#19872)

Signed-off-by: bigsheeper <[email protected]>

Signed-off-by: bigsheeper <[email protected]>

Author: bigsheeper

configs/milvus.yaml

@@ -449,7 +449,7 @@ quotaAndLimits:
     diskProtection:
       # When the total file size of object storage is greater than `diskQuota`, all dml requests would be rejected;
       enabled: true
-      diskQuota: -1 # GB, (0, +inf), default no limit
+      diskQuota: -1 # MB, (0, +inf), default no limit
 
   # limitReading decides whether dql requests are allowed.
   limitReading:

internal/datacoord/meta.go

@@ -205,6 +205,7 @@ func (m *meta) GetTotalBinlogSize() int64 {
 	for _, segment := range segments {
 		ret += segment.getSegmentSize()
 	}
+	metrics.DataCoordStoredBinlogSize.WithLabelValues().Set(float64(ret))
 	return ret
 }
 

internal/metrics/datacoord_metrics.go

@@ -84,6 +84,14 @@ var (
 			Help:      "synchronized unix epoch per physical channel",
 		}, []string{channelNameLabelName})
 
+	DataCoordStoredBinlogSize = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: milvusNamespace,
+			Subsystem: typeutil.DataCoordRole,
+			Name:      "stored_binlog_size",
+			Help:      "binlog size of all collections/segments",
+		}, []string{})
+
 	/* hard to implement, commented now
 	DataCoordSegmentSizeRatio = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -139,4 +147,5 @@ func RegisterDataCoord(registry *prometheus.Registry) {
 	registry.MustRegister(DataCoordNumStoredRows)
 	registry.MustRegister(DataCoordNumStoredRowsCounter)
 	registry.MustRegister(DataCoordSyncEpoch)
+	registry.MustRegister(DataCoordStoredBinlogSize)
 }

internal/metrics/rootcoord_metrics.go

@@ -133,6 +133,17 @@ var (
 			Name:      "num_of_roles",
 			Help:      "The number of roles",
 		})
+
+	// RootCoordTtDelay records the max time tick delay of flow graphs in DataNodes and QueryNodes.
+	RootCoordTtDelay = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: milvusNamespace,
+			Subsystem: typeutil.RootCoordRole,
+			Name:      "time_tick_delay",
+			Help:      "The max time tick delay of flow graphs",
+		}, []string{
+			nodeIDLabelName,
+		})
 )
 
 //RegisterRootCoord registers RootCoord metrics
@@ -163,4 +174,5 @@ func RegisterRootCoord(registry *prometheus.Registry) {
 	registry.MustRegister(RootCoordNumOfCredentials)
 
 	registry.MustRegister(RootCoordNumOfRoles)
+	registry.MustRegister(RootCoordTtDelay)
 }

internal/rootcoord/quota_center.go

@@ -19,6 +19,7 @@ package rootcoord
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"time"
 
@@ -27,14 +28,14 @@ import (
 
 	"github.com/milvus-io/milvus-proto/go-api/commonpb"
 	"github.com/milvus-io/milvus/internal/log"
+	"github.com/milvus-io/milvus/internal/metrics"
 	"github.com/milvus-io/milvus/internal/proto/internalpb"
 	"github.com/milvus-io/milvus/internal/proto/proxypb"
 	"github.com/milvus-io/milvus/internal/tso"
 	"github.com/milvus-io/milvus/internal/types"
 	"github.com/milvus-io/milvus/internal/util/metricsinfo"
 	"github.com/milvus-io/milvus/internal/util/ratelimitutil"
 	"github.com/milvus-io/milvus/internal/util/tsoutil"
-	"github.com/milvus-io/milvus/internal/util/typeutil"
 )
 
 const (
@@ -86,9 +87,9 @@ type QuotaCenter struct {
 	dataCoord  types.DataCoord
 
 	// metrics
-	queryNodeMetrics []*metricsinfo.QueryNodeQuotaMetrics
-	dataNodeMetrics  []*metricsinfo.DataNodeQuotaMetrics
-	proxyMetrics     []*metricsinfo.ProxyQuotaMetrics
+	queryNodeMetrics map[UniqueID]*metricsinfo.QueryNodeQuotaMetrics
+	dataNodeMetrics  map[UniqueID]*metricsinfo.DataNodeQuotaMetrics
+	proxyMetrics     map[UniqueID]*metricsinfo.ProxyQuotaMetrics
 	dataCoordMetrics *metricsinfo.DataCoordQuotaMetrics
 
 	currentRates map[internalpb.RateType]Limit
@@ -152,9 +153,9 @@ func (q *QuotaCenter) stop() {
 
 //  clearMetrics removes all metrics stored in QuotaCenter.
 func (q *QuotaCenter) clearMetrics() {
-	q.dataNodeMetrics = make([]*metricsinfo.DataNodeQuotaMetrics, 0)
-	q.queryNodeMetrics = make([]*metricsinfo.QueryNodeQuotaMetrics, 0)
-	q.proxyMetrics = make([]*metricsinfo.ProxyQuotaMetrics, 0)
+	q.dataNodeMetrics = make(map[UniqueID]*metricsinfo.DataNodeQuotaMetrics, 0)
+	q.queryNodeMetrics = make(map[UniqueID]*metricsinfo.QueryNodeQuotaMetrics, 0)
+	q.proxyMetrics = make(map[UniqueID]*metricsinfo.ProxyQuotaMetrics, 0)
 }
 
 // syncMetrics sends GetMetrics requests to DataCoord and QueryCoord to sync the metrics in DataNodes and QueryNodes.
@@ -185,7 +186,7 @@ func (q *QuotaCenter) syncMetrics() error {
 		}
 		for _, queryNodeMetric := range queryCoordTopology.Cluster.ConnectedNodes {
 			if queryNodeMetric.QuotaMetrics != nil {
-				q.queryNodeMetrics = append(q.queryNodeMetrics, queryNodeMetric.QuotaMetrics)
+				q.queryNodeMetrics[queryNodeMetric.ID] = queryNodeMetric.QuotaMetrics
 			}
 		}
 		return nil
@@ -206,7 +207,7 @@ func (q *QuotaCenter) syncMetrics() error {
 		}
 		for _, dataNodeMetric := range dataCoordTopology.Cluster.ConnectedNodes {
 			if dataNodeMetric.QuotaMetrics != nil {
-				q.dataNodeMetrics = append(q.dataNodeMetrics, dataNodeMetric.QuotaMetrics)
+				q.dataNodeMetrics[dataNodeMetric.ID] = dataNodeMetric.QuotaMetrics
 			}
 		}
 		if dataCoordTopology.Cluster.Self.QuotaMetrics != nil {
@@ -228,7 +229,7 @@ func (q *QuotaCenter) syncMetrics() error {
 				return err
 			}
 			if proxyMetric.QuotaMetrics != nil {
-				q.proxyMetrics = append(q.proxyMetrics, proxyMetric.QuotaMetrics)
+				q.proxyMetrics[proxyMetric.ID] = proxyMetric.QuotaMetrics
 			}
 		}
 		return nil
@@ -339,10 +340,11 @@ func (q *QuotaCenter) calculateWriteRates() error {
 	}
 	log.Debug("QuotaCenter check diskQuota done", zap.Bool("exceeded", exceeded))
 
-	ttFactor, err := q.timeTickDelay()
+	ts, err := q.tsoAllocator.GenerateTSO(1)
 	if err != nil {
 		return err
 	}
+	ttFactor := q.timeTickDelay(ts)
 	if ttFactor <= 0 {
 		q.forceDenyWriting(TimeTickLongDelay) // tt protection
 		return nil
@@ -409,43 +411,46 @@ func (q *QuotaCenter) resetCurrentRates() {
 
 // timeTickDelay gets time tick delay of DataNodes and QueryNodes,
 // and return the factor according to max tolerable time tick delay.
-func (q *QuotaCenter) timeTickDelay() (float64, error) {
+func (q *QuotaCenter) timeTickDelay(ts Timestamp) float64 {
+	t1, _ := tsoutil.ParseTS(ts)
+
+	var maxDelay time.Duration
+	for nodeID, metric := range q.queryNodeMetrics {
+		if metric.Fgm.NumFlowGraph > 0 {
+			t2, _ := tsoutil.ParseTS(metric.Fgm.MinFlowGraphTt)
+			delay := t1.Sub(t2)
+			if delay.Nanoseconds() > maxDelay.Nanoseconds() {
+				maxDelay = delay
+			}
+			metrics.RootCoordTtDelay.WithLabelValues(strconv.FormatInt(nodeID, 10)).Set(float64(maxDelay.Milliseconds()))
+		}
+	}
+	for nodeID, metric := range q.dataNodeMetrics {
+		if metric.Fgm.NumFlowGraph > 0 {
+			t2, _ := tsoutil.ParseTS(metric.Fgm.MinFlowGraphTt)
+			delay := t1.Sub(t2)
+			if delay.Nanoseconds() > maxDelay.Nanoseconds() {
+				maxDelay = delay
+			}
+			metrics.RootCoordTtDelay.WithLabelValues(strconv.FormatInt(nodeID, 10)).Set(float64(maxDelay.Milliseconds()))
+		}
+	}
+
 	if !Params.QuotaConfig.TtProtectionEnabled {
-		return 1, nil
+		return 1
 	}
 
 	maxTt := Params.QuotaConfig.MaxTimeTickDelay
 	if maxTt < 0 {
 		// < 0 means disable tt protection
-		return 1, nil
+		return 1
 	}
 
-	minTs := typeutil.MaxTimestamp
-	for _, metric := range q.queryNodeMetrics {
-		if metric.Fgm.NumFlowGraph > 0 && metric.Fgm.MinFlowGraphTt < minTs {
-			minTs = metric.Fgm.MinFlowGraphTt
-		}
-	}
-	for _, metric := range q.dataNodeMetrics {
-		if metric.Fgm.NumFlowGraph > 0 && metric.Fgm.MinFlowGraphTt < minTs {
-			minTs = metric.Fgm.MinFlowGraphTt
-		}
-	}
-	ts, err := q.tsoAllocator.GenerateTSO(1)
-	if err != nil {
-		return 0, err
-	}
-	if minTs >= ts {
-		return 1, nil
-	}
-	t1, _ := tsoutil.ParseTS(minTs)
-	t2, _ := tsoutil.ParseTS(ts)
-	delay := t2.Sub(t1)
-	log.Debug("QuotaCenter check timeTick delay", zap.Time("minTs", t1), zap.Time("curTs", t2), zap.Duration("delay", delay))
-	if delay.Nanoseconds() >= maxTt.Nanoseconds() {
-		return 0, nil
+	log.Debug("QuotaCenter check timeTick delay", zap.Time("curTs", t1), zap.Duration("maxDelay", maxDelay))
+	if maxDelay.Nanoseconds() >= maxTt.Nanoseconds() {
+		return 0
 	}
-	return float64(maxTt.Nanoseconds()-delay.Nanoseconds()) / float64(maxTt.Nanoseconds()), nil
+	return float64(maxTt.Nanoseconds()-maxDelay.Nanoseconds()) / float64(maxTt.Nanoseconds())
 }
 
 // checkNQInQuery checks search&query nq in QueryNode,