Support Ruler to query Query Frontend

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

Author: SungJin1212

CHANGELOG.md

@@ -20,6 +20,7 @@
 * [FEATURE] Distributor: Add `validation.max-native-histogram-buckets` to limit max number of bucket count. Distributor will try to automatically reduce histogram resolution until it is within the bucket limit or resolution cannot be reduced anymore. #6104
 * [FEATURE] Store Gateway: Token bucket limiter. #6016
 * [FEATURE] Ruler: Add support for `query_offset` field on RuleGroup and new `ruler_query_offset` per-tenant limit. #6085
+* [FEATURE] Ruler: Add `ruler.frontend-address` and `ruler.frontend-timeout' to allow query to query frontends instead of ingesters. #6151
 * [ENHANCEMENT] rulers: Add support to persist tokens in rulers. #5987
 * [ENHANCEMENT] Query Frontend/Querier: Added store gateway postings touched count and touched size in Querier stats and log in Query Frontend. #5892
 * [ENHANCEMENT] Query Frontend/Querier: Returns `warnings` on prometheus query responses. #5916

docs/configuration/config-file-reference.md

@@ -4062,6 +4062,15 @@ The `redis_config` configures the Redis backend cache.
 The `ruler_config` configures the Cortex ruler.
 
 ```yaml
+# Address of the Query Frontend service, in host:port format. Ruler queries to
+# Query Frontends. If not set, ruler queries to Ingesters directly.
+# CLI flag: -ruler.frontend-address
+[frontend_address: <string> | default = ""]
+
+# HTTP timeout duration when querying to the Query Frontend.
+# CLI flag: -ruler.frontend-timeout
+[frontend_timeout: <duration> | default = 10s]
+
 # URL of alerts return path.
 # CLI flag: -ruler.external.url
 [external_url: <url> | default = ]

pkg/cortex/modules.go

@@ -583,10 +583,15 @@ func (t *Cortex) initRuler() (serv services.Service, err error) {
 	} else {
 		rulerRegisterer := prometheus.WrapRegistererWith(prometheus.Labels{"engine": "ruler"}, prometheus.DefaultRegisterer)
 		// TODO: Consider wrapping logger to differentiate from querier module logger
-		queryable, _, engine := querier.New(t.Cfg.Querier, t.Overrides, t.Distributor, t.StoreQueryables, rulerRegisterer, util_log.Logger)
-
-		managerFactory := ruler.DefaultTenantManagerFactory(t.Cfg.Ruler, t.Distributor, queryable, engine, t.Overrides, metrics, prometheus.DefaultRegisterer)
-		manager, err = ruler.NewDefaultMultiTenantManager(t.Cfg.Ruler, managerFactory, metrics, prometheus.DefaultRegisterer, util_log.Logger)
+		if t.Cfg.Ruler.FrontendAddress == "" {
+			queryable, _, engine := querier.New(t.Cfg.Querier, t.Overrides, t.Distributor, t.StoreQueryables, rulerRegisterer, util_log.Logger)
+			managerFactory := ruler.DefaultTenantManagerFactory(t.Cfg.Ruler, t.Distributor, queryable, engine, t.Overrides, metrics, prometheus.DefaultRegisterer)
+			manager, err = ruler.NewDefaultMultiTenantManager(t.Cfg.Ruler, managerFactory, metrics, prometheus.DefaultRegisterer, util_log.Logger)
+		} else {
+			queryable, _, _ := querier.New(t.Cfg.Querier, t.Overrides, t.Distributor, t.StoreQueryables, rulerRegisterer, util_log.Logger)
+			managerFactory := ruler.TenantManagerFactory(t.Cfg.Ruler, t.Distributor, queryable, t.Overrides, metrics)
+			manager, err = ruler.NewDefaultMultiTenantManager(t.Cfg.Ruler, managerFactory, metrics, prometheus.DefaultRegisterer, util_log.Logger)
+		}
 	}
 
 	if err != nil {

pkg/ruler/compat.go

@@ -226,6 +226,33 @@ func EngineQueryFunc(engine promql.QueryEngine, q storage.Queryable, overrides R
 	}
 }
 
