Merge branch 'main' of ../private

proxy/integration-tests/features/remedy_strategy_based_throttling.feature

@@ -12,6 +12,20 @@ Feature: Strategy Based Throttling Remedy
 
         Then Responses have 200, 200, 429, 200 status codes in order
 
+    Scenario: Requests which exceed the limit with spillover
+        Given API Provider is up
+        And   Lunar Proxy is up
+        When  policies.yaml file is updated
+        And   policies.yaml includes a strategy_based_throttling remedy for GET httpbinmock /anything requests with 2 requests per 1 seconds spillover is enabled
+        And   policies.yaml file is saved
+        And   apply_policies command is run without waiting for Fluent to reload
+        And   wait 1 seconds
+        And   1 request is sent to httpbinmock /anything through Lunar Proxy
+        And   wait 1 seconds
+        And   3 requests are sent to httpbinmock /anything through Lunar Proxy
+        And   1 request is sent to httpbinmock /anything through Lunar Proxy
+        Then Responses have 200, 200, 200, 200, 429 status codes in order
+
     Scenario: Requests which exceed the limit per endpoint defined by the remedy receive a rate limit error response
         Given API Provider is up
         And   Lunar Proxy is up

proxy/integration-tests/features/steps/remedy_strategy_based_throttling.py

@@ -74,6 +74,31 @@ def parse_default_behaviour_name(behavior_text: str) -> DefaultBehavior:
 register_type(DefaultBehavior=parse_default_behaviour)
 
 
+def add_strategy_based_throttling_to_policies(
+    policies_requests: PoliciesRequests,
+    allowed_requests: int,
+    time_window: int,
+    method: str,
+    host: str,
+    path: str,
+    spillover_enabled: bool = False,
+    group_by: str = None,
+    quota_allocations: list[QuotaAllocation] = None,
+    default_behavior: DefaultBehavior = None,
+):
+    remedy = _build_remedy(
+        allowed_requests=allowed_requests,
+        time_window=time_window,
+        spillover_enabled=spillover_enabled,
+        group_by=group_by,
+        quota_allocations=quota_allocations,
+        default=default_behavior,
+    )
+    policies_requests.endpoints.append(
+        EndpointPolicy(method, f"{host}{path}", remedies=[remedy])
+    )
+
+
 @when(
     "policies.yaml includes a strategy_based_throttling remedy for {method} {host} {path} requests with {allowed_requests:Int} requests per {time_window:Int} seconds"
 )
@@ -86,10 +111,37 @@ async def step_impl(
     allowed_requests: int,
     time_window: int,
 ):
