Add querier.ingester-query-max-attempts to retry on partial data.

CHANGELOG.md

@@ -25,6 +25,7 @@
 * [ENHANCEMENT] OTLP: Support otlp metadata ingestion. #6617
 * [ENHANCEMENT] AlertManager: Add `keep_instance_in_the_ring_on_shutdown` and `tokens_file_path` configs for alertmanager ring. #6628
 * [ENHANCEMENT] Querier: Add metric and enhanced logging for query partial data. #6676
+* [ENHANCEMENT] Querier: Add `querier.ingester-query-max-attempts` to retry on partial data. #6714
 * [BUGFIX] Ingester: Avoid error or early throttling when READONLY ingesters are present in the ring #6517
 * [BUGFIX] Ingester: Fix labelset data race condition. #6573
 * [BUGFIX] Compactor: Cleaner should not put deletion marker for blocks with no-compact marker. #6576

docs/blocks-storage/querier.md

@@ -237,6 +237,11 @@ querier:
   # CLI flag: -querier.store-gateway-consistency-check-max-attempts
   [store_gateway_consistency_check_max_attempts: <int> | default = 3]
 
+  # The maximum number of times we attempt fetching data from ingesters for
+  # retryable errors (ex. partial data returned).
+  # CLI flag: -querier.ingester-query-max-attempts
+  [ingester_query_max_attempts: <int> | default = 1]
+
   # 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

docs/configuration/config-file-reference.md

@@ -4165,6 +4165,11 @@ store_gateway_client:
 # CLI flag: -querier.store-gateway-consistency-check-max-attempts
 [store_gateway_consistency_check_max_attempts: <int> | default = 3]
 
