diff --git a/CHANGELOG.md b/CHANGELOG.md index 68a1af0..221c30b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -64,6 +64,15 @@ APIs change between minors without a deprecation period. four, as the room left holds, and its memory figure is for that pool. A plan says what the model was trained for (`trained`). Strata stays at its 32768; vLLM's form is unchanged. +- **A model that failed to start was started in the same place again.** + Placement given the same inputs chooses the same node and card, so an + on-demand model whose replica failed to load was retried there, past each + backoff, until it was parked, with another card or node free all along. + The control plane now remembers where a model's replicas failed (node and + cards, up to eight places) and places the next one elsewhere when anywhere + else fits; with nowhere else it is tried there as before. Coming up, a + change to its definition or `ctl retry` forgets the places. The log says + `placing a model away from where it failed to start`, with both places. - **Deleted model files came back after a control plane restart.** An agent registers with the files its startup scan found, and sends the list again only when it changes; a file deleted after the agent started was listed diff --git a/internal/control/placement/placement.go b/internal/control/placement/placement.go index 944a66b..d8f114f 100644 --- a/internal/control/placement/placement.go +++ b/internal/control/placement/placement.go @@ -154,6 +154,9 @@ func (r Result) Explain() string { // Scheduler places replicas onto an inventory. type Scheduler struct { inv *inventory.Store + // shun is where a replica of the model being placed failed to start; + // set on a copy by Away, never on the scheduler the reconciler shares. + shun []Spot // now is the clock power readings are judged fresh by; nil is the wall // clock. now func() time.Time @@ -162,6 +165,41 @@ type Scheduler struct { // New builds a scheduler over an inventory. func New(inv *inventory.Store) *Scheduler { return &Scheduler{inv: inv} } +// Spot is where a replica ran: a node, and the cards it held there (none +// for one on the node's CPU). +type Spot struct { + Node string + GPUs []int +} + +// Away is this scheduler placing nowhere in spots: not those cards, nor a +// CPU replica on those nodes. A copy, so the one shared is not changed. +// +// For a model that failed to start somewhere: the same inputs pick the same +// node and the same card every time, so without it every retry went back to +// where the last one had failed until the model was parked, with another +// card or node free all along. +func (s *Scheduler) Away(spots []Spot) *Scheduler { + c := *s + c.shun = spots + return &c +} + +// shunned is the cards of node n a model failed on, and whether it failed +// on the node's CPU. +func (s *Scheduler) shunned(node string) (gpus []int, cpu bool) { + for _, sp := range s.shun { + if sp.Node != node { + continue + } + if len(sp.GPUs) == 0 { + cpu = true + } + gpus = append(gpus, sp.GPUs...) + } + return gpus, cpu +} + // Place picks the first profile, in priority order, that fits. // // avoid lists nodes already hosting this model, so replicas of one model spread @@ -686,7 +724,12 @@ func (s *Scheduler) eligibleNodes(p Profile, fabric string, avoid []string, reje } } + badGPUs, badCPU := s.shunned(n.ID) if p.CPUOnly() { + if badCPU { + reject("a replica of it failed to start here; trying elsewhere first") + continue + } // Every ready node can run it; there is no card to be free. c := candidate{node: n} if why := budget.check(n, s.shardWatts(p, c)); why != "" { @@ -703,8 +746,16 @@ func (s *Scheduler) eligibleNodes(p Profile, fabric string, avoid []string, reje if g.Class() != p.GPUClass || !slices.Contains(freeIdx, g.Index) { continue } + if slices.Contains(badGPUs, g.Index) { + continue + } free = append(free, g) } + if len(free) < p.GPUsPerNode && len(badGPUs) > 0 { + reject(fmt.Sprintf("has %d free %s GPUs besides the ones a replica of it failed to start on, needs %d", + len(free), p.GPUClass, p.GPUsPerNode)) + continue + } if len(free) < p.GPUsPerNode { reject(fmt.Sprintf("has %d free %s GPUs, needs %d", len(free), p.GPUClass, p.GPUsPerNode)) continue diff --git a/internal/control/reconciler/away_test.go b/internal/control/reconciler/away_test.go new file mode 100644 index 0000000..b3fbe72 --- /dev/null +++ b/internal/control/reconciler/away_test.go @@ -0,0 +1,80 @@ +package reconciler + +import ( + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +// failAndRetry fails the last replica started, and returns where the next +// one is started once the backoff is over. +func failAndRetry(t *testing.T, h *harness) ShardCommand { + t.Helper() + started := h.disp.startedShards() + if len(started) == 0 { + t.Fatal("nothing was started") + } + s := started[len(started)-1] + h.r.OnShardStatus(s.ReplicaID, s.ShardID, supervisor.Failed, "the engine exited") + h.r.Reconcile(t.Context()) // reaps it + h.disp.reset() + h.advance(DefaultConfig().MaxBackoff) + h.r.Reconcile(t.Context()) + again := h.disp.startedShards() + if len(again) == 0 { + t.Fatal("it was not tried again") + } + return again[0] +} + +func oneCard() catalog.Model { + return testModel(1, catalog.Profile{Name: "one", GPUClass: "h100-80g", GPUsPerNode: 1}) +} + +func TestAReplicaThatFailedIsTriedOnAnotherNode(t *testing.T) { + // The same inputs give the same node every time: an on-demand 3D model + // that failed to load was started on that node again, and again, until + // it was parked, with another node free all along. + h := newHarness(t, oneCard(), gpuNode("1", 1), gpuNode("2", 1)) + h.r.Reconcile(t.Context()) + first := h.disp.startedShards()[0].NodeID + + if next := failAndRetry(t, h); next.NodeID == first { + t.Errorf("tried on node %s again, where it had just failed", first) + } +} + +func TestAReplicaThatFailedIsTriedOnAnotherCardOfTheNode(t *testing.T) { + h := newHarness(t, oneCard(), gpuNode("1", 2)) + h.r.Reconcile(t.Context()) + first := h.disp.startedShards()[0].Spec.GPUIndices + + next := failAndRetry(t, h) + if len(next.Spec.GPUIndices) != 1 || len(first) != 1 || next.Spec.GPUIndices[0] == first[0] { + t.Errorf("tried on card %v again, where it had just failed", next.Spec.GPUIndices) + } +} + +func TestWithNowhereElseItIsTriedWhereItFailed(t *testing.T) { + // A preference, not a rule: a model that has one place is not left + // unplaced because it failed there once. + h := newHarness(t, oneCard(), gpuNode("1", 1)) + h.r.Reconcile(t.Context()) + if next := failAndRetry(t, h); next.NodeID != "1" { + t.Errorf("started on %q", next.NodeID) + } +} + +func TestComingUpForgetsWhereItFailed(t *testing.T) { + h := newHarness(t, oneCard(), gpuNode("1", 1), gpuNode("2", 1)) + h.r.Reconcile(t.Context()) + next := failAndRetry(t, h) + h.r.OnShardStatus(next.ReplicaID, next.ShardID, supervisor.Ready, "") + h.r.mu.Lock() + left := len(h.r.failedAt["llama"]) + h.r.mu.Unlock() + if left != 0 { + t.Errorf("a model that came up still remembers %d failures' places", left) + } +} diff --git a/internal/control/reconciler/health.go b/internal/control/reconciler/health.go index 006b3eb..d2a4761 100644 --- a/internal/control/reconciler/health.go +++ b/internal/control/reconciler/health.go @@ -61,6 +61,7 @@ func (r *Reconciler) Retry(model string) bool { delete(r.parked, model) r.attempts[model] = 0 delete(r.nextAttempt, model) + delete(r.failedAt, model) r.mu.Unlock() r.Notify() return was diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index 020a6fe..bafc766 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -7,6 +7,8 @@ import ( "log/slog" "math/rand/v2" "slices" + "strconv" + "strings" "sync" "sync/atomic" "time" @@ -251,6 +253,9 @@ type Reconciler struct { nextAttempt map[string]time.Time // attempts counts consecutive failures per model, for the backoff. attempts map[string]int + // failedAt is where a model's replicas failed to start since it last + // came up, most recent last: the next one is placed elsewhere if it can. + failedAt map[string][]placement.Spot // lastExplain keeps the most recent placement failure per model, so // `ctl explain-placement` can answer without re-running placement. lastExplain map[string]placement.Result @@ -334,6 +339,7 @@ func New(cfg Config, cat *catalog.Live, inv *inventory.Store, disp Dispatcher, l replicas: map[string]*Replica{}, nextAttempt: map[string]time.Time{}, attempts: map[string]int{}, + failedAt: map[string][]placement.Spot{}, reservedPorts: map[int]struct{}{}, wanted: map[string]time.Time{}, coldStart: map[string]time.Duration{}, @@ -726,6 +732,7 @@ func (r *Reconciler) reapFailed(ctx context.Context) { // replica is about to disappear, and the backoff has to survive it. r.attempts[rep.Model]++ r.nextAttempt[rep.Model] = r.now().Add(r.backoff(r.attempts[rep.Model])) + r.noteFailedAt(rep) if n := r.attempts[rep.Model]; n >= r.cfg.ParkAfter { if _, already := r.parked[rep.Model]; !already { m, _ := r.cat.Load().Resolve(rep.Model) @@ -875,6 +882,7 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) { delete(r.parked, m.Name) r.attempts[m.Name] = 0 delete(r.nextAttempt, m.Name) + delete(r.failedAt, m.Name) isParked, blockedUntil = false, time.Time{} r.log.Info("model unparked: its definition changed", "model", m.Name) } @@ -1002,7 +1010,29 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin return notTried } } - res := r.sched.Place(profiles, avoid) + // Away from where it last failed to start, when there is anywhere else: + // the same inputs give the same node and card, and an on-demand model + // that failed there was otherwise started there again until it was + // parked. Only a preference too: nowhere else, and it is tried there. + r.mu.Lock() + failed := slices.Clone(r.failedAt[m.Name]) + r.mu.Unlock() + var res placement.Result + if len(failed) > 0 { + away := r.sched.Away(failed) + res = away.Place(profiles, avoid) + if res.Placement == nil && len(avoid) > 0 { + res = away.Place(profiles, nil) + } + pspan.SetAttributes(attribute.Bool("llmgw.away_from_failure", res.Placement != nil)) + if res.Placement != nil { + r.log.Info("placing a model away from where it failed to start", "model", m.Name, + "failed_on", spotsText(failed), "now_on", spotsText(spotsOf(res.Placement))) + } + } + if res.Placement == nil { + res = r.sched.Place(profiles, avoid) + } if res.Placement == nil && len(avoid) > 0 { // Spreading across failure domains is a preference, not a rule. If it // is the only thing standing in the way, take the placement anyway: @@ -1054,6 +1084,51 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin return placed } +// maxFailedSpots is how many places a model's failures are remembered at. +const maxFailedSpots = 8 + +// noteFailedAt remembers where a failed replica ran. Called under r.mu. +func (r *Reconciler) noteFailedAt(rep *Replica) { + spots := r.failedAt[rep.Model] + for _, sh := range rep.Shards { + sp := placement.Spot{Node: sh.NodeID, GPUs: slices.Clone(sh.GPUIndices)} + spots = slices.DeleteFunc(spots, func(o placement.Spot) bool { + return o.Node == sp.Node && slices.Equal(o.GPUs, sp.GPUs) + }) + spots = append(spots, sp) + } + if len(spots) > maxFailedSpots { + spots = spots[len(spots)-maxFailedSpots:] + } + r.failedAt[rep.Model] = spots +} + +// spotsOf is where a placement puts its shards. +func spotsOf(pl *placement.Placement) []placement.Spot { + var out []placement.Spot + for _, n := range pl.Nodes { + out = append(out, placement.Spot{Node: n.NodeID, GPUs: n.GPUIndices}) + } + return out +} + +// spotsText is spots as a log reads them: "node01 gpu 0,1; node02 cpu". +func spotsText(spots []placement.Spot) string { + parts := make([]string, 0, len(spots)) + for _, sp := range spots { + if len(sp.GPUs) == 0 { + parts = append(parts, sp.Node+" cpu") + continue + } + gs := make([]string, len(sp.GPUs)) + for i, g := range sp.GPUs { + gs[i] = strconv.Itoa(g) + } + parts = append(parts, sp.Node+" gpu "+strings.Join(gs, ",")) + } + return strings.Join(parts, "; ") +} + // scaleOutcome is what one scaleUp did. type scaleOutcome int @@ -1433,6 +1508,7 @@ func (r *Reconciler) markShard(replicaID, shardID string, st supervisor.State, e // counting from scratch rather than from an old streak. r.attempts[rep.Model] = 0 delete(r.nextAttempt, rep.Model) + delete(r.failedAt, rep.Model) if !rep.CreatedAt.IsZero() { r.coldStart[rep.Model] = r.now().Sub(rep.CreatedAt) }