diff --git a/BACKLOG.md b/BACKLOG.md index de71367..24a8e64 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -1944,11 +1944,23 @@ one control plane's router can. In order: the PII patterns (~12%). Ways on, a guardrail decision: judge shadow output in larger windows, or a sample of it. Next: the same run on a machine the size of a customer's control plane, then O2. - O2. Every control plane answers `/v1` itself (active-active): - - the leader's replica table followed live by the standbys, so their routers route; - - rate limits and budgets kept cluster-wide, not per process (J6), or three routers allow three times as much; - - a session's requests to one router and one replica (consistent hashing on the session), so the engines' prefix - caches keep paying; - - in front: IPVS on the keepalived the setup already installs, DNS round-robin, or the customer's balancer (J7, L9). + - [x] the leader's replica table followed live by the standbys, so their routers route (2026-11-01, behind + `router.active_active`, off by default). Every 2 s a standby posts what its router used to the leader's + `POST /api/v1/routing/activity` over the agent port with its control-plane certificate, and is answered with + the Ready replicas: no keys travel, each control plane derives a replica's key from the cluster secret. The + 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). + - [ ] 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 + (J7, L9). Until then the standbys' own addresses are what spread the load. + - [ ] tried on the PoC: 140 and 150 both taking `/v1`, the O1 load split between them. + - Known and accepted for now, per router: the precise KV index is empty on a standby (agents stream to the leader + only), a revoked key works on a standby until `auth.cache_ttl`, canary promotion sees the leader's share only, + the answer cache and prefix tree are split, an MCP caller's rate is per router. - O3. Only if O2 is not enough: stateless gateway machines (`llmgw server --role gateway`) in front of the leader's control plane, as many as the traffic needs. diff --git a/CHANGELOG.md b/CHANGELOG.md index 07bd6a5..930049b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,16 @@ APIs change between minors without a deprecation period. ### Added +- **Standbys that answer the client API themselves.** With + `router.active_active: true` in the cluster's settings, a standby control + plane routes `/v1` to the leader's replicas instead of passing every + request to the leader, so more machines carry the traffic. It hears the + 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. - **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 161f71c..3cf0d95 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -110,6 +110,9 @@ { "name": "restart" }, + { + "name": "routing" + }, { "name": "settings" }, @@ -4793,6 +4796,77 @@ } } }, + "/api/v1/routing": { + "get": { + "operationId": "getRouting", + "summary": "The Ready replicas a standby routes to", + "description": "Read by a standby control plane when router.active_active is on: each replica's address, model, profile, pool and weight, without its key, and which catalog is in force.", + "tags": [ + "routing" + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/control.Routing" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, + "/api/v1/routing/activity": { + "post": { + "operationId": "postRoutingActivity", + "summary": "A standby's router reports what it used", + "description": "When it last sent each replica a request, and the on-demand models it was asked for, so the leader does not evict a replica busy behind a standby. Answered with the routing, so one request a tick does both.", + "tags": [ + "routing" + ], + "parameters": [ + { + "$ref": "#/components/parameters/Actor" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/control.Activity" + } + } + } + }, + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/control.Routing" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, "/api/v1/settings": { "get": { "operationId": "getSettings", @@ -10554,6 +10628,65 @@ } } }, + "control.Activity": { + "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": { + "last_used": { + "type": "object", + "additionalProperties": { + "type": "string", + "format": "date-time" + } + }, + "want": { + "type": "array", + "items": { + "type": "string" + } + } + } + }, + "control.RoutedReplica": { + "type": "object", + "description": "RoutedReplica is a Ready replica as a router needs it: where it is, what it serves and how much it may take. Not its key — every control plane derives that from the replica's id and the cluster's secret, so none travels.", + "properties": { + "id": { + "type": "string" + }, + "model": { + "type": "string" + }, + "profile": { + "type": "string" + }, + "pool": { + "type": "string" + }, + "endpoint": { + "type": "string" + }, + "weight": { + "type": "number" + } + } + }, + "control.Routing": { + "type": "object", + "description": "Routing is what a standby that answers requests itself reads from the leader: the replicas it may send to, and which catalog they belong to.", + "properties": { + "replicas": { + "type": "array", + "items": { + "$ref": "#/components/schemas/control.RoutedReplica" + } + }, + "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." + } + } + }, "eval.Live": { "type": "object", "description": "Live is a run as it stands: one running, with how many of its cases are done, or one finished.", diff --git a/docs/reference.md b/docs/reference.md index 47351a3..5da77a0 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -320,6 +320,7 @@ rather than repeated. | `batch` | `Batch` | [Router](#router) | | `cache` | `Cache` | [Router](#router) | | `trusted_proxies` | `[]string` | [Router](#router) | +| `active_active` | `bool` | [Router](#router) | | `timeouts` | `Timeouts` | [Router](#router) | | `scoring` | `Scoring` | [Router](#router) | | `admission` | `Admission` | [Router](#router) | @@ -2810,6 +2811,21 @@ as addresses or CIDR prefixes. Their X-Forwarded-For, -Proto and -Host are believed; anyone else's are not, since they decide the address written into the setup scripts and whose attempts are counted where. +### `active_active` + +`bool` + +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 +default. Which machine a client reaches is the load balancer's or +DNS's to say; the floating address stays on the leader. + ### `timeouts` `Timeouts` diff --git a/internal/accounting/accounting.go b/internal/accounting/accounting.go index 2c26add..27e4c08 100644 --- a/internal/accounting/accounting.go +++ b/internal/accounting/accounting.go @@ -78,10 +78,41 @@ func (a *Accountant) RestoreBudgets(ctx context.Context, s store.Store, period s if start.IsZero() { return nil } - rows, err := s.SpendSinceByScope(ctx, start) + spend, keys, err := storedSpend(ctx, s, start) if err != nil { return fmt.Errorf("restoring budgets: %w", err) } + a.budgets.Restore(period, spend) + a.log.Info("restored budgets from the store", + "keys", keys, "counters", len(spend), "since", start) + return nil +} + +// CatchUpBudgets raises each counter to what the store has recorded, where +// that is more. Run every so often by a router that is one of several taking +// requests: the others' spend reaches it only through the store. Never +// lowers a counter, which leads the store by whatever this process has not +// written yet. +func (a *Accountant) CatchUpBudgets(ctx context.Context, s store.Store, period store.Period) error { + start := PeriodStart(period, time.Now()) + if start.IsZero() { + return nil + } + spend, _, err := storedSpend(ctx, s, start) + if err != nil { + return fmt.Errorf("reading budgets: %w", err) + } + a.budgets.Raise(period, spend) + return nil +} + +// storedSpend is the spend since start the store has recorded, by counter, +// and how many keys it came from. +func storedSpend(ctx context.Context, s store.Store, start time.Time) (map[string]int64, int, error) { + rows, err := s.SpendSinceByScope(ctx, start) + if err != nil { + return nil, 0, err + } // Every scope a key sits under, not just the key. A budget owned by a team // is one counter fed by every key beneath it, and restoring only per-key // totals would bring the control plane back up believing the team had spent @@ -113,10 +144,7 @@ func (a *Accountant) RestoreBudgets(ctx context.Context, s store.Store, period s } } } - a.budgets.Restore(period, spend) - a.log.Info("restored budgets from the store", - "keys", len(rows), "counters", len(spend), "since", start) - return nil + return spend, len(rows), nil } // Reservation is what Check returns: the estimate it charged, so Record can diff --git a/internal/accounting/accounting_test.go b/internal/accounting/accounting_test.go index ceb96f5..c1ef1d1 100644 --- a/internal/accounting/accounting_test.go +++ b/internal/accounting/accounting_test.go @@ -216,6 +216,22 @@ func TestBudgetSurvivesARestart(t *testing.T) { } } +func TestCatchingUpRaisesButNeverLowers(t *testing.T) { + // Another router's spend arrives through the store; this one's own, not + // written yet, must not be wiped by a store that is behind. + b := NewBudgets() + b.Add("mine", store.PeriodMonth, 900) + b.Add("shared", store.PeriodMonth, 100) + + b.Raise(store.PeriodMonth, map[string]int64{"mine": 400, "shared": 700, "theirs": 50}) + + for key, want := range map[string]int64{"mine": 900, "shared": 700, "theirs": 50} { + if got := b.Spent(key, store.PeriodMonth); got != want { + t.Errorf("%s: spent %d, want %d", key, got, want) + } + } +} + func TestBudgetsResetAtThePeriodBoundary(t *testing.T) { b := NewBudgets() now := time.Date(2026, 1, 31, 23, 59, 0, 0, time.UTC) diff --git a/internal/accounting/budget.go b/internal/accounting/budget.go index 0b47d79..f44d804 100644 --- a/internal/accounting/budget.go +++ b/internal/accounting/budget.go @@ -61,6 +61,19 @@ func (b *Budgets) Restore(period store.Period, spend map[string]int64) { } } +// Raise lifts counters to the spend given, where it is more. +func (b *Budgets) Raise(period store.Period, spend map[string]int64) { + start := PeriodStart(period, b.now()) + b.mu.Lock() + defer b.mu.Unlock() + for key, micros := range spend { + if c := b.currentLocked(key, start); micros > c.micros { + c.micros = micros + b.spend[key] = c + } + } +} + // Spent returns a key's spend in the current period, resetting it if the period // has rolled over. func (b *Budgets) Spent(keyID string, period store.Period) int64 { diff --git a/internal/audit/audit.go b/internal/audit/audit.go index 5229a42..1be99be 100644 --- a/internal/audit/audit.go +++ b/internal/audit/audit.go @@ -112,8 +112,9 @@ func (w *statusWriter) Flush() { } // notChanges are POSTs that change nothing and carry text that must not be -// kept: asking a judge about a sentence. -var notChanges = []string{"/api/v1/guard/judges/ask"} +// kept: asking a judge about a sentence. And a standby's report of what its +// router sent where, every two seconds: activity, not an operator's change. +var notChanges = []string{"/api/v1/guard/judges/ask", "/api/v1/routing/activity"} // API records every change to the admin API: a request that is not a GET or // HEAD, under /api/. actor says who made it. diff --git a/internal/cli/leaderforward.go b/internal/cli/leaderforward.go index a50f3a8..0372689 100644 --- a/internal/cli/leaderforward.go +++ b/internal/cli/leaderforward.go @@ -49,6 +49,9 @@ type leaderForward struct { self string tls tlsconf.Config log *slog.Logger + // routes, with router.active_active, is the leader's replicas this + // standby routes to; nil when every request goes to the leader. + routes *standbyRoutes } // standby wraps the admin handler: /api/ goes to the leader while this @@ -117,7 +120,8 @@ func (f *leaderForward) public(next http.Handler, port string) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // Once only: a request another control plane forwarded is answered // here, leading or not, rather than sent on while the lease moves. - if publicStaysHere(r.URL.Path) || f.lead.IsLeader() || r.Header.Get(hdrForwardedBy) != "" { + if publicStaysHere(r.URL.Path) || f.lead.IsLeader() || r.Header.Get(hdrForwardedBy) != "" || + f.answersHere(r.URL.Path) { next.ServeHTTP(w, r) return } @@ -154,6 +158,15 @@ func (f *leaderForward) public(next http.Handler, port string) http.Handler { }) } +// answersHere is whether a standby answers this client request itself: the +// API under /v1, with router.active_active, while it has the leader's +// replicas to send it to. Not the portal, its admin pages or /mcp, whose +// llmgw:admin tools read the cluster from memory only the leader has; and +// not while the leader has gone unheard, when it is passed on as before. +func (f *leaderForward) answersHere(path string) bool { + return f.routes != nil && strings.HasPrefix(path, "/v1/") && f.routes.serving() +} + // trustControlPlanes keeps the client port trusting the other control // planes' X-Forwarded-For, besides router.trusted_proxies: a standby passes // requests on, and the client is the address it was given, not the standby. diff --git a/internal/cli/leaderforward_test.go b/internal/cli/leaderforward_test.go index 70e3fbd..ab02f36 100644 --- a/internal/cli/leaderforward_test.go +++ b/internal/cli/leaderforward_test.go @@ -32,7 +32,11 @@ type forwardCluster struct { leaderURL string } -func newForwardCluster(t *testing.T) *forwardCluster { +func newForwardCluster(t *testing.T) *forwardCluster { return newForwardClusterWith(t, nil) } + +// newForwardClusterWith answers forwarded requests on the leader with admin, +// or, given nil, with one that says what it was asked. +func newForwardClusterWith(t *testing.T, leaderAdmin http.Handler) *forwardCluster { t.Helper() dir := t.TempDir() ca, err := tlsconf.NewCA("test CA") @@ -81,6 +85,9 @@ func newForwardCluster(t *testing.T) *forwardCluster { "auth": r.Header.Get("Authorization"), "actor": r.Header.Get("X-Llmgw-Actor")}) }) mux := http.NewServeMux() + if leaderAdmin != nil { + admin = leaderAdmin.ServeHTTP + } mux.Handle(forwardPrefix+"/", c.leader.leaderSide(admin)) srv := httptest.NewUnstartedServer(mux) tc, err := cfgFor("ctl-a").ServerConfig() diff --git a/internal/cli/server.go b/internal/cli/server.go index 511bb01..5e02e35 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -691,6 +691,21 @@ func newServerCmd() *cobra.Command { if admin != nil && admin.Leadership() != nil && rt.Store() != nil && cfg.AgentTLS.Enabled() { self, _ := os.Hostname() forward = &leaderForward{st: rt.Store(), lead: admin.Leadership(), self: self, tls: cfg.AgentTLS, log: log} + // And with router.active_active, a standby answers the client + // API itself, to the leader's replicas. + if cfg.Router.ActiveActive { + var refresh func(context.Context) error + if catalogSvc != nil { + refresh = catalogSvc.Refresh + } + routes := newStandbyRoutes(forward, rt, refresh, admin.Publish, log) + forward.routes = routes + admin.SetReplicaFeed(routes) + rt.SetActivator(standbyActivator{Reconciler: admin.Reconciler(), routes: routes}) + go routes.run(ctx) + go rt.ShareBudgets(ctx) + log.Info("standbys answer the client API themselves (router.active_active)") + } } served := adminHandler if forward != nil { diff --git a/internal/cli/standbyroutes.go b/internal/cli/standbyroutes.go new file mode 100644 index 0000000..8216b4b --- /dev/null +++ b/internal/cli/standbyroutes.go @@ -0,0 +1,240 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "sync" + "time" + + "golang.org/x/net/http2" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" +) + +// How often a standby answering requests itself hears from the leader, and +// how long it keeps routing to the replicas it last heard of when it does +// not. Long enough to ride out the leader changing hands — the engines are +// still where they were, and a dead one opens its breaker — short enough +// that a standby cut off from the cluster stops sending to a list it cannot +// check. +const ( + standbyPoll = 2 * time.Second + standbyTrust = time.Minute +) + +// 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 +// request over the agent port with its control-plane certificate. +type standbyRoutes struct { + fwd *leaderForward + usage control.UsageSource + refresh func(context.Context) error // reads the catalog again from the store + publish func(context.Context) // hands the router the new list now + log *slog.Logger + now func() time.Time + + mu sync.Mutex + reps []control.RoutedReplica + heard time.Time + catalog string // the leader's catalog revision last seen + sent map[string]time.Time // the last use already reported, by replica + want map[string]bool // on-demand models asked for since the last report + kick chan struct{} + + // The connection to the leader, kept between ticks: one per leader + // address, the previous one's closed when the lease moves. + client *http.Client + clientHost string +} + +func newStandbyRoutes(fwd *leaderForward, usage control.UsageSource, refresh func(context.Context) error, + publish func(context.Context), log *slog.Logger, +) *standbyRoutes { + return &standbyRoutes{fwd: fwd, usage: usage, refresh: refresh, publish: publish, log: log, now: time.Now, + sent: map[string]time.Time{}, want: map[string]bool{}, kick: make(chan struct{}, 1)} +} + +// Ready satisfies control.ReplicaFeed. +func (s *standbyRoutes) Ready() ([]control.RoutedReplica, bool) { + s.mu.Lock() + defer s.mu.Unlock() + if s.heard.IsZero() || s.now().Sub(s.heard) > standbyTrust { + return nil, false + } + return s.reps, true +} + +// serving says whether this standby has replicas to send a request to, which +// is when it keeps the client API rather than passing it to the leader. +func (s *standbyRoutes) serving() bool { + reps, ok := s.Ready() + return ok && len(reps) > 0 +} + +// wanted queues an on-demand model for the leader, and reports at once +// rather than on the next tick: someone is waiting on it. +func (s *standbyRoutes) wanted(model string) { + s.mu.Lock() + s.want[model] = true + s.mu.Unlock() + select { + case s.kick <- struct{}{}: + default: + } +} + +func (s *standbyRoutes) run(ctx context.Context) { + t := time.NewTicker(standbyPoll) + defer t.Stop() + for { + if s.fwd.lead.IsLeader() { + // Its own reconciler's replicas now. What it heard as a standby + // is forgotten, so becoming one again starts afresh. + s.mu.Lock() + s.reps, s.heard, s.catalog = nil, time.Time{}, "" + clear(s.sent) + clear(s.want) + s.mu.Unlock() + } else if err := s.exchange(ctx); err != nil { + s.log.Debug("could not hear from the leader", "err", err) + } + select { + case <-ctx.Done(): + return + case <-t.C: + case <-s.kick: + } + } +} + +// exchange reports activity and takes the replicas in return. +func (s *standbyRoutes) exchange(ctx context.Context) error { + act, sent := s.activity() + body, err := json.Marshal(act) + if err != nil { + return err + } + target, name, err := s.fwd.leader(ctx) + if err != nil { + return err + } + client, err := s.clientFor(target.Host, target.Hostname()) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + "https://"+target.Host+forwardPrefix+"/api/v1/routing/activity", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set(hdrForwardedBy, s.fwd.self) + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("%s answered %s: %s", name, resp.Status, bytes.TrimSpace(b)) + } + var rt control.Routing + if err := json.NewDecoder(resp.Body).Decode(&rt); err != nil { + return fmt.Errorf("%s's routing: %w", name, err) + } + + s.mu.Lock() + s.reps, s.heard = rt.Replicas, s.now() + stale := rt.Catalog != s.catalog + s.catalog = rt.Catalog + for id, at := range sent { + s.sent[id] = at + } + for _, m := range act.Want { + delete(s.want, m) + } + s.mu.Unlock() + + // 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 { + if err := s.refresh(ctx); err != nil { + s.log.Warn("could not read the catalog the leader changed", "err", err) + s.mu.Lock() + s.catalog = "" // tried again next tick + s.mu.Unlock() + } + } + if s.publish != nil { + s.publish(ctx) + } + return nil +} + +// clientFor is the client for the leader at host, made once per leader. +// Only the run loop calls it. +func (s *standbyRoutes) clientFor(host, name string) (*http.Client, error) { + if s.client != nil && s.clientHost == host { + return s.client, nil + } + tc, err := s.fwd.tls.ClientConfig(name) + if err != nil { + return nil, err + } + if s.client != nil { + s.client.CloseIdleConnections() + } + s.client = &http.Client{Transport: &http2.Transport{TLSClientConfig: tc, ReadIdleTimeout: 30 * time.Second}} + s.clientHost = host + return s.client, nil +} + +// activity is what to report: uses newer than the last reported, and the +// models asked for. sent is what to record as reported once the leader has +// it. +func (s *standbyRoutes) activity() (control.Activity, map[string]time.Time) { + var used map[string]time.Time + if s.usage != nil { + used = s.usage.LastUsed() + } + s.mu.Lock() + defer s.mu.Unlock() + act := control.Activity{LastUsed: map[string]time.Time{}} + for id, at := range used { + if at.After(s.sent[id]) { + act.LastUsed[id] = at + } + } + for m := range s.want { + act.Want = append(act.Want, m) + } + return act, act.LastUsed +} + +// standbyActivator asks the leader for an on-demand model when a request for +// it reaches a standby: the standby's own reconciler places nothing. Which +// models load on demand, and why one cannot, are still the reconciler's to +// answer from the catalog. +type standbyActivator struct { + *reconciler.Reconciler + routes *standbyRoutes +} + +func (a standbyActivator) Want(model string) bool { + if !a.Reconciler.Want(model) { + return false + } + if !a.routes.fwd.lead.IsLeader() { + a.routes.wanted(model) + } + return true +} diff --git a/internal/cli/standbyroutes_test.go b/internal/cli/standbyroutes_test.go new file mode 100644 index 0000000..3c3ef32 --- /dev/null +++ b/internal/cli/standbyroutes_test.go @@ -0,0 +1,153 @@ +package cli + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "slices" + "strings" + "sync" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control" +) + +// fakeLeaderRouting is the leader's routing API: what it was told, and the +// replicas and catalog it answers with. +type fakeLeaderRouting struct { + mu sync.Mutex + got []control.Activity + routing control.Routing +} + +func (f *fakeLeaderRouting) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/api/v1/routing/activity" { + http.NotFound(w, r) + return + } + var act control.Activity + if err := json.NewDecoder(r.Body).Decode(&act); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + f.mu.Lock() + defer f.mu.Unlock() + f.got = append(f.got, act) + _ = json.NewEncoder(w).Encode(f.routing) +} + +func (f *fakeLeaderRouting) last() control.Activity { + f.mu.Lock() + defer f.mu.Unlock() + return f.got[len(f.got)-1] +} + +type fixedUsage map[string]time.Time + +func (u fixedUsage) LastUsed() map[string]time.Time { return u } + +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"}}}} + c := newForwardClusterWith(t, leader) + used := time.Date(2026, 11, 1, 12, 0, 0, 0, time.UTC) + refreshed := 0 + s := newStandbyRoutes(c.standby, fixedUsage{"llama-r0": used}, func(_ context.Context) error { + refreshed++ + return nil + }, nil, slog.New(slog.DiscardHandler)) + + if _, ok := s.Ready(); ok { + t.Fatal("ready before hearing from the leader") + } + if err := s.exchange(t.Context()); err != nil { + t.Fatal(err) + } + if reps, ok := s.Ready(); !ok || len(reps) != 1 || reps[0].ID != "llama-r0" { + t.Fatalf("replicas %+v, %v", reps, ok) + } + if got := leader.last().LastUsed["llama-r0"]; !got.Equal(used) { + t.Errorf("the leader was told the replica was last used %v", got) + } + if refreshed != 1 { + t.Errorf("the catalog was read %d times on first hearing from the leader", refreshed) + } + + // Only what is new is reported, and the catalog is read again only when + // the leader's changed. + s.wanted("flux") + if err := s.exchange(t.Context()); err != nil { + t.Fatal(err) + } + if act := leader.last(); len(act.LastUsed) != 0 || !slices.Equal(act.Want, []string{"flux"}) { + t.Errorf("second report %+v", act) + } + if refreshed != 1 { + t.Errorf("an unchanged catalog was read again") + } + leader.mu.Lock() + leader.routing.Catalog = "a/2" + leader.mu.Unlock() + if err := s.exchange(t.Context()); err != nil { + t.Fatal(err) + } + if act := leader.last(); len(act.Want) != 0 { + t.Errorf("an on-demand model reported twice: %+v", act) + } + if refreshed != 2 { + t.Errorf("the changed catalog was not read: %d", refreshed) + } + + // Unheard for longer than it is trusted, it routes nothing, and requests + // go to the leader again. + c.standby.routes = s + if !c.standby.answersHere("/v1/chat/completions") { + t.Error("a standby with replicas passed the API on") + } + s.now = func() time.Time { return time.Now().Add(standbyTrust + time.Second) } + if _, ok := s.Ready(); ok || c.standby.answersHere("/v1/chat/completions") { + t.Error("a standby the leader has not answered for a minute still routes") + } +} + +// The client API stays on a standby with replicas; the portal, its admin +// pages and /mcp still go to the leader. +func TestAStandbyAnswersOnlyTheAPIItself(t *testing.T) { + leader := &fakeLeaderRouting{routing: control.Routing{ + Replicas: []control.RoutedReplica{{ID: "llama-r0", Model: "llama", Endpoint: "http://10.0.0.5:20001"}}}} + c := newForwardClusterWith(t, leader) + s := newStandbyRoutes(c.standby, nil, nil, nil, slog.New(slog.DiscardHandler)) + if err := s.exchange(t.Context()); err != nil { + t.Fatal(err) + } + c.standby.routes = s + + port := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("leader " + r.URL.Path)) + })) + defer port.Close() + _, p, _ := net.SplitHostPort(strings.TrimPrefix(port.URL, "http://")) + local := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("standby " + r.URL.Path)) }) + srv := httptest.NewServer(c.standby.public(local, p)) + defer srv.Close() + + for path, want := range map[string]string{ + "/v1/chat/completions": "standby", "/v1/models": "standby", + "/admin/replicas": "leader", "/mcp": "leader", "/": "leader", + } { + resp, err := http.Get(srv.URL + path) //nolint:noctx // a test + if err != nil { + t.Fatal(err) + } + b, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if got, _, _ := strings.Cut(string(b), " "); got != want { + t.Errorf("%s answered by %q, want the %s", path, b, want) + } + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 506feed..19c3276 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -641,6 +641,18 @@ type Router struct { // written into the setup scripts and whose attempts are counted where. TrustedProxies []string `koanf:"trusted_proxies"` + // 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 + // 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"` + Timeouts Timeouts `koanf:"timeouts"` Scoring Scoring `koanf:"scoring"` Admission Admission `koanf:"admission"` diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 9bb87b5..638271d 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -182,6 +182,7 @@ func (a *API) Register(mux Mux) { a.registerPresets(mux) mux.HandleFunc("GET /api/v1/models", a.listModels) mux.HandleFunc("GET /api/v1/replicas", a.listReplicas) + a.registerRouting(mux) // Listing is a read and deleting is a separate verb, so a mistyped // command cannot remove anything. mux.HandleFunc("GET /api/v1/weights/orphans", a.listOrphans) diff --git a/internal/control/adminapi/routes.go b/internal/control/adminapi/routes.go index db0a7ec..225815d 100644 --- a/internal/control/adminapi/routes.go +++ b/internal/control/adminapi/routes.go @@ -6,6 +6,7 @@ import ( llmgw "gitea.mynodes.xyz/ilker/llmgw" "gitea.mynodes.xyz/ilker/llmgw/internal/bench" "gitea.mynodes.xyz/ilker/llmgw/internal/bench/eval" + "gitea.mynodes.xyz/ilker/llmgw/internal/control" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrics" "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" @@ -194,6 +195,15 @@ var Routes = []Route{ {Method: "GET", Path: "/api/v1/replicas", Summary: "Every replica, with its shards", Query: []Param{{Name: "model", Description: "Only this model's replicas."}}, Response: []ReplicaView{}}, + {Method: "GET", Path: "/api/v1/routing", Summary: "The Ready replicas a standby routes to", + Description: "Read by a standby control plane when router.active_active is on: each replica's address, " + + "model, profile, pool and weight, without its key, and which catalog is in force.", + Response: control.Routing{}}, + {Method: "POST", Path: "/api/v1/routing/activity", Summary: "A standby's router reports what it used", + Description: "When it last sent each replica a request, and the on-demand models it was asked for, " + + "so the leader does not evict a replica busy behind a standby. Answered with the routing, " + + "so one request a tick does both.", + Request: control.Activity{}, Response: control.Routing{}}, {Method: "GET", Path: "/api/v1/explain-placement", Summary: "Why a model could not be placed", Query: []Param{{Name: "model", Required: true, Description: "The model's name or an alias."}}, Response: ExplainView{}}, diff --git a/internal/control/adminapi/routing.go b/internal/control/adminapi/routing.go new file mode 100644 index 0000000..cef008f --- /dev/null +++ b/internal/control/adminapi/routing.go @@ -0,0 +1,27 @@ +package adminapi + +import ( + "encoding/json" + "io" + "net/http" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control" +) + +// registerRouting is what a standby answering requests itself reads from the +// leader, and reports back to it. Reached by the standby through the agent +// port with its control-plane certificate, like the rest of the forwarded API. +func (a *API) registerRouting(mux Mux) { + mux.HandleFunc("GET /api/v1/routing", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, a.ctl.Routing()) + }) + mux.HandleFunc("POST /api/v1/routing/activity", func(w http.ResponseWriter, r *http.Request) { + var act control.Activity + if err := json.NewDecoder(io.LimitReader(r.Body, 8<<20)).Decode(&act); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + a.ctl.RecordActivity(act) + writeJSON(w, http.StatusOK, a.ctl.Routing()) + }) +} diff --git a/internal/control/control.go b/internal/control/control.go index 96b3584..1b3f8dc 100644 --- a/internal/control/control.go +++ b/internal/control/control.go @@ -12,6 +12,7 @@ import ( "os" "strings" "sync" + "sync/atomic" "time" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" @@ -95,6 +96,13 @@ type Controller struct { mu sync.Mutex last string // fingerprint of the last published list + // feed is where a standby's router learns the leader's Ready replicas, + // when standbys answer requests themselves; nil when they do not. + feed ReplicaFeed + // catRev changes with every catalog put in force here, so a standby + // following this one knows to read the catalog again. + catRev atomic.Uint64 + stop chan struct{} done chan struct{} once sync.Once @@ -193,6 +201,7 @@ type LoadSource interface { // killed. func (c *Controller) ReplaceCatalog(ctx context.Context, next *catalog.Catalog) { c.cat.Store(next) + c.catRev.Add(1) if cs, ok := c.sink.(CatalogSink); ok { cs.SetCatalog(next) } @@ -345,7 +354,7 @@ func (c *Controller) Publish(ctx context.Context) { // endpoints converts Ready replicas into the router's view. func (c *Controller) endpoints() []endpoint.Info { - ready := c.rec.ReadyReplicas() + ready := c.routable() out := make([]endpoint.Info, 0, len(ready)) for _, rep := range ready { diff --git a/internal/control/routing.go b/internal/control/routing.go new file mode 100644 index 0000000..a04c4fe --- /dev/null +++ b/internal/control/routing.go @@ -0,0 +1,91 @@ +package control + +import ( + "strconv" + "time" +) + +// RoutedReplica is a Ready replica as a router needs it: where it is, what it +// serves and how much it may take. Not its key — every control plane derives +// that from the replica's id and the cluster's secret, so none travels. +type RoutedReplica struct { + ID string `json:"id"` + Model string `json:"model"` + Profile string `json:"profile"` + Pool string `json:"pool"` + Endpoint string `json:"endpoint"` + Weight float64 `json:"weight"` +} + +// Routing is what a standby that answers requests itself reads from the +// leader: the replicas it may send to, and which catalog they belong to. +type Routing struct { + Replicas []RoutedReplica `json:"replicas"` + // 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. + Catalog string `json:"catalog"` +} + +// 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. +type Activity struct { + LastUsed map[string]time.Time `json:"last_used,omitempty"` + Want []string `json:"want,omitempty"` +} + +// ReplicaFeed is a standby's copy of the leader's Ready replicas. ok is false +// when it has none worth using — standbys do not answer requests, or the +// leader has not been heard from lately — and the standby routes nothing. +type ReplicaFeed interface { + Ready() (reps []RoutedReplica, ok bool) +} + +// SetReplicaFeed lets a standby route to the leader's replicas. +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 { + reps := c.rec.ReadyReplicas() + out := Routing{Replicas: make([]RoutedReplica, 0, len(reps)), Catalog: c.CatalogRev()} + for _, r := range reps { + out.Replicas = append(out.Replicas, RoutedReplica{ + ID: r.ID, Model: r.Model, Profile: r.Profile, Pool: r.Pool, Endpoint: r.Endpoint, Weight: r.Weight, + }) + } + return out +} + +// 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. +func (c *Controller) CatalogRev() string { + owner := "" + if c.lead != nil { + owner = c.lead.Owner() + } + return owner + "/" + strconv.FormatUint(c.catRev.Load(), 10) +} + +// RecordActivity takes in what a standby's router did since its last report. +func (c *Controller) RecordActivity(a Activity) { + c.rec.RecordUsage(a.LastUsed) + for _, m := range a.Want { + c.rec.Want(m) + } +} + +// routable is what this process's router may send to: the replicas the +// reconciler holds Ready, or, on a standby following the leader, the leader's. +func (c *Controller) routable() []RoutedReplica { + if c.feed != nil && c.lead != nil && !c.lead.IsLeader() { + if reps, ok := c.feed.Ready(); ok { + return reps + } + return nil + } + return c.Routing().Replicas +} diff --git a/internal/control/routing_test.go b/internal/control/routing_test.go new file mode 100644 index 0000000..f3f9641 --- /dev/null +++ b/internal/control/routing_test.go @@ -0,0 +1,92 @@ +package control + +import ( + "log/slog" + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +type fixedFeed struct { + reps []RoutedReplica + ok bool +} + +func (f fixedFeed) Ready() ([]RoutedReplica, bool) { return f.reps, f.ok } + +func routingController(t *testing.T, sink *recordSink) *Controller { + t.Helper() + cat, err := catalog.New([]catalog.Model{{Name: "llama", Source: "/models/llama", Aliases: []string{"ll"}, + Scaling: catalog.Scaling{Min: 1, Max: 1}, + Profiles: []catalog.Profile{{Name: "one", GPUClass: "h100-80g", Engine: "vllm", Priority: 3}}}}) + if err != nil { + t.Fatal(err) + } + cfg := DefaultConfig() + cfg.Reconciler.EngineKey = EngineKeys([]byte("secret")) + return New(cfg, catalog.NewLive(cat), noDispatch{}, sink, slog.New(slog.DiscardHandler)) +} + +// standby is a leadership that does not lead: one without a store always +// would. +func standby() *Leadership { + l := NewLeadership(nil, 0, nil) + l.leader.Store(false) + return l +} + +// A standby following the leader routes to the leader's replicas, each with +// the key it derives itself and the engine and priority its own catalog +// gives the profile; a replica of a model its catalog lacks is left out. +func TestAStandbyRoutesToTheLeadersReplicas(t *testing.T) { + sink := &recordSink{} + c := routingController(t, sink) + c.SetLeadership(standby()) + c.SetReplicaFeed(fixedFeed{ok: true, reps: []RoutedReplica{ + {ID: "llama-r0", Model: "llama", Profile: "one", Pool: "a", Endpoint: "http://10.0.0.5:20001/", Weight: 2}, + {ID: "gone-r0", Model: "gone", Profile: "x", Endpoint: "http://10.0.0.6:20001"}, + }}) + c.Publish(t.Context()) + + if len(sink.infos) != 1 { + t.Fatalf("published %+v", sink.infos) + } + e := sink.infos[0] + if e.ID != "llama-r0" || e.URL != "http://10.0.0.5:20001" || e.Pool != "a" || e.Weight != 2 || + e.Engine != "vllm" || e.Priority != 3 || len(e.Models) != 2 || e.Models[1] != "ll" { + t.Errorf("endpoint %+v", e) + } + if e.APIKey == "" || e.APIKey != EngineKeys([]byte("secret"))("llama-r0") { + t.Errorf("the key is not the one the leader gave the engine: %q", e.APIKey) + } +} + +// One that has not heard from the leader lately routes to nothing, and the +// leader never reads a feed: its replicas are its reconciler's. +func TestAStandbyCutOffRoutesNothingAndTheLeaderIgnoresTheFeed(t *testing.T) { + sink := &recordSink{} + c := routingController(t, sink) + l := standby() + c.SetLeadership(l) + c.SetReplicaFeed(fixedFeed{ok: false, reps: []RoutedReplica{{ID: "llama-r0", Model: "llama", Profile: "one"}}}) + if got := c.endpoints(); len(got) != 0 { + t.Errorf("a standby that has not heard from the leader routes to %+v", got) + } + + l.leader.Store(true) + c.SetReplicaFeed(fixedFeed{ok: true, reps: []RoutedReplica{{ID: "llama-r0", Model: "llama", Profile: "one"}}}) + if got := c.endpoints(); len(got) != 0 { + t.Errorf("the leader routed to a feed rather than its own replicas: %+v", got) + } +} + +// The catalog's revision moves with every catalog put in force, so a +// standby knows to read it again. +func TestTheCatalogRevisionMovesWithTheCatalog(t *testing.T) { + c := routingController(t, &recordSink{}) + before := c.Routing().Catalog + c.ReplaceCatalog(t.Context(), c.cat.Load()) + if after := c.Routing().Catalog; after == before { + t.Errorf("the revision stayed %q", after) + } +} diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index 19d5a75..4f97dfd 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -406,6 +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: "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 743094f..eebaf73 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -726,6 +726,27 @@ func (r *Router) Stop() { } } +// ShareBudgets keeps budgets honest while other control planes take requests +// too: every half minute each counter is raised to what the store recorded, +// which includes their spend. Until ctx ends. +func (r *Router) ShareBudgets(ctx context.Context) { + if r.acct == nil || r.store == nil { + return + } + t := time.NewTicker(30 * time.Second) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if err := r.acct.CatchUpBudgets(ctx, r.store, storepkg.PeriodMonth); err != nil { + r.log.Warn("could not read the budgets other control planes spent", "err", err) + } + } + } +} + // SetLeader tells /leader how to know whether this process leads. func (r *Router) SetLeader(f func() bool) { r.isLeader.Store(&f) }