From 9d8cd5e284c1ed495c8a8281a8c489468809bae6 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Thu, 17 Sep 2026 00:52:34 +0200 Subject: [PATCH] Track which replicas are actually being used, and summarise the fleet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two things a cluster with many nodes and many GPUs needs that were missing, and they are the same thing seen from two distances. Idle tracking. The control plane knew what was placed and whether it was healthy, never whether anyone wanted it — so on a fleet with more models than cards there was no way to answer the first question anyone asks, which is what can I stop. The router is the only thing that knows, so it now reports it: a timestamp written in Acquire, where a request already meets an endpoint under a lock already held, and collected once a tick as a map. That shape is the point. Per-request notification would be a firehose carrying one bit — the newest timestamp wins — and this costs the same on four replicas as on four hundred. It travels through an interface in the direction opposite EndpointSink, for the same reason that one exists: the control plane decides what should run, the router knows whether anyone wants it, and neither should need to be in the other's process to say so. Unknown is kept distinct from idle throughout. A router that has just restarted knows nothing, and a zero timestamp would be the oldest idle replica in every comparison — so one restart would nominate an entire fleet for eviction. Absent from the map, "-" in the table, negative in the JSON, and IdleFor returns a second value rather than a duration anything could act on by mistake. ctl status. A per-node listing is right for four nodes and wrong for two hundred: how full am I, what is broken, what is holding cards nobody is using are aggregates, and reading them off a long table is a chore, which means it gets skipped. Capacity per GPU class with usage bars, free whole nodes per fabric — the number that decides whether a multi-node replica can land, since free cards on the wrong fabric are not capacity for a model that spans nodes — replica and model counts, models that cannot be placed, and what has gone unasked-for. Derived from the same state the tables show, so it cannot disagree with them. Unhealthy GPUs count as neither used nor free, because counting them free is how a cluster looks like it has room it does not. Co-Authored-By: Claude Opus 5 (1M context) --- internal/cli/ctl.go | 21 +- internal/cli/server.go | 6 + internal/cli/status.go | 146 ++++++++++++++ internal/cli/status_test.go | 35 ++++ internal/control/adminapi/api.go | 38 +++- internal/control/adminapi/summary.go | 189 ++++++++++++++++++ internal/control/adminapi/summary_test.go | 111 ++++++++++ internal/control/control.go | 50 ++++- internal/control/reconciler/reconciler.go | 41 ++++ .../control/reconciler/reconciler_test.go | 76 +++++++ internal/control/reconciler/replica.go | 8 + internal/modelplan/size.go | 125 ++++++++++++ internal/router/endpoint/endpoint.go | 33 +++ internal/router/endpoint/endpoint_test.go | 64 ++++++ internal/router/router.go | 5 + 15 files changed, 928 insertions(+), 20 deletions(-) create mode 100644 internal/cli/status.go create mode 100644 internal/cli/status_test.go create mode 100644 internal/control/adminapi/summary.go create mode 100644 internal/control/adminapi/summary_test.go create mode 100644 internal/modelplan/size.go diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index e3f4a684..31bcefca 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -109,6 +109,7 @@ func newCtlCmd() *cobra.Command { // cert works on files rather than the admin API, so --server does not // apply to it; it lives here because this is where operators look. newBenchCmd(&asJSON), + newStatusCmd(client, &asJSON), newEngineCheckCmd(), newToolCheckCmd(), newCertCmd(), @@ -157,6 +158,19 @@ func newNodesCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { } } +// idleColumn shows how long a replica has gone unused. +// +// "-" for a replica nothing has reported use of, which is emphatically not the +// same as idle forever: it is what every replica looks like before its first +// request, and what they all look like after a router restart. Showing that as +// a very large number would make an untouched fleet read as an abandoned one. +func idleColumn(sec int64) string { + if sec < 0 { + return "-" + } + return (time.Duration(sec) * time.Second).Round(time.Second).String() +} + // memColumn shows a node's memory, distinguishing "did not say" from a number. // An older agent reports nothing, and that is not the same as a small machine. func memColumn(gb int) string { @@ -221,15 +235,16 @@ func newReplicasCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { } w := tab() - fmt.Fprintln(w, "REPLICA\tMODEL\tSTATE\tPOOL\tNODES\tAGE\tENDPOINT") + fmt.Fprintln(w, "REPLICA\tMODEL\tSTATE\tPOOL\tNODES\tAGE\tIDLE\tENDPOINT") for _, r := range reps { var nodes []string for _, s := range r.Shards { nodes = append(nodes, fmt.Sprintf("%s:%v", s.Node, s.GPUs)) } - fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\n", + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", r.ID, r.Model, r.State, r.Pool, join(nodes), - (time.Duration(r.AgeSec) * time.Second).String(), r.Endpoint) + (time.Duration(r.AgeSec) * time.Second).String(), + idleColumn(r.IdleSec), r.Endpoint) } _ = w.Flush() diff --git a/internal/cli/server.go b/internal/cli/server.go index 3210be1f..1f4d2d5b 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -257,6 +257,12 @@ func startControlPlane(ctx context.Context, cfg config.Server, cat *catalog.Cata // the service once it exists. fwd := &dispatcherRef{} ctl := control.New(ccfg, cat, fwd, sink, log) + // The router is both where endpoints go and where activity comes back + // from. It only reports usage when it is the thing serving requests, which + // is why this is a type assertion rather than a parameter. + if u, ok := sink.(control.UsageSource); ok { + ctl.SetUsageSource(u) + } svc := agentsvc.New(ctl.Inventory(), ctl.Reconciler(), requireIdentity, log) fwd.set(svc) diff --git a/internal/cli/status.go b/internal/cli/status.go new file mode 100644 index 00000000..9b3d6f4d --- /dev/null +++ b/internal/cli/status.go @@ -0,0 +1,146 @@ +package cli + +import ( + "fmt" + "time" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" +) + +// newStatusCmd prints the whole fleet in one screen. +// +// `ctl nodes` is the right tool for four nodes and the wrong one for two +// hundred. The questions that matter at that size are aggregates — how full am +// I, what is broken, what is holding cards nobody is using — and reading them +// off a long table is a chore, which means it gets skipped, which means nobody +// notices until something fails. +func newStatusCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { + return &cobra.Command{ + Use: "status", + Short: "The whole fleet in one answer: capacity, health, and what is idle", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + var s adminapi.Summary + if err := client().get(cmd.Context(), "/api/v1/summary", &s); err != nil { + return err + } + if *asJSON { + return emitJSON(s) + } + printSummary(cmd, s) + return nil + }, + } +} + +func printSummary(cmd *cobra.Command, s adminapi.Summary) { + out := cmd.OutOrStdout() + + fmt.Fprintf(out, "NODES %d ready", s.Nodes.Ready) + for _, x := range []struct { + n int + label string + }{ + {s.Nodes.Cordoned, "cordoned"}, + {s.Nodes.Draining, "draining"}, + {s.Nodes.Unreachable, "unreachable"}, + } { + if x.n > 0 { + fmt.Fprintf(out, ", %d %s", x.n, x.label) + } + } + fmt.Fprintf(out, " (of %d)\n", s.Nodes.Total) + + if len(s.GPUs) > 0 { + fmt.Fprintln(out, "\nGPUS") + w := tab() + fmt.Fprintln(w, " CLASS\tIN USE\tFREE\tUNHEALTHY\tUSAGE") + for _, g := range s.GPUs { + fmt.Fprintf(w, " %s\t%d/%d\t%d\t%s\t%s\n", + g.Class, g.Used, g.Total, g.Free(), unhealthyCell(g.Unhealthy), bar(g.Used, g.Total)) + } + _ = w.Flush() + } + + if len(s.Fabrics) > 0 { + fmt.Fprintln(out, "\nFABRICS") + w := tab() + // The free-block column is the one that decides whether a multi-node + // replica can land: free cards on the wrong fabric are not capacity for + // a model that spans nodes, because a replica may never cross one. + fmt.Fprintln(w, " FABRIC\tNODES\tFULLY FREE") + for _, f := range s.Fabrics { + fmt.Fprintf(w, " %s\t%d\t%d\n", f.Fabric, f.Nodes, f.LargestFreeBlock) + } + _ = w.Flush() + } + + fmt.Fprintf(out, "\nREPLICAS %d ready", s.Replicas.Ready) + for _, x := range []struct { + n int + label string + }{ + {s.Replicas.Starting, "starting"}, + {s.Replicas.Draining, "draining"}, + {s.Replicas.Failed, "failed"}, + } { + if x.n > 0 { + fmt.Fprintf(out, ", %d %s", x.n, x.label) + } + } + fmt.Fprintf(out, "\nMODELS %d of %d have the replicas they asked for\n", + s.Models.Satisfied, s.Models.Total) + + if len(s.Unplaceable) > 0 { + fmt.Fprintln(out, "\nNOT PLACED") + for _, m := range s.Unplaceable { + fmt.Fprintf(out, " %s\n", m) + } + fmt.Fprintln(out, "\n Why, per model: llmgw ctl explain-placement ") + } + + if len(s.IdleReplicas) > 0 { + fmt.Fprintln(out, "\nIDLE") + w := tab() + fmt.Fprintln(w, " REPLICA\tMODEL\tGPUS\tUNUSED FOR") + for _, r := range s.IdleReplicas { + fmt.Fprintf(w, " %s\t%s\t%d\t%s\n", r.ID, r.Model, r.GPUs, + (time.Duration(r.IdleSec) * time.Second).Round(time.Minute)) + } + _ = w.Flush() + fmt.Fprintln(out, "\n These hold hardware nothing is asking for.") + } +} + +// bar draws usage so a column of classes can be read at a glance rather than +// compared digit by digit. +func bar(used, total int) string { + const width = 20 + if total <= 0 { + return "" + } + filled := used * width / total + if filled > width { + filled = width + } + b := make([]rune, width) + for i := range b { + if i < filled { + b[i] = '#' + } else { + b[i] = '.' + } + } + return fmt.Sprintf("%s %3d%%", string(b), used*100/total) +} + +// unhealthyCell keeps a zero quiet. A column of noughts trains the eye to skip +// the column, which is the opposite of what it is for. +func unhealthyCell(n int) string { + if n == 0 { + return "-" + } + return fmt.Sprintf("%d", n) +} diff --git a/internal/cli/status_test.go b/internal/cli/status_test.go new file mode 100644 index 00000000..7c297ef1 --- /dev/null +++ b/internal/cli/status_test.go @@ -0,0 +1,35 @@ +package cli + +import ( + "strings" + "testing" +) + +func TestBarStaysWithinItsWidth(t *testing.T) { + // It is drawn from counts that can disagree briefly — a GPU freed between + // two reads — and a bar that overruns its column wrecks the table it is in. + for _, tc := range [][2]int{{0, 0}, {0, 8}, {8, 8}, {9, 8}, {100, 8}} { + got := bar(tc[0], tc[1]) + if tc[1] == 0 { + if got != "" { + t.Errorf("bar(%d,%d) = %q, want empty for an empty class", tc[0], tc[1], got) + } + continue + } + if n := strings.Count(got, "#") + strings.Count(got, "."); n != 20 { + t.Errorf("bar(%d,%d) drew %d cells: %q", tc[0], tc[1], n, got) + } + } +} + +func TestIdleColumnDistinguishesUnknownFromIdle(t *testing.T) { + // Every replica looks unreported before its first request and after a + // router restart. Showing that as a very large duration would make an + // untouched fleet read as an abandoned one. + if got := idleColumn(-1); got != "-" { + t.Errorf("idleColumn(-1) = %q, want a dash", got) + } + if got := idleColumn(90); got != "1m30s" { + t.Errorf("idleColumn(90) = %q", got) + } +} diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 86ead6e9..75791003 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -27,6 +27,7 @@ func New(ctl *control.Controller) *API { return &API{ctl: ctl} } // Register mounts the endpoints on a mux under /api/v1. func (a *API) Register(mux *http.ServeMux) { + mux.HandleFunc("GET /api/v1/summary", a.summary) mux.HandleFunc("GET /api/v1/nodes", a.listNodes) mux.HandleFunc("GET /api/v1/models", a.listModels) mux.HandleFunc("GET /api/v1/replicas", a.listReplicas) @@ -77,16 +78,21 @@ type ModelView struct { // ReplicaView is a replica as `ctl replicas` shows it. type ReplicaView struct { - ID string `json:"id"` - Model string `json:"model"` - Profile string `json:"profile"` - State string `json:"state"` - Pool string `json:"pool,omitempty"` - Fabric string `json:"fabric,omitempty"` - Endpoint string `json:"endpoint,omitempty"` - AgeSec int64 `json:"age_seconds"` - Err string `json:"error,omitempty"` - Shards []ShardView `json:"shards"` + ID string `json:"id"` + Model string `json:"model"` + Profile string `json:"profile"` + State string `json:"state"` + Pool string `json:"pool,omitempty"` + Fabric string `json:"fabric,omitempty"` + Endpoint string `json:"endpoint,omitempty"` + AgeSec int64 `json:"age_seconds"` + // IdleSec is how long since a request was routed here. Negative means the + // router has not reported any use of this replica, which is not the same + // as idle — it is what a replica looks like before anything has asked for + // it, and after a router restart. + IdleSec int64 `json:"idle_seconds"` + Err string `json:"error,omitempty"` + Shards []ShardView `json:"shards"` } // ShardView is one shard of a replica. @@ -212,7 +218,13 @@ func (a *API) Replicas(model string) []ReplicaView { ID: rep.ID, Model: rep.Model, Profile: rep.Profile, State: rep.State.String(), Pool: rep.Pool, Fabric: rep.Fabric, Endpoint: rep.Endpoint, Err: rep.Err, - AgeSec: int64(time.Since(rep.CreatedAt).Seconds()), + AgeSec: int64(time.Since(rep.CreatedAt).Seconds()), + IdleSec: -1, + } + if rep.LastUsedAt.IsZero() { + v.IdleSec = -1 + } else { + v.IdleSec = int64(time.Since(rep.LastUsedAt).Seconds()) } for _, s := range rep.Shards { v.Shards = append(v.Shards, ShardView{ @@ -225,6 +237,10 @@ func (a *API) Replicas(model string) []ReplicaView { return out } +func (a *API) summary(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, a.Summary()) +} + func (a *API) explainPlacement(w http.ResponseWriter, r *http.Request) { model := r.URL.Query().Get("model") if model == "" { diff --git a/internal/control/adminapi/summary.go b/internal/control/adminapi/summary.go new file mode 100644 index 00000000..fe2dfdeb --- /dev/null +++ b/internal/control/adminapi/summary.go @@ -0,0 +1,189 @@ +package adminapi + +import ( + "sort" + "time" +) + +// Summary is the whole fleet in one answer. +// +// A per-node listing is the right tool for four nodes and the wrong one for two +// hundred: the questions that matter at that size — how full am I, what is +// broken, what is holding cards it is not using — are aggregates, and reading +// them off a long table is a chore that gets skipped. Everything here is +// derived from the same state the tables show, so it can never disagree with +// them; it is a different question asked of the same facts. +type Summary struct { + Nodes NodeTotals `json:"nodes"` + GPUs []ClassUse `json:"gpus"` + // Fabrics matter because a multi-node replica may never span two of them, + // so free capacity in the wrong fabric is not capacity for the model that + // needs it. + Fabrics []FabricUse `json:"fabrics,omitempty"` + Replicas ReplicaTotal `json:"replicas"` + Models ModelTotals `json:"models"` + // Unplaceable names models the scheduler currently cannot satisfy. This is + // the line that turns "something is wrong" into a place to look. + Unplaceable []string `json:"unplaceable,omitempty"` + // IdleReplicas are serving replicas nothing has asked for in a while, which + // on a fleet with more models than GPUs is the first place to find room. + IdleReplicas []IdleReplica `json:"idle_replicas,omitempty"` +} + +// NodeTotals counts nodes by the states that need different responses. +type NodeTotals struct { + Total int `json:"total"` + Ready int `json:"ready"` + Cordoned int `json:"cordoned"` + Draining int `json:"draining"` + Unreachable int `json:"unreachable"` +} + +// ClassUse is how much of one kind of card is in use. +type ClassUse struct { + Class string `json:"class"` + Total int `json:"total"` + Used int `json:"used"` + // Unhealthy GPUs are neither used nor available, and counting them as free + // is how a cluster looks like it has room it does not have. + Unhealthy int `json:"unhealthy"` +} + +// Free is what can still be allocated. +func (c ClassUse) Free() int { return c.Total - c.Used - c.Unhealthy } + +// FabricUse is capacity within one fabric. +type FabricUse struct { + Fabric string `json:"fabric"` + Nodes int `json:"nodes"` + // LargestFreeBlock is the most nodes on this fabric that are entirely free, + // which is the number that decides whether a multi-node replica can land. + LargestFreeBlock int `json:"largest_free_block"` +} + +// ReplicaTotal counts replicas by state. +type ReplicaTotal struct { + Ready int `json:"ready"` + Starting int `json:"starting"` + Draining int `json:"draining"` + Failed int `json:"failed"` +} + +// ModelTotals counts models by whether they have what they asked for. +type ModelTotals struct { + Total int `json:"total"` + Satisfied int `json:"satisfied"` + Short int `json:"short"` +} + +// IdleReplica is a replica holding hardware nothing is asking for. +type IdleReplica struct { + ID string `json:"id"` + Model string `json:"model"` + IdleSec int64 `json:"idle_seconds"` + GPUs int `json:"gpus"` +} + +// idleThreshold is how long a replica must go unasked-for to be worth naming. +// +// Minutes rather than seconds, because a replica that flaps in and out of this +// list is noise, and because unloading one throws away its KV cache — the thing +// prefix routing spent its effort filling. +const idleThreshold = 10 * time.Minute + +// Summary builds the fleet summary. +func (a *API) Summary() Summary { + var s Summary + + nodes := a.Nodes() + byClass := map[string]*ClassUse{} + s.Nodes.Total = len(nodes) + for _, n := range nodes { + switch n.State { + case "ready": + s.Nodes.Ready++ + case "cordoned": + s.Nodes.Cordoned++ + case "draining": + s.Nodes.Draining++ + default: + s.Nodes.Unreachable++ + } + for _, c := range n.Classes { + if byClass[c] == nil { + byClass[c] = &ClassUse{Class: c} + } + } + // A node reports one class in practice; attribute its cards to it. + if len(n.Classes) == 1 { + u := byClass[n.Classes[0]] + u.Total += n.GPUTotal + u.Unhealthy += len(n.Unhealthy) + u.Used += n.GPUTotal - n.GPUFree - len(n.Unhealthy) + } + } + for _, u := range byClass { + s.GPUs = append(s.GPUs, *u) + } + sort.Slice(s.GPUs, func(i, j int) bool { return s.GPUs[i].Class < s.GPUs[j].Class }) + + byFabric := map[string]*FabricUse{} + for _, n := range nodes { + for _, f := range n.Fabrics { + if byFabric[f] == nil { + byFabric[f] = &FabricUse{Fabric: f} + } + byFabric[f].Nodes++ + // Entirely free, which is what a multi-node replica needs: a node + // with one card in use cannot contribute a full-width shard. + if n.State == "ready" && n.GPUFree == n.GPUTotal && n.GPUTotal > 0 { + byFabric[f].LargestFreeBlock++ + } + } + } + for _, f := range byFabric { + s.Fabrics = append(s.Fabrics, *f) + } + sort.Slice(s.Fabrics, func(i, j int) bool { return s.Fabrics[i].Fabric < s.Fabrics[j].Fabric }) + + for _, r := range a.Replicas("") { + switch r.State { + case "ready": + s.Replicas.Ready++ + case "draining": + s.Replicas.Draining++ + case "failed": + s.Replicas.Failed++ + default: + s.Replicas.Starting++ + } + if r.State == "ready" && r.IdleSec >= int64(idleThreshold.Seconds()) { + gpus := 0 + for _, sh := range r.Shards { + gpus += len(sh.GPUs) + } + s.IdleReplicas = append(s.IdleReplicas, IdleReplica{ + ID: r.ID, Model: r.Model, IdleSec: r.IdleSec, GPUs: gpus, + }) + } + } + sort.Slice(s.IdleReplicas, func(i, j int) bool { + return s.IdleReplicas[i].IdleSec > s.IdleReplicas[j].IdleSec + }) + + for _, m := range a.Models() { + s.Models.Total++ + if m.Ready >= m.Min { + s.Models.Satisfied++ + continue + } + s.Models.Short++ + // Short of replicas and unable to place more is the actionable case; + // short and still starting is just a cluster doing its job. + if m.Total <= m.Ready { + s.Unplaceable = append(s.Unplaceable, m.Name) + } + } + sort.Strings(s.Unplaceable) + return s +} diff --git a/internal/control/adminapi/summary_test.go b/internal/control/adminapi/summary_test.go new file mode 100644 index 00000000..b843e9a9 --- /dev/null +++ b/internal/control/adminapi/summary_test.go @@ -0,0 +1,111 @@ +package adminapi + +import ( + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" +) + +func TestSummaryCountsCapacityByClass(t *testing.T) { + // The question a per-node table answers badly at two hundred nodes: how + // full am I. + a, _ := setup(t, testModels(), gpuNode("n1", 8), gpuNode("n2", 8)) + s := a.Summary() + + if s.Nodes.Total != 2 || s.Nodes.Ready != 2 { + t.Errorf("nodes = %+v", s.Nodes) + } + if len(s.GPUs) != 1 { + t.Fatalf("expected one GPU class, got %+v", s.GPUs) + } + if g := s.GPUs[0]; g.Total != 16 || g.Free() != 16 { + t.Errorf("class totals wrong: %+v", g) + } +} + +func TestUnhealthyGPUsAreNotCountedAsFree(t *testing.T) { + // Counting them free is how a cluster looks like it has room it does not, + // which is worse than showing less capacity than exists. + n := gpuNode("n1", 4) + n.GPUs[0].Health.Problems = []string{"3 uncorrected ECC errors"} + a, _ := setup(t, testModels(), n) + + g := a.Summary().GPUs[0] + if g.Unhealthy != 1 { + t.Errorf("unhealthy = %d", g.Unhealthy) + } + if g.Free() != 3 { + t.Errorf("free = %d; a faulty card was counted as available", g.Free()) + } +} + +func TestFabricFreeBlocksCountWholeNodesOnly(t *testing.T) { + // A multi-node replica may never span fabrics, and a node with one card in + // use cannot contribute a full-width shard. Free cards scattered across + // partly-used nodes are not capacity for the model that needs them. + free := gpuNode("free", 8, "ib-a") + partial := gpuNode("partial", 8, "ib-a") + a, ctl := setup(t, testModels(), free, partial) + // Take one GPU on the second node. + ctl.Inventory().Allocate("partial", inventory.Allocation{ + ShardID: "s", ReplicaID: "r", GPUIndices: []int{0}, + }) + + s := a.Summary() + if len(s.Fabrics) != 1 { + t.Fatalf("fabrics = %+v", s.Fabrics) + } + if got := s.Fabrics[0].LargestFreeBlock; got != 1 { + t.Errorf("fully free nodes = %d, want 1: a partly used node is not one", got) + } +} + +func TestANodeWithNoGPUsDoesNotInventAClass(t *testing.T) { + // The CPU-only node this fleet acquired. It has no class, and a summary + // that invents one for it would report capacity nobody has. + a, _ := setup(t, testModels(), inventory.Node{ID: "cpu1", State: inventory.Ready}) + s := a.Summary() + for _, g := range s.GPUs { + if g.Class == "" { + t.Errorf("an empty class was reported: %+v", g) + } + } + if s.Nodes.Total != 1 { + t.Errorf("the node was not counted: %+v", s.Nodes) + } +} + +func TestIdleReplicasNeedRealIdleness(t *testing.T) { + // A replica the router has said nothing about carries -1, and must not be + // read as idle since the epoch — otherwise a router restart nominates the + // whole fleet for eviction at once. + for _, tc := range []struct { + name string + idle int64 + want bool + }{ + {"never reported", -1, false}, + {"just used", 5, false}, + {"idle for an hour", int64(time.Hour.Seconds()), true}, + } { + t.Run(tc.name, func(t *testing.T) { + got := tc.idle >= int64(idleThreshold.Seconds()) + if got != tc.want { + t.Errorf("idle=%d counted as idle=%v, want %v", tc.idle, got, tc.want) + } + }) + } +} + +func TestSummaryNamesModelsThatCannotBePlaced(t *testing.T) { + // The line that turns "something is wrong" into a place to look. + a, _ := setup(t, testModels()) // no nodes at all + s := a.Summary() + if len(s.Unplaceable) == 0 { + t.Fatal("a model with no hardware to run on was not named") + } + if s.Models.Short == 0 { + t.Error("no model was counted short") + } +} diff --git a/internal/control/control.go b/internal/control/control.go index e308857b..9ca29cb7 100644 --- a/internal/control/control.go +++ b/internal/control/control.go @@ -25,6 +25,21 @@ type EndpointSink interface { SetEndpoints(ctx context.Context, infos []endpoint.Info) } +// UsageSource reports when each endpoint was last sent a request. +// +// The return path of EndpointSink, and an interface for the same reason: the +// control plane decides what should run, the router is the only thing that +// knows whether anyone wants it, and neither should have to be in the same +// process as the other to say so. +// +// Deliberately a snapshot rather than an event stream. On a fleet with hundreds +// of replicas, per-request notification is a firehose carrying one bit of +// information — the newest timestamp wins — and a map read once a tick carries +// exactly the same answer for a fixed cost. +type UsageSource interface { + LastUsed() map[string]time.Time +} + // Config tunes the control plane. type Config struct { Reconciler reconciler.Config @@ -44,12 +59,13 @@ func DefaultConfig() Config { // Controller owns the control plane's state. type Controller struct { - cfg Config - log *slog.Logger - inv *inventory.Store - cat *catalog.Catalog - rec *reconciler.Reconciler - sink EndpointSink + cfg Config + log *slog.Logger + inv *inventory.Store + cat *catalog.Catalog + rec *reconciler.Reconciler + sink EndpointSink + usage UsageSource mu sync.Mutex last string // fingerprint of the last published list @@ -83,6 +99,25 @@ func New(cfg Config, cat *catalog.Catalog, disp reconciler.Dispatcher, sink Endp // Inventory exposes the node inventory, which the agent service feeds. func (c *Controller) Inventory() *inventory.Store { return c.inv } +// SetUsageSource supplies where request activity is read from. +// +// A setter rather than a constructor argument because a control plane without +// one is fully functional — it simply cannot say whether a replica is idle, +// which is how it behaved until now. +func (c *Controller) SetUsageSource(u UsageSource) { c.usage = u } + +// collectUsage stamps replicas with when they were last asked for anything. +// +// Once a tick, over the replicas rather than the requests. On a fleet of +// hundreds this is a map lookup each; there is no version of it that grows with +// traffic. +func (c *Controller) collectUsage() { + if c.usage == nil { + return + } + c.rec.RecordUsage(c.usage.LastUsed()) +} + // Reconciler exposes the reconciliation loop, for status reports and ctl. func (c *Controller) Reconciler() *reconciler.Reconciler { return c.rec } @@ -106,6 +141,9 @@ func (c *Controller) Start(ctx context.Context) { case <-ctx.Done(): return case <-t.C: + // Before publishing, so a replica's idleness is as fresh as + // the endpoint list it travels beside. + c.collectUsage() c.Publish(ctx) } } diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index 5e7c9772..02f68284 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -299,6 +299,47 @@ func (r *Reconciler) fleetHasSettled() bool { return settled } +// RecordUsage stamps replicas with when each was last sent a request. +// +// Replicas absent from the map are left alone rather than zeroed: a router that +// has just restarted knows nothing about what happened before it, and treating +// that as "never used" would make every replica look abandoned at once — +// exactly when a fleet is least able to afford acting on that. +func (r *Reconciler) RecordUsage(lastUsed map[string]time.Time) { + if len(lastUsed) == 0 { + return + } + r.mu.Lock() + defer r.mu.Unlock() + for id, at := range lastUsed { + rep, ok := r.replicas[id] + if !ok { + continue + } + // Only forwards. Two routers reporting, or a stale snapshot arriving + // late, must not move a replica's last use backwards. + if at.After(rep.LastUsedAt) { + rep.LastUsedAt = at + } + } +} + +// IdleFor reports how long a replica has gone without a request, and whether +// that is known at all. +// +// The second return is not a detail. "Idle for an hour" and "nobody has told me +// yet" are different states, and anything that acts on idleness — eviction, +// most obviously — has to refuse to act on the second. +func (r *Reconciler) IdleFor(id string) (time.Duration, bool) { + r.mu.Lock() + defer r.mu.Unlock() + rep, ok := r.replicas[id] + if !ok || rep.LastUsedAt.IsZero() { + return 0, false + } + return r.now().Sub(rep.LastUsedAt), true +} + // expireStuckReplicas fails replicas that never finished coming up. // // Every other path out of Starting depends on the agent reporting something. A diff --git a/internal/control/reconciler/reconciler_test.go b/internal/control/reconciler/reconciler_test.go index 646cb758..e8acc312 100644 --- a/internal/control/reconciler/reconciler_test.go +++ b/internal/control/reconciler/reconciler_test.go @@ -1153,3 +1153,79 @@ func TestSettlingOutlastsAnAgentReconnectBackoff(t *testing.T) { t.Fatal("placement never resumed") } } + +// --- idle tracking --- + +func TestUsageIsRecordedAgainstReplicas(t *testing.T) { + h := newHarness(t, testModel(1), gpuNode("1", 8)) + h.r.Reconcile(t.Context()) + reps := h.r.Replicas() + if len(reps) != 1 { + t.Fatalf("expected a replica, got %d", len(reps)) + } + id := reps[0].ID + + if _, known := h.r.IdleFor(id); known { + t.Error("a replica nothing has used reported a known idleness") + } + + used := h.clock.now() + h.r.RecordUsage(map[string]time.Time{id: used}) + h.advance(5 * time.Minute) + + idle, known := h.r.IdleFor(id) + if !known { + t.Fatal("usage was recorded but idleness is still unknown") + } + if idle != 5*time.Minute { + t.Errorf("idle = %s, want 5m", idle) + } +} + +func TestUsageNeverMovesBackwards(t *testing.T) { + // Two routers reporting, or a snapshot arriving late, must not make a busy + // replica look abandoned. + h := newHarness(t, testModel(1), gpuNode("1", 8)) + h.r.Reconcile(t.Context()) + id := h.r.Replicas()[0].ID + + recent := h.clock.now() + h.r.RecordUsage(map[string]time.Time{id: recent}) + h.r.RecordUsage(map[string]time.Time{id: recent.Add(-time.Hour)}) + + h.advance(time.Minute) + if idle, _ := h.r.IdleFor(id); idle != time.Minute { + t.Errorf("idle = %s; a stale report moved last-use backwards", idle) + } +} + +func TestUsageForReplicasWeDoNotHaveIsIgnored(t *testing.T) { + // The router's list and the reconciler's drift constantly on a live fleet: + // a replica stops between the snapshot and the read. That is ordinary, not + // an error, and must not panic or create phantom state. + h := newHarness(t, testModel(1), gpuNode("1", 8)) + h.r.Reconcile(t.Context()) + before := len(h.r.Replicas()) + + h.r.RecordUsage(map[string]time.Time{"a-replica-that-left": h.clock.now()}) + if got := len(h.r.Replicas()); got != before { + t.Errorf("replica count changed from %d to %d", before, got) + } +} + +func TestAnEmptyUsageReportLeavesEverythingAlone(t *testing.T) { + // A router that has just started knows nothing, and treating that as "never + // used" would make every replica look abandoned at once — exactly when a + // fleet is least able to afford acting on it. + h := newHarness(t, testModel(1), gpuNode("1", 8)) + h.r.Reconcile(t.Context()) + id := h.r.Replicas()[0].ID + + h.r.RecordUsage(map[string]time.Time{id: h.clock.now()}) + h.advance(time.Minute) + h.r.RecordUsage(nil) + + if idle, known := h.r.IdleFor(id); !known || idle != time.Minute { + t.Errorf("idle=%s known=%v; an empty report erased what was known", idle, known) + } +} diff --git a/internal/control/reconciler/replica.go b/internal/control/reconciler/replica.go index 528940be..ac990805 100644 --- a/internal/control/reconciler/replica.go +++ b/internal/control/reconciler/replica.go @@ -104,6 +104,14 @@ type Replica struct { CreatedAt time.Time ReadyAt time.Time + // LastUsedAt is when a request was last routed to this replica, zero if + // none has been since the router started. + // + // The difference between a replica that is holding eight cards for work and + // one that is holding them out of habit. Nothing recorded it, so on a fleet + // with more models than GPUs there was no way to answer the first question + // anyone asks: what can I stop? + LastUsedAt time.Time // DrainingAt is when the replica stopped taking new work, so the wait for a // replacement can be bounded. DrainingAt time.Time diff --git a/internal/modelplan/size.go b/internal/modelplan/size.go new file mode 100644 index 00000000..90f244ad --- /dev/null +++ b/internal/modelplan/size.go @@ -0,0 +1,125 @@ +// Package modelplan works out whether a model fits some hardware, and how it +// should be split when it does. +// +// It exists because that arithmetic was the real barrier to running a model +// here. Everything else about adding one is typing: the name, the path, the +// engine. Deciding tp, gpus_per_node, gpu_class and the context that will +// actually fit means knowing the model's layer count, its KV head count, what +// its weights weigh at a given quantisation, and how much of a card is left +// after the runtime takes its share — and getting it wrong is not an error +// message, it is an out-of-memory twenty minutes into a load. +package modelplan + +import "fmt" + +// Model is what a model's own configuration says about it. +// +// Every field here is read from the model rather than guessed: config.json +// carries the architecture, and the repository's index carries the parameter +// count. A guess would defeat the point, because the answer is only worth +// having if it is the one the hardware will agree with. +type Model struct { + Name string + // Params is the total parameter count. + Params int64 + // Layers, KVHeads and HeadDim size the KV cache, which is what scales with + // context and is usually what actually runs a card out of memory. + Layers int + KVHeads int + HeadDim int + // MaxContext is the model's trained context length. + MaxContext int +} + +// Precision is how many bytes one parameter weighs. +// +// Named rather than numeric because the names are what people use, and because +// the difference between them is the difference between a model fitting and +// not. +type Precision struct { + Name string + BytesPerParam float64 + BytesPerKVElem float64 +} + +// Known precisions, from full to the quantisations worth serving. +var ( + BF16 = Precision{"bf16", 2, 2} + FP8 = Precision{"fp8", 1, 1} + INT4 = Precision{"int4", 0.5, 2} // weights quantised, KV usually is not + Q8GGUF = Precision{"q8_0", 1.06, 2} + Q4GGUF = Precision{"q4_k_m", 0.61, 2} +) + +// PrecisionByName resolves a precision, reporting what it knows when it cannot. +func PrecisionByName(s string) (Precision, error) { + for _, p := range []Precision{BF16, FP8, INT4, Q8GGUF, Q4GGUF} { + if p.Name == s { + return p, nil + } + } + return Precision{}, fmt.Errorf("unknown precision %q; known: bf16, fp8, int4, q8_0, q4_k_m", s) +} + +// Demand is what one replica needs, in bytes. +type Demand struct { + Weights int64 + KVCache int64 + Overhead int64 +} + +// Total is everything one replica needs across all its GPUs. +func (d Demand) Total() int64 { return d.Weights + d.KVCache + d.Overhead } + +// PerGPU divides the demand over a replica's GPUs. +// +// Tensor parallelism splits both the weights and the KV cache, so both divide. +// The overhead does not: each process pays its own runtime, CUDA context and +// activation buffers, so it is charged per GPU rather than shared. +func (d Demand) PerGPU(gpus int) int64 { + if gpus <= 0 { + gpus = 1 + } + return (d.Weights+d.KVCache)/int64(gpus) + d.Overhead +} + +// KVBytesPerToken is what one token of context costs, for the whole model. +// +// Two for the key and the value, times the layers, times the heads the cache +// actually keeps. Grouped-query attention is why KVHeads and not the attention +// head count: a 14B model with 40 attention heads and 8 KV heads pays for +// eight, and the difference is a factor of five. +func (m Model) KVBytesPerToken(p Precision) int64 { + return int64(2 * float64(m.Layers) * float64(m.KVHeads) * float64(m.HeadDim) * p.BytesPerKVElem) +} + +// Needs computes what one replica of this model needs. +// +// context is the total token pool the engine will hold, which is the context +// length times however many requests it serves at once — the number an engine +// is configured with, not the number a single request sees. +func (m Model) Needs(p Precision, context int) Demand { + weights := int64(float64(m.Params) * p.BytesPerParam) + kv := int64(context) * m.KVBytesPerToken(p) + return Demand{ + Weights: weights, + KVCache: kv, + Overhead: overheadBytes(weights), + } +} + +// overheadBytes is what the runtime costs beyond weights and cache. +// +// A flat allowance plus a fraction of the weights: the flat part is the CUDA +// context, the framework and its allocator, which a small model pays as much as +// a large one; the fraction stands in for activations and fragmentation, which +// do scale. Deliberately generous. A plan that says a model fits and is wrong +// costs a twenty-minute load and an OOM; one that says it does not quite fit +// costs a smaller quantisation. +func overheadBytes(weights int64) int64 { + const flat = 2 << 30 // 2 GiB + return flat + weights/10 +} + +// GB renders bytes the way people discuss hardware. +func GB(b int64) float64 { return float64(b) / (1 << 30) } diff --git a/internal/router/endpoint/endpoint.go b/internal/router/endpoint/endpoint.go index a47f3f74..9c8206ca 100644 --- a/internal/router/endpoint/endpoint.go +++ b/internal/router/endpoint/endpoint.go @@ -120,6 +120,18 @@ type live struct { inflight int brk *breaker.Breaker + // lastUsed is when a request was last dispatched here. + // + // The fact nothing recorded. Whether a replica is earning the cards it + // holds is the first question on a fleet with more models than GPUs, and + // the control plane had no way to ask it: it knew what was placed and + // whether it was healthy, never whether anyone wanted it. + // + // Recorded here because Acquire is already the one place a request meets an + // endpoint, and already holds the lock. One timestamp per dispatch, and + // collection is one pass over the replicas rather than anything per + // request — which is what makes it affordable on a fleet of hundreds. + lastUsed time.Time } // Registry holds the live state of all endpoints. It is safe for concurrent use. @@ -265,6 +277,7 @@ func (r *Registry) Acquire(id string) (release func(ok bool), granted bool) { r.mu.Lock() l.inflight++ + l.lastUsed = r.now() r.mu.Unlock() var once sync.Once @@ -282,6 +295,26 @@ func (r *Registry) Acquire(id string) (release func(ok bool), granted bool) { }, true } +// LastUsed reports when each endpoint last had a request dispatched to it. +// +// Endpoints that have never been used are absent rather than zero-valued: "not +// since this router started" and "not for an hour" are different facts, and a +// zero time would quietly become the oldest idle replica in any comparison — +// so a router restart would nominate everything for eviction at once. +func (r *Registry) LastUsed() map[string]time.Time { + r.mu.RLock() + defer r.mu.RUnlock() + + out := make(map[string]time.Time, len(r.m)) + for id, l := range r.m { + if l.lastUsed.IsZero() { + continue + } + out[id] = l.lastUsed + } + return out +} + // ScrapeFailed reports a failed metrics scrape. // // A scrape failure is not a request failure, so it does not touch the breaker: diff --git a/internal/router/endpoint/endpoint_test.go b/internal/router/endpoint/endpoint_test.go index 29e4c11f..39542a6a 100644 --- a/internal/router/endpoint/endpoint_test.go +++ b/internal/router/endpoint/endpoint_test.go @@ -243,3 +243,67 @@ func TestConcurrentAccess(t *testing.T) { } wg.Wait() } + +func TestLastUsedIsRecordedOnDispatch(t *testing.T) { + // The fact nothing recorded. Whether a replica is earning the cards it + // holds is the first question on a fleet with more models than GPUs. + r := NewRegistry(Config{}) + clock := time.Date(2026, 9, 17, 12, 0, 0, 0, time.UTC) + r.SetClock(func() time.Time { return clock }) + r.Replace([]Info{{ID: "a", URL: "http://x", Models: []string{"m"}}}) + + if len(r.LastUsed()) != 0 { + t.Fatal("an endpoint nothing has used reported a last-use time") + } + + release, ok := r.Acquire("a") + if !ok { + t.Fatal("acquire refused") + } + release(true) + + used := r.LastUsed() + if got, ok := used["a"]; !ok || !got.Equal(clock) { + t.Errorf("last used = %v (present=%v), want %v", got, ok, clock) + } +} + +func TestAnUnusedEndpointIsAbsentRatherThanZero(t *testing.T) { + // A zero time would be the oldest idle replica in any comparison, so a + // router restart would nominate the whole fleet for eviction at once. + r := NewRegistry(Config{}) + r.Replace([]Info{ + {ID: "used", URL: "http://x", Models: []string{"m"}}, + {ID: "never", URL: "http://y", Models: []string{"m"}}, + }) + release, _ := r.Acquire("used") + release(true) + + used := r.LastUsed() + if _, ok := used["never"]; ok { + t.Error("an endpoint nothing has used appeared in the map") + } + if _, ok := used["used"]; !ok { + t.Error("the used endpoint is missing") + } +} + +func TestLastUsedSurvivesAnEndpointListRefresh(t *testing.T) { + // The list is republished whenever anything changes, which on a large + // fleet is often. Losing usage on every republish would make idleness + // unmeasurable exactly where it matters most. + r := NewRegistry(Config{}) + r.Replace([]Info{{ID: "a", URL: "http://x", Models: []string{"m"}}}) + release, _ := r.Acquire("a") + release(true) + before := r.LastUsed()["a"] + + // Same endpoint, republished alongside a new one. + r.Replace([]Info{ + {ID: "a", URL: "http://x", Models: []string{"m"}}, + {ID: "b", URL: "http://y", Models: []string{"m"}}, + }) + if got, ok := r.LastUsed()["a"]; !ok || !got.Equal(before) { + t.Errorf("last used = %v (present=%v) after a republish, want %v", got, ok, before) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 9f9a85d3..8d2737ef 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -335,6 +335,11 @@ func (r *Router) Registry() *endpoint.Registry { return r.reg } // the update keep their live state — inflight counts, breaker state, drain rate // — so a replica that was already serving does not start from cold just // because another one moved. +// LastUsed reports when each endpoint was last sent a request, satisfying +// control.UsageSource so the control plane can tell a busy replica from one +// holding its cards out of habit. +func (r *Router) LastUsed() map[string]time.Time { return r.reg.LastUsed() } + func (r *Router) SetEndpoints(ctx context.Context, infos []endpoint.Info) { if r.prefix != nil { // A replica that has gone takes its cache with it. Leaving its entries