From 540b518fdeb35b18dace39bd4f67d978c6404158 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 1 Nov 2026 17:30:08 +0100 Subject: [PATCH] Rate limits counted across the control planes with router.active_active Each router's per-counter request and token counts, this minute's and the last, ride the standby's two-second exchange with the leader. The leader answers each standby with everyone else's sum, its own included, and puts the standbys' sum in force on its own router; each limiter adds the others' counts to its sliding window. Co-Authored-By: Claude Opus 5.5 (1M context) --- BACKLOG.md | 10 ++- CHANGELOG.md | 8 +- docs/admin-api.openapi.json | 48 ++++++++++++ docs/install.md | 13 +++- docs/reference.md | 11 +-- internal/accounting/limiter.go | 13 +++- internal/accounting/shared.go | 110 +++++++++++++++++++++++++++ internal/accounting/shared_test.go | 71 +++++++++++++++++ internal/cli/standbyroutes.go | 16 +++- internal/cli/standbyroutes_test.go | 33 ++++++++ internal/config/config.go | 11 +-- internal/control/adminapi/routing.go | 2 +- internal/control/control.go | 3 + internal/control/routing.go | 103 ++++++++++++++++++++++++- internal/control/routing_test.go | 51 +++++++++++++ internal/docgen/reference_gen.go | 2 +- internal/router/router.go | 17 +++++ 17 files changed, 494 insertions(+), 28 deletions(-) create mode 100644 internal/accounting/shared.go create mode 100644 internal/accounting/shared_test.go diff --git a/BACKLOG.md b/BACKLOG.md index 4026f4a..cc7b03f 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -1688,6 +1688,8 @@ how to fix it by hand (`ctl model resources`, then `ctl restart`). Rate limits are counted in each control plane's memory. A failover starts every key's minute afresh. Documented in install.md; worth a shared count in the store only if a customer's limits must hold across a failover. +With `router.active_active` (O2, 2026-11-01) the counts are shared every 2 s, +so a standby taking over keeps the minute; without it, as above. ### J7. nginx in front of several control planes (S) `deploy/proxy/Caddyfile` sends to whichever control plane answers `/leader`. @@ -1951,8 +1953,12 @@ one control plane's router can. In order: leader takes the last uses in (else a replica busy behind a standby is evicted as idle) and the on-demand models asked for; a catalog changed on the leader is read again from the store. Unheard from the leader for a minute, a standby routes nothing and passes `/v1` on as before. The portal, `/admin` and `/mcp` still go to the leader. - - [~] budgets: each router raises its counters to the store's every 30 s, so another's spend is seen within that. - Rate limits are still per process: up to N× a key's limit across N control planes (J6). + - [x] budgets: each router raises its counters to the store's every 30 s, so another's spend is seen within that. + - [x] rate limits cluster-wide (2026-11-01): each router's per-counter request and token counts, this minute's + and the last as absolute figures, ride the same 2 s exchange; the leader answers each standby with everyone + else's sum, its own included, and puts the standbys' sum in force on its own router. Each limiter adds the + others' counts to its sliding window. Overshoot is bounded by what N routers allow in 2 s; a standby quiet for + 10 s stops counting. (J6's single-process case was never the problem.) - [ ] a session's requests to one router and one replica: the session ring already picks the same replica on every router for the same replica set; what is left is the same router, which is the balancer's (source hash). - [ ] in front: IPVS on the keepalived the setup already installs, DNS round-robin, or the customer's balancer diff --git a/CHANGELOG.md b/CHANGELOG.md index 930049b..5df569c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,9 +37,11 @@ APIs change between minors without a deprecation period. replicas from the leader every two seconds over the agent port (no engine keys travel) and reports back what it used, so the leader does not evict a replica busy behind a standby, and the on-demand models it was asked - for. Budgets are brought up to the store's every 30 seconds; rate limits - are still counted per control plane. Off by default; the portal, `/admin` - and `/mcp` still go to the leader. + for. Rate limits are the cluster's: each router's request and token + counts reach the others through the leader every two seconds, so a key + spread over several control planes gets its limit, not that many times + it. Budgets are brought up to the store's every 30 seconds. Off by + default; the portal, `/admin` and `/mcp` still go to the leader. - **A served model's longest answer, changed in place.** `llmgw ctl model context --max-output N` (and `max_output` on `POST /api/v1/models/{name}/context`, and the console's Context length… form) diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index 3cf0d95..8680d4b 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -9770,6 +9770,42 @@ } } }, + "accounting.Counts": { + "type": "object", + "description": "Counts are one limiter's counts by counter: this minute's and the one before, as absolute figures, so reporting them twice changes nothing.", + "properties": { + "minute": { + "type": "string", + "format": "date-time" + }, + "current": { + "type": "object", + "additionalProperties": { + "type": "integer", + "format": "int64" + } + }, + "previous": { + "type": "object", + "additionalProperties": { + "type": "integer", + "format": "int64" + } + } + } + }, + "accounting.Rates": { + "type": "object", + "description": "Rates are an accountant's two limiters' counts.", + "properties": { + "requests": { + "$ref": "#/components/schemas/accounting.Counts" + }, + "tokens": { + "$ref": "#/components/schemas/accounting.Counts" + } + } + }, "bench.Capacity": { "type": "object", "description": "Capacity is what one replica was measured to serve, kept so the measurement outlives the terminal it was printed in.\n\nThe conclusion of a run, not the run. The ramp itself — every concurrency level, its latencies, its errors — is worth reading once, while deciding whether the run was sound; it is not worth carrying around afterwards, and storing it would grow without bound as a model is re-benchmarked. What a later reader needs is the number and enough provenance to judge whether it is still true.", @@ -10632,6 +10668,10 @@ "type": "object", "description": "Activity is what a standby's router reports back: when it last sent each replica a request, and the on-demand models it was asked for. The leader evicts and loads by these, and without them a replica busy behind a standby looks idle to the one machine that decides.", "properties": { + "from": { + "type": "string", + "description": "From names the standby, so its counts replace its own last ones and are not given back to it." + }, "last_used": { "type": "object", "additionalProperties": { @@ -10644,6 +10684,10 @@ "items": { "type": "string" } + }, + "rates": { + "$ref": "#/components/schemas/accounting.Rates", + "description": "Rates are its router's own counts against each limit." } } }, @@ -10684,6 +10728,10 @@ "catalog": { "type": "string", "description": "Catalog changes whenever the leader puts a catalog in force. Only compared with the last one a standby saw, never with its own: a model added, scaled or given a canary is in the store, and the standby reads it again from there." + }, + "rates": { + "$ref": "#/components/schemas/accounting.Rates", + "description": "Rates are what every other control plane's router has counted against each limit, the leader's among them: added to the asker's own, they make a key's limits the cluster's." } } }, diff --git a/docs/install.md b/docs/install.md index 6e28113..228c5ba 100644 --- a/docs/install.md +++ b/docs/install.md @@ -2821,9 +2821,16 @@ Caddy doing that for three control planes, and `server add` checks, and says what to add to the mirror's unit if it refuses the new control plane — which would otherwise find out the first time it leads and fetches a model. -- **Limits are per process.** Rate limits are counted in memory, and clients - reach the leader through the floating address, so a failover starts each - key's minute afresh. +- **Standbys can answer clients too.** With `router.active_active: true` + (`llmgw ctl config set router.active_active true`) a standby answers `/v1` + itself, routing to the leader's replicas, instead of passing it on: more + machines carry the traffic. Point clients at every control plane — DNS, + or your own load balancer checking `/ready` — since the floating address + stays on the leader. Rate limits stay the cluster's: each router's counts + reach the others every two seconds, and a standby that becomes the leader + keeps the minute it was counting. Without the setting, rate limits are + counted in the leader's memory and a failover starts each key's minute + afresh. [deploy/ha/](../deploy/ha/) describes what each half does, for anyone putting their own load balancer in front. diff --git a/docs/reference.md b/docs/reference.md index 5da77a0..b56a2f4 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -2818,11 +2818,12 @@ written into the setup scripts and whose attempts are counted where. ActiveActive has standby control planes answer the client API themselves, routing to the leader's replicas, rather than passing every request to the leader. More machines carry the traffic, which -is the point; what each router keeps in memory is then its own: -rate limits are counted per control plane, so a key may make up to -as many times its limit as there are control planes taking its -requests; a budget is checked against the store every half minute; -cached answers and prefix affinity are split between them. Off by +is the point. Rate limits are the cluster's: each router's counts +reach the others through the leader every two seconds, so a key +spread over N control planes may pass its limit by what they allow +in those two seconds. A budget is checked against the store every +half minute; cached answers and prefix affinity are split between +the routers. Off by default. Which machine a client reaches is the load balancer's or DNS's to say; the floating address stays on the leader. diff --git a/internal/accounting/limiter.go b/internal/accounting/limiter.go index e85efee..89804ac 100644 --- a/internal/accounting/limiter.go +++ b/internal/accounting/limiter.go @@ -25,7 +25,11 @@ type Limiter struct { type window struct { // current and previous are counts in this minute and the one before. current, previous int64 - start time.Time + // othersCurrent and othersPrevious are the same counts on the other + // control planes taking requests, as last heard: what makes a limit the + // cluster's rather than each process's. + othersCurrent, othersPrevious int64 + start time.Time } // NewLimiter builds an empty limiter. @@ -104,21 +108,24 @@ func (w *window) roll(now time.Time) { return case start.Sub(w.start) == windowLen: w.previous, w.current = w.current, 0 + w.othersPrevious, w.othersCurrent = w.othersCurrent, 0 default: // More than one window has passed with no traffic. w.previous, w.current = 0, 0 + w.othersPrevious, w.othersCurrent = 0, 0 } w.start = start } // rate weights the previous window by how much of it is still in view. func (w *window) rate(now time.Time) float64 { + cur, prev := w.current+w.othersCurrent, w.previous+w.othersPrevious elapsed := now.Sub(w.start) if elapsed >= windowLen { - return float64(w.current) + return float64(cur) } remaining := 1 - float64(elapsed)/float64(windowLen) - return float64(w.previous)*remaining + float64(w.current) + return float64(prev)*remaining + float64(cur) } // Forget drops a key's window. Used when a key is revoked; keeping it would diff --git a/internal/accounting/shared.go b/internal/accounting/shared.go new file mode 100644 index 0000000..776567d --- /dev/null +++ b/internal/accounting/shared.go @@ -0,0 +1,110 @@ +package accounting + +import "time" + +// Counts are one limiter's counts by counter: this minute's and the one +// before, as absolute figures, so reporting them twice changes nothing. +type Counts struct { + Minute time.Time `json:"minute"` + Current map[string]int64 `json:"current,omitempty"` + Previous map[string]int64 `json:"previous,omitempty"` +} + +// Add sums other into c, aligned to c's minute: counts of a minute that is +// c's previous one land there, older ones are dropped. Clocks a second apart +// at a minute's turn are why. +func (c *Counts) Add(other Counts) { + switch { + case other.Minute.Equal(c.Minute): + c.Current = addInto(c.Current, other.Current) + c.Previous = addInto(c.Previous, other.Previous) + case other.Minute.Add(windowLen).Equal(c.Minute): + c.Previous = addInto(c.Previous, other.Current) + } +} + +func addInto(dst, src map[string]int64) map[string]int64 { + if len(src) == 0 { + return dst + } + if dst == nil { + dst = make(map[string]int64, len(src)) + } + for k, v := range src { + dst[k] += v + } + return dst +} + +// Counts reads this limiter's own counts, without the others'. +func (l *Limiter) Counts() Counts { + now := l.now() + out := Counts{Minute: now.Truncate(windowLen), Current: map[string]int64{}, Previous: map[string]int64{}} + l.mu.Lock() + defer l.mu.Unlock() + for k, w := range l.keys { + w.roll(now) + if w.current > 0 { + out.Current[k] = w.current + } + if w.previous > 0 { + out.Previous[k] = w.previous + } + } + return out +} + +// SetOthers replaces what the other control planes have counted. A counter +// they did not mention has nothing from them. +func (l *Limiter) SetOthers(c Counts) { + now := l.now() + aligned := Counts{Minute: now.Truncate(windowLen)} + aligned.Add(c) + l.mu.Lock() + defer l.mu.Unlock() + for k, w := range l.keys { + w.roll(now) + w.othersCurrent, w.othersPrevious = aligned.Current[k], aligned.Previous[k] + delete(aligned.Current, k) + delete(aligned.Previous, k) + } + // Counters only others have used yet: a key spread over the cluster is + // limited here from its first request. + for k := range aligned.Current { + w := l.windowLocked(k, now) + w.othersCurrent, w.othersPrevious = aligned.Current[k], aligned.Previous[k] + delete(aligned.Previous, k) + } + for k, v := range aligned.Previous { + l.windowLocked(k, now).othersPrevious = v + } +} + +// Rates are an accountant's two limiters' counts. +type Rates struct { + Requests Counts `json:"requests"` + Tokens Counts `json:"tokens"` +} + +// Add sums other into r. +func (r *Rates) Add(other Rates) { + r.Requests.Add(other.Requests) + r.Tokens.Add(other.Tokens) +} + +// NewRates is an empty sum for the minute now is in. +func NewRates(now time.Time) Rates { + m := now.Truncate(windowLen) + return Rates{Requests: Counts{Minute: m}, Tokens: Counts{Minute: m}} +} + +// Rates reads this process's own request and token counts. +func (a *Accountant) Rates() Rates { + return Rates{Requests: a.requests.Counts(), Tokens: a.tokens.Counts()} +} + +// SetOtherRates puts the other control planes' counts in force here. +func (a *Accountant) SetOtherRates(r Rates) { + a.requests.SetOthers(r.Requests) + a.tokens.SetOthers(r.Tokens) +} diff --git a/internal/accounting/shared_test.go b/internal/accounting/shared_test.go new file mode 100644 index 0000000..0db0836 --- /dev/null +++ b/internal/accounting/shared_test.go @@ -0,0 +1,71 @@ +package accounting + +import ( + "testing" + "time" +) + +// Two control planes each allowing a key six of its ten a minute: once each +// has heard the other's count, neither allows another. +func TestALimitIsTheClustersOnceTheCountsAreShared(t *testing.T) { + now := time.Date(2026, 11, 1, 12, 0, 10, 0, time.UTC) + a, b := NewLimiter(), NewLimiter() + a.now, b.now = func() time.Time { return now }, func() time.Time { return now } + for range 6 { + if !a.Allow("k", 10, 1) || !b.Allow("k", 10, 1) { + t.Fatal("refused within the limit") + } + } + a.SetOthers(b.Counts()) + b.SetOthers(a.Counts()) + if a.Allow("k", 10, 1) || b.Allow("k", 10, 1) { + t.Error("twelve of ten allowed once both knew") + } + // Reporting the same counts again changes nothing: they are absolute. + a.SetOthers(b.Counts()) + if got := a.Rate("k"); got != 12 { + t.Errorf("rate %v after the same counts twice", got) + } + // What the others counted is never charged back to them. + if c := a.Counts(); c.Current["k"] != 6 { + t.Errorf("own count %d", c.Current["k"]) + } +} + +// A key only the others have used is limited here from its first request, +// and their counts move into the previous minute with the clock. +func TestOthersCountsFollowTheMinute(t *testing.T) { + now := time.Date(2026, 11, 1, 12, 0, 50, 0, time.UTC) + a, b := NewLimiter(), NewLimiter() + a.now, b.now = func() time.Time { return now }, func() time.Time { return now } + for range 10 { + b.Allow("k", 10, 1) + } + a.SetOthers(b.Counts()) + if a.Allow("k", 10, 1) { + t.Error("a key the cluster has used up was allowed where it had not been seen") + } + now = now.Add(20 * time.Second) // 12:01:10, a sixth into the next minute + if got := a.Rate("k"); got < 8 || got > 8.5 { + t.Errorf("rate %v: the others' last minute should weigh five sixths", got) + } + // Counts of a minute ago, sent late, land in the previous minute. + old := b.Counts() + old.Minute = now.Truncate(time.Minute).Add(-time.Minute) + old.Current, old.Previous = map[string]int64{"k": 3}, nil + a.SetOthers(old) + if got := a.Rate("k"); got < 2 || got > 3 { + t.Errorf("rate %v from late counts", got) + } +} + +func TestRatesAddAlignsMinutes(t *testing.T) { + m := time.Date(2026, 11, 1, 12, 1, 0, 0, time.UTC) + sum := NewRates(m) + sum.Add(Rates{Requests: Counts{Minute: m, Current: map[string]int64{"k": 2}, Previous: map[string]int64{"k": 1}}}) + sum.Add(Rates{Requests: Counts{Minute: m.Add(-time.Minute), Current: map[string]int64{"k": 5}}}) + sum.Add(Rates{Requests: Counts{Minute: m.Add(-2 * time.Minute), Current: map[string]int64{"k": 100}}}) + if sum.Requests.Current["k"] != 2 || sum.Requests.Previous["k"] != 6 { + t.Errorf("sum %+v", sum.Requests) + } +} diff --git a/internal/cli/standbyroutes.go b/internal/cli/standbyroutes.go index 8216b4b..8983cc0 100644 --- a/internal/cli/standbyroutes.go +++ b/internal/cli/standbyroutes.go @@ -15,6 +15,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" + "gitea.mynodes.xyz/ilker/llmgw/internal/router" ) // How often a standby answering requests itself hears from the leader, and @@ -28,6 +29,9 @@ const ( standbyTrust = time.Minute ) +// The router is what counts against limits, and shares the counts. +var _ control.RateSharer = (*router.Router)(nil) + // standbyRoutes is a standby's copy of the leader's Ready replicas, when // standbys answer requests themselves (router.active_active). Each tick it // reports what its router used and is answered with the replicas, in one @@ -164,6 +168,12 @@ func (s *standbyRoutes) exchange(ctx context.Context) error { } s.mu.Unlock() + // What the leader and the other standbys counted, so this router's + // limits are the cluster's. + if rs, ok := s.usage.(control.RateSharer); ok && rt.Rates != nil { + rs.SetOtherRates(*rt.Rates) + } + // A model added or changed through the leader is in the store: the // catalog first, so the replicas published next resolve against it. if stale && s.refresh != nil { @@ -208,7 +218,11 @@ func (s *standbyRoutes) activity() (control.Activity, map[string]time.Time) { } s.mu.Lock() defer s.mu.Unlock() - act := control.Activity{LastUsed: map[string]time.Time{}} + act := control.Activity{From: s.fwd.self, LastUsed: map[string]time.Time{}} + if rs, ok := s.usage.(control.RateSharer); ok { + r := rs.Rates() + act.Rates = &r + } for id, at := range used { if at.After(s.sent[id]) { act.LastUsed[id] = at diff --git a/internal/cli/standbyroutes_test.go b/internal/cli/standbyroutes_test.go index 3c3ef32..1b32f6a 100644 --- a/internal/cli/standbyroutes_test.go +++ b/internal/cli/standbyroutes_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + "gitea.mynodes.xyz/ilker/llmgw/internal/accounting" "gitea.mynodes.xyz/ilker/llmgw/internal/control" ) @@ -51,6 +52,38 @@ type fixedUsage map[string]time.Time func (u fixedUsage) LastUsed() map[string]time.Time { return u } +// sharingUsage is a router whose limits are counted with the others'. +type sharingUsage struct { + fixedUsage + own accounting.Rates + others *accounting.Rates +} + +func (u *sharingUsage) Rates() accounting.Rates { return u.own } +func (u *sharingUsage) SetOtherRates(r accounting.Rates) { u.others = &r } + +// A standby reports its counts under its name and puts what the leader says +// the rest of the cluster counted in force on its router. +func TestAStandbysLimitsAreTheClusters(t *testing.T) { + theirs := accounting.NewRates(time.Now()) + theirs.Tokens.Current = map[string]int64{"key-1": 900} + leader := &fakeLeaderRouting{routing: control.Routing{Rates: &theirs}} + c := newForwardClusterWith(t, leader) + u := &sharingUsage{own: accounting.NewRates(time.Now())} + u.own.Requests.Current = map[string]int64{"key-1": 4} + s := newStandbyRoutes(c.standby, u, nil, nil, slog.New(slog.DiscardHandler)) + if err := s.exchange(t.Context()); err != nil { + t.Fatal(err) + } + act := leader.last() + if act.From != "ctl-b" || act.Rates == nil || act.Rates.Requests.Current["key-1"] != 4 { + t.Errorf("reported %+v", act) + } + if u.others == nil || u.others.Tokens.Current["key-1"] != 900 { + t.Errorf("the router was told %+v", u.others) + } +} + func TestAStandbyFollowsTheLeadersReplicas(t *testing.T) { leader := &fakeLeaderRouting{routing: control.Routing{Catalog: "a/1", Replicas: []control.RoutedReplica{{ID: "llama-r0", Model: "llama", Endpoint: "http://10.0.0.5:20001"}}}} diff --git a/internal/config/config.go b/internal/config/config.go index 19c3276..f56c46f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -644,11 +644,12 @@ type Router struct { // ActiveActive has standby control planes answer the client API // themselves, routing to the leader's replicas, rather than passing // every request to the leader. More machines carry the traffic, which - // is the point; what each router keeps in memory is then its own: - // rate limits are counted per control plane, so a key may make up to - // as many times its limit as there are control planes taking its - // requests; a budget is checked against the store every half minute; - // cached answers and prefix affinity are split between them. Off by + // is the point. Rate limits are the cluster's: each router's counts + // reach the others through the leader every two seconds, so a key + // spread over N control planes may pass its limit by what they allow + // in those two seconds. A budget is checked against the store every + // half minute; cached answers and prefix affinity are split between + // the routers. Off by // default. Which machine a client reaches is the load balancer's or // DNS's to say; the floating address stays on the leader. ActiveActive bool `koanf:"active_active"` diff --git a/internal/control/adminapi/routing.go b/internal/control/adminapi/routing.go index cef008f..5e5aded 100644 --- a/internal/control/adminapi/routing.go +++ b/internal/control/adminapi/routing.go @@ -22,6 +22,6 @@ func (a *API) registerRouting(mux Mux) { return } a.ctl.RecordActivity(act) - writeJSON(w, http.StatusOK, a.ctl.Routing()) + writeJSON(w, http.StatusOK, a.ctl.RoutingFor(act.From)) }) } diff --git a/internal/control/control.go b/internal/control/control.go index 1b3f8dc..9b782db 100644 --- a/internal/control/control.go +++ b/internal/control/control.go @@ -102,6 +102,8 @@ type Controller struct { // catRev changes with every catalog put in force here, so a standby // following this one knows to read the catalog again. catRev atomic.Uint64 + // rates are the standbys' counts against each limit, while this leads. + rates sharedRates stop chan struct{} done chan struct{} @@ -315,6 +317,7 @@ func (c *Controller) Start(ctx context.Context) { // Before publishing, so a replica's idleness is as fresh as // the endpoint list it travels beside. c.collectUsage() + c.shareRates() c.Publish(ctx) } } diff --git a/internal/control/routing.go b/internal/control/routing.go index a04c4fe..775af70 100644 --- a/internal/control/routing.go +++ b/internal/control/routing.go @@ -2,7 +2,10 @@ package control import ( "strconv" + "sync" "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/accounting" ) // RoutedReplica is a Ready replica as a router needs it: where it is, what it @@ -26,6 +29,10 @@ type Routing struct { // added, scaled or given a canary is in the store, and the standby reads // it again from there. Catalog string `json:"catalog"` + // Rates are what every other control plane's router has counted against + // each limit, the leader's among them: added to the asker's own, they + // make a key's limits the cluster's. + Rates *accounting.Rates `json:"rates,omitempty"` } // Activity is what a standby's router reports back: when it last sent each @@ -33,8 +40,66 @@ type Routing struct { // evicts and loads by these, and without them a replica busy behind a // standby looks idle to the one machine that decides. type Activity struct { + // From names the standby, so its counts replace its own last ones and + // are not given back to it. + From string `json:"from,omitempty"` LastUsed map[string]time.Time `json:"last_used,omitempty"` Want []string `json:"want,omitempty"` + // Rates are its router's own counts against each limit. + Rates *accounting.Rates `json:"rates,omitempty"` +} + +// RateSharer is a router whose limits can be counted with other routers'. +type RateSharer interface { + Rates() accounting.Rates + SetOtherRates(accounting.Rates) +} + +// ratesFresh is how long a standby's counts are believed after it last +// reported: a standby gone quiet stops counting against anyone's limits +// shortly after it stops taking requests. +const ratesFresh = 10 * time.Second + +// sharedRates are the standbys' last counts, by name. +type sharedRates struct { + mu sync.Mutex + by map[string]rateReport + seen bool // whether any standby has ever reported +} + +type rateReport struct { + rates accounting.Rates + at time.Time +} + +// put keeps a standby's latest counts. +func (s *sharedRates) put(from string, r accounting.Rates, now time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + if s.by == nil { + s.by = map[string]rateReport{} + } + s.by[from] = rateReport{rates: r, at: now} + s.seen = true +} + +// sum adds every fresh standby's counts but except's; stale ones are +// forgotten. ok is false when no standby has ever reported, and there is +// nothing to put in force. +func (s *sharedRates) sum(except string, now time.Time) (accounting.Rates, bool) { + out := accounting.NewRates(now) + s.mu.Lock() + defer s.mu.Unlock() + for name, r := range s.by { + if now.Sub(r.at) > ratesFresh { + delete(s.by, name) + continue + } + if name != except { + out.Add(r.rates) + } + } + return out, s.seen } // ReplicaFeed is a standby's copy of the leader's Ready replicas. ok is false @@ -48,17 +113,44 @@ type ReplicaFeed interface { func (c *Controller) SetReplicaFeed(f ReplicaFeed) { c.feed = f } // Routing is this control plane's Ready replicas, for a standby to route to. -func (c *Controller) Routing() Routing { +func (c *Controller) Routing() Routing { return c.RoutingFor("") } + +// RoutingFor is Routing for the standby named from, with what every other +// router has counted against each limit. +func (c *Controller) RoutingFor(from string) Routing { + out := Routing{Replicas: c.ready(), Catalog: c.CatalogRev()} + if rs, ok := c.sink.(RateSharer); ok { + others, _ := c.rates.sum(from, time.Now()) + others.Add(rs.Rates()) + out.Rates = &others + } + return out +} + +// ready is the reconciler's Ready replicas, as a router needs them. +func (c *Controller) ready() []RoutedReplica { reps := c.rec.ReadyReplicas() - out := Routing{Replicas: make([]RoutedReplica, 0, len(reps)), Catalog: c.CatalogRev()} + out := make([]RoutedReplica, 0, len(reps)) for _, r := range reps { - out.Replicas = append(out.Replicas, RoutedReplica{ + out = append(out, RoutedReplica{ ID: r.ID, Model: r.Model, Profile: r.Profile, Pool: r.Pool, Endpoint: r.Endpoint, Weight: r.Weight, }) } return out } +// shareRates puts the standbys' counts in force on this control plane's own +// router, while it leads. Once a tick, beside publishing. +func (c *Controller) shareRates() { + rs, ok := c.sink.(RateSharer) + if !ok || (c.lead != nil && !c.lead.IsLeader()) { + return + } + if others, seen := c.rates.sum("", time.Now()); seen { + rs.SetOtherRates(others) + } +} + // CatalogRev names the catalog in force here. The process's owner id with a // counter: a new leader, or a restarted one, is a different catalog as far as // any standby knows. @@ -72,6 +164,9 @@ func (c *Controller) CatalogRev() string { // RecordActivity takes in what a standby's router did since its last report. func (c *Controller) RecordActivity(a Activity) { + if a.From != "" && a.Rates != nil { + c.rates.put(a.From, *a.Rates, time.Now()) + } c.rec.RecordUsage(a.LastUsed) for _, m := range a.Want { c.rec.Want(m) @@ -87,5 +182,5 @@ func (c *Controller) routable() []RoutedReplica { } return nil } - return c.Routing().Replicas + return c.ready() } diff --git a/internal/control/routing_test.go b/internal/control/routing_test.go index f3f9641..bb0a936 100644 --- a/internal/control/routing_test.go +++ b/internal/control/routing_test.go @@ -3,7 +3,9 @@ package control import ( "log/slog" "testing" + "time" + "gitea.mynodes.xyz/ilker/llmgw/internal/accounting" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" ) @@ -90,3 +92,52 @@ func TestTheCatalogRevisionMovesWithTheCatalog(t *testing.T) { t.Errorf("the revision stayed %q", after) } } + +// rateSink is a router whose counts can be shared: its own, and what it was +// last told the others counted. +type rateSink struct { + recordSink + own accounting.Rates + others *accounting.Rates +} + +func (s *rateSink) Rates() accounting.Rates { return s.own } +func (s *rateSink) SetOtherRates(r accounting.Rates) { + s.others = &r +} + +func counts(k string, n int64) accounting.Rates { + r := accounting.NewRates(time.Now()) + r.Requests.Current = map[string]int64{k: n} + return r +} + +// Each standby is given what the leader and the other standbys counted, never +// its own; the leader puts the standbys' counts in force on its own router. +func TestLimitsAreCountedAcrossTheControlPlanes(t *testing.T) { + sink := &rateSink{own: counts("key-1", 5)} + cat, _ := catalog.New(nil) + c := New(DefaultConfig(), catalog.NewLive(cat), noDispatch{}, sink, slog.New(slog.DiscardHandler)) + + b, d := counts("key-1", 3), counts("key-1", 7) + c.RecordActivity(Activity{From: "ctl-b", Rates: &b}) + c.RecordActivity(Activity{From: "ctl-d", Rates: &d}) + + if got := c.RoutingFor("ctl-b").Rates.Requests.Current["key-1"]; got != 12 { + t.Errorf("ctl-b was told %d, want the leader's 5 and ctl-d's 7", got) + } + c.shareRates() + if sink.others == nil || sink.others.Requests.Current["key-1"] != 10 { + t.Errorf("the leader's router was told %+v, want the standbys' 10", sink.others) + } + + // A standby that stopped reporting stops counting. + c.rates.mu.Lock() + r := c.rates.by["ctl-d"] + r.at = r.at.Add(-ratesFresh - time.Second) + c.rates.by["ctl-d"] = r + c.rates.mu.Unlock() + if got := c.RoutingFor("ctl-b").Rates.Requests.Current["key-1"]; got != 5 { + t.Errorf("a quiet standby still counts: %d", got) + } +} diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index 4f97dfd..c109d05 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -406,7 +406,7 @@ var Reference = []Block{ {Key: "batch", Type: "Batch", Doc: "Batch is the Files and Batch APIs (/v1/files, /v1/batches): a file of\nrequests served at the lowest priority, its answers written to\nanother. Needs the store.", Nested: "config.Batch"}, {Key: "cache", Type: "Cache", Doc: "Cache bounds the answers kept for the projects that reuse them\n(llmgw keys cache). Which projects do, and for how long, is theirs\nto set; this is only how much memory it may take.", Nested: "config.Cache"}, {Key: "trusted_proxies", Type: "[]string", Doc: "TrustedProxies are the reverse proxies in front of the client port,\nas addresses or CIDR prefixes. Their X-Forwarded-For, -Proto and -Host\nare believed; anyone else's are not, since they decide the address\nwritten into the setup scripts and whose attempts are counted where.", Nested: ""}, - {Key: "active_active", Type: "bool", Doc: "ActiveActive has standby control planes answer the client API\nthemselves, routing to the leader's replicas, rather than passing\nevery request to the leader. More machines carry the traffic, which\nis the point; what each router keeps in memory is then its own:\nrate limits are counted per control plane, so a key may make up to\nas many times its limit as there are control planes taking its\nrequests; a budget is checked against the store every half minute;\ncached answers and prefix affinity are split between them. Off by\ndefault. Which machine a client reaches is the load balancer's or\nDNS's to say; the floating address stays on the leader.", Nested: ""}, + {Key: "active_active", Type: "bool", Doc: "ActiveActive has standby control planes answer the client API\nthemselves, routing to the leader's replicas, rather than passing\nevery request to the leader. More machines carry the traffic, which\nis the point. Rate limits are the cluster's: each router's counts\nreach the others through the leader every two seconds, so a key\nspread over N control planes may pass its limit by what they allow\nin those two seconds. A budget is checked against the store every\nhalf minute; cached answers and prefix affinity are split between\nthe routers. Off by\ndefault. Which machine a client reaches is the load balancer's or\nDNS's to say; the floating address stays on the leader.", Nested: ""}, {Key: "timeouts", Type: "Timeouts", Doc: "", Nested: "config.Timeouts"}, {Key: "scoring", Type: "Scoring", Doc: "", Nested: "config.Scoring"}, {Key: "admission", Type: "Admission", Doc: "", Nested: "config.Admission"}, diff --git a/internal/router/router.go b/internal/router/router.go index eebaf73..cf60adf 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -747,6 +747,23 @@ func (r *Router) ShareBudgets(ctx context.Context) { } } +// Rates are this router's own request and token counts, for the other +// control planes taking requests to add to theirs. +func (r *Router) Rates() accounting.Rates { + if r.acct == nil { + return accounting.Rates{} + } + return r.acct.Rates() +} + +// SetOtherRates puts what the other control planes counted in force here, so +// a key's limits are the cluster's. +func (r *Router) SetOtherRates(o accounting.Rates) { + if r.acct != nil { + r.acct.SetOtherRates(o) + } +} + // SetLeader tells /leader how to know whether this process leads. func (r *Router) SetLeader(f func() bool) { r.isLeader.Store(&f) }