Skip index queries for high cardinality labels.

Firstly, cache the length of index rows we query (by the hash and range key).  Secondly, fail for rows with > 100k, either because the cache told us so, or because we read them.  Finally, allow matchers to fail on cardinality errors but proceed with the query (as long as at least one matcher succeeds), and then filter results.

Notably, after this change, queries on two high-cardinality labels that would have results in a small number of series will fail.

Signed-off-by: Tom Wilkie <[email protected]>

Author: tomwilkie

pkg/chunk/chunk_store.go

@@ -52,15 +52,21 @@ func init() {
 type StoreConfig struct {
 	CacheConfig cache.Config
 
-	MinChunkAge     time.Duration
-	QueryChunkLimit int
+	MinChunkAge              time.Duration
+	QueryChunkLimit          int
+	CardinalityCacheSize     int
+	CardinalityCacheValidity time.Duration
+	CardinalityLimit         int
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
 func (cfg *StoreConfig) RegisterFlags(f *flag.FlagSet) {
 	cfg.CacheConfig.RegisterFlags(f)
 	f.DurationVar(&cfg.MinChunkAge, "store.min-chunk-age", 0, "Minimum time between chunk update and being saved to the store.")
 	f.IntVar(&cfg.QueryChunkLimit, "store.query-chunk-limit", 2e6, "Maximum number of chunks that can be fetched in a single query.")
+	f.IntVar(&cfg.CardinalityCacheSize, "store.cardinality-cache-size", 0, "Size of in-memory cardinality cache, 0 to disable.")
+	f.DurationVar(&cfg.CardinalityCacheValidity, "store.cardinality-cache-validity", 1*time.Hour, "Period for which entries in the cardinality cache are valid.")
+	f.IntVar(&cfg.CardinalityLimit, "store.cardinality-limit", 1e5, "Cardinality limit for index queries.")
 }
 
 // store implements Store

pkg/chunk/series_store.go

@@ -2,7 +2,9 @@ package chunk
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"time"
 
 	"github.com/go-kit/kit/log/level"
 	"github.com/prometheus/common/model"
@@ -13,9 +15,14 @@ import (
 	"github.com/weaveworks/cortex/pkg/util/extract"
 )
 
+var (
+	errCardinalityExceeded = errors.New("cardinality limit exceeded")
+)
+
 // seriesStore implements Store
 type seriesStore struct {
 	store
+	cardinalityCache *fifoCache
 }
 
 func newSeriesStore(cfg StoreConfig, schema Schema, storage StorageClient) (Store, error) {
@@ -31,6 +38,7 @@ func newSeriesStore(cfg StoreConfig, schema Schema, storage StorageClient) (Stor
 			schema:       schema,
 			chunkFetcher: fetcher,
 		},
+		cardinalityCache: newFifoCache(cfg.CardinalityCacheSize),
 	}, nil
 }
 
@@ -49,15 +57,15 @@ func (c *seriesStore) Get(ctx context.Context, from, through model.Time, allMatc
 	}
 
 	// Ensure this query includes a metric name.
-	metricNameMatcher, matchers, ok := extract.MetricNameMatcherFromMatchers(allMatchers)
+	metricNameMatcher, allMatchers, ok := extract.MetricNameMatcherFromMatchers(allMatchers)
 	if !ok || metricNameMatcher.Type != labels.MatchEqual {
 		return nil, fmt.Errorf("query must contain metric name")
 	}
 	level.Debug(log).Log("metric", metricNameMatcher.Value)
 
 	// Fetch the series IDs from the index, based on non-empty matchers from
 	// the query.
-	filters, matchers := util.SplitFiltersAndMatchers(matchers)
+	_, matchers := util.SplitFiltersAndMatchers(allMatchers)
 	seriesIDs, err := c.lookupSeriesByMetricNameMatchers(ctx, from, through, metricNameMatcher.Value, matchers)
 	if err != nil {
 		return nil, err
@@ -94,7 +102,7 @@ func (c *seriesStore) Get(ctx context.Context, from, through model.Time, allMatc
 	}
 
 	// Filter out chunks based on the empty matchers in the query.
-	filteredChunks := filterChunksByMatchers(allChunks, filters)
+	filteredChunks := filterChunksByMatchers(allChunks, allMatchers)
 	return filteredChunks, nil
 }
 
@@ -124,6 +132,7 @@ func (c *seriesStore) lookupSeriesByMetricNameMatchers(ctx context.Context, from
 	// Receive chunkSets from all matchers
 	var ids []string
 	var lastErr error
+	var cardinalityExceededErrors int
 	for i := 0; i < len(matchers); i++ {
 		select {
 		case incoming := <-incomingIDs:
@@ -133,10 +142,16 @@ func (c *seriesStore) lookupSeriesByMetricNameMatchers(ctx context.Context, from
 				ids = intersectStrings(ids, incoming)
 			}
 		case err := <-incomingErrors:
-			lastErr = err
+			if err == errCardinalityExceeded {
+				cardinalityExceededErrors++
+			} else {
+				lastErr = err
+			}
 		}
 	}
-	if lastErr != nil {
+	if cardinalityExceededErrors == len(matchers) {
+		return nil, errCardinalityExceeded
+	} else if lastErr != nil {
 		return nil, lastErr
 	}
 
@@ -166,12 +181,32 @@ func (c *seriesStore) lookupSeriesByMetricNameMatcher(ctx context.Context, from,
 	}
 	level.Debug(log).Log("queries", len(queries))
 
+	for _, query := range queries {
+		value, updated, ok := c.cardinalityCache.get(query.HashValue)
+		if !ok {
+			continue
+		}
+		entryAge := time.Now().Sub(updated)
+		cardinality := value.(int)
+		if entryAge < c.cfg.CardinalityCacheValidity && cardinality > c.cfg.CardinalityLimit {
+			return nil, errCardinalityExceeded
+		}
+	}
+
 	entries, err := c.lookupEntriesByQueries(ctx, queries)
 	if err != nil {
 		return nil, err
 	}
 	level.Debug(log).Log("entries", len(entries))
 
+	// TODO This is not correct, will overcount for queries > 24hrs
+	for _, query := range queries {
+		c.cardinalityCache.put(query.HashValue, len(entries))
+	}
+	if len(entries) > c.cfg.CardinalityLimit {
+		return nil, errCardinalityExceeded
+	}
+
 	ids, err := c.parseIndexEntries(ctx, entries, matcher)
 	if err != nil {
 		return nil, err