Limit series per metric (cortexproject#276)

* Limit series per metric

Fixes cortexproject#47

* Fix error wrapping for series-per-metric exceeded

* Move num-metrics-per-series handling to userState

* Review feeback

Author: juliusv

distributor/http_server.go

@@ -24,15 +24,20 @@ func (d *Distributor) PushHandler(w http.ResponseWriter, r *http.Request) {
 
 	_, err := d.Push(ctx, &req)
 	if err != nil {
-		if grpc.Code(err) == codes.ResourceExhausted && grpc.ErrorDesc(err) == util.ErrUserSeriesLimitExceeded.Error() {
-			err = util.ErrUserSeriesLimitExceeded
+		if grpc.Code(err) == codes.ResourceExhausted {
+			switch grpc.ErrorDesc(err) {
+			case util.ErrUserSeriesLimitExceeded.Error():
+				err = util.ErrUserSeriesLimitExceeded
+			case util.ErrMetricSeriesLimitExceeded.Error():
+				err = util.ErrMetricSeriesLimitExceeded
+			}
 		}
 
 		var code int
 		switch err {
 		case errIngestionRateLimitExceeded:
 			code = http.StatusTooManyRequests
-		case util.ErrUserSeriesLimitExceeded:
+		case util.ErrUserSeriesLimitExceeded, util.ErrMetricSeriesLimitExceeded:
 			code = http.StatusInsufficientStorage
 		default:
 			code = http.StatusInternalServerError

ingester/ingester.go

@@ -38,6 +38,8 @@ const (
 	DefaultConcurrentFlush = 50
 	// DefaultMaxSeriesPerUser is the maximum number of series allowed per user.
 	DefaultMaxSeriesPerUser = 5000000
+	// DefaultMaxSeriesPerMetric is the maximum number of series in one metric (of a single user).
+	DefaultMaxSeriesPerMetric = 50000
 
 	minReadyDuration = 1 * time.Minute
 )
@@ -118,6 +120,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.StringVar(&cfg.ChunkEncoding, "ingester.chunk-encoding", "1", "Encoding version to use for chunks.")
 	f.DurationVar(&cfg.UserStatesConfig.RateUpdatePeriod, "ingester.rate-update-period", 15*time.Second, "Period with which to update the per-user ingestion rates.")
 	f.IntVar(&cfg.UserStatesConfig.MaxSeriesPerUser, "ingester.max-series-per-user", DefaultMaxSeriesPerUser, "Maximum number of active series per user.")
+	f.IntVar(&cfg.UserStatesConfig.MaxSeriesPerMetric, "ingester.max-series-per-metric", DefaultMaxSeriesPerMetric, "Maximum number of active series per metric name.")
 }
 
 type flushOp struct {
@@ -149,11 +152,14 @@ func New(cfg Config, chunkStore cortex_chunk.Store, ring *ring.Ring) (*Ingester,
 	if cfg.ChunkEncoding == "" {
 		cfg.ChunkEncoding = "1"
 	}
+	if cfg.UserStatesConfig.RateUpdatePeriod == 0 {
+		cfg.UserStatesConfig.RateUpdatePeriod = 15 * time.Second
+	}
 	if cfg.UserStatesConfig.MaxSeriesPerUser <= 0 {
 		cfg.UserStatesConfig.MaxSeriesPerUser = DefaultMaxSeriesPerUser
 	}
-	if cfg.UserStatesConfig.RateUpdatePeriod == 0 {
-		cfg.UserStatesConfig.RateUpdatePeriod = 15 * time.Second
+	if cfg.UserStatesConfig.MaxSeriesPerMetric <= 0 {
+		cfg.UserStatesConfig.MaxSeriesPerMetric = DefaultMaxSeriesPerMetric
 	}
 
 	if err := chunk.DefaultEncoding.Set(cfg.ChunkEncoding); err != nil {
@@ -249,7 +255,7 @@ func (i *Ingester) Push(ctx context.Context, req *remote.WriteRequest) (*cortex.
 	var lastPartialErr error
 	for _, sample := range util.FromWriteRequest(req) {
 		if err := i.append(ctx, sample); err != nil {
-			if err == util.ErrUserSeriesLimitExceeded {
+			if err == util.ErrUserSeriesLimitExceeded || err == util.ErrMetricSeriesLimitExceeded {
 				lastPartialErr = grpc.Errorf(codes.ResourceExhausted, err.Error())
 				continue
 			}
@@ -594,8 +600,7 @@ func (i *Ingester) flushUserSeries(userID string, fp model.Fingerprint, immediat
 	series.chunkDescs = series.chunkDescs[len(chunks):]
 	i.memoryChunks.Sub(float64(len(chunks)))
 	if len(series.chunkDescs) == 0 {
-		userState.fpToSeries.del(fp)
-		userState.index.delete(series.metric, fp)
+		userState.removeSeries(fp, series.metric)
 	}
 	userState.fpLocker.Unlock(fp)
 	return nil

ingester/ingester_test.go

@@ -149,7 +149,7 @@ func TestIngesterAppend(t *testing.T) {
 	}
 }
 
-func TestIngesterSeriesLimitExceeded(t *testing.T) {
+func TestIngesterUserSeriesLimitExceeded(t *testing.T) {
 	cfg := Config{
 		FlushCheckPeriod: 99999 * time.Hour,
 		MaxChunkIdle:     99999 * time.Hour,
@@ -192,7 +192,93 @@ func TestIngesterSeriesLimitExceeded(t *testing.T) {
 	// Append to two series, expect series-exceeded error.
 	_, err = ing.Push(ctx, util.ToWriteRequest([]*model.Sample{sample2, sample3}))
 	if grpc.ErrorDesc(err) != util.ErrUserSeriesLimitExceeded.Error() {
-		t.Fatalf("expected series-exceeded error, got %v", err)
+		t.Fatalf("expected error about exceeding metrics per user, got %v", err)
+	}
+
+	// Read samples back via ingester queries.
+	matcher, err := metric.NewLabelMatcher(metric.Equal, model.MetricNameLabel, "testmetric")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	req, err := util.ToQueryRequest(model.Earliest, model.Latest, []*metric.LabelMatcher{matcher})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	resp, err := ing.Query(ctx, req)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	res := util.FromQueryResponse(resp)
+	sort.Sort(res)
+
+	expected := model.Matrix{
+		{
+			Metric: sample1.Metric,
+			Values: []model.SamplePair{
+				{
+					Timestamp: sample1.Timestamp,
+					Value:     sample1.Value,
+				},
+				{
+					Timestamp: sample2.Timestamp,
+					Value:     sample2.Value,
+				},
+			},
+		},
+	}
+
+	if !reflect.DeepEqual(res, expected) {
+		t.Fatalf("unexpected query result\n\nwant:\n\n%v\n\ngot:\n\n%v\n\n", expected, res)
+	}
+}
+
+func TestIngesterMetricSeriesLimitExceeded(t *testing.T) {
+	cfg := Config{
+		FlushCheckPeriod: 99999 * time.Hour,
+		MaxChunkIdle:     99999 * time.Hour,
+		UserStatesConfig: UserStatesConfig{
+			MaxSeriesPerMetric: 1,
+		},
+	}
+	store := &testStore{
+		chunks: map[string][]chunk.Chunk{},
+	}
+	ing, err := New(cfg, store, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	userID := "1"
+	sample1 := &model.Sample{
+		Metric:    model.Metric{model.MetricNameLabel: "testmetric", "foo": "bar"},
+		Timestamp: 0,
+		Value:     1,
+	}
+	sample2 := &model.Sample{
+		Metric:    model.Metric{model.MetricNameLabel: "testmetric", "foo": "bar"},
+		Timestamp: 1,
+		Value:     2,
+	}
+	sample3 := &model.Sample{
+		Metric:    model.Metric{model.MetricNameLabel: "testmetric", "foo": "biz"},
+		Timestamp: 1,
+		Value:     3,
+	}
+
+	// Append only one series first, expect no error.
+	ctx := user.WithID(context.Background(), userID)
+	_, err = ing.Push(ctx, util.ToWriteRequest([]*model.Sample{sample1}))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Append to two series, expect series-exceeded error.
+	_, err = ing.Push(ctx, util.ToWriteRequest([]*model.Sample{sample2, sample3}))
+	if grpc.ErrorDesc(err) != util.ErrMetricSeriesLimitExceeded.Error() {
+		t.Fatalf("expected error about exceeding series per metric, got %v", err)
 	}
 
 	// Read samples back via ingester queries.

ingester/user_state.go

@@ -26,12 +26,16 @@ type userState struct {
 	mapper          *fpMapper
 	index           *invertedIndex
 	ingestedSamples *ewmaRate
+
+	seriesInMetricMtx sync.Mutex
+	seriesInMetric    map[model.LabelValue]int
 }
 
 // UserStatesConfig configures userStates properties.
 type UserStatesConfig struct {
-	RateUpdatePeriod time.Duration
-	MaxSeriesPerUser int
+	RateUpdatePeriod   time.Duration
+	MaxSeriesPerUser   int
+	MaxSeriesPerMetric int
 }
 
 func newUserStates(cfg *UserStatesConfig) *userStates {
@@ -130,7 +134,7 @@ func (us *userStates) getOrCreateSeries(ctx context.Context, metric model.Metric
 	us.mtx.RLock()
 	state, ok = us.states[userID]
 	if ok {
-		fp, series, err = state.unlockedGet(metric, us.cfg.MaxSeriesPerUser)
+		fp, series, err = state.unlockedGet(metric, us.cfg)
 		if err != nil {
 			us.mtx.RUnlock()
 			return nil, fp, nil, err
@@ -144,7 +148,7 @@ func (us *userStates) getOrCreateSeries(ctx context.Context, metric model.Metric
 	us.mtx.Lock()
 	defer us.mtx.Unlock()
 	state = us.unlockedGetOrCreate(userID)
-	fp, series, err = state.unlockedGet(metric, us.cfg.MaxSeriesPerUser)
+	fp, series, err = state.unlockedGet(metric, us.cfg)
 	return state, fp, series, err
 }
 
@@ -157,14 +161,15 @@ func (us *userStates) unlockedGetOrCreate(userID string) *userState {
 			fpLocker:        newFingerprintLocker(16),
 			index:           newInvertedIndex(),
 			ingestedSamples: newEWMARate(0.2, us.cfg.RateUpdatePeriod),
+			seriesInMetric:  map[model.LabelValue]int{},
 		}
 		state.mapper = newFPMapper(state.fpToSeries)
 		us.states[userID] = state
 	}
 	return state
 }
 
-func (u *userState) unlockedGet(metric model.Metric, maxSeries int) (model.Fingerprint, *memorySeries, error) {
+func (u *userState) unlockedGet(metric model.Metric, cfg *UserStatesConfig) (model.Fingerprint, *memorySeries, error) {
 	rawFP := metric.FastFingerprint()
 	u.fpLocker.Lock(rawFP)
 	fp := u.mapper.mapFP(rawFP, metric)
@@ -183,17 +188,59 @@ func (u *userState) unlockedGet(metric model.Metric, maxSeries int) (model.Finge
 	// all proceed to add a new series. This is likely not worth addressing,
 	// as this should happen rarely (all samples from one push are added
 	// serially), and the overshoot in allowed series would be minimal.
-	if u.fpToSeries.length() >= maxSeries {
+	if u.fpToSeries.length() >= cfg.MaxSeriesPerUser {
 		u.fpLocker.Unlock(fp)
 		return fp, nil, util.ErrUserSeriesLimitExceeded
 	}
 
+	metricName, err := util.ExtractMetricNameFromMetric(metric)
+	if err != nil {
+		u.fpLocker.Unlock(fp)
+		return fp, nil, err
+	}
+
+	if !u.canAddSeriesFor(metricName, cfg) {
+		u.fpLocker.Unlock(fp)
+		return fp, nil, util.ErrMetricSeriesLimitExceeded
+	}
+
 	series = newMemorySeries(metric)
 	u.fpToSeries.put(fp, series)
 	u.index.add(metric, fp)
 	return fp, series, nil
 }
 
+func (u *userState) canAddSeriesFor(metric model.LabelValue, cfg *UserStatesConfig) bool {
+	u.seriesInMetricMtx.Lock()
+	defer u.seriesInMetricMtx.Unlock()
+
+	if u.seriesInMetric[metric] >= cfg.MaxSeriesPerMetric {
+		return false
+	}
+	u.seriesInMetric[metric]++
+	return true
+}
+
+func (u *userState) removeSeries(fp model.Fingerprint, metric model.Metric) {
+	u.fpToSeries.del(fp)
+	u.index.delete(metric, fp)
+
+	metricName, err := util.ExtractMetricNameFromMetric(metric)
+	if err != nil {
+		// Series without a metric name should never be able to make it into
+		// the ingester's memory storage.
+		panic(err)
+	}
+
+	u.seriesInMetricMtx.Lock()
+	defer u.seriesInMetricMtx.Unlock()
+
+	u.seriesInMetric[metricName]--
+	if u.seriesInMetric[metricName] == 0 {
+		delete(u.seriesInMetric, metricName)
+	}
+}
+
 // forSeriesMatching passes all series matching the given matchers to the provided callback.
 // Deals with locking and the quirks of zero-length matcher values.
 func (u *userState) forSeriesMatching(allMatchers []*metric.LabelMatcher, callback func(model.Fingerprint, *memorySeries) error) error {

util/error.go

@@ -7,8 +7,9 @@ func (e Error) Error() string { return string(e) }
 
 // Errors returned by Cortex components.
 const (
-	ErrMissingMetricName       = Error("sample missing metric name")
-	ErrInvalidMetricName       = Error("sample invalid metric name")
-	ErrInvalidLabel            = Error("sample invalid label")
-	ErrUserSeriesLimitExceeded = Error("per-user series limit exceeded")
+	ErrMissingMetricName         = Error("sample missing metric name")
+	ErrInvalidMetricName         = Error("sample invalid metric name")
+	ErrInvalidLabel              = Error("sample invalid label")
+	ErrUserSeriesLimitExceeded   = Error("per-user series limit exceeded")
+	ErrMetricSeriesLimitExceeded = Error("per-metric series limit exceeded")
 )