From 5ae870e1b256d9aa7ebfc90c18554ec23ce3bfa6 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Wed, 7 Oct 2026 13:09:10 +0200 Subject: [PATCH] ctl restart --max-surge and --max-unavailable A model's replicas restarted one at a time, always. RestartPace says how many restart at once: MaxSurge beside themselves (replacement first), MaxUnavailable stopped first (no room needed, out while loading); the reconciler begins that many per model per pass. The admin API takes max_surge and max_unavailable on POST /api/v1/restart. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/admin-api.openapi.json | 7 +++ internal/cli/ctl_restart.go | 17 ++++-- internal/control/adminapi/api.go | 14 +++-- internal/control/reconciler/reconciler.go | 42 ++++++++++---- .../control/reconciler/reconciler_test.go | 57 +++++++++++++++++++ internal/control/reconciler/restart.go | 36 ++++++++++++ 6 files changed, 155 insertions(+), 18 deletions(-) diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index 8efa8bae..2df4bd12 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -5592,6 +5592,13 @@ "in_place": { "type": "boolean", "description": "InPlace stops each replica before starting its replacement, rather than starting the replacement beside it first." + }, + "max_surge": { + "type": "integer", + "description": "MaxSurge is how many of a model's replicas are replaced beside themselves at once, MaxUnavailable how many are stopped first at once; both zero is one at a time, replaced beside itself." + }, + "max_unavailable": { + "type": "integer" } } }, diff --git a/internal/cli/ctl_restart.go b/internal/cli/ctl_restart.go index 59d33847..d0133838 100644 --- a/internal/cli/ctl_restart.go +++ b/internal/cli/ctl_restart.go @@ -12,9 +12,10 @@ import ( func newRestartCmd(client func() *ctlClient) *cobra.Command { var ( - models []string - all bool - inPlace bool + models []string + all bool + inPlace bool + surge, unavailable int ) c := &cobra.Command{ Use: "restart [REPLICA...]", @@ -31,6 +32,10 @@ func newRestartCmd(client func() *ctlClient) *cobra.Command { "profile edited to need more than any node has — the restart is given up\n" + "and the old replica keeps serving: llmgw ctl replicas says so, and why.\n\n" + "--in-place stops each replica first, whatever room there is.\n\n" + + "A model's replicas are replaced one at a time. --max-surge N replaces N at\n" + + "once, each beside itself, needing room for N more; --max-unavailable N\n" + + "stops N at once first, needing no room and leaving N fewer serving while\n" + + "they load. Together, the first N are stopped first and the rest surged.\n\n" + " llmgw ctl restart qwen2.5-7b-3f2a91c0\n" + " llmgw ctl restart --model qwen2.5-7b\n" + " llmgw ctl restart --all", @@ -38,7 +43,8 @@ func newRestartCmd(client func() *ctlClient) *cobra.Command { if len(args) == 0 && len(models) == 0 && !all { return errors.New("name replicas, --model, or --all") } - body, _ := json.Marshal(adminapi.RestartRequest{Replicas: args, Models: models, All: all, InPlace: inPlace}) + body, _ := json.Marshal(adminapi.RestartRequest{Replicas: args, Models: models, All: all, InPlace: inPlace, + MaxSurge: surge, MaxUnavailable: unavailable}) var res adminapi.RestartResult if err := client().post(cmd.Context(), "/api/v1/restart", body, &res); err != nil { return err @@ -50,6 +56,9 @@ func newRestartCmd(client func() *ctlClient) *cobra.Command { c.Flags().StringArrayVar(&models, "model", nil, "every replica of this model; repeatable") c.Flags().BoolVar(&all, "all", false, "every ready replica in the cluster") c.Flags().BoolVar(&inPlace, "in-place", false, "stop each replica before starting its replacement") + c.Flags().IntVar(&surge, "max-surge", 0, "replicas of a model replaced beside themselves at once (default 1)") + c.Flags().IntVar(&unavailable, "max-unavailable", 0, + "replicas of a model stopped first at once, each out while its replacement loads") return c } diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 52551e71..87fe5e43 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -1517,6 +1517,11 @@ type RestartRequest struct { // InPlace stops each replica before starting its replacement, rather // than starting the replacement beside it first. InPlace bool `json:"in_place,omitempty"` + // MaxSurge is how many of a model's replicas are replaced beside + // themselves at once, MaxUnavailable how many are stopped first at once; + // both zero is one at a time, replaced beside itself. + MaxSurge int `json:"max_surge,omitempty"` + MaxUnavailable int `json:"max_unavailable,omitempty"` } // RestartResult lists the replicas queued. @@ -1540,11 +1545,12 @@ func (a *API) restart(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "nothing ready matches; name replicas, models, or all") return } - restart := a.ctl.Reconciler().Restart - if req.InPlace { - restart = a.ctl.Reconciler().RestartInPlace + if req.MaxSurge < 0 || req.MaxUnavailable < 0 { + writeError(w, http.StatusBadRequest, "max_surge and max_unavailable are counts of replicas, not negative") + return } - if err := restart(ids...); err != nil { + pace := reconciler.RestartPace{MaxSurge: req.MaxSurge, MaxUnavailable: req.MaxUnavailable} + if err := a.ctl.Reconciler().RestartPaced(pace, req.InPlace, ids...); err != nil { writeError(w, http.StatusNotFound, err.Error()) return } diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index 104f83c2..d82d4224 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -270,6 +270,9 @@ type Reconciler struct { restartQueue map[string][]string // restartInPlace are the replicas asked to restart by stopping first. restartInPlace map[string]bool + // restartPace is, per model, how many of its replicas may be restarting + // at once: one unless an operator said otherwise. + restartPace map[string]RestartPace // restartRefused says, by replica, why a restart was given up rather // than leave the model without it. restartRefused map[string]string @@ -323,6 +326,7 @@ func New(cfg Config, cat *catalog.Live, inv *inventory.Store, disp Dispatcher, l wanted: map[string]time.Time{}, restartQueue: map[string][]string{}, restartInPlace: map[string]bool{}, + restartPace: map[string]RestartPace{}, restartRefused: map[string]string{}, parked: map[string]parkRecord{}, extra: map[string]int{}, @@ -775,19 +779,37 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) { drainingIDs = append(drainingIDs, rep.ID) } } - // The next restart an operator queued, once the last one is done. + // The next restarts an operator queued, as many at once as the pace + // allows: one by default, each begun once the last is done. restarting := "" - if len(retiringIDs) == 0 { - if id, ok := r.nextRestartLocked(m.Name); ok { - r.replicas[id].RetiringAt = r.now() - retiringIDs = append(retiringIDs, id) - liveIDs = slices.DeleteFunc(liveIDs, func(x string) bool { return x == id }) - readyLive-- - restarting = id - r.log.Info("restarting a replica: placing its replacement first", - "replica", id, "model", m.Name) + pace := r.restartPace[m.Name] + inPlaceNow := 0 + for _, id := range retiringIDs { + if r.restartInPlace[id] { + inPlaceNow++ } } + for len(retiringIDs) < pace.width() { + id, ok := r.nextRestartLocked(m.Name) + if !ok { + break + } + // Up to MaxUnavailable of them stop first; the rest are surged. + if inPlaceNow < pace.MaxUnavailable { + r.restartInPlace[id] = true + inPlaceNow++ + } + r.replicas[id].RetiringAt = r.now() + retiringIDs = append(retiringIDs, id) + liveIDs = slices.DeleteFunc(liveIDs, func(x string) bool { return x == id }) + readyLive-- + restarting = id + r.log.Info("restarting a replica", "replica", id, "model", m.Name, + "in_place", r.restartInPlace[id], "at_once", pace.width()) + } + if _, queued := r.restartQueue[m.Name]; !queued && len(retiringIDs) == 0 { + delete(r.restartPace, m.Name) + } blockedUntil := r.nextAttempt[m.Name] park, isParked := r.parked[m.Name] if isParked && park.fp != modelFingerprint(m) { diff --git a/internal/control/reconciler/reconciler_test.go b/internal/control/reconciler/reconciler_test.go index f9e35d80..e7b9765b 100644 --- a/internal/control/reconciler/reconciler_test.go +++ b/internal/control/reconciler/reconciler_test.go @@ -2063,3 +2063,60 @@ func TestLoadAddsReplicasUpToMaxAndQuietTakesThemBack(t *testing.T) { t.Errorf("%d replicas after a long quiet spell; min is 1", n) } } + +func TestReplicasRestartSeveralAtOnceWhenAsked(t *testing.T) { + h := newHarness(t, testModel(3), gpuNode("1", 8), gpuNode("2", 8), gpuNode("3", 8), gpuNode("4", 8)) + for i := 0; i < 3; i++ { + h.r.Reconcile(t.Context()) + h.markReady() + } + var ids []string + for _, rep := range h.r.ReadyReplicas() { + ids = append(ids, rep.ID) + } + if err := h.r.RestartPaced(RestartPace{MaxSurge: 2}, false, ids...); err != nil { + t.Fatal(err) + } + h.r.Reconcile(t.Context()) + replacing, queued := 0, 0 + for _, id := range ids { + switch h.r.RestartState(id) { + case "replacing": + replacing++ + case "queued": + queued++ + } + } + if replacing != 2 || queued != 1 { + t.Errorf("max surge 2: %d replacing, %d queued", replacing, queued) + } +} + +func TestMaxUnavailableStopsThatManyFirst(t *testing.T) { + h := newHarness(t, testModel(2), gpuNode("1", 8), gpuNode("2", 8), gpuNode("3", 8)) + for i := 0; i < 3; i++ { + h.r.Reconcile(t.Context()) + h.markReady() + } + var ids []string + for _, rep := range h.r.ReadyReplicas() { + ids = append(ids, rep.ID) + } + if err := h.r.RestartPaced(RestartPace{MaxSurge: 1, MaxUnavailable: 1}, false, ids...); err != nil { + t.Fatal(err) + } + h.r.Reconcile(t.Context()) + stopped, beside := 0, 0 + for _, id := range ids { + rep, ok := replicaByID(h, id) + switch { + case !ok || rep.State != Ready: + stopped++ + case h.r.RestartState(id) == "replacing": + beside++ + } + } + if stopped != 1 || beside != 1 { + t.Errorf("one stopped first and one replaced beside itself: %d stopped, %d beside", stopped, beside) + } +} diff --git a/internal/control/reconciler/restart.go b/internal/control/reconciler/restart.go index 1af8031b..550cbeea 100644 --- a/internal/control/reconciler/restart.go +++ b/internal/control/reconciler/restart.go @@ -127,6 +127,42 @@ func (r *Reconciler) nextRestartLocked(model string) (string, bool) { return "", false } +// RestartPace is how many of a model's replicas restart at once. +type RestartPace struct { + // MaxSurge are replaced beside themselves at once: their replacements + // placed first, then they stop. Zero is one. + MaxSurge int + // MaxUnavailable are stopped first at once, their replacements starting + // in their place: an outage for each while it loads, needing no room. + MaxUnavailable int +} + +// width is how many may be restarting at once. +func (p RestartPace) width() int { + surge := p.MaxSurge + if surge <= 0 && p.MaxUnavailable <= 0 { + surge = 1 + } + return surge + max(p.MaxUnavailable, 0) +} + +// RestartPaced queues replicas like Restart (or RestartInPlace), as many of +// each model's at once as pace says. +func (r *Reconciler) RestartPaced(pace RestartPace, inPlace bool, ids ...string) error { + if err := r.restart(inPlace, ids); err != nil { + return err + } + r.mu.Lock() + for _, id := range ids { + if rep, ok := r.replicas[id]; ok { + r.restartPace[rep.Model] = pace + } + } + r.mu.Unlock() + r.Notify() + return nil +} + // ErrNoSuchReplica means no ready replica has that id. var ErrNoSuchReplica = errors.New("no ready replica with that id")