Added option to ignore specific metrics when checking max-series-per-metric limit. (#4302)

* Added option to ignore specific metrics when checking max-series-per-metric limit.

Signed-off-by: Peter Štibraný <[email protected]>

Author: pstibrany

CHANGELOG.md

@@ -49,6 +49,7 @@
   * `cortex_ruler_write_requests_failed_total`
   * `cortex_ruler_queries_total`
   * `cortex_ruler_queries_failed_total`
+* [ENHANCEMENT] Ingester: Added option `-ingester.ignore-series-limit-for-metric-names` with comma-separated list of metric names that will be ignored in max series per metric limit. #4302
 * [BUGFIX] Purger: fix `Invalid null value in condition for column range` caused by `nil` value in range for WriteBatch query. #4128
 * [BUGFIX] Ingester: fixed infrequent panic caused by a race condition between TSDB mmap-ed head chunks truncation and queries. #4176
 * [BUGFIX] Alertmanager: fix Alertmanager status page if clustering via gossip is disabled or sharding is enabled. #4184

docs/configuration/config-file-reference.md

@@ -810,6 +810,13 @@ instance_limits:
   # tenants). Additional requests will be rejected. 0 = unlimited.
   # CLI flag: -ingester.instance-limits.max-inflight-push-requests
   [max_inflight_push_requests: <int> | default = 0]
+
+# Comma-separated list of metric names, for which
+# -ingester.max-series-per-metric and -ingester.max-global-series-per-metric
+# limits will be ignored. Does not affect max-series-per-user or
+# max-global-series-per-metric limits.
+# CLI flag: -ingester.ignore-series-limit-for-metric-names
+[ignore_series_limit_for_metric_names: <string> | default = ""]
 ```
 
 ### `querier_config`

pkg/ingester/ingester.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -98,6 +99,8 @@ type Config struct {
 	DefaultLimits    InstanceLimits         `yaml:"instance_limits"`
 	InstanceLimitsFn func() *InstanceLimits `yaml:"-"`
 
+	IgnoreSeriesLimitForMetricNames string `yaml:"ignore_series_limit_for_metric_names"`
+
 	// For testing, you can override the address and ID of this ingester.
 	ingesterClientFactory func(addr string, cfg client.Config) (client.HealthAndIngesterClient, error)
 }
@@ -131,6 +134,29 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.Int64Var(&cfg.DefaultLimits.MaxInMemoryTenants, "ingester.instance-limits.max-tenants", 0, "Max users that this ingester can hold. Requests from additional users will be rejected. This limit only works when using blocks engine. 0 = unlimited.")
 	f.Int64Var(&cfg.DefaultLimits.MaxInMemorySeries, "ingester.instance-limits.max-series", 0, "Max series that this ingester can hold (across all tenants). Requests to create additional series will be rejected. This limit only works when using blocks engine. 0 = unlimited.")
 	f.Int64Var(&cfg.DefaultLimits.MaxInflightPushRequests, "ingester.instance-limits.max-inflight-push-requests", 0, "Max inflight push requests that this ingester can handle (across all tenants). Additional requests will be rejected. 0 = unlimited.")
+
+	f.StringVar(&cfg.IgnoreSeriesLimitForMetricNames, "ingester.ignore-series-limit-for-metric-names", "", "Comma-separated list of metric names, for which -ingester.max-series-per-metric and -ingester.max-global-series-per-metric limits will be ignored. Does not affect max-series-per-user or max-global-series-per-metric limits.")
+}
+
+func (cfg *Config) getIgnoreSeriesLimitForMetricNamesMap() map[string]struct{} {
+	if cfg.IgnoreSeriesLimitForMetricNames == "" {
+		return nil
+	}
+
+	result := map[string]struct{}{}
+
+	for _, s := range strings.Split(cfg.IgnoreSeriesLimitForMetricNames, ",") {
+		tr := strings.TrimSpace(s)
+		if tr != "" {
+			result[tr] = struct{}{}
+		}
+	}
+
+	if len(result) == 0 {
+		return nil
+	}
+
+	return result
 }
 
 // Ingester deals with "in flight" chunks.  Based on Prometheus 1.x

pkg/ingester/ingester_test.go

@@ -1044,3 +1044,15 @@ func TestIngesterActiveSeries(t *testing.T) {
 		})
 	}
 }
+
+func TestGetIgnoreSeriesLimitForMetricNamesMap(t *testing.T) {
+	cfg := Config{}
+
+	require.Nil(t, cfg.getIgnoreSeriesLimitForMetricNamesMap())
+
+	cfg.IgnoreSeriesLimitForMetricNames = ", ,,,"
+	require.Nil(t, cfg.getIgnoreSeriesLimitForMetricNamesMap())
+
+	cfg.IgnoreSeriesLimitForMetricNames = "foo, bar, ,"
+	require.Equal(t, map[string]struct{}{"foo": {}, "bar": {}}, cfg.getIgnoreSeriesLimitForMetricNamesMap())
+}

pkg/ingester/ingester_v2.go

@@ -1558,7 +1558,7 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	userDB := &userTSDB{
 		userID:              userID,
 		activeSeries:        NewActiveSeries(),
-		seriesInMetric:      newMetricCounter(i.limiter),
+		seriesInMetric:      newMetricCounter(i.limiter, i.cfg.getIgnoreSeriesLimitForMetricNamesMap()),
 		ingestedAPISamples:  util_math.NewEWMARate(0.2, i.cfg.RateUpdatePeriod),
 		ingestedRuleSamples: util_math.NewEWMARate(0.2, i.cfg.RateUpdatePeriod),
 

pkg/ingester/user_state.go

@@ -145,7 +145,7 @@ func (us *userStates) getOrCreate(userID string) *userState {
 			index:               index.New(),
 			ingestedAPISamples:  util_math.NewEWMARate(0.2, us.cfg.RateUpdatePeriod),
 			ingestedRuleSamples: util_math.NewEWMARate(0.2, us.cfg.RateUpdatePeriod),
-			seriesInMetric:      newMetricCounter(us.limiter),
+			seriesInMetric:      newMetricCounter(us.limiter, us.cfg.getIgnoreSeriesLimitForMetricNamesMap()),
 			logger:              logger,
 
 			memSeries:             us.metrics.memSeries,
@@ -362,9 +362,11 @@ type metricCounterShard struct {
 type metricCounter struct {
 	limiter *Limiter
 	shards  []metricCounterShard
+
+	ignoredMetrics map[string]struct{}
 }
 
-func newMetricCounter(limiter *Limiter) *metricCounter {
+func newMetricCounter(limiter *Limiter, ignoredMetricsForSeriesCount map[string]struct{}) *metricCounter {
 	shards := make([]metricCounterShard, 0, numMetricCounterShards)
 	for i := 0; i < numMetricCounterShards; i++ {
 		shards = append(shards, metricCounterShard{
@@ -374,6 +376,8 @@ func newMetricCounter(limiter *Limiter) *metricCounter {
 	return &metricCounter{
 		limiter: limiter,
 		shards:  shards,
+
+		ignoredMetrics: ignoredMetricsForSeriesCount,
 	}
 }
 
@@ -394,6 +398,10 @@ func (m *metricCounter) getShard(metricName string) *metricCounterShard {
 }
 
 func (m *metricCounter) canAddSeriesFor(userID, metric string) error {
+	if _, ok := m.ignoredMetrics[metric]; ok {
+		return nil
+	}
+
 	shard := m.getShard(metric)
 	shard.mtx.Lock()
 	defer shard.mtx.Unlock()

pkg/ingester/user_state_test.go

@@ -15,6 +15,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/weaveworks/common/user"
+
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
 // Test forSeriesMatching correctly batches up series.
@@ -153,3 +156,86 @@ func TestTeardown(t *testing.T) {
 	cortex_ingester_active_series{user="3"} 0
 	`), metricNames...))
 }
