From f2979963d3d0a82a27b63e080305a4029b3b9f5c Mon Sep 17 00:00:00 2001 From: Vladimir Varankin Date: Sat, 11 Jul 2026 12:12:53 +0200 Subject: [PATCH 1/7] querier: limit number of blocks queried in one store-gw request --- cmd/mimir/config-descriptor.json | 11 + cmd/mimir/help-all.txt.tmpl | 2 + .../mimir/configure/about-versioning.md | 1 + .../configuration-parameters/index.md | 6 + operations/mimir/mimir-flags-defaults.json | 1 + pkg/querier/blocks_store_queryable.go | 499 +++++++++--------- ...locks_store_queryable_compartments_test.go | 48 +- pkg/querier/blocks_store_queryable_search.go | 96 ++-- .../blocks_store_queryable_search_test.go | 48 +- pkg/querier/blocks_store_queryable_test.go | 249 ++++----- pkg/querier/blocks_store_replicated_set.go | 16 +- .../blocks_store_replicated_set_test.go | 136 +++-- pkg/util/validation/limits.go | 7 + 13 files changed, 611 insertions(+), 509 deletions(-) diff --git a/cmd/mimir/config-descriptor.json b/cmd/mimir/config-descriptor.json index c6fa6ca4e3e..2d4e14e017f 100644 --- a/cmd/mimir/config-descriptor.json +++ b/cmd/mimir/config-descriptor.json @@ -6085,6 +6085,17 @@ "fieldType": "int", "fieldCategory": "experimental" }, + { + "kind": "field", + "name": "max_blocks_per_store_request", + "required": false, + "desc": "Maximum number of blocks that a querier will reference in a single request to a store-gateway. When a request would exceed this, it is split into multiple requests to the same store-gateway. 0 disables the limit.", + "fieldValue": null, + "fieldDefaultValue": 0, + "fieldFlag": "querier.max-blocks-per-store-request", + "fieldType": "int", + "fieldCategory": "experimental" + }, { "kind": "field", "name": "max_query_lookback", diff --git a/cmd/mimir/help-all.txt.tmpl b/cmd/mimir/help-all.txt.tmpl index 5aa33b913b3..52bfc9ac881 100644 --- a/cmd/mimir/help-all.txt.tmpl +++ b/cmd/mimir/help-all.txt.tmpl @@ -2473,6 +2473,8 @@ Usage of ./cmd/mimir/mimir: Maximum number of label names allowed to be queried in a single /api/v1/cardinality/label_values API call. (default 100) -querier.lookback-delta duration Time since the last sample after which a time series is considered stale and ignored by expression evaluations. This config option should be set on query-frontend too when query sharding is enabled. (default 5m0s) + -querier.max-blocks-per-store-request int + [experimental] Maximum number of blocks that a querier will reference in a single request to a store-gateway. When a request would exceed this, it is split into multiple requests to the same store-gateway. 0 disables the limit. -querier.max-concurrent int The number of workers running in each querier process. This setting limits the maximum number of concurrent queries in each querier. The minimum value is four; lower values are ignored and set to the minimum (default 8) -querier.max-concurrent-remote-read-queries int diff --git a/docs/sources/mimir/configure/about-versioning.md b/docs/sources/mimir/configure/about-versioning.md index 79f3ae7bec9..c0c462bda33 100644 --- a/docs/sources/mimir/configure/about-versioning.md +++ b/docs/sources/mimir/configure/about-versioning.md @@ -232,6 +232,7 @@ The following features are currently experimental: - Max concurrency for tenant federated queries (`-tenant-federation.max-concurrent`) - [Mimir query engine](https://grafana.com/docs/mimir//references/architecture/mimir-query-engine) (`-querier.query-engine` and `-querier.enable-query-engine-fallback`, and all flags beginning with `-querier.mimir-query-engine`) - Maximum estimated memory consumption per query limit (`-querier.max-estimated-memory-consumption-per-query`) + - Maximum number of blocks per store-gateway request (`-querier.max-blocks-per-store-request`) - Enable the experimental Prometheus feature for delayed name removal (`-querier.enable-delayed-name-removal`) - Ignore deletion marks while querying delay (`-blocks-storage.bucket-store.ignore-deletion-marks-while-querying-delay`) - Querier ring (all flags beginning with `-querier.ring`) diff --git a/docs/sources/mimir/configure/configuration-parameters/index.md b/docs/sources/mimir/configure/configuration-parameters/index.md index 4796792e21e..9c046d1658c 100644 --- a/docs/sources/mimir/configure/configuration-parameters/index.md +++ b/docs/sources/mimir/configure/configuration-parameters/index.md @@ -4743,6 +4743,12 @@ The `limits` block configures default and per-tenant limits imposed by component # CLI flag: -querier.max-estimated-memory-consumption-per-query [max_estimated_memory_consumption_per_query: | default = 0] +# (experimental) Maximum number of blocks that a querier will reference in a +# single request to a store-gateway. When a request would exceed this, it is +# split into multiple requests to the same store-gateway. 0 disables the limit. +# CLI flag: -querier.max-blocks-per-store-request +[max_blocks_per_store_request: | default = 0] + # Limit how long back data (series and metadata) can be queried, up until # duration ago. This limit is enforced in the query-frontend, querier # and ruler for instant, range and remote read queries. For metadata queries diff --git a/operations/mimir/mimir-flags-defaults.json b/operations/mimir/mimir-flags-defaults.json index f2f9693f33f..65b844f6bd7 100644 --- a/operations/mimir/mimir-flags-defaults.json +++ b/operations/mimir/mimir-flags-defaults.json @@ -424,6 +424,7 @@ "querier.max-fetched-series-per-query": 0, "querier.max-fetched-chunk-bytes-per-query": 0, "querier.max-estimated-memory-consumption-per-query": 0, + "querier.max-blocks-per-store-request": 0, "querier.max-query-lookback": 0, "querier.max-partial-query-length": 0, "querier.max-query-parallelism": 14, diff --git a/pkg/querier/blocks_store_queryable.go b/pkg/querier/blocks_store_queryable.go index ef4521b11ed..e7329b6b38f 100644 --- a/pkg/querier/blocks_store_queryable.go +++ b/pkg/querier/blocks_store_queryable.go @@ -65,10 +65,11 @@ var errAlreadyClosed = errors.New("querier already closed") type BlocksStoreSet interface { services.Service - // GetClientsFor returns the store gateway clients that should be used to - // query the set of blocks in input. The exclude parameter is the map of - // blocks -> store-gateway addresses that should be excluded. - GetClientsFor(userID string, blocks bucketindex.Blocks, exclude map[ulid.ULID][]string) (map[BlocksStoreClient][]ulid.ULID, error) + // GetClientsFor returns the store-gateway clients that should be used to query the given set of blocks, and for + // each client the blocks it should be queried for. The blocks of a client are grouped into one or more partitions, + // each to be queried separately. + // The exclude parameter is the map of blocks -> store-gateway addresses that should be excluded. + GetClientsFor(userID string, blocks bucketindex.Blocks, exclude map[ulid.ULID][]string) (map[BlocksStoreClient][][]ulid.ULID, error) } // BlocksFinder is the interface used to find blocks for a given user and time range. @@ -100,6 +101,7 @@ type BlocksStoreLimits interface { MaxLabelsQueryLength(userID string) time.Duration MaxChunksPerQuery(userID string) int + MaxBlocksPerStoreRequest(userID string) int StoreGatewayTenantShardSize(userID string) int StoreGatewayTenantShardSizePerZone(userID string) int } @@ -515,7 +517,7 @@ func (q *blocksStoreQuerier) LabelNames(ctx context.Context, hints *storage.Labe resWarnings annotations.Annotations ) - queryF := func(ctx context.Context, clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { + queryF := func(ctx context.Context, clients map[BlocksStoreClient][][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { nameSets, warnings, queriedBlocks, err := q.fetchLabelNamesFromStore(ctx, clients, minT, maxT, tenantID, hints, convertedMatchers, indexMeta) if err != nil { return nil, err @@ -562,7 +564,7 @@ func (q *blocksStoreQuerier) LabelValues(ctx context.Context, name string, hints resWarnings annotations.Annotations ) - queryF := func(ctx context.Context, clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { + queryF := func(ctx context.Context, clients map[BlocksStoreClient][][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { valueSets, warnings, queriedBlocks, err := q.fetchLabelValuesFromStore(ctx, name, clients, minT, maxT, tenantID, hints, matchers, indexMeta) if err != nil { return nil, err @@ -623,7 +625,7 @@ func (q *blocksStoreQuerier) selectSorted(ctx context.Context, sp *storage.Selec return storage.ErrSeriesSet(err) } - queryF := func(ctx context.Context, clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { + queryF := func(ctx context.Context, clients map[BlocksStoreClient][][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { seriesSets, queriedBlocks, warnings, streamReaders, chunkEstimator, err := q.fetchSeriesFromStores(ctx, sp, clients, minT, maxT, tenantID, convertedMatchers, memoryTracker, indexMeta) if err != nil { return nil, err @@ -699,7 +701,7 @@ func (q *blocksStoreQuerier) startBuffering(streamReaders []*storeGatewayStreamR return nil } -type queryFunc func(ctx context.Context, clients map[BlocksStoreClient][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) +type queryFunc func(ctx context.Context, clients map[BlocksStoreClient][][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) // storeGatewayQueryStats holds the per-query store-gateway histogram values. queryWithConsistencyCheck // returns it (rather than observing the histograms itself) so the caller observes them once per query. @@ -793,11 +795,13 @@ func (q *blocksStoreQuerier) queryWithConsistencyCheck( spanLog.DebugLog("msg", "received series from all store-gateways", "queried blocks", strings.Join(convertULIDsToString(queriedBlocks), " ")) // Update the map of blocks we attempted to query. - for client, blockIDs := range clients { + for client, partitions := range clients { touchedStores[client.RemoteAddress()] = struct{}{} - for _, blockID := range blockIDs { - attemptedBlocks[blockID] = append(attemptedBlocks[blockID], client.RemoteAddress()) + for _, part := range partitions { + for _, blockID := range part { + attemptedBlocks[blockID] = append(attemptedBlocks[blockID], client.RemoteAddress()) + } } } @@ -1013,7 +1017,7 @@ func canBlockWithCompactorShardIndexContainQueryShard(queryShardIndex, queryShar func (q *blocksStoreQuerier) fetchSeriesFromStores( ctx context.Context, sp *storage.SelectHints, - clients map[BlocksStoreClient][]ulid.ULID, + clients map[BlocksStoreClient][][]ulid.ULID, minT int64, maxT int64, tenantID string, @@ -1041,162 +1045,165 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores( debugQuery := chunkinfologger.IsChunkInfoLoggingEnabled(ctx) - // Concurrently fetch series from all clients. - for c, blockIDs := range clients { - g.Go(func() error { - clientSpanLog, clientCtx := spanlogger.New(reqCtx, spanLog, tracer, "blocksStoreQuerier.fetchSeriesFromStores") - defer clientSpanLog.Finish() - clientSpanLog.SetTag("store_gateway_address", c.RemoteAddress()) - - var skipChunks bool - if sp != nil { - // See: https://github.com/prometheus/prometheus/pull/8050 - // TODO(goutham): we should ideally be passing the hints down to the storage layer - // and let the TSDB return us data with no chunks as in prometheus#8050. - // But this is an acceptable workaround for now. - skipChunks = sp.Func == "series" - } - - req, err := createSeriesRequest(minT, maxT, matchers, skipChunks, sp, blockIDs, q.streamingChunksBatchSize) - if err != nil { - return errors.Wrapf(err, "failed to create series request") - } - - stream, err := c.Series(clientCtx, req) - if err == nil { - mtx.Lock() - streams = append(streams, stream) - mtx.Unlock() - err = gCtx.Err() - } - if err != nil { - if shouldRetry(err) { - level.Warn(clientSpanLog).Log("msg", "failed to fetch series", "remote", c.RemoteAddress(), "err", err) - return nil + // Concurrently fetch series from all clients. A client may have multiple partitions; each + // partition is one independent RPC to the same store-gateway. + for c, partitions := range clients { + for _, blockIDs := range partitions { + g.Go(func() error { + clientSpanLog, clientCtx := spanlogger.New(reqCtx, spanLog, tracer, "blocksStoreQuerier.fetchSeriesFromStores") + defer clientSpanLog.Finish() + clientSpanLog.SetTag("store_gateway_address", c.RemoteAddress()) + + var skipChunks bool + if sp != nil { + // See: https://github.com/prometheus/prometheus/pull/8050 + // TODO(goutham): we should ideally be passing the hints down to the storage layer + // and let the TSDB return us data with no chunks as in prometheus#8050. + // But this is an acceptable workaround for now. + skipChunks = sp.Func == "series" } - return err - } - - myStreamingSeriesLabels := []labels.Labels(nil) - var myWarnings annotations.Annotations - myQueriedBlocks := []ulid.ULID(nil) - indexBytesFetched := uint64(0) + req, err := createSeriesRequest(minT, maxT, matchers, skipChunks, sp, blockIDs, q.streamingChunksBatchSize) + if err != nil { + return errors.Wrapf(err, "failed to create series request") + } - deduplicator, err := limiter.SeriesLabelsDeduplicatorFromContext(ctx) - if err != nil { - return err - } + stream, err := c.Series(clientCtx, req) + if err == nil { + mtx.Lock() + streams = append(streams, stream) + mtx.Unlock() + err = gCtx.Err() + } + if err != nil { + if shouldRetry(err) { + level.Warn(clientSpanLog).Log("msg", "failed to fetch series", "remote", c.RemoteAddress(), "err", err) + return nil + } - for { - // Ensure the context hasn't been canceled in the meanwhile (eg. an error occurred - // in another goroutine). - if gCtx.Err() != nil { - return gCtx.Err() + return err } - var err error - var isEOS bool - var shouldRetry bool + myStreamingSeriesLabels := []labels.Labels(nil) + var myWarnings annotations.Annotations + myQueriedBlocks := []ulid.ULID(nil) + indexBytesFetched := uint64(0) - myWarnings, myQueriedBlocks, myStreamingSeriesLabels, indexBytesFetched, isEOS, shouldRetry, err = q.receiveMessage( - c, stream, queryLimiter, memoryTracker, deduplicator, myWarnings, myQueriedBlocks, myStreamingSeriesLabels, indexBytesFetched, - ) - if errors.Is(err, io.EOF) { - util.CloseAndExhaust[*storepb.SeriesResponse](stream) //nolint:errcheck - break - } - if shouldRetry { - level.Warn(clientSpanLog).Log("msg", "failed to receive series", "remote", c.RemoteAddress(), "err", err) - return nil - } + deduplicator, err := limiter.SeriesLabelsDeduplicatorFromContext(ctx) if err != nil { return err } - if isEOS { - // If we aren't expecting any series from this stream, close it now. - if len(myStreamingSeriesLabels) == 0 { - util.CloseAndExhaust[*storepb.SeriesResponse](stream) //nolint:errcheck + for { + // Ensure the context hasn't been canceled in the meanwhile (eg. an error occurred + // in another goroutine). + if gCtx.Err() != nil { + return gCtx.Err() } - // We expect "end of stream" to be sent after the hints and the stats have been sent, so we can break out of the loop now. - break - } - } + var err error + var isEOS bool + var shouldRetry bool - // debug - var chunkInfo *chunkinfologger.ChunkInfoLogger - if debugQuery { - traceID, spanID, _ := tracing.ExtractTraceSpanID(ctx) - chunkInfo = chunkinfologger.NewChunkInfoLogger("store-gateway message", traceID, spanID, q.logger, chunkinfologger.ChunkInfoLoggingFromContext(ctx)) - chunkInfo.LogSelect("store-gateway", minT, maxT) - } + myWarnings, myQueriedBlocks, myStreamingSeriesLabels, indexBytesFetched, isEOS, shouldRetry, err = q.receiveMessage( + c, stream, queryLimiter, memoryTracker, deduplicator, myWarnings, myQueriedBlocks, myStreamingSeriesLabels, indexBytesFetched, + ) + if errors.Is(err, io.EOF) { + util.CloseAndExhaust[*storepb.SeriesResponse](stream) //nolint:errcheck + break + } + if shouldRetry { + level.Warn(clientSpanLog).Log("msg", "failed to receive series", "remote", c.RemoteAddress(), "err", err) + return nil + } + if err != nil { + return err + } - reqStats.AddFetchedIndexBytes(indexBytesFetched) - var streamReader *storeGatewayStreamReader - if len(myStreamingSeriesLabels) > 0 { - // FetchedChunks and FetchedChunkBytes are added by the SeriesChunksStreamReader. - reqStats.AddFetchedSeries(uint64(len(myStreamingSeriesLabels))) - - if req.SkipChunks { - // If we aren't creating a stream reader for reading chunks, we need to close the stream - // ourselves. It's safe to close the stream multiple times so we don't worry about closing - // all streams below when there's an error. - if err := util.CloseAndExhaust[*storepb.SeriesResponse](stream); err != nil { - level.Warn(clientSpanLog).Log("msg", "closing store-gateway client stream failed", "err", err) + if isEOS { + // If we aren't expecting any series from this stream, close it now. + if len(myStreamingSeriesLabels) == 0 { + util.CloseAndExhaust[*storepb.SeriesResponse](stream) //nolint:errcheck + } + + // We expect "end of stream" to be sent after the hints and the stats have been sent, so we can break out of the loop now. + break } - } else { - streamReader = newStoreGatewayStreamReader(clientCtx, stream, len(myStreamingSeriesLabels), queryLimiter, memoryTracker, reqStats, q.metrics, q.logger) } - clientSpanLog.DebugLog( - "msg", "received streaming series from store-gateway", - "instance", c.RemoteAddress(), - "fetched series", len(myStreamingSeriesLabels), - "fetched index bytes", indexBytesFetched, - "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), - "queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " "), - ) - } else { - clientSpanLog.DebugLog( - "msg", "received no series from store-gateway", - "instance", c.RemoteAddress(), - "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), - "queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " "), - ) - } - - // Store the result. - mtx.Lock() - if len(myStreamingSeriesLabels) > 0 { - if chunkInfo != nil { - chunkInfo.SetMsg("store-gateway streaming") + // debug + var chunkInfo *chunkinfologger.ChunkInfoLogger + if debugQuery { + traceID, spanID, _ := tracing.ExtractTraceSpanID(ctx) + chunkInfo = chunkinfologger.NewChunkInfoLogger("store-gateway message", traceID, spanID, q.logger, chunkinfologger.ChunkInfoLoggingFromContext(ctx)) + chunkInfo.LogSelect("store-gateway", minT, maxT) } - if req.SkipChunks { - noChunkSeries := make([]storage.Series, 0, len(myStreamingSeriesLabels)) - for _, lbls := range myStreamingSeriesLabels { - noChunkSeries = append(noChunkSeries, series.NewConcreteSeries(lbls, nil, nil)) + reqStats.AddFetchedIndexBytes(indexBytesFetched) + var streamReader *storeGatewayStreamReader + if len(myStreamingSeriesLabels) > 0 { + // FetchedChunks and FetchedChunkBytes are added by the SeriesChunksStreamReader. + reqStats.AddFetchedSeries(uint64(len(myStreamingSeriesLabels))) + + if req.SkipChunks { + // If we aren't creating a stream reader for reading chunks, we need to close the stream + // ourselves. It's safe to close the stream multiple times so we don't worry about closing + // all streams below when there's an error. + if err := util.CloseAndExhaust[*storepb.SeriesResponse](stream); err != nil { + level.Warn(clientSpanLog).Log("msg", "closing store-gateway client stream failed", "err", err) + } + } else { + streamReader = newStoreGatewayStreamReader(clientCtx, stream, len(myStreamingSeriesLabels), queryLimiter, memoryTracker, reqStats, q.metrics, q.logger) } - seriesSets = append(seriesSets, series.NewConcreteSeriesSetFromSortedSeries(noChunkSeries)) + clientSpanLog.DebugLog( + "msg", "received streaming series from store-gateway", + "instance", c.RemoteAddress(), + "fetched series", len(myStreamingSeriesLabels), + "fetched index bytes", indexBytesFetched, + "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), + "queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " "), + ) } else { - seriesSets = append(seriesSets, &blockStreamingQuerierSeriesSet{ - series: myStreamingSeriesLabels, - streamReader: streamReader, - chunkInfo: chunkInfo, - remoteAddress: c.RemoteAddress(), - }) - streamReaders = append(streamReaders, streamReader) + clientSpanLog.DebugLog( + "msg", "received no series from store-gateway", + "instance", c.RemoteAddress(), + "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), + "queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " "), + ) } - } - warnings.Merge(myWarnings) - queriedBlocks = append(queriedBlocks, myQueriedBlocks...) - mtx.Unlock() - return nil - }) + // Store the result. + mtx.Lock() + if len(myStreamingSeriesLabels) > 0 { + if chunkInfo != nil { + chunkInfo.SetMsg("store-gateway streaming") + } + + if req.SkipChunks { + noChunkSeries := make([]storage.Series, 0, len(myStreamingSeriesLabels)) + for _, lbls := range myStreamingSeriesLabels { + noChunkSeries = append(noChunkSeries, series.NewConcreteSeries(lbls, nil, nil)) + } + + seriesSets = append(seriesSets, series.NewConcreteSeriesSetFromSortedSeries(noChunkSeries)) + } else { + seriesSets = append(seriesSets, &blockStreamingQuerierSeriesSet{ + series: myStreamingSeriesLabels, + streamReader: streamReader, + chunkInfo: chunkInfo, + remoteAddress: c.RemoteAddress(), + }) + streamReaders = append(streamReaders, streamReader) + } + } + warnings.Merge(myWarnings) + queriedBlocks = append(queriedBlocks, myQueriedBlocks...) + mtx.Unlock() + + return nil + }) + } } // Wait until all client requests complete. @@ -1316,7 +1323,7 @@ func shouldRetry(err error) bool { func (q *blocksStoreQuerier) fetchLabelNamesFromStore( ctx context.Context, - clients map[BlocksStoreClient][]ulid.ULID, + clients map[BlocksStoreClient][][]ulid.ULID, minT int64, maxT int64, tenantID string, @@ -1335,64 +1342,67 @@ func (q *blocksStoreQuerier) fetchLabelNamesFromStore( spanLog = spanlogger.FromContext(ctx, q.logger) ) - // Concurrently fetch series from all clients. - for c, blockIDs := range clients { - g.Go(func() error { - req, err := createLabelNamesRequest(minT, maxT, blockIDs, hints, matchers) - if err != nil { - return errors.Wrapf(err, "failed to create label names request") - } - - namesResp, err := c.LabelNames(gCtx, req) - if err != nil { - if shouldRetry(err) { - level.Warn(spanLog).Log("msg", "failed to fetch label names; error is retriable", "remote", c.RemoteAddress(), "err", err) - return nil + // Concurrently fetch series from all clients. A client may have multiple partitions; each + // partition is one independent RPC to the same store-gateway. + for c, partitions := range clients { + for _, blockIDs := range partitions { + g.Go(func() error { + req, err := createLabelNamesRequest(minT, maxT, blockIDs, hints, matchers) + if err != nil { + return errors.Wrapf(err, "failed to create label names request") } - return fmt.Errorf("non-retriable error while fetching label names from store: %w", err) - } - myQueriedBlocks := []ulid.ULID(nil) - if namesResp.ResponseHints != nil { - ids, err := convertBlockHintsToULIDs(namesResp.ResponseHints.QueriedBlocks) + namesResp, err := c.LabelNames(gCtx, req) if err != nil { - return errors.Wrapf(err, "failed to parse queried block IDs from received hints") + if shouldRetry(err) { + level.Warn(spanLog).Log("msg", "failed to fetch label names; error is retriable", "remote", c.RemoteAddress(), "err", err) + return nil + } + return fmt.Errorf("non-retriable error while fetching label names from store: %w", err) } - myQueriedBlocks = ids - } else if namesResp.Hints != nil { //nolint:staticcheck // Ignore SA1019. This use will be removed in Mimir 3.2 - // Note that we use a different but equivalent hints type for the opaque field. - resHints := hintspb.LabelNamesResponseHints{} - //nolint:staticcheck // Ignore SA1019. This use will be removed in Mimir 3.2 - if err := types.UnmarshalAny(namesResp.Hints, &resHints); err != nil { - return errors.Wrapf(err, "failed to unmarshal label names hints from %s", c.RemoteAddress()) - } + myQueriedBlocks := []ulid.ULID(nil) + if namesResp.ResponseHints != nil { + ids, err := convertBlockHintsToULIDs(namesResp.ResponseHints.QueriedBlocks) + if err != nil { + return errors.Wrapf(err, "failed to parse queried block IDs from received hints") + } - ids, err := convertBlockHintsToULIDsOpaque(resHints.QueriedBlocks) - if err != nil { - return errors.Wrapf(err, "failed to parse queried block IDs from received hints") + myQueriedBlocks = ids + } else if namesResp.Hints != nil { //nolint:staticcheck // Ignore SA1019. This use will be removed in Mimir 3.2 + // Note that we use a different but equivalent hints type for the opaque field. + resHints := hintspb.LabelNamesResponseHints{} + //nolint:staticcheck // Ignore SA1019. This use will be removed in Mimir 3.2 + if err := types.UnmarshalAny(namesResp.Hints, &resHints); err != nil { + return errors.Wrapf(err, "failed to unmarshal label names hints from %s", c.RemoteAddress()) + } + + ids, err := convertBlockHintsToULIDsOpaque(resHints.QueriedBlocks) + if err != nil { + return errors.Wrapf(err, "failed to parse queried block IDs from received hints") + } + + myQueriedBlocks = ids } - myQueriedBlocks = ids - } + spanLog.DebugLog("msg", "received label names from store-gateway", + "instance", c, + "num labels", len(namesResp.Names), + "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), + "queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " ")) - spanLog.DebugLog("msg", "received label names from store-gateway", - "instance", c, - "num labels", len(namesResp.Names), - "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), - "queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " ")) - - // Store the result. - mtx.Lock() - nameSets = append(nameSets, namesResp.Names) - for _, w := range namesResp.Warnings { - warnings.Add(errors.New(w)) - } - queriedBlocks = append(queriedBlocks, myQueriedBlocks...) - mtx.Unlock() + // Store the result. + mtx.Lock() + nameSets = append(nameSets, namesResp.Names) + for _, w := range namesResp.Warnings { + warnings.Add(errors.New(w)) + } + queriedBlocks = append(queriedBlocks, myQueriedBlocks...) + mtx.Unlock() - return nil - }) + return nil + }) + } } // Wait until all client requests complete. @@ -1406,7 +1416,7 @@ func (q *blocksStoreQuerier) fetchLabelNamesFromStore( func (q *blocksStoreQuerier) fetchLabelValuesFromStore( ctx context.Context, name string, - clients map[BlocksStoreClient][]ulid.ULID, + clients map[BlocksStoreClient][][]ulid.ULID, minT int64, maxT int64, tenantID string, @@ -1425,67 +1435,70 @@ func (q *blocksStoreQuerier) fetchLabelValuesFromStore( spanLog = spanlogger.FromContext(ctx, q.logger) ) - // Concurrently fetch series from all clients. - for c, blockIDs := range clients { - g.Go(func() error { - req, err := createLabelValuesRequest(minT, maxT, name, blockIDs, hints, matchers...) - if err != nil { - return errors.Wrapf(err, "failed to create label values request") - } - - valuesResp, err := c.LabelValues(gCtx, req) - if err != nil { - if shouldRetry(err) { - level.Warn(spanLog).Log("msg", "failed to fetch label values; error is retriable", "remote", c.RemoteAddress(), "err", err) - return nil + // Concurrently fetch series from all clients. A client may have multiple partitions; each + // partition is one independent RPC to the same store-gateway. + for c, partitions := range clients { + for _, blockIDs := range partitions { + g.Go(func() error { + req, err := createLabelValuesRequest(minT, maxT, name, blockIDs, hints, matchers...) + if err != nil { + return errors.Wrapf(err, "failed to create label values request") } - return fmt.Errorf("non-retriable error while fetching label values from store: %w", err) - } - myQueriedBlocks := []ulid.ULID(nil) - if valuesResp.ResponseHints != nil { - ids, err := convertBlockHintsToULIDs(valuesResp.ResponseHints.QueriedBlocks) + valuesResp, err := c.LabelValues(gCtx, req) if err != nil { - return errors.Wrapf(err, "failed to parse queried block IDs from received hints") + if shouldRetry(err) { + level.Warn(spanLog).Log("msg", "failed to fetch label values; error is retriable", "remote", c.RemoteAddress(), "err", err) + return nil + } + return fmt.Errorf("non-retriable error while fetching label values from store: %w", err) } - myQueriedBlocks = ids - } else if valuesResp.Hints != nil { //nolint:staticcheck // Ignore SA1019. This use will be removed in Mimir 3.2 - // Note that we use a different but equivalent hints type for the opaque field. - resHints := hintspb.LabelValuesResponseHints{} - //nolint:staticcheck // Ignore SA1019. This use will be removed in Mimir 3.2 - if err := types.UnmarshalAny(valuesResp.Hints, &resHints); err != nil { - return errors.Wrapf(err, "failed to unmarshal label values hints from %s", c.RemoteAddress()) - } + myQueriedBlocks := []ulid.ULID(nil) + if valuesResp.ResponseHints != nil { + ids, err := convertBlockHintsToULIDs(valuesResp.ResponseHints.QueriedBlocks) + if err != nil { + return errors.Wrapf(err, "failed to parse queried block IDs from received hints") + } - ids, err := convertBlockHintsToULIDsOpaque(resHints.QueriedBlocks) - if err != nil { - return errors.Wrapf(err, "failed to parse queried block IDs from received hints") - } + myQueriedBlocks = ids + } else if valuesResp.Hints != nil { //nolint:staticcheck // Ignore SA1019. This use will be removed in Mimir 3.2 + // Note that we use a different but equivalent hints type for the opaque field. + resHints := hintspb.LabelValuesResponseHints{} + //nolint:staticcheck // Ignore SA1019. This use will be removed in Mimir 3.2 + if err := types.UnmarshalAny(valuesResp.Hints, &resHints); err != nil { + return errors.Wrapf(err, "failed to unmarshal label values hints from %s", c.RemoteAddress()) + } - myQueriedBlocks = ids - } + ids, err := convertBlockHintsToULIDsOpaque(resHints.QueriedBlocks) + if err != nil { + return errors.Wrapf(err, "failed to parse queried block IDs from received hints") + } - spanLog.DebugLog("msg", "received label values from store-gateway", - "instance", c.RemoteAddress(), - "num values", len(valuesResp.Values), - "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), - "queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " ")) + myQueriedBlocks = ids + } - // Values returned need not be sorted, but we need them to be sorted so we can merge. - slices.Sort(valuesResp.Values) + spanLog.DebugLog("msg", "received label values from store-gateway", + "instance", c.RemoteAddress(), + "num values", len(valuesResp.Values), + "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), + "queried blocks", strings.Join(convertULIDsToString(myQueriedBlocks), " ")) - // Store the result. - mtx.Lock() - valueSets = append(valueSets, valuesResp.Values) - for _, w := range valuesResp.Warnings { - warnings.Add(errors.New(w)) - } - queriedBlocks = append(queriedBlocks, myQueriedBlocks...) - mtx.Unlock() + // Values returned need not be sorted, but we need them to be sorted so we can merge. + slices.Sort(valuesResp.Values) - return nil - }) + // Store the result. + mtx.Lock() + valueSets = append(valueSets, valuesResp.Values) + for _, w := range valuesResp.Warnings { + warnings.Add(errors.New(w)) + } + queriedBlocks = append(queriedBlocks, myQueriedBlocks...) + mtx.Unlock() + + return nil + }) + } } // Wait until all client requests complete. diff --git a/pkg/querier/blocks_store_queryable_compartments_test.go b/pkg/querier/blocks_store_queryable_compartments_test.go index 1e7ae98ac5b..5043331b291 100644 --- a/pkg/querier/blocks_store_queryable_compartments_test.go +++ b/pkg/querier/blocks_store_queryable_compartments_test.go @@ -149,25 +149,25 @@ func TestBlocksStoreQuerier_Compartments_LabelNames(t *testing.T) { // within each response, as the store-gateways return them. stores := []BlocksStoreSet{ &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ Names: []string{"label_from_compartment_0", "shared_label"}, ResponseHints: mockNamesResponseHints(block0), }, - }: {block0}, + }: {{block0}}, }, }}, &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ Names: []string{"label_from_compartment_1", "shared_label"}, ResponseHints: mockNamesResponseHints(block1), }, - }: {block1}, + }: {{block1}}, }, }}, } @@ -324,25 +324,25 @@ func TestBlocksStoreQuerier_Compartments_LabelValues(t *testing.T) { // within each response, as the store-gateways return them. stores := []BlocksStoreSet{ &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelValuesResponse: &storepb.LabelValuesResponse{ Values: []string{"shared_value", "value_from_compartment_0"}, ResponseHints: mockValuesResponseHints(block0), }, - }: {block0}, + }: {{block0}}, }, }}, &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedLabelValuesResponse: &storepb.LabelValuesResponse{ Values: []string{"shared_value", "value_from_compartment_1"}, ResponseHints: mockValuesResponseHints(block1), }, - }: {block1}, + }: {{block1}}, }, }}, } @@ -408,21 +408,21 @@ func TestBlocksStoreQuerier_Compartments_Select(t *testing.T) { stores := []BlocksStoreSet{ &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(seriesA, minT, 1). addBlocks(block0). build(), - }: {block0}, + }: {{block0}}, }, }}, &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(seriesB, minT, 2). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, }, }}, } @@ -474,25 +474,25 @@ func TestBlocksStoreQuerier_Compartments_Select(t *testing.T) { stores := []BlocksStoreSet{ &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(seriesA, minT, 1). addBlocks(block0). build(), - }: {block0}, + }: {{block0}}, }, }}, &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(seriesB, minT, 2). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, }, }}, } @@ -575,7 +575,7 @@ func TestBlocksStoreQuerier_Compartments_SearchLabelNames(t *testing.T) { // store-gateway returns its values sorted by value ascending (the default search ordering). stores := []BlocksStoreSet{ &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &searchStoreGatewayClientMock{ storeGatewayClientMock: storeGatewayClientMock{remoteAddr: "1.1.1.1"}, searchLabelNamesBatches: []*storepb.SearchResultBatch{{ @@ -586,11 +586,11 @@ func TestBlocksStoreQuerier_Compartments_SearchLabelNames(t *testing.T) { }, }}, queriedBlockIDs: []ulid.ULID{block0}, - }: {block0}, + }: {{block0}}, }, }}, &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &searchStoreGatewayClientMock{ storeGatewayClientMock: storeGatewayClientMock{remoteAddr: "2.2.2.2"}, searchLabelNamesBatches: []*storepb.SearchResultBatch{{ @@ -601,7 +601,7 @@ func TestBlocksStoreQuerier_Compartments_SearchLabelNames(t *testing.T) { }, }}, queriedBlockIDs: []ulid.ULID{block1}, - }: {block1}, + }: {{block1}}, }, }}, } @@ -641,7 +641,7 @@ func TestBlocksStoreQuerier_Compartments_SearchLabelValues(t *testing.T) { // store-gateway returns its values sorted by value ascending (the default search ordering). stores := []BlocksStoreSet{ &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &searchStoreGatewayClientMock{ storeGatewayClientMock: storeGatewayClientMock{remoteAddr: "1.1.1.1"}, searchLabelValuesBatches: []*storepb.SearchResultBatch{{ @@ -652,11 +652,11 @@ func TestBlocksStoreQuerier_Compartments_SearchLabelValues(t *testing.T) { }, }}, queriedBlockIDs: []ulid.ULID{block0}, - }: {block0}, + }: {{block0}}, }, }}, &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &searchStoreGatewayClientMock{ storeGatewayClientMock: storeGatewayClientMock{remoteAddr: "2.2.2.2"}, searchLabelValuesBatches: []*storepb.SearchResultBatch{{ @@ -667,7 +667,7 @@ func TestBlocksStoreQuerier_Compartments_SearchLabelValues(t *testing.T) { }, }}, queriedBlockIDs: []ulid.ULID{block1}, - }: {block1}, + }: {{block1}}, }, }}, } diff --git a/pkg/querier/blocks_store_queryable_search.go b/pkg/querier/blocks_store_queryable_search.go index 7c712ce7641..d247278c575 100644 --- a/pkg/querier/blocks_store_queryable_search.go +++ b/pkg/querier/blocks_store_queryable_search.go @@ -83,7 +83,7 @@ func (q *blocksStoreQuerier) SearchLabelNames( resSources []storage.SearchResultSet ) - queryF := func(ctx context.Context, clients map[BlocksStoreClient][]ulid.ULID, qMinT, qMaxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { + queryF := func(ctx context.Context, clients map[BlocksStoreClient][][]ulid.ULID, qMinT, qMaxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { sources, queriedBlocks, err := q.fetchSearchLabelNamesFromStore(ctx, clients, qMinT, qMaxT, tenantID, params, hints, convertedMatchers, indexMeta) if err != nil { return nil, err @@ -136,7 +136,7 @@ func (q *blocksStoreQuerier) SearchLabelValues( resSources []storage.SearchResultSet ) - queryF := func(ctx context.Context, clients map[BlocksStoreClient][]ulid.ULID, qMinT, qMaxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { + queryF := func(ctx context.Context, clients map[BlocksStoreClient][][]ulid.ULID, qMinT, qMaxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) { sources, queriedBlocks, err := q.fetchSearchLabelValuesFromStore(ctx, name, clients, qMinT, qMaxT, tenantID, params, hints, convertedMatchers, indexMeta) if err != nil { return nil, err @@ -169,7 +169,7 @@ func (q *blocksStoreQuerier) SearchLabelValues( // every source already collected is closed before returning. func (q *blocksStoreQuerier) fetchSearchLabelNamesFromStore( ctx context.Context, - clients map[BlocksStoreClient][]ulid.ULID, + clients map[BlocksStoreClient][][]ulid.ULID, minT int64, maxT int64, tenantID string, @@ -188,28 +188,30 @@ func (q *blocksStoreQuerier) fetchSearchLabelNamesFromStore( queriedBlocks []ulid.ULID ) - for c, blockIDs := range clients { - g.Go(func() error { - source, myBlocks, retriable, err := q.openSearchLabelNamesStream(reqCtx, gCtx, c, blockIDs, minT, maxT, params, hints, wireMatchers) - if err != nil { - if retriable { - level.Warn(spanLog).Log("msg", "failed to open search label names stream; error is retriable", "remote", c.RemoteAddress(), "err", err) - return nil + for c, partitions := range clients { + for _, blockIDs := range partitions { + g.Go(func() error { + source, myBlocks, retriable, err := q.openSearchLabelNamesStream(reqCtx, gCtx, c, blockIDs, minT, maxT, params, hints, wireMatchers) + if err != nil { + if retriable { + level.Warn(spanLog).Log("msg", "failed to open search label names stream; error is retriable", "remote", c.RemoteAddress(), "err", err) + return nil + } + return errors.Wrapf(err, "non-retriable error while fetching search label names from store %s", c.RemoteAddress()) } - return errors.Wrapf(err, "non-retriable error while fetching search label names from store %s", c.RemoteAddress()) - } - spanLog.DebugLog("msg", "received header from store-gateway", - "instance", c, - "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), - "queried blocks", strings.Join(convertULIDsToString(myBlocks), " ")) - - mtx.Lock() - sources = append(sources, source) - queriedBlocks = append(queriedBlocks, myBlocks...) - mtx.Unlock() - return nil - }) + spanLog.DebugLog("msg", "received header from store-gateway", + "instance", c, + "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), + "queried blocks", strings.Join(convertULIDsToString(myBlocks), " ")) + + mtx.Lock() + sources = append(sources, source) + queriedBlocks = append(queriedBlocks, myBlocks...) + mtx.Unlock() + return nil + }) + } } if err := g.Wait(); err != nil { @@ -223,7 +225,7 @@ func (q *blocksStoreQuerier) fetchSearchLabelNamesFromStore( func (q *blocksStoreQuerier) fetchSearchLabelValuesFromStore( ctx context.Context, name string, - clients map[BlocksStoreClient][]ulid.ULID, + clients map[BlocksStoreClient][][]ulid.ULID, minT int64, maxT int64, tenantID string, @@ -242,29 +244,31 @@ func (q *blocksStoreQuerier) fetchSearchLabelValuesFromStore( queriedBlocks []ulid.ULID ) - for c, blockIDs := range clients { - g.Go(func() error { - source, myBlocks, retriable, err := q.openSearchLabelValuesStream(reqCtx, gCtx, c, name, blockIDs, minT, maxT, params, hints, wireMatchers) - if err != nil { - if retriable { - level.Warn(spanLog).Log("msg", "failed to open search label values stream; error is retriable", "remote", c.RemoteAddress(), "err", err) - return nil + for c, partitions := range clients { + for _, blockIDs := range partitions { + g.Go(func() error { + source, myBlocks, retriable, err := q.openSearchLabelValuesStream(reqCtx, gCtx, c, name, blockIDs, minT, maxT, params, hints, wireMatchers) + if err != nil { + if retriable { + level.Warn(spanLog).Log("msg", "failed to open search label values stream; error is retriable", "remote", c.RemoteAddress(), "err", err) + return nil + } + return errors.Wrapf(err, "non-retriable error while fetching search label values from store %s", c.RemoteAddress()) } - return errors.Wrapf(err, "non-retriable error while fetching search label values from store %s", c.RemoteAddress()) - } - - spanLog.DebugLog("msg", "received header from store-gateway", - "instance", c, - "label", name, - "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), - "queried blocks", strings.Join(convertULIDsToString(myBlocks), " ")) - - mtx.Lock() - sources = append(sources, source) - queriedBlocks = append(queriedBlocks, myBlocks...) - mtx.Unlock() - return nil - }) + + spanLog.DebugLog("msg", "received header from store-gateway", + "instance", c, + "label", name, + "requested blocks", strings.Join(convertULIDsToString(blockIDs), " "), + "queried blocks", strings.Join(convertULIDsToString(myBlocks), " ")) + + mtx.Lock() + sources = append(sources, source) + queriedBlocks = append(queriedBlocks, myBlocks...) + mtx.Unlock() + return nil + }) + } } if err := g.Wait(); err != nil { diff --git a/pkg/querier/blocks_store_queryable_search_test.go b/pkg/querier/blocks_store_queryable_search_test.go index 5dc9dac6312..a39fcb9649c 100644 --- a/pkg/querier/blocks_store_queryable_search_test.go +++ b/pkg/querier/blocks_store_queryable_search_test.go @@ -261,9 +261,9 @@ func TestBlocksStoreQuerier_SearchLabelNames_HappyPath(t *testing.T) { } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ - store1: {block1}, - store2: {block2}, + map[BlocksStoreClient][][]ulid.ULID{ + store1: {{block1}}, + store2: {{block2}}, }, }} finder := &blocksFinderMock{} @@ -318,8 +318,8 @@ func TestBlocksStoreQuerier_SearchLabelNames_RetriesMissingBlock(t *testing.T) { } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storePartial: {block1, block2}}, - map[BlocksStoreClient][]ulid.ULID{storeRetry: {block2}}, + map[BlocksStoreClient][][]ulid.ULID{storePartial: {{block1, block2}}}, + map[BlocksStoreClient][][]ulid.ULID{storeRetry: {{block2}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -365,8 +365,8 @@ func TestBlocksStoreQuerier_SearchLabelNames_PartialReplicaFailureWithRetry(t *t } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeRetriable: {block1}}, - map[BlocksStoreClient][]ulid.ULID{storeOK: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{storeRetriable: {{block1}}}, + map[BlocksStoreClient][][]ulid.ULID{storeOK: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -407,7 +407,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_WarningsPropagated(t *testing.T) { queriedBlockIDs: []ulid.ULID{block1}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{store: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{store: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -451,7 +451,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_WarningsOnlyNoResults(t *testing.T) queriedBlockIDs: []ulid.ULID{block1}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{store: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{store: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -492,7 +492,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_NonRetriableErrorBubblesUp(t *testi searchLabelNamesErr: status.Error(codes.Code(http.StatusUnprocessableEntity), "validation"), } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{store: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{store: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -537,7 +537,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_PeerNonRetriableCancelsBlockedHeade queriedBlockIDs: []ulid.ULID{block2}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeA: {block1}, storeB: {block2}}, + map[BlocksStoreClient][][]ulid.ULID{storeA: {{block1}}, storeB: {{block2}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -590,7 +590,7 @@ func TestBlocksStoreQuerier_SearchLabelValues_PeerNonRetriableCancelsBlockedHead queriedBlockIDs: []ulid.ULID{block2}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeA: {block1}, storeB: {block2}}, + map[BlocksStoreClient][][]ulid.ULID{storeA: {{block1}}, storeB: {{block2}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -651,7 +651,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_MidStreamRecvError(t *testing.T) { queriedBlockIDs: []ulid.ULID{block1}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{store: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{store: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -696,8 +696,8 @@ func TestBlocksStoreQuerier_SearchLabelNames_HeaderRecvErrorIsRetriable(t *testi queriedBlockIDs: []ulid.ULID{block1}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeHeaderErr: {block1}}, - map[BlocksStoreClient][]ulid.ULID{storeOK: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{storeHeaderErr: {{block1}}}, + map[BlocksStoreClient][][]ulid.ULID{storeOK: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -735,7 +735,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_MissingHeaderIsProtocolViolation(t searchLabelNamesOmitHeader: true, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeNoHeader: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{storeNoHeader: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -776,7 +776,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_ResultsOnHeaderIsProtocolViolation( searchLabelNamesOmitHeader: true, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeResultsOnHeader: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{storeResultsOnHeader: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -829,9 +829,9 @@ func TestBlocksStoreQuerier_SearchLabelValues_HappyPath(t *testing.T) { } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ - store1: {block1}, - store2: {block2}, + map[BlocksStoreClient][][]ulid.ULID{ + store1: {{block1}}, + store2: {{block2}}, }, }} finder := &blocksFinderMock{} @@ -878,7 +878,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_PropagatesBucketStoreMetadata(t *te queriedBlockIDs: []ulid.ULID{block1}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{store: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{store: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -919,7 +919,7 @@ func TestBlocksStoreQuerier_SearchLabelValues_PropagatesBucketStoreMetadata(t *t queriedBlockIDs: []ulid.ULID{block1}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{store: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{store: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -962,7 +962,7 @@ func TestBlocksStoreQuerier_SearchLabelValues_PassesLabelName(t *testing.T) { queriedBlockIDs: []ulid.ULID{block1}, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{store: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{store: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ @@ -1051,7 +1051,7 @@ func TestBlocksStoreQuerier_SearchLabelNames_MalformedHeaderBlockHintHardFails(t searchLabelNamesOmitHeader: true, } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeBadHeader: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{storeBadHeader: {{block1}}}, }} finder := &blocksFinderMock{} finder.On("GetBlocks", mock.Anything, "user-1", minT, maxT).Return(bucketindex.Blocks{ diff --git a/pkg/querier/blocks_store_queryable_test.go b/pkg/querier/blocks_store_queryable_test.go index 0cf929e8153..d3302fc6900 100644 --- a/pkg/querier/blocks_store_queryable_test.go +++ b/pkg/querier/blocks_store_queryable_test.go @@ -135,14 +135,14 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT, 1). addValue(metricNameLabel, minT+1, 2). addBlocks(block1, block2). addFetchedIndexBytes(50). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -216,7 +216,7 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addChunks(metricNameLabel, createAggrChunkWithSamples(promql.FPoint{T: minT, F: 1}), @@ -225,7 +225,7 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { addBlocks(block1, block2). addFetchedIndexBytes(50). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -299,14 +299,14 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addValue(series1Label, minT+1, 2). addValue(series2Label, minT, 3). addBlocks(block1, block2). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -385,17 +385,17 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT, 1). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT+1, 2). addBlocks(block2). build(), - }: {block2}, + }: {{block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -469,18 +469,18 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT+1, 2). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT, 1). addValue(metricNameLabel, minT+1, 2). addBlocks(block2). build(), - }: {block2}, + }: {{block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -554,25 +554,25 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT+1, 2). addValue(series2Label, minT, 1). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addValue(series1Label, minT+1, 2). addBlocks(block2). build(), - }: {block2}, + }: {{block2}}, &storeGatewayClientMock{remoteAddr: "3.3.3.3", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series2Label, minT, 1). addValue(series2Label, minT+1, 3). addBlocks(block3). build(), - }: {block3}, + }: {{block3}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -649,15 +649,15 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedSeriesResponses: nil, - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder().addBlocks(block1).build(), - }: {block1}, + }: {{block1}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -669,15 +669,15 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedSeriesResponses: nil, - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedSeriesResponses: nil, - }: {block1}, + }: {{block1}}, }, // Third attempt returns an error because there are no other store-gateways left. errors.New("no store-gateway remaining after exclude"), @@ -693,13 +693,13 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addValue(series1Label, minT+1, 2). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, }, // Second attempt returns an error because there are no other store-gateways left. errors.New("no store-gateway remaining after exclude"), @@ -770,17 +770,17 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT+1, 2). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT+1, 2). addBlocks(block2). build(), - }: {block2}, + }: {{block2}}, }, // Second attempt returns an error because there are no other store-gateways left. errors.New("no store-gateway remaining after exclude"), @@ -851,33 +851,33 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addBlocks(block1). build(), - }: {block1, block3}, + }: {{block1, block3}}, &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series2Label, minT, 2). addBlocks(block2). build(), - }: {block2, block4}, + }: {{block2, block4}}, }, // Second attempt returns 1 missing block. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "3.3.3.3", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT+1, 2). addBlocks(block3). build(), - }: {block3, block4}, + }: {{block3, block4}}, }, // Third attempt returns the last missing block. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "4.4.4.4", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series2Label, minT+1, 3). addBlocks(block4). build(), - }: {block4}, + }: {{block4}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -953,38 +953,38 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block1}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesErr: errors.New("failed to receive from store-gateway"), - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesErr: errors.New("failed to receive from store-gateway"), - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "3.3.3.3", mockedSeriesErr: errors.New("failed to receive from store-gateway"), - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "4.4.4.4", mockedSeriesErr: errors.New("failed to receive from store-gateway"), - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "5.5.5.5", mockedSeriesErr: errors.New("failed to receive from store-gateway"), - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "6.6.6.6", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 2). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1054,13 +1054,13 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addValue(series1Label, minT+1, 2). addBlocks(block1, block2). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{maxChunksPerQuery: 3}, @@ -1134,13 +1134,13 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addValue(series1Label, minT+1, 2). addBlocks(block1, block2). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1206,13 +1206,13 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addValue(series1Label, minT+1, 2). addBlocks(block1, block2). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1281,33 +1281,33 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addBlocks(block1). build(), - }: {block1, block3}, + }: {{block1, block3}}, &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series2Label, minT, 2). addBlocks(block2). build(), - }: {block2, block4}, + }: {{block2, block4}}, }, // Second attempt returns 1 missing block. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "3.3.3.3", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT+1, 2). addBlocks(block3). build(), - }: {block3, block4}, + }: {{block3, block4}}, }, // Third attempt returns the last missing block. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "4.4.4.4", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series2Label, minT+1, 3). addBlocks(block4). build(), - }: {block4}, + }: {{block4}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1320,13 +1320,13 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addValue(series2Label, minT+1, 2). addBlocks(block1, block2). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1339,13 +1339,13 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 1). addValue(series1Label, minT+1, 2). addBlocks(block1, block2). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{maxChunksPerQuery: 1}, @@ -1361,13 +1361,13 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { }, queryShardID: "2_of_4", storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT, 1). addValue(metricNameLabel, minT+1, 2). addBlocks(block2). build(), - }: {block2}, // Only block2 will be queried + }: {{block2}}, // Only block2 will be queried }, }, limits: &blocksStoreLimitsMock{}, @@ -1443,13 +1443,13 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { }, queryShardID: "3_of_5", storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(metricNameLabel, minT, 1). addValue(metricNameLabel, minT+1, 2). addBlocks(block1, block2, block3, block4). build(), - }: {block1, block2, block3, block4}, + }: {{block1, block2, block3, block4}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1521,18 +1521,18 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block1}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedSeriesErr: errors.New("failed to receive from store-gateway"), - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(series1Label, minT, 2). addBlocks(block1). build(), - }: {block1}, + }: {{block1}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1604,7 +1604,7 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addFloatHistogramSamples(metricNameLabel, promql.HPoint{T: minT + 1, H: test.GenerateTestFloatHistogram(40)}, @@ -1620,7 +1620,7 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { addBlocks(block1, block2). addFetchedIndexBytes(50). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1646,7 +1646,7 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addFloatHistogramSamples(metricNameLabel, promql.HPoint{T: minT + 1, H: test.GenerateTestFloatHistogram(40)}, @@ -1666,7 +1666,7 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { addBlocks(block1, block2). addFetchedIndexBytes(50). build(), - }: {block1, block2}, + }: {{block1, block2}}, }, }, limits: &blocksStoreLimitsMock{}, @@ -1691,7 +1691,7 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { // Count the number of series and chunks to check the stats later. seriesCount, chunksCount := 0, 0 for _, res := range testData.storeSetResponses { - m, ok := res.(map[BlocksStoreClient][]ulid.ULID) + m, ok := res.(map[BlocksStoreClient][][]ulid.ULID) if !ok { // If this isn't a success response we can't count series or chunks. continue @@ -1824,14 +1824,14 @@ func TestBlocksStoreQuerier_Select_ClosedBeforeSelectFinishes(t *testing.T) { block := ulid.MustNew(1, nil) storeSetResponses := []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). addValue(labels.FromStrings(model.MetricNameLabel, "some_metric"), minT, 1). addBlocks(block). build(), - }: {block}, + }: {{block}}, }, } @@ -1923,9 +1923,9 @@ func TestBlocksStoreQuerier_ShouldReturnContextCanceledIfContextWasCanceledWhile stores := &blocksStoreSetMock{mockedResponses: []interface{}{ // These tests only require 1 mocked response, but we mock it multiple times to make debugging easier // when the tests fail because the request is retried (even if we expect not to be retried). - map[BlocksStoreClient][]ulid.ULID{client: {block1}}, - map[BlocksStoreClient][]ulid.ULID{client: {block1}}, - map[BlocksStoreClient][]ulid.ULID{client: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{client: {{block1}}}, + map[BlocksStoreClient][][]ulid.ULID{client: {{block1}}}, + map[BlocksStoreClient][][]ulid.ULID{client: {{block1}}}, }} reg := prometheus.NewPedanticRegistry() @@ -2130,7 +2130,7 @@ func TestBlocksStoreQuerier_Select_cancelledContext(t *testing.T) { } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeGateway: {block}}, + map[BlocksStoreClient][][]ulid.ULID{storeGateway: {{block}}}, errors.New("no store-gateway remaining after exclude"), }} @@ -2219,7 +2219,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2234,7 +2234,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block1, block2), ResponseHints: mockValuesResponseHints(block1, block2), }, - }: {block1, block2}, + }: {{block1, block2}}, }, }, expectedLabelNames: namesFromSeries(series1, series2), @@ -2246,7 +2246,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2259,7 +2259,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Warnings: []string{}, ResponseHints: mockValuesResponseHints(block1, block2), }, - }: {block1, block2}, + }: {{block1, block2}}, }, }, expectedLabelNames: namesFromSeries(series1, series2), @@ -2271,7 +2271,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2286,7 +2286,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block1), ResponseHints: mockValuesResponseHints(block1), }, - }: {block1}, + }: {{block1}}, &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2301,7 +2301,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block2), ResponseHints: mockValuesResponseHints(block2), }, - }: {block2}, + }: {{block2}}, }, }, expectedLabelNames: namesFromSeries(series1, series2), @@ -2313,7 +2313,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2328,7 +2328,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block1), ResponseHints: mockValuesResponseHints(block1), }, - }: {block1}, + }: {{block1}}, &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2343,7 +2343,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block2), ResponseHints: mockValuesResponseHints(block2), }, - }: {block2}, + }: {{block2}}, }, }, expectedLabelNames: namesFromSeries(series1), @@ -2358,7 +2358,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { // Block2 has only series1 // Block3 has only series2 storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2373,7 +2373,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block1), ResponseHints: mockValuesResponseHints(block1), }, - }: {block1}, + }: {{block1}}, &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2388,7 +2388,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block2), ResponseHints: mockValuesResponseHints(block2), }, - }: {block2}, + }: {{block2}}, &storeGatewayClientMock{ remoteAddr: "3.3.3.3", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2403,7 +2403,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block3), ResponseHints: mockValuesResponseHints(block3), }, - }: {block3}, + }: {{block3}}, }, }, expectedLabelNames: namesFromSeries(series1, series2), @@ -2445,7 +2445,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2460,7 +2460,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block1), ResponseHints: mockValuesResponseHints(block1), }, - }: {block1}, + }: {{block1}}, }, // Second attempt returns an error because there are no other store-gateways left. errors.New("no store-gateway remaining after exclude"), @@ -2476,7 +2476,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2491,7 +2491,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block1), ResponseHints: mockValuesResponseHints(block1), }, - }: {block1}, + }: {{block1}}, &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2506,7 +2506,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block2), ResponseHints: mockValuesResponseHints(block2), }, - }: {block2}, + }: {{block2}}, }, // Second attempt returns an error because there are no other store-gateways left. errors.New("no store-gateway remaining after exclude"), @@ -2526,7 +2526,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { }, storeSetResponses: []interface{}{ // First attempt returns a client whose response does not include all expected blocks. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2541,7 +2541,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block1), ResponseHints: mockValuesResponseHints(block1), }, - }: {block1, block3}, + }: {{block1, block3}}, &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2556,10 +2556,10 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block2), ResponseHints: mockValuesResponseHints(block2), }, - }: {block2, block4}, + }: {{block2, block4}}, }, // Second attempt returns 1 missing block. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "3.3.3.3", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2574,10 +2574,10 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block3), ResponseHints: mockValuesResponseHints(block3), }, - }: {block3, block4}, + }: {{block3, block4}}, }, // Third attempt returns the last missing block. - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "4.4.4.4", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2592,7 +2592,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block4), ResponseHints: mockValuesResponseHints(block4), }, - }: {block4}, + }: {{block4}}, }, }, expectedLabelNames: namesFromSeries(series1, series2), @@ -2635,14 +2635,14 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { {ID: block1}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesErr: errors.New("failed to receive from store-gateway"), mockedLabelValuesErr: errors.New("failed to receive from store-gateway"), - }: {block1}, + }: {{block1}}, }, - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "2.2.2.2", mockedLabelNamesResponse: &storepb.LabelNamesResponse{ @@ -2657,7 +2657,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { Hints: mockValuesHints(block1), ResponseHints: mockValuesResponseHints(block1), }, - }: {block1}, + }: {{block1}}, }, }, expectedLabelNames: namesFromSeries(series1), @@ -2698,12 +2698,12 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { {ID: block2}, }, storeSetResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ + map[BlocksStoreClient][][]ulid.ULID{ &storeGatewayClientMock{ remoteAddr: "1.1.1.1", mockedLabelNamesErr: status.Error(http.StatusUnprocessableEntity, "limit exceeded"), mockedLabelValuesErr: status.Error(http.StatusUnprocessableEntity, "limit exceeded"), - }: {block1, block2}, + }: {{block1, block2}}, }, }, expectedErrRegex: "non-retriable error while fetching label (names|values) from store: rpc error: code = Code\\(422\\) desc = limit exceeded", @@ -2783,7 +2783,7 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { } stores := &blocksStoreSetMock{mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{storeGateway: {block1}}, + map[BlocksStoreClient][][]ulid.ULID{storeGateway: {{block1}}}, errors.New("no store-gateway remaining after exclude"), }} @@ -3119,9 +3119,9 @@ func TestBlocksStoreQuerier_PromQLExecution(t *testing.T) { stores := &blocksStoreSetMock{ Service: services.NewIdleService(nil, nil), mockedResponses: []interface{}{ - map[BlocksStoreClient][]ulid.ULID{ - gateway1: {block1}, - gateway2: {block2}, + map[BlocksStoreClient][][]ulid.ULID{ + gateway1: {{block1}}, + gateway2: {{block2}}, }, }, } @@ -3264,7 +3264,7 @@ type blocksStoreSetMock struct { nextResult int } -func (m *blocksStoreSetMock) GetClientsFor(_ string, _ bucketindex.Blocks, _ map[ulid.ULID][]string) (map[BlocksStoreClient][]ulid.ULID, error) { +func (m *blocksStoreSetMock) GetClientsFor(_ string, _ bucketindex.Blocks, _ map[ulid.ULID][]string) (map[BlocksStoreClient][][]ulid.ULID, error) { if m.nextResult >= len(m.mockedResponses) { panic("not enough mocked results") } @@ -3275,7 +3275,7 @@ func (m *blocksStoreSetMock) GetClientsFor(_ string, _ bucketindex.Blocks, _ map if err, ok := res.(error); ok { return nil, err } - if clients, ok := res.(map[BlocksStoreClient][]ulid.ULID); ok { + if clients, ok := res.(map[BlocksStoreClient][][]ulid.ULID); ok { return clients, nil } @@ -3437,6 +3437,7 @@ func (m *cancelerStoreGatewayClientMock) RemoteZone() string { type blocksStoreLimitsMock struct { maxLabelsQueryLength time.Duration maxChunksPerQuery int + maxBlocksPerStoreRequest int storeGatewayTenantShardSize int storeGatewayTenantShardSizePerZone int storeGatewayExpandedReplication bool @@ -3450,6 +3451,10 @@ func (m *blocksStoreLimitsMock) MaxChunksPerQuery(_ string) int { return m.maxChunksPerQuery } +func (m *blocksStoreLimitsMock) MaxBlocksPerStoreRequest(_ string) int { + return m.maxBlocksPerStoreRequest +} + func (m *blocksStoreLimitsMock) StoreGatewayTenantShardSize(_ string) int { return m.storeGatewayTenantShardSize } diff --git a/pkg/querier/blocks_store_replicated_set.go b/pkg/querier/blocks_store_replicated_set.go index b174c2541fe..3f969f38f9c 100644 --- a/pkg/querier/blocks_store_replicated_set.go +++ b/pkg/querier/blocks_store_replicated_set.go @@ -104,7 +104,7 @@ func (s *blocksStoreReplicationSet) stopping(_ error) error { return services.StopManagerAndAwaitStopped(context.Background(), s.subservices) } -func (s *blocksStoreReplicationSet) GetClientsFor(userID string, blocks bucketindex.Blocks, exclude map[ulid.ULID][]string) (map[BlocksStoreClient][]ulid.ULID, error) { +func (s *blocksStoreReplicationSet) GetClientsFor(userID string, blocks bucketindex.Blocks, exclude map[ulid.ULID][]string) (map[BlocksStoreClient][][]ulid.ULID, error) { blocksByAddr := make(map[string][]ulid.ULID) instances := make(map[string]ring.InstanceDesc) userRing := storegateway.GetShuffleShardingSubring(s.storesRing, userID, s.limits) @@ -137,7 +137,8 @@ func (s *blocksStoreReplicationSet) GetClientsFor(userID string, blocks bucketin blocksByAddr[inst.Addr] = append(blocksByAddr[inst.Addr], block.ID) } - clients := map[BlocksStoreClient][]ulid.ULID{} + clients := map[BlocksStoreClient][][]ulid.ULID{} + maxBlocksPerClient := s.limits.MaxBlocksPerStoreRequest(userID) // Get the client for each store-gateway. for addr, instance := range instances { @@ -146,12 +147,21 @@ func (s *blocksStoreReplicationSet) GetClientsFor(userID string, blocks bucketin return nil, errors.Wrapf(err, "failed to get store-gateway client for %s %s", instance.Id, addr) } - clients[c.(BlocksStoreClient)] = blocksByAddr[addr] + clients[c.(BlocksStoreClient)] = partitionBlocks(blocksByAddr[addr], maxBlocksPerClient) } return clients, nil } +// partitionBlocks splits blocks into consecutive partitions of at most max blocks each. +// When max is 0 the function returns all blocks in a single partition. +func partitionBlocks(blocks []ulid.ULID, max int) [][]ulid.ULID { + if max == 0 { + return [][]ulid.ULID{blocks} + } + return slices.Collect(slices.Chunk(blocks, max)) +} + func getNonExcludedInstance(set ring.ReplicationSet, exclude []string, balancingStrategy loadBalancingStrategy, preferredZones []string) *ring.InstanceDesc { if balancingStrategy == randomLoadBalancing { // Randomize the list of instances to not always query the same one. diff --git a/pkg/querier/blocks_store_replicated_set_test.go b/pkg/querier/blocks_store_replicated_set_test.go index b616a8e8b3a..40d936d7911 100644 --- a/pkg/querier/blocks_store_replicated_set_test.go +++ b/pkg/querier/blocks_store_replicated_set_test.go @@ -69,7 +69,7 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { setup func(*ring.Desc) queryBlocks bucketindex.Blocks exclude map[ulid.ULID][]string - expectedClients map[string][]ulid.ULID + expectedClients map[string][][]ulid.ULID expectedErr error }{ "shard size 0, single instance in the ring with RF = 1": { @@ -79,8 +79,8 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-1", "127.0.0.1", "", []uint32{block1Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block2}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1, blockID2}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID2}}, }, }, "shard size 0, single instance in the ring with RF = 1 but excluded": { @@ -105,8 +105,8 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { exclude: map[ulid.ULID][]string{ blockID3: {"127.0.0.1"}, }, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1, blockID2}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID2}}, }, }, "shard size 0, single instance in the ring with RF = 2": { @@ -116,8 +116,8 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-1", "127.0.0.1", "", []uint32{block1Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block2}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1, blockID2}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID2}}, }, }, "shard size 0, multiple instances in the ring with each requested block belonging to a different store-gateway and RF = 1": { @@ -130,10 +130,10 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-4", "127.0.0.4", "", []uint32{block4Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block3, block4}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1}, - "127.0.0.3": {blockID3}, - "127.0.0.4": {blockID4}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1}}, + "127.0.0.3": {{blockID3}}, + "127.0.0.4": {{blockID4}}, }, }, "shard size 0, multiple instances in the ring with each requested block belonging to a different store-gateway and RF = 1 but excluded": { @@ -161,10 +161,10 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-4", "127.0.0.4", "", []uint32{block4Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block3, block4}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1}, - "127.0.0.3": {blockID3}, - "127.0.0.4": {blockID4}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1}}, + "127.0.0.3": {{blockID3}}, + "127.0.0.4": {{blockID4}}, }, }, "shard size 0, multiple instances in the ring with multiple requested blocks belonging to the same store-gateway and RF = 2": { @@ -175,9 +175,9 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-2", "127.0.0.2", "", []uint32{block3Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block2, block3, block4}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1, blockID4}, - "127.0.0.2": {blockID2, blockID3}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID4}}, + "127.0.0.2": {{blockID2, blockID3}}, }, }, "shard size 0, multiple instances in the ring with each requested block belonging to a different store-gateway and RF = 2 and some blocks excluded but with replacement available": { @@ -194,9 +194,9 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { blockID3: {"127.0.0.3"}, blockID1: {"127.0.0.1"}, }, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.2": {blockID1}, - "127.0.0.4": {blockID3, blockID4}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.2": {{blockID1}}, + "127.0.0.4": {{blockID3, blockID4}}, }, }, "shard size 0, multiple instances in the ring are JOINING, the requested block + its replicas only belongs to JOINING instances": { @@ -209,8 +209,8 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-4", "127.0.0.4", "", []uint32{block4Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.4": {blockID1}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.4": {{blockID1}}, }, }, "shard size 1, single instance in the ring with RF = 1": { @@ -220,8 +220,8 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-1", "127.0.0.1", "", []uint32{block1Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block2}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1, blockID2}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID2}}, }, }, "shard size 1, single instance in the ring with RF = 1, but store-gateway excluded": { @@ -243,8 +243,8 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-1", "127.0.0.1", "", []uint32{block1Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block2}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1, blockID2}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID2}}, }, }, "shard size 1, multiple instances in the ring with RF = 1": { @@ -257,8 +257,8 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-4", "127.0.0.4", "", []uint32{block4Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block2, block4}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1, blockID2, blockID4}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID2, blockID4}}, }, }, "shard size 2, shuffle sharding, multiple instances in the ring with RF = 1": { @@ -271,9 +271,9 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-4", "127.0.0.4", "", []uint32{block4Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block2, block4}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1, blockID4}, - "127.0.0.3": {blockID2}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID4}}, + "127.0.0.3": {{blockID2}}, }, }, "shard size 4, multiple instances in the ring with RF = 1": { @@ -286,10 +286,10 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { d.AddIngester("instance-4", "127.0.0.4", "", []uint32{block4Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) }, queryBlocks: []*bucketindex.Block{block1, block2, block4}, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.1": {blockID1}, - "127.0.0.2": {blockID2}, - "127.0.0.4": {blockID4}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1}}, + "127.0.0.2": {{blockID2}}, + "127.0.0.4": {{blockID4}}, }, }, "shard size 2, multiple instances in the ring with RF = 2, with excluded blocks but some replacement available": { @@ -306,8 +306,8 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { blockID1: {"127.0.0.1"}, blockID2: {"127.0.0.1"}, }, - expectedClients: map[string][]ulid.ULID{ - "127.0.0.3": {blockID1, blockID2}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.3": {{blockID1, blockID2}}, }, }, "shard size 2, multiple instances in the ring with RF = 2, SS = 2 with excluded blocks and no replacement available": { @@ -709,11 +709,11 @@ func TestBlocksStoreReplicationSet_GetClientsFor_BufferReuseSafety(t *testing.T) // Verify the block assignment is correct. clientAddrs := getStoreGatewayClientAddrs(clients) - expectedClients := map[string][]ulid.ULID{ - "127.0.0.1": {blockID1}, - "127.0.0.2": {blockID2}, - "127.0.0.3": {blockID3}, - "127.0.0.4": {blockID4}, + expectedClients := map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1}}, + "127.0.0.2": {{blockID2}}, + "127.0.0.3": {{blockID3}}, + "127.0.0.4": {{blockID4}}, } assert.Equal(t, expectedClients, clientAddrs) } @@ -839,10 +839,52 @@ func BenchmarkBlocksStoreReplicationSet_GetClientsFor(b *testing.B) { } } -func getStoreGatewayClientAddrs(clients map[BlocksStoreClient][]ulid.ULID) map[string][]ulid.ULID { - addrs := map[string][]ulid.ULID{} - for c, blockIDs := range clients { - addrs[c.RemoteAddress()] = blockIDs +func TestPartitionBlocks(t *testing.T) { + id1 := ulid.MustNew(1, nil) + id2 := ulid.MustNew(2, nil) + id3 := ulid.MustNew(3, nil) + id4 := ulid.MustNew(4, nil) + id5 := ulid.MustNew(5, nil) + + tests := map[string]struct { + blocks []ulid.ULID + max int + expected [][]ulid.ULID + }{ + "disabled keeps input order in a single partition": { + blocks: []ulid.ULID{id3, id1, id2}, + max: 0, + expected: [][]ulid.ULID{{id3, id1, id2}}, + }, + "exactly max stays in one partition": { + blocks: []ulid.ULID{id1, id2, id3}, + max: 3, + expected: [][]ulid.ULID{{id1, id2, id3}}, + }, + "one over max splits into two": { + blocks: []ulid.ULID{id1, id2, id3, id4}, + max: 3, + expected: [][]ulid.ULID{{id1, id2, id3}, {id4}}, + }, + "splits into multiple partitions of max size": { + blocks: []ulid.ULID{id1, id2, id3, id4, id5}, + max: 2, + expected: [][]ulid.ULID{{id1, id2}, {id3, id4}, {id5}}, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + require.Equal(t, tc.expected, partitionBlocks(tc.blocks, tc.max)) + }) + } +} + +// getStoreGatewayClientAddrs returns each client's partitions keyed by store-gateway address. +func getStoreGatewayClientAddrs(clients map[BlocksStoreClient][][]ulid.ULID) map[string][][]ulid.ULID { + addrs := map[string][][]ulid.ULID{} + for c, partitions := range clients { + addrs[c.RemoteAddress()] = partitions } return addrs } diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 5b0ef85858b..2dd8401d890 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -49,6 +49,7 @@ const ( MaxSeriesPerQueryFlag = "querier.max-fetched-series-per-query" MaxEstimatedChunksPerQueryMultiplierFlag = "querier.max-estimated-fetched-chunks-per-query-multiplier" MaxEstimatedMemoryConsumptionPerQueryFlag = "querier.max-estimated-memory-consumption-per-query" + MaxBlocksPerStoreRequestFlag = "querier.max-blocks-per-store-request" MaxLabelNamesPerSeriesFlag = "validation.max-label-names-per-series" MaxLabelNamesPerInfoSeriesFlag = "validation.max-label-names-per-info-series" MaxLabelNameLengthFlag = "validation.max-length-label-name" @@ -215,6 +216,7 @@ type Limits struct { MaxFetchedSeriesPerQuery int `yaml:"max_fetched_series_per_query" json:"max_fetched_series_per_query"` MaxFetchedChunkBytesPerQuery int `yaml:"max_fetched_chunk_bytes_per_query" json:"max_fetched_chunk_bytes_per_query"` MaxEstimatedMemoryConsumptionPerQuery uint64 `yaml:"max_estimated_memory_consumption_per_query" json:"max_estimated_memory_consumption_per_query" category:"experimental"` + MaxBlocksPerStoreRequest int `yaml:"max_blocks_per_store_request" json:"max_blocks_per_store_request" category:"experimental"` MaxQueryLookback model.Duration `yaml:"max_query_lookback" json:"max_query_lookback"` MaxPartialQueryLength model.Duration `yaml:"max_partial_query_length" json:"max_partial_query_length"` MaxQueryParallelism int `yaml:"max_query_parallelism" json:"max_query_parallelism"` @@ -431,6 +433,7 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) { f.IntVar(&l.MaxFetchedSeriesPerQuery, MaxSeriesPerQueryFlag, 0, "The maximum number of unique series for which a query can fetch samples from ingesters and store-gateways. This limit is enforced in the querier, ruler and store-gateway. 0 to disable") f.IntVar(&l.MaxFetchedChunkBytesPerQuery, MaxChunkBytesPerQueryFlag, 0, "The maximum size of all chunks in bytes that a query can fetch from ingesters and store-gateways. This limit is enforced in the querier and ruler. 0 to disable.") f.Uint64Var(&l.MaxEstimatedMemoryConsumptionPerQuery, MaxEstimatedMemoryConsumptionPerQueryFlag, 0, "The maximum estimated memory a single query can consume at once, in bytes. This limit is only enforced when Mimir's query engine is in use. This limit is enforced in the querier. 0 to disable.") + f.IntVar(&l.MaxBlocksPerStoreRequest, MaxBlocksPerStoreRequestFlag, 0, "Maximum number of blocks that a querier will reference in a single request to a store-gateway. When a request would exceed this, it is split into multiple requests to the same store-gateway. 0 disables the limit.") f.Var(&l.MaxPartialQueryLength, MaxPartialQueryLengthFlag, "Limit the time range for partial queries at the querier level.") f.Var(&l.MaxQueryLookback, "querier.max-query-lookback", "Limit how long back data (series and metadata) can be queried, up until duration ago. This limit is enforced in the query-frontend, querier and ruler for instant, range and remote read queries. For metadata queries like series, label names, label values queries the limit is enforced in the querier and ruler. If the requested time range is outside the allowed range, the request will not fail but will be manipulated to only query data within the allowed time range. 0 to disable.") f.IntVar(&l.MaxQueryParallelism, "querier.max-query-parallelism", 14, "Maximum number of split (by time) or partial (by shard) queries that will be scheduled in parallel by the query-frontend for a single input query. This limit is introduced to have a fairer query scheduling and avoid a single query over a large time range saturating all available queriers.") @@ -1044,6 +1047,10 @@ func (o *Overrides) MaxChunksPerQuery(userID string) int { return o.getOverridesForUser(userID).MaxChunksPerQuery } +func (o *Overrides) MaxBlocksPerStoreRequest(userID string) int { + return o.getOverridesForUser(userID).MaxBlocksPerStoreRequest +} + func (o *Overrides) MaxEstimatedChunksPerQuery(userID string) int { overridesForUser := o.getOverridesForUser(userID) return int(overridesForUser.MaxEstimatedChunksPerQueryMultiplier * float64(overridesForUser.MaxChunksPerQuery)) From e79f96d245248e8fa0525a4d233a921b398b0025 Mon Sep 17 00:00:00 2001 From: Vladimir Varankin Date: Sat, 11 Jul 2026 12:12:53 +0200 Subject: [PATCH 2/7] update changelog --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index db4b023d5b1..d59c6f06992 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ * [ENHANCEMENT] Distributor: Add the experimental `cortex_distributor_otlp_requests_with_job_or_instance_resource_attribute_total{user}` counter to track OTLP requests carrying `job` or `instance` as a resource attribute. #16285 * [ENHANCEMENT] Ruler: Split oversized remote distributor writes to keep resulting calls within the configured gRPC maximum send size, and expose the number of generated requests in `cortex_ruler_remote_distributor_requests_per_write_request`. #16160 * [ENHANCEMENT] Alerts: Don't fire `MimirMemberlistZoneAwareRoutingAutoFailover` while the node is still joining the cluster. #16315 +* [FEATURE] Querier: Add experimental per-tenant limit `-querier.max-blocks-per-store-request` to cap the number of blocks a single store-gateway request may reference. Disabled by default. #16292 * [BUGFIX] Query-frontend: Wait for the querier ring to be populated during startup, up to 30 seconds, before reporting the query-frontend as ready. Previously a query-frontend could become ready before it had seen any querier in the ring and fail every query it received until the ring was populated. Only applies when remote execution is enabled, and can be disabled with the experimental `-query-frontend.wait-for-querier-ring-on-startup=false`. #16333 * [BUGFIX] Query-frontend: Fail queries with a clear error, rather than planning them against an invalid maximum supported query plan version, when the querier ring contains only unhealthy queriers. #16333 * [BUGFIX] Query-frontend: Return a HTTP 500 error rather than a HTTP 400 when a querier receives a query plan that is too new. #16233 From ca1427a5b6607afa31a79ce63c3bee1b3502af8a Mon Sep 17 00:00:00 2001 From: Vladimir Varankin Date: Thu, 6 Aug 2026 12:47:14 +0200 Subject: [PATCH 3/7] add metrics --- pkg/querier/blocks_store_queryable.go | 32 ++++++++++++++++++--------- 1 file changed, 21 insertions(+), 11 deletions(-) diff --git a/pkg/querier/blocks_store_queryable.go b/pkg/querier/blocks_store_queryable.go index e7329b6b38f..7b9f887d95f 100644 --- a/pkg/querier/blocks_store_queryable.go +++ b/pkg/querier/blocks_store_queryable.go @@ -110,6 +110,7 @@ type blocksStoreQueryableMetrics struct { storesHit prometheus.Histogram refetches prometheus.Histogram + storesQueried prometheus.Counter blocksFound prometheus.Counter blocksQueried prometheus.Counter blocksWithCompactorShardButIncompatibleQueryShard prometheus.Counter @@ -133,7 +134,10 @@ func newBlocksStoreQueryableMetrics(compartmentsCfg compartments.Config, reg pro Help: "Number of re-fetches attempted while querying store-gateway instances due to missing blocks.", Buckets: []float64{0, 1, 2}, }), - + storesQueried: promauto.With(reg).NewCounter(prometheus.CounterOpts{ + Name: "cortex_querier_storegateway_queried_total", + Help: "Total number of requests to store-gateway instances for a single query. Splitting queried blocks into partitions can produce more than one request per store-gateway instance per query.", + }), blocksFound: promauto.With(reg).NewCounter(prometheus.CounterOpts{ Name: "cortex_querier_blocks_found_total", Help: "Number of blocks found based on query time range.", @@ -703,12 +707,13 @@ func (q *blocksStoreQuerier) startBuffering(streamReaders []*storeGatewayStreamR type queryFunc func(ctx context.Context, clients map[BlocksStoreClient][][]ulid.ULID, minT, maxT int64, indexMeta *bucketindex.Metadata) ([]ulid.ULID, error) -// storeGatewayQueryStats holds the per-query store-gateway histogram values. queryWithConsistencyCheck -// returns it (rather than observing the histograms itself) so the caller observes them once per query. +// storeGatewayQueryStats holds the per-query store-gateway metric values. queryWithConsistencyCheck +// returns it (rather than updating the metrics itself) so the caller aggregates them once per query. type storeGatewayQueryStats struct { - storesHit int // number of distinct store-gateway instances queried - refetches int // number of retries due to missing blocks - queried bool // whether the block store was actually queried (true only on the success path) + storesHit int // number of distinct store-gateway instances queried + storesQueried int // number of requests to store-gateway instances (one per block partition, so >= storesHit) + refetches int // number of retries due to missing blocks + queried bool // whether the block store was actually queried (true only on the success path) } func (q *blocksStoreQuerier) queryWithConsistencyCheck( @@ -756,6 +761,7 @@ func (q *blocksStoreQuerier) queryWithConsistencyCheck( remainingBlocks = knownBlocks attemptedBlocks = map[ulid.ULID][]string{} touchedStores = map[string]struct{}{} + queriedStores = 0 ) consistencyTracker := q.consistency.NewTracker(knownBlocks, spanLog) @@ -797,6 +803,7 @@ func (q *blocksStoreQuerier) queryWithConsistencyCheck( // Update the map of blocks we attempted to query. for client, partitions := range clients { touchedStores[client.RemoteAddress()] = struct{}{} + queriedStores += len(partitions) for _, part := range partitions { for _, blockID := range part { @@ -809,7 +816,7 @@ func (q *blocksStoreQuerier) queryWithConsistencyCheck( // The next attempt should just query the missing blocks. remainingBlocks = consistencyTracker.Check(queriedBlocks) if len(remainingBlocks) == 0 { - return storeGatewayQueryStats{storesHit: len(touchedStores), refetches: attempt - 1, queried: true}, nil + return storeGatewayQueryStats{storesHit: len(touchedStores), storesQueried: queriedStores, refetches: attempt - 1, queried: true}, nil } spanLog.DebugLog("msg", "couldn't query all blocks", "attempt", attempt, "missing blocks", strings.Join(convertULIDsToString(remainingBlocks.GetULIDs()), " ")) @@ -861,10 +868,11 @@ func (q *blocksStoreQuerier) forEachCompartment(ctx context.Context, targets []i // whole query, summed across the compartments that were queried. func (q *blocksStoreQuerier) queryCompartmentsWithConsistencyCheck(ctx context.Context, targets []int, spanLog *spanlogger.SpanLogger, minT, maxT int64, tenantID string, shard *sharding.ShardSelector, queryF queryFunc) error { var ( - statsMtx sync.Mutex - storesHit int - refetches int - anyQueried bool + statsMtx sync.Mutex + storesHit int + storesQueried int + refetches int + anyQueried bool ) err := q.forEachCompartment(ctx, targets, func(ctx context.Context, c blocksStoreCompartment) error { @@ -875,6 +883,7 @@ func (q *blocksStoreQuerier) queryCompartmentsWithConsistencyCheck(ctx context.C statsMtx.Lock() storesHit += stats.storesHit + storesQueried += stats.storesQueried refetches += stats.refetches anyQueried = anyQueried || stats.queried statsMtx.Unlock() @@ -885,6 +894,7 @@ func (q *blocksStoreQuerier) queryCompartmentsWithConsistencyCheck(ctx context.C } q.metrics.storesHit.Observe(float64(storesHit)) + q.metrics.storesQueried.Add(float64(storesQueried)) if q.metrics.compartmentsHit != nil { q.metrics.compartmentsHit.Observe(float64(len(targets))) } From f18839abab2f529fc92061e495a18ab74717516e Mon Sep 17 00:00:00 2001 From: Vladimir Varankin Date: Thu, 6 Aug 2026 12:57:24 +0200 Subject: [PATCH 4/7] more tests --- .../blocks_store_replicated_set_test.go | 28 ++++++++++++++----- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/pkg/querier/blocks_store_replicated_set_test.go b/pkg/querier/blocks_store_replicated_set_test.go index 40d936d7911..860a6faf4ad 100644 --- a/pkg/querier/blocks_store_replicated_set_test.go +++ b/pkg/querier/blocks_store_replicated_set_test.go @@ -64,13 +64,14 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { registeredAt := time.Now() tests := map[string]struct { - tenantShardSize int - replicationFactor int - setup func(*ring.Desc) - queryBlocks bucketindex.Blocks - exclude map[ulid.ULID][]string - expectedClients map[string][][]ulid.ULID - expectedErr error + tenantShardSize int + replicationFactor int + maxBlocksPerStoreRequest int + setup func(*ring.Desc) + queryBlocks bucketindex.Blocks + exclude map[ulid.ULID][]string + expectedClients map[string][][]ulid.ULID + expectedErr error }{ "shard size 0, single instance in the ring with RF = 1": { tenantShardSize: 0, @@ -326,6 +327,18 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { }, expectedErr: fmt.Errorf("no store-gateway instance left after checking exclude for block %s", blockID1.String()), }, + "max blocks per store request splits a single store-gateway's blocks into partitions": { + tenantShardSize: 0, + replicationFactor: 1, + maxBlocksPerStoreRequest: 2, + setup: func(d *ring.Desc) { + d.AddIngester("instance-1", "127.0.0.1", "", []uint32{block1Hash + 1}, ring.ACTIVE, registeredAt, false, time.Time{}, nil) + }, + queryBlocks: []*bucketindex.Block{block1, block2, block3, block4}, + expectedClients: map[string][][]ulid.ULID{ + "127.0.0.1": {{blockID1, blockID2}, {blockID3, blockID4}}, + }, + }, } for testName, testData := range tests { @@ -354,6 +367,7 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) { limits := &blocksStoreLimitsMock{ storeGatewayTenantShardSize: testData.tenantShardSize, + maxBlocksPerStoreRequest: testData.maxBlocksPerStoreRequest, } reg := prometheus.NewPedanticRegistry() From bfbd1437d7fcdd5d0a49948f0ff66f03e8f69082 Mon Sep 17 00:00:00 2001 From: Vladimir Varankin Date: Thu, 6 Aug 2026 13:14:32 +0200 Subject: [PATCH 5/7] validate limit's value --- pkg/util/validation/limits.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 2dd8401d890..600dc02cd0d 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -92,6 +92,7 @@ var ( errInvalidIngestStorageReadConsistency = fmt.Errorf("invalid ingest storage read consistency (supported values: %s)", strings.Join(api.ReadConsistencies, ", ")) errInvalidMaxEstimatedChunksPerQueryMultiplier = fmt.Errorf("invalid value for -%s: must be 0 or greater than or equal to 1", MaxEstimatedChunksPerQueryMultiplierFlag) errNegativeUpdateTimeoutJitterMax = errors.New("HA tracker max update timeout jitter shouldn't be negative") + errNegativeMaxBlocksPerStoreRequest = fmt.Errorf("-%s must be 0 or greater", MaxBlocksPerStoreRequestFlag) errInvalidFloatChunkEncoding = fmt.Errorf("invalid float chunk encoding (supported values: %q, %q)", promcfg.FloatChunkEncodingXOR, promcfg.FloatChunkEncodingXOR2) ) @@ -726,6 +727,10 @@ func (l *Limits) Validate() error { return errNegativeUpdateTimeoutJitterMax } + if l.MaxBlocksPerStoreRequest < 0 { + return errNegativeMaxBlocksPerStoreRequest + } + switch l.FloatChunkEncoding { case "", promcfg.FloatChunkEncodingXOR, promcfg.FloatChunkEncodingXOR2: default: From e3b55844ae61b815c1055750bac07f49cf66d265 Mon Sep 17 00:00:00 2001 From: Vladimir Varankin Date: Wed, 12 Aug 2026 15:04:08 +0200 Subject: [PATCH 6/7] make metric a histogram --- pkg/querier/blocks_store_queryable.go | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/pkg/querier/blocks_store_queryable.go b/pkg/querier/blocks_store_queryable.go index 7b9f887d95f..9bf15c60e3e 100644 --- a/pkg/querier/blocks_store_queryable.go +++ b/pkg/querier/blocks_store_queryable.go @@ -107,10 +107,10 @@ type BlocksStoreLimits interface { } type blocksStoreQueryableMetrics struct { - storesHit prometheus.Histogram - refetches prometheus.Histogram + storesHit prometheus.Histogram + storesQueried prometheus.Histogram + refetches prometheus.Histogram - storesQueried prometheus.Counter blocksFound prometheus.Counter blocksQueried prometheus.Counter blocksWithCompactorShardButIncompatibleQueryShard prometheus.Counter @@ -129,15 +129,18 @@ func newBlocksStoreQueryableMetrics(compartmentsCfg compartments.Config, reg pro Help: "Number of store-gateway instances hit for a single query.", Buckets: []float64{0, 1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048}, }), + storesQueried: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{ + Name: "cortex_querier_storegateway_queried_per_query", + Help: "Number of requests to store-gateway instances for a single query. Splitting queried blocks into partitions can produce more than one request per store-gateway instance per query.", + NativeHistogramBucketFactor: 1.1, + NativeHistogramMaxBucketNumber: 100, + NativeHistogramMinResetDuration: 1 * time.Hour, + }), refetches: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{ Name: "cortex_querier_storegateway_refetches_per_query", Help: "Number of re-fetches attempted while querying store-gateway instances due to missing blocks.", Buckets: []float64{0, 1, 2}, }), - storesQueried: promauto.With(reg).NewCounter(prometheus.CounterOpts{ - Name: "cortex_querier_storegateway_queried_total", - Help: "Total number of requests to store-gateway instances for a single query. Splitting queried blocks into partitions can produce more than one request per store-gateway instance per query.", - }), blocksFound: promauto.With(reg).NewCounter(prometheus.CounterOpts{ Name: "cortex_querier_blocks_found_total", Help: "Number of blocks found based on query time range.", @@ -894,7 +897,7 @@ func (q *blocksStoreQuerier) queryCompartmentsWithConsistencyCheck(ctx context.C } q.metrics.storesHit.Observe(float64(storesHit)) - q.metrics.storesQueried.Add(float64(storesQueried)) + q.metrics.storesQueried.Observe(float64(storesQueried)) if q.metrics.compartmentsHit != nil { q.metrics.compartmentsHit.Observe(float64(len(targets))) } From 13b4433ce690d1fd125f352a449de76a215730a7 Mon Sep 17 00:00:00 2001 From: Vladimir Varankin Date: Wed, 12 Aug 2026 15:37:19 +0200 Subject: [PATCH 7/7] more tests --- pkg/querier/blocks_store_queryable_test.go | 63 ++++++++++++++++++++-- 1 file changed, 60 insertions(+), 3 deletions(-) diff --git a/pkg/querier/blocks_store_queryable_test.go b/pkg/querier/blocks_store_queryable_test.go index d3302fc6900..23c76de7ac1 100644 --- a/pkg/querier/blocks_store_queryable_test.go +++ b/pkg/querier/blocks_store_queryable_test.go @@ -210,6 +210,34 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { cortex_querier_blocks_consistency_checks_total 1 `, }, + "a single store-gateway instance holds the required blocks split into multiple partitions": { + finderResult: bucketindex.Blocks{ + {ID: block1}, + {ID: block2}, + }, + storeSetResponses: []interface{}{ + map[BlocksStoreClient][][]ulid.ULID{ + &storeGatewayClientMock{remoteAddr: "1.1.1.1", mockedSeriesResponses: newSeriesResponseBuilder(). + addValue(metricNameLabel, minT, 1). + addValue(metricNameLabel, minT+1, 2). + addBlocks(block1, block2). + addFetchedIndexBytes(50). + build(), + }: {{block1}, {block2}}, + }, + }, + limits: &blocksStoreLimitsMock{}, + queryLimiter: noOpQueryLimiter, + expectedSeries: []seriesResult{ + { + lbls: metricNameLabel, + values: []valueResult{ + {t: minT, v: 1}, + {t: minT + 1, v: 2}, + }, + }, + }, + }, "a single store-gateway instance holds the required blocks (single returned series) - multiple chunks per series for stats": { finderResult: bucketindex.Blocks{ {ID: block1}, @@ -1697,15 +1725,17 @@ func TestBlocksStoreQuerier_Select(t *testing.T) { continue } - for k := range m { + for k, partitions := range m { mockClient := k.(*storeGatewayClientMock) + // Each partition is a separate RPC to the same client, so the mocked responses + // (and thus the fetched series/chunks) are received once per partition. for _, sr := range mockClient.mockedSeriesResponses { if s := sr.GetStreamingSeries(); s != nil { - seriesCount += len(s.Series) + seriesCount += len(s.Series) * len(partitions) } if c := sr.GetStreamingChunksEstimate(); c != nil { - chunksCount += int(c.EstimatedChunkCount) + chunksCount += int(c.EstimatedChunkCount) * len(partitions) } } } @@ -2240,6 +2270,33 @@ func TestBlocksStoreQuerier_Labels(t *testing.T) { expectedLabelNames: namesFromSeries(series1, series2), expectedLabelValues: valuesFromSeries(model.MetricNameLabel, series1, series2), }, + "a single store-gateway instance holds the required blocks split into multiple partitions": { + finderResult: bucketindex.Blocks{ + {ID: block1}, + {ID: block2}, + }, + storeSetResponses: []interface{}{ + map[BlocksStoreClient][][]ulid.ULID{ + &storeGatewayClientMock{ + remoteAddr: "1.1.1.1", + mockedLabelNamesResponse: &storepb.LabelNamesResponse{ + Names: namesFromSeries(series1, series2), + Warnings: []string{}, + Hints: mockNamesHints(block1, block2), + ResponseHints: mockNamesResponseHints(block1, block2), + }, + mockedLabelValuesResponse: &storepb.LabelValuesResponse{ + Values: valuesFromSeries(model.MetricNameLabel, series1, series2), + Warnings: []string{}, + Hints: mockValuesHints(block1, block2), + ResponseHints: mockValuesResponseHints(block1, block2), + }, + }: {{block1}, {block2}}, + }, + }, + expectedLabelNames: namesFromSeries(series1, series2), + expectedLabelValues: valuesFromSeries(model.MetricNameLabel, series1, series2), + }, "a single store-gateway instance holds the required blocks with only non-opaque response hints": { finderResult: bucketindex.Blocks{ {ID: block1},