Total Worker Parallelism in the Querier

CHANGELOG.md

@@ -24,6 +24,7 @@
 * [ENHANCEMENT] Experimental TSDB: Added `cortex_querier_blocks_meta_synced`, which reflects current state of synced blocks over all tenants. #2392
 * [ENHANCEMENT] Added `cortex_distributor_latest_seen_sample_timestamp_seconds` metric to see how far behind Prometheus servers are in sending data. #2371
 * [ENHANCEMENT] FIFO cache to support eviction based on memory usage. The `-<prefix>.fifocache.size` CLI flag has been renamed to `-<prefix>.fifocache.max-size-items` as well as its YAML config option `size` renamed to `max_size_items`. Added `-<prefix>.fifocache.max-size-bytes` CLI flag and YAML config option `max_size_bytes` to specify memory limit of the cache. #2319
+* [ENHANCEMENT] Added `-querier.worker-match-max-concurrent`.  Force worker concurrency to match the `-querier.max-concurrent` option.  Overrides `-querier.worker-parallelism`.  #2456
 * [ENHANCEMENT] Added the following metrics for monitoring delete requests: #2445
   - `cortex_purger_delete_requests_received_total`: Number of delete requests received per user.
   - `cortex_purger_delete_requests_processed_total`: Number of delete requests processed per user.

docs/configuration/arguments.md

@@ -16,7 +16,7 @@ Duration arguments should be specified with a unit like `5s` or `3h`. Valid time
 - `-querier.max-concurrent`
 
    The maximum number of top-level PromQL queries that will execute at the same time, per querier process.
-   If using the query frontend, this should be set to at least (`querier.worker-parallelism` * number of query frontend replicas). Otherwise queries may queue in the queriers and not the frontend, which will affect QoS.
+   If using the query frontend, this should be set to at least (`-querier.worker-parallelism` * number of query frontend replicas). Otherwise queries may queue in the queriers and not the frontend, which will affect QoS.  Alternatively, consider using `-querier.worker-match-max-concurrent` to force worker parallelism to match `-querier.max-concurrent`.
 
 - `-querier.query-parallelism`
 
@@ -42,9 +42,15 @@ The next three options only apply when the querier is used together with the Que
 
 - `-querier.worker-parallelism`
 
-   Number of simultaneous queries to process, per worker process.
+   Number of simultaneous queries to process, per query frontend.
    See note on `-querier.max-concurrent`
 
+- `-querier.worker-match-max-concurrent`
+
+   Force worker concurrency to match the -querier.max-concurrent option.  Overrides `-querier.worker-parallelism`.
+   See note on `-querier.max-concurrent`
+
+
 ## Querier and Ruler
 
 The ingester query API was improved over time, but defaults to the old behaviour for backwards-compatibility. For best results both of these next two flags should be set to `true`:

docs/configuration/config-file-reference.md

@@ -1801,10 +1801,15 @@ The `frontend_worker_config` configures the worker - running within the Cortex q
 # CLI flag: -querier.frontend-address
 [frontend_address: <string> | default = ""]
 
-# Number of simultaneous queries to process.
+# Number of simultaneous queries to process per query frontend.
 # CLI flag: -querier.worker-parallelism
 [parallelism: <int> | default = 10]
 
+# Force worker concurrency to match the -querier.max-concurrent option.
+# Overrides querier.worker-parallelism.
+# CLI flag: -querier.worker-match-max-concurrent
+[match_max_concurrent: <boolean> | default = false]
+
 # How often to query DNS.
 # CLI flag: -querier.dns-lookup-period
 [dns_lookup_duration: <duration> | default = 10s]

docs/configuration/single-process-config-blocks-gossip-1.yaml

@@ -80,3 +80,6 @@ tsdb:
   backend: filesystem # s3, gcs, azure or filesystem are valid options
   filesystem:
     dir: /tmp/cortex/storage
+
+frontend_worker:
+  match_max_concurrent: true

docs/configuration/single-process-config-blocks-gossip-2.yaml

@@ -79,3 +79,6 @@ tsdb:
   backend: filesystem # s3, gcs, azure or filesystem are valid options
   filesystem:
     dir: /tmp/cortex/storage
+
+frontend_worker:
+  match_max_concurrent: true

