Address comments

Signed-off-by: 🌲 Harry 🌊 John 🏔 <[email protected]>

Author: harry671003

pkg/configs/userconfig/config.go

@@ -9,10 +9,10 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/model/rulefmt"
-	"github.com/prometheus/prometheus/promql/parser"
 	"github.com/prometheus/prometheus/rules"
 	"gopkg.in/yaml.v3"
 
+	"github.com/cortexproject/cortex/pkg/parser"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 )
 

pkg/configs/userconfig/config_test.go

@@ -12,12 +12,12 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/model/rulefmt"
-	"github.com/prometheus/prometheus/promql/parser"
 	"github.com/prometheus/prometheus/rules"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
 
+	"github.com/cortexproject/cortex/pkg/parser"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 )
 

pkg/cortex/cortex.go

@@ -379,6 +379,7 @@ func New(cfg Config) (*Cortex, error) {
 		return nil, err
 	}
 
+	cortex.setupPromQLFunctions()
 	return cortex, nil
 }
 
@@ -537,3 +538,9 @@ func (t *Cortex) readyHandler(sm *services.Manager) http.HandlerFunc {
 		util.WriteTextResponse(w, "ready")
 	}
 }
+
+func (t *Cortex) setupPromQLFunctions() {
+	// The holt_winters function is renamed to double_exponential_smoothing and has been experimental since Prometheus v3. (https://github.com/prometheus/prometheus/pull/14930)
+	// The cortex supports holt_winters for users using this function.
+	querier.EnableExperimentalPromQLFunctions(t.Cfg.Querier.EnablePromQLExperimentalFunctions, true)
+}

pkg/cortex/modules.go

@@ -555,7 +555,6 @@ func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err erro
 		t.Cfg.Querier.DefaultEvaluationInterval,
 		t.Cfg.Querier.MaxSubQuerySteps,
 		t.Cfg.Querier.LookbackDelta,
