Add query metrics to ruler query stats

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

Author: SungJin1212

CHANGELOG.md

@@ -53,6 +53,7 @@
 * [ENHANCEMENT] Compactor: Introduce cleaner visit marker. #6113
 * [ENHANCEMENT] Query Frontend: Add cortex_query_samples_total metric. #6142
 * [ENHANCEMENT] Ingester: Implement metadata API limit. #6128
+* [ENHANCEMENT] Ruler: Add query statistics metrics when --ruler.query-stats-enabled=true. #6173
 * [BUGFIX] Configsdb: Fix endline issue in db password. #5920
 * [BUGFIX] Ingester: Fix `user` and `type` labels for the `cortex_ingester_tsdb_head_samples_appended_total` TSDB metric. #5952
 * [BUGFIX] Querier: Enforce max query length check for `/api/v1/series` API even though `ignoreMaxQueryLength` is set to true. #6018

docs/configuration/config-file-reference.md

@@ -4366,8 +4366,8 @@ ring:
 # CLI flag: -ruler.disabled-tenants
 [disabled_tenants: <string> | default = ""]
 
-# Report the wall time for ruler queries to complete as a per user metric and as
-# an info level log message.
+# Report query statistics for ruler queries to complete as a per user metric and
+# as an info level log message.
 # CLI flag: -ruler.query-stats-enabled
 [query_stats_enabled: <boolean> | default = false]
 

pkg/ruler/compat.go

@@ -229,19 +229,19 @@ func MetricsQueryFunc(qf rules.QueryFunc, queries, failedQueries prometheus.Coun
 	}
 }
 
