Support Ruler to query Query Frontend

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

Author: SungJin1212

CHANGELOG.md

@@ -28,6 +28,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

@@ -4063,6 +4063,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

@@ -157,7 +157,7 @@ type RulesLimits interface {
 // EngineQueryFunc returns a new engine query function validating max queryLength.
 // 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)
@@ -174,25 +174,34 @@ func EngineQueryFunc(engine promql.QueryEngine, q storage.Queryable, overrides R
 			}
 		}
 
-		q, err := engine.NewInstantQuery(ctx, q, nil, qs, t)
-		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)
+			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)
+			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")
+			}
 		}
 	}
 }
@@ -294,16 +303,18 @@ type RulesManager interface {
 }
 
 // ManagerFactory is a function that creates new RulesManager for given user and notifier.Manager.
-type ManagerFactory func(ctx context.Context, userID string, notifier *notifier.Manager, logger log.Logger, reg prometheus.Registerer) RulesManager
+type ManagerFactory func(ctx context.Context, userID string, notifier *notifier.Manager, logger log.Logger, reg prometheus.Registerer) (RulesManager, error)
 
 func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engine promql.QueryEngine, overrides RulesLimits, evalMetrics *RuleEvalMetrics, reg prometheus.Registerer) 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 {
+	return func(ctx context.Context, userID string, notifier *notifier.Manager, logger log.Logger, reg prometheus.Registerer) (RulesManager, error) {
 		var queryTime prometheus.Counter
+		var promClient *promClient
+		var err error
 		if evalMetrics.RulerQuerySeconds != nil {
 			queryTime = evalMetrics.RulerQuerySeconds.WithLabelValues(userID)
 		}
@@ -313,7 +324,13 @@ 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)
+		if cfg.FrontendAddress != "" {
+			promClient, err = NewPromClient(cfg.FrontendAddress, cfg.PrometheusHTTPPrefix, userID, cfg.FrontendTimeout)
+			if err != nil {
+				return nil, err
+			}
+		}
+		engineQueryFunc := EngineQueryFunc(engine, promClient, q, overrides, userID, cfg.LookbackDelta)
 		metricsQueryFunc := MetricsQueryFunc(engineQueryFunc, totalQueries, failedQueries)
 
 		return rules.NewManager(&rules.ManagerOptions{
@@ -333,7 +350,7 @@ func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engi
 			DefaultRuleQueryOffset: func() time.Duration {
 				return overrides.RulerQueryOffset(userID)
 			},
-		})
+		}), nil
 	}
 }
 

pkg/ruler/manager.go

@@ -269,7 +269,7 @@ func (r *DefaultMultiTenantManager) newManager(ctx context.Context, userID strin
 		return nil, err
 	}
 
-	return r.managerFactory(ctx, userID, notifier, r.logger, reg), nil
+	return r.managerFactory(ctx, userID, notifier, r.logger, reg)
 }
 
 func (r *DefaultMultiTenantManager) removeNotifier(userID string) {

pkg/ruler/manager_test.go

@@ -304,13 +304,13 @@ func getManager(m *DefaultMultiTenantManager, user string) RulesManager {
 }
 
 func RuleManagerFactory(groupsToReturn [][]*promRules.Group, waitDurations []time.Duration) ManagerFactory {
-	return func(_ context.Context, _ string, _ *notifier.Manager, _ log.Logger, _ prometheus.Registerer) RulesManager {
+	return func(_ context.Context, _ string, _ *notifier.Manager, _ log.Logger, _ prometheus.Registerer) (RulesManager, error) {
 		return &mockRulesManager{
 			done:           make(chan struct{}),
 			groupsToReturn: groupsToReturn,
 			waitDurations:  waitDurations,
 			iteration:      -1,
-		}
+		}, nil
 	}
 }
 

pkg/ruler/prom_client.go

@@ -0,0 +1,107 @@
+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/histogram"
+	"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, error) {
+	client, err := promapi.NewClient(promapi.Config{
+		Address:      queryFrontendUrl + prometheusHTTPPrefix,
+		RoundTripper: &addOrgIDRoundTripper{orgID: orgID, next: http.DefaultTransport},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &promClient{
+		promAPI: promv1.NewAPI(client),
+		timeout: timeout,
+	}, nil
+}
+
+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),
+				})
+			}
+
+			if sample.Histogram != nil {
+				v = append(v, promql.Sample{
+					T: sample.Timestamp.Unix(),
+					H: &histogram.FloatHistogram{
+						Count: float64(sample.Histogram.Count),
+						Sum:   float64(sample.Histogram.Sum),
+					},
+					Metric: metric,
+				})
+			} else {
+				v = append(v, promql.Sample{
+					T:      sample.Timestamp.Unix(),
+					F:      float64(sample.Value),
+					Metric: metric,
+				})
+			}
+		}
+		return v, nil
+	case model.ValScalar:
+		scalar := value.(*model.Scalar)
+		return promql.Vector{promql.Sample{
+			T:      scalar.Timestamp.Unix(),
+			F:      float64(scalar.Value),
+			Metric: labels.Labels{},
+		}}, nil
+	default:
+		return nil, errors.New("rule result is not a vector or scalar")
+	}
+}
+
+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,46 @@
+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
+		isError          bool
+		description      string
+	}{
+		{
+			queryFrontendUrl: "http://\tlocalhost:3333",
+			isError:          true,
+			description:      "invalid url of query frontend url should return nil",
+		},
+		{
+			queryFrontendUrl: "service.namespace:80",
+			isError:          false,
+			description:      "k8s service address should return not nil",
+		},
+		{
+			queryFrontendUrl: "service.namespace.svc.cluster.local",
+			isError:          false,
+			description:      "k8s service address should return not nil",
+		},
+	}
+
+	for _, test := range tests {
+		if test.isError {
+			_, err := NewPromClient(test.queryFrontendUrl, prometheusHTTPPrefix, orgId, timeout)
+			require.Error(t, err, test.description)
+		} else {
+			_, err := NewPromClient(test.queryFrontendUrl, prometheusHTTPPrefix, orgId, timeout)
+			require.NoError(t, err, test.description)
+		}
+	}
+}

pkg/ruler/ruler.go

@@ -93,6 +93,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 +151,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"`
@@ -190,6 +195,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")