add touch series limit

Author: lisuo

cmd/thanos/store.go

@@ -67,6 +67,9 @@ func registerStore(app *extkingpin.App) {
 	maxSampleCount := cmd.Flag("store.grpc.series-sample-limit",
 		"Maximum amount of samples returned via a single Series call. The Series call fails if this limit is exceeded. 0 means no limit. NOTE: For efficiency the limit is internally implemented as 'chunks limit' considering each chunk contains 120 samples (it's the max number of samples each chunk can contain), so the actual number of samples might be lower, even though the maximum could be hit.").
 		Default("0").Uint()
+	maxTouchSeriesCount := cmd.Flag("store.grpc.touch-series-limit",
+		"Maximum amount of touch series returned via a single Series call. The Series call fails if this limit is exceeded. 0 means no limit.").
+		Default("0").Uint()
 
 	maxConcurrent := cmd.Flag("store.grpc.series-max-concurrency", "Maximum number of concurrent Series calls.").Default("20").Int()
 
@@ -136,6 +139,7 @@ func registerStore(app *extkingpin.App) {
 			uint64(*indexCacheSize),
 			uint64(*chunkPoolSize),
 			uint64(*maxSampleCount),
+			uint64(*maxTouchSeriesCount),
 			*maxConcurrent,
 			component.Store,
 			debugLogging,
@@ -173,7 +177,7 @@ func runStore(
 	grpcGracePeriod time.Duration,
 	grpcCert, grpcKey, grpcClientCA, httpBindAddr string,
 	httpGracePeriod time.Duration,
-	indexCacheSizeBytes, chunkPoolSizeBytes, maxSampleCount uint64,
+	indexCacheSizeBytes, chunkPoolSizeBytes, maxSampleCount, maxSeriesCount uint64,
 	maxConcurrency int,
 	component component.Component,
 	verbose bool,
@@ -303,6 +307,7 @@ func runStore(
 		queriesGate,
 		chunkPoolSizeBytes,
 		store.NewChunksLimiterFactory(maxSampleCount/store.MaxSamplesPerChunk), // The samples limit is an approximation based on the max number of samples per chunk.
+		store.NewSeriesLimiterFactory(maxSeriesCount),
 		verbose,
 		blockSyncConcurrency,
 		filterConf,

pkg/store/bucket.go

@@ -105,6 +105,7 @@ type bucketStoreMetrics struct {
 	resultSeriesCount     prometheus.Summary
 	chunkSizeBytes        prometheus.Histogram
 	queriesDropped        prometheus.Counter
+	queriesSeriesDropped  prometheus.Counter
 	seriesRefetches       prometheus.Counter
 
 	cachedPostingsCompressions           *prometheus.CounterVec
@@ -190,6 +191,10 @@ func newBucketStoreMetrics(reg prometheus.Registerer) *bucketStoreMetrics {
 		Name: "thanos_bucket_store_queries_dropped_total",
 		Help: "Number of queries that were dropped due to the sample limit.",
 	})
+	m.queriesSeriesDropped = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name: "thanos_bucket_store_queries_series_dropped_total",
+		Help: "Number of queries that were dropped due to the series limit.",
+	})
 	m.seriesRefetches = promauto.With(reg).NewCounter(prometheus.CounterOpts{
 		Name: "thanos_bucket_store_series_refetches_total",
 		Help: fmt.Sprintf("Total number of cases where %v bytes was not enough was to fetch series from index, resulting in refetch.", maxSeriesSize),
@@ -276,6 +281,8 @@ type BucketStore struct {
 
 	// chunksLimiterFactory creates a new limiter used to limit the number of chunks fetched by each Series() call.
 	chunksLimiterFactory ChunksLimiterFactory
+	// seriesLimiterFactory creates a new limiter used to limit the number of touch series by each Series() call.
+	seriesLimiterFactory SeriesLimiterFactory
 	partitioner          partitioner
 
 	filterConfig             *FilterConfig
@@ -300,6 +307,7 @@ func NewBucketStore(
 	queryGate gate.Gate,
 	maxChunkPoolBytes uint64,
 	chunksLimiterFactory ChunksLimiterFactory,
+	seriesLimiterFactory SeriesLimiterFactory,
 	debugLogging bool,
 	blockSyncConcurrency int,
 	filterConfig *FilterConfig,
@@ -333,6 +341,7 @@ func NewBucketStore(
 		filterConfig:                filterConfig,
 		queryGate:                   queryGate,
 		chunksLimiterFactory:        chunksLimiterFactory,
+		seriesLimiterFactory:        seriesLimiterFactory,
 		partitioner:                 gapBasedPartitioner{maxGapSize: partitionerMaxGapSize},
 		enableCompatibilityLabel:    enableCompatibilityLabel,
 		postingOffsetsInMemSampling: postingOffsetsInMemSampling,
@@ -683,6 +692,7 @@ func blockSeries(
 	matchers []*labels.Matcher,
 	req *storepb.SeriesRequest,
 	chunksLimiter ChunksLimiter,
+	seriesLimiter SeriesLimiter,
 ) (storepb.SeriesSet, *queryStats, error) {
 	ps, err := indexr.ExpandedPostings(matchers)
 	if err != nil {
@@ -693,6 +703,11 @@ func blockSeries(
 		return storepb.EmptySeriesSet(), indexr.stats, nil
 	}
 
+	// Reserve seriesLimiter
+	if err := seriesLimiter.Reserve(uint64(len(ps))); err != nil {
+		return nil, nil, errors.Wrap(err, "exceeded series limit")
+	}
+
 	// Preload all series index data.
 	// TODO(bwplotka): Consider not keeping all series in memory all the time.
 	// TODO(bwplotka): Do lazy loading in one step as `ExpandingPostings` method.
@@ -887,6 +902,7 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 		resHints         = &hintspb.SeriesResponseHints{}
 		reqBlockMatchers []*labels.Matcher
 		chunksLimiter    = s.chunksLimiterFactory(s.metrics.queriesDropped)
+		seriesLimiter    = s.seriesLimiterFactory(s.metrics.queriesSeriesDropped)
 	)
 
 	if req.Hints != nil {
@@ -942,6 +958,7 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 					blockMatchers,
 					req,
 					chunksLimiter,
+					seriesLimiter,
 				)
 				if err != nil {
 					return errors.Wrapf(err, "fetch series for block %s", b.meta.ULID)

pkg/store/limiter.go

@@ -18,10 +18,19 @@ type ChunksLimiter interface {
 	Reserve(num uint64) error
 }
 
+type SeriesLimiter interface {
+	// Reserve num series out of the total number of series enforced by the limiter.
+	// Returns an error if the limit has been exceeded. This function must be
+	// goroutine safe.
+	Reserve(num uint64) error
+}
+
 // ChunksLimiterFactory is used to create a new ChunksLimiter. The factory is useful for
 // projects depending on Thanos (eg. Cortex) which have dynamic limits.
 type ChunksLimiterFactory func(failedCounter prometheus.Counter) ChunksLimiter
 
+type SeriesLimiterFactory func(failedCounter prometheus.Counter) SeriesLimiter
+
 // Limiter is a simple mechanism for checking if something has passed a certain threshold.
 type Limiter struct {
 	limit    uint64
@@ -57,3 +66,10 @@ func NewChunksLimiterFactory(limit uint64) ChunksLimiterFactory {
 		return NewLimiter(limit, failedCounter)
 	}
 }
+
+// NewSeriesLimiterFactory makes a new NewSeriesLimiterFactory with a static limit.
+func NewSeriesLimiterFactory(limit uint64) SeriesLimiterFactory {
+	return func(failedCounter prometheus.Counter) SeriesLimiter {
+		return NewLimiter(limit, failedCounter)
+	}
+}