Skip to content

Commit 0ae15ec

Browse files
committed
Creating Limits per LabelSet
1 parent fe2047e commit 0ae15ec

File tree

6 files changed

+517
-33
lines changed

6 files changed

+517
-33
lines changed

pkg/ingester/ingester.go

Lines changed: 26 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -254,11 +254,12 @@ func (r tsdbCloseCheckResult) shouldClose() bool {
254254
}
255255

256256
type userTSDB struct {
257-
db *tsdb.DB
258-
userID string
259-
activeSeries *ActiveSeries
260-
seriesInMetric *metricCounter
261-
limiter *Limiter
257+
db *tsdb.DB
258+
userID string
259+
activeSeries *ActiveSeries
260+
seriesInMetric *metricCounter
261+
labelSetCounter *labelSetCounter
262+
limiter *Limiter
262263

263264
instanceSeriesCount *atomic.Int64 // Shared across all userTSDB instances created by ingester.
264265
instanceLimitsFn func() *InstanceLimits
@@ -399,6 +400,10 @@ func (u *userTSDB) PreCreation(metric labels.Labels) error {
399400
return err
400401
}
401402

403+
if err := u.labelSetCounter.canAddSeriesForLabelSet(context.TODO(), u, metric); err != nil {
404+
return err
405+
}
406+
402407
return nil
403408
}
404409

@@ -412,6 +417,7 @@ func (u *userTSDB) PostCreation(metric labels.Labels) {
412417
return
413418
}
414419
u.seriesInMetric.increaseSeriesForMetric(metricName)
420+
u.labelSetCounter.increaseSeriesLabelSet(u, metric)
415421
}
416422

417423
// PostDeletion implements SeriesLifecycleCallback interface.
@@ -425,6 +431,7 @@ func (u *userTSDB) PostDeletion(metrics map[chunks.HeadSeriesRef]labels.Labels)
425431
continue
426432
}
427433
u.seriesInMetric.decreaseSeriesForMetric(metricName)
434+
u.labelSetCounter.decreaseSeriesLabelSet(u, metric)
428435
}
429436
}
430437

@@ -924,6 +931,7 @@ func (i *Ingester) updateActiveSeries() {
924931

925932
userDB.activeSeries.Purge(purgeTime)
926933
i.metrics.activeSeriesPerUser.WithLabelValues(userID).Set(float64(userDB.activeSeries.Active()))
934+
userDB.labelSetCounter.UpdateMetric(userDB, i.metrics.activeSeriesPerLabelSet)
927935
}
928936
}
929937

@@ -1100,38 +1108,43 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
11001108
// of it, so that we can return it back to the distributor, which will return a
11011109
// 400 error to the client. The client (Prometheus) will not retry on 400, and
11021110
// we actually ingested all samples which haven't failed.
1103-
switch cause := errors.Cause(err); cause {
1104-
case storage.ErrOutOfBounds:
1111+
switch cause := errors.Cause(err); {
1112+
case errors.Is(cause, storage.ErrOutOfBounds):
11051113
sampleOutOfBoundsCount++
11061114
updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(s.TimestampMs), ts.Labels) })
11071115
continue
11081116

1109-
case storage.ErrOutOfOrderSample:
1117+
case errors.Is(cause, storage.ErrOutOfOrderSample):
11101118
sampleOutOfOrderCount++
11111119
updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(s.TimestampMs), ts.Labels) })
11121120
continue
11131121

1114-
case storage.ErrDuplicateSampleForTimestamp:
1122+
case errors.Is(cause, storage.ErrDuplicateSampleForTimestamp):
11151123
newValueForTimestampCount++
11161124
updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(s.TimestampMs), ts.Labels) })
11171125
continue
11181126

1119-
case storage.ErrTooOldSample:
1127+
case errors.Is(cause, storage.ErrTooOldSample):
11201128
sampleTooOldCount++
11211129
updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(s.TimestampMs), ts.Labels) })
11221130
continue
11231131

1124-
case errMaxSeriesPerUserLimitExceeded:
1132+
case errors.Is(cause, errMaxSeriesPerUserLimitExceeded):
11251133
perUserSeriesLimitCount++
11261134
updateFirstPartial(func() error { return makeLimitError(perUserSeriesLimit, i.limiter.FormatError(userID, cause)) })
11271135
continue
11281136

1129-
case errMaxSeriesPerMetricLimitExceeded:
1137+
case errors.Is(cause, errMaxSeriesPerMetricLimitExceeded):
11301138
perMetricSeriesLimitCount++
11311139
updateFirstPartial(func() error {
11321140
return makeMetricLimitError(perMetricSeriesLimit, copiedLabels, i.limiter.FormatError(userID, cause))
11331141
})
11341142
continue
1143+
case errors.As(cause, &errMaxSeriesPerLabelSetLimitExceeded{}):
1144+
updateFirstPartial(func() error {
1145+
return makeMetricLimitError(perLabelsetSeriesLimit, copiedLabels, i.limiter.FormatError(userID, cause))
1146+
})
1147+
continue
11351148
}
11361149

11371150
// The error looks an issue on our side, so we should rollback
@@ -2018,6 +2031,7 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
20182031
userID: userID,
20192032
activeSeries: NewActiveSeries(),
20202033
seriesInMetric: newMetricCounter(i.limiter, i.cfg.getIgnoreSeriesLimitForMetricNamesMap()),
2034+
labelSetCounter: newLabelSetCounter(i.limiter),
20212035
ingestedAPISamples: util_math.NewEWMARate(0.2, i.cfg.RateUpdatePeriod),
20222036
ingestedRuleSamples: util_math.NewEWMARate(0.2, i.cfg.RateUpdatePeriod),
20232037

0 commit comments

Comments
 (0)