Limit queried chunks by bytes #3089
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -64,6 +64,9 @@ const ( | |
| maxChunkSize = 16000 | ||
| maxSeriesSize = 64 * 1024 | ||
|
|
||
| noChunksLimit = 0 | ||
| noChunksSizeLimit = 0 | ||
|
|
||
| // CompatibilityTypeLabelName is an artificial label that Store Gateway can optionally advertise. This is required for compatibility | ||
| // with pre v0.8.0 Querier. Previous Queriers was strict about duplicated external labels of all StoreAPIs that had any labels. | ||
| // Now with newer Store Gateway advertising all the external labels it has access to, there was simple case where | ||
|
|
@@ -270,7 +273,9 @@ type BucketStore struct { | |
|
|
||
| // chunksLimiterFactory creates a new limiter used to limit the number of chunks fetched by each Series() call. | ||
| chunksLimiterFactory ChunksLimiterFactory | ||
| // chunksSizeLimiterFactory creates a new limiter used to limit the size of chunks fetched by each Series() call. | ||
| chunksSizeLimiterFactory ChunksLimiterFactory | ||
| partitioner partitioner | ||
|
|
||
| filterConfig *FilterConfig | ||
| advLabelSets []labelpb.ZLabelSet | ||
|
|
@@ -286,6 +291,42 @@ type BucketStore struct { | |
| enableSeriesResponseHints bool | ||
| } | ||
|
|
||
| type bucketStoreOptions struct { | ||
| chunksLimitFactory ChunksLimiterFactory | ||
| chunksSizeLimitFactory ChunksLimiterFactory | ||
| } | ||
|
|
||
| // BucketStoreOption overrides options of the bucket store. | ||
| type BucketStoreOption interface { | ||
| apply(*bucketStoreOptions) | ||
| } | ||
|
|
||
| type bucketStoreOptionFunc func(*bucketStoreOptions) | ||
|
|
||
| func (f bucketStoreOptionFunc) apply(o *bucketStoreOptions) { | ||
| f(o) | ||
| } | ||
|
|
||
| // WithChunksLimit sets chunks limit for the bucket store. | ||
| func WithChunksLimit(f ChunksLimiterFactory) BucketStoreOption { | ||
| return bucketStoreOptionFunc(func(lo *bucketStoreOptions) { | ||
| lo.chunksLimitFactory = f | ||
| }) | ||
| } | ||
|
|
||
| // WithChunksSizeLimit sets chunks size limit for the bucket store. | ||
| func WithChunksSizeLimit(f ChunksLimiterFactory) BucketStoreOption { | ||
| return bucketStoreOptionFunc(func(lo *bucketStoreOptions) { | ||
| lo.chunksSizeLimitFactory = f | ||
| }) | ||
| } | ||
|
|
||
| var defaultBucketStoreOptions = bucketStoreOptions{ | ||
| chunksLimitFactory: NewChunksLimiterFactory(noChunksLimit), | ||
| chunksSizeLimitFactory: NewChunksLimiterFactory(noChunksSizeLimit), | ||
| } | ||
|
|
||
| // Deprecated. Use NewBucketStoreWithOptions instead. | ||
| // NewBucketStore creates a new bucket backed store that implements the store API against | ||
| // an object store bucket. It is optimized to work against high latency backends. | ||
| func NewBucketStore( | ||
|
|
@@ -305,6 +346,49 @@ func NewBucketStore( | |
| enablePostingsCompression bool, | ||
| postingOffsetsInMemSampling int, | ||
| enableSeriesResponseHints bool, // TODO(pracucci) Thanos 0.12 and below doesn't gracefully handle new fields in SeriesResponse. Drop this flag and always enable hints once we can drop backward compatibility. | ||
| ) (*BucketStore, error) { | ||
| bucketStoreOpts := []BucketStoreOption{ | ||
| WithChunksLimit(chunksLimiterFactory), | ||
| WithChunksSizeLimit(NewChunksLimiterFactory(noChunksSizeLimit)), | ||
| } | ||
| return NewBucketStoreWithOptions(logger, | ||
| reg, | ||
| bkt, | ||
| fetcher, | ||
| dir, | ||
| indexCache, | ||
| queryGate, | ||
| maxChunkPoolBytes, | ||
| debugLogging, | ||
| blockSyncConcurrency, | ||
| filterConfig, | ||
| enableCompatibilityLabel, | ||
| enablePostingsCompression, | ||
| postingOffsetsInMemSampling, | ||
| enableSeriesResponseHints, | ||
| bucketStoreOpts..., | ||
| ) | ||
| } | ||
|
|
||
| // NewBucketStoreWithOptions creates a new bucket backed store that implements the store API against | ||
| // an object store bucket. It is optimized to work against high latency backends. | ||
| func NewBucketStoreWithOptions( | ||
|
|
||
| logger log.Logger, | ||
| reg prometheus.Registerer, | ||
| bkt objstore.InstrumentedBucketReader, | ||
| fetcher block.MetadataFetcher, | ||
| dir string, | ||
| indexCache storecache.IndexCache, | ||
| queryGate gate.Gate, | ||
| maxChunkPoolBytes uint64, | ||
| debugLogging bool, | ||
| blockSyncConcurrency int, | ||
| filterConfig *FilterConfig, | ||
| enableCompatibilityLabel bool, | ||
| enablePostingsCompression bool, | ||
| postingOffsetsInMemSampling int, | ||
| enableSeriesResponseHints bool, // TODO(pracucci) Thanos 0.12 and below doesn't gracefully handle new fields in SeriesResponse. Drop this flag and always enable hints once we can drop backward compatibility. | ||
| opt ...BucketStoreOption, | ||
| ) (*BucketStore, error) { | ||
| if logger == nil { | ||
| logger = log.NewNopLogger() | ||
|
|
@@ -315,6 +399,11 @@ func NewBucketStore( | |
| return nil, errors.Wrap(err, "create chunk pool") | ||
| } | ||
|
|
||
| opts := defaultBucketStoreOptions | ||
| for _, o := range opt { | ||
| o.apply(&opts) | ||
| } | ||
|
|
||
| s := &BucketStore{ | ||
| logger: logger, | ||
| bkt: bkt, | ||
|
|
@@ -328,7 +417,8 @@ func NewBucketStore( | |
| blockSyncConcurrency: blockSyncConcurrency, | ||
| filterConfig: filterConfig, | ||
| queryGate: queryGate, | ||
| chunksLimiterFactory: opts.chunksLimitFactory, | ||
| chunksSizeLimiterFactory: opts.chunksSizeLimitFactory, | ||
| partitioner: gapBasedPartitioner{maxGapSize: partitionerMaxGapSize}, | ||
| enableCompatibilityLabel: enableCompatibilityLabel, | ||
| enablePostingsCompression: enablePostingsCompression, | ||
|
|
@@ -679,6 +769,7 @@ func blockSeries( | |
| matchers []*labels.Matcher, | ||
| req *storepb.SeriesRequest, | ||
| chunksLimiter ChunksLimiter, | ||
| chunksSizeLimiter ChunksLimiter, | ||
| ) (storepb.SeriesSet, *queryStats, error) { | ||
| ps, err := indexr.ExpandedPostings(matchers) | ||
| if err != nil { | ||
|
|
@@ -774,6 +865,10 @@ func blockSeries( | |
| return nil, nil, errors.Wrap(err, "preload chunks") | ||
| } | ||
|
|
||
| if err := chunksSizeLimiter.Reserve(uint64(chunkr.stats.chunksFetchedSizeSum)); err != nil { | ||
| return nil, nil, errors.Wrap(err, "exceeded chunks size limit") | ||
| } | ||
|
|
||
| // Transform all chunks into the response format. | ||
| for _, s := range res { | ||
| for i, ref := range s.refs { | ||
|
|
@@ -894,16 +989,22 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie | |
| req.MaxTime = s.limitMaxTime(req.MaxTime) | ||
|
|
||
| var ( | ||
| ctx = srv.Context() | ||
| stats = &queryStats{} | ||
| res []storepb.SeriesSet | ||
| mtx sync.Mutex | ||
| g, gctx = errgroup.WithContext(ctx) | ||
| resHints = &hintspb.SeriesResponseHints{} | ||
| reqBlockMatchers []*labels.Matcher | ||
| chunksLimiter = s.chunksLimiterFactory(s.metrics.queriesDropped) | ||
| chunksSizeLimiter = s.chunksSizeLimiterFactory(s.metrics.queriesDropped) | ||
| ) | ||
|
|
||
| chunksSizeLimiter, err = chunksSizeLimiter.NewWithFailedCounterFrom(chunksLimiter) | ||
|
There was a problem hiding this comment. Could you clarify why you need There was a problem hiding this comment. @pracucci thanks for reviewing this. There was a problem hiding this comment. more info on that: There was a problem hiding this comment. I think There's no problem updating a metric concurrently, but we may want to distinguish the reason why a query was dropped. An option may be adding a "reason" label to |
||
| if err != nil { | ||
| return status.Error(codes.InvalidArgument, err.Error()) | ||
| } | ||
|
|
||
| if req.Hints != nil { | ||
| reqHints := &hintspb.SeriesRequestHints{} | ||
| if err := types.UnmarshalAny(req.Hints, reqHints); err != nil { | ||
|
|
@@ -957,6 +1058,7 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie | |
| blockMatchers, | ||
| req, | ||
| chunksLimiter, | ||
| chunksSizeLimiter, | ||
| ) | ||
| if err != nil { | ||
| return errors.Wrapf(err, "fetch series for block %s", b.meta.ULID) | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@bwplotka What do you think about this? I've mixed feelings about it, just to avoid adding another argument to
NewBucketStore(). Maybe we can rollback options in this PR and follow up the discussion about functional options in #3931?