-func RecordAndReportRuleQueryMetrics(qf rules.QueryFunc, queryTime prometheus.Counter, logger log.Logger) rules.QueryFunc {
-	if queryTime == nil {
-		return qf
-	}
-
+func RecordAndReportRuleQueryMetrics(qf rules.QueryFunc, queryTime, querySeries, querySample, queryChunkBytes, queryDataBytes prometheus.Counter, logger log.Logger) rules.QueryFunc {
 	return func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
 		queryStats, ctx := stats.ContextWithEmptyStats(ctx)
 		// If we've been passed a counter we want to record the wall time spent executing this request.
 		timer := prometheus.NewTimer(nil)
+
 		defer func() {
 			querySeconds := timer.ObserveDuration().Seconds()
 			queryTime.Add(querySeconds)
-
+			querySeries.Add(float64(queryStats.FetchedSeriesCount))
+			querySample.Add(float64(queryStats.FetchedSamplesCount))
+			queryChunkBytes.Add(float64(queryStats.FetchedChunkBytes))
+			queryDataBytes.Add(float64(queryStats.FetchedDataBytes))
 			// Log ruler query stats.
 			logMessage := []interface{}{
 				"msg", "query stats",
@@ -303,23 +303,29 @@ func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engi
 	q = querier.NewErrorTranslateQueryableWithFn(q, WrapQueryableErrors)
 
 	return func(ctx context.Context, userID string, notifier *notifier.Manager, logger log.Logger, reg prometheus.Registerer) RulesManager {
-		var queryTime prometheus.Counter
-		if evalMetrics.RulerQuerySeconds != nil {
-			queryTime = evalMetrics.RulerQuerySeconds.WithLabelValues(userID)
-		}
-
 		failedQueries := evalMetrics.FailedQueriesVec.WithLabelValues(userID)
 		totalQueries := evalMetrics.TotalQueriesVec.WithLabelValues(userID)
 		totalWrites := evalMetrics.TotalWritesVec.WithLabelValues(userID)
 		failedWrites := evalMetrics.FailedWritesVec.WithLabelValues(userID)
 
+		var queryFunc rules.QueryFunc
 		engineQueryFunc := EngineQueryFunc(engine, q, overrides, userID, cfg.LookbackDelta)
 		metricsQueryFunc := MetricsQueryFunc(engineQueryFunc, totalQueries, failedQueries)
+		if cfg.EnableQueryStats {
+			queryTime := evalMetrics.RulerQuerySeconds.WithLabelValues(userID)
+			querySeries := evalMetrics.RulerQuerySeries.WithLabelValues(userID)
+			querySample := evalMetrics.RulerQuerySamples.WithLabelValues(userID)
+			queryChunkBytes := evalMetrics.RulerQuerySamples.WithLabelValues(userID)
+			queryDataBytes := evalMetrics.RulerQueryDataBytes.WithLabelValues(userID)
+			queryFunc = RecordAndReportRuleQueryMetrics(metricsQueryFunc, queryTime, querySeries, querySample, queryChunkBytes, queryDataBytes, logger)
+		} else {
+			queryFunc = metricsQueryFunc
+		}
 
 		return rules.NewManager(&rules.ManagerOptions{
 			Appendable:             NewPusherAppendable(p, userID, overrides, totalWrites, failedWrites),
 			Queryable:              q,
-			QueryFunc:              RecordAndReportRuleQueryMetrics(metricsQueryFunc, queryTime, logger),
+			QueryFunc:              queryFunc,
 			Context:                user.InjectOrgID(ctx, userID),
 			ExternalURL:            cfg.ExternalURL.URL,
 			NotifyFunc:             SendAlerts(notifier, cfg.ExternalURL.URL.String()),

pkg/ruler/compat_test.go

@@ -393,13 +393,33 @@ func TestMetricsQueryFuncErrors(t *testing.T) {
 
 func TestRecordAndReportRuleQueryMetrics(t *testing.T) {
 	queryTime := prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user"})
+	querySeries := prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user"})
+	querySample := prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user"})
+	queryChunkBytes := prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user"})
+	queryDataBytes := prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user"})
 
 	mockFunc := func(ctx context.Context, q string, t time.Time) (promql.Vector, error) {
 		time.Sleep(1 * time.Second)
-		return promql.Vector{}, nil
+		return promql.Vector{
+			promql.Sample{
+				T:      1,
+				F:      1,
+				Metric: labels.FromStrings("a", "a1", "b", "b1"),
+			},
+			promql.Sample{
+				T:      2,
+				F:      2,
+				Metric: labels.FromStrings("a", "a2", "b", "b2"),
+			},
+		}, nil
 	}
-	qf := RecordAndReportRuleQueryMetrics(mockFunc, queryTime.WithLabelValues("userID"), log.NewNopLogger())
+	qf := RecordAndReportRuleQueryMetrics(mockFunc, queryTime.WithLabelValues("userID"), querySeries.WithLabelValues("userID"), querySample.WithLabelValues("userID"), queryChunkBytes.WithLabelValues("userID"), queryDataBytes.WithLabelValues("userID"), log.NewNopLogger())
 	_, _ = qf(context.Background(), "test", time.Now())
 
 	require.GreaterOrEqual(t, testutil.ToFloat64(queryTime.WithLabelValues("userID")), float64(1))
+	fmt.Println("testutil.ToFloat64(querySeries.WithLabelValues(\"userID\"))", testutil.ToFloat64(querySeries.WithLabelValues("userID")))
+	require.Equal(t, testutil.ToFloat64(querySeries.WithLabelValues("userID")), float64(2))
+	require.Equal(t, testutil.ToFloat64(querySample.WithLabelValues("userID")), float64(2))
+	fmt.Println(testutil.ToFloat64(queryChunkBytes.WithLabelValues("userID")))
+	fmt.Println(testutil.ToFloat64(queryDataBytes.WithLabelValues("userID")))
 }

pkg/ruler/manager_metrics.go

@@ -225,11 +225,15 @@ func (m *ManagerMetrics) Collect(out chan<- prometheus.Metric) {
 }
 
 type RuleEvalMetrics struct {
-	TotalWritesVec    *prometheus.CounterVec
-	FailedWritesVec   *prometheus.CounterVec
-	TotalQueriesVec   *prometheus.CounterVec
-	FailedQueriesVec  *prometheus.CounterVec
-	RulerQuerySeconds *prometheus.CounterVec
+	TotalWritesVec       *prometheus.CounterVec
+	FailedWritesVec      *prometheus.CounterVec
+	TotalQueriesVec      *prometheus.CounterVec
+	FailedQueriesVec     *prometheus.CounterVec
+	RulerQuerySeconds    *prometheus.CounterVec
+	RulerQuerySeries     *prometheus.CounterVec
+	RulerQuerySamples    *prometheus.CounterVec
+	RulerQueryChunkBytes *prometheus.CounterVec
+	RulerQueryDataBytes  *prometheus.CounterVec
 }
 
 func NewRuleEvalMetrics(cfg Config, reg prometheus.Registerer) *RuleEvalMetrics {
@@ -256,6 +260,22 @@ func NewRuleEvalMetrics(cfg Config, reg prometheus.Registerer) *RuleEvalMetrics
 			Name: "cortex_ruler_query_seconds_total",
 			Help: "Total amount of wall clock time spent processing queries by the ruler.",
 		}, []string{"user"})
+		m.RulerQuerySeries = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "cortex_ruler_fetched_series_total",
+			Help: "Number of series fetched to execute a query by the ruler.",
+		}, []string{"user"})
+		m.RulerQuerySamples = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "cortex_ruler_samples_total",
+			Help: "Number of samples fetched to execute a query by the ruler.",
+		}, []string{"user"})
+		m.RulerQueryChunkBytes = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "cortex_ruler_fetched_chunks_bytes_total",
+			Help: "Size of all chunks fetched to execute a query in bytes by the ruler.",
+		}, []string{"user"})
+		m.RulerQueryDataBytes = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "cortex_ruler_fetched_data_bytes_total",
+			Help: "Size of all data fetched to execute a query in bytes by the ruler.",
+		}, []string{"user"})
 	}
 
 	return m
