diff --git a/99-experiments/global-ratelimit/README.md b/99-experiments/global-ratelimit/README.md new file mode 100644 index 0000000..2f43053 --- /dev/null +++ b/99-experiments/global-ratelimit/README.md @@ -0,0 +1,118 @@ +# Experiment — a global rate limiter as a service + +Exercise 03 built an in-process token bucket. That limits **one process**. +When N replicas of a service must share one quota ("tenant-a gets 10 req/s, +total"), the bucket has to live somewhere they can all reach. This is the +smallest honest version of that: an HTTP service holding per-key buckets, and +a demo client that plays one replica. + +``` +99-experiments/global-ratelimit/ + limiter/ token bucket (ex03) + sharded keyed store (ex09) + idle eviction go test -race + server/ HTTP service: /allow, /stats, /healthz + client/ demo replica: asks /allow before each unit of work +``` + +## Run it + +``` +# terminal 1 — the limiter +go run ./99-experiments/global-ratelimit/server -rate 10 -burst 10 + +# quick check +for i in $(seq 12); do curl -s -o /dev/null -w '%{http_code} ' 'localhost:8100/allow?key=t'; done +# 200 200 200 200 200 200 200 200 200 200 429 429 +curl -si 'localhost:8100/allow?key=t' | grep -iE 'HTTP/|retry|ratelimit' + +# terminals 2–4 — three replicas, each WANTS 20/s, all on key tenant-a +go run ./99-experiments/global-ratelimit/client -id inst-1 -rps 20 +go run ./99-experiments/global-ratelimit/client -id inst-2 -rps 20 +go run ./99-experiments/global-ratelimit/client -id inst-3 -rps 20 + +curl -s localhost:8100/stats +``` + +Observed (3 × 20 rps attempted, limit 10/s, 8 s): + +``` +inst-1 summary: attempts=159 allowed=31 denied=127 → work done=32 +inst-2 summary: attempts=159 allowed=28 denied=130 → work done=29 +inst-3 summary: attempts=160 allowed=29 denied=130 → work done=30 + ─── + 88 ≈ 10/s × 8 s + burst 10 +inst-1 /allow latency: p50=0.6ms p99=11ms +``` + +The **sum** stays at the limit, not each instance — that's the whole point of +centralising it. Each replica gets roughly a third, with no coordination +between replicas and no per-replica configuration. + +## API + +``` +POST|GET /allow?key=K[&n=1] + 200 {"allowed":true,"remaining":9,"limit":10,"rate":10} + 429 {"allowed":false,"remaining":0.3,"limit":10,"rate":10,"retry_after_ms":70} + headers: X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After (429; whole seconds, rounded up) + 400 missing key / bad n +GET /stats {"keys":2,"allowed_total":99,"denied_total":391} +GET /healthz ok +``` + +Server flags: `-addr`, `-rate`/`-burst` (defaults for unknown keys), +`-limits "tenant-a=100:200,tenant-b=5:5"` (per-key overrides), `-sweep 1m`. + +Client flags: `-id`, `-limiter`, `-key`, `-rps`, `-duration`, `-timeout 200ms`, +`-fail open|closed`. + +## What to observe + +**The hop.** Every decision is an HTTP round trip: p50 ≈ 0.6 ms on loopback, +1–2 ms across an AZ. The in-process limiter is ~100 ns. That's the price of +"global", and why you only pay it at the edge (per-tenant quota), not in inner +loops. + +**Fail-open vs fail-closed.** Start a client with each mode, then Ctrl-C the +server mid-run: + +``` +open allowed=0 denied=0 errors=10 → work done keeps going (49 of 59) +closed allowed=0 denied=0 errors=10 → work stops (18 of 59) +``` + +The limiter is now a dependency on the request path. Fail-open protects your +users and risks overloading what the limiter guards; fail-closed protects the +backend and turns a limiter outage into *your* outage. Pick deliberately, per +call site, and alert on the error rate either way. `-timeout` matters for the +same reason: a *slow* limiter must not make you slow. + +**Idle eviction.** `-sweep` removes buckets untouched for that long. An idle +bucket is full anyway (once `idle ≥ burst/rate`), so recreating it later is +indistinguishable from keeping it — `last` doubles as the idle timestamp, +nothing extra to track. Without this, one bucket per API key ever seen lives +forever. + +**Burst semantics across replicas.** The first second admits ~20 (burst 10 + +10/s). Replicas that start first eat the burst. If that's unfair for your use +case, lower `burst` — global buckets make the burst global too. + +## What this is not + +- **Not HA.** One process, in memory. Restart = all buckets full again + (a free burst for everyone) and, for the duration, fail-open or fail-closed + as configured. Real deployments put the bucket in Redis and run the + refill+consume as one Lua script so it's atomic — same arithmetic as + `limiter.Bucket.AllowN`, different storage. +- **Not accurate under Redis Cluster / multiple limiter nodes** unless keys are + routed consistently to one node. Keys → node is what makes a "global" bucket + global. +- **No local tier.** Production usually also keeps a coarse in-process limiter + (ex03) as a pre-filter, so a flooding client is cut off before it costs a + network hop per request, and caches "denied until T" locally. + +## References + +- Stripe, "Scaling your API with rate limiters" — the token bucket + Redis + pattern and why to shed load with it +- Envoy `ratelimit` service / `local_ratelimit` filter — the two-tier design +- `golang.org/x/time/rate` — the in-process reference implementation diff --git a/99-experiments/global-ratelimit/client/main.go b/99-experiments/global-ratelimit/client/main.go new file mode 100644 index 0000000..916e0b4 --- /dev/null +++ b/99-experiments/global-ratelimit/client/main.go @@ -0,0 +1,144 @@ +// Demo downstream service replica that consults the global rate limiter +// before each unit of work. +// +// Run several in parallel with different -id against one server and watch +// the SUM of their allowed rates stay at the key's limit: +// +// go run ./99-experiments/global-ratelimit/client -id inst-1 -rps 20 +// go run ./99-experiments/global-ratelimit/client -id inst-2 -rps 20 +// go run ./99-experiments/global-ratelimit/client -id inst-3 -rps 20 +// +// Then Ctrl-C the server mid-run and compare -fail open vs -fail closed. +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "net/url" + "os" + "os/signal" + "sort" + "time" +) + +type allowResponse struct { + Allowed bool `json:"allowed"` + RetryAfterMS int64 `json:"retry_after_ms"` +} + +type outcome int + +const ( + allowed outcome = iota + denied + errored // limiter unreachable / timed out / bad response +) + +// ask performs one /allow call. It is the ONLY place that talks to the limiter. +func ask(ctx context.Context, client *http.Client, endpoint string) (outcome, time.Duration) { + start := time.Now() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, nil) + if err != nil { + return errored, time.Since(start) + } + resp, err := client.Do(req) + if err != nil { + return errored, time.Since(start) + } + defer resp.Body.Close() + // Decode fully (which also drains the body → connection reuse). + var ar allowResponse + if err := json.NewDecoder(resp.Body).Decode(&ar); err != nil { + io.Copy(io.Discard, resp.Body) + return errored, time.Since(start) + } + switch resp.StatusCode { + case http.StatusOK: + return allowed, time.Since(start) + case http.StatusTooManyRequests: + return denied, time.Since(start) + default: + return errored, time.Since(start) + } +} + +func percentile(sorted []time.Duration, p float64) time.Duration { + if len(sorted) == 0 { + return 0 + } + i := int(float64(len(sorted)-1) * p) + return sorted[i] +} + +func main() { + id := flag.String("id", "inst-1", "this replica's name (for output only)") + limiterURL := flag.String("limiter", "http://127.0.0.1:8100", "limiter service base URL") + key := flag.String("key", "tenant-a", "rate-limit key to consume from") + rps := flag.Float64("rps", 20, "how many units of work this replica ATTEMPTS per second") + duration := flag.Duration("duration", 30*time.Second, "how long to run") + failMode := flag.String("fail", "open", "what to do when the limiter is unreachable: open (proceed) | closed (deny)") + timeout := flag.Duration("timeout", 200*time.Millisecond, "per-call timeout to the limiter") + flag.Parse() + + if *failMode != "open" && *failMode != "closed" { + fmt.Fprintln(os.Stderr, "-fail must be open or closed") + os.Exit(2) + } + + endpoint := *limiterURL + "/allow?" + url.Values{"key": {*key}}.Encode() + + // One shared client, with a timeout: a limiter that is slow must not make + // US slow beyond `timeout`. Keep-alive means one TCP conn for the whole run. + client := &http.Client{Timeout: *timeout} + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + ctx, cancel := context.WithTimeout(ctx, *duration) + defer cancel() + + tick := time.NewTicker(time.Duration(float64(time.Second) / *rps)) + defer tick.Stop() + report := time.NewTicker(time.Second) + defer report.Stop() + + var sec, total [3]int // per-second and cumulative counts by outcome + var latencies []time.Duration + workDone := 0 // units of work actually performed (allowed + fail-open errors) + + fmt.Printf("%s: attempting %.0f/s on key %q via %s, fail-%s, timeout %v\n", + *id, *rps, *key, *limiterURL, *failMode, *timeout) + +loop: + for { + select { + case <-ctx.Done(): + break loop + case <-report.C: + fmt.Printf("%s allowed=%-3d denied=%-3d errors=%-3d (last 1s)\n", *id, sec[allowed], sec[denied], sec[errored]) + sec = [3]int{} + case <-tick.C: + o, lat := ask(ctx, client, endpoint) + if ctx.Err() != nil { + break loop // aborted by our own shutdown, not a limiter failure + } + sec[o]++ + total[o]++ + latencies = append(latencies, lat) + // Do the actual "work" only if allowed — or on error, if fail-open. + if o == allowed || (o == errored && *failMode == "open") { + workDone++ + } + } + } + + sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) + n := total[allowed] + total[denied] + total[errored] + fmt.Printf("\n%s summary: attempts=%d allowed=%d denied=%d errors=%d → work done=%d (fail-%s)\n", + *id, n, total[allowed], total[denied], total[errored], workDone, *failMode) + fmt.Printf("%s /allow latency: p50=%v p99=%v max=%v\n", + *id, percentile(latencies, 0.50), percentile(latencies, 0.99), percentile(latencies, 1)) +} diff --git a/99-experiments/global-ratelimit/limiter/limiter.go b/99-experiments/global-ratelimit/limiter/limiter.go new file mode 100644 index 0000000..cf89337 --- /dev/null +++ b/99-experiments/global-ratelimit/limiter/limiter.go @@ -0,0 +1,185 @@ +// Package limiter is the token bucket from exercise 03 turned into a keyed +// store suitable for a shared rate-limiter service: many buckets addressed by +// a string key (tenant, client id, API key), sharded like exercise 09, with +// eviction of idle buckets so the map does not grow forever. +package limiter + +import ( + "hash/fnv" + "sync" + "sync/atomic" + "time" +) + +// ---- Bucket: a single token bucket (exercise 03) ---- + +// Bucket is a lazily refilled token bucket. Safe for concurrent use. +type Bucket struct { + mu sync.Mutex + tokens float64 + max float64 + rate float64 // tokens per second + last time.Time +} + +// NewBucket returns a full bucket allowing `rate` tokens/s with bursts up to `burst`. +func NewBucket(rate float64, burst int) *Bucket { + return &Bucket{tokens: float64(burst), max: float64(burst), rate: rate, last: time.Now()} +} + +// Decision is the outcome of an AllowN call. +type Decision struct { + Allowed bool + Remaining float64 // tokens left after this decision + Limit int // burst (bucket capacity) + Rate float64 // refill rate, tokens/s + RetryAfter time.Duration // when denied: how long until n tokens would be available +} + +// AllowN tries to take n tokens. Non-blocking. +func (b *Bucket) AllowN(n int) Decision { + b.mu.Lock() + defer b.mu.Unlock() + + // 1. lazy refill, capped at max; `last` advances on every call + now := time.Now() + b.tokens += now.Sub(b.last).Seconds() * b.rate + if b.tokens > b.max { + b.tokens = b.max + } + b.last = now + + d := Decision{Limit: int(b.max), Rate: b.rate} + // 2. consume or compute how long until we could + if b.tokens >= float64(n) { + b.tokens -= float64(n) + d.Allowed = true + } else if b.rate > 0 { + d.RetryAfter = time.Duration((float64(n) - b.tokens) / b.rate * float64(time.Second)) + } + d.Remaining = b.tokens + return d +} + +// idleSince reports the last time the bucket was touched and whether it is +// full (i.e. nothing would be lost by forgetting it). +func (b *Bucket) idleSince() (time.Time, bool) { + b.mu.Lock() + defer b.mu.Unlock() + // A bucket untouched for `max/rate` seconds is full again by definition, + // so idle time alone decides; we don't need to refill here. + return b.last, true +} + +// ---- Store: many buckets by key (exercise 09's sharding) ---- + +const shards = 16 + +type shard struct { + mu sync.Mutex + m map[string]*Bucket +} + +// Limit is a (rate, burst) pair. +type Limit struct { + Rate float64 + Burst int +} + +// Store holds one Bucket per key, created lazily with the default Limit or a +// per-key override. +type Store struct { + def Limit + overrides map[string]Limit // read-only after construction + shards [shards]shard + + allowed atomic.Int64 + denied atomic.Int64 +} + +// NewStore creates a store; unknown keys get `def`, keys in `overrides` get theirs. +func NewStore(def Limit, overrides map[string]Limit) *Store { + s := &Store{def: def, overrides: overrides} + for i := range s.shards { + s.shards[i].m = make(map[string]*Bucket) + } + return s +} + +func (s *Store) shardFor(key string) *shard { + h := fnv.New32a() + h.Write([]byte(key)) + return &s.shards[h.Sum32()%shards] +} + +func (s *Store) limitFor(key string) Limit { + if l, ok := s.overrides[key]; ok { + return l + } + return s.def +} + +// bucket returns the bucket for key, creating it on first sight. +func (s *Store) bucket(key string) *Bucket { + sh := s.shardFor(key) + sh.mu.Lock() + defer sh.mu.Unlock() + b, ok := sh.m[key] + if !ok { + l := s.limitFor(key) + b = NewBucket(l.Rate, l.Burst) + sh.m[key] = b + } + return b +} + +// Allow tries to take n tokens from key's bucket. +func (s *Store) Allow(key string, n int) Decision { + d := s.bucket(key).AllowN(n) + if d.Allowed { + s.allowed.Add(1) + } else { + s.denied.Add(1) + } + return d +} + +// Sweep removes buckets not touched for at least `idle`. Such a bucket has +// refilled to full anyway (as long as idle >= burst/rate), so dropping it and +// recreating it later is unobservable to clients. Returns how many were removed. +func (s *Store) Sweep(idle time.Duration) int { + cutoff := time.Now().Add(-idle) + removed := 0 + for i := range s.shards { + sh := &s.shards[i] + sh.mu.Lock() + for k, b := range sh.m { + if last, _ := b.idleSince(); last.Before(cutoff) { + delete(sh.m, k) + removed++ + } + } + sh.mu.Unlock() + } + return removed +} + +// Stats is a point-in-time snapshot for observability. +type Stats struct { + Keys int `json:"keys"` + Allowed int64 `json:"allowed_total"` + Denied int64 `json:"denied_total"` +} + +// Stats returns counters. Keys is counted shard by shard, not atomically +// across shards (fine for metrics; see exercise 09 Snapshot discussion). +func (s *Store) Stats() Stats { + st := Stats{Allowed: s.allowed.Load(), Denied: s.denied.Load()} + for i := range s.shards { + sh := &s.shards[i] + sh.mu.Lock() + st.Keys += len(sh.m) + sh.mu.Unlock() + } + return st +} diff --git a/99-experiments/global-ratelimit/limiter/limiter_test.go b/99-experiments/global-ratelimit/limiter/limiter_test.go new file mode 100644 index 0000000..a2bad3a --- /dev/null +++ b/99-experiments/global-ratelimit/limiter/limiter_test.go @@ -0,0 +1,128 @@ +package limiter + +import ( + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestBurstThenDeny(t *testing.T) { + s := NewStore(Limit{Rate: 1, Burst: 3}, nil) + for i := 0; i < 3; i++ { + if !s.Allow("k", 1).Allowed { + t.Fatalf("request %d within burst should be allowed", i) + } + } + d := s.Allow("k", 1) + if d.Allowed { + t.Fatal("4th immediate request should be denied") + } + if d.RetryAfter <= 0 || d.RetryAfter > time.Second { + t.Fatalf("retry-after should be ~1s at rate 1, got %v", d.RetryAfter) + } +} + +func TestRefill(t *testing.T) { + s := NewStore(Limit{Rate: 20, Burst: 1}, nil) + if !s.Allow("k", 1).Allowed { + t.Fatal("initial token missing") + } + if s.Allow("k", 1).Allowed { + t.Fatal("should be empty immediately after") + } + time.Sleep(120 * time.Millisecond) + if !s.Allow("k", 1).Allowed { + t.Fatal("should have refilled") + } +} + +func TestBurstIsCapped(t *testing.T) { + s := NewStore(Limit{Rate: 1000, Burst: 2}, nil) + time.Sleep(50 * time.Millisecond) + allowed := 0 + for i := 0; i < 10; i++ { + if s.Allow("k", 1).Allowed { + allowed++ + } + } + if allowed > 2 { + t.Fatalf("idle time built credit beyond burst: %d", allowed) + } +} + +func TestKeysAreIsolatedAndOverridden(t *testing.T) { + s := NewStore(Limit{Rate: 0.0001, Burst: 1}, map[string]Limit{"vip": {Rate: 0.0001, Burst: 5}}) + if !s.Allow("a", 1).Allowed || s.Allow("a", 1).Allowed { + t.Fatal("key a should allow exactly one") + } + if !s.Allow("b", 1).Allowed { + t.Fatal("key b must have its own bucket") + } + for i := 0; i < 5; i++ { + if !s.Allow("vip", 1).Allowed { + t.Fatalf("vip request %d should be allowed (burst 5)", i) + } + } + if s.Allow("vip", 1).Allowed { + t.Fatal("vip 6th should be denied") + } + if got := s.Stats().Keys; got != 3 { + t.Fatalf("expected 3 keys, got %d", got) + } +} + +func TestAllowN(t *testing.T) { + s := NewStore(Limit{Rate: 0.0001, Burst: 10}, nil) + if !s.Allow("k", 7).Allowed { + t.Fatal("7 of 10 should be allowed") + } + if s.Allow("k", 4).Allowed { + t.Fatal("4 more (3 left) should be denied") + } + if !s.Allow("k", 3).Allowed { + t.Fatal("exactly 3 left should be allowed") + } +} + +func TestConcurrentNoOverAdmission(t *testing.T) { + s := NewStore(Limit{Rate: 0.0001, Burst: 100}, nil) + var allowed atomic.Int64 + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 10; j++ { + if s.Allow("shared", 1).Allowed { + allowed.Add(1) + } + } + }() + } + wg.Wait() + if got := allowed.Load(); got != 100 { + t.Fatalf("expected exactly 100 admissions, got %d", got) + } + st := s.Stats() + if st.Allowed != 100 || st.Denied != 400 { + t.Fatalf("stats mismatch: %+v", st) + } +} + +func TestSweepEvictsIdleOnly(t *testing.T) { + s := NewStore(Limit{Rate: 1000, Burst: 1}, nil) + s.Allow("old", 1) + time.Sleep(30 * time.Millisecond) + s.Allow("fresh", 1) + if n := s.Sweep(20 * time.Millisecond); n != 1 { + t.Fatalf("expected 1 evicted, got %d", n) + } + if got := s.Stats().Keys; got != 1 { + t.Fatalf("expected 1 key left, got %d", got) + } + // evicted key comes back full, which is what an idle bucket would be anyway + if !s.Allow("old", 1).Allowed { + t.Fatal("recreated bucket should start full") + } +} diff --git a/99-experiments/global-ratelimit/server/main.go b/99-experiments/global-ratelimit/server/main.go new file mode 100644 index 0000000..071a99b --- /dev/null +++ b/99-experiments/global-ratelimit/server/main.go @@ -0,0 +1,182 @@ +// Global rate limiter service. +// +// One process holds per-key token buckets; any number of downstream service +// replicas ask it "may I?" before doing work, so the quota is enforced +// globally rather than per replica. +// +// go run ./99-experiments/global-ratelimit/server -rate 10 -burst 10 +// curl -i 'localhost:8100/allow?key=tenant-a' +// curl -s 'localhost:8100/stats' +// +// API +// +// POST|GET /allow?key=K[&n=1] +// 200 {"allowed":true, "remaining":9, "limit":10, "rate":10} +// 429 {"allowed":false, "remaining":0.3, "limit":10, "rate":10, "retry_after_ms":70} +// headers: X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After (429 only) +// 400 when key is missing or n is not a positive integer +// GET /stats → {"keys":N,"allowed_total":A,"denied_total":D} +// GET /healthz → 200 ok +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "log" + "math" + "net/http" + "os" + "os/signal" + "strconv" + "strings" + "time" + + "gosre-exercises/99-experiments/global-ratelimit/limiter" +) + +type allowResponse struct { + Allowed bool `json:"allowed"` + Remaining float64 `json:"remaining"` + Limit int `json:"limit"` + Rate float64 `json:"rate"` + RetryAfterMS int64 `json:"retry_after_ms,omitempty"` +} + +func allowHandler(store *limiter.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + key := r.URL.Query().Get("key") + if key == "" { + http.Error(w, `missing "key" query parameter`, http.StatusBadRequest) + return + } + n := 1 + if raw := r.URL.Query().Get("n"); raw != "" { + v, err := strconv.Atoi(raw) + if err != nil || v <= 0 { + http.Error(w, `"n" must be a positive integer`, http.StatusBadRequest) + return + } + n = v + } + + d := store.Allow(key, n) + + w.Header().Set("Content-Type", "application/json") + w.Header().Set("X-RateLimit-Limit", strconv.Itoa(d.Limit)) + w.Header().Set("X-RateLimit-Remaining", strconv.Itoa(int(math.Floor(d.Remaining)))) + resp := allowResponse{Allowed: d.Allowed, Remaining: d.Remaining, Limit: d.Limit, Rate: d.Rate} + status := http.StatusOK + if !d.Allowed { + status = http.StatusTooManyRequests + resp.RetryAfterMS = d.RetryAfter.Milliseconds() + // Retry-After is whole seconds; round UP so clients don't retry too early. + w.Header().Set("Retry-After", strconv.Itoa(int(math.Ceil(d.RetryAfter.Seconds())))) + } + w.WriteHeader(status) + json.NewEncoder(w).Encode(resp) + } +} + +func statsHandler(store *limiter.Store) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(store.Stats()) + } +} + +// parseLimits parses "key=rate:burst,key2=rate:burst". +func parseLimits(s string) (map[string]limiter.Limit, error) { + out := map[string]limiter.Limit{} + if strings.TrimSpace(s) == "" { + return out, nil + } + for _, item := range strings.Split(s, ",") { + key, spec, ok := strings.Cut(strings.TrimSpace(item), "=") + if !ok { + return nil, fmt.Errorf("bad limit %q: want key=rate:burst", item) + } + rateS, burstS, ok := strings.Cut(spec, ":") + if !ok { + return nil, fmt.Errorf("bad limit %q: want key=rate:burst", item) + } + rate, err1 := strconv.ParseFloat(rateS, 64) + burst, err2 := strconv.Atoi(burstS) + if err1 != nil || err2 != nil || rate <= 0 || burst <= 0 { + return nil, fmt.Errorf("bad limit %q: rate and burst must be positive", item) + } + out[key] = limiter.Limit{Rate: rate, Burst: burst} + } + return out, nil +} + +func main() { + addr := flag.String("addr", ":8100", "listen address") + rate := flag.Float64("rate", 10, "default tokens/second for keys without an override") + burst := flag.Int("burst", 20, "default burst for keys without an override") + limits := flag.String("limits", "", `per-key overrides, e.g. "tenant-a=100:200,tenant-b=5:5"`) + sweep := flag.Duration("sweep", time.Minute, "evict buckets idle for at least this long (0 = never)") + flag.Parse() + + overrides, err := parseLimits(*limits) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(2) + } + store := limiter.NewStore(limiter.Limit{Rate: *rate, Burst: *burst}, overrides) + + mux := http.NewServeMux() + mux.HandleFunc("/allow", allowHandler(store)) + mux.HandleFunc("/stats", statsHandler(store)) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { fmt.Fprintln(w, "ok") }) + + srv := &http.Server{ + Addr: *addr, + Handler: mux, + // Never run a server with zero timeouts (see 99-experiments/body-drain). + ReadHeaderTimeout: 5 * time.Second, + WriteTimeout: 5 * time.Second, + IdleTimeout: 60 * time.Second, + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + + // Idle-key eviction: the "map grows forever" fix. Also a periodic stats line. + go func() { + if *sweep <= 0 { + return + } + t := time.NewTicker(*sweep) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + removed := store.Sweep(*sweep) + st := store.Stats() + log.Printf("sweep: evicted=%d keys=%d allowed=%d denied=%d", removed, st.Keys, st.Allowed, st.Denied) + } + } + }() + + go func() { + log.Printf("global rate limiter listening on %s (default %v/s burst %d, %d overrides)", *addr, *rate, *burst, len(overrides)) + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Fatal(err) + } + }() + + <-ctx.Done() + log.Println("shutting down") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := srv.Shutdown(shutdownCtx); err != nil { + log.Printf("shutdown: %v", err) + } + st := store.Stats() + log.Printf("final: keys=%d allowed=%d denied=%d", st.Keys, st.Allowed, st.Denied) +}