-
Notifications
You must be signed in to change notification settings - Fork 322
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
1204bb0
commit d8302b1
Showing
17 changed files
with
568 additions
and
369 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| package adaptivethrottlercounter | ||
|
|
||
| import ( | ||
| "context" | ||
| "net/http" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/rudderlabs/rudder-go-kit/config" | ||
| ) | ||
|
|
||
| type timer struct { | ||
| frequency *config.Reloadable[time.Duration] | ||
| limitReached bool | ||
| mu sync.Mutex | ||
| limitSet bool | ||
| cancel context.CancelFunc | ||
| } | ||
|
|
||
| type Adaptive struct { | ||
| shortTimer *timer | ||
| longTimer *timer | ||
| } | ||
|
|
||
| func New(config *config.Config) *Adaptive { | ||
| shortTimeFrequency := config.GetReloadableDurationVar(5, time.Second, "Router.throttler.adaptiveRateLimit.shortTimeFrequency") | ||
| longTimeFrequency := config.GetReloadableDurationVar(15, time.Second, "Router.throttler.adaptiveRateLimit.longTimeFrequency") | ||
|
|
||
| shortTimer := &timer{ | ||
| frequency: shortTimeFrequency, | ||
| } | ||
| longTimer := &timer{ | ||
| frequency: longTimeFrequency, | ||
| } | ||
|
|
||
| go shortTimer.run() | ||
| go longTimer.run() | ||
|
|
||
| return &Adaptive{ | ||
| shortTimer: shortTimer, | ||
| longTimer: longTimer, | ||
| } | ||
| } | ||
|
|
||
| func (a *Adaptive) LimitFactor() float64 { | ||
| decreaseLimitPercentage := config.GetInt64("Router.throttler.adaptiveRateLimit.decreaseLimitPercentage", 30) | ||
| increaseChangePercentage := config.GetInt64("Router.throttler.adaptiveRateLimit.increaseChangePercentage", 10) | ||
| if a.shortTimer.getLimitReached() && !a.shortTimer.limitSet { | ||
| a.shortTimer.limitSet = true | ||
| return float64(-decreaseLimitPercentage) / 100 | ||
| } else if !a.longTimer.getLimitReached() && !a.longTimer.limitSet { | ||
| a.longTimer.limitSet = true | ||
| return float64(increaseChangePercentage) / 100 | ||
| } | ||
| return 0.0 | ||
| } | ||
|
|
||
| func (a *Adaptive) ResponseCodeReceived(code int) { | ||
| if code == http.StatusTooManyRequests { | ||
| a.shortTimer.updateLimitReached() | ||
| a.longTimer.updateLimitReached() | ||
| } | ||
| } | ||
|
|
||
| func (a *Adaptive) ShutDown() { | ||
| a.shortTimer.cancel() | ||
| a.longTimer.cancel() | ||
| } | ||
|
|
||
| func (t *timer) run() { | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| t.cancel = cancel | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case <-time.After(t.frequency.Load()): | ||
| t.resetLimitReached() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (t *timer) updateLimitReached() { | ||
| t.mu.Lock() | ||
| defer t.mu.Unlock() | ||
| t.limitReached = true | ||
| } | ||
|
|
||
| func (t *timer) getLimitReached() bool { | ||
| t.mu.Lock() | ||
| defer t.mu.Unlock() | ||
| return t.limitReached | ||
| } | ||
|
|
||
| func (t *timer) resetLimitReached() { | ||
| t.mu.Lock() | ||
| defer t.mu.Unlock() | ||
| t.limitReached = false | ||
| t.limitSet = false | ||
| } |
30 changes: 30 additions & 0 deletions
30
router/throttler/adaptiveAlgorithmCounter/algorithm_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package adaptivethrottlercounter | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/rudderlabs/rudder-go-kit/config" | ||
| ) | ||
|
|
||
| func TestAdaptiveRateLimit(t *testing.T) { | ||
| config := config.New() // TODO: change to config.New() | ||
| config.Set("Router.throttler.adaptiveRateLimit.shortTimeFrequency", 1*time.Second) | ||
| config.Set("Router.throttler.adaptiveRateLimit.longTimeFrequency", 2*time.Second) | ||
| al := New(config) | ||
|
|
||
| t.Run("when there is a 429 in the last shortTimeFrequency", func(t *testing.T) { | ||
| al.ResponseCodeReceived(429) | ||
| require.Eventually(t, func() bool { | ||
| return al.LimitFactor() == float64(-0.3) // reduces by 30% since there is an error in the last 1 second | ||
| }, time.Second, 10*time.Millisecond) | ||
| }) | ||
|
|
||
| t.Run("when there are no 429 ins the last longTimeFrequency", func(t *testing.T) { | ||
| require.Eventually(t, func() bool { | ||
| return al.LimitFactor() == float64(+0.1) // increases by 10% since there is no error in the last 2 seconds | ||
| }, 2*time.Second, 10*time.Millisecond) | ||
| }) | ||
| } |
Oops, something went wrong.