chore: add env to enable throttlerV2 (#4313)

rudderlabs · Jan 15, 2024 · 139d46e · 139d46e
1 parent d7cbc8c
commit 139d46e
Show file tree

Hide file tree

Showing 2 changed files with 37 additions and 15 deletions.
diff --git a/router/throttler/factory.go b/router/throttler/factory.go
@@ -63,20 +63,23 @@ func (f *factory) Get(destName, destID string) Throttler {
 		config:  conf,
 	}
 
-	var adaptiveConf adaptiveThrottleConfig
-	adaptiveConf.readThrottlingConfig(f.config, destName, destID)
-	var limitFactorMeasurement stats.Measurement = nil
-	if f.Stats != nil {
-		limitFactorMeasurement = f.Stats.NewTaggedStat("adaptive_throttler_limit_factor", stats.GaugeType, stats.Tags{
-			"destinationId": destID,
-			"destType":      destName,
-		})
-	}
-	at := &adaptiveThrottler{
-		limiter:                f.adaptiveLimiter,
-		algorithm:              newAdaptiveAlgorithm(f.config, adaptiveConf.window),
-		config:                 adaptiveConf,
-		limitFactorMeasurement: limitFactorMeasurement,
+	var at Throttler = &noOpThrottler{}
+	if f.config.GetBool("Router.throttlerV2.enabled", true) {
+		var adaptiveConf adaptiveThrottleConfig
+		adaptiveConf.readThrottlingConfig(f.config, destName, destID)
+		var limitFactorMeasurement stats.Measurement = nil
+		if f.Stats != nil {
+			limitFactorMeasurement = f.Stats.NewTaggedStat("adaptive_throttler_limit_factor", stats.GaugeType, stats.Tags{
+				"destinationId": destID,
+				"destType":      destName,
+			})
+		}
+		at = &adaptiveThrottler{
+			limiter:                f.adaptiveLimiter,
+			algorithm:              newAdaptiveAlgorithm(f.config, adaptiveConf.window),
+			config:                 adaptiveConf,
+			limitFactorMeasurement: limitFactorMeasurement,
+		}
 	}
 
 	f.throttlers[destID] = &switchingThrottler{

diff --git a/router/throttler/factory_test.go b/router/throttler/factory_test.go
@@ -11,8 +11,8 @@ import (
 )
 
 func TestFactory(t *testing.T) {
-	config := config.New()
 	t.Run("when adaptive throttling is enabled", func(t *testing.T) {
+		config := config.New()
 		config.Set("Router.throttler.adaptive.enabled", true)
 		maxLimit := int64(300)
 		config.Set("Router.throttler.adaptive.maxLimit", maxLimit)
@@ -52,6 +52,25 @@ func TestFactory(t *testing.T) {
 			}, 2*time.Second, 100*time.Millisecond, "limit: %d, expectedLimit: %d", ta.getLimit(), maxLimit*5/10)
 		})
 	})
+
+	t.Run("when throttlerV2 is false", func(t *testing.T) {
+		config := config.New()
+		config.Set("Router.throttlerV2.enabled", false)
+		config.Set("Router.throttler.destName.timeWindow", time.Second)
+		config.Set("Router.throttler.destName.limit", int64(100))
+		f, err := NewFactory(config, nil)
+		require.NoError(t, err)
+		defer f.Shutdown()
+		ta := f.Get("destName", "destID")
+		require.Eventually(t, func() bool {
+			return ta.getLimit() == int64(100)
+		}, 2*time.Second, 100*time.Millisecond)
+
+		config.Set("Router.throttler.adaptive.enabled", true)
+		require.Eventually(t, func() bool {
+			return ta.getLimit() == int64(0)
+		}, 2*time.Second, 100*time.Millisecond)
+	})
 }
 
 func floatCheck(a, b int64) bool {