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