+# The maximum number of times we attempt fetching data from ingesters for
+# retryable errors (ex. partial data returned).
+# CLI flag: -querier.ingester-query-max-attempts
+[ingester_query_max_attempts: <int> | default = 1]
+
 # 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

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 = time.Millisecond
+const retryMaxBackoff = 5 * time.Millisecond
+
 // Distributor is the read interface to the distributor, made an interface here
 // to reduce package coupling.
 type Distributor interface {
@@ -38,36 +42,39 @@ type Distributor interface {
 	MetricsMetadata(ctx context.Context, req *client.MetricsMetadataRequest) ([]scrape.MetricMetadata, error)
 }
 
-func newDistributorQueryable(distributor Distributor, streamingMetdata bool, labelNamesWithMatchers bool, iteratorFn chunkIteratorFunc, queryIngestersWithin time.Duration, isPartialDataEnabled partialdata.IsCfgEnabledFunc) QueryableWithFilter {
+func newDistributorQueryable(distributor Distributor, streamingMetdata bool, labelNamesWithMatchers bool, iteratorFn chunkIteratorFunc, queryIngestersWithin time.Duration, isPartialDataEnabled partialdata.IsCfgEnabledFunc, ingesterQueryMaxAttempts int) QueryableWithFilter {
 	return distributorQueryable{
-		distributor:            distributor,
-		streamingMetdata:       streamingMetdata,
-		labelNamesWithMatchers: labelNamesWithMatchers,
-		iteratorFn:             iteratorFn,
-		queryIngestersWithin:   queryIngestersWithin,
-		isPartialDataEnabled:   isPartialDataEnabled,
+		distributor:              distributor,
+		streamingMetdata:         streamingMetdata,
+		labelNamesWithMatchers:   labelNamesWithMatchers,
+		iteratorFn:               iteratorFn,
+		queryIngestersWithin:     queryIngestersWithin,
+		isPartialDataEnabled:     isPartialDataEnabled,
+		ingesterQueryMaxAttempts: ingesterQueryMaxAttempts,
 	}
 }
 
 type distributorQueryable struct {
-	distributor            Distributor
-	streamingMetdata       bool
-	labelNamesWithMatchers bool
-	iteratorFn             chunkIteratorFunc
-	queryIngestersWithin   time.Duration
-	isPartialDataEnabled   partialdata.IsCfgEnabledFunc
+	distributor              Distributor
+	streamingMetdata         bool
+	labelNamesWithMatchers   bool
+	iteratorFn               chunkIteratorFunc
+	queryIngestersWithin     time.Duration
+	isPartialDataEnabled     partialdata.IsCfgEnabledFunc
+	ingesterQueryMaxAttempts int
 }
 
 func (d distributorQueryable) Querier(mint, maxt int64) (storage.Querier, error) {
 	return &distributorQuerier{
-		distributor:          d.distributor,
-		mint:                 mint,
-		maxt:                 maxt,
-		streamingMetadata:    d.streamingMetdata,
-		labelNamesMatchers:   d.labelNamesWithMatchers,
-		chunkIterFn:          d.iteratorFn,
-		queryIngestersWithin: d.queryIngestersWithin,
-		isPartialDataEnabled: d.isPartialDataEnabled,
+		distributor:              d.distributor,
+		mint:                     mint,
+		maxt:                     maxt,
+		streamingMetadata:        d.streamingMetdata,
+		labelNamesMatchers:       d.labelNamesWithMatchers,
+		chunkIterFn:              d.iteratorFn,
+		queryIngestersWithin:     d.queryIngestersWithin,
+		isPartialDataEnabled:     d.isPartialDataEnabled,
+		ingesterQueryMaxAttempts: d.ingesterQueryMaxAttempts,
 	}, nil
 }
 
@@ -77,13 +84,14 @@ func (d distributorQueryable) UseQueryable(now time.Time, _, queryMaxT int64) bo
 }
 
 type distributorQuerier struct {
-	distributor          Distributor
-	mint, maxt           int64
-	streamingMetadata    bool
-	labelNamesMatchers   bool
-	chunkIterFn          chunkIteratorFunc
-	queryIngestersWithin time.Duration
-	isPartialDataEnabled partialdata.IsCfgEnabledFunc
+	distributor              Distributor
+	mint, maxt               int64
+	streamingMetadata        bool
+	labelNamesMatchers       bool
+	chunkIterFn              chunkIteratorFunc
+	queryIngestersWithin     time.Duration
+	isPartialDataEnabled     partialdata.IsCfgEnabledFunc
+	ingesterQueryMaxAttempts int
 }
 
 // Select implements storage.Querier interface.
@@ -150,7 +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.distributor.QueryStream(ctx, model.Time(minT), model.Time(maxT), partialDataEnabled, matchers...)
+	results, err := q.queryWithRetry(ctx, func() (*client.QueryStreamResponse, error) {
+		return q.distributor.QueryStream(ctx, model.Time(minT), model.Time(maxT), partialDataEnabled, matchers...)
+	})
 
 	if err != nil && !partialdata.IsPartialDataError(err) {
 		return storage.ErrSeriesSet(err)
@@ -192,6 +202,33 @@ func (q *distributorQuerier) streamingSelect(ctx context.Context, sortSeries, pa
 	return seriesSet
 }
 
+func (q *distributorQuerier) queryWithRetry(ctx context.Context, queryFunc func() (*client.QueryStreamResponse, error)) (*client.QueryStreamResponse, error) {
+	if q.ingesterQueryMaxAttempts <= 1 {
+		return queryFunc()
+	}
+
+	var result *client.QueryStreamResponse
+	var err error
+
+	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
+}
+
 func (q *distributorQuerier) LabelValues(ctx context.Context, name string, hints *storage.LabelHints, matchers ...*labels.Matcher) ([]string, annotations.Annotations, error) {
 	var (
 		lvs []string
@@ -201,9 +238,13 @@ func (q *distributorQuerier) LabelValues(ctx context.Context, name string, hints
 	partialDataEnabled := q.partialDataEnabled(ctx)
 
 	if q.streamingMetadata {
-		lvs, err = q.distributor.LabelValuesForLabelNameStream(ctx, model.Time(q.mint), model.Time(q.maxt), model.LabelName(name), hints, partialDataEnabled, matchers...)
+		lvs, err = q.labelsWithRetry(ctx, func() ([]string, error) {
+			return q.distributor.LabelValuesForLabelNameStream(ctx, model.Time(q.mint), model.Time(q.maxt), model.LabelName(name), hints, partialDataEnabled, matchers...)
+		})
 	} else {
-		lvs, err = q.distributor.LabelValuesForLabelName(ctx, model.Time(q.mint), model.Time(q.maxt), model.LabelName(name), hints, partialDataEnabled, matchers...)
+		lvs, err = q.labelsWithRetry(ctx, func() ([]string, error) {
+			return q.distributor.LabelValuesForLabelName(ctx, model.Time(q.mint), model.Time(q.maxt), model.LabelName(name), hints, partialDataEnabled, matchers...)
+		})
 	}
 
 	if partialdata.IsPartialDataError(err) {
@@ -230,9 +271,13 @@ func (q *distributorQuerier) LabelNames(ctx context.Context, hints *storage.Labe
 	)
 
 	if q.streamingMetadata {
-		ln, err = q.distributor.LabelNamesStream(ctx, model.Time(q.mint), model.Time(q.maxt), hints, partialDataEnabled, matchers...)
+		ln, err = q.labelsWithRetry(ctx, func() ([]string, error) {
+			return q.distributor.LabelNamesStream(ctx, model.Time(q.mint), model.Time(q.maxt), hints, partialDataEnabled, matchers...)
+		})
 	} else {
-		ln, err = q.distributor.LabelNames(ctx, model.Time(q.mint), model.Time(q.maxt), hints, partialDataEnabled, matchers...)
+		ln, err = q.labelsWithRetry(ctx, func() ([]string, error) {
+			return q.distributor.LabelNames(ctx, model.Time(q.mint), model.Time(q.maxt), hints, partialDataEnabled, matchers...)
+		})
 	}
 
 	if partialdata.IsPartialDataError(err) {
@@ -243,6 +288,33 @@ func (q *distributorQuerier) LabelNames(ctx context.Context, hints *storage.Labe
 	return ln, nil, err
 }
 
+func (q *distributorQuerier) labelsWithRetry(ctx context.Context, labelsFunc func() ([]string, error)) ([]string, error) {
+	if q.ingesterQueryMaxAttempts == 1 {
+		return labelsFunc()
+	}
+
+	var result []string
+	var err error
+
+	retries := backoff.New(ctx, backoff.Config{
+		MinBackoff: retryMinBackoff,
+		MaxBackoff: retryMaxBackoff,
+		MaxRetries: q.ingesterQueryMaxAttempts,
+	})
+
+	for retries.Ongoing() {
+		result, err = labelsFunc()
+
+		if err == nil || !q.isRetryableError(err) {
+			return result, err
+		}
+
+		retries.Wait()
+	}
+
+	return result, err
+}
+
 // labelNamesWithMatchers performs the LabelNames call by calling ingester's MetricsForLabelMatchers method
 func (q *distributorQuerier) labelNamesWithMatchers(ctx context.Context, hints *storage.LabelHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]string, annotations.Annotations, error) {
 	log, ctx := spanlogger.New(ctx, "distributorQuerier.labelNamesWithMatchers")
@@ -297,6 +369,10 @@ func (q *distributorQuerier) partialDataEnabled(ctx context.Context) bool {
 	return q.isPartialDataEnabled != nil && q.isPartialDataEnabled(userID)
 }
 
+func (q *distributorQuerier) isRetryableError(err error) bool {
+	return partialdata.IsPartialDataError(err)
+}
+
 type distributorExemplarQueryable struct {
 	distributor Distributor
 }