Deprecating querier.max-outstanding-requests-per-tenant and -query-scheduler.max-outstanding-requests-per-tenant flags (#6146)

* deprecating querier.max-outstanding-requests-per-tenant flag

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

* fix scheduler

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

* changelog

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

* fix test

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

---------

Signed-off-by: alanprot <[email protected]>
Signed-off-by: Alan Protasio <[email protected]>

Author: alanprot

CHANGELOG.md

@@ -7,6 +7,7 @@
 * [CHANGE] Distributor/Ingester: Change `cortex_distributor_ingester_appends_total`, `cortex_distributor_ingester_append_failures_total`, `cortex_distributor_ingester_queries_total`, and `cortex_distributor_ingester_query_failures_total` metrics to use the ingester ID instead of its IP as the label value. #6078
 * [CHANGE] OTLP: Set `AddMetricSuffixes` to true to always enable metric name normalization. #6136
 * [CHANGE] Querier: Deprecate and enable by default `querier.ingester-metadata-streaming` flag. #6147
+* [CHANGE] QueryFrontend/QueryScheduler: Deprecate `-querier.max-outstanding-requests-per-tenant` and `-query-scheduler.max-outstanding-requests-per-tenant` flags. Use frontend.max-outstanding-requests-per-tenant instead. #6146
 * [FEATURE] Ingester/Distributor: Experimental: Enable native histogram ingestion via `-blocks-storage.tsdb.enable-native-histograms` flag. #5986 #6010 #6020
 * [FEATURE] Querier: Enable querying native histogram chunks. #5944 #6031
 * [FEATURE] Query Frontend: Support native histogram in query frontend response. #5996 #6043

docs/configuration/config-file-reference.md

@@ -176,13 +176,6 @@ tenant_federation:
 [memberlist: <memberlist_config>]
 
 query_scheduler:
-  # Deprecated (use frontend.max-outstanding-requests-per-tenant instead) and
-  # will be removed in v1.17.0: Maximum number of outstanding requests per
-  # tenant per query-scheduler. In-flight requests above this limit will fail
-  # with HTTP response status code 429.
-  # CLI flag: -query-scheduler.max-outstanding-requests-per-tenant
-  [max_outstanding_requests_per_tenant: <int> | default = 0]
-
   # If a querier disconnects without sending notification about graceful
   # shutdown, the query-scheduler will keep the querier in the tenant's shard
   # until the forget delay has passed. This feature is useful to reduce the
@@ -3835,12 +3828,6 @@ The `query_frontend_config` configures the Cortex query-frontend.
 # CLI flag: -frontend.query-stats-enabled
 [query_stats_enabled: <boolean> | default = false]
 
-# Deprecated (use frontend.max-outstanding-requests-per-tenant instead) and will
-# be removed in v1.17.0: Maximum number of outstanding requests per tenant per
-# frontend; requests beyond this error with HTTP 429.
-# CLI flag: -querier.max-outstanding-requests-per-tenant
-[max_outstanding_per_tenant: <int> | default = 0]
-
 # If a querier disconnects without sending notification about graceful shutdown,
 # the query-frontend will keep the querier in the tenant's shard until the
 # forget delay has passed. This feature is useful to reduce the blast radius

pkg/frontend/v1/frontend.go

@@ -21,7 +21,9 @@ import (
 	"github.com/cortexproject/cortex/pkg/scheduler/queue"
 	"github.com/cortexproject/cortex/pkg/tenant"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/httpgrpcutil"
+	util_log "github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
@@ -32,13 +34,14 @@ var (
 
 // Config for a Frontend.
 type Config struct {
-	MaxOutstandingPerTenant int           `yaml:"max_outstanding_per_tenant"`
-	QuerierForgetDelay      time.Duration `yaml:"querier_forget_delay"`
+	QuerierForgetDelay time.Duration `yaml:"querier_forget_delay"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
-	f.IntVar(&cfg.MaxOutstandingPerTenant, "querier.max-outstanding-requests-per-tenant", 0, "Deprecated (use frontend.max-outstanding-requests-per-tenant instead) and will be removed in v1.17.0: Maximum number of outstanding requests per tenant per frontend; requests beyond this error with HTTP 429.")
+	//lint:ignore faillint Need to pass the global logger like this for warning on deprecated methods
+	flagext.DeprecatedFlag(f, "querier.max-outstanding-requests-per-tenant", "Deprecated: Use frontend.max-outstanding-requests-per-tenant instead.", util_log.Logger)
+
 	f.DurationVar(&cfg.QuerierForgetDelay, "query-frontend.querier-forget-delay", 0, "If a querier disconnects without sending notification about graceful shutdown, the query-frontend will keep the querier in the tenant's shard until the forget delay has passed. This feature is useful to reduce the blast radius when shuffle-sharding is enabled.")
 }
 
@@ -129,7 +132,7 @@ func New(cfg Config, limits Limits, log log.Logger, registerer prometheus.Regist
 		}),
 	}
 
-	f.requestQueue = queue.NewRequestQueue(cfg.MaxOutstandingPerTenant, cfg.QuerierForgetDelay, f.queueLength, f.discardedRequests, f.limits, registerer)
+	f.requestQueue = queue.NewRequestQueue(cfg.QuerierForgetDelay, f.queueLength, f.discardedRequests, f.limits, registerer)
 	f.activeUsers = util.NewActiveUsersCleanupWithDefaultValues(f.cleanupInactiveUserMetrics)
 
 	var err error

pkg/frontend/v1/frontend_test.go

@@ -129,7 +129,7 @@ func TestFrontendCheckReady(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			f := &Frontend{
 				log: log.NewNopLogger(),
-				requestQueue: queue.NewRequestQueue(5, 0,
+				requestQueue: queue.NewRequestQueue(5,
 					prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user"}),
 					prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user"}),
 					limits,

pkg/frontend/v1/queue_test.go

@@ -22,10 +22,10 @@ import (
 	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
-func setupFrontend(t *testing.T, config Config) (*Frontend, error) {
+func setupFrontend(t *testing.T, maxOutstanding int, config Config) (*Frontend, error) {
 	logger := log.NewNopLogger()
 
-	limits := MockLimits{Queriers: 3, MockLimits: queue.MockLimits{MaxOutstanding: 100}}
+	limits := MockLimits{Queriers: 3, MockLimits: queue.MockLimits{MaxOutstanding: maxOutstanding}}
 	frontend, err := New(config, limits, logger, nil, transport.NewRetry(0, nil))
 	require.NoError(t, err)
 
@@ -51,18 +51,17 @@ func testReq(ctx context.Context, reqID, user string) *request {
 func TestDequeuesExpiredRequests(t *testing.T) {
 	var config Config
 	flagext.DefaultValues(&config)
-	config.MaxOutstandingPerTenant = 10
 	userID := "1"
 
-	f, err := setupFrontend(t, config)
+	f, err := setupFrontend(t, 10, config)
 	require.NoError(t, err)
 
 	ctx := user.InjectOrgID(context.Background(), userID)
 	expired, cancel := context.WithCancel(ctx)
 	cancel()
 
 	good := 0
-	for i := 0; i < config.MaxOutstandingPerTenant; i++ {
+	for i := 0; i < 10; i++ {
 		var err error
 		if i%5 == 0 {
 			good++
@@ -99,9 +98,7 @@ func TestRoundRobinQueues(t *testing.T) {
 		tenants  = 10
 	)
 
-	config.MaxOutstandingPerTenant = requests
-
-	f, err := setupFrontend(t, config)
+	f, err := setupFrontend(t, requests, config)
 	require.NoError(t, err)
 
 	for i := 0; i < requests; i++ {

pkg/scheduler/queue/queue.go

@@ -66,9 +66,9 @@ type RequestQueue struct {
 	discardedRequests *prometheus.CounterVec // Per user and priority.
 }
 
-func NewRequestQueue(maxOutstandingPerTenant int, forgetDelay time.Duration, queueLength *prometheus.GaugeVec, discardedRequests *prometheus.CounterVec, limits Limits, registerer prometheus.Registerer) *RequestQueue {
+func NewRequestQueue(forgetDelay time.Duration, queueLength *prometheus.GaugeVec, discardedRequests *prometheus.CounterVec, limits Limits, registerer prometheus.Registerer) *RequestQueue {
 	q := &RequestQueue{
-		queues:                  newUserQueues(maxOutstandingPerTenant, forgetDelay, limits, queueLength),
+		queues:                  newUserQueues(forgetDelay, limits, queueLength),
 		connectedQuerierWorkers: atomic.NewInt32(0),
 		totalRequests: promauto.With(registerer).NewCounterVec(prometheus.CounterOpts{
 			Name: "cortex_request_queue_requests_total",

pkg/scheduler/queue/queue_test.go

@@ -24,7 +24,7 @@ func BenchmarkGetNextRequest(b *testing.B) {
 	queues := make([]*RequestQueue, 0, b.N)
 
 	for n := 0; n < b.N; n++ {
-		queue := NewRequestQueue(maxOutstandingPerTenant, 0,
+		queue := NewRequestQueue(maxOutstandingPerTenant,
 			prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user", "priority", "type"}),
 			prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user", "priority"}),
 			MockLimits{MaxOutstanding: 100},
@@ -83,7 +83,7 @@ func BenchmarkQueueRequest(b *testing.B) {
 	requests := make([]MockRequest, 0, numTenants)
 
 	for n := 0; n < b.N; n++ {
-		q := NewRequestQueue(maxOutstandingPerTenant, 0,
+		q := NewRequestQueue(maxOutstandingPerTenant,
 			prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user", "priority", "type"}),
 			prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user", "priority"}),
 			MockLimits{MaxOutstanding: 100},
@@ -123,7 +123,7 @@ func BenchmarkGetNextRequestPriorityQueue(b *testing.B) {
 	queues := make([]*RequestQueue, 0, b.N)
 
 	for n := 0; n < b.N; n++ {
-		queue := NewRequestQueue(maxOutstandingPerTenant, 0,
+		queue := NewRequestQueue(maxOutstandingPerTenant,
 			prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user", "priority", "type"}),
 			prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user", "priority"}),
 			MockLimits{MaxOutstanding: 100, QueryPriorityVal: validation.QueryPriority{Enabled: true}},
@@ -182,7 +182,7 @@ func BenchmarkQueueRequestPriorityQueue(b *testing.B) {
 	requests := make([]MockRequest, 0, numTenants)
 
 	for n := 0; n < b.N; n++ {
-		q := NewRequestQueue(maxOutstandingPerTenant, 0,
+		q := NewRequestQueue(maxOutstandingPerTenant,
 			prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user", "priority", "type"}),
 			prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user", "priority"}),
 			MockLimits{MaxOutstanding: 100, QueryPriorityVal: validation.QueryPriority{Enabled: true}},
@@ -217,7 +217,7 @@ func BenchmarkQueueRequestPriorityQueue(b *testing.B) {
 func TestRequestQueue_GetNextRequestForQuerier_ShouldGetRequestAfterReshardingBecauseQuerierHasBeenForgotten(t *testing.T) {
 	const forgetDelay = 3 * time.Second
 
-	queue := NewRequestQueue(1, forgetDelay,
+	queue := NewRequestQueue(forgetDelay,
 		prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user", "priority", "type"}),
 		prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user", "priority"}),
 		MockLimits{MaxOutstanding: 100},
@@ -260,7 +260,7 @@ func TestRequestQueue_GetNextRequestForQuerier_ShouldGetRequestAfterReshardingBe
 }
 
 func TestQueriersShouldGetHighPriorityQueryFirst(t *testing.T) {
-	queue := NewRequestQueue(0, 0,
+	queue := NewRequestQueue(0,
 		prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user", "priority", "type"}),
 		prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user", "priority"}),
 		MockLimits{MaxOutstanding: 3, QueryPriorityVal: validation.QueryPriority{Enabled: true}},
@@ -290,7 +290,7 @@ func TestQueriersShouldGetHighPriorityQueryFirst(t *testing.T) {
 }
 
 func TestReservedQueriersShouldOnlyGetHighPriorityQueries(t *testing.T) {
-	queue := NewRequestQueue(0, 0,
+	queue := NewRequestQueue(0,
 		prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user", "priority", "type"}),
 		prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user", "priority"}),
 		MockLimits{
@@ -356,7 +356,7 @@ func TestExitingRequestsShouldPersistEvenIfTheConfigHasChanged(t *testing.T) {
 	limits := MockLimits{
 		MaxOutstanding: 3,
 	}
-	queue := NewRequestQueue(0, 0,
+	queue := NewRequestQueue(0,
 		prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"user", "priority", "type"}),
 		prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"user", "priority"}),
 		limits,

pkg/scheduler/queue/user_queues.go

@@ -46,8 +46,6 @@ type queues struct {
 	// this list when there are ""'s at the end of it.
 	users []string
 
-	maxUserQueueSize int
-
 	// How long to wait before removing a querier which has got disconnected
 	// but hasn't notified about a graceful shutdown.
 	forgetDelay time.Duration
@@ -87,16 +85,15 @@ type userQueue struct {
 	index int
 }
 
-func newUserQueues(maxUserQueueSize int, forgetDelay time.Duration, limits Limits, queueLength *prometheus.GaugeVec) *queues {
+func newUserQueues(forgetDelay time.Duration, limits Limits, queueLength *prometheus.GaugeVec) *queues {
 	return &queues{
-		userQueues:       map[string]*userQueue{},
-		users:            nil,
-		maxUserQueueSize: maxUserQueueSize,
-		forgetDelay:      forgetDelay,
-		queriers:         map[string]*querier{},
-		sortedQueriers:   nil,
-		limits:           limits,
-		queueLength:      queueLength,
+		userQueues:     map[string]*userQueue{},
+		users:          nil,
+		forgetDelay:    forgetDelay,
+		queriers:       map[string]*querier{},
+		sortedQueriers: nil,
+		limits:         limits,
+		queueLength:    queueLength,
 	}
 }
 
@@ -216,12 +213,6 @@ func (q *queues) createUserRequestQueue(userID string) userRequestQueue {
 
 	queueSize := q.limits.MaxOutstandingPerTenant(userID)
 
-	// 0 is the default value of the flag. If the old flag is set
-	// then we use its value for compatibility reason.
-	if q.maxUserQueueSize != 0 {
-		queueSize = q.maxUserQueueSize
-	}
-
 	return NewFIFORequestQueue(make(chan Request, queueSize), userID, q.queueLength)
 }
 

pkg/scheduler/queue/user_queues_test.go

@@ -16,7 +16,7 @@ import (
 )
 
 func TestQueues(t *testing.T) {
-	uq := newUserQueues(0, 0, MockLimits{}, nil)
+	uq := newUserQueues(0, MockLimits{}, nil)
 	assert.NotNil(t, uq)
 	assert.NoError(t, isConsistent(uq))
 
@@ -71,7 +71,7 @@ func TestQueues(t *testing.T) {
 }
 
 func TestQueuesWithQueriers(t *testing.T) {
-	uq := newUserQueues(0, 0, MockLimits{}, nil)
+	uq := newUserQueues(0, MockLimits{}, nil)
 	assert.NotNil(t, uq)
 	assert.NoError(t, isConsistent(uq))
 
@@ -148,7 +148,7 @@ func TestQueuesConsistency(t *testing.T) {
 
 	for testName, testData := range tests {
 		t.Run(testName, func(t *testing.T) {
-			uq := newUserQueues(0, testData.forgetDelay, MockLimits{}, nil)
+			uq := newUserQueues(testData.forgetDelay, MockLimits{}, nil)
 			assert.NotNil(t, uq)
 			assert.NoError(t, isConsistent(uq))
 
@@ -197,7 +197,7 @@ func TestQueues_ForgetDelay(t *testing.T) {
 	)
 
 	now := time.Now()
-	uq := newUserQueues(0, forgetDelay, MockLimits{}, nil)
+	uq := newUserQueues(forgetDelay, MockLimits{}, nil)
 	assert.NotNil(t, uq)
 	assert.NoError(t, isConsistent(uq))
 
@@ -289,7 +289,7 @@ func TestQueues_ForgetDelay_ShouldCorrectlyHandleQuerierReconnectingBeforeForget
 	)
 
 	now := time.Now()
-	uq := newUserQueues(0, forgetDelay, MockLimits{}, nil)
+	uq := newUserQueues(forgetDelay, MockLimits{}, nil)
 	assert.NotNil(t, uq)
 	assert.NoError(t, isConsistent(uq))
 
@@ -358,7 +358,7 @@ func TestGetOrAddQueueShouldUpdateProperties(t *testing.T) {
 	limits := MockLimits{
 		MaxOutstanding: 3,
 	}
-	q := newUserQueues(0, 0, limits, nil)
+	q := newUserQueues(0, limits, nil)
 	q.addQuerierConnection("q-1")
 	q.addQuerierConnection("q-2")
 	q.addQuerierConnection("q-3")
@@ -463,7 +463,7 @@ func TestQueueConcurrency(t *testing.T) {
 	limits := MockLimits{
 		MaxOutstanding: 50,
 	}
-	q := newUserQueues(0, 0, limits, nil)
+	q := newUserQueues(0, limits, nil)
 	q.addQuerierConnection("q-1")
 	q.addQuerierConnection("q-2")
 	q.addQuerierConnection("q-3")

pkg/scheduler/scheduler.go

@@ -27,8 +27,10 @@ import (
 	"github.com/cortexproject/cortex/pkg/scheduler/schedulerpb"
 	"github.com/cortexproject/cortex/pkg/tenant"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/grpcclient"
 	"github.com/cortexproject/cortex/pkg/util/httpgrpcutil"
+	util_log "github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
@@ -82,13 +84,12 @@ type connectedFrontend struct {
 }
 
 type Config struct {
-	MaxOutstandingPerTenant int               `yaml:"max_outstanding_requests_per_tenant"`
-	QuerierForgetDelay      time.Duration     `yaml:"querier_forget_delay"`
-	GRPCClientConfig        grpcclient.Config `yaml:"grpc_client_config" doc:"description=This configures the gRPC client used to report errors back to the query-frontend."`
+	QuerierForgetDelay time.Duration     `yaml:"querier_forget_delay"`
+	GRPCClientConfig   grpcclient.Config `yaml:"grpc_client_config" doc:"description=This configures the gRPC client used to report errors back to the query-frontend."`
 }
 
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
-	f.IntVar(&cfg.MaxOutstandingPerTenant, "query-scheduler.max-outstanding-requests-per-tenant", 0, "Deprecated (use frontend.max-outstanding-requests-per-tenant instead) and will be removed in v1.17.0: Maximum number of outstanding requests per tenant per query-scheduler. In-flight requests above this limit will fail with HTTP response status code 429.")
+	flagext.DeprecatedFlag(f, "query-scheduler.max-outstanding-requests-per-tenant", "Deprecated: Use frontend.max-outstanding-requests-per-tenant instead.", util_log.Logger)
 	f.DurationVar(&cfg.QuerierForgetDelay, "query-scheduler.querier-forget-delay", 0, "If a querier disconnects without sending notification about graceful shutdown, the query-scheduler will keep the querier in the tenant's shard until the forget delay has passed. This feature is useful to reduce the blast radius when shuffle-sharding is enabled.")
 	cfg.GRPCClientConfig.RegisterFlagsWithPrefix("query-scheduler.grpc-client-config", f)
 }
@@ -114,7 +115,7 @@ func NewScheduler(cfg Config, limits Limits, log log.Logger, registerer promethe
 		Help: "Total number of query requests discarded.",
 	}, []string{"user", "priority"})
 
-	s.requestQueue = queue.NewRequestQueue(cfg.MaxOutstandingPerTenant, cfg.QuerierForgetDelay, s.queueLength, s.discardedRequests, s.limits, registerer)
+	s.requestQueue = queue.NewRequestQueue(cfg.QuerierForgetDelay, s.queueLength, s.discardedRequests, s.limits, registerer)
 
 	s.queueDuration = promauto.With(registerer).NewHistogram(prometheus.HistogramOpts{
 		Name:    "cortex_query_scheduler_queue_duration_seconds",

pkg/scheduler/scheduler_test.go

@@ -35,8 +35,6 @@ const testMaxOutstandingPerTenant = 5
 func setupScheduler(t *testing.T, reg prometheus.Registerer) (*Scheduler, schedulerpb.SchedulerForFrontendClient, schedulerpb.SchedulerForQuerierClient) {
 	cfg := Config{}
 	flagext.DefaultValues(&cfg)
-	cfg.MaxOutstandingPerTenant = testMaxOutstandingPerTenant
-
 	s, err := NewScheduler(cfg, frontendv1.MockLimits{Queriers: 2, MockLimits: queue.MockLimits{MaxOutstanding: testMaxOutstandingPerTenant}}, log.NewNopLogger(), reg)
 	require.NoError(t, err)