-    policies_requests: PoliciesRequests = context.policies_requests
-    remedy = _build_remedy(allowed_requests, time_window)
-    policies_requests.endpoints.append(
-        EndpointPolicy(method, f"{host}{path}", remedies=[remedy])
+    add_strategy_based_throttling_to_policies(
+        policies_requests=context.policies_requests,
+        allowed_requests=allowed_requests,
+        time_window=time_window,
+        method=method,
+        host=host,
+        path=path,
+    )
+
+
+@when(
+    "policies.yaml includes a strategy_based_throttling remedy for {method} {host} {path} requests with {allowed_requests:Int} requests per {time_window:Int} seconds spillover is {spillover_enabled:Enabled}"
+)
+@async_run_until_complete
+async def step_impl(
+    context: Any,
+    method: str,
+    host: str,
+    path: str,
+    allowed_requests: int,
+    time_window: int,
+    spillover_enabled: bool,
+):
+    add_strategy_based_throttling_to_policies(
+        policies_requests=context.policies_requests,
+        allowed_requests=allowed_requests,
+        time_window=time_window,
+        spillover_enabled=spillover_enabled,
+        method=method,
+        host=host,
+        path=path,
     )
 
 
@@ -105,10 +157,14 @@ async def step_impl(
     time_window: int,
     group_by: str,
 ):
-    policies_requests: PoliciesRequests = context.policies_requests
-    remedy = _build_remedy(allowed_requests, time_window, group_by)
-    policies_requests.endpoints.append(
-        EndpointPolicy("GET", f"{host}{path}", remedies=[remedy])
+    add_strategy_based_throttling_to_policies(
+        context.policies_requests,
+        allowed_requests=allowed_requests,
+        time_window=time_window,
+        group_by=group_by,
+        method="GET",
+        host=host,
+        path=path,
     )
 
 
@@ -125,10 +181,15 @@ async def step_impl(
     group_by: str,
     quota_allocations: list[QuotaAllocation],
 ):
-    policies_requests: PoliciesRequests = context.policies_requests
-    remedy = _build_remedy(allowed_requests, time_window, group_by, quota_allocations)
-    policies_requests.endpoints.append(
-        EndpointPolicy("GET", f"{host}{path}", remedies=[remedy])
+    add_strategy_based_throttling_to_policies(
+        context.policies_requests,
+        allowed_requests=allowed_requests,
+        time_window=time_window,
+        group_by=group_by,
+        quota_allocations=quota_allocations,
+        method="GET",
+        host=host,
+        path=path,
     )
 
 
@@ -146,12 +207,16 @@ async def step_impl(
     quota_allocations: list[QuotaAllocation],
     default_behaviour: DefaultBehavior,
 ):
-    policies_requests: PoliciesRequests = context.policies_requests
-    remedy = _build_remedy(
-        allowed_requests, time_window, group_by, quota_allocations, default_behaviour
-    )
-    policies_requests.endpoints.append(
-        EndpointPolicy("GET", f"{host}{path}", remedies=[remedy])
+    add_strategy_based_throttling_to_policies(
+        context.policies_requests,
+        allowed_requests=allowed_requests,
+        time_window=time_window,
+        group_by=group_by,
+        quota_allocations=quota_allocations,
+        default_behaviour=default_behaviour,
+        method="GET",
+        host=host,
+        path=path,
     )
 
 
@@ -288,6 +353,7 @@ def _build_remedy(
     default: DefaultBehavior = None,
     default_allocation_percentage: float = None,
     remedy_name: str = "test",
+    spillover_enabled: bool = False,
 ):
     remedy = {
         "name": f"{remedy_name} {uuid.uuid4()}",
@@ -312,4 +378,9 @@ def _build_remedy(
             ],
         }
 
+    remedy["config"]["strategy_based_throttling"]["spillover_config"] = {
+        "enabled": spillover_enabled,
+        "renew_on_day": 0,
+    }
+
     return remedy

proxy/src/services/lunar-engine/services/remedies/strategy_based_throttling_plugin_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"lunar/engine/actions"
 	"lunar/engine/config"
+	"lunar/engine/messages"
 	"lunar/engine/services/remedies"
 	"lunar/engine/utils"
 	"lunar/engine/utils/limit"
@@ -20,20 +21,68 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestWhenOnRequestIsCalledMoreThanAllowedRequestsRateLimitErrorIsReturned(
+func TestWhenOnRequestIsCalledMoreThanAllowedRequestsRateLimitWithSpillover(
 	t *testing.T,
 ) {
 	t.Parallel()
-	allowedRequests := 2
+	allowedRequests := 3
 	windowSizeInSeconds := 1
+	clock, plugin, onRequestArgs, scopedRemedy := setTest(
+		allowedRequests, windowSizeInSeconds, true)
+
+	assertNoOpAction(allowedRequests, plugin, onRequestArgs, scopedRemedy, t)
+
+	action, err := plugin.OnRequest(onRequestArgs, scopedRemedy)
+
+	assert.Nil(t, err)
+	wantAction := getEarlyResponseAction()
+	assert.Equal(t, &wantAction, action)
+
+	clock.AdvanceTime(time.Duration(windowSizeInSeconds) * time.Second)
+
+	action, err = plugin.OnRequest(onRequestArgs, scopedRemedy)
+
+	assert.Nil(t, err)
+	assert.Equal(t, &actions.NoOpAction{}, action)
+
+	clock.AdvanceTime(time.Duration(windowSizeInSeconds) * time.Second)
+	assertNoOpAction(5, plugin, onRequestArgs, scopedRemedy, t)
+}
+
+func assertNoOpAction(
+	runNTimes int, plugin *remedies.StrategyBasedThrottlingPlugin,
+	onRequestArgs messages.OnRequest, scopedRemedy config.ScopedRemedy,
+	t *testing.T,
+) {
+	for i := 0; i < runNTimes; i++ {
+		action, err := plugin.OnRequest(onRequestArgs, scopedRemedy)
+
+		assert.Nil(t, err)
+		assert.Equal(t, &actions.NoOpAction{}, action)
+	}
+}
+
+func getEarlyResponseAction() actions.EarlyResponseAction {
+	return actions.EarlyResponseAction{
+		Status:  429,
+		Headers: map[string]string{"Content-Type": "text/plain"},
+		Body:    "Too many requests",
+	}
+}
+
+func setTest(
+	allowedRequests int, windowSizeInSeconds int, spilloverEnabled bool) (
+	*clock.MockClock, *remedies.StrategyBasedThrottlingPlugin, messages.OnRequest,
+	config.ScopedRemedy,
+) {
 	clock := clock.NewMockClock()
 	obfuscator := obfuscation.Obfuscator{Hasher: obfuscation.MD5Hasher{}}
 	ctx := context.Background()
 	rateLimitState := limit.NewRateLimitState(clock, logging.ContextLogger{})
 	plugin, _ := remedies.NewStrategyBasedThrottlingPlugin(
 		ctx, clock, nil, rateLimitState, obfuscator)
 	remedyConfig := strategyBasedThrottlingRemedyConfig(
-		allowedRequests, windowSizeInSeconds, nil, false)
+		allowedRequests, windowSizeInSeconds, nil, spilloverEnabled)
 	onRequestArgs := onRequestArgs()
 
 	scopedRemedy := config.ScopedRemedy{
@@ -47,22 +96,23 @@ func TestWhenOnRequestIsCalledMoreThanAllowedRequestsRateLimitErrorIsReturned(
 			},
 		},
 	}
+	return clock, plugin, onRequestArgs, scopedRemedy
+}
 
-	for i := 0; i < allowedRequests; i++ {
-		action, err := plugin.OnRequest(onRequestArgs, scopedRemedy)
-
-		assert.Nil(t, err)
-		assert.Equal(t, &actions.NoOpAction{}, action)
-	}
+func TestWhenOnRequestIsCalledMoreThanAllowedRequestsRateLimitErrorIsReturned(
+	t *testing.T,
+) {
+	t.Parallel()
+	allowedRequests := 2
+	windowSizeInSeconds := 1
+	clock, plugin, onRequestArgs, scopedRemedy := setTest(
+		allowedRequests, windowSizeInSeconds, false)
 
+	assertNoOpAction(allowedRequests, plugin, onRequestArgs, scopedRemedy, t)
 	action, err := plugin.OnRequest(onRequestArgs, scopedRemedy)
 
 	assert.Nil(t, err)
-	wantAction := actions.EarlyResponseAction{
-		Status:  429,
-		Headers: map[string]string{"Content-Type": "text/plain"},
-		Body:    "Too many requests",
-	}
+	wantAction := getEarlyResponseAction()
 	assert.Equal(t, &wantAction, action)
 
 	clock.AdvanceTime(1 * time.Second)

proxy/src/services/lunar-engine/utils/limit/rate_limit_state_by_limiter_test.go

@@ -12,10 +12,43 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestHappyFlowForRateLimitState(
+func TestHappyFlowForRateLimitStateWithSpillover(
 	t *testing.T,
 ) {
 	t.Parallel()
+	testArgs := createTestArgs()
+
+	for _, requestArgs := range testArgs {
+		mockClock := clock.NewMockClock()
+		windowSize := time.Duration(1) * time.Second
+		state := limit.NewRateLimitState(
+			mockClock,
+			logging.ContextLogger{},
+		).(*limit.RateLimitState)
+
+		windowsData := limit.WindowData{
+			WindowSize:           windowSize,
+			AllowedRequestCount:  3,
+			QuotaAllocationRatio: 1,
+			SpilloverEnabled:     true,
+			SpilloverRenewOnDay:  0,
+		}
+
+		counter := incrementNTimes(t, 2, state, requestArgs, windowsData)
+		assert.Equal(t, int64(2), counter)
+
+		mockClock.AdvanceTime(time.Duration(1) * time.Second)
+
+		counter = incrementNTimes(t, 4, state, requestArgs, windowsData)
+		assert.Equal(t, int64(4), counter)
+
+		limitState, err := state.TryToIncrement(requestArgs, windowsData)
+		assert.Nil(t, err)
+		assert.Equal(t, limit.Block, limitState.LimitSate)
+	}
+}
+
+func createTestArgs() []limit.RequestArguments {
 	testArgs := []limit.RequestArguments{
 		{
 			LimiterID: "Global",
@@ -36,7 +69,14 @@ func TestHappyFlowForRateLimitState(
 			GroupID:   "group-id",
 		},
 	}
+	return testArgs
+}
 
+func TestHappyFlowForRateLimitState(
+	t *testing.T,
+) {
+	t.Parallel()
+	testArgs := createTestArgs()
 	for _, requestArgs := range testArgs {
 		mockClock := clock.NewMockClock()
 		windowSize := time.Duration(1) * time.Second