@@ -270,4 +290,16 @@ func (m *RuleEvalMetrics) deletePerUserMetrics(userID string) {
 	if m.RulerQuerySeconds != nil {
 		m.RulerQuerySeconds.DeleteLabelValues(userID)
 	}
+	if m.RulerQuerySeries != nil {
+		m.RulerQuerySeries.DeleteLabelValues(userID)
+	}
+	if m.RulerQuerySamples != nil {
+		m.RulerQuerySamples.DeleteLabelValues(userID)
+	}
+	if m.RulerQueryChunkBytes != nil {
+		m.RulerQueryChunkBytes.DeleteLabelValues(userID)
+	}
+	if m.RulerQueryDataBytes != nil {
+		m.RulerQueryDataBytes.DeleteLabelValues(userID)
+	}
 }

pkg/ruler/manager_metrics_test.go

@@ -574,8 +574,16 @@ func TestRuleEvalMetricsDeletePerUserMetrics(t *testing.T) {
 	m.FailedQueriesVec.WithLabelValues("fake2").Add(10)
 	m.RulerQuerySeconds.WithLabelValues("fake1").Add(10)
 	m.RulerQuerySeconds.WithLabelValues("fake2").Add(10)
-
-	metricNames := []string{"cortex_ruler_write_requests_total", "cortex_ruler_write_requests_failed_total", "cortex_ruler_queries_total", "cortex_ruler_queries_failed_total", "cortex_ruler_query_seconds_total"}
+	m.RulerQuerySeries.WithLabelValues("fake1").Add(10)
+	m.RulerQuerySeries.WithLabelValues("fake2").Add(10)
+	m.RulerQuerySamples.WithLabelValues("fake1").Add(10)
+	m.RulerQuerySamples.WithLabelValues("fake2").Add(10)
+	m.RulerQueryChunkBytes.WithLabelValues("fake1").Add(10)
+	m.RulerQueryChunkBytes.WithLabelValues("fake2").Add(10)
+	m.RulerQueryDataBytes.WithLabelValues("fake1").Add(10)
+	m.RulerQueryDataBytes.WithLabelValues("fake2").Add(10)
+
+	metricNames := []string{"cortex_ruler_write_requests_total", "cortex_ruler_write_requests_failed_total", "cortex_ruler_queries_total", "cortex_ruler_queries_failed_total", "cortex_ruler_query_seconds_total", "cortex_ruler_fetched_series_total", "cortex_ruler_samples_total", "cortex_ruler_fetched_chunks_bytes_total", "cortex_ruler_fetched_data_bytes_total"}
 	gm, err := reg.Gather()
 	require.NoError(t, err)
 	mfm, err := util.NewMetricFamilyMap(gm)

pkg/ruler/ruler.go

@@ -217,7 +217,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.Var(&cfg.EnabledTenants, "ruler.enabled-tenants", "Comma separated list of tenants whose rules this ruler can evaluate. If specified, only these tenants will be handled by ruler, otherwise this ruler can process rules from all tenants. Subject to sharding.")
 	f.Var(&cfg.DisabledTenants, "ruler.disabled-tenants", "Comma separated list of tenants whose rules this ruler cannot evaluate. If specified, a ruler that would normally pick the specified tenant(s) for processing will ignore them instead. Subject to sharding.")
 
-	f.BoolVar(&cfg.EnableQueryStats, "ruler.query-stats-enabled", false, "Report the wall time for ruler queries to complete as a per user metric and as an info level log message.")
+	f.BoolVar(&cfg.EnableQueryStats, "ruler.query-stats-enabled", false, "Report query statistics for ruler queries to complete as a per user metric and as an info level log message.")
 	f.BoolVar(&cfg.DisableRuleGroupLabel, "ruler.disable-rule-group-label", false, "Disable the rule_group label on exported metrics")
 
 	cfg.RingCheckPeriod = 5 * time.Second