diff --git a/CHANGELOG.md b/CHANGELOG.md index 25f51a840a..dc11656c0a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ * [FEATURE] Distributor: Accept multiple HA Tracker pairs in the same request. #6256 * [FEATURE] Ruler: Add support for per-user external labels #6340 * [FEATURE] Query Frontend: Support an exemplar federated query when `-tenant-federation.enabled=true`. #6455 +* [FEATURE] Ingester: Add support for cache query matchers via `-ingester.matchers-cache-max-items. #6477 * [ENHANCEMENT] Querier: Add a `-tenant-federation.max-concurrent` flags to configure the number of worker processing federated query and add a `cortex_querier_federated_tenants_per_query` histogram to track the number of tenants per query. #6449 * [ENHANCEMENT] Query Frontend: Add a number of series in the query response to the query stat log. #6423 * [ENHANCEMENT] Store Gateway: Add a hedged request to reduce the tail latency. #6388 diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 6a36afefb0..d10a027131 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -3181,6 +3181,10 @@ instance_limits: # change by changing this option. # CLI flag: -ingester.disable-chunk-trimming [disable_chunk_trimming: | default = false] + +# Maximum number of entries in the matchers cache. 0 to disable. +# CLI flag: -ingester.matchers-cache-max-items +[matchers_cache_max_items: | default = 0] ``` ### `ingester_client_config` diff --git a/integration/query_fuzz_test.go b/integration/query_fuzz_test.go index 85fe9fca2e..cb2e851425 100644 --- a/integration/query_fuzz_test.go +++ b/integration/query_fuzz_test.go @@ -378,8 +378,9 @@ func TestExpandedPostingsCacheFuzz(t *testing.T) { "-blocks-storage.expanded_postings_cache.head.enabled": "true", "-blocks-storage.expanded_postings_cache.block.enabled": "true", // Ingester. - "-ring.store": "consul", - "-consul.hostname": consul2.NetworkHTTPEndpoint(), + "-ring.store": "consul", + "-consul.hostname": consul2.NetworkHTTPEndpoint(), + "-ingester.matchers-cache-max-items": "10000", // Distributor. "-distributor.replication-factor": "1", // Store-gateway. diff --git a/pkg/distributor/distributor_test.go b/pkg/distributor/distributor_test.go index 4ac187ffd7..b6d1cee24c 100644 --- a/pkg/distributor/distributor_test.go +++ b/pkg/distributor/distributor_test.go @@ -26,6 +26,7 @@ import ( "github.com/prometheus/prometheus/tsdb/tsdbutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + storecache "github.com/thanos-io/thanos/pkg/store/cache" "github.com/weaveworks/common/httpgrpc" "github.com/weaveworks/common/user" "go.uber.org/atomic" @@ -3374,7 +3375,7 @@ func (i *mockIngester) Query(ctx context.Context, req *client.QueryRequest, opts return nil, errFail } - _, _, matchers, err := client.FromQueryRequest(req) + _, _, matchers, err := client.FromQueryRequest(storecache.NewNoopMatcherCache(), req) if err != nil { return nil, err } @@ -3400,7 +3401,7 @@ func (i *mockIngester) QueryStream(ctx context.Context, req *client.QueryRequest return nil, errFail } - _, _, matchers, err := client.FromQueryRequest(req) + _, _, matchers, err := client.FromQueryRequest(storecache.NewNoopMatcherCache(), req) if err != nil { return nil, err } @@ -3459,7 +3460,7 @@ func (i *mockIngester) MetricsForLabelMatchersStream(ctx context.Context, req *c return nil, errFail } - _, _, _, multiMatchers, err := client.FromMetricsForLabelMatchersRequest(req) + _, _, _, multiMatchers, err := client.FromMetricsForLabelMatchersRequest(storecache.NewNoopMatcherCache(), req) if err != nil { return nil, err } @@ -3491,7 +3492,7 @@ func (i *mockIngester) MetricsForLabelMatchers(ctx context.Context, req *client. return nil, errFail } - _, _, _, multiMatchers, err := client.FromMetricsForLabelMatchersRequest(req) + _, _, _, multiMatchers, err := client.FromMetricsForLabelMatchersRequest(storecache.NewNoopMatcherCache(), req) if err != nil { return nil, err } diff --git a/pkg/ingester/client/compat.go b/pkg/ingester/client/compat.go index 1a8e178641..3a81ca6ea6 100644 --- a/pkg/ingester/client/compat.go +++ b/pkg/ingester/client/compat.go @@ -7,6 +7,7 @@ import ( "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/tsdb/chunkenc" + storecache "github.com/thanos-io/thanos/pkg/store/cache" "github.com/cortexproject/cortex/pkg/cortexpb" ) @@ -26,8 +27,8 @@ func ToQueryRequest(from, to model.Time, matchers []*labels.Matcher) (*QueryRequ } // FromQueryRequest unpacks a QueryRequest proto. -func FromQueryRequest(req *QueryRequest) (model.Time, model.Time, []*labels.Matcher, error) { - matchers, err := FromLabelMatchers(req.Matchers) +func FromQueryRequest(cache storecache.MatchersCache, req *QueryRequest) (model.Time, model.Time, []*labels.Matcher, error) { + matchers, err := FromLabelMatchers(cache, req.Matchers) if err != nil { return 0, 0, nil, err } @@ -55,10 +56,10 @@ func ToExemplarQueryRequest(from, to model.Time, matchers ...[]*labels.Matcher) } // FromExemplarQueryRequest unpacks a ExemplarQueryRequest proto. -func FromExemplarQueryRequest(req *ExemplarQueryRequest) (int64, int64, [][]*labels.Matcher, error) { +func FromExemplarQueryRequest(cache storecache.MatchersCache, req *ExemplarQueryRequest) (int64, int64, [][]*labels.Matcher, error) { var result [][]*labels.Matcher for _, m := range req.Matchers { - matchers, err := FromLabelMatchers(m.Matchers) + matchers, err := FromLabelMatchers(cache, m.Matchers) if err != nil { return 0, 0, nil, err } @@ -175,10 +176,10 @@ func SeriesSetToQueryResponse(s storage.SeriesSet) (*QueryResponse, error) { } // FromMetricsForLabelMatchersRequest unpacks a MetricsForLabelMatchersRequest proto -func FromMetricsForLabelMatchersRequest(req *MetricsForLabelMatchersRequest) (model.Time, model.Time, int, [][]*labels.Matcher, error) { +func FromMetricsForLabelMatchersRequest(cache storecache.MatchersCache, req *MetricsForLabelMatchersRequest) (model.Time, model.Time, int, [][]*labels.Matcher, error) { matchersSet := make([][]*labels.Matcher, 0, len(req.MatchersSet)) for _, matchers := range req.MatchersSet { - matchers, err := FromLabelMatchers(matchers.Matchers) + matchers, err := FromLabelMatchers(cache, matchers.Matchers) if err != nil { return 0, 0, 0, nil, err } @@ -206,12 +207,12 @@ func ToLabelValuesRequest(labelName model.LabelName, from, to model.Time, limit } // FromLabelValuesRequest unpacks a LabelValuesRequest proto -func FromLabelValuesRequest(req *LabelValuesRequest) (string, int64, int64, int, []*labels.Matcher, error) { +func FromLabelValuesRequest(cache storecache.MatchersCache, req *LabelValuesRequest) (string, int64, int64, int, []*labels.Matcher, error) { var err error var matchers []*labels.Matcher if req.Matchers != nil { - matchers, err = FromLabelMatchers(req.Matchers.Matchers) + matchers, err = FromLabelMatchers(cache, req.Matchers.Matchers) if err != nil { return "", 0, 0, 0, nil, err } @@ -236,12 +237,12 @@ func ToLabelNamesRequest(from, to model.Time, limit int, matchers []*labels.Matc } // FromLabelNamesRequest unpacks a LabelNamesRequest proto -func FromLabelNamesRequest(req *LabelNamesRequest) (int64, int64, int, []*labels.Matcher, error) { +func FromLabelNamesRequest(cache storecache.MatchersCache, req *LabelNamesRequest) (int64, int64, int, []*labels.Matcher, error) { var err error var matchers []*labels.Matcher if req.Matchers != nil { - matchers, err = FromLabelMatchers(req.Matchers.Matchers) + matchers, err = FromLabelMatchers(cache, req.Matchers.Matchers) if err != nil { return 0, 0, 0, nil, err } @@ -275,27 +276,31 @@ func toLabelMatchers(matchers []*labels.Matcher) ([]*LabelMatcher, error) { return result, nil } -func FromLabelMatchers(matchers []*LabelMatcher) ([]*labels.Matcher, error) { +func FromLabelMatchers(cache storecache.MatchersCache, matchers []*LabelMatcher) ([]*labels.Matcher, error) { result := make([]*labels.Matcher, 0, len(matchers)) for _, matcher := range matchers { - var mtype labels.MatchType - switch matcher.Type { - case EQUAL: - mtype = labels.MatchEqual - case NOT_EQUAL: - mtype = labels.MatchNotEqual - case REGEX_MATCH: - mtype = labels.MatchRegexp - case REGEX_NO_MATCH: - mtype = labels.MatchNotRegexp - default: - return nil, fmt.Errorf("invalid matcher type") - } - matcher, err := labels.NewMatcher(mtype, matcher.Name, matcher.Value) + m, err := cache.GetOrSet(matcher.String(), func() (*labels.Matcher, error) { + var mtype labels.MatchType + switch matcher.Type { + case EQUAL: + mtype = labels.MatchEqual + case NOT_EQUAL: + mtype = labels.MatchNotEqual + case REGEX_MATCH: + mtype = labels.MatchRegexp + case REGEX_NO_MATCH: + mtype = labels.MatchNotRegexp + default: + return nil, fmt.Errorf("invalid matcher type") + } + return labels.NewMatcher(mtype, matcher.GetName(), matcher.GetValue()) + }) + if err != nil { return nil, err } - result = append(result, matcher) + + result = append(result, m) } return result, nil } diff --git a/pkg/ingester/client/compat_test.go b/pkg/ingester/client/compat_test.go index c9467abd0a..43e1ac3c7e 100644 --- a/pkg/ingester/client/compat_test.go +++ b/pkg/ingester/client/compat_test.go @@ -7,6 +7,7 @@ import ( "github.com/prometheus/common/model" "github.com/prometheus/prometheus/model/labels" + storecache "github.com/thanos-io/thanos/pkg/store/cache" ) func TestQueryRequest(t *testing.T) { @@ -41,7 +42,7 @@ func TestQueryRequest(t *testing.T) { t.Fatal(err) } - haveFrom, haveTo, haveMatchers, err := FromQueryRequest(req) + haveFrom, haveTo, haveMatchers, err := FromQueryRequest(storecache.NewNoopMatcherCache(), req) if err != nil { t.Fatal(err) } diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index 1d29b8b95d..3d01c10199 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -37,6 +37,7 @@ import ( "github.com/thanos-io/objstore" "github.com/thanos-io/thanos/pkg/block/metadata" "github.com/thanos-io/thanos/pkg/shipper" + storecache "github.com/thanos-io/thanos/pkg/store/cache" "github.com/thanos-io/thanos/pkg/store/storepb" "github.com/weaveworks/common/httpgrpc" "go.uber.org/atomic" @@ -145,6 +146,9 @@ type Config struct { // When disabled, the result may contain samples outside the queried time range but Select() performances // may be improved. DisableChunkTrimming bool `yaml:"disable_chunk_trimming"` + + // Maximum number of entries in the matchers cache. 0 to disable. + MatchersCacheMaxItems int `yaml:"matchers_cache_max_items"` } // RegisterFlags adds the flags required to config this to the given FlagSet @@ -173,6 +177,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) { f.BoolVar(&cfg.LabelsStringInterningEnabled, "ingester.labels-string-interning-enabled", false, "Experimental: Enable string interning for metrics labels.") f.BoolVar(&cfg.DisableChunkTrimming, "ingester.disable-chunk-trimming", false, "Disable trimming of matching series chunks based on query Start and End time. When disabled, the result may contain samples outside the queried time range but select performances may be improved. Note that certain query results might change by changing this option.") + f.IntVar(&cfg.MatchersCacheMaxItems, "ingester.matchers-cache-max-items", 0, "Maximum number of entries in the matchers cache. 0 to disable.") } func (cfg *Config) Validate() error { @@ -243,6 +248,7 @@ type Ingester struct { inflightQueryRequests atomic.Int64 maxInflightQueryRequests util_math.MaxTracker + matchersCache storecache.MatchersCache expandedPostingsCacheFactory *cortex_tsdb.ExpandedPostingsCacheFactory } @@ -708,7 +714,18 @@ func New(cfg Config, limits *validation.Overrides, registerer prometheus.Registe logger: logger, ingestionRate: util_math.NewEWMARate(0.2, instanceIngestionRateTickInterval), expandedPostingsCacheFactory: cortex_tsdb.NewExpandedPostingsCacheFactory(cfg.BlocksStorageConfig.TSDB.PostingsCache), + matchersCache: storecache.NewNoopMatcherCache(), + } + + if cfg.MatchersCacheMaxItems > 0 { + r := prometheus.NewRegistry() + registerer.MustRegister(newMatchCacheMetrics(r, logger)) + i.matchersCache, err = storecache.NewMatchersCache(storecache.WithSize(cfg.MatchersCacheMaxItems), storecache.WithPromRegistry(r)) + if err != nil { + return nil, err + } } + i.metrics = newIngesterMetrics(registerer, false, cfg.ActiveSeriesMetricsEnabled, @@ -1474,7 +1491,7 @@ func (i *Ingester) QueryExemplars(ctx context.Context, req *client.ExemplarQuery return nil, err } - from, through, matchers, err := client.FromExemplarQueryRequest(req) + from, through, matchers, err := client.FromExemplarQueryRequest(i.matchersCache, req) if err != nil { return nil, err } @@ -1564,7 +1581,7 @@ func (i *Ingester) labelsValuesCommon(ctx context.Context, req *client.LabelValu return nil, cleanup, err } - labelName, startTimestampMs, endTimestampMs, limit, matchers, err := client.FromLabelValuesRequest(req) + labelName, startTimestampMs, endTimestampMs, limit, matchers, err := client.FromLabelValuesRequest(i.matchersCache, req) if err != nil { return nil, cleanup, err } @@ -1654,7 +1671,7 @@ func (i *Ingester) labelNamesCommon(ctx context.Context, req *client.LabelNamesR return nil, cleanup, err } - startTimestampMs, endTimestampMs, limit, matchers, err := client.FromLabelNamesRequest(req) + startTimestampMs, endTimestampMs, limit, matchers, err := client.FromLabelNamesRequest(i.matchersCache, req) if err != nil { return nil, cleanup, err } @@ -1768,7 +1785,7 @@ func (i *Ingester) metricsForLabelMatchersCommon(ctx context.Context, req *clien } // Parse the request - _, _, limit, matchersSet, err := client.FromMetricsForLabelMatchersRequest(req) + _, _, limit, matchersSet, err := client.FromMetricsForLabelMatchersRequest(i.matchersCache, req) if err != nil { return cleanup, err } @@ -1982,7 +1999,7 @@ func (i *Ingester) QueryStream(req *client.QueryRequest, stream client.Ingester_ return err } - from, through, matchers, err := client.FromQueryRequest(req) + from, through, matchers, err := client.FromQueryRequest(i.matchersCache, req) if err != nil { return err } diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go index b4852d4076..44c7742a9b 100644 --- a/pkg/ingester/ingester_test.go +++ b/pkg/ingester/ingester_test.go @@ -119,6 +119,64 @@ func seriesSetFromResponseStream(s *mockQueryStreamServer) (storage.SeriesSet, e return set, nil } +func TestMatcherCache(t *testing.T) { + limits := defaultLimitsTestConfig() + userID := "1" + tenantLimits := newMockTenantLimits(map[string]*validation.Limits{userID: &limits}) + registry := prometheus.NewRegistry() + + dir := t.TempDir() + chunksDir := filepath.Join(dir, "chunks") + blocksDir := filepath.Join(dir, "blocks") + require.NoError(t, os.Mkdir(chunksDir, os.ModePerm)) + require.NoError(t, os.Mkdir(blocksDir, os.ModePerm)) + cfg := defaultIngesterTestConfig(t) + cfg.MatchersCacheMaxItems = 50 + ing, err := prepareIngesterWithBlocksStorageAndLimits(t, cfg, limits, tenantLimits, blocksDir, registry, true) + require.NoError(t, err) + require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing)) + + defer services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck + + // Wait until it's ACTIVE + test.Poll(t, time.Second, ring.ACTIVE, func() interface{} { + return ing.lifecycler.GetState() + }) + ctx := user.InjectOrgID(context.Background(), userID) + // Lets have 1 key evicted + numberOfDifferentMatchers := cfg.MatchersCacheMaxItems + 1 + callPerMatcher := 10 + for j := 0; j < numberOfDifferentMatchers; j++ { + for i := 0; i < callPerMatcher; i++ { + s := &mockQueryStreamServer{ctx: ctx} + err = ing.QueryStream(&client.QueryRequest{ + StartTimestampMs: math.MinInt64, + EndTimestampMs: math.MaxInt64, + Matchers: []*client.LabelMatcher{{Type: client.EQUAL, Name: labels.MetricName, Value: fmt.Sprintf("%d", j)}}, + }, s) + require.NoError(t, err) + } + } + + require.NoError(t, testutil.GatherAndCompare(registry, bytes.NewBufferString(fmt.Sprintf(` + # HELP ingester_matchers_cache_evicted_total Total number of items evicted from the cache + # TYPE ingester_matchers_cache_evicted_total counter + ingester_matchers_cache_evicted_total 1 + # HELP ingester_matchers_cache_hits_total Total number of cache hits for series matchers + # TYPE ingester_matchers_cache_hits_total counter + ingester_matchers_cache_hits_total %v + # HELP ingester_matchers_cache_items Total number of cached items + # TYPE ingester_matchers_cache_items gauge + ingester_matchers_cache_items %v + # HELP ingester_matchers_cache_max_items Maximum number of items that can be cached + # TYPE ingester_matchers_cache_max_items gauge + ingester_matchers_cache_max_items 0 + # HELP ingester_matchers_cache_requests_total Total number of cache requests for series matchers + # TYPE ingester_matchers_cache_requests_total counter + ingester_matchers_cache_requests_total %v + `, callPerMatcher*numberOfDifferentMatchers-numberOfDifferentMatchers, cfg.MatchersCacheMaxItems, callPerMatcher*numberOfDifferentMatchers)), "ingester_matchers_cache_requests_total", "ingester_matchers_cache_hits_total", "ingester_matchers_cache_items", "ingester_matchers_cache_max_items", "ingester_matchers_cache_evicted_total")) +} + func TestIngesterPerLabelsetLimitExceeded(t *testing.T) { limits := defaultLimitsTestConfig() userID := "1" diff --git a/pkg/ingester/lifecycle_test.go b/pkg/ingester/lifecycle_test.go index efa739b426..4fab7d716e 100644 --- a/pkg/ingester/lifecycle_test.go +++ b/pkg/ingester/lifecycle_test.go @@ -43,6 +43,7 @@ func defaultIngesterTestConfig(t testing.TB) Config { cfg.LifecyclerConfig.FinalSleep = 0 cfg.ActiveSeriesMetricsEnabled = true cfg.LabelsStringInterningEnabled = true + cfg.MatchersCacheMaxItems = 1024 return cfg } diff --git a/pkg/ingester/metrics.go b/pkg/ingester/metrics.go index 6cfe49dc1a..fdbe5cd4fa 100644 --- a/pkg/ingester/metrics.go +++ b/pkg/ingester/metrics.go @@ -1,6 +1,8 @@ package ingester import ( + "github.com/go-kit/log" + "github.com/go-kit/log/level" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" @@ -676,3 +678,71 @@ func (sm *tsdbMetrics) setRegistryForUser(userID string, registry *prometheus.Re func (sm *tsdbMetrics) removeRegistryForUser(userID string) { sm.regs.RemoveUserRegistry(userID, false) } + +type matcherCacheMetrics struct { + r *prometheus.Registry + logger log.Logger + + requestsTotal *prometheus.Desc + hitsTotal *prometheus.Desc + numItems *prometheus.Desc + maxItems *prometheus.Desc + evicted *prometheus.Desc +} + +func newMatchCacheMetrics(r *prometheus.Registry, l log.Logger) *matcherCacheMetrics { + m := &matcherCacheMetrics{ + r: r, + logger: l, + requestsTotal: prometheus.NewDesc( + "ingester_matchers_cache_requests_total", + "Total number of cache requests for series matchers", + nil, nil), + hitsTotal: prometheus.NewDesc( + "ingester_matchers_cache_hits_total", + "Total number of cache hits for series matchers", + nil, nil), + numItems: prometheus.NewDesc( + "ingester_matchers_cache_items", + "Total number of cached items", + nil, nil), + maxItems: prometheus.NewDesc( + "ingester_matchers_cache_max_items", + "Maximum number of items that can be cached", + nil, nil), + evicted: prometheus.NewDesc( + "ingester_matchers_cache_evicted_total", + "Total number of items evicted from the cache", + nil, nil), + } + return m +} + +func (m *matcherCacheMetrics) Describe(out chan<- *prometheus.Desc) { + out <- m.requestsTotal + out <- m.hitsTotal + out <- m.numItems + out <- m.maxItems + out <- m.evicted +} + +func (m *matcherCacheMetrics) Collect(out chan<- prometheus.Metric) { + gm, err := m.r.Gather() + if err != nil { + level.Warn(m.logger).Log("msg", "failed to gather metrics from registry", "err", err) + return + } + + mfm, err := util.NewMetricFamilyMap(gm) + + if err != nil { + level.Warn(m.logger).Log("msg", "failed to create metric family map", "err", err) + return + } + + out <- prometheus.MustNewConstMetric(m.requestsTotal, prometheus.CounterValue, mfm.SumCounters("thanos_matchers_cache_requests_total")) + out <- prometheus.MustNewConstMetric(m.hitsTotal, prometheus.CounterValue, mfm.SumCounters("thanos_matchers_cache_hits_total")) + out <- prometheus.MustNewConstMetric(m.numItems, prometheus.GaugeValue, mfm.SumGauges("thanos_matchers_cache_items")) + out <- prometheus.MustNewConstMetric(m.maxItems, prometheus.GaugeValue, mfm.SumGauges("thanos_matchers_cache_max_items")) + out <- prometheus.MustNewConstMetric(m.evicted, prometheus.CounterValue, mfm.SumCounters("thanos_matchers_cache_evicted_total")) +} diff --git a/pkg/querier/remote_read.go b/pkg/querier/remote_read.go index a7e86b96e5..a40c0b2e3a 100644 --- a/pkg/querier/remote_read.go +++ b/pkg/querier/remote_read.go @@ -6,6 +6,7 @@ import ( "github.com/go-kit/log" "github.com/go-kit/log/level" "github.com/prometheus/prometheus/storage" + storecache "github.com/thanos-io/thanos/pkg/store/cache" "github.com/cortexproject/cortex/pkg/ingester/client" "github.com/cortexproject/cortex/pkg/util" @@ -34,7 +35,7 @@ func RemoteReadHandler(q storage.Queryable, logger log.Logger) http.Handler { errors := make(chan error) for i, qr := range req.Queries { go func(i int, qr *client.QueryRequest) { - from, to, matchers, err := client.FromQueryRequest(qr) + from, to, matchers, err := client.FromQueryRequest(storecache.NewNoopMatcherCache(), qr) if err != nil { errors <- err return