mirror of
https://github.com/lukaszraczylo/graphql-monitoring-proxy.git
synced 2026-11-09 08:54:27 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3e187790f2 | ||
|
|
ea30694dc9 | ||
|
|
c128a0f4d9 |
@@ -821,7 +821,7 @@ Enable rate limiting based on user roles using the `ROLE_RATE_LIMIT` environment
|
|||||||
- `ROLE_RATE_LIMIT`: Enable role-based rate limiting (default: `false`)
|
- `ROLE_RATE_LIMIT`: Enable role-based rate limiting (default: `false`)
|
||||||
|
|
||||||
**Features:**
|
**Features:**
|
||||||
- **Burst control**: Optional burst limits for handling traffic spikes
|
- **Burst control**: Optional `burst` caps requests in a short window (one tenth of the interval). `req` counts requests per `interval`. A `burst` at or above `req` has no effect.
|
||||||
|
|
||||||
`ratelimit.json` is loaded once at startup; changing the file requires a restart.
|
`ratelimit.json` is loaded once at startup; changing the file requires a restart.
|
||||||
|
|
||||||
|
|||||||
+1
-1
@@ -118,7 +118,7 @@ func (bhm *BackendHealthManager) WaitForBackendReady(timeout time.Duration) erro
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Exponential backoff (1.5x growth, capped at maxDelay), no jitter
|
// Exponential backoff (1.5x growth, capped at maxDelay), no jitter
|
||||||
time.Sleep(currentDelay)
|
time.Sleep(min(currentDelay, time.Until(deadline)))
|
||||||
currentDelay = time.Duration(float64(currentDelay) * 1.5)
|
currentDelay = time.Duration(float64(currentDelay) * 1.5)
|
||||||
if currentDelay > maxDelay {
|
if currentDelay > maxDelay {
|
||||||
currentDelay = maxDelay
|
currentDelay = maxDelay
|
||||||
|
|||||||
@@ -0,0 +1,32 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
|
||||||
|
"github.com/valyala/fasthttp"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Pins theorems/Theorems/ReadyBackoff.lean: the startup wait never sleeps past its timeout.
|
||||||
|
func TestWaitForBackendReady_HonoursTimeout(t *testing.T) {
|
||||||
|
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
addr := ln.Addr().String()
|
||||||
|
_ = ln.Close() // closed port: every probe is refused immediately
|
||||||
|
|
||||||
|
bhm := NewBackendHealthManager(&fasthttp.Client{}, "http://"+addr, "", libpack_logger.New())
|
||||||
|
defer bhm.cancel()
|
||||||
|
|
||||||
|
const timeout = 3 * time.Second
|
||||||
|
start := time.Now()
|
||||||
|
if err := bhm.WaitForBackendReady(timeout); err == nil {
|
||||||
|
t.Fatal("want error for unreachable backend")
|
||||||
|
}
|
||||||
|
if elapsed := time.Since(start); elapsed > timeout+500*time.Millisecond {
|
||||||
|
t.Fatalf("waited %v, want <= %v", elapsed, timeout+500*time.Millisecond)
|
||||||
|
}
|
||||||
|
}
|
||||||
Vendored
+2
-1
@@ -158,7 +158,8 @@ func (c *Cache) Set(key string, value []byte, ttl time.Duration) {
|
|||||||
c.evictToFreeMemory(memoryToFree)
|
c.evictToFreeMemory(memoryToFree)
|
||||||
} else if atomic.LoadInt64(&c.entryCount) >= c.maxCacheSize {
|
} else if atomic.LoadInt64(&c.entryCount) >= c.maxCacheSize {
|
||||||
// Fall back to count-based eviction for backward compatibility
|
// Fall back to count-based eviction for backward compatibility
|
||||||
c.evictOldest(int(c.maxCacheSize / 10)) // Evict 10% of entries
|
// 10% of entries, at least one: max/10 is 0 for caps below 10.
|
||||||
|
c.evictOldest(int(max(1, c.maxCacheSize/10)))
|
||||||
}
|
}
|
||||||
|
|
||||||
// A non-positive TTL is clamped to defaultTTL so this backend expires
|
// A non-positive TTL is clamped to defaultTTL so this backend expires
|
||||||
|
|||||||
Vendored
+23
@@ -0,0 +1,23 @@
|
|||||||
|
package libpack_cache_memory
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Pins theorems/Theorems/MemCacheCap.lean: every cap >= 1 bounds the entry count.
|
||||||
|
func TestSet_SmallEntryCapIsEnforced(t *testing.T) {
|
||||||
|
for _, maxEntries := range []int64{1, 2, 5, 9, 10, 25} {
|
||||||
|
t.Run(fmt.Sprintf("cap_%d", maxEntries), func(t *testing.T) {
|
||||||
|
c := NewWithSize(time.Minute, DefaultMaxMemorySize, maxEntries)
|
||||||
|
defer c.Shutdown()
|
||||||
|
for i := 0; i < 100; i++ {
|
||||||
|
c.Set(fmt.Sprintf("k%d", i), []byte("v"), time.Minute)
|
||||||
|
assert.LessOrEqual(t, c.CountQueries(), maxEntries)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
+10
-26
@@ -304,35 +304,19 @@ func rateLimitedRequest(userID, userRole string) bool {
|
|||||||
|
|
||||||
// checkRateLimit performs the actual rate limit check.
|
// checkRateLimit performs the actual rate limit check.
|
||||||
//
|
//
|
||||||
// Two independent gates apply, both must pass:
|
// Two gates apply, both must pass:
|
||||||
//
|
//
|
||||||
// 1. Sustained: tickerRate (requests measured over the full Interval) must
|
// 1. Sustained: the request count over the full Interval must stay at or
|
||||||
// stay at or below Req. This is the only gate when Burst <= 0, and its
|
// below Req (Req is per Interval, not per second).
|
||||||
// comparison is unchanged from before the burst fix.
|
// 2. Burst (only when Burst > 0): the request count within a short trailing
|
||||||
// 2. Burst (only when roleConfig.burstCounter != nil, i.e. Burst > 0): the
|
// window (see burstWindowDuration) must stay at or below Burst.
|
||||||
// raw request count within a short trailing window (see
|
|
||||||
// burstWindowDuration), much shorter than Interval, must stay at or
|
|
||||||
// below Burst.
|
|
||||||
//
|
//
|
||||||
// Because the burst window is short relative to Interval, a client can
|
// The burst window lies inside the Interval, so its count never exceeds the
|
||||||
// legally land up to Burst requests in a tight cluster - more than a
|
// sustained count. Burst >= Req therefore never rejects anything the sustained
|
||||||
// steady Req-paced stream would deliver in that same short span - as long
|
// gate admits; Burst only takes effect when it is below Req.
|
||||||
// as the sustained gate still holds over the full Interval. That is what
|
|
||||||
// makes burst > req a real spike allowance instead of a no-op. Burst is a
|
|
||||||
// short-window ceiling layered ON TOP of the sustained Req rate, not a
|
|
||||||
// replacement for it. burst < req is an unusual config: the burst gate
|
|
||||||
// becomes the tighter, binding constraint whenever requests cluster in
|
|
||||||
// time, but the sustained cap for evenly-spread traffic is still Req
|
|
||||||
// alone. The pre-fix code compared the SAME sustained-window tickerRate
|
|
||||||
// against both Burst and Req, so when burst < req its Burst check tripped
|
|
||||||
// first and Burst was effectively the sustained cap too. This gate's
|
|
||||||
// Req-only sustained check can therefore allow MORE through than the
|
|
||||||
// pre-fix code did for evenly-spread traffic under a burst < req config.
|
|
||||||
// In both cases the burst gate never allows more than Burst requests
|
|
||||||
// within its own window (fail-safe).
|
|
||||||
func checkRateLimit(userID, userRole string, roleConfig RateLimitConfig, endpoint string) bool {
|
func checkRateLimit(userID, userRole string, roleConfig RateLimitConfig, endpoint string) bool {
|
||||||
roleConfig.RateCounterTicker.Incr(1)
|
roleConfig.RateCounterTicker.Incr(1)
|
||||||
tickerRate := roleConfig.RateCounterTicker.GetRate()
|
tickerRate := roleConfig.RateCounterTicker.Get()
|
||||||
|
|
||||||
logDetails := map[string]any{
|
logDetails := map[string]any{
|
||||||
"user_role": userRole,
|
"user_role": userRole,
|
||||||
@@ -348,7 +332,7 @@ func checkRateLimit(userID, userRole string, roleConfig RateLimitConfig, endpoin
|
|||||||
Pairs: map[string]any{"log_details": logDetails},
|
Pairs: map[string]any{"log_details": logDetails},
|
||||||
})
|
})
|
||||||
|
|
||||||
if tickerRate > float64(roleConfig.Req) {
|
if tickerRate > int64(roleConfig.Req) {
|
||||||
cfg.Logger.Debug(&libpack_logger.LogMessage{
|
cfg.Logger.Debug(&libpack_logger.LogMessage{
|
||||||
Message: "Rate limit exceeded",
|
Message: "Rate limit exceeded",
|
||||||
Pairs: map[string]any{"log_details": logDetails},
|
Pairs: map[string]any{"log_details": logDetails},
|
||||||
|
|||||||
+7
-40
@@ -372,59 +372,26 @@ func Test_checkRateLimit_BurstSemantics(t *testing.T) {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
t.Run("b_burst_above_req_allows_short_spike_sustained_still_capped", func(t *testing.T) {
|
t.Run("b_burst_above_req_is_noop_sustained_caps_at_req", func(t *testing.T) {
|
||||||
const (
|
const (
|
||||||
interval = 10 * time.Second
|
interval = 10 * time.Second
|
||||||
req = 3
|
req = 3
|
||||||
burst = 6 // burst > req
|
burst = 6 // burst >= req: dominated by the sustained gate
|
||||||
)
|
)
|
||||||
rc := newBurstTestRoleConfig(interval, req, burst)
|
rc := newBurstTestRoleConfig(interval, req, burst)
|
||||||
fc := newFakeClock(time.Now())
|
fc := newFakeClock(time.Now())
|
||||||
rc.burstCounter.now = fc.Now
|
rc.burstCounter.now = fc.Now
|
||||||
|
|
||||||
// A rapid cluster of `burst` requests must all be allowed, even
|
for i := 1; i <= req; i++ {
|
||||||
// though burst (6) > req (3): the sustained gate only bounds the
|
|
||||||
// total over the full 10s Interval (tickerRate <= 3 needs count <=
|
|
||||||
// 30), so a small cluster like this never trips it.
|
|
||||||
for i := 1; i <= burst; i++ {
|
|
||||||
if !checkRateLimit("u", "role-b", rc, "") {
|
if !checkRateLimit("u", "role-b", rc, "") {
|
||||||
t.Fatalf("request %d in first burst: got denied, want allowed (<= Burst=%d)", i, burst)
|
t.Fatalf("request %d: got denied, want allowed (<= Req=%d per Interval)", i, req)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// The (burst+1)th request in the SAME short window must be denied
|
// Clearing the burst window must not readmit: Req is per Interval.
|
||||||
// by the burst gate specifically - not by the sustained gate,
|
fc.Advance(2 * burstWindowDuration(interval))
|
||||||
// which alone would still allow it (tickerRate = 7/10 = 0.7 <= 3).
|
|
||||||
// This is the proof burst > req is not a no-op: it is the gate
|
|
||||||
// that actually fires here.
|
|
||||||
if checkRateLimit("u", "role-b", rc, "") {
|
if checkRateLimit("u", "role-b", rc, "") {
|
||||||
t.Fatalf("request %d: got allowed, want denied by burst gate (> Burst=%d)", burst+1, burst)
|
t.Fatalf("request %d: got allowed, want denied by the sustained gate (Req=%d per Interval)", req+1, req)
|
||||||
}
|
|
||||||
|
|
||||||
// Roll the burst window forward (fake clock only - the real
|
|
||||||
// sustained ticker has not aged, since no real time passed) and
|
|
||||||
// repeat clusters of `burst` requests. Each cluster alone passes
|
|
||||||
// burst, but the cumulative sustained count keeps climbing until
|
|
||||||
// it exceeds req*interval.Seconds() = 30, at which point the
|
|
||||||
// sustained gate must deny regardless of how the burst window
|
|
||||||
// resets. This proves "sustained still capped at Req".
|
|
||||||
sustainedTripped := false
|
|
||||||
for batch := 0; batch < 10 && !sustainedTripped; batch++ {
|
|
||||||
fc.Advance(2 * burstWindowDuration(interval)) // clears the burst window
|
|
||||||
for i := 1; i <= burst; i++ {
|
|
||||||
allowed := checkRateLimit("u", "role-b", rc, "")
|
|
||||||
rate := rc.RateCounterTicker.GetRate()
|
|
||||||
if rate > float64(req) {
|
|
||||||
if allowed {
|
|
||||||
t.Fatalf("batch %d req %d: allowed with sustained rate %.2f > Req=%d", batch, i, rate, req)
|
|
||||||
}
|
|
||||||
sustainedTripped = true
|
|
||||||
break
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if !sustainedTripped {
|
|
||||||
t.Fatalf("sustained gate never tripped after repeated bursts; expected it to cap total requests at Req over the Interval")
|
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
|
||||||
|
)
|
||||||
|
|
||||||
|
func setupRateLimitCfg() {
|
||||||
|
cfg = &config{}
|
||||||
|
parseConfig()
|
||||||
|
cfg.Logger = libpack_logger.New()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Pins theorems/Theorems/RateLimitUnits.lean: Req counts requests per Interval, not per second.
|
||||||
|
func TestCheckRateLimit_ReqIsPerInterval(t *testing.T) {
|
||||||
|
const req = 3
|
||||||
|
setupRateLimitCfg()
|
||||||
|
rc := newBurstTestRoleConfig(time.Minute, req, 0)
|
||||||
|
for i := 1; i <= req; i++ {
|
||||||
|
if !checkRateLimit("u", "r", rc, "") {
|
||||||
|
t.Fatalf("request %d denied, want allowed (Req=%d per minute)", i, req)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if checkRateLimit("u", "r", rc, "") {
|
||||||
|
t.Fatalf("request %d allowed, want denied: Req=%d per minute", req+1, req)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// With Burst >= Req the burst gate never rejects what the sustained gate admits.
|
||||||
|
func TestCheckRateLimit_BurstAboveReqIsNoOp(t *testing.T) {
|
||||||
|
const req, burst = 3, 6
|
||||||
|
setupRateLimitCfg()
|
||||||
|
rc := newBurstTestRoleConfig(time.Minute, req, burst)
|
||||||
|
for i := 1; i <= req; i++ {
|
||||||
|
if !checkRateLimit("u", "r", rc, "") {
|
||||||
|
t.Fatalf("request %d denied", i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if checkRateLimit("u", "r", rc, "") {
|
||||||
|
t.Fatalf("request %d allowed, want denied by the sustained gate", req+1)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user