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

@@ -548,6 +548,7 @@ func (t *Cortex) initRuler() (serv services.Service, err error) {
 	}
 
 	t.Cfg.Ruler.LookbackDelta = t.Cfg.Querier.LookbackDelta
+	t.Cfg.Ruler.PrometheusHTTPPrefix = t.Cfg.API.PrometheusHTTPPrefix
 	t.Cfg.Ruler.Ring.ListenPort = t.Cfg.Server.GRPCListenPort
 	metrics := ruler.NewRuleEvalMetrics(t.Cfg.Ruler, prometheus.DefaultRegisterer)
 

pkg/ruler/compat.go

@@ -185,7 +185,7 @@ type RulesLimits interface {
 // EngineQueryFunc returns a new engine query function by passing an altered timestamp.
 // Modified from Prometheus rules.EngineQueryFunc
 // https://github.com/prometheus/prometheus/blob/v2.39.1/rules/manager.go#L189.
-func EngineQueryFunc(engine promql.QueryEngine, q storage.Queryable, overrides RulesLimits, userID string, lookbackDelta time.Duration) rules.QueryFunc {
+func EngineQueryFunc(engine promql.QueryEngine, promClient *promClient, q storage.Queryable, 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)
@@ -203,25 +203,34 @@ func EngineQueryFunc(engine promql.QueryEngine, q storage.Queryable, overrides R
 		}
 
 		evaluationDelay := overrides.EvaluationDelay(userID)
-		q, err := engine.NewInstantQuery(ctx, q, nil, qs, t.Add(-evaluationDelay))
-		if err != nil {
-			return nil, err
-		}
-		res := q.Exec(ctx)
-		if res.Err != nil {
-			return nil, res.Err
-		}
-		switch v := res.Value.(type) {
-		case promql.Vector:
+		if promClient != nil {
+			v, err := promClient.InstantQuery(ctx, qs, t.Add(-evaluationDelay))
+			if err != nil {
+				return nil, err
+			}
+
 			return v, nil
-		case promql.Scalar:
-			return promql.Vector{promql.Sample{
-				T:      v.T,
-				F:      v.V,
-				Metric: labels.Labels{},
-			}}, nil
-		default:
-			return nil, errors.New("rule result is not a vector or scalar")
+		} else {
+			q, err := engine.NewInstantQuery(ctx, q, nil, qs, t.Add(-evaluationDelay))
+			if err != nil {
+				return nil, err
+			}
+			res := q.Exec(ctx)
+			if res.Err != nil {
+				return nil, res.Err
+			}
+			switch v := res.Value.(type) {
+			case promql.Vector:
+				return v, nil
+			case promql.Scalar:
+				return promql.Vector{promql.Sample{
+					T:      v.T,
+					F:      v.V,
+					Metric: labels.Labels{},
+				}}, nil
+			default:
+				return nil, errors.New("rule result is not a vector or scalar")
+			}
 		}
 	}
 }
@@ -342,7 +351,8 @@ func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engi
 		totalWrites := evalMetrics.TotalWritesVec.WithLabelValues(userID)
 		failedWrites := evalMetrics.FailedWritesVec.WithLabelValues(userID)
 
-		engineQueryFunc := EngineQueryFunc(engine, q, overrides, userID, cfg.LookbackDelta)
+		promClient := NewPromClient(cfg.FrontendAddress, cfg.PrometheusHTTPPrefix, userID, cfg.FrontendTimeout)
+		engineQueryFunc := EngineQueryFunc(engine, promClient, q, overrides, userID, cfg.LookbackDelta)
 		metricsQueryFunc := MetricsQueryFunc(engineQueryFunc, totalQueries, failedQueries)
 
 		return rules.NewManager(&rules.ManagerOptions{

pkg/ruler/prom_client.go

@@ -0,0 +1,91 @@
+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, prometheusHTTPPrefix, orgID string, timeout time.Duration) *promClient {
+	if queryFrontendUrl == "" {
+		return nil
+	}
+	client, err := promapi.NewClient(promapi.Config{
+		Address:      queryFrontendUrl + prometheusHTTPPrefix,
+		RoundTripper: &addOrgIDRoundTripper{orgID: orgID, next: http.DefaultTransport},
+	})
+	if err != nil {
+		// use distributor queryable instead
+		return nil
+	}
+
+	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/prom_client_test.go

@@ -0,0 +1,49 @@
+package ruler
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewPromClient(t *testing.T) {
+	prometheusHTTPPrefix := "/prometheus"
+	timeout := time.Second * 10
+	orgId := "user"
+
+	tests := []struct {
+		queryFrontendUrl string
+		isNil            bool
+		description      string
+	}{
+		{
+			queryFrontendUrl: "",
+			isNil:            true,
+			description:      "empty string of query frontend url should return nil",
+		},
+		{
+			queryFrontendUrl: "http://\tlocalhost:3333",
+			isNil:            true,
+			description:      "invalid url of query frontend url should return nil",
+		},
+		{
+			queryFrontendUrl: "service.namespace:80",
+			isNil:            false,
+			description:      "k8s service address should return not nil",
+		},
+		{
+			queryFrontendUrl: "service.namespace.svc.cluster.local",
+			isNil:            false,
+			description:      "k8s service address should return not nil",
+		},
+	}
+
+	for _, test := range tests {
+		if test.isNil {
+			require.Nil(t, NewPromClient(test.queryFrontendUrl, prometheusHTTPPrefix, orgId, timeout), test.description)
+		} else {
+			require.NotNil(t, NewPromClient(test.queryFrontendUrl, prometheusHTTPPrefix, orgId, timeout), test.description)
+		}
+	}
+}

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
@@ -147,7 +152,8 @@ type Config struct {
 	RingCheckPeriod time.Duration `yaml:"-"`
 
 	// Field will be populated during runtime.
-	LookbackDelta time.Duration `yaml:"-"`
+	LookbackDelta        time.Duration `yaml:"-"`
+	PrometheusHTTPPrefix string        `yaml:"-"`
 
 	EnableQueryStats      bool `yaml:"query_stats_enabled"`
 	DisableRuleGroupLabel bool `yaml:"disable_rule_group_label"`
@@ -167,6 +173,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 +200,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")