QueryFrontend|Query: Create new arg to enable extended functions

.dockerignore

@@ -25,3 +25,5 @@ website/docs-pre-processed/
 !website/data
 
 tmp/
+
+.build/

CHANGELOG.md

@@ -27,6 +27,8 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#6984](https://github.com/thanos-io/thanos/pull/6984) Store Gateway: Added `--store.index-header-lazy-download-strategy` to specify how to lazily download index headers when lazy mmap is enabled.
 
 - [#6887](https://github.com/thanos-io/thanos/pull/6887) Query Frontend: *breaking :warning:* Add tenant label to relevant exported metrics. Note that this change may cause some pre-existing custom dashboard queries to be incorrect due to the added label.
+- [#7028](https://github.com/thanos-io/thanos/pull/7028): Query Frontend: Add new `--query-frontend.enable-x-functions` flag to enable experimental extended functions.
+- [#7028](https://github.com/thanos-io/thanos/pull/7028): Query: Add new `--query.enable-x-functions` flag to enable experimental extended functions.
 
 ### Changed
 

cmd/thanos/query.go

@@ -102,7 +102,7 @@ func registerQuery(app *extkingpin.App) {
 
  defaultEngine := cmd.Flag("query.promql-engine", "Default PromQL engine to use.").Default(string(apiv1.PromqlEnginePrometheus)).
  Enum(string(apiv1.PromqlEnginePrometheus), string(apiv1.PromqlEngineThanos))
-
+ extendedFunctionsEnabled := cmd.Flag("query.enable-x-functions", "Whether to enable extended rate functions (xrate, xincrease and xdelta). Only has effect when used with Thanos engine.").Default("false").Bool()
  promqlQueryMode := cmd.Flag("query.mode", "PromQL query mode. One of: local, distributed.").
  Hidden().
  Default(string(queryModeLocal)).
@@ -342,6 +342,7 @@ func registerQuery(app *extkingpin.App) {
  *queryTelemetrySeriesQuantiles,
  *defaultEngine,
  storeRateLimits,
+ *extendedFunctionsEnabled,
  queryMode(*promqlQueryMode),
  *tenantHeader,
  *defaultTenant,
@@ -421,6 +422,7 @@ func runQuery(
  queryTelemetrySeriesQuantiles []float64,
  defaultEngine string,
  storeRateLimits store.SeriesSelectLimits,
+ extendedFunctionsEnabled bool,
  queryMode queryMode,
  tenantHeader string,
  defaultTenant string,
@@ -652,6 +654,7 @@ func runQuery(
  engineFactory := apiv1.NewQueryEngineFactory(
  engineOpts,
  remoteEngineEndpoints,
+ extendedFunctionsEnabled,
  )
 
  lookbackDeltaCreator := LookbackDeltaFactory(engineOpts, dynamicLookbackDelta)

cmd/thanos/query_frontend.go

@@ -16,6 +16,8 @@ import (
  "github.com/opentracing/opentracing-go"
  "github.com/pkg/errors"
  "github.com/prometheus/client_golang/prometheus"
+ "github.com/prometheus/prometheus/promql/parser"
+ "github.com/thanos-io/promql-engine/execution/parse"
  "github.com/weaveworks/common/user"
  "gopkg.in/yaml.v2"
 
@@ -92,6 +94,9 @@ func registerQueryFrontend(app *extkingpin.App) {
  cmd.Flag("query-range.max-retries-per-request", "Maximum number of retries for a single query range request; beyond this, the downstream error is returned.").
  Default("5").IntVar(&cfg.QueryRangeConfig.MaxRetries)
 
+ cmd.Flag("query-frontend.enable-x-functions", "Enable experimental x- functions in query-frontend. --no-query-frontend.enable-x-functions for disabling.").
+ Default("false").BoolVar(&cfg.EnableXFunctions)
+
  cmd.Flag("query-range.max-query-length", "Limit the query time range (end - start time) in the query-frontend, 0 disables it.").
  Default("0").DurationVar((*time.Duration)(&cfg.QueryRangeConfig.Limits.MaxQueryLength))
 
@@ -285,6 +290,12 @@ func runQueryFrontend(
  return errors.Wrap(err, "error validating the config")
  }
 
+ if cfg.EnableXFunctions {
+ for fname, v := range parse.XFunctions {
+ parser.Functions[fname] = v
+ }
+ }
+
  tripperWare, err := queryfrontend.NewTripperware(cfg.Config, reg, logger)
  if err != nil {
  return errors.Wrap(err, "setup tripperwares")

docs/components/query-frontend.md

@@ -252,6 +252,11 @@ Flags:
  --query-frontend.downstream-url="http://localhost:9090"
  URL of downstream Prometheus Query compatible
  API.
+ --query-frontend.enable-x-functions
+ Enable experimental x-
+ functions in query-frontend.
+ --no-query-frontend.enable-x-functions for
+ disabling.
  --query-frontend.forward-header=<http-header-name> ...
  List of headers forwarded by the query-frontend
  to downstream queriers, default is empty

docs/components/query.md

@@ -363,6 +363,10 @@ Flags:
  --query.default-tenant-id="default-tenant"
  Default tenant ID to use if tenant header is
  not present
+ --query.enable-x-functions
+ Whether to enable extended rate functions
+ (xrate, xincrease and xdelta). Only has effect
+ when used with Thanos engine.
  --query.lookback-delta=QUERY.LOOKBACK-DELTA
  The maximum lookback duration for retrieving
  metrics during expression evaluations.

pkg/api/query/v1.go

@@ -99,6 +99,7 @@ type QueryEngineFactory struct {
 
  createThanosEngine sync.Once
  thanosEngine v1.QueryEngine
+ enableXFunctions bool
 }
 
 func (f *QueryEngineFactory) GetPrometheusEngine() v1.QueryEngine {
@@ -118,7 +119,7 @@ func (f *QueryEngineFactory) GetThanosEngine() v1.QueryEngine {
  return
  }
  if f.remoteEngineEndpoints == nil {
- f.thanosEngine = engine.New(engine.Opts{EngineOpts: f.engineOpts, Engine: f.GetPrometheusEngine(), EnableAnalysis: true})
+ f.thanosEngine = engine.New(engine.Opts{EngineOpts: f.engineOpts, Engine: f.GetPrometheusEngine(), EnableAnalysis: true, EnableXFunctions: f.enableXFunctions})
  } else {
  f.thanosEngine = engine.NewDistributedEngine(engine.Opts{EngineOpts: f.engineOpts, Engine: f.GetPrometheusEngine(), EnableAnalysis: true}, f.remoteEngineEndpoints)
  }
@@ -127,13 +128,11 @@ func (f *QueryEngineFactory) GetThanosEngine() v1.QueryEngine {
  return f.thanosEngine
 }
 
-func NewQueryEngineFactory(
- engineOpts promql.EngineOpts,
- remoteEngineEndpoints promqlapi.RemoteEndpoints,
-) *QueryEngineFactory {
+func NewQueryEngineFactory(engineOpts promql.EngineOpts, remoteEngineEndpoints promqlapi.RemoteEndpoints, enableExtendedFunctions bool) *QueryEngineFactory {
  return &QueryEngineFactory{
  engineOpts: engineOpts,
  remoteEngineEndpoints: remoteEngineEndpoints,
+ enableXFunctions: enableExtendedFunctions,
  }
 }
 

pkg/api/query/v1_test.go

@@ -190,7 +190,7 @@ func TestQueryEndpoints(t *testing.T) {
  Reg: nil,
  MaxSamples: 10000,
  Timeout: timeout,
- }, nil)
+ }, nil, false)
  api := &QueryAPI{
  baseAPI: &baseAPI.BaseAPI{
  Now: func() time.Time { return now },
@@ -643,7 +643,7 @@ func TestQueryExplainEndpoints(t *testing.T) {
  Reg: nil,
  MaxSamples: 10000,
  Timeout: timeout,
- }, nil)
+ }, nil, false)
  api := &QueryAPI{
  baseAPI: &baseAPI.BaseAPI{
  Now: func() time.Time { return now },
@@ -707,7 +707,7 @@ func TestQueryAnalyzeEndpoints(t *testing.T) {
  Reg: nil,
  MaxSamples: 10000,
  Timeout: timeout,
- }, nil)
+ }, nil, false)
  api := &QueryAPI{
  baseAPI: &baseAPI.BaseAPI{
  Now: func() time.Time { return now },
@@ -881,7 +881,7 @@ func TestMetadataEndpoints(t *testing.T) {
  Reg: nil,
  MaxSamples: 10000,
  Timeout: timeout,
- }, nil)
+ }, nil, false)
  api := &QueryAPI{
  baseAPI: &baseAPI.BaseAPI{
  Now: func() time.Time { return now },

pkg/queryfrontend/config.go

@@ -209,6 +209,7 @@ type Config struct {
  TenantHeader string
  DefaultTenant string
  TenantCertField string
+ EnableXFunctions bool
 }
 
 // QueryRangeConfig holds the config for query range tripperware.

test/e2e/e2ethanos/services.go

@@ -29,6 +29,7 @@ import (
  "github.com/thanos-io/objstore/exthttp"
 
  "github.com/thanos-io/thanos/pkg/alert"
+ apiv1 "github.com/thanos-io/thanos/pkg/api/query"
  "github.com/thanos-io/thanos/pkg/clientconfig"
  "github.com/thanos-io/thanos/pkg/queryfrontend"
  "github.com/thanos-io/thanos/pkg/receive"
@@ -253,8 +254,9 @@ type QuerierBuilder struct {
  endpoints []string
  strictEndpoints []string
 
- engine string
- queryMode string
+ engine apiv1.PromqlEngineType
+ queryMode string
+ enableXFunctions bool
 
  replicaLabels []string
  tracingConfig string
@@ -362,7 +364,7 @@ func (q *QuerierBuilder) WithDisablePartialResponses(disable bool) *QuerierBuild
  return q
 }
 
-func (q *QuerierBuilder) WithEngine(engine string) *QuerierBuilder {
+func (q *QuerierBuilder) WithEngine(engine apiv1.PromqlEngineType) *QuerierBuilder {
  q.engine = engine
  return q
 }
@@ -372,6 +374,11 @@ func (q *QuerierBuilder) WithQueryMode(mode string) *QuerierBuilder {
  return q
 }
 
+func (q *QuerierBuilder) WithEnableXFunctions(enable bool) *QuerierBuilder {
+ q.enableXFunctions = enable
+ return q
+}
+
 func (q *QuerierBuilder) WithEnvVars(envVars map[string]string) *QuerierBuilder {
  q.envVars = envVars
  return q
@@ -484,6 +491,13 @@ func (q *QuerierBuilder) collectArgs() ([]string, error) {
  for _, bucket := range q.telemetrySeriesQuantiles {
  args = append(args, "--query.telemetry.request-series-seconds-quantiles="+strconv.FormatFloat(bucket, 'f', -1, 64))
  }
+ if q.enableXFunctions {
+ args = append(args, "--query.enable-x-functions")
+ }
+ if q.engine != "" {
+ args = append(args, "--query.promql-engine="+string(q.engine))
+ }
+
  return args, nil
 }
 

test/e2e/query_test.go

@@ -315,6 +315,49 @@ func TestQuery(t *testing.T) {
  })
 }
 
+func TestQueryWithExtendedFunctions(t *testing.T) {
+ t.Parallel()
+
+ e, err := e2e.New(e2e.WithName("e2e-qry-xfunc"))
+ testutil.Ok(t, err)
+ t.Cleanup(e2ethanos.CleanScenario(t, e))
+
+ // create prom + sidecar
+ prom, sidecar := e2ethanos.NewPrometheusWithSidecar(e, "prom", e2ethanos.DefaultPromConfig("prom", 0, "", "", e2ethanos.LocalPrometheusTarget), "", e2ethanos.DefaultPrometheusImage(), "", "remote-write-receiver")
+ testutil.Ok(t, e2e.StartAndWaitReady(prom, sidecar))
+
+ // create querier
+ q := e2ethanos.NewQuerierBuilder(e, "1", sidecar.InternalEndpoint("grpc")).WithEngine("thanos").WithEnableXFunctions(true).Init()
+ testutil.Ok(t, e2e.StartAndWaitReady(q))
+
+ ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
+ t.Cleanup(cancel)
+
+ // send series to prom
+ samples := []seriesWithLabels{
+ {intLabels: labels.FromStrings("__name__", "my_fake_metric", "a", "1", "b", "1", "instance", "1")},
+ {intLabels: labels.FromStrings("__name__", "my_fake_metric", "a", "1", "b", "2", "instance", "1")},
+ {intLabels: labels.FromStrings("__name__", "my_fake_metric", "a", "2", "b", "1", "instance", "1")},
+ {intLabels: labels.FromStrings("__name__", "my_fake_metric", "a", "2", "b", "2", "instance", "1")},
+ }
+ testutil.Ok(t, remoteWriteSeriesWithLabels(ctx, prom, samples))
+
+ // query
+ queryAndAssertSeries(t, ctx, q.Endpoint("http"), func() string {
+ return `xrate(my_fake_metric{a="1", b="1"}[1m])`
+ }, time.Now, promclient.QueryOptions{
+ Deduplicate: false,
+ }, []model.Metric{
+ {
+ "a": "1",
+ "b": "1",
+ "instance": "1",
+ "prometheus": "prom",
+ "replica": "0",
+ },
+ })
+}
+
 func TestQueryExternalPrefixWithoutReverseProxy(t *testing.T) {
  t.Parallel()