[ingester/stream]: Add a byte stream rate limit. #4191

cstyan · 2021-08-18T19:30:32Z

Adds a limiter to the stream struct.

Signed-off-by: Callum Styan callumstyan@gmail.com

owen-d

The outline for this PR is looking good; I've left the first round of feedback.

owen-d · 2021-08-19T14:23:35Z

pkg/ingester/ingester.go

@@ -140,6 +140,7 @@ type Ingester struct {
 	cfg           Config
 	clientConfig  client.Config
 	tenantConfigs *runtime.TenantConfigs
+	limits        *validation.Overrides


The overrides are already embedded in the limiter field, which I think we should use for consistency.

owen-d · 2021-08-19T14:24:47Z

pkg/ingester/limiter.go

+}
+
+func (s *localStrategy) Limit(userID string) float64 {
+	return float64(s.limits.MaxLocalStreamRateMB(userID))


I think this should use bytes, not MB

owen-d · 2021-08-19T14:30:42Z

pkg/ingester/stream.go

+		// Check if this this should be rate limited.
+		now := time.Now()
+		if !s.limiter.AllowN(now, s.tenant, len(entries[i].Line)) {
+			validation.DiscardedSamples.WithLabelValues(validation.StreamRateLimit, s.tenant).Add(float64(len(entries) - i))


Please use accumulator counters for the discarded samples, similar to how out_of_orders are handled on
line 206-212. This is the hot path and it's nice to call this once per batch rather than once per sample.

owen-d · 2021-08-19T14:33:52Z

pkg/ingester/stream.go

+				rateLimitesBytes += len(entries[j].Line)
+			}
+			validation.DiscardedBytes.WithLabelValues(validation.StreamRateLimit, s.tenant).Add(float64(rateLimitesBytes))
+			return 0, httpgrpc.Errorf(http.StatusTooManyRequests, validation.StreamRateLimitErrorMsg, int(s.limiter.Limit(now, s.tenant)), len(entries)-i, rateLimitesBytes)


We shouldn't return early here for a few reasons:

We'd never run the tailer code.

We need to ultimately return the correct bytesAdded for partial batches.

We want the metrics to be correct for the whole batch, not just one sample.

Instead, try to incorporate these new failures into failedEntriesWithError.

owen-d · 2021-08-19T14:55:47Z

pkg/validation/limits.go

+	_ = l.MaxLocalStreamRateBytes.Set("10000")
+	f.Var(&l.MaxLocalStreamRateBytes, "ingester.max-stream-rate-bytes", "Maximum bytes per second rate per active stream.")
+	_ = l.MaxLocalStreamBurstRateBytes.Set("20000")
+	f.Var(&l.MaxLocalStreamBurstRateBytes, "ingester.max-stream-rate-bytes", "Maximum burst bytes per second rate per active stream.")


Please mimic how we set similar configs:
https://github.com/grafana/loki/blob/main/pkg/ingester/wal.go#L24-L50

Notably:

Use a readable const default ${VARIABLE} with appropriate comment if applicable

Default to a powers of 2 form.

I'd like us to default to the following

const defaultPerStreamRateLimit = 1<<20 // 1MB const defaultPerStreamBurstLimit = 2 * defaultPerStreamRateLimit

owen-d · 2021-08-19T14:57:20Z

pkg/validation/limits.go

+func (o *Overrides) MaxLocalStreamRateMB(userID string) int {
+	return o.getOverridesForUser(userID).MaxLocalStreamRateBytes.Val()
+}
+
+func (o *Overrides) MaxLocalStreamBurstRateMB(userID string) int {
+	return int(o.getOverridesForUser(userID).MaxLocalStreamBurstRateBytes)
+}
+


Why are these calculated in different ways (int(x) vs x.Val())? Let's consistently use .Val().

Signed-off-by: Callum Styan <callumstyan@gmail.com>

config value. Signed-off-by: Callum Styan <callumstyan@gmail.com>

Signed-off-by: Callum Styan <callumstyan@gmail.com>

owen-d · 2021-08-19T18:07:47Z

pkg/validation/validate.go

+	//StreamRateLimit is a reason for discarding lines when the streams own rate limit is hit
+	// rather than the overall ingestion rate limit.
+	StreamRateLimit         = "per_stream_rate_limt"
+	StreamRateLimitErrorMsg = "Per stream rate limit exceeded (limit: %d bytes/sec) while attempting to ingest '%d' lines totaling '%d' bytes, consider splitting a stream via additional labels or contact your Loki administrator to see if the limt can be increased"


Suggested change

StreamRateLimitErrorMsg = "Per stream rate limit exceeded (limit: %d bytes/sec) while attempting to ingest '%d' lines totaling '%d' bytes, consider splitting a stream via additional labels or contact your Loki administrator to see if the limt can be increased"

This was unused

owen-d · 2021-08-19T18:17:59Z

pkg/ingester/stream.go

 	return &stream{
+		limiter:         limiter.NewRateLimiter(newLocalStreamRateStrategy(limits), 10*time.Second),


I looked at the the Cortex rate limiter utility and realized it has a tenancy concept built into it. This has some nice benefits (automatically rechecking dynamically loaded overrides) as well as some costs (mutex). I suspect this is low cost enough that we don't have to worry about it (we'll have a map with one entry, gated by a RWMutex). However, if we end up seeing profile issues here in the future, we may have to roll our own or contribute upstream.

Easy enough to roll our own in the near future, just makes use of an actual rate limiter library under the hood. I can file an issue for this.

owen-d · 2021-08-19T18:20:16Z

pkg/ingester/stream.go

@@ -113,8 +117,9 @@ type entryWithError struct {
 	e     error
 }

-func newStream(cfg *Config, tenant string, fp model.Fingerprint, labels labels.Labels, unorderedWrites bool, metrics *ingesterMetrics) *stream {
+func newStream(cfg *Config, limits *validation.Overrides, tenant string, fp model.Fingerprint, labels labels.Labels, unorderedWrites bool, metrics *ingesterMetrics) *stream {


Suggested change

func newStream(cfg *Config, limits *validation.Overrides, tenant string, fp model.Fingerprint, labels labels.Labels, unorderedWrites bool, metrics *ingesterMetrics) *stream {

func newStream(cfg *Config, limits limiter.RateLimiterStrategy, tenant string, fp model.Fingerprint, labels labels.Labels, unorderedWrites bool, metrics *ingesterMetrics) *stream {

nit: Let's replace this with the strategy interface so we don't depend on the overrides itself here.

owen-d · 2021-08-19T18:21:45Z

pkg/ingester/instance_test.go

@@ -450,7 +450,7 @@ func Test_ChunkFilter(t *testing.T) {
 	overrides, err := validation.NewOverrides(defaultLimits, nil)
 	require.NoError(t, err)
 	instance := newInstance(
-		&ingesterConfig, "fake", NewLimiter(overrides, NilMetrics, &ringCountMock{count: 1}, 1), loki_runtime.DefaultTenantConfigs(), noopWAL{}, NilMetrics, nil, &testFilter{})
+		&ingesterConfig, &validation.Overrides{}, "fake", NewLimiter(overrides, NilMetrics, &ringCountMock{count: 1}, 1), loki_runtime.DefaultTenantConfigs(), noopWAL{}, NilMetrics, nil, &testFilter{})


The instance type already has access to the overrides via
https://github.com/grafana/loki/blob/main/pkg/ingester/instance.go#L83

Let's remove this from the newInstance function to reduce arguments and potential confusion from specifying the limits twice.

owen-d · 2021-08-19T18:22:37Z

pkg/ingester/instance.go

@@ -63,6 +63,7 @@ var (

 type instance struct {
 	cfg        *Config
+	limits     *validation.Overrides


This should already be covered here:
https://github.com/grafana/loki/blob/main/pkg/ingester/instance.go#L83

Signed-off-by: Callum Styan <callumstyan@gmail.com>

slim-bean · 2021-08-20T20:48:32Z

pkg/ingester/stream.go

@@ -49,7 +50,8 @@ var (
 )

 var (
-	ErrEntriesExist = errors.New("duplicate push - entries already exist")
+	ErrEntriesExist    = errors.New("duplicate push - entries already exist")
+	ErrStreamRateLimit = errors.New("stream rate limit exceeded")


Sorry I missed this before this was merged, but could you open a separate PR to move this error message to the validate.go file and also update it to include an action for the user to take (similar to other errors in that file).

We should always try to provide an actionable error message for anything returned by the API.

owen-d · 2021-08-30T14:02:38Z

ref #1544

pull-request-size bot added the size/L label Aug 18, 2021

owen-d reviewed Aug 19, 2021

View reviewed changes

cstyan added 3 commits August 19, 2021 10:24

Add a byte stream rate limit.

c318c5c

Signed-off-by: Callum Styan <callumstyan@gmail.com>

Track discarded lines due to stream rate limit + use ByteSize for the

69430df

config value. Signed-off-by: Callum Styan <callumstyan@gmail.com>

Address review comments

9d1849b

Signed-off-by: Callum Styan <callumstyan@gmail.com>

owen-d reviewed Aug 19, 2021

View reviewed changes

cstyan added 5 commits August 19, 2021 12:18

Another round of review fixes.

d5e64d4

Signed-off-by: Callum Styan <callumstyan@gmail.com>

Fix goimports issue.

c89bba2

Signed-off-by: Callum Styan <callumstyan@gmail.com>

Fix duplicate flag registration + stream rate limit test.

5fe344c

Signed-off-by: Callum Styan <callumstyan@gmail.com>

Fix stream rate/burst limit flags yaml naming.

a7aa7e4

Signed-off-by: Callum Styan <callumstyan@gmail.com>

Fix tests now that we have stream rate limit.

4651dee

Signed-off-by: Callum Styan <callumstyan@gmail.com>

cstyan marked this pull request as ready for review August 19, 2021 22:46

cstyan requested a review from a team as a code owner August 19, 2021 22:46

cstyan changed the title ~~WIP: Add a byte stream rate limit.~~ [ingester/stream]: Add a byte stream rate limit. Aug 19, 2021

fix some comments

31edbb8

Signed-off-by: Callum Styan <callumstyan@gmail.com>

owen-d approved these changes Aug 20, 2021

View reviewed changes

owen-d merged commit 757eeac into grafana:main Aug 20, 2021

slim-bean reviewed Aug 20, 2021

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[ingester/stream]: Add a byte stream rate limit. #4191

[ingester/stream]: Add a byte stream rate limit. #4191

cstyan commented Aug 18, 2021

owen-d left a comment

owen-d Aug 19, 2021

owen-d Aug 19, 2021

owen-d Aug 19, 2021

owen-d Aug 19, 2021

owen-d Aug 19, 2021

owen-d Aug 19, 2021

owen-d Aug 19, 2021

owen-d Aug 19, 2021

cstyan Aug 19, 2021

owen-d Aug 19, 2021

owen-d Aug 19, 2021

owen-d Aug 19, 2021

slim-bean Aug 20, 2021

owen-d commented Aug 30, 2021

		return &stream{
		limiter: limiter.NewRateLimiter(newLocalStreamRateStrategy(limits), 10*time.Second),

	func newStream(cfg Config, limits validation.Overrides, tenant string, fp model.Fingerprint, labels labels.Labels, unorderedWrites bool, metrics ingesterMetrics) stream {
	func newStream(cfg Config, limits limiter.RateLimiterStrategy, tenant string, fp model.Fingerprint, labels labels.Labels, unorderedWrites bool, metrics ingesterMetrics) *stream {

[ingester/stream]: Add a byte stream rate limit. #4191

[ingester/stream]: Add a byte stream rate limit. #4191

Conversation

cstyan commented Aug 18, 2021

owen-d left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

owen-d commented Aug 30, 2021