Add a new config and metric for reporting ruler query execution wall time.

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## master / unreleased
+* [FEATURE] Ruler: Add new `-ruler.query-stats-enabled` which when enabled will report the `cortex_ruler_query_seconds_total` metric that tracks the sum of the wall time of executing queries in the ruler in seconds. #4317
 
 ## 1.10.0-rc.0 / 2021-06-28
 

docs/configuration/config-file-reference.md

@@ -1616,6 +1616,10 @@ ring:
 # processing will ignore them instead. Subject to sharding.
 # CLI flag: -ruler.disabled-tenants
 [disabled_tenants: <string> | default = ""]
+
+# Report the wall time for ruler queries to complete as a metric.
+# CLI flag: -ruler.query-stats-enabled
+[query_stats_enabled: <boolean> | default = false]
 ```
 
 ### `ruler_storage_config`

pkg/ruler/compat.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/prometheus/notifier"
@@ -20,6 +21,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/querier"
+	util_log "github.com/cortexproject/cortex/pkg/util/log"
 )
 
 // Pusher is an ingester server that accepts pushes.
@@ -146,9 +148,7 @@ func EngineQueryFunc(engine *promql.Engine, q storage.Queryable, overrides Rules
 func MetricsQueryFunc(qf rules.QueryFunc, queries, failedQueries prometheus.Counter) rules.QueryFunc {
 	return func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
 		queries.Inc()
-
 		result, err := qf(ctx, qs, t)
-
 		// We rely on TranslateToPromqlApiError to do its job here... it returns nil, if err is nil.
 		// It returns promql.ErrStorage, if error should be reported back as 500.
 		// Other errors it returns are either for canceled or timed-out queriers (we're not reporting those as failures),
@@ -163,6 +163,29 @@ func MetricsQueryFunc(qf rules.QueryFunc, queries, failedQueries prometheus.Coun
 	}
 }
 
+func RecordAndReportRuleQueryMetrics(qf rules.QueryFunc, queryTime prometheus.Counter, logger log.Logger) rules.QueryFunc {
+	return func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
+		// If we've been passed a counter we want to record the wall time spent executing this request.
+		if queryTime != nil {
+			timer := prometheus.NewTimer(nil)
+			defer func() {
+				querySeconds := timer.ObserveDuration().Seconds()
+				queryTime.Add(querySeconds)
+
+				// Log ruler query stats.
+				logMessage := append([]interface{}{
+					"msg", "ruler query stats",
+					"cortex_ruler_query_seconds_total", querySeconds,
+				}, qs)
+				level.Info(util_log.WithContext(ctx, logger)).Log(logMessage...)
+			}()
+		}
+
+		result, err := qf(ctx, qs, t)
+		return result, err
+	}
+}
+
 // This interface mimicks rules.Manager API. Interface is used to simplify tests.
 type RulesManager interface {
 	// Starts rules manager. Blocks until Stop is called.
@@ -199,12 +222,24 @@ func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engi
 		Name: "cortex_ruler_queries_failed_total",
 		Help: "Number of failed queries by ruler.",
 	})
+	var rulerQuerySeconds *prometheus.CounterVec
+	if cfg.EnableQueryStats {
+		rulerQuerySeconds = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "cortex_ruler_query_seconds_total",
+			Help: "Total amount of wall clock time spent processing queries by the ruler.",
+		}, []string{"user"})
+	}
 
 	return func(ctx context.Context, userID string, notifier *notifier.Manager, logger log.Logger, reg prometheus.Registerer) RulesManager {
+		var queryTime prometheus.Counter = nil
+		if rulerQuerySeconds != nil {
+			queryTime = rulerQuerySeconds.WithLabelValues(userID)
+		}
+
 		return rules.NewManager(&rules.ManagerOptions{
 			Appendable:      NewPusherAppendable(p, userID, overrides, totalWrites, failedWrites),
 			Queryable:       q,
-			QueryFunc:       MetricsQueryFunc(EngineQueryFunc(engine, q, overrides, userID), totalQueries, failedQueries),
+			QueryFunc:       RecordAndReportRuleQueryMetrics(MetricsQueryFunc(EngineQueryFunc(engine, q, overrides, userID), totalQueries, failedQueries), queryTime, logger),
 			Context:         user.InjectOrgID(ctx, userID),
 			ExternalURL:     cfg.ExternalURL.URL,
 			NotifyFunc:      SendAlerts(notifier, cfg.ExternalURL.URL.String()),

pkg/ruler/compat_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-kit/kit/log"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/common/model"
@@ -230,7 +231,6 @@ func TestMetricsQueryFuncErrors(t *testing.T) {
 			mockFunc := func(ctx context.Context, q string, t time.Time) (promql.Vector, error) {
 				return promql.Vector{}, tc.returnedError
 			}
-
 			qf := MetricsQueryFunc(mockFunc, queries, failures)
 
 			_, err := qf(context.Background(), "test", time.Now())
@@ -241,3 +241,21 @@ func TestMetricsQueryFuncErrors(t *testing.T) {
 		})
 	}
 }
+
+func TestMetricsQueryFuncMetrics(t *testing.T) {
+	queries := prometheus.NewCounter(prometheus.CounterOpts{})
+	failures := prometheus.NewCounter(prometheus.CounterOpts{})
+	queryTime := 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
+	}
+	qf := RecordAndReportRuleQueryMetrics(MetricsQueryFunc(mockFunc, queries, failures), queryTime.WithLabelValues("userID"), log.NewNopLogger())
+
+	_, _ = qf(context.Background(), "test", time.Now())
+
+	require.Equal(t, 1, int(testutil.ToFloat64(queries)))
+	require.Equal(t, 0, int(testutil.ToFloat64(failures)))
+	require.LessOrEqual(t, float64(1), testutil.ToFloat64(queryTime.WithLabelValues("userID")))
+}

pkg/ruler/ruler.go

@@ -115,6 +115,8 @@ type Config struct {
 	DisabledTenants flagext.StringSliceCSV `yaml:"disabled_tenants"`
 
 	RingCheckPeriod time.Duration `yaml:"-"`
+
+	EnableQueryStats bool `yaml:"query_stats_enabled"`
 }
 
 // Validate config and returns error on failure
@@ -173,6 +175,8 @@ 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 metric.")
+
 	cfg.RingCheckPeriod = 5 * time.Second
 }
 

pkg/ruler/ruler_test.go

@@ -60,6 +60,7 @@ func defaultRulerConfig(store rulestore.RuleStore) (Config, func()) {
 	cfg.Ring.ListenPort = 0
 	cfg.Ring.InstanceAddr = "localhost"
 	cfg.Ring.InstanceID = "localhost"
+	cfg.EnableQueryStats = false
 
 	// Create a cleanup function that will be called at the end of the test
 	cleanup := func() {