docs/configuration/single-process-config-blocks.yaml

@@ -82,3 +82,6 @@ compactor:
   sharding_ring:
     kvstore:
       store: inmemory
+
+frontend_worker:
+  match_max_concurrent: true

docs/configuration/single-process-config.md

@@ -71,4 +71,9 @@ storage:
 
   filesystem:
     directory: /tmp/cortex/chunks
+
+# Configure the frontend worker in the querier to match worker count
+#  to max_concurrent on the queriers.
+frontend_worker:
+  match_max_concurrent: true
 ```

docs/configuration/single-process-config.yaml

@@ -75,3 +75,6 @@ storage:
 purger:
   enable: true
   object_store_type: filesystem
+
+frontend_worker:
+  match_max_concurrent: true

pkg/cortex/modules.go

@@ -210,7 +210,7 @@ func (t *Cortex) initQuerier(cfg *Config) (serv services.Service, err error) {
 
 	// Query frontend worker will only be started after all its dependencies are started, not here.
 	// Worker may also be nil, if not configured, which is OK.
-	worker, err := frontend.NewWorker(cfg.Worker, httpgrpc_server.NewServer(handler), util.Logger)
+	worker, err := frontend.NewWorker(cfg.Worker, cfg.Querier, httpgrpc_server.NewServer(handler), util.Logger)
 	if err != nil {
 		return
 	}

pkg/querier/frontend/frontend_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/weaveworks/common/user"
 	"google.golang.org/grpc"
 
+	"github.com/cortexproject/cortex/pkg/querier"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/services"
 )
@@ -55,7 +56,8 @@ func TestFrontend(t *testing.T) {
 
 		assert.Equal(t, "Hello World", string(body))
 	}
-	testFrontend(t, handler, test)
+	testFrontend(t, handler, test, false)
+	testFrontend(t, handler, test, true)
 }
 
 func TestFrontendPropagateTrace(t *testing.T) {
@@ -104,7 +106,8 @@ func TestFrontendPropagateTrace(t *testing.T) {
 		// Query should do one calls.
 		assert.Equal(t, traceID, <-observedTraceID)
 	}
-	testFrontend(t, handler, test)
+	testFrontend(t, handler, test, false)
+	testFrontend(t, handler, test, true)
 }
 
 // TestFrontendCancel ensures that when client requests are cancelled,
@@ -135,7 +138,9 @@ func TestFrontendCancel(t *testing.T) {
 		time.Sleep(100 * time.Millisecond)
 		assert.Equal(t, int32(1), atomic.LoadInt32(&tries))
 	}
-	testFrontend(t, handler, test)
+	testFrontend(t, handler, test, false)
+	tries = 0
+	testFrontend(t, handler, test, true)
 }
 
 func TestFrontendCancelStatusCode(t *testing.T) {
@@ -156,15 +161,18 @@ func TestFrontendCancelStatusCode(t *testing.T) {
 	}
 }
 
-func testFrontend(t *testing.T, handler http.Handler, test func(addr string)) {
+func testFrontend(t *testing.T, handler http.Handler, test func(addr string), matchMaxConcurrency bool) {
 	logger := log.NewNopLogger()
 
 	var (
-		config       Config
-		workerConfig WorkerConfig
+		config        Config
+		workerConfig  WorkerConfig
+		querierConfig querier.Config
 	)
 	flagext.DefaultValues(&config, &workerConfig)
 	workerConfig.Parallelism = 1
+	workerConfig.MatchMaxConcurrency = matchMaxConcurrency
+	querierConfig.MaxConcurrent = 1
 
 	// localhost:0 prevents firewall warnings on Mac OS X.
 	grpcListen, err := net.Listen("tcp", "localhost:0")
@@ -196,7 +204,7 @@ func testFrontend(t *testing.T, handler http.Handler, test func(addr string)) {
 	go httpServer.Serve(httpListen) //nolint:errcheck
 	go grpcServer.Serve(grpcListen) //nolint:errcheck
 
-	worker, err := NewWorker(workerConfig, httpgrpc_server.NewServer(handler), logger)
+	worker, err := NewWorker(workerConfig, querierConfig, httpgrpc_server.NewServer(handler), logger)
 	require.NoError(t, err)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), worker))