Use backoff

Signed-off-by: Justin Jung <[email protected]>

Author: justinjung04

pkg/querier/blocks_store_queryable.go

@@ -61,8 +61,6 @@ const (
 	// store-gateways. If no more store-gateways are left (ie. due to lower replication
 	// factor) than we'll end the retries earlier.
 	maxFetchSeriesAttempts = 3
-
-	ingesterQueryMaxAttempts = 3
 )
 
 var (

pkg/querier/distributor_queryable.go

@@ -20,10 +20,14 @@ import (
 	"github.com/cortexproject/cortex/pkg/querier/series"
 	"github.com/cortexproject/cortex/pkg/tenant"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/backoff"
 	"github.com/cortexproject/cortex/pkg/util/chunkcompat"
 	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
 
+const retryMinBackoff = 10 * time.Second
+const retryMaxBackoff = time.Minute
+
 // Distributor is the read interface to the distributor, made an interface here
 // to reduce package coupling.
 type Distributor interface {
@@ -154,9 +158,9 @@ func (q *distributorQuerier) Select(ctx context.Context, sortSeries bool, sp *st
 }
 
 func (q *distributorQuerier) streamingSelect(ctx context.Context, sortSeries, partialDataEnabled bool, minT, maxT int64, matchers []*labels.Matcher) storage.SeriesSet {
-	results, err := q.queryWithRetry(func() (*client.QueryStreamResponse, error) {
+	results, err := q.queryWithRetry(ctx, func() (*client.QueryStreamResponse, error) {
 		return q.distributor.QueryStream(ctx, model.Time(minT), model.Time(maxT), partialDataEnabled, matchers...)
-	}, q.ingesterQueryMaxAttempts)
+	})
 
 	if err != nil && !partialdata.IsPartialDataError(err) {
 		return storage.ErrSeriesSet(err)
@@ -198,16 +202,24 @@ func (q *distributorQuerier) streamingSelect(ctx context.Context, sortSeries, pa
 	return seriesSet
 }
 
-func (q *distributorQuerier) queryWithRetry(queryFunc func() (*client.QueryStreamResponse, error), retryAttempt int) (*client.QueryStreamResponse, error) {
+func (q *distributorQuerier) queryWithRetry(ctx context.Context, queryFunc func() (*client.QueryStreamResponse, error)) (*client.QueryStreamResponse, error) {
 	var result *client.QueryStreamResponse
 	var err error
 
-	for i := 0; i < retryAttempt; i++ {
+	retries := backoff.New(ctx, backoff.Config{
+		MinBackoff: retryMinBackoff,
+		MaxBackoff: retryMaxBackoff,
+		MaxRetries: q.ingesterQueryMaxAttempts,
+	})
+
+	for retries.Ongoing() {
 		result, err = queryFunc()
 
 		if err == nil || !q.isRetryableError(err) {
 			return result, err
 		}
+
+		retries.Wait()
 	}
 
 	return result, err

pkg/querier/querier.go

@@ -128,7 +128,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.StringVar(&cfg.StoreGatewayAddresses, "querier.store-gateway-addresses", "", "Comma separated list of store-gateway addresses in DNS Service Discovery format. This option should be set when using the blocks storage and the store-gateway sharding is disabled (when enabled, the store-gateway instances form a ring and addresses are picked from the ring).")
 	f.BoolVar(&cfg.StoreGatewayQueryStatsEnabled, "querier.store-gateway-query-stats-enabled", true, "If enabled, store gateway query stats will be logged using `info` log level.")
 	f.IntVar(&cfg.StoreGatewayConsistencyCheckMaxAttempts, "querier.store-gateway-consistency-check-max-attempts", maxFetchSeriesAttempts, "The maximum number of times we attempt fetching missing blocks from different store-gateways. If no more store-gateways are left (ie. due to lower replication factor) than we'll end the retries earlier")
-	f.IntVar(&cfg.IngesterQueryMaxAttempts, "querier.ingester-query-max-attempts", ingesterQueryMaxAttempts, "The maximum number of times we attempt fetching data from ingesters for retryable errors.")
+	f.IntVar(&cfg.IngesterQueryMaxAttempts, "querier.ingester-query-max-attempts", 3, "The maximum number of times we attempt fetching data from ingesters for retryable errors.")
 	f.DurationVar(&cfg.LookbackDelta, "querier.lookback-delta", 5*time.Minute, "Time since the last sample after which a time series is considered stale and ignored by expression evaluations.")
 	f.DurationVar(&cfg.ShuffleShardingIngestersLookbackPeriod, "querier.shuffle-sharding-ingesters-lookback-period", 0, "When distributor's sharding strategy is shuffle-sharding and this setting is > 0, queriers fetch in-memory series from the minimum set of required ingesters, selecting only ingesters which may have received series since 'now - lookback period'. The lookback period should be greater or equal than the configured 'query store after' and 'query ingesters within'. If this setting is 0, queriers always query all ingesters (ingesters shuffle sharding on read path is disabled).")
 	f.BoolVar(&cfg.ThanosEngine, "querier.thanos-engine", false, "Experimental. Use Thanos promql engine https://github.com/thanos-io/promql-engine rather than the Prometheus promql engine.")