+func QueryFrontendQueryFunc(promClient *promClient, overrides RulesLimits, userID string, lookbackDelta time.Duration) rules.QueryFunc {
+	return func(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
+		// Enforce the max query length.
+		maxQueryLength := overrides.MaxQueryLength(userID)
+		if maxQueryLength > 0 {
+			expr, err := parser.ParseExpr(qs)
+			// If failed to parse expression, skip checking select range.
+			// Fail the query in the engine.
+			if err == nil {
+				// Enforce query length across all selectors in the query.
+				length := promql_util.FindNonOverlapQueryLength(expr, 0, 0, lookbackDelta)
+				if length > maxQueryLength {
+					return nil, validation.LimitError(fmt.Sprintf(validation.ErrQueryTooLong, length, maxQueryLength))
+				}
+			}
+		}
+
+		evaluationDelay := overrides.EvaluationDelay(userID)
+		v, err := promClient.InstantQuery(ctx, qs, t.Add(-evaluationDelay))
+		if err != nil {
+			return nil, err
+		}
+
+		return v, nil
+	}
+}
+
 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()
@@ -366,6 +393,48 @@ func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engi
 	}
 }
 
+// TenantManagerFactory return a rule manager factory making rule manager using query frontend query func
+func TenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, overrides RulesLimits, evalMetrics *RuleEvalMetrics) ManagerFactory {
+	// Wrap errors returned by Queryable to our wrapper, so that we can distinguish between those errors
+	// and errors returned by PromQL engine. Errors from Queryable can be either caused by user (limits) or internal errors.
+	// Errors from PromQL are always "user" errors.
+	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)
+
+		promClient := NewPromClient(cfg.FrontendAddress, userID, cfg.FrontendTimeout)
+		queryFrontendQueryFunc := QueryFrontendQueryFunc(promClient, overrides, userID, cfg.LookbackDelta)
+		metricsQueryFunc := MetricsQueryFunc(queryFrontendQueryFunc, totalQueries, failedQueries)
+
+		return rules.NewManager(&rules.ManagerOptions{
+			Appendable:             NewPusherAppendable(p, userID, overrides, totalWrites, failedWrites),
+			Queryable:              q,
+			QueryFunc:              RecordAndReportRuleQueryMetrics(metricsQueryFunc, queryTime, logger),
+			Context:                user.InjectOrgID(ctx, userID),
+			ExternalURL:            cfg.ExternalURL.URL,
+			NotifyFunc:             SendAlerts(notifier, cfg.ExternalURL.URL.String()),
+			Logger:                 log.With(logger, "user", userID),
+			Registerer:             reg,
+			OutageTolerance:        cfg.OutageTolerance,
+			ForGracePeriod:         cfg.ForGracePeriod,
+			ResendDelay:            cfg.ResendDelay,
+			ConcurrentEvalsEnabled: cfg.ConcurrentEvalsEnabled,
+			MaxConcurrentEvals:     cfg.MaxConcurrentEvals,
+			DefaultRuleQueryOffset: func() time.Duration {
+				return overrides.RulerQueryOffset(userID)
+			},
+		})
+	}
+}
+
 type QueryableError struct {
 	err error
 }

pkg/ruler/prom_client.go