+
+func TestMetricCounter(t *testing.T) {
+	const metric = "metric"
+
+	for name, tc := range map[string]struct {
+		ignored                   []string
+		localLimit                int
+		series                    int
+		expectedErrorOnLastSeries error
+	}{
+		"no ignored metrics, limit not reached": {
+			ignored:                   nil,
+			localLimit:                5,
+			series:                    5,
+			expectedErrorOnLastSeries: nil,
+		},
+
+		"no ignored metrics, limit reached": {
+			ignored:                   nil,
+			localLimit:                5,
+			series:                    6,
+			expectedErrorOnLastSeries: errMaxSeriesPerMetricLimitExceeded,
+		},
+
+		"ignored metric, limit not reached": {
+			ignored:                   []string{metric},
+			localLimit:                5,
+			series:                    5,
+			expectedErrorOnLastSeries: nil,
+		},
+
+		"ignored metric, over limit": {
+			ignored:                   []string{metric},
+			localLimit:                5,
+			series:                    10,
+			expectedErrorOnLastSeries: nil, // this metric is not checked for series-count.
+		},
+
+		"ignored different metric, limit not reached": {
+			ignored:                   []string{"another_metric1", "another_metric2"},
+			localLimit:                5,
+			series:                    5,
+			expectedErrorOnLastSeries: nil,
+		},
+
+		"ignored different metric, over limit": {
+			ignored:                   []string{"another_metric1", "another_metric2"},
+			localLimit:                5,
+			series:                    6,
+			expectedErrorOnLastSeries: errMaxSeriesPerMetricLimitExceeded,
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			var ignored map[string]struct{}
+			if tc.ignored != nil {
+				ignored = make(map[string]struct{})
+				for _, v := range tc.ignored {
+					ignored[v] = struct{}{}
+				}
+			}
+
+			limits := validation.Limits{MaxLocalSeriesPerMetric: tc.localLimit}
+
+			overrides, err := validation.NewOverrides(limits, nil)
+			require.NoError(t, err)
+
+			// We're testing code that's not dependant on sharding strategy, replication factor, etc. To simplify the test,
+			// we use local limit only.
+			limiter := NewLimiter(overrides, nil, util.ShardingStrategyDefault, true, 3, false)
+			mc := newMetricCounter(limiter, ignored)
+
+			for i := 0; i < tc.series; i++ {
+				err := mc.canAddSeriesFor("user", metric)
+				if i < tc.series-1 {
+					assert.NoError(t, err)
+					mc.increaseSeriesForMetric(metric)
+				} else {
+					assert.Equal(t, tc.expectedErrorOnLastSeries, err)
+				}
+			}
+		})
+	}
+}