Allow dynamic limiter to scale ingest rate below static limit (#851)

Author: lahsivjar

processor/ratelimitprocessor/README.md

@@ -313,7 +313,7 @@ Telemetry and metrics:
 
 ### Throttling rate based on custom logic using `window_configurator`
 
-The `window_configurator` option configures a custom OpenTelemetry Collector extension to dynamically choose a window multiplier for the current period. The value is the extension ID (the extension's configured name). The extension MUST implement the `WindowConfigurator` interface. The multiplier can be used to scale up the rate limit from previous window (by returning a multiplier greater than `1`) or scale down the rate limit from previous window (by returning a multiplier less than `1`, greater than `0`). If the extension returns a negative value then the `default_window_multiplier` will be used. Note that the dynamic limit MUST be at least the configured `static_rate`, ensuring a minimum level of throughput. An example configuration including the window configurator:
+The `window_configurator` option configures a custom OpenTelemetry Collector extension to dynamically choose a window multiplier for the current period. The value is the extension ID (the extension's configured name). The extension MUST implement the `WindowConfigurator` interface. The multiplier can be used to scale up the rate limit from the previous window (by returning a multiplier greater than `1`) or scale down the rate limit from the previous window (by returning a multiplier less than `1`, greater than `0`). The lowest possible value of the rate is `1/s`. If the extension returns a negative value, then the `default_window_multiplier` will be used. An example configuration including the window configurator:
 
 ```yaml
 processors:

processor/ratelimitprocessor/config.go

@@ -83,9 +83,8 @@ type DynamicRateLimiting struct {
 	// WindowConfigurator is the component ID of the extension to dynamically
 	// determine the window multiplier. The extension is expected to implement
 	// the `WindowConfigurator` interface. The window configurator is used in
-	// the hot path so it should respond fast. The effective rate cannot go
-	// below the configured static rate limit settings. If the configurator
-	// returns a negative multiplier then the default multiplier will be used.
+	// the hot path so it should respond fast.If the configurator returns a
+	// negative multiplier then the default multiplier will be used.
 	WindowConfigurator component.ID `mapstructure:"window_configurator"`
 }
 

processor/ratelimitprocessor/gubernator.go

@@ -407,11 +407,21 @@ func (r *gubernatorRateLimiter) getDynamicLimit(ctx context.Context,
 	}
 	// Only record the incoming hits when the current rate is within the allowed
 	// range, otherwise, do not record the hits and return the calculated rate.
-	// MaxAllowed sets a ceiling on the rate with the window duration.
+	// MaxAllowed sets a ceiling on the rate with the window duration. If the
+	// previous period had hits and the window multiplier is suggesting lowering
+	// the ingestion rate then the MaxAllowed will be allowed to go below the
+	// static rate (to as low as `1`). As soon as the window multiplier suggests
+	// increasing the ingestion rate, the MaxAllowed will jump to a minimum of
+	// static rate.
 	//
 	// NOTE(marclop) We may want to add a follow-up static ceiling to avoid
 	// unbounded growth.
-	maxAllowed := math.Max(staticRate, previous*windowMultiplier)
+	var maxAllowed float64
+	if previous > 0 && windowMultiplier <= 1 {
+		maxAllowed = max(1, previous*windowMultiplier)
+	} else {
+		maxAllowed = max(staticRate, previous*windowMultiplier)
+	}
 	// Normalise the current rate assuming no more events will occur during the
 	// rest of the window. This will ensure that we record hits based on the
 	// currently observed hits and NOT based on extrapolated data.
@@ -423,6 +433,19 @@ func (r *gubernatorRateLimiter) getDynamicLimit(ctx context.Context,
 			return -1, err
 		}
 	}
+	if r.logger.Level() == zap.DebugLevel {
+		r.logger.Debug(
+			"Dynamic rate limiting applied",
+			zap.Dict(
+				"ratelimit",
+				zap.String("unique_key", uniqueKey),
+				zap.Float64("multiplier", windowMultiplier),
+				zap.Float64("static_rate", staticRate),
+				zap.Float64("previous_rate", previous),
+				zap.Float64("limit", maxAllowed),
+			),
+		)
+	}
 	return maxAllowed, nil
 }
 

processor/ratelimitprocessor/gubernator_test.go

@@ -734,8 +734,8 @@ func TestGubernatorRateLimiter_WindowConfigurator(t *testing.T) {
 	// Set the windowConfigurator manually
 	rateLimiter.windowConfigurator = &fakeWindowConfigurator{
 		mapping: map[string][]float64{
-			"x-tenant-id:12345": {1.5},           // more ingest data
-			"x-tenant-id:67890": {-1, 2, 2, 0.5}, // throttle ingest data
+			"x-tenant-id:12345": {1.5},                 // more ingest data
+			"x-tenant-id:67890": {-1, 2, 2, 0.5, 0.01}, // throttle ingest data
 		},
 		count: make(map[string]int),
 	}
@@ -747,7 +747,8 @@ func TestGubernatorRateLimiter_WindowConfigurator(t *testing.T) {
 		})
 
 		waitUntilNextPeriod(windowDuration)
-		// 1st non-zero window: establish baseline with static limit
+		// 1st non-zero window: establish baseline with static limit. Since the previous
+		// hits at this point were `0` the limit will always be equal to static limit.
 		actual := int64(staticRatePerSec) * int64(windowDuration) / int64(time.Second)
 		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
 		drainEvents(eventChannel)
@@ -756,11 +757,12 @@ func TestGubernatorRateLimiter_WindowConfigurator(t *testing.T) {
 		// 2nd window: ramp up by 1.5 factor as configured for the tenant
 		actual = int64(100) * int64(windowDuration) / int64(time.Second)
 		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
+		expectedLimit := int64(float64(staticRatePerSec) * 1.5) // 150% of previous hits/s
 		verify := func(evc <-chan gubernator.HitEvent) {
 			t.Helper()
 			event := lastReqLimitEvent(drainEvents(evc), t)
 			assertRequestRateLimitEvent(t, "x-tenant-id:12345", event,
-				actual, 750, 750-actual, gubernator.Status_UNDER_LIMIT,
+				actual, expectedLimit, expectedLimit-actual, gubernator.Status_UNDER_LIMIT,
 			)
 		}
 		verify(eventChannel)
@@ -775,49 +777,73 @@ func TestGubernatorRateLimiter_WindowConfigurator(t *testing.T) {
 		uniqueKey := "x-tenant-id:67890"
 
 		waitUntilNextPeriod(windowDuration)
-		// 1st non-zero window: establish baseline with static limit
+		// 1st non-zero window: establish baseline with static limit. Since the previous
+		// hits at this point were `0` the limit will always be equal to static limit.
 		actual := int64(staticRatePerSec) * int64(windowDuration) / int64(time.Second)
 		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
 		drainEvents(eventChannel)
 
 		waitUntilNextPeriod(windowDuration)
 		// 2nd window: scale up by 2 factor as configured for the tenant
-		actual = int64(1000) * int64(windowDuration) / int64(time.Second)
+		expectedLimit := int64(float64(staticRatePerSec) * 2) // 200% of previous hits/s
+		// maximize the hits so that the rate is fully utilized
+		actual = expectedLimit * int64(windowDuration) / int64(time.Second)
 		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
 		assertRequestRateLimitEvent(
 			t, uniqueKey,
 			lastReqLimitEvent(drainEvents(eventChannel), t),
-			actual, 1000, 1000-actual, gubernator.Status_UNDER_LIMIT,
+			actual, expectedLimit, expectedLimit-actual, gubernator.Status_UNDER_LIMIT,
 		)
 
 		waitUntilNextPeriod(windowDuration)
 		// 3nd window: scale up by 2 factor as configured for the tenant
-		actual = int64(2000) * int64(windowDuration) / int64(time.Second)
+		expectedLimit = expectedLimit * 2 // 200% of previous hits/s
+		// maximize the hits so that the rate is fully utilized
+		actual = expectedLimit * int64(windowDuration) / int64(time.Second)
 		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
 		assertRequestRateLimitEvent(
 			t, uniqueKey,
 			lastReqLimitEvent(drainEvents(eventChannel), t),
-			actual, 2000, 2000-actual, gubernator.Status_UNDER_LIMIT,
+			actual, expectedLimit, expectedLimit-actual, gubernator.Status_UNDER_LIMIT,
 		)
 
 		waitUntilNextPeriod(windowDuration)
 		// 4th window: scale down by 50% as configured for the tenant
-		actual = int64(100) * int64(windowDuration) / int64(time.Second)
+		expectedLimit = int64(float64(expectedLimit) * 0.5) // 50% of previous hits/s
+		// maximize the hits so that the rate is fully utilized
+		actual = expectedLimit * int64(windowDuration) / int64(time.Second)
 		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
 		assertRequestRateLimitEvent(
 			t, uniqueKey,
 			lastReqLimitEvent(drainEvents(eventChannel), t),
-			actual, 1000, 1000-actual, gubernator.Status_UNDER_LIMIT,
+			actual, expectedLimit, expectedLimit-actual, gubernator.Status_UNDER_LIMIT,
 		)
 
-		// wait for a couple of periods to assert static rate as the lower bound
-		waitUntilNextPeriod(2 * windowDuration)
-		actual = int64(100) * int64(windowDuration) / int64(time.Second)
+		waitUntilNextPeriod(windowDuration)
+		// 5th window: scale down by 1% as configured for the tenant and assert
+		// that the rate goes below static limit
+		expectedLimit = int64(float64(expectedLimit) * 0.01) // 1% of previous hits/s
+		// maximize the hits so that the rate is fully utilized
+		actual = expectedLimit * int64(windowDuration) / int64(time.Second)
+		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
+		assertRequestRateLimitEvent(
+			t, uniqueKey,
+			lastReqLimitEvent(drainEvents(eventChannel), t),
+			actual, expectedLimit, expectedLimit-actual, gubernator.Status_UNDER_LIMIT,
+		)
+
+		// wait for a few periods to assert rate that the rate goes to be equal to the
+		// static rate as data is NOT sent in the first waiting window causing the
+		// previous rate to go to zero in the next window which we will assert.
+		waitUntilNextPeriod(windowDuration)
+		waitUntilNextPeriod(windowDuration)
+		expectedLimit = int64(staticRatePerSec)
+		actual = int64(100) * int64(windowDuration) / int64(time.Second) // send 100 hits
 		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
 		assertRequestRateLimitEvent(
 			t, uniqueKey,
 			lastReqLimitEvent(drainEvents(eventChannel), t),
-			actual, 500, 500-actual, gubernator.Status_UNDER_LIMIT,
+			actual, expectedLimit, expectedLimit-actual, gubernator.Status_UNDER_LIMIT,
 		)
 	})
 	t.Run("unknown tenant, use default multiplier", func(t *testing.T) {
@@ -828,12 +854,13 @@ func TestGubernatorRateLimiter_WindowConfigurator(t *testing.T) {
 		})
 
 		actual := int64(400)
+		expectedLimit := int64(staticRatePerSec)
 		require.NoError(t, rateLimiter.RateLimit(ctx, int(actual)))
 		verify := func(evc <-chan gubernator.HitEvent) {
 			t.Helper()
 			event := lastReqLimitEvent(drainEvents(evc), t)
 			assertRequestRateLimitEvent(t, "x-tenant-id:unknown", event,
-				actual, 500, 500-actual, gubernator.Status_UNDER_LIMIT,
+				actual, expectedLimit, expectedLimit-actual, gubernator.Status_UNDER_LIMIT,
 			)
 		}
 		verify(eventChannel)