From cfc38d35cc23200a24d870e8e335a8d9062ce5d7 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Mon, 19 Oct 2026 12:05:13 +0200 Subject: [PATCH] A node's console opens on what the machine is doing sudo llmgw on a node showed only agent.yaml, the manual, wipe and leave. It now opens on an Overview read on the node itself: the agent's state and version, the control planes it dials and whether they answer, each GPU with the models on it, the engines from the supervisor's state files, the free space for weights and engines, and the agent's warnings of the last hour. No admin token is needed, every command is bounded by two seconds, and a missing tool is said in a line. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 4 + docs/console.md | 21 +- internal/agent/discovery/sensors.go | 7 + internal/cli/tui_console.go | 29 +- internal/cli/tui_node_overview.go | 715 +++++++++++++++++++++++++ internal/cli/tui_node_overview_test.go | 206 +++++++ 6 files changed, 964 insertions(+), 18 deletions(-) create mode 100644 internal/cli/tui_node_overview.go create mode 100644 internal/cli/tui_node_overview_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 0324524..4de8947 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,10 @@ APIs change between minors without a deprecation period. - **A node's page says what it runs.** It now shows the agent's version, marked when the control plane runs another, and the engines with their versions. Each GPU row now names the models loaded on that card, and each engine row its GPUs. +- **`sudo llmgw` on a node opens on what the machine is doing.** A new Overview screen shows the agent's state and + version, each control plane it dials and whether it answers, every GPU with the models on it, the engines running + here, the free space for weights and engines, and the agent's warnings of the last hour. It is read on the node + itself, every three seconds, so it needs no admin token and works while the control plane is out of reach. ### Fixed diff --git a/docs/console.md b/docs/console.md index 6176783..e26e578 100644 --- a/docs/console.md +++ b/docs/console.md @@ -2,8 +2,8 @@ `sudo llmgw`, with nothing after it, on a machine llmgw runs on. On a machine with nothing installed it opens the setup screens; on a control plane, the -console described here; on a node, the node's own console, which edits -`agent.yaml`. Everything on these screens can also be done by command — each +console described here; on a node, the node's own console, which shows +what runs on that machine and edits `agent.yaml`. Everything on these screens can also be done by command — each screen says which — so a script never needs them and a person never needs to remember the commands. @@ -357,5 +357,18 @@ to restart it. ## The node's console -On a node, `sudo llmgw` opens `agent.yaml` in the same settings view, and the -manual. The cluster is watched and changed from a control plane. +On a node, `sudo llmgw` opens on **Overview**: what this machine is doing, read +on the machine itself, so it works without an admin token and while the control +plane is out of reach. Refreshed every three seconds, it shows + +- llmgw-agent's state and since when, and the version of llmgw; +- each control plane the agent dials, from `agent.yaml` and the list the + leader last sent, and whether it can be reached; +- each GPU: how busy, its memory, its heat, and the models on it; +- the engines running here, from the agent's own records: replica, model, + engine, GPUs, port, whether the process is still there, and for how long; +- the free space where weights and engines are kept; +- the agent's warnings and errors of the last hour. + +Below it, `agent.yaml` in the same settings view, and the manual. The cluster +is watched and changed from a control plane. diff --git a/internal/agent/discovery/sensors.go b/internal/agent/discovery/sensors.go index d34dde9..f5e5d66 100644 --- a/internal/agent/discovery/sensors.go +++ b/internal/agent/discovery/sensors.go @@ -74,6 +74,13 @@ func NewSensorSampler() *SensorSampler { return &SensorSampler{sys: "/sys", smi: runSMI, now: time.Now} } +// SampleGPUs reads every card's heat, power and how busy it is now, without +// a sampler's ten seconds of memory: for a screen that asks every few seconds +// and wants what it asked for. Nil on a machine without nvidia-smi or a card. +func SampleGPUs(ctx context.Context) []GPUSensor { + return (&SensorSampler{smi: runSMI}).gpus(ctx) +} + func runSMI(ctx context.Context, fields string) ([]byte, error) { ctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() diff --git a/internal/cli/tui_console.go b/internal/cli/tui_console.go index 5494248..1f2cdcb 100644 --- a/internal/cli/tui_console.go +++ b/internal/cli/tui_console.go @@ -141,8 +141,8 @@ type consoleModel struct { detail bool scroll int - // node is the console of a node: its agent.yaml and nothing to watch, - // since the cluster is watched from the control plane. + // node is the console of a node: what runs on the machine and its + // agent.yaml, since the cluster is watched from the control plane. node bool // The settings. @@ -192,6 +192,11 @@ type consoleModel struct { docs docViewer mon monitorState det monitorState + + // nodeOv is a node's overview as last read; nodeBusy says a reading is + // under way. + nodeOv nodeOverview + nodeBusy bool } var usageGroupings = []string{"team", "project", "org"} @@ -200,9 +205,6 @@ func newConsoleModel(addr, file string) consoleModel { m := consoleModel{client: newCtlClient(addr), spin: newSpinner(), width: 100, height: 30, file: file, st: &storeHolder{file: file}, docs: newDocViewer()} m.node = strings.HasSuffix(file, "agent.yaml") - if m.node { - m.menu = cSettings - } m.loadSettings() return m } @@ -259,7 +261,7 @@ func (m *consoleModel) loadSettings() { func (m consoleModel) Init() tea.Cmd { if m.node { - return m.spin.Tick + return tea.Batch(m.fetchNodeOverview(), m.spin.Tick, tea.Tick(3*time.Second, func(time.Time) tea.Msg { return consoleTickMsg{} })) } return tea.Batch(m.fetch(), m.spin.Tick, tea.Tick(3*time.Second, func(time.Time) tea.Msg { return consoleTickMsg{} })) } @@ -322,7 +324,7 @@ func (m consoleModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { return m, cmd case consoleTickMsg: if m.node { - return m, nil + return m.nodeTick() } cmds := []tea.Cmd{m.fetch(), tea.Tick(3*time.Second, func(time.Time) tea.Msg { return consoleTickMsg{} })} if m.menu == cUsage { @@ -356,6 +358,8 @@ func (m consoleModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { cmds = append(cmds, m.fetchLogs(m.logs.node)) } return m, tea.Batch(cmds...) + case nodeOverviewMsg: + m.nodeOv, m.nodeBusy = nodeOverview(msg), false case consoleDataMsg: m.loaded = true if msg.err != nil { @@ -430,7 +434,7 @@ func (m consoleModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { func (m consoleModel) entryOK(i int) bool { if m.node { - return i == cSettings || i == cManual || i == cWipe || i == cLeave + return i == cOverview || i == cSettings || i == cManual || i == cWipe || i == cLeave } return i != cReplicate || !m.storeReplicated() } @@ -1106,9 +1110,6 @@ func (m consoleModel) rowCount() int { func (m consoleModel) View() string { host, _ := os.Hostname() nav := []navItem{{Label: "Watch", Heading: true}} - if m.node { - nav = nil - } for i := 0; i < cCount; i++ { if m.node && !m.entryOK(i) { continue @@ -1203,6 +1204,9 @@ func (m consoleModel) panel() (title, body, keys string) { } switch m.menu { case cOverview: + if m.node { + return m.nodeOverviewPanel(pw) + } var b bytes.Buffer c := &cobra.Command{} c.SetOut(&b) @@ -1676,9 +1680,6 @@ func runConsole(ctx context.Context, out io.Writer, file string) error { } } menu := cOverview - if strings.HasSuffix(file, "agent.yaml") { - menu = cSettings - } for { m := newConsoleModel(addr, file) m.menu = menu diff --git a/internal/cli/tui_node_overview.go b/internal/cli/tui_node_overview.go new file mode 100644 index 0000000..b7955e1 --- /dev/null +++ b/internal/cli/tui_node_overview.go @@ -0,0 +1,715 @@ +package cli + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "slices" + "sort" + "strconv" + "strings" + "sync" + "syscall" + "time" + + tea "github.com/charmbracelet/bubbletea" + "github.com/charmbracelet/lipgloss" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent" + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines" + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/provision" + "gitea.mynodes.xyz/ilker/llmgw/internal/version" +) + +// A node's overview: what a bare `sudo llmgw` opens on a machine that runs +// only llmgw-agent. Everything on it is read here — the agent's own files, +// systemd, the cards and the disks — because a node has no admin token, and +// the control plane may be the very thing it cannot reach. + +// nodeCmdTimeout bounds every command and check the overview runs: a screen +// refreshed every three seconds cannot wait on a tool that hangs. +const nodeCmdTimeout = 2 * time.Second + +// nodeControlPlanesFile is where the agent keeps the control planes the +// leader last named (controlPlanesFile in the agent package). +var nodeControlPlanesFile = "/var/lib/llmgw-agent/control-planes.json" + +// nodeProblemLines is how many of the agent's warnings the overview shows. +const nodeProblemLines = 6 + +type nodeOverviewMsg nodeOverview + +// nodeOverview is everything the overview shows, read in one go. +type nodeOverview struct { + at time.Time + version string + agent agentService + planes []planeCheck + gpus []nodeGPU + // gpuNote says why there are no GPU rows: no card, or no answer. + gpuNote string + // names is the cards' names by index, which do not change and are asked + // for once rather than every refresh. + names map[int]string + engines []engineRun + enginesNote string + disks []diskUse + problems []string + // problemsNote replaces the problems when the journal cannot be read. + problemsNote string +} + +// agentService is what systemd says of llmgw-agent. +type agentService struct { + state, sub string + since time.Time + // note replaces the rest when systemd cannot say. + note string +} + +// planeCheck is one control plane the agent dials, and whether it answers. +type planeCheck struct { + addr string + // learned is an address the leader named rather than one in agent.yaml. + learned bool + err string +} + +type nodeGPU struct { + index int + name string + // util is -1 where the driver does not report it. + util float64 + memUsedMB, memTotalMB float64 + tempC float64 + models []string +} + +// engineRun is one engine the supervisor started here, from its state file. +type engineRun struct { + replica, model, engine string + gpus []int + port, pid int + alive bool + started time.Time +} + +type diskUse struct { + what, path string + free, total uint64 + err string +} + +// nodeProbe reads this machine. Every path and tool is a field, so a test +// reads a directory of its own instead. +type nodeProbe struct { + configured []string + planesFile string + workDir string + weightsDir string + enginesDir string + root bool + names map[int]string + + run func(ctx context.Context, name string, args ...string) ([]byte, error) + dial func(ctx context.Context, addr string) error + gpus func(ctx context.Context) []discovery.GPUSensor + cards func(ctx context.Context) ([]discovery.GPU, error) + alive func(pid int) bool + statfs func(path string) (free, total uint64, err error) + now func() time.Time +} + +// newNodeProbe reads the machine agent.yaml describes, as the agent would. +func newNodeProbe(cfg any, names map[int]string) nodeProbe { + c, ok := cfg.(agent.Config) + if !ok { + c = agent.DefaultConfig() + } + var configured []string + if c.Server != "" { + configured = append(configured, c.Server) + } + configured = append(configured, c.Servers...) + // The same defaults the agent applies as it starts. + weights := c.WeightsCache + if weights == "" && len(c.ModelDirs) > 0 { + weights = c.ModelDirs[0] + } + if weights == "" { + weights = agent.DefaultWeightsCache + } + enginesDir := c.EnginesDir + if enginesDir == "" { + enginesDir = provision.DefaultRoot + } + work := c.Superv.WorkDir + if work == "" { + work = "/var/lib/llmgw-agent" + } + return nodeProbe{ + configured: configured, planesFile: nodeControlPlanesFile, workDir: work, + weightsDir: weights, enginesDir: enginesDir, root: os.Geteuid() == 0, names: names, + run: runLocal, dial: dialTCP, gpus: discovery.SampleGPUs, + cards: func(ctx context.Context) ([]discovery.GPU, error) { return discovery.DiscoverGPUsSMI(ctx, "") }, + alive: pidAlive, statfs: statfs, now: time.Now, + } +} + +// fetchNodeOverview reads the machine in the background. +func (m consoleModel) fetchNodeOverview() tea.Cmd { + p := newNodeProbe(m.cfg, m.nodeOv.names) + return func() tea.Msg { return nodeOverviewMsg(p.gather(context.Background())) } +} + +// nodeTick refreshes the overview while it is on screen, one reading at a +// time; the other screens of a node watch nothing. +func (m consoleModel) nodeTick() (consoleModel, tea.Cmd) { + tick := tea.Tick(3*time.Second, func(time.Time) tea.Msg { return consoleTickMsg{} }) + if m.menu != cOverview || m.nodeBusy { + return m, tick + } + m.nodeBusy = true + return m, tea.Batch(m.fetchNodeOverview(), tick) +} + +// gather reads everything at once, each part in its own goroutine, so the +// slowest part bounds the whole. +func (p nodeProbe) gather(ctx context.Context) nodeOverview { + ov := nodeOverview{version: version.Version} + var wg sync.WaitGroup + do := func(f func()) { + wg.Add(1) + go func() { defer wg.Done(); f() }() + } + var engines []engineRun + do(func() { ov.agent = p.agentState(ctx) }) + do(func() { ov.planes = p.checkPlanes(ctx) }) + do(func() { ov.gpus, ov.names, ov.gpuNote = p.readGPUs(ctx) }) + do(func() { engines, ov.enginesNote = readShardStates(p.workDir, p.alive) }) + do(func() { ov.disks = p.readDisks() }) + do(func() { ov.problems, ov.problemsNote = p.readProblems(ctx) }) + wg.Wait() + ov.engines = engines + ov.gpus = modelsOnGPUs(ov.gpus, engines) + ov.at = p.now() + return ov +} + +// --- the agent --- + +func (p nodeProbe) agentState(ctx context.Context) agentService { + out, err := p.run(ctx, "systemctl", "show", "llmgw-agent", + "-p", "LoadState", "-p", "ActiveState", "-p", "SubState", "-p", "ActiveEnterTimestamp") + if err != nil && len(out) == 0 { + return agentService{note: "systemctl is not available: " + err.Error()} + } + return parseSystemctlShow(string(out)) +} + +// parseSystemctlShow reads `systemctl show -p …`: one key=value a line. +func parseSystemctlShow(out string) agentService { + kv := map[string]string{} + for _, line := range strings.Split(out, "\n") { + if k, v, ok := strings.Cut(strings.TrimSpace(line), "="); ok { + kv[k] = v + } + } + if kv["LoadState"] == "not-found" { + return agentService{note: "llmgw-agent is not installed as a service here"} + } + s := agentService{state: kv["ActiveState"], sub: kv["SubState"]} + if s.state == "" { + s.note = "systemd does not say" + } + s.since = parseSystemdTime(kv["ActiveEnterTimestamp"]) + return s +} + +// parseSystemdTime reads systemd's "Mon 2026-10-19 08:12:03 CEST", in this +// machine's zone; zero when it is empty or n/a. +func parseSystemdTime(v string) time.Time { + v = strings.TrimSpace(v) + if v == "" || v == "n/a" { + return time.Time{} + } + for _, layout := range []string{"Mon 2006-01-02 15:04:05 MST", "Mon 2006-01-02 15:04:05 -0700", "Mon 2006-01-02 15:04:05"} { + if t, err := time.ParseInLocation(layout, v, time.Local); err == nil { + return t + } + } + return time.Time{} +} + +// --- the control planes --- + +// readControlPlanes is every address the agent dials, in its order: the +// configured ones, then those the leader named that agent.yaml lacks. +func readControlPlanes(file string, configured []string) []planeCheck { + var out []planeCheck + seen := map[string]bool{} + add := func(s string, learned bool) { + hp := hostPort(s) + if hp == "" || seen[hp] { + return + } + seen[hp] = true + out = append(out, planeCheck{addr: hp, learned: learned}) + } + for _, s := range configured { + add(s, false) + } + if b, err := os.ReadFile(file); err == nil { + var told []string + if json.Unmarshal(b, &told) == nil { + for _, s := range told { + add(s, true) + } + } + } + return out +} + +// hostPort is the host:port a control plane's address dials, whether it was +// written as a URL or not. +func hostPort(s string) string { + s = strings.TrimSpace(s) + port := "" + switch { + case strings.HasPrefix(s, "https://"): + s, port = strings.TrimPrefix(s, "https://"), "443" + case strings.HasPrefix(s, "http://"): + s, port = strings.TrimPrefix(s, "http://"), "80" + } + s, _, _ = strings.Cut(s, "/") + if _, _, err := net.SplitHostPort(s); err != nil && port != "" && s != "" { + s = net.JoinHostPort(strings.Trim(s, "[]"), port) + } + return s +} + +func (p nodeProbe) checkPlanes(ctx context.Context) []planeCheck { + planes := readControlPlanes(p.planesFile, p.configured) + var wg sync.WaitGroup + for i := range planes { + wg.Add(1) + go func(c *planeCheck) { + defer wg.Done() + if err := p.dial(ctx, c.addr); err != nil { + c.err = err.Error() + } + }(&planes[i]) + } + wg.Wait() + return planes +} + +func dialTCP(ctx context.Context, addr string) error { + d := net.Dialer{Timeout: nodeCmdTimeout} + c, err := d.DialContext(ctx, "tcp", addr) + if err != nil { + return err + } + return c.Close() +} + +// --- the GPUs --- + +// readGPUs samples the cards, naming them from discovery the first time. +func (p nodeProbe) readGPUs(ctx context.Context) ([]nodeGPU, map[int]string, string) { + ctx, cancel := context.WithTimeout(ctx, nodeCmdTimeout) + defer cancel() + sensors := p.gpus(ctx) + if len(sensors) == 0 { + // Without nvidia-smi discovery says nothing and no error: a machine + // without a card. With it and no answer, a driver in trouble. + if _, err := p.cards(ctx); err != nil { + return nil, p.names, "The GPUs do not answer: " + err.Error() + } + return nil, p.names, "No GPU on this machine." + } + names := p.names + for _, s := range sensors { + if names[s.Index] == "" { + names = map[int]string{} + if cards, err := p.cards(ctx); err == nil { + for _, c := range cards { + if c.MIG == nil { + names[c.Index] = c.Name + } + } + } + break + } + } + var out []nodeGPU + for _, s := range sensors { + g := nodeGPU{index: s.Index, name: names[s.Index], util: s.UtilPercent, tempC: s.TempC, + memUsedMB: s.MemoryUsedMB, memTotalMB: s.MemoryTotalMB} + if s.MemoryTotalMB == 0 { + g.util = -1 + } + out = append(out, g) + } + return out, names, "" +} + +// cardOf is the card a device number is on: itself, or a MIG instance's card. +func cardOf(i int) int { + if discovery.IsMIGIndex(i) { + return (i - discovery.MIGIndex(0, 0)) / 100 + } + return i +} + +// modelsOnGPUs names, on each card, the models whose engines hold it. +func modelsOnGPUs(gpus []nodeGPU, runs []engineRun) []nodeGPU { + for i := range gpus { + gpus[i].models = nil + for _, r := range runs { + for _, d := range r.gpus { + if cardOf(d) == gpus[i].index && !slices.Contains(gpus[i].models, r.model) { + gpus[i].models = append(gpus[i].models, r.model) + } + } + } + } + return gpus +} + +// --- the engines --- + +// shardStateFile is the part of the supervisor's shard-*.json the overview +// reads (shardState in the supervisor package). +type shardStateFile struct { + ShardID string `json:"shard_id"` + ReplicaID string `json:"replica_id"` + PID int `json:"pid"` + StartedAt time.Time `json:"started_at"` + Spec engines.ShardSpec `json:"spec"` +} + +// readShardStates is the engines the supervisor keeps state files for in +// its work directory, with whether each process is still there. +func readShardStates(dir string, alive func(int) bool) ([]engineRun, string) { + entries, err := os.ReadDir(dir) + switch { + case errors.Is(err, os.ErrPermission): + return nil, "The engines' state is root's: run with sudo." + case errors.Is(err, os.ErrNotExist): + return nil, "" + case err != nil: + return nil, err.Error() + } + var runs []engineRun + note := "" + for _, e := range entries { + name := e.Name() + if !strings.HasPrefix(name, "shard-") || !strings.HasSuffix(name, ".json") { + continue + } + b, err := os.ReadFile(filepath.Join(dir, name)) + if errors.Is(err, os.ErrPermission) { + note = "The engines' state is root's: run with sudo." + continue + } + var st shardStateFile + if err != nil || json.Unmarshal(b, &st) != nil { + continue + } + replica := st.ReplicaID + if replica == "" { + replica = st.Spec.ReplicaID + } + model := st.Spec.Profile.ServedModelName + if model == "" { + model = modelOfReplica(replica) + } + runs = append(runs, engineRun{replica: replica, model: model, engine: st.Spec.Profile.Engine, + gpus: st.Spec.GPUIndices, port: st.Spec.HTTPPort, pid: st.PID, alive: st.PID > 0 && alive(st.PID), + started: st.StartedAt}) + } + sort.Slice(runs, func(i, j int) bool { return runs[i].replica < runs[j].replica }) + return runs, note +} + +// modelOfReplica is the model a replica id names: the control plane calls a +// replica the model's name and a suffix. +func modelOfReplica(id string) string { + if i := strings.LastIndexByte(id, '-'); i > 0 { + return id[:i] + } + return id +} + +func pidAlive(pid int) bool { + err := syscall.Kill(pid, 0) + return err == nil || errors.Is(err, syscall.EPERM) +} + +// --- the disks --- + +func (p nodeProbe) readDisks() []diskUse { + disks := []diskUse{{what: "weights", path: p.weightsDir}, {what: "engines", path: p.enginesDir}} + for i := range disks { + // In its own goroutine: statfs on a network mount that has gone away + // does not return. + type res struct { + free, total uint64 + err error + } + ch := make(chan res, 1) + go func(path string) { + f, t, err := p.statfs(path) + ch <- res{f, t, err} + }(disks[i].path) + select { + case r := <-ch: + disks[i].free, disks[i].total = r.free, r.total + if r.err != nil { + disks[i].err = r.err.Error() + if errors.Is(r.err, os.ErrNotExist) { + disks[i].err = "not there yet" + } + } + case <-time.After(nodeCmdTimeout): + disks[i].err = "does not answer" + } + } + return disks +} + +func statfs(path string) (free, total uint64, err error) { + var st syscall.Statfs_t + if err := syscall.Statfs(path, &st); err != nil { + return 0, 0, &os.PathError{Op: "statfs", Path: path, Err: err} + } + //nolint:unconvert // the field types differ between systems + return uint64(st.Bavail) * uint64(st.Bsize), uint64(st.Blocks) * uint64(st.Bsize), nil +} + +// --- the problems --- + +func (p nodeProbe) readProblems(ctx context.Context) ([]string, string) { + out, err := p.run(ctx, "journalctl", "-u", "llmgw-agent", "-p", "warning", "--since", "-1h", + "--no-pager", "-o", "cat", "-n", strconv.Itoa(nodeProblemLines)) + if err != nil && len(out) == 0 { + return nil, "The journal cannot be read: " + err.Error() + } + lines := parseJournal(string(out), nodeProblemLines) + if len(lines) == 0 && !p.root { + return nil, "None that this user can see: run with sudo to read the agent's journal." + } + return lines, "" +} + +// parseJournal keeps the last n lines journalctl printed, without its own +// remarks. +func parseJournal(out string, n int) []string { + var lines []string + sc := bufio.NewScanner(strings.NewReader(out)) + sc.Buffer(make([]byte, 64*1024), 1024*1024) + for sc.Scan() { + l := strings.TrimSpace(sc.Text()) + if l == "" || strings.HasPrefix(l, "-- ") || strings.HasPrefix(l, "Hint: ") { + continue + } + lines = append(lines, l) + } + if len(lines) > n { + lines = lines[len(lines)-n:] + } + return lines +} + +// runLocal runs a tool on this machine, bounded by nodeCmdTimeout. A tool +// that is not installed is said so, not waited for. +func runLocal(ctx context.Context, name string, args ...string) ([]byte, error) { + if _, err := exec.LookPath(name); err != nil { + return nil, fmt.Errorf("%s is not installed", name) + } + ctx, cancel := context.WithTimeout(ctx, nodeCmdTimeout) + defer cancel() + //nolint:gosec // fixed tools and arguments, from this file + return exec.CommandContext(ctx, name, args...).Output() +} + +// --- the panel --- + +func (m consoleModel) nodeOverviewPanel(pw int) (string, string, string) { + keys := "↑↓ choose · PgUp/PgDn scroll · Q leave · refreshed every three seconds" + if m.nodeOv.at.IsZero() { + return "Overview", "\n " + m.spin.View() + " reading this machine…", keys + } + return "Overview", renderNodeOverview(m.nodeOv, pw, time.Now()), keys +} + +// renderNodeOverview draws the overview from what was read. +func renderNodeOverview(ov nodeOverview, pw int, now time.Time) string { + var b strings.Builder + head := func(s string) { b.WriteString("\n " + lipgloss.NewStyle().Bold(true).Render(s) + "\n") } + kv := func(k, v string) { fmt.Fprintf(&b, " %-14s %s\n", sDim.Render(k), v) } + line := func(s string) { b.WriteString(" " + s + "\n") } + + head("Agent") + a := ov.agent + switch { + case a.note != "": + kv("llmgw-agent", sWarn.Render(a.note)) + default: + state := a.state + if a.sub != "" { + state += " (" + a.sub + ")" + } + if a.state == "active" { + state = sOK.Render(state) + } else { + state = sBad.Render(state) + } + if !a.since.IsZero() { + state += sDim.Render(" since " + a.since.Format("2006-01-02 15:04") + ", " + uptime(now.Sub(a.since)) + " ago") + } + kv("llmgw-agent", state) + } + kv("llmgw", ov.version) + + head("Control planes") + if len(ov.planes) == 0 { + line(sWarn.Render("None configured: agent.yaml names no server.")) + } + for _, c := range ov.planes { + state := sOK.Render("reachable") + if c.err != "" { + state = sBad.Render("✗ ") + truncate(c.err, max(20, pw-30)) + } + if c.learned { + state += sDim.Render(" (named by the leader)") + } + fmt.Fprintf(&b, " %-22s %s\n", c.addr, state) + } + + head("GPUs") + if len(ov.gpus) == 0 { + line(sDim.Render(ov.gpuNote)) + } else { + var rows [][]string + for _, g := range ov.gpus { + busy, mem, temp := "—", "—", "—" + if g.util >= 0 { + busy = fmt.Sprintf("%.0f%%", g.util) + } + if g.memTotalMB > 0 { + mem = fmt.Sprintf("%.1f/%.0f GB", g.memUsedMB/1024, g.memTotalMB/1024) + } + if g.tempC > 0 { + temp = fmt.Sprintf("%.0f°C", g.tempC) + } + models := strings.Join(g.models, ", ") + if models == "" { + models = sDim.Render("free") + } + rows = append(rows, []string{strconv.Itoa(g.index), g.name, busy, mem, temp, models}) + } + b.WriteString(indent(table([]string{"GPU", "CARD", "BUSY", "MEMORY", "TEMP", "MODELS"}, rows, -1, pw-2, false), " ") + "\n") + } + + head("Engines running here") + switch { + case len(ov.engines) > 0: + var rows [][]string + for _, r := range ov.engines { + // The process by its pid, marked when it is no longer there. + proc := sOK.Render(strconv.Itoa(r.pid)) + if !r.alive { + proc = sBad.Render(strconv.Itoa(r.pid) + " gone") + } + gpus := strings.Join(discovery.DeviceLabels(r.gpus), ",") + if gpus == "" { + gpus = sDim.Render("CPU") + } + up := "—" + if !r.started.IsZero() { + up = uptime(now.Sub(r.started)) + } + port := "—" + if r.port > 0 { + port = strconv.Itoa(r.port) + } + rows = append(rows, []string{r.replica, r.model, r.engine, gpus, port, proc, up}) + } + b.WriteString(indent(table([]string{"REPLICA", "MODEL", "ENGINE", "GPUS", "PORT", "PID", "UP"}, rows, -1, pw-2, false), " ") + "\n") + if ov.enginesNote != "" { + line(sWarn.Render(ov.enginesNote)) + } + case ov.enginesNote != "": + line(sWarn.Render(ov.enginesNote)) + default: + line(sDim.Render("None: the control plane has placed nothing here.")) + } + + head("Disk") + for _, d := range ov.disks { + use := sWarn.Render(d.err) + if d.err == "" { + use = fmt.Sprintf("%s free of %s", diskSize(d.free), diskSize(d.total)) + if d.total > 0 && d.free*10 < d.total { + use = sWarn.Render(use) + } + } + fmt.Fprintf(&b, " %-14s %-28s %s\n", sDim.Render(d.what), d.path, use) + } + + head("Recent problems") + switch { + case ov.problemsNote != "": + line(sDim.Render(ov.problemsNote)) + case len(ov.problems) == 0: + line(sOK.Render("None in the last hour.")) + default: + for _, l := range ov.problems { + l = truncateWidth(l, max(20, pw-4)) + if strings.Contains(l, "level=ERROR") { + l = sBad.Render(l) + } else { + l = sWarn.Render(l) + } + line(l) + } + } + b.WriteString("\n " + sDim.Render("read at "+ov.at.Format("15:04:05")) + "\n") + return b.String() +} + +// uptime is a duration as people say it: 3d4h, 2h15m, 40s. +func uptime(d time.Duration) string { + d = d.Round(time.Second) + switch { + case d < 0: + return "0s" + case d >= 24*time.Hour: + return fmt.Sprintf("%dd%dh", d/(24*time.Hour), (d%(24*time.Hour))/time.Hour) + case d >= time.Hour: + return fmt.Sprintf("%dh%dm", d/time.Hour, (d%time.Hour)/time.Minute) + case d >= time.Minute: + return fmt.Sprintf("%dm%ds", d/time.Minute, (d%time.Minute)/time.Second) + } + return d.String() +} + +// diskSize is a byte count in the unit disks are sold in. +func diskSize(n uint64) string { + switch { + case n >= 1<<40: + return fmt.Sprintf("%.1f TB", float64(n)/(1<<40)) + case n >= 1<<30: + return fmt.Sprintf("%.0f GB", float64(n)/(1<<30)) + } + return fmt.Sprintf("%.0f MB", float64(n)/(1<<20)) +} diff --git a/internal/cli/tui_node_overview_test.go b/internal/cli/tui_node_overview_test.go new file mode 100644 index 0000000..623e6dd --- /dev/null +++ b/internal/cli/tui_node_overview_test.go @@ -0,0 +1,206 @@ +package cli + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/charmbracelet/x/ansi" + + "gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery" +) + +func writeNodeFile(t *testing.T, path, body string) { + t.Helper() + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } +} + +func TestANodesOverviewDialsTheConfiguredControlPlanesThenTheLeadersOwn(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "control-planes.json") + writeNodeFile(t, file, `["https://10.0.0.150:7443","10.0.0.140:7443","https://cp3.example/agent"]`) + got := readControlPlanes(file, []string{"10.0.0.140:7443", "http://10.0.0.141"}) + want := []planeCheck{{addr: "10.0.0.140:7443"}, {addr: "10.0.0.141:80"}, + {addr: "10.0.0.150:7443", learned: true}, {addr: "cp3.example:443", learned: true}} + if len(got) != len(want) { + t.Fatalf("got %+v, want %+v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("%d: got %+v, want %+v", i, got[i], want[i]) + } + } + // Without the file, what agent.yaml says. + if got := readControlPlanes(filepath.Join(dir, "none.json"), []string{"cp:7443"}); len(got) != 1 || got[0].addr != "cp:7443" { + t.Errorf("without the file: %+v", got) + } +} + +func TestANodesOverviewReadsTheEnginesFromTheSupervisorsStateFiles(t *testing.T) { + dir := t.TempDir() + writeNodeFile(t, filepath.Join(dir, "shard-qwen3-8b-ab12-0.json"), `{"shard_id":"qwen3-8b-ab12-0","replica_id":"qwen3-8b-ab12", + "pid":4242,"started_at":"2026-10-19T08:00:00Z","spec":{"ReplicaID":"qwen3-8b-ab12","GPUIndices":[0,1],"HTTPPort":8101, + "Profile":{"Name":"vllm-tp2","Engine":"vllm","ServedModelName":"qwen3-8b"}}}`) + writeNodeFile(t, filepath.Join(dir, "shard-bge-m3-7-0.json"), `{"replica_id":"bge-m3-7","pid":99, + "spec":{"GPUIndices":[1002],"HTTPPort":8102,"Profile":{"Engine":"llamacpp"}}}`) + writeNodeFile(t, filepath.Join(dir, "shard-x.json.tmp"), `{"replica_id":"half-written"}`) + writeNodeFile(t, filepath.Join(dir, "control-planes.json"), `[]`) + runs, note := readShardStates(dir, func(pid int) bool { return pid == 4242 }) + if note != "" { + t.Fatalf("note %q", note) + } + if len(runs) != 2 { + t.Fatalf("got %+v", runs) + } + b, q := runs[0], runs[1] + if q.replica != "qwen3-8b-ab12" || q.model != "qwen3-8b" || q.engine != "vllm" || q.port != 8101 || !q.alive || + len(q.gpus) != 2 || !q.started.Equal(time.Date(2026, 10, 19, 8, 0, 0, 0, time.UTC)) { + t.Errorf("qwen: %+v", q) + } + // No served name: the model from the replica's id. + if b.model != "bge-m3" || b.alive || b.engine != "llamacpp" { + t.Errorf("bge: %+v", b) + } + if runs, note := readShardStates(filepath.Join(dir, "absent"), pidAlive); runs != nil || note != "" { + t.Errorf("no work dir: %+v %q", runs, note) + } +} + +func TestANodesOverviewNamesTheModelOnEachGPU(t *testing.T) { + gpus := []nodeGPU{{index: 0}, {index: 1}, {index: 2}} + runs := []engineRun{ + {model: "qwen3-8b", gpus: []int{0, 1}}, + {model: "bge-m3", gpus: []int{discovery.MIGIndex(1, 2)}}, + {model: "qwen3-8b", gpus: []int{1}}, + } + got := modelsOnGPUs(gpus, runs) + if strings.Join(got[0].models, ",") != "qwen3-8b" || strings.Join(got[1].models, ",") != "qwen3-8b,bge-m3" || + len(got[2].models) != 0 { + t.Errorf("got %+v", got) + } +} + +func TestANodesOverviewReadsWhatSystemdSaysOfTheAgent(t *testing.T) { + s := parseSystemctlShow("LoadState=loaded\nActiveState=active\nSubState=running\nActiveEnterTimestamp=Mon 2026-10-19 08:12:03 UTC\n") + if s.state != "active" || s.sub != "running" || s.note != "" || s.since.IsZero() || s.since.Minute() != 12 { + t.Errorf("got %+v", s) + } + if s := parseSystemctlShow("LoadState=not-found\nActiveState=inactive\nActiveEnterTimestamp=\n"); s.note == "" { + t.Errorf("a missing service: %+v", s) + } + if s := parseSystemctlShow("LoadState=loaded\nActiveState=failed\nSubState=failed\nActiveEnterTimestamp=n/a\n"); s.state != "failed" || + !s.since.IsZero() { + t.Errorf("a failed service: %+v", s) + } +} + +func TestANodesOverviewKeepsTheLastWarningsOfTheJournal(t *testing.T) { + var b strings.Builder + for i := 1; i <= 9; i++ { + b.WriteString("level=WARN msg=w" + string(rune('0'+i)) + "\n") + } + got := parseJournal(b.String(), 6) + if len(got) != 6 || got[0] != "level=WARN msg=w4" || got[5] != "level=WARN msg=w9" { + t.Errorf("got %q", got) + } + if got := parseJournal("-- No entries --\n", 6); len(got) != 0 { + t.Errorf("no entries: %q", got) + } +} + +// fakeProbe is a node with two cards, one engine and a broken second control +// plane, and no tool of the machine running the test touched. +func fakeProbe(t *testing.T) nodeProbe { + dir := t.TempDir() + writeNodeFile(t, filepath.Join(dir, "shard-qwen3-8b-ab12-0.json"), `{"replica_id":"qwen3-8b-ab12","pid":4242, + "started_at":"2026-10-19T08:00:00Z","spec":{"GPUIndices":[0],"HTTPPort":8101,"Profile":{"Engine":"vllm","ServedModelName":"qwen3-8b"}}}`) + return nodeProbe{ + configured: []string{"10.0.0.140:7443", "10.0.0.150:7443"}, + planesFile: filepath.Join(dir, "control-planes.json"), + workDir: dir, weightsDir: "/var/lib/llmgw-weights", enginesDir: "/opt/llmgw/engines", root: true, + run: func(_ context.Context, name string, _ ...string) ([]byte, error) { + switch name { + case "systemctl": + return []byte("LoadState=loaded\nActiveState=active\nSubState=running\nActiveEnterTimestamp=Mon 2026-10-19 08:00:00 UTC\n"), nil + case "journalctl": + return []byte("level=WARN msg=\"the control plane does not answer\" server=10.0.0.150:7443\n"), nil + } + return nil, errors.New(name + " is not installed") + }, + dial: func(_ context.Context, addr string) error { + if addr == "10.0.0.150:7443" { + return errors.New("dial tcp 10.0.0.150:7443: connect: connection refused") + } + return nil + }, + gpus: func(context.Context) []discovery.GPUSensor { + return []discovery.GPUSensor{ + {Index: 0, TempC: 61, UtilPercent: 87, MemoryUsedMB: 20480, MemoryTotalMB: 24576}, + {Index: 1, TempC: 35, UtilPercent: 0, MemoryUsedMB: 300, MemoryTotalMB: 24576}, + } + }, + cards: func(context.Context) ([]discovery.GPU, error) { + return []discovery.GPU{{Index: 0, Name: "NVIDIA RTX A5000"}, {Index: 1, Name: "NVIDIA RTX A5000"}}, nil + }, + alive: func(int) bool { return true }, + statfs: func(string) (uint64, uint64, error) { return 120 << 30, 500 << 30, nil }, + now: func() time.Time { return time.Date(2026, 10, 19, 11, 12, 0, 0, time.UTC) }, + } +} + +func TestANodesOverviewShowsWhatRunsOnTheMachine(t *testing.T) { + ov := fakeProbe(t).gather(context.Background()) + out := ansi.Strip(renderNodeOverview(ov, 100, ov.at)) + t.Log("\n" + out) + for _, want := range []string{ + "llmgw-agent", "active (running)", "3h12m ago", + "10.0.0.140:7443", "reachable", "connection refused", + "NVIDIA RTX A5000", "87%", "20.0/24 GB", "61°C", "qwen3-8b", "free", + "qwen3-8b-ab12", "vllm", "8101", "4242", "3h12m", + "/var/lib/llmgw-weights", "120 GB free of 500 GB", "/opt/llmgw/engines", + "the control plane does not answer", + } { + if !strings.Contains(out, want) { + t.Errorf("the overview lacks %q", want) + } + } + if ov.names[0] != "NVIDIA RTX A5000" { + t.Errorf("the names are not kept for the next reading: %v", ov.names) + } +} + +func TestANodeWithoutAGPUSaysSoInOneLine(t *testing.T) { + p := fakeProbe(t) + p.gpus = func(context.Context) []discovery.GPUSensor { return nil } + p.cards = func(context.Context) ([]discovery.GPU, error) { return nil, nil } + p.run = func(context.Context, string, ...string) ([]byte, error) { return nil, errors.New("not installed") } + ov := p.gather(context.Background()) + out := ansi.Strip(renderNodeOverview(ov, 100, ov.at)) + if !strings.Contains(out, "No GPU on this machine.") { + t.Errorf("no line saying there is no GPU:\n%s", out) + } + if strings.Contains(out, "BUSY") { + t.Errorf("an empty GPU table:\n%s", out) + } + // No systemctl or journalctl is said, not a crash. + if !strings.Contains(out, "systemctl is not available") || !strings.Contains(out, "The journal cannot be read") { + t.Errorf("missing tools are not said:\n%s", out) + } +} + +func TestANodesConsoleOpensOnItsOverview(t *testing.T) { + m := newConsoleModel("127.0.0.1:1", filepath.Join(t.TempDir(), "agent.yaml")) + if m.menu != cOverview || !m.entryOK(cOverview) { + t.Fatalf("menu %d", m.menu) + } + _, body, _ := m.panel() + if !strings.Contains(body, "reading this machine") { + t.Errorf("before the first reading: %q", body) + } +}