From adb1b3cf2ac7a674e4fbcd94d01a1fcfc227db11 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Wed, 30 Sep 2026 03:18:07 +0200 Subject: [PATCH] Add replicas under load, up to a model's max, and take them back when quiet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A model's max never did anything: desiredFor returned the minimum, or one for an on-demand model someone was asking for, whatever the queue. Six small CPU nodes run out of slots quickly, and a model with room to grow stayed at its minimum while requests waited. The router already scrapes each engine's running and waiting requests. The control plane now reads them per model on every tick, alongside when each replica was last used, and the reconciler keeps how many replicas load has added above a model's base. One or more requests waiting per ready replica, held for a minute, adds one; ten minutes with nothing waiting removes one of those again. It never goes above max or below the base, moves one step at a time, and does not add while the last addition is still placing or starting — on this fleet a replica is minutes of weights and gigabytes of memory, and a spike should not leave a pile of replicas behind it. Decisions are logged with the load behind them. Nothing live changes until a model's max is raised above its min. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 11 ++ internal/control/control.go | 10 ++ internal/control/reconciler/reconciler.go | 113 ++++++++++++++++++ .../control/reconciler/reconciler_test.go | 71 +++++++++++ internal/router/router.go | 19 +++ 5 files changed, 224 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d52d8b63..747850da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -77,6 +77,17 @@ more than 26 hours old or there has never been one. docs/install.md now says how to restore, including why running engines survive it: the engine-key secret comes back with the store. +### Replicas follow load between min and max + +A model's `max` was a ceiling nothing reached: every model ran exactly its +minimum however long its queue. Now, when requests are waiting at the engines — +every slot of every replica busy — at one or more per replica for a minute, one +replica is added, up to `max`; ten minutes with nothing waiting takes one of +those away, down to `min`. One step at a time, and not while the previous one +is still starting, so a spike that is over before a replica loads does not +leave several behind. Each decision is logged with the load that made it. +`llmgw ctl scale MODEL --min 1 --max 3` turns it on for a model. + ### A model from the registry in one command `llmgw ctl model serve OWNER/REPO` takes a model from the registry to a replica diff --git a/internal/control/control.go b/internal/control/control.go index d742eda5..2ebf98a6 100644 --- a/internal/control/control.go +++ b/internal/control/control.go @@ -134,6 +134,16 @@ func (c *Controller) collectUsage() { return } c.rec.RecordUsage(c.usage.LastUsed()) + // And what the replicas are doing, when the source can say: what adds + // and removes replicas between a model's min and max. + if l, ok := c.usage.(LoadSource); ok { + c.rec.RecordLoad(l.ModelLoad()) + } +} + +// LoadSource reports each model's replicas' load, keyed by catalog name. +type LoadSource interface { + ModelLoad() map[string]reconciler.Load } // ReplaceCatalog puts a new catalog in force, without a restart. diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index 3a922b96..c0a3d3ba 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -70,6 +70,13 @@ type Config struct { // Without a bound, draining a node whose model cannot be placed anywhere // else would never finish. DrainGrace time.Duration + // ScaleUpWaiting is the average number of requests waiting per ready + // replica, held for ScaleUpAfter, that adds a replica — up to the model's + // max. ScaleDownAfter is how long with nothing waiting before one of the + // replicas added that way is taken away again, down to its min. + ScaleUpWaiting float64 + ScaleUpAfter time.Duration + ScaleDownAfter time.Duration // ParkAfter is how many failures in a row stop a model being retried. // // Backoff alone never gives up: it grows to MaxBackoff and keeps trying, @@ -140,6 +147,9 @@ func DefaultConfig() Config { MinLifetime: 5 * time.Minute, DrainGrace: 5 * time.Minute, ParkAfter: 5, + ScaleUpWaiting: 1, + ScaleUpAfter: time.Minute, + ScaleDownAfter: 10 * time.Minute, PortRange: [2]int{20000, 20999}, MgmtInterface: "eth0", } @@ -182,6 +192,15 @@ func (c *Config) normalize() { if c.ParkAfter <= 0 { c.ParkAfter = d.ParkAfter } + if c.ScaleUpWaiting <= 0 { + c.ScaleUpWaiting = d.ScaleUpWaiting + } + if c.ScaleUpAfter <= 0 { + c.ScaleUpAfter = d.ScaleUpAfter + } + if c.ScaleDownAfter <= 0 { + c.ScaleDownAfter = d.ScaleDownAfter + } if c.PortRange[1] <= c.PortRange[0] { c.PortRange = d.PortRange } @@ -228,6 +247,11 @@ type Reconciler struct { restartQueue map[string][]string // parked holds models that failed ParkAfter times in a row. parked map[string]parkRecord + // extra is how many replicas load has added above a model's base, and + // busySince/quietSince when its load last crossed the thresholds. + extra map[string]int + busySince map[string]time.Time + quietSince map[string]time.Time // placementLogged remembers the last "no placement" said per model, so it // is said when it changes rather than on every five-second pass. placementLogged map[string]placementNote @@ -261,6 +285,9 @@ func New(cfg Config, cat *catalog.Live, inv *inventory.Store, disp Dispatcher, l wanted: map[string]time.Time{}, restartQueue: map[string][]string{}, parked: map[string]parkRecord{}, + extra: map[string]int{}, + busySince: map[string]time.Time{}, + quietSince: map[string]time.Time{}, placementLogged: map[string]placementNote{}, lastExplain: map[string]placement.Result{}, wake: make(chan struct{}, 1), @@ -508,6 +535,19 @@ func (r *Reconciler) Want(model string) bool { // feature: a cluster with more models than GPUs can hold every model in its // catalog and only the wanted ones in its cards. func (r *Reconciler) desiredFor(m catalog.Model) int { + base := r.baseFor(m) + if base == 0 { + return 0 + } + r.mu.Lock() + extra := r.extra[m.Name] + r.mu.Unlock() + return min(base+extra, max(m.Scaling.Max, base)) +} + +// baseFor is what a model runs without load: its minimum, or for an on-demand +// model one while somebody is asking and none once they stop. +func (r *Reconciler) baseFor(m catalog.Model) int { if !m.Scaling.OnDemand() { return m.Scaling.Min } @@ -1715,3 +1755,76 @@ func firstLine(s string) string { } return s } + +// Load is what a model's replicas are doing, as the router sees it. +type Load struct { + Replicas int + Running int + Waiting int +} + +// RecordLoad adds or takes away replicas between a model's min and max as its +// load says. +// +// Max was a ceiling nothing reached: every model ran exactly its minimum +// however long its queue. Now requests waiting at the engines — every slot of +// every replica busy — held for ScaleUpAfter add one replica, and a quiet spell +// of ScaleDownAfter takes one of those back. One step at a time, and not while +// the last one is still starting: on a small CPU fleet each replica is minutes +// of weights and gigabytes of memory, and a spike that is over by the time it +// loads should not leave three behind. +func (r *Reconciler) RecordLoad(loads map[string]Load) { + now := r.now() + for _, m := range r.cat.Load().Models() { + base := r.baseFor(m) + r.mu.Lock() + if base == 0 || m.Scaling.Max <= base { + delete(r.extra, m.Name) + delete(r.busySince, m.Name) + delete(r.quietSince, m.Name) + r.mu.Unlock() + continue + } + starting := false + for _, rep := range r.replicas { + if rep.Model == m.Name && (rep.State == Placing || rep.State == Starting) { + starting = true + } + } + l := loads[m.Name] + perReplica := float64(l.Waiting) / float64(max(l.Replicas, 1)) + extra := r.extra[m.Name] + switch { + case perReplica >= r.cfg.ScaleUpWaiting: + delete(r.quietSince, m.Name) + since, ok := r.busySince[m.Name] + if !ok { + r.busySince[m.Name] = now + break + } + if now.Sub(since) >= r.cfg.ScaleUpAfter && base+extra < m.Scaling.Max && !starting { + r.extra[m.Name] = extra + 1 + r.busySince[m.Name] = now + r.log.Info("scaling up for load", "model", m.Name, "replicas", base+extra+1, + "waiting_per_replica", perReplica, "for", now.Sub(since).Round(time.Second)) + } + case l.Waiting == 0 && extra > 0: + delete(r.busySince, m.Name) + since, ok := r.quietSince[m.Name] + if !ok { + r.quietSince[m.Name] = now + break + } + if now.Sub(since) >= r.cfg.ScaleDownAfter { + r.extra[m.Name] = extra - 1 + r.quietSince[m.Name] = now + r.log.Info("scaling down after a quiet spell", "model", m.Name, + "replicas", base+extra-1, "quiet_for", now.Sub(since).Round(time.Second)) + } + default: + delete(r.busySince, m.Name) + delete(r.quietSince, m.Name) + } + r.mu.Unlock() + } +} diff --git a/internal/control/reconciler/reconciler_test.go b/internal/control/reconciler/reconciler_test.go index c1394b17..872c4e0d 100644 --- a/internal/control/reconciler/reconciler_test.go +++ b/internal/control/reconciler/reconciler_test.go @@ -1932,3 +1932,74 @@ func TestAPlacementFailureIsLoggedWhenItChangesNotEveryPass(t *testing.T) { t.Errorf("health %+v", mh) } } + +func liveCount(h *harness) int { + n := 0 + for _, r := range h.r.Replicas() { + if r.State.Counts() { + n++ + } + } + return n +} + +func TestLoadAddsReplicasUpToMaxAndQuietTakesThemBack(t *testing.T) { + m := testModel(1) + m.Scaling.Max = 3 + h := newHarness(t, m, gpuNode("1", 8), gpuNode("2", 8), gpuNode("3", 8)) + h.r.Reconcile(t.Context()) + h.markReady() + busy := map[string]Load{"llama": {Replicas: 1, Running: 4, Waiting: 3}} + + // A spike shorter than ScaleUpAfter adds nothing. + h.r.RecordLoad(busy) + h.advance(30 * time.Second) + h.r.RecordLoad(busy) + h.r.Reconcile(t.Context()) + if n := liveCount(h); n != 1 { + t.Fatalf("%d replicas after 30s of load", n) + } + + // Held past it, one more. + h.advance(31 * time.Second) + h.r.RecordLoad(busy) + h.r.Reconcile(t.Context()) + if n := liveCount(h); n != 2 { + t.Fatalf("%d replicas after a minute of load; expected 2", n) + } + + // Not another while that one is still starting. + h.advance(2 * time.Minute) + h.r.RecordLoad(busy) + h.r.Reconcile(t.Context()) + if n := liveCount(h); n != 2 { + t.Fatalf("stacked a replica on one still starting: %d", n) + } + + // Once it is up and load holds, the third — and never a fourth. + h.markReady() + for i := 0; i < 5; i++ { + h.r.RecordLoad(busy) + h.advance(61 * time.Second) + h.r.RecordLoad(busy) + h.r.Reconcile(t.Context()) + h.markReady() + } + if n := liveCount(h); n != 3 { + t.Fatalf("%d replicas under sustained load; max is 3", n) + } + + // Quiet: back down one step per ScaleDownAfter, to min and no further. + quiet := map[string]Load{"llama": {Replicas: 3}} + for i := 0; i < 6; i++ { + h.r.RecordLoad(quiet) + h.advance(DefaultConfig().ScaleDownAfter + time.Second) + h.r.RecordLoad(quiet) + for j := 0; j < 3; j++ { + h.r.Reconcile(t.Context()) + } + } + if n := liveCount(h); n != 1 { + t.Errorf("%d replicas after a long quiet spell; min is 1", n) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 4d43f5db..92d393e9 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -9,6 +9,7 @@ import ( "context" "errors" "fmt" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "log/slog" "math" "net/http" @@ -510,6 +511,24 @@ func (r *Router) SetCatalog(c *catalog.Catalog) { // holding its cards out of habit. func (r *Router) LastUsed() map[string]time.Time { return r.reg.LastUsed() } +// ModelLoad sums what each model's replicas report: how many there are, and +// requests running and waiting at their engines. Keyed by the catalog name, +// which the control plane publishes first among each endpoint's names. +func (r *Router) ModelLoad() map[string]reconciler.Load { + out := map[string]reconciler.Load{} + for _, s := range r.reg.Snapshots("") { + if len(s.Models) == 0 { + continue + } + l := out[s.Models[0]] + l.Replicas++ + l.Running += s.Stats.Running + l.Waiting += s.Stats.Waiting + out[s.Models[0]] = l + } + return out +} + 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