Adding decodingConcurrency flag for Thanos Engine (#7118)

Signed-off-by: Paurush Garg <[email protected]>

Author: PaurushGarg

CHANGELOG.md

@@ -7,6 +7,7 @@
 * [ENHANCEMENT] Ingester: Add `enable_matcher_optimization` config to apply low selectivity matchers lazily. #7063
 * [ENHANCEMENT] Distributor: Add a label references validation for remote write v2 request. #7074
 * [ENHANCEMENT] Distributor: Add count, spans, and buckets validations for native histogram. #7072
+* [ENHANCEMENT] Ruler: Add DecodingConcurrency config flag for Thanos Engine. #7118
 * [BUGFIX] Ring: Change DynamoDB KV to retry indefinitely for WatchKey. #7088
 * [BUGFIX] Ruler: Add XFunctions validation support. #7111
 

docs/blocks-storage/querier.md

@@ -269,6 +269,11 @@ querier:
     # CLI flag: -querier.optimizers
     [optimizers: <string> | default = "default"]
 
+    # Maximum number of goroutines that can be used to decode samples. 0
+    # defaults to GOMAXPROCS / 2.
+    # CLI flag: -querier.decoding-concurrency
+    [decoding_concurrency: <int> | default = 0]
+
   # If enabled, ignore max query length check at Querier select method. Users
   # can choose to ignore it since the validation can be done before Querier
   # evaluation like at Query Frontend or Ruler.

docs/configuration/config-file-reference.md

@@ -4758,6 +4758,11 @@ thanos_engine:
   # CLI flag: -querier.optimizers
   [optimizers: <string> | default = "default"]
 
+  # Maximum number of goroutines that can be used to decode samples. 0 defaults
+  # to GOMAXPROCS / 2.
+  # CLI flag: -querier.decoding-concurrency
+  [decoding_concurrency: <int> | default = 0]
+
 # If enabled, ignore max query length check at Querier select method. Users can
 # choose to ignore it since the validation can be done before Querier evaluation
 # like at Query Frontend or Ruler.
@@ -5519,6 +5524,11 @@ thanos_engine:
   # sort-matchers, merge-selects, detect-histogram-stats
   # CLI flag: -ruler.optimizers
   [optimizers: <string> | default = "default"]
+
+  # Maximum number of goroutines that can be used to decode samples. 0 defaults
+  # to GOMAXPROCS / 2.
+  # CLI flag: -ruler.decoding-concurrency
+  [decoding_concurrency: <int> | default = 0]
 ```
 
 ### `ruler_storage_config`

pkg/engine/config.go

@@ -10,18 +10,20 @@ import (
 
 var supportedOptimizers = []string{"default", "all", "propagate-matchers", "sort-matchers", "merge-selects", "detect-histogram-stats"}
 
-// ThanosEngineConfig contains the configuration to create engine
+// 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:"-"`
+	Enabled             bool                    `yaml:"enabled"`
+	EnableXFunctions    bool                    `yaml:"enable_x_functions"`
+	Optimizers          string                  `yaml:"optimizers"`
+	DecodingConcurrency int                     `yaml:"decoding_concurrency"`
+	LogicalOptimizers   []logicalplan.Optimizer `yaml:"-"`
 }
 
 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, ", "))
+	f.IntVar(&cfg.DecodingConcurrency, prefix+"decoding-concurrency", 0, "Maximum number of goroutines that can be used to decode samples. 0 defaults to GOMAXPROCS / 2.")
 }
 
 func (cfg *ThanosEngineConfig) Validate() error {

pkg/engine/engine.go

@@ -64,10 +64,11 @@ func New(opts promql.EngineOpts, thanosEngineCfg ThanosEngineConfig, reg prometh
 	var thanosEngine *thanosengine.Engine
 	if thanosEngineCfg.Enabled {
 		thanosEngine = thanosengine.New(thanosengine.Opts{
-			EngineOpts:        opts,
-			LogicalOptimizers: thanosEngineCfg.LogicalOptimizers,
-			EnableAnalysis:    true,
-			EnableXFunctions:  thanosEngineCfg.EnableXFunctions,
+			EngineOpts:          opts,
+			LogicalOptimizers:   thanosEngineCfg.LogicalOptimizers,
+			EnableAnalysis:      true,
+			EnableXFunctions:    thanosEngineCfg.EnableXFunctions,
+			DecodingConcurrency: thanosEngineCfg.DecodingConcurrency,
 		})
 	}
 

schemas/cortex-config-schema.json

@@ -5928,6 +5928,12 @@
         },
         "thanos_engine": {
           "properties": {
+            "decoding_concurrency": {
+              "default": 0,
+              "description": "Maximum number of goroutines that can be used to decode samples. 0 defaults to GOMAXPROCS / 2.",
+              "type": "number",
+              "x-cli-flag": "querier.decoding-concurrency"
+            },
             "enable_x_functions": {
               "default": false,
               "description": "Enable xincrease, xdelta, xrate etc from Thanos engine.",
@@ -6943,6 +6949,12 @@
         },
         "thanos_engine": {
           "properties": {
+            "decoding_concurrency": {
+              "default": 0,
+              "description": "Maximum number of goroutines that can be used to decode samples. 0 defaults to GOMAXPROCS / 2.",
+              "type": "number",
+              "x-cli-flag": "ruler.decoding-concurrency"
+            },
             "enable_x_functions": {
               "default": false,
               "description": "Enable xincrease, xdelta, xrate etc from Thanos engine.",