-		t.Cfg.Querier.EnablePromQLExperimentalFunctions,
 	)
 
 	return services.NewIdleService(nil, func(_ error) error {
@@ -648,7 +647,7 @@ func (t *Cortex) initRuler() (serv services.Service, err error) {
 				return t.Cfg.Querier.DefaultEvaluationInterval.Milliseconds()
 			},
 		}
-		queryEngine := engine.New(opts, t.Cfg.Querier.Engine, rulerRegisterer)
+		queryEngine := engine.New(opts, t.Cfg.Ruler.ThanosEngine, rulerRegisterer)
 
 		managerFactory := ruler.DefaultTenantManagerFactory(t.Cfg.Ruler, t.Cfg.ExternalPusher, t.Cfg.ExternalQueryable, queryEngine, t.Overrides, metrics, prometheus.DefaultRegisterer)
 		manager, err = ruler.NewDefaultMultiTenantManager(t.Cfg.Ruler, t.Overrides, managerFactory, metrics, prometheus.DefaultRegisterer, util_log.Logger)

pkg/engine/config.go

@@ -10,22 +10,21 @@ import (
 
 var supportedOptimizers = []string{"default", "all", "propagate-matchers", "sort-matchers", "merge-selects", "detect-histogram-stats"}
 
-// Config contains the configuration to create engine
-type Config struct {
-	EnableThanosEngine bool                    `yaml:"enable_thanos_engine"`
-	EnableXFunctions   bool                    `yaml:"enable_x_functions"`
-	Optimizers         string                  `yaml:"optimizers"`
-	LogicalOptimizers  []logicalplan.Optimizer `yaml:"-"`
+// ThanosEngineConfig contains the configuration to create engine
+type ThanosEngineConfig struct {
+	Enabled           bool                    `yaml:"enabled"`
+	EnableXFunctions  bool                    `yaml:"enable_x_functions"`
+	Optimizers        string                  `yaml:"optimizers"`
+	LogicalOptimizers []logicalplan.Optimizer `yaml:"-"`
 }
 
-// RegisterFlags adds the flags required to config this to the given FlagSet.
-func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
-	f.BoolVar(&cfg.EnableThanosEngine, "querier.thanos-engine", false, "Experimental. Use Thanos promql engine https://github.com/thanos-io/promql-engine rather than the Prometheus promql engine.")
-	f.BoolVar(&cfg.EnableXFunctions, "querier.enable-x-functions", false, "Enable xincrease, xdelta, xrate etc from Thanos engine.")
-	f.StringVar(&cfg.Optimizers, "querier.optimizers", "default", "Logical plan optimizers. Multiple optimizers can be provided as a comma-separated list. Supported values: "+strings.Join(supportedOptimizers, ", "))
+func (cfg *ThanosEngineConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	f.BoolVar(&cfg.Enabled, prefix+"thanos-engine", false, "Experimental. Use Thanos promql engine https://github.com/thanos-io/promql-engine rather than the Prometheus promql engine.")
+	f.BoolVar(&cfg.EnableXFunctions, prefix+"enable-x-functions", false, "Enable xincrease, xdelta, xrate etc from Thanos engine.")
+	f.StringVar(&cfg.Optimizers, prefix+"optimizers", "default", "Logical plan optimizers. Multiple optimizers can be provided as a comma-separated list. Supported values: "+strings.Join(supportedOptimizers, ", "))
 }
 
-func (cfg *Config) Validate() error {
+func (cfg *ThanosEngineConfig) Validate() error {
 	splitOptimizers := strings.Split(cfg.Optimizers, ",")
 
 	for _, optimizer := range splitOptimizers {

pkg/engine/engine.go

@@ -51,16 +51,16 @@ type Engine struct {
 	engineSwitchQueriesTotal *prometheus.CounterVec
 }
 
-func New(opts promql.EngineOpts, engineCfg Config, reg prometheus.Registerer) *Engine {
+func New(opts promql.EngineOpts, thanosEngineCfg ThanosEngineConfig, reg prometheus.Registerer) *Engine {
 	prometheusEngine := promql.NewEngine(opts)
 
 	var thanosEngine *thanosengine.Engine
-	if engineCfg.EnableThanosEngine {
+	if thanosEngineCfg.Enabled {
 		thanosEngine = thanosengine.New(thanosengine.Opts{
 			EngineOpts:        opts,
-			LogicalOptimizers: engineCfg.LogicalOptimizers,
+			LogicalOptimizers: thanosEngineCfg.LogicalOptimizers,
 			EnableAnalysis:    true,
-			EnableXFunctions:  engineCfg.EnableXFunctions,
+			EnableXFunctions:  thanosEngineCfg.EnableXFunctions,
 		})
 	}
 

pkg/engine/engine_test.go

@@ -39,7 +39,7 @@ func TestEngine_Fallback(t *testing.T) {
 		Logger: utillog.GoKitLogToSlog(log.NewNopLogger()),
 		Reg:    reg,
 	}
-	queryEngine := New(opts, Config{EnableThanosEngine: true}, reg)
+	queryEngine := New(opts, ThanosEngineConfig{Enabled: true}, reg)
 
 	// instant query, should go to fallback
 	_, _ = queryEngine.NewInstantQuery(ctx, queryable, nil, "unimplemented(foo)", now)
@@ -70,7 +70,7 @@ func TestEngine_Switch(t *testing.T) {
 		Logger: utillog.GoKitLogToSlog(log.NewNopLogger()),
 		Reg:    reg,
 	}
-	queryEngine := New(opts, Config{EnableThanosEngine: true}, reg)
+	queryEngine := New(opts, ThanosEngineConfig{Enabled: true}, reg)
 
 	// Query Prometheus engine
 	r := &http.Request{Header: http.Header{}}
@@ -111,7 +111,7 @@ func TestEngine_XFunctions(t *testing.T) {
 		Logger: utillog.GoKitLogToSlog(log.NewNopLogger()),
 		Reg:    reg,
 	}
-	queryEngine := New(opts, Config{EnableThanosEngine: true, EnableXFunctions: true}, reg)
+	queryEngine := New(opts, ThanosEngineConfig{Enabled: true, EnableXFunctions: true}, reg)
 
 	for name := range parse.XFunctions {
 		t.Run(name, func(t *testing.T) {

pkg/parser/parser.go

@@ -0,0 +1,21 @@
+package parser
+
+import (
+	"maps"
+
+	promqlparser "github.com/prometheus/prometheus/promql/parser"
+	"github.com/thanos-io/promql-engine/execution/parse"
+)
+
+var functions = buildFunctions()
+
+func buildFunctions() map[string]*promqlparser.Function {
+	fns := make(map[string]*promqlparser.Function, len(promqlparser.Functions))
+	maps.Copy(fns, promqlparser.Functions)
+	maps.Copy(fns, parse.XFunctions)
+	return fns
+}
+
+func ParseExpr(qs string) (promqlparser.Expr, error) {
+	return promqlparser.NewParser(qs, promqlparser.WithFunctions(functions)).ParseExpr()
+}

pkg/querier/batch/batch_test.go

@@ -149,7 +149,7 @@ func TestSeekCorrectlyDealWithSinglePointChunks(t *testing.T) {
 			require.Equal(t, histograms[0], val)
 			require.Equal(t, int64(1*time.Second/time.Millisecond), actual)
 
-			// Histogram chunk should support querying float histograms since it is what Query Engine does.
+			// Histogram chunk should support querying float histograms since it is what Query ThanosEngine does.
 			actualT, fh := sut.AtFloatHistogram(nil)
 			require.Equal(t, histograms[0].ToFloat(nil), fh)
 			require.Equal(t, int64(1*time.Second/time.Millisecond), actualT)

pkg/querier/querier.go

@@ -85,7 +85,7 @@ type Config struct {
 
 	ShuffleShardingIngestersLookbackPeriod time.Duration `yaml:"shuffle_sharding_ingesters_lookback_period"`
 
-	Engine engine.Config `yaml:"thanos_engine"`
+	ThanosEngine engine.ThanosEngineConfig `yaml:"thanos_engine"`
 
 	// Ignore max query length check at Querier.
 	IgnoreMaxQueryLength              bool `yaml:"ignore_max_query_length"`
@@ -109,6 +109,8 @@ var (
 
 // RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
+	cfg.ThanosEngine.RegisterFlagsWithPrefix("querier.", f)
+
 	//lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods
 	flagext.DeprecatedFlag(f, "querier.ingester-streaming", "Deprecated: Use streaming RPCs to query ingester. QueryStream is always enabled and the flag is not effective anymore.", util_log.Logger)
 	//lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods
@@ -143,8 +145,6 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&cfg.EnableParquetQueryable, "querier.enable-parquet-queryable", false, "[Experimental] If true, querier will try to query the parquet files if available.")
 	f.IntVar(&cfg.ParquetQueryableShardCacheSize, "querier.parquet-queryable-shard-cache-size", 512, "[Experimental] [Experimental] Maximum size of the Parquet queryable shard cache. 0 to disable.")
 	f.StringVar(&cfg.ParquetQueryableDefaultBlockStore, "querier.parquet-queryable-default-block-store", string(parquetBlockStore), "Parquet queryable's default block store to query. Valid options are tsdb and parquet. If it is set to tsdb, parquet queryable always fallback to store gateway.")
-
-	cfg.Engine.RegisterFlags(f)
 }
 
 // Validate the config
@@ -180,7 +180,7 @@ func (cfg *Config) Validate() error {
 		}
 	}
 
-	if err := cfg.Engine.Validate(); err != nil {
+	if err := cfg.ThanosEngine.Validate(); err != nil {
 		return err
 	}
 
@@ -231,10 +231,6 @@ func New(cfg Config, limits *validation.Overrides, distributor Distributor, stor
 	})
 	maxConcurrentMetric.Set(float64(cfg.MaxConcurrent))
 
-	// The holt_winters function is renamed to double_exponential_smoothing and has been experimental since Prometheus v3. (https://github.com/prometheus/prometheus/pull/14930)
-	// The cortex supports holt_winters for users using this function.
-	EnableExperimentalPromQLFunctions(cfg.EnablePromQLExperimentalFunctions, true)
-
 	opts := promql.EngineOpts{
 		Logger:               util_log.GoKitLogToSlog(logger),
 		Reg:                  reg,
@@ -249,7 +245,7 @@ func New(cfg Config, limits *validation.Overrides, distributor Distributor, stor
 			return cfg.DefaultEvaluationInterval.Milliseconds()
 		},
 	}
-	queryEngine := engine.New(opts, cfg.Engine, reg)
+	queryEngine := engine.New(opts, cfg.ThanosEngine, reg)
 	return NewSampleAndChunkQueryable(lazyQueryable), exemplarQueryable, queryEngine
 }