Skip to content

Return rate limit error according to ingestion rate strategy #3485

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ie-pham merged 6 commits into from
Apr 9, 2024
Merged

Return rate limit error according to ingestion rate strategy #3485

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
aef0c0f
print limit according to ingestion rate strategy
ie-pham Mar 13, 2024
a80b652
clean
ie-pham Mar 13, 2024
d2a3090
overrides always return default overrides for ingestion rate strategy
ie-pham Mar 14, 2024
5e37852
remove test logs
ie-pham Mar 14, 2024
1a08ca3
babo
ie-pham Mar 14, 2024
0fc1492
add both local and global limits
ie-pham Mar 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion modules/distributor/distributor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -280,10 +280,14 @@ func (d *Distributor) checkForRateLimits(tracesSize, spanCount int, userID strin
now := time.Now()
if !d.ingestionRateLimiter.AllowN(now, userID, tracesSize) {
overrides.RecordDiscardedSpans(spanCount, reasonRateLimited, userID)
limit := int(d.ingestionRateLimiter.Limit(now, userID))
if d.overrides.IngestionRateStrategy() == overrides.GlobalIngestionRateStrategy {
limit = limit * d.DistributorRing.InstancesCount()
}
return status.Errorf(codes.ResourceExhausted,
"%s: ingestion rate limit (%d bytes) exceeded while adding %d bytes for user %s",
overrides.ErrorPrefixRateLimited,
int(d.ingestionRateLimiter.Limit(now, userID)),
limit,
tracesSize, userID)
}

Expand Down
3 changes: 3 additions & 0 deletions modules/overrides/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,9 @@ const (
// GlobalIngestionRateStrategy indicates that an attempt should be made to consider this limit across the entire Tempo cluster
GlobalIngestionRateStrategy = "global"

// IngestionRateUser since ingestion rate strategy is not per tenant config
IngestionRateUserVariable = "ingestion rate strategy user variable"

// ErrorPrefixLiveTracesExceeded is used to flag batches from the ingester that were rejected b/c they had too many traces
ErrorPrefixLiveTracesExceeded = "LIVE_TRACES_EXCEEDED"
// ErrorPrefixTraceTooLarge is used to flag batches from the ingester that were rejected b/c they exceeded the single trace limit
Expand Down
6 changes: 3 additions & 3 deletions modules/overrides/runtime_config_overrides.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -275,8 +275,8 @@ func (o *runtimeConfigOverridesManager) GetRuntimeOverridesFor(userID string) *O
// to each distributor instance (local) or evenly shared across the cluster (global).
func (o *runtimeConfigOverridesManager) IngestionRateStrategy() string {
// The ingestion rate strategy can't be overridden on a per-tenant basis,
// so here we just pick the value for a not-existing user ID (empty string).
return o.getOverridesForUser("").Ingestion.RateStrategy
// so here we just pass this specific variable to fetch the defaults value.
return o.getOverridesForUser(IngestionRateUserVariable).Ingestion.RateStrategy
Comment thread
yvrhdn marked this conversation as resolved.
Outdated
}

// MaxLocalTracesPerUser returns the maximum number of traces a user is allowed to store
Expand Down Expand Up @@ -484,7 +484,7 @@ func (o *runtimeConfigOverridesManager) DedicatedColumns(userID string) backend.
}

func (o *runtimeConfigOverridesManager) getOverridesForUser(userID string) *Overrides {
if tenantOverrides := o.tenantOverrides(); tenantOverrides != nil {
if tenantOverrides := o.tenantOverrides(); tenantOverrides != nil && userID != IngestionRateUserVariable {
l := tenantOverrides.forUser(userID)
if l != nil {
return l
Expand Down