@@ -0,0 +1,85 @@
+package ruler
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/go-kit/log/level"
+	promapi "github.com/prometheus/client_golang/api"
+	promv1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/promql"
+
+	"github.com/cortexproject/cortex/pkg/util/spanlogger"
+)
+
+const orgIDHeader = "X-Scope-OrgId"
+
+type promClient struct {
+	promAPI promv1.API
+	timeout time.Duration
+}
+
+func NewPromClient(queryFrontendUrl, orgID string, timeout time.Duration) *promClient {
+	client, _ := promapi.NewClient(promapi.Config{
+		Address:      queryFrontendUrl + "/api/prom",
+		RoundTripper: &addOrgIDRoundTripper{orgID: orgID, next: http.DefaultTransport},
+	})
+
+	return &promClient{
+		promAPI: promv1.NewAPI(client),
+		timeout: timeout,
+	}
+}
+
+func (p *promClient) InstantQuery(ctx context.Context, qs string, t time.Time) (promql.Vector, error) {
+	log, ctx := spanlogger.New(ctx, "promClient.InstantQuery")
+	defer log.Span.Finish()
+
+	value, warns, err := p.promAPI.Query(ctx, qs, t, promv1.WithTimeout(p.timeout))
+	if err != nil {
+		level.Error(log).Log("err", err, "query", qs)
+		return nil, err
+	}
+	if len(warns) > 0 {
+		level.Warn(log).Log("warnings", strings.Join(warns, ", "), "query", qs)
+	}
+
+	switch value.Type() {
+	case model.ValVector:
+		vector := value.(model.Vector)
+		v := make([]promql.Sample, 0, len(vector))
+		for _, sample := range value.(model.Vector) {
+			metric := make([]labels.Label, 0, len(sample.Metric))
+			for k, v := range sample.Metric {
+				metric = append(metric, labels.Label{
+					Name:  string(k),
+					Value: string(v),
+				})
+			}
+			v = append(v, promql.Sample{
+				T:      sample.Timestamp.Unix(),
+				F:      float64(sample.Value),
+				Metric: metric,
+			})
+		}
+		return v, nil
+	default:
+		return nil, errors.New("rule result is not a vector")
+	}
+}
+
+type addOrgIDRoundTripper struct {
+	orgID string
+	next  http.RoundTripper
+}
+
+func (r *addOrgIDRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	req.Header.Set(orgIDHeader, r.orgID)
+
+	return r.next.RoundTrip(req)
+}

pkg/ruler/ruler.go

@@ -50,6 +50,7 @@ var (
 	errInvalidShardingStrategy   = errors.New("invalid sharding strategy")
 	errInvalidTenantShardSize    = errors.New("invalid tenant shard size, the value must be greater than 0")
 	errInvalidMaxConcurrentEvals = errors.New("invalid max concurrent evals, the value must be greater than 0")
+	errInvalidFrontendAddress    = errors.New("invalid query frontend address")
 )
 
 const (
@@ -93,6 +94,10 @@ func (e *DisabledRuleGroupErr) Error() string {
 
 // Config is the configuration for the recording rules server.
 type Config struct {
+	// This is used for query to query frontend to evaluate rules
+	FrontendAddress string `yaml:"frontend_address"`
+	// HTTP timeout duration when querying to query frontend to evaluate rules
+	FrontendTimeout time.Duration `yaml:"frontend_timeout"`
 	// This is used for template expansion in alerts; must be a valid URL.
 	ExternalURL flagext.URLValue `yaml:"external_url"`
 	// Labels to add to all alerts
@@ -167,6 +172,10 @@ func (cfg *Config) Validate(limits validation.Limits, log log.Logger) error {
 		return errors.Wrap(err, "invalid ruler gRPC client config")
 	}
 
+	if _, err := url.Parse(cfg.FrontendAddress); err != nil {
+		return errInvalidFrontendAddress
+	}
+
 	if cfg.ConcurrentEvalsEnabled && cfg.MaxConcurrentEvals <= 0 {
 		return errInvalidMaxConcurrentEvals
 	}
@@ -190,6 +199,8 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	//lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods
 	flagext.DeprecatedFlag(f, "ruler.alertmanager-use-v2", "This flag is no longer functional. V1 API is deprecated and removed", util_log.Logger)
 
+	f.StringVar(&cfg.FrontendAddress, "ruler.frontend-address", "", "Address of the Query Frontend service, in host:port format. Ruler queries to Query Frontends. If not set, ruler queries to Ingesters directly.")
+	f.DurationVar(&cfg.FrontendTimeout, "ruler.frontend-timeout", 10*time.Second, "HTTP timeout duration when querying to the Query Frontend.")
 	cfg.ExternalURL.URL, _ = url.Parse("") // Must be non-nil
 	f.Var(&cfg.ExternalURL, "ruler.external.url", "URL of alerts return path.")
 	f.DurationVar(&cfg.EvaluationInterval, "ruler.evaluation-interval", 1*time.Minute, "How frequently to evaluate rules")