From 12464ce0f6e31c5affef19cf8de70ae16b64ee78 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Fri, 30 Oct 2026 11:14:02 +0100 Subject: [PATCH] Cluster upgrades: nodes told together with progress said, done when back on the version, the download kept past a broken stream Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 12 ++ deploy/systemd/llmgw-agent.service | 3 + internal/agent/agent.go | 4 +- internal/agent/upgrade.go | 24 +++- internal/cli/ctl_upgrade.go | 117 ++++++++++++++++---- internal/cli/ctl_upgrade_test.go | 29 ++++- internal/cli/serverinit_cluster.go | 3 + internal/cli/tui_actions.go | 20 +++- internal/control/agentsvc/heartbeat_test.go | 23 ++++ internal/control/agentsvc/service.go | 44 ++++++-- 10 files changed, 244 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7393ac3..50a97de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,18 @@ APIs change between minors without a deprecation period. ### Fixed +- **A cluster upgrade that sat silent, or stopped on a node that had upgraded.** + - The nodes of a wave are told together, each saying when it is done, and what is still awaited is said every 30 + seconds; before, each was told in turn, up to five minutes each, with nothing on the screen. + - The control plane takes the node registering again on the version as done: the agent's answer, sent a moment + before it exits, could be lost, and a node that had upgraded was failed as "older than v0.10.4". + - The agent's download no longer ends with its connection (a control plane restarting, leadership moving), and the + control plane waits longer than the download may take; a second request while one runs is refused. + - A cordoned or draining node upgraded is done; a node whose idle model stopped meanwhile is done once nothing on it + is starting; a node not connected is passed over and named at the end rather than stopping every node after it. + - The console's node upgrade waits as long as the control plane, and offers no development version. + - systemd never gives up restarting the agent (`StartLimitIntervalSec=0`), in units written from now on. + - **Strata.** One named by hand in `agent.yaml` was never reported, so never placed on; its engine, which leaves the server's process group, is now stopped with it; Pascal and Volta, which its CUDA 12 build runs, were refused. - **A name with a slash in it, asked through a standby control plane.** `llmgw ctl kb delete Demo/hr` on a standby diff --git a/deploy/systemd/llmgw-agent.service b/deploy/systemd/llmgw-agent.service index 75cadfa..882c2f9 100644 --- a/deploy/systemd/llmgw-agent.service +++ b/deploy/systemd/llmgw-agent.service @@ -2,6 +2,9 @@ Description=llmgw node agent After=network-online.target Wants=network-online.target +# Never given up on: a new release that fails its first starts, or an upgrade +# that restarts it twice in a minute, would otherwise leave the node off. +StartLimitIntervalSec=0 [Service] ExecStart=/usr/local/bin/llmgw agent --config /etc/llmgw/agent.yaml diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 1540dca..32e6b5e 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -47,7 +47,9 @@ import ( // Agent runs on a GPU node. type Agent struct { - cfg Config + // upgrading is true while an upgrade downloads. + upgrading atomic.Bool + cfg Config // answered says the current session heard from its control plane, which // starts the reconnect backoff again when it ends. answered atomic.Bool diff --git a/internal/agent/upgrade.go b/internal/agent/upgrade.go index 80fa591..7b803e2 100644 --- a/internal/agent/upgrade.go +++ b/internal/agent/upgrade.go @@ -35,12 +35,25 @@ func (a *Agent) upgrade(ctx context.Context, m *llmgwv1.UpgradeAgent, out chan<- select { case out <- &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Upgrade{Upgrade: reply}}: case <-ctx.Done(): + case <-time.After(5 * time.Second): } } + // One at a time: a second request while one downloads would race it to + // the same file. + if !a.upgrading.CompareAndSwap(false, true) { + reply.Error = "an upgrade is already under way on this node" + send() + return + } + defer a.upgrading.Store(false) a.log.Info("upgrading", "to", m.GetVersion(), "from", a.cfg.Version) self, err := upgradeTarget() if err == nil { - dctx, cancel := context.WithTimeout(ctx, 10*time.Minute) + // Not the stream's context: a stream that breaks while the release + // downloads — a control plane restarting, leadership moving — cancelled + // the upgrade with nobody told. The node registering again on the + // version is what the control plane watches for. + dctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), upgradeDownload) _, err = release.Install(dctx, nil, m.GetReleases(), m.GetVersion(), runtime.GOOS, runtime.GOARCH, self) cancel() } @@ -53,6 +66,13 @@ func (a *Agent) upgrade(ctx context.Context, m *llmgwv1.UpgradeAgent, out chan<- send() a.log.Info("upgraded; restarting into the new version", "to", m.GetVersion()) // A moment for the reply to leave before the stream goes. - time.Sleep(500 * time.Millisecond) + time.Sleep(upgradeLinger) exitForUpgrade() } + +// upgradeDownload bounds an upgrade's download; the control plane waits +// longer (agentsvc.UpgradeWait). upgradeLinger is the moment its answer has +// to leave before the agent exits; a variable so tests need not wait it. +const upgradeDownload = 10 * time.Minute + +var upgradeLinger = 2 * time.Second diff --git a/internal/cli/ctl_upgrade.go b/internal/cli/ctl_upgrade.go index a87c9f5..323920d 100644 --- a/internal/cli/ctl_upgrade.go +++ b/internal/cli/ctl_upgrade.go @@ -11,11 +11,13 @@ import ( "runtime" "slices" "strings" + "sync" "time" "github.com/spf13/cobra" "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/onboard" "gitea.mynodes.xyz/ilker/llmgw/internal/release" "gitea.mynodes.xyz/ilker/llmgw/internal/version" @@ -244,6 +246,7 @@ func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient, } slices.SortFunc(nodes, func(a, b adminapi.NodeView) int { return strings.Compare(a.ID, b.ID) }) var todo []adminapi.NodeView + var away []string skipped := 0 for _, n := range nodes { if len(only) > 0 && !slices.Contains(only, n.ID) { @@ -253,8 +256,17 @@ func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient, skipped++ continue } + // A node that is not connected cannot be told, and one tried first + // stopped every run at it: passed over, and named at the end. + if n.State == "lost" || n.State == "unreachable" { + away = append(away, n.ID) + continue + } todo = append(todo, n) } + if len(away) > 0 { + fmt.Fprintf(out, "Passed over, not connected: %s\n", strings.Join(away, ", ")) + } parallel = max(parallel, 1) start, size, done := time.Now(), 1, 0 for len(todo) > 0 { @@ -296,7 +308,10 @@ func upgradeNodes(ctx context.Context, out io.Writer, client func() *ctlClient, } // upgradeWave tells every node in the wave at once, and waits until each is -// back on v, ready, with as many ready replicas as it had. +// back on v and serving what it served. +// +// Told together, not one after another: each answer can take minutes, and a +// wave of twenty waited for them in turn with nothing on the screen. func upgradeWave(ctx context.Context, out io.Writer, client func() *ctlClient, v string, wave []adminapi.NodeView, perNode time.Duration, ) error { @@ -308,35 +323,61 @@ func upgradeWave(ctx context.Context, out io.Writer, client func() *ctlClient, v } fmt.Fprintf(out, "== %d node(s): %s\n", len(wave), abbreviate(ids, 8)) body, _ := json.Marshal(adminapi.UpgradeRequest{Version: v}) - var failed []string + var ( + mu sync.Mutex + failed []string + wg sync.WaitGroup + ) for _, n := range wave { - // The control plane waits for the agent to fetch and check the - // release, up to five minutes; fifteen seconds gave up on a node - // that was upgrading. - if err := client().patient(6*time.Minute).post(ctx, "/api/v1/nodes/"+n.ID+"/upgrade", body, nil); err != nil { - failed = append(failed, fmt.Sprintf("%s: %v", n.ID, err)) - } + wg.Add(1) + go func() { + defer wg.Done() + // The control plane waits for the agent's answer, or for it to + // come back on v, longer than the agent gives its download. + err := client().patient(agentsvc.UpgradeWait+time.Minute).post(ctx, "/api/v1/nodes/"+n.ID+"/upgrade", body, nil) + mu.Lock() + defer mu.Unlock() + if err != nil { + failed = append(failed, fmt.Sprintf("%s: %v", n.ID, err)) + fmt.Fprintf(out, " %s: not upgraded: %v\n", n.ID, err) + return + } + fmt.Fprintf(out, " %s: on %s, coming back\n", n.ID, v) + }() } - if len(failed) > 0 { - return fmt.Errorf("not told to upgrade:\n %s\n`llmgw ctl logs NODE` says more", strings.Join(failed, "\n ")) - } - waiting := slices.Clone(ids) + wg.Wait() + waiting := slices.DeleteFunc(slices.Clone(ids), func(id string) bool { + return slices.ContainsFunc(failed, func(f string) bool { return strings.HasPrefix(f, id+": ") }) + }) deadline := time.Now().Add(perNode) - for { + backAt := map[string]time.Time{} + said := time.Now() + for len(waiting) > 0 { var now []adminapi.NodeView _ = client().get(ctx, "/api/v1/nodes", &now) - after := readyReplicasByNode(ctx, client) + var reps []adminapi.ReplicaView + _ = client().get(ctx, "/api/v1/replicas", &reps) + after, starting := replicasByNode(reps) state := map[string]adminapi.NodeView{} for _, x := range now { state[x.ID] = x } waiting = slices.DeleteFunc(waiting, func(id string) bool { cur := state[id] - return cur.AgentVersion == v && cur.State == "ready" && after[id] >= before[id] + // Connected and on v: ready, or cordoned or draining as it + // was, which an upgrade does not undo. + if cur.AgentVersion != v || (cur.State != "ready" && cur.State != "cordoned" && cur.State != "draining") { + return false + } + if _, ok := backAt[id]; !ok { + backAt[id] = time.Now() + } + // Serving what it served; or, a while on, nothing on it still + // starting: a model that went idle meanwhile is not one lost. + return after[id] >= before[id] || (time.Since(backAt[id]) > 90*time.Second && starting[id] == 0) }) if len(waiting) == 0 { - fmt.Fprintf(out, " done in %s\n", time.Since(start).Round(time.Second)) - return nil + break } if time.Now().After(deadline) { var why []string @@ -345,11 +386,49 @@ func upgradeWave(ctx context.Context, out io.Writer, client func() *ctlClient, v why = append(why, fmt.Sprintf("%s (version %q, state %q, replicas %d/%d)", id, cur.AgentVersion, cur.State, after[id], before[id])) } - return fmt.Errorf("not back within %s:\n %s\n`llmgw ctl logs NODE` says why, and each keeps its "+ - "previous binary beside the new one", perNode, strings.Join(why, "\n ")) + failed = append(failed, fmt.Sprintf("not back within %s: %s", perNode, strings.Join(why, ", "))) + break + } + if time.Since(said) >= 30*time.Second { + var w []string + for _, id := range waiting { + cur := state[id] + w = append(w, fmt.Sprintf("%s (%s, %s, replicas %d/%d)", id, orUnknown(cur.AgentVersion), cur.State, + after[id], before[id])) + } + fmt.Fprintf(out, " waiting for %s\n", abbreviate(w, 8)) + said = time.Now() } time.Sleep(2 * time.Second) } + if len(failed) > 0 { + return fmt.Errorf("not upgraded:\n %s\n`llmgw ctl logs NODE` says why; each keeps its previous binary beside the new one", + strings.Join(failed, "\n ")) + } + fmt.Fprintf(out, " done in %s\n", time.Since(start).Round(time.Second)) + return nil +} + +// replicasByNode counts each node's ready replicas, and those still placing +// or starting. +func replicasByNode(reps []adminapi.ReplicaView) (ready, starting map[string]int) { + ready, starting = map[string]int{}, map[string]int{} + for _, r := range reps { + seen := map[string]bool{} + for _, sh := range r.Shards { + if seen[sh.Node] { + continue + } + seen[sh.Node] = true + switch r.State { + case "ready": + ready[sh.Node]++ + case "placing", "starting": + starting[sh.Node]++ + } + } + } + return ready, starting } // abbreviate lists up to n names, and says how many more. diff --git a/internal/cli/ctl_upgrade_test.go b/internal/cli/ctl_upgrade_test.go index f65376b..a8ded02 100644 --- a/internal/cli/ctl_upgrade_test.go +++ b/internal/cli/ctl_upgrade_test.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "slices" "strings" "sync" "testing" @@ -21,6 +22,8 @@ type fakeFleet struct { versions map[string]string stuck map[string]bool told []string + // states are nodes' states other than ready. + states map[string]string } func (f *fakeFleet) server(t *testing.T) *httptest.Server { @@ -31,7 +34,11 @@ func (f *fakeFleet) server(t *testing.T) *httptest.Server { case r.URL.Path == "/api/v1/nodes": var out []adminapi.NodeView for id, v := range f.versions { - out = append(out, adminapi.NodeView{ID: id, State: "ready", AgentVersion: v}) + st := "ready" + if f.states[id] != "" { + st = f.states[id] + } + out = append(out, adminapi.NodeView{ID: id, State: st, AgentVersion: v}) } _ = json.NewEncoder(w).Encode(out) case r.URL.Path == "/api/v1/replicas": @@ -201,3 +208,23 @@ func TestUpgradingAControlPlaneAloneSaysItsNodesWereNot(t *testing.T) { t.Errorf("said something with every node on the version: %s", out.String()) } } + +// A cordoned node upgraded is done, as it was; a node not connected is passed +// over rather than stopping every node after it. +func TestAnUpgradeTakesACordonedNodeAndPassesOverOneAway(t *testing.T) { + f := &fakeFleet{versions: map[string]string{"a": "v1.0.0", "b": "v1.0.0", "c": "v1.0.0"}, + states: map[string]string{"a": "lost", "b": "cordoned"}} + srv := f.server(t) + var out bytes.Buffer + err := upgradeNodes(t.Context(), &out, func() *ctlClient { return newCtlClient(srv.URL) }, "v1.0.1", nil, 5*time.Second, 2) + if slices.Contains(f.told, "a") || !slices.Contains(f.told, "b") || !slices.Contains(f.told, "c") { + t.Errorf("told %v", f.told) + } + if !strings.Contains(out.String(), "Passed over, not connected: a") { + t.Errorf("output:\n%s", out.String()) + } + // The one away is named at the end, behind. + if err == nil || !strings.Contains(err.Error(), "a (v1.0.0, lost)") { + t.Errorf("err %v", err) + } +} diff --git a/internal/cli/serverinit_cluster.go b/internal/cli/serverinit_cluster.go index bcf307d..856d2a0 100644 --- a/internal/cli/serverinit_cluster.go +++ b/internal/cli/serverinit_cluster.go @@ -64,6 +64,9 @@ const agentUnit = `[Unit] Description=llmgw node agent After=network-online.target Wants=network-online.target +# Never given up on: a new release that fails its first starts, or an upgrade +# that restarts it twice in a minute, would otherwise leave the node off. +StartLimitIntervalSec=0 [Service] ExecStart=/usr/local/bin/llmgw agent --config /etc/llmgw/agent.yaml diff --git a/internal/cli/tui_actions.go b/internal/cli/tui_actions.go index 00367d9..2439820 100644 --- a/internal/cli/tui_actions.go +++ b/internal/cli/tui_actions.go @@ -16,6 +16,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control" "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/facility" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve" @@ -571,9 +572,15 @@ func (m consoleModel) nodeMenu(n adminapi.NodeView) *menuDialog { menuItem{"Upgrade its agent…", func(m *consoleModel) tea.Cmd { m.openForm(&formDialog{title: "Upgrade " + n.ID, note: "It fetches the release for its platform, checks it, swaps it in and restarts; engines keep serving.", - fields: []formField{textF("version", version.Version, "a release, like v0.11.1")}, + fields: []formField{textF("version", releaseOrEmpty(version.Version), "a release, like v0.11.1")}, submit: func(m *consoleModel, v []string) (tea.Cmd, error) { - return m.call("Upgrade "+n.ID, http.MethodPost, path+"/upgrade", adminapi.UpgradeRequest{Version: v[0]}, nil), nil + if !release.Tag.MatchString(v[0]) { + return nil, fmt.Errorf("a release is vMAJOR.MINOR.PATCH, like v0.15.3") + } + // As long as the control plane waits: the download, the + // restart, the node back on the version. + return m.callWaiting("Upgrade "+n.ID, http.MethodPost, path+"/upgrade", adminapi.UpgradeRequest{Version: v[0]}, + func(json.RawMessage) string { return n.ID + " is on " + v[0] + "." }, agentsvc.UpgradeWait+time.Minute), nil }}) return nil }}, @@ -1056,3 +1063,12 @@ func purgeSays(raw json.RawMessage) string { } return s } + +// releaseOrEmpty is this build's version when it is a release, else empty: +// a development build's is not one a node can fetch. +func releaseOrEmpty(v string) string { + if release.Tag.MatchString(v) { + return v + } + return "" +} diff --git a/internal/control/agentsvc/heartbeat_test.go b/internal/control/agentsvc/heartbeat_test.go index a21821c..a9c02fe 100644 --- a/internal/control/agentsvc/heartbeat_test.go +++ b/internal/control/agentsvc/heartbeat_test.go @@ -3,6 +3,7 @@ package agentsvc import ( "log/slog" "testing" + "time" llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" @@ -284,6 +285,28 @@ func TestAnUpgradeWaitsForTheAgentsAnswer(t *testing.T) { } } +// The agent's answer lost — it exits a moment after sending it — and the node +// back on the version: the upgrade is done, not five minutes of waiting and a +// failure blaming an old agent. +func TestAnUpgradeIsDoneWhenTheNodeComesBackOnTheVersion(t *testing.T) { + s := newTestService() + nodeWith(s, inventory.Node{ID: "n1", State: inventory.Ready, AgentVersion: "v1.2.2"}) + c := &conn{nodeID: "n1", send: make(chan *llmgwv1.ControlMessage, 4), closed: make(chan struct{})} + s.conns["n1"] = c + done := make(chan error, 1) + go func() { done <- s.Upgrade(t.Context(), "n1", "v1.2.3", "https://r") }() + <-c.send + nodeWith(s, inventory.Node{ID: "n1", State: inventory.Ready, AgentVersion: "v1.2.3"}) + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(10 * time.Second): + t.Fatal("still waiting for an answer the node will never send") + } +} + func TestAnotherTenantOnACardIsJudgedEveryHeartbeat(t *testing.T) { // Found on cuda01: ComfyUI held 5 GB when the agent started, so the card // was kept out of placement. ComfyUI let go of it and the card stayed diff --git a/internal/control/agentsvc/service.go b/internal/control/agentsvc/service.go index d3dafe7..70c3b41 100644 --- a/internal/control/agentsvc/service.go +++ b/internal/control/agentsvc/service.go @@ -1158,21 +1158,45 @@ func (s *Service) Upgrade(ctx context.Context, nodeID, version, releases string) return err } s.log.Info("told a node to upgrade", "node", nodeID, "version", version) - // A download of some tens of megabytes on a slow link, and no longer: an - // agent too old to know this message never answers. - ctx, cancel := context.WithTimeout(ctx, 5*time.Minute) + // Longer than the agent gives its download, so its own answer — done, or + // why not — is what is heard. Its answer can still be lost: it exits a + // moment after sending it, and a stream that breaks takes it along. The + // node registering again on the version says the same, and is watched for. + ctx, cancel := context.WithTimeout(ctx, UpgradeWait) defer cancel() - select { - case r := <-ch: - if r.GetError() != "" { - return errors.New(r.GetError()) + tick := time.NewTicker(2 * time.Second) + defer tick.Stop() + for { + select { + case r := <-ch: + if r.GetError() != "" { + return errors.New(r.GetError()) + } + return nil + case <-tick.C: + if n, ok := s.inv.Get(nodeID); ok && n.AgentVersion == version { + return nil + } + case <-ctx.Done(): + n, ok := s.inv.Get(nodeID) + switch { + case !slices.Contains(s.ConnectedNodes(), nodeID): + return fmt.Errorf("node %s went away during the upgrade and has not come back: on it, "+ + "`systemctl status llmgw-agent` and `journalctl -u llmgw-agent` say why", nodeID) + case ok && n.AgentVersion != "": + return fmt.Errorf("node %s did not answer the upgrade and still runs %s: its download is slow or stuck, "+ + "or its agent is older than v0.10.4, which cannot upgrade itself", nodeID, n.AgentVersion) + default: + return fmt.Errorf("node %s did not answer the upgrade; an agent older than v0.10.4 cannot upgrade itself", nodeID) + } } - return nil - case <-ctx.Done(): - return fmt.Errorf("node %s did not answer the upgrade; an agent older than v0.10.4 cannot upgrade itself", nodeID) } } +// UpgradeWait is how long the control plane waits for an agent to upgrade: +// longer than the agent gives its download (agent.upgradeDownload). +const UpgradeWait = 11 * time.Minute + // wearFrom is a card's wear, when the agent read it. func wearFrom(g *llmgwv1.GPUSensor) *inventory.GPUWear { if !g.GetHasWear() {