Standbys that answer the client API themselves, behind router.active_active
CI / test (push) Successful in 9m35s

A standby follows the leader's Ready replicas over the agent port, every
two seconds, and reports back the replicas its router used and the
on-demand models it was asked for. No engine keys travel: every control
plane derives them from the cluster secret. Budgets are raised to the
store's every 30 s. Unheard from the leader for a minute, a standby
passes /v1 on as before. Off by default.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ilker ManapandClaude Opus 5.5 committed 2026-11-01 13:47:47 +01:00
1 parent f664592948
commit 8b7eefef16
22 files changed
+936 -15

No files matched your search

+17 -5
View File
@@ -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.
+10
View File
@@ -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 <model> --max-output N` (and `max_output` on `POST
/api/v1/models/{name}/context`, and the console's Context length… form)
+133
View File
@@ -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.",
+16
View File
@@ -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`
+33 -5
View File
@@ -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
+16
View File
@@ -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)
+13
View File
@@ -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 {
+3 -2
View File
@@ -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.
+14 -1
View File
@@ -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.
+8 -1
View File
@@ -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()
+15
View File
@@ -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 {
+240
View File
@@ -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
}
+153
View File
@@ -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)
}
}
}
+12
View File
@@ -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"`
+1
View File
@@ -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)
+10
View File
@@ -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{}},
+27
View File
@@ -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())
})
}
+10 -1
View File
@@ -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 {
+91
View File
@@ -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
}
+92
View File
@@ -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)
}
}
+1
View File
@@ -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"},
+21
View File
@@ -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) }