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