diff --git a/CHANGELOG.md b/CHANGELOG.md index 07dbef25..45d6e08e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -76,6 +76,26 @@ is not an update. A model from `server.yaml` is reported but not changed: its source is edited there. +### The monitor: the cluster drawn like btop + +The console has a **Monitor** under Watch: CPU, memory, GPU busy and memory, +requests, tokens, time to first token, in flight and queued, network and +disk, as Braille charts filled and coloured by height, over 5 minutes, an +hour, a day or a week (1–4 or R); then each node's CPU, memory and network, +each card's busy, memory, heat and power, and each model's requests, as +sparklines. A node's and a model's own pages carry their charts too. + +The history is read from the Prometheus in `monitoring.prometheus`, through +the admin API (`POST /api/v1/metrics/range`); without one, the charts draw +what the console has seen since it opened. + +The control plane now lists the cluster's machines for Prometheus' service +discovery, at `/prometheus/targets` on its metrics port — nodes, itself and +the store's members, each labelled with its llmgw id — and `llmgw node add` +installs `prometheus-node-exporter` on the machine (`--no-node-exporter` +leaves it off). `deploy/monitoring/prometheus.yml` scrapes them as the +`nodes` job. + ### Serving and updating run in the background Serving a model or moving one to new weights from the console starts a job diff --git a/deploy/monitoring/prometheus.yml b/deploy/monitoring/prometheus.yml index f69c501f..d47be64b 100644 --- a/deploy/monitoring/prometheus.yml +++ b/deploy/monitoring/prometheus.yml @@ -27,6 +27,17 @@ scrape_configs: labels: instance: control-plane + # Every machine of the cluster, for its node exporter: network and disk + # traffic, filesystems, CPU time. The control plane lists them — nodes, + # itself, the store's members — so a node added is a node scraped with + # nothing edited here, and each carries its llmgw id as the node label. + # `llmgw node add` installs the exporter; on a machine added before that, + # apt-get install prometheus-node-exporter. + - job_name: nodes + http_sd_configs: + - url: http://10.0.0.140:9091/prometheus/targets + refresh_interval: 1m + - job_name: prometheus static_configs: - targets: ['localhost:9090'] diff --git a/docs/console.md b/docs/console.md index 8cc5a2fe..458347cd 100644 --- a/docs/console.md +++ b/docs/console.md @@ -57,6 +57,19 @@ read, and Enter brings the console back. **Overview** is `llmgw ctl status`: nodes, GPUs by class, fabrics, replicas, models, the store's members, and anything parked, unplaced or idle. +**Monitor** draws the cluster the way btop draws a machine: CPU, memory, +GPU busy and GPU memory, requests and tokens a second, time to first token, +requests in flight and queued, network and disk — Braille charts, filled +and coloured green to red by height — over 5 minutes, an hour, a day or a +week (**1**–**4**, or **R** for the next). Below them each node's CPU, +memory and network now and as a sparkline, each card's busy, memory, heat +and power, and each model's requests a second. The history comes from the +Prometheus named by `monitoring.prometheus` in server.yaml; without one the +charts draw what this console has seen since it opened, and say so. Network +and disk come from each machine's node exporter, which `llmgw node add` +installs. A node's and a model's own pages have their charts too, over the +same range. + **Nodes** lists every node with its state, load, GPUs, replicas, disk and agent version. On a row: diff --git a/internal/cli/node.go b/internal/cli/node.go index 2e4c14bf..92f73afe 100644 --- a/internal/cli/node.go +++ b/internal/cli/node.go @@ -121,6 +121,7 @@ engines are kept.`, f.StringVar(&na.opts.PKI, "pki", defaultPKIDir, "directory holding the certificate authority") f.StringVar(&na.opts.Releases, "releases", "https://gitea.mynodes.xyz/ilker/llmgw/releases/download", "where release files are downloaded from") f.BoolVar(&na.opts.IgnoreChecks, "ignore-checks", false, "add the node even when doctor finds a problem") + f.BoolVar(&na.opts.NoNodeExporter, "no-node-exporter", false, "leave Prometheus' node exporter off the machine") f.StringArrayVar(&na.opts.DisableEngines, "disable-engine", nil, "an engine this node never installs, e.g. vllm on GPUs it does not run on; repeatable") f.StringVar(&na.admin, "admin", "", "the control plane's admin API, to watch the node join (default: from server.yaml, else 127.0.0.1:9090)") f.StringVar(&na.serverCfg, "config", "/etc/llmgw/server.yaml", "the control plane's configuration, for the admin API's address") diff --git a/internal/cli/server.go b/internal/cli/server.go index 6b7698cf..d8501c4c 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -287,6 +287,10 @@ func newServerCmd() *cobra.Command { var history web.History if h := promq.New(cfg.Monitoring.Prometheus, cfg.Monitoring.Timeout); h != nil { history = h + // And the console's charts, over the admin API. + if adminAPI != nil { + adminAPI.SetHistory(h) + } log.Info("dashboard history configured", "prometheus", cfg.Monitoring.Prometheus) } @@ -382,6 +386,15 @@ func newServerCmd() *cobra.Command { _ = adminLn.Close() return fmt.Errorf("listening on %s: %w", cfg.MetricsListen, err) } + // Beside the metrics, the machines to scrape for their node + // exporters, for Prometheus' http_sd_configs: on this port + // because Prometheus already reaches it. + if adminAPI != nil { + mux := http.NewServeMux() + mux.HandleFunc("GET /prometheus/targets", adminAPI.PromSD) + mux.Handle("/", h) + h = mux + } metricsSrv = &http.Server{Handler: h, ReadHeaderTimeout: 10 * time.Second} go func() { if err := metricsSrv.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { diff --git a/internal/cli/tui_actions.go b/internal/cli/tui_actions.go index 1f2613cc..4c8f2511 100644 --- a/internal/cli/tui_actions.go +++ b/internal/cli/tui_actions.go @@ -276,7 +276,7 @@ func (m consoleModel) actionDialog() (title, body string) { func details(m *consoleModel) tea.Cmd { m.detail, m.scroll = true, 0 - return nil + return m.fetchDetail() } // --- models --- diff --git a/internal/cli/tui_actions_test.go b/internal/cli/tui_actions_test.go index 38d7ce65..95889078 100644 --- a/internal/cli/tui_actions_test.go +++ b/internal/cli/tui_actions_test.go @@ -17,6 +17,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve" "gitea.mynodes.xyz/ilker/llmgw/internal/modelplan" + "gitea.mynodes.xyz/ilker/llmgw/internal/promq" "gitea.mynodes.xyz/ilker/llmgw/internal/store" "gitea.mynodes.xyz/ilker/llmgw/internal/version" ) @@ -492,3 +493,101 @@ func TestJobsAreFollowedFromTheirOwnView(t *testing.T) { t.Errorf("steps not shown:\n%s", v) } } + +func TestTheMonitorDrawsFromPrometheusAndWithoutIt(t *testing.T) { + pts := func(vals ...float64) []promq.Series { + s := promq.Series{Name: "x"} + for _, v := range vals { + s.Points = append(s.Points, promq.Point{Value: v}) + } + return []promq.Series{s} + } + var asked adminapi.MetricsRequest + admin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/v1/metrics/range" { + _, _ = w.Write([]byte("[]")) + return + } + _ = json.NewDecoder(r.Body).Decode(&asked) + res := adminapi.MetricsResult{Configured: true, Series: map[string][]promq.Series{ + "cpu": pts(10, 40, 80), "mem": pts(50, 55), "rps": pts(1, 2, 3), "rpsby": pts(1, 2, 3), "tokout": pts(100, 200), + "netin": pts(1e6, 2e6), + }} + res.Series["rpsby"][0].Name = "qwen" + _ = json.NewEncoder(w).Encode(res) + })) + defer admin.Close() + m := newConsoleModel(admin.URL, "/nonexistent/server.yaml") + m.loaded, m.menu, m.inPanel, m.width, m.height = true, cMonitor, true, 160, 60 + var tm tea.Model = m + tm, cmd := tm.Update(tea.KeyMsg{Type: tea.KeyRunes, Runes: []rune("2")}) + tm, _ = tm.Update(cmd()) + if asked.Window != time.Hour || len(asked.Queries) < 10 { + t.Fatalf("asked %+v", asked) + } + v := tm.View() + for _, want := range []string{"[1h]", "CPU", "80%", "Requests", "qwen", "from Prometheus", "Network", "1.9 MB/s"} { + if !strings.Contains(v, want) { + t.Errorf("no %q:\n%s", want, v) + } + } + if !strings.ContainsAny(v, "⣀⣤⣶⣿") { + t.Errorf("no braille chart:\n%s", v) + } + + // No Prometheus: what the console saw, and it says so. + m2, _ := consoleWith(t, cMonitor) + m2.width, m2.height = 160, 60 + for _, c := range []float64{20, 60} { + m2.data.nodes = []adminapi.NodeView{{ID: "n1", State: "ready", Load: &adminapi.LoadView{CPUPercent: c, MemPercent: 30, + MemUsed: 3, MemTotal: 10}}} + m2.mon.sample(m2.data) + } + if v := m2.View(); !strings.Contains(v, "since it opened") || !strings.Contains(v, "60%") { + t.Errorf("local:\n%s", v) + } +} + +func TestBrailleChartsFillToTheirHeight(t *testing.T) { + rows := brailleChart([]chartSeries{{values: []float64{0, 50, 100, 100}}}, 2, 2, 100) + if len(rows) != 2 { + t.Fatal(rows) + } + // The right cell is full top to bottom; the left holds a low point and a middle one. + if !strings.Contains(rows[0], "⣿") || !strings.Contains(rows[1], "⣿") { + t.Errorf("%q", rows) + } + if s := sparkline([]float64{0, 1, 2, 3}, 4, 3); !strings.Contains(s, "▁") || !strings.Contains(s, "█") { + t.Errorf("%q", s) + } +} + +func TestANodesPageHasItsCharts(t *testing.T) { + var asked adminapi.MetricsRequest + admin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/v1/metrics/range" { + _, _ = w.Write([]byte("[]")) + return + } + _ = json.NewDecoder(r.Body).Decode(&asked) + _ = json.NewEncoder(w).Encode(adminapi.MetricsResult{Configured: true, Series: map[string][]promq.Series{ + "cpu": {{Name: "n1", Points: []promq.Point{{Value: 12}, {Value: 34}}}}}}) + })) + defer admin.Close() + m := newConsoleModel(admin.URL, "/nonexistent/server.yaml") + m.loaded, m.menu, m.inPanel, m.width, m.height = true, cNodes, true, 160, 60 + m.data.nodes = []adminapi.NodeView{{ID: "n1", State: "ready"}} + var tm tea.Model = m + tm = press(tm, "enter") // the menu; Details first + tm, cmd := tm.Update(tea.KeyMsg{Type: tea.KeyEnter}) + if cmd == nil { + t.Fatal("opening the page read nothing") + } + tm, _ = tm.Update(cmd()) + if len(asked.Queries) == 0 || !strings.Contains(asked.Queries[0].Expr, `node="n1"`) { + t.Fatalf("asked %+v", asked.Queries) + } + if v := tm.View(); !strings.Contains(v, "Over the last") || !strings.Contains(v, "34%") { + t.Errorf("%s", v) + } +} diff --git a/internal/cli/tui_chart.go b/internal/cli/tui_chart.go new file mode 100644 index 00000000..aadf0884 --- /dev/null +++ b/internal/cli/tui_chart.go @@ -0,0 +1,230 @@ +package cli + +import ( + "fmt" + "math" + "strings" + + "github.com/charmbracelet/lipgloss" +) + +// Charts drawn in a terminal the way btop draws them: Braille characters, +// each cell two dots across and four down, so a chart forty cells wide holds +// eighty points and five rows twenty levels. Filled below the line, and +// coloured by height — green low, yellow, red near the top — so how hard +// something is working reads before any number does. + +// Braille dot bits by column (0 left, 1 right) and row (0 top .. 3 bottom). +var brailleBit = [2][4]rune{{0x01, 0x02, 0x04, 0x40}, {0x08, 0x10, 0x20, 0x80}} + +// gradient is the colour at a fraction of a chart's height, 0 bottom, 1 top. +func gradient(f float64) lipgloss.Color { + stops := [][3]float64{{80, 200, 120}, {230, 200, 60}, {230, 80, 70}} + f = math.Max(0, math.Min(1, f)) + seg := f * float64(len(stops)-1) + i := int(seg) + if i >= len(stops)-1 { + i, seg = len(stops)-2, float64(len(stops)-1) + } + t := seg - float64(i) + c := [3]float64{} + for k := range c { + c[k] = stops[i][k] + (stops[i+1][k]-stops[i][k])*t + } + return lipgloss.Color(fmt.Sprintf("#%02x%02x%02x", int(c[0]), int(c[1]), int(c[2]))) +} + +// lineColors are the colours of the lines of a chart with several. +var lineColors = []lipgloss.Color{"#61afef", "#e5c07b", "#c678dd", "#56b6c2", "#e06c75", "#98c379", "#d19a66", "#abb2bf"} + +// chartSeries is one line: its values, oldest first, and a name for the +// legend. NaN is no sample. +type chartSeries struct { + name string + values []float64 +} + +// brailleChart draws series into width×height cells. top is the value at the +// top of the chart; 0 takes the largest value. One series is filled and +// coloured by height; several are lines, each its own colour. +func brailleChart(series []chartSeries, width, height int, top float64) []string { + width, height = max(width, 4), max(height, 1) + if top <= 0 { + for _, s := range series { + for _, v := range s.values { + if !math.IsNaN(v) && v > top { + top = v + } + } + } + } + if top <= 0 { + top = 1 + } + cols, rows := width*2, height*4 + dots := make([][]rune, height) + owner := make([][]int, height) + for r := range dots { + dots[r] = make([]rune, width) + owner[r] = make([]int, width) + for c := range owner[r] { + owner[r][c] = -1 + } + } + set := func(x, y, who int) { // y from the bottom + if x < 0 || x >= cols || y < 0 || y >= rows { + return + } + r := height - 1 - y/4 + dots[r][x/2] |= brailleBit[x%2][3-y%4] + owner[r][x/2] = who + } + area := len(series) == 1 + for si, s := range series { + vals := resample(s.values, cols) + prev := -1 + for x, v := range vals { + if math.IsNaN(v) { + prev = -1 + continue + } + h := int(math.Round(v / top * float64(rows-1))) + h = max(0, min(rows-1, h)) + if area { + for y := 0; y <= h; y++ { + set(x, y, si) + } + continue + } + // A line: this point, and the rise or fall from the last. + lo, hi := h, h + if prev >= 0 { + lo, hi = min(prev, h), max(prev, h) + } + for y := lo; y <= hi; y++ { + set(x, y, si) + } + prev = h + } + } + out := make([]string, height) + for r := range dots { + var b strings.Builder + for c, d := range dots[r] { + if owner[r][c] < 0 { + b.WriteString(" ") + continue + } + color := lineColors[owner[r][c]%len(lineColors)] + if area { + color = gradient(float64(height-1-r) / float64(max(1, height-1))) + } + b.WriteString(lipgloss.NewStyle().Foreground(color).Render(string(0x2800 + d))) + } + out[r] = b.String() + } + return out +} + +// resample fits values into n points: the most recent n, each the average of +// what falls in it, the left padded with no sample when there are fewer. +func resample(vals []float64, n int) []float64 { + out := make([]float64, n) + for i := range out { + out[i] = math.NaN() + } + if len(vals) == 0 { + return out + } + if len(vals) <= n { + copy(out[n-len(vals):], vals) + return out + } + per := float64(len(vals)) / float64(n) + for i := range out { + from, to := int(float64(i)*per), int(float64(i+1)*per) + sum, k := 0.0, 0 + for _, v := range vals[from:max(to, from+1)] { + if !math.IsNaN(v) { + sum += v + k++ + } + } + if k > 0 { + out[i] = sum / float64(k) + } + } + return out +} + +var sparkRunes = []rune("▁▂▃▄▅▆▇█") + +// sparkline is values in one row of block characters, coloured by height. +func sparkline(vals []float64, width int, top float64) string { + vals = resample(vals, width) + if top <= 0 { + for _, v := range vals { + if !math.IsNaN(v) && v > top { + top = v + } + } + } + if top <= 0 { + top = 1 + } + var b strings.Builder + for _, v := range vals { + if math.IsNaN(v) { + b.WriteString(" ") + continue + } + f := math.Max(0, math.Min(1, v/top)) + r := sparkRunes[int(math.Round(f*float64(len(sparkRunes)-1)))] + b.WriteString(lipgloss.NewStyle().Foreground(gradient(f)).Render(string(r))) + } + return b.String() +} + +// meter is a fraction as a bar of width cells, btop's ■, coloured by level. +func meter(f float64, width int) string { + f = math.Max(0, math.Min(1, f)) + n := int(math.Round(f * float64(width))) + var b strings.Builder + for i := 0; i < width; i++ { + if i < n { + b.WriteString(lipgloss.NewStyle().Foreground(gradient(float64(i) / float64(max(1, width-1)))).Render("■")) + } else { + b.WriteString(sDim.Render("■")) + } + } + return b.String() +} + +// chartBox is a chart with its title and current value above it, and the +// window's span below: what a panel of the monitor shows. +func chartBox(title, now string, series []chartSeries, width, height int, top float64, span string) string { + var b strings.Builder + head := mdBold.Render(title) + if now != "" { + head += " " + now + } + if len(series) > 1 { + var leg []string + for i, s := range series { + if i == 6 { + leg = append(leg, sDim.Render(fmt.Sprintf("+%d", len(series)-6))) + break + } + leg = append(leg, lipgloss.NewStyle().Foreground(lineColors[i%len(lineColors)]).Render("━ ")+truncate(s.name, 18)) + } + head += " " + strings.Join(leg, " ") + } + b.WriteString(truncateWidth(head, width) + "\n") + for _, l := range brailleChart(series, width, height, top) { + b.WriteString(l + "\n") + } + left, right := "-"+span, "now" + gap := max(1, width-len(left)-len(right)) + b.WriteString(sDim.Render(left + strings.Repeat("─", gap) + right)) + return b.String() +} diff --git a/internal/cli/tui_console.go b/internal/cli/tui_console.go index 06ef9a36..947b138a 100644 --- a/internal/cli/tui_console.go +++ b/internal/cli/tui_console.go @@ -33,6 +33,7 @@ import ( // Sidebar entries, in order. Headings are not entries. const ( cOverview = iota + cMonitor cNodes cGPUs cModels @@ -57,7 +58,7 @@ const ( cCount ) -var consoleLabels = [cCount]string{"Overview", "Nodes", "GPUs", "Models", "Replicas", "Usage", "Quotas", "Mirror", "Jobs", "Logs", "Accounts", "API keys", "Organizations", "Manual", +var consoleLabels = [cCount]string{"Overview", "Monitor", "Nodes", "GPUs", "Models", "Replicas", "Usage", "Quotas", "Mirror", "Jobs", "Logs", "Accounts", "API keys", "Organizations", "Manual", "server.yaml", "Add a node", "Replicate the store", "Serve a model", "Upgrade the cluster", "Restart all replicas", "Back up now", "Leave"} @@ -159,6 +160,8 @@ type consoleModel struct { mirror consoleMirrorMsg hf hfState docs docViewer + mon monitorState + det monitorState } var usageGroupings = []string{"team", "project", "org"} @@ -275,6 +278,12 @@ func (m consoleModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { if m.menu == cMirror { cmds = append(cmds, m.fetchMirror()) } + if m.menu == cMonitor && m.monitorDue() { + cmds = append(cmds, m.fetchMonitor()) + } + if m.detailDue() { + cmds = append(cmds, m.fetchDetail()) + } if m.peopleView() { cmds = append(cmds, m.fetchPeople()) } @@ -288,6 +297,16 @@ func (m consoleModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { m.data.err = msg.err } else { m.data = consoleData(msg) + m.mon.sample(m.data) + } + case detailMsg: + if msg.key == m.detailKey() { + m.det.res, m.det.err, m.det.key = msg.res, msg.err, msg.key + } + case monitorMsg: + // Kept if it is still what the screen wants: the range and width. + if msg.key == m.monitorKey() { + m.mon.res, m.mon.err, m.mon.key = msg.res, msg.err, msg.key } case consoleUsageMsg: m.usage, m.usageErr = msg.rows, msg.err @@ -467,6 +486,18 @@ func (m consoleModel) key(k tea.KeyMsg) (tea.Model, tea.Cmd) { m.docs.show(m.row) m.detail = true } + case "1", "2", "3", "4", "r": + if kind, _ := m.detailTarget(); m.menu == cMonitor || kind != "" { + if k.String() == "r" { + m.mon.rng = (m.mon.rng + 1) % len(monRanges) + } else { + m.mon.rng = int(k.String()[0] - '1') + } + if kind != "" { + return m, m.fetchDetail() + } + return m, m.fetchMonitor() + } case "/": if m.menu == cServe { m.hfSearchDialog() @@ -524,6 +555,8 @@ func (m consoleModel) onEnter() tea.Cmd { return m.fetchQuotas() case cMirror: return m.fetchMirror() + case cMonitor: + return m.fetchMonitor() case cServe: if m.hf.models == nil { return m.fetchHF() @@ -961,7 +994,7 @@ func (m consoleModel) panel() (title, body, keys string) { if m.menu == cLogs { return "Agent log · " + n.ID, m.logView(), "← back · PgUp/PgDn · refreshed every three seconds" } - return "Node · " + n.ID, m.nodeDetail(n), "← back · PgUp/PgDn · Q leave" + return "Node · " + n.ID, m.nodeDetail(n) + m.detailCharts(pw), "← back · 1–4 or R range · PgUp/PgDn · Q leave" } var rows [][]string for _, n := range m.data.nodes { @@ -993,7 +1026,8 @@ func (m consoleModel) panel() (title, body, keys string) { m.gpuRows(), m.row, pw, m.inPanel), keys case cModels: if m.detail && m.row < len(m.data.models) { - return "Model · " + m.data.models[m.row].Name, m.modelDetail(m.data.models[m.row]), "← back · PgUp/PgDn · Q leave" + return "Model · " + m.data.models[m.row].Name, m.modelDetail(m.data.models[m.row]) + m.detailCharts(pw), + "← back · 1–4 or R range · PgUp/PgDn · Q leave" } var rows [][]string for _, md := range m.data.models { @@ -1074,6 +1108,8 @@ func (m consoleModel) panel() (title, body, keys string) { return m.mirrorPanel(pw) case cJobs: return m.jobsPanel(pw) + case cMonitor: + return m.monitorPanel(pw) case cServe: return m.hfPanel(pw) case cSettings: diff --git a/internal/cli/tui_monitor.go b/internal/cli/tui_monitor.go new file mode 100644 index 00000000..a974f5ee --- /dev/null +++ b/internal/cli/tui_monitor.go @@ -0,0 +1,534 @@ +package cli + +import ( + "context" + "fmt" + "math" + "net/http" + "sort" + "strconv" + "strings" + "time" + + tea "github.com/charmbracelet/bubbletea" + "github.com/charmbracelet/lipgloss" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/promq" +) + +// The monitor: the cluster drawn the way btop draws a machine. History comes +// from the Prometheus that scrapes the control plane (monitoring.prometheus in +// server.yaml), read through the admin API; without one, from what this +// console has seen since it opened. + +var monRanges = []struct { + label string + window time.Duration + refresh time.Duration +}{ + {"5m", 5 * time.Minute, 5 * time.Second}, + {"1h", time.Hour, 15 * time.Second}, + {"24h", 24 * time.Hour, time.Minute}, + {"7d", 7 * 24 * time.Hour, 5 * time.Minute}, +} + +type monitorState struct { + rng int + res adminapi.MetricsResult + err error + at time.Time + key string + // local is what this console sampled itself, every refresh, by series: + // the charts when there is no Prometheus. + local map[string][]float64 + localAt []time.Time +} + +type monitorMsg struct { + key string + res adminapi.MetricsResult + err error +} + +const localKeep = 1200 // an hour of three-second samples + +// sample keeps what the cluster looks like now, for the charts without a +// Prometheus. +func (s *monitorState) sample(d consoleData) { + if s.local == nil { + s.local = map[string][]float64{} + } + add := func(k string, v float64) { + s.local[k] = append(s.local[k], v) + if len(s.local[k]) > localKeep { + s.local[k] = s.local[k][len(s.local[k])-localKeep:] + } + } + var cpu, used, total float64 + n := 0 + for _, nd := range d.nodes { + if nd.Load == nil { + continue + } + l := nd.Load + cpu += l.CPUPercent + n++ + used, total = used+float64(l.MemUsed), total+float64(l.MemTotal) + add("nodecpu/"+nd.ID, l.CPUPercent) + add("nodemem/"+nd.ID, l.MemPercent) + for _, g := range l.GPUs { + card := nd.ID + ":" + strconv.Itoa(g.Index) + add("gpu/"+card, g.UtilPercent) + if g.MemoryTotalMB > 0 { + add("vram/"+card, 100*g.MemoryUsedMB/g.MemoryTotalMB) + } + add("gputemp/"+card, g.TempC) + add("gpupow/"+card, g.PowerW) + } + } + if n > 0 { + add("cpu/", cpu/float64(n)) + } + if total > 0 { + add("mem/", 100*used/total) + } +} + +// localSeries is the series sampled under an id, as charts take them. +func (s monitorState) localSeries(id string) []promq.Series { + var out []promq.Series + for k, vals := range s.local { + pre, name, ok := strings.Cut(k, "/") + if !ok || pre != id { + continue + } + ser := promq.Series{Name: name} + for _, v := range vals { + ser.Points = append(ser.Points, promq.Point{Value: v}) + } + out = append(out, ser) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + return out +} + +// netDevs and diskDevs leave out what is not the machine's own traffic: +// loopback, container bridges, device-mapper over the disks it is made of. +const ( + netDevs = `job="nodes",device!~"lo|veth.*|docker.*|br-.*|cni.*|flannel.*|cali.*|virbr.*|tap.*"` + diskDevs = `job="nodes",device!~"loop.*|ram.*|dm-.*|sr.*"` +) + +// bps is bytes a second as a person reads it. +func bps(v float64) string { return bytesShort(int64(v)) + "/s" } + +// promDur is a duration as PromQL writes one. +func promDur(d time.Duration) string { + if d%time.Minute == 0 { + return strconv.Itoa(int(d/time.Minute)) + "m" + } + return strconv.Itoa(int(d/time.Second)) + "s" +} + +// monitorQueries are what the monitor draws, for a window drawn at points +// points wide. +func monitorQueries(window time.Duration, points int) (time.Duration, []adminapi.MetricsQuery) { + step := max(5*time.Second, (window / time.Duration(max(points, 1))).Round(time.Second)) + // A rate over at least four scrapes, or it is noise. + rw := promDur(max(time.Minute, 4*step).Round(time.Second)) + card := func(expr string) string { return `label_join(` + expr + `, "card", ":", "node", "gpu")` } + return step, []adminapi.MetricsQuery{ + {ID: "cpu", Expr: `avg(llmgw_node_cpu_utilization_percent)`}, + {ID: "mem", Expr: `100 * sum(llmgw_node_memory_used_bytes) / sum(llmgw_node_memory_bytes)`}, + {ID: "gpu", Expr: card(`llmgw_gpu_utilization_percent`), By: "card"}, + {ID: "vram", Expr: card(`100 * llmgw_gpu_memory_used_bytes / llmgw_gpu_memory_total_bytes`), By: "card"}, + {ID: "gputemp", Expr: card(`llmgw_gpu_temperature_celsius`), By: "card"}, + {ID: "gpupow", Expr: card(`llmgw_gpu_power_watts`), By: "card"}, + // Totals fall back to zero: a counter not yet incremented since the + // control plane started is no series at all, and a quiet hour should + // read as none, not as missing. + {ID: "rps", Expr: `sum(rate(llmgw_requests_total[` + rw + `])) or vector(0)`}, + {ID: "rpsby", Expr: `sum by (model) (rate(llmgw_requests_total[` + rw + `]))`, By: "model"}, + {ID: "errs", Expr: `sum(rate(llmgw_requests_total{status=~"5..|429"}[` + rw + `])) or vector(0)`}, + {ID: "tokin", Expr: `sum(rate(llmgw_tokens_total{direction="prompt"}[` + rw + `])) or vector(0)`}, + {ID: "tokout", Expr: `sum(rate(llmgw_tokens_total{direction="completion"}[` + rw + `])) or vector(0)`}, + // The machines' own, from their node exporters. + {ID: "netin", Expr: `sum(rate(node_network_receive_bytes_total{` + netDevs + `}[` + rw + `]))`}, + {ID: "netout", Expr: `sum(rate(node_network_transmit_bytes_total{` + netDevs + `}[` + rw + `]))`}, + {ID: "diskr", Expr: `sum(rate(node_disk_read_bytes_total{` + diskDevs + `}[` + rw + `]))`}, + {ID: "diskw", Expr: `sum(rate(node_disk_written_bytes_total{` + diskDevs + `}[` + rw + `]))`}, + {ID: "nodenet", Expr: `sum by (node) (rate(node_network_receive_bytes_total{` + netDevs + `}[` + rw + `]) + rate(node_network_transmit_bytes_total{` + netDevs + `}[` + rw + `]))`, By: "node"}, + {ID: "ttft50", Expr: `histogram_quantile(0.5, sum by (le) (rate(llmgw_ttft_seconds_bucket[` + rw + `])))`}, + {ID: "ttft99", Expr: `histogram_quantile(0.99, sum by (le) (rate(llmgw_ttft_seconds_bucket[` + rw + `])))`}, + {ID: "inflight", Expr: `sum(llmgw_endpoint_inflight)`}, + {ID: "queue", Expr: `sum(llmgw_admission_queue_depth)`}, + {ID: "nodecpu", Expr: `llmgw_node_cpu_utilization_percent`, By: "node"}, + {ID: "nodemem", Expr: `100 * llmgw_node_memory_used_bytes / llmgw_node_memory_bytes`, By: "node"}, + } +} + +func (m consoleModel) monitorWidth() int { return max(40, m.width-navWidth-8) } + +// monitorKey says what is drawn: the range and the width it is drawn at. +func (m consoleModel) monitorKey() string { + return fmt.Sprintf("%d/%d", m.mon.rng, m.monitorWidth()) +} + +// monitorDue is whether the charts should be read again. +func (m consoleModel) monitorDue() bool { + return m.mon.key != m.monitorKey() || time.Since(m.mon.at) >= monRanges[m.mon.rng].refresh +} + +func (m *consoleModel) fetchMonitor() tea.Cmd { + r := monRanges[m.mon.rng] + step, qs := monitorQueries(r.window, (m.monitorWidth()/2-2)*2) + key, c := m.monitorKey(), m.client + m.mon.at, m.mon.key = time.Now(), key + return func() tea.Msg { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + var res adminapi.MetricsResult + err := c.do(ctx, http.MethodPost, "/api/v1/metrics/range", + adminapi.MetricsRequest{Window: r.window, Step: step, Queries: qs}, &res) + return monitorMsg{key: key, res: res, err: err} + } +} + +// series is what is drawn under an id: from Prometheus, or sampled here. +func (m consoleModel) series(id string) []chartSeries { + src := m.mon.res.Series[id] + if !m.mon.res.Configured { + src = m.mon.localSeries(id) + } + out := make([]chartSeries, 0, len(src)) + for _, s := range src { + cs := chartSeries{name: s.Name} + for _, p := range s.Points { + cs.values = append(cs.values, p.Value) + } + out = append(out, cs) + } + return out +} + +func lastOf(s []chartSeries) float64 { + t := 0.0 + for _, x := range s { + for i := len(x.values) - 1; i >= 0; i-- { + if !math.IsNaN(x.values[i]) { + t += x.values[i] + break + } + } + } + return t +} + +func rate(v float64) string { + switch { + case v >= 1000: + return fmt.Sprintf("%.1fk/s", v/1000) + case v >= 10: + return fmt.Sprintf("%.0f/s", v) + } + return fmt.Sprintf("%.2f/s", v) +} + +// side puts two blocks next to each other, each w wide. +func side(w int, a, b string) string { + st := lipgloss.NewStyle().Width(w) + return lipgloss.JoinHorizontal(lipgloss.Top, st.Render(a), " ", st.Render(b)) +} + +func (m consoleModel) monitorPanel(pw int) (string, string, string) { + keys := "→ or Enter open · ↑↓ choose · Q leave" + if m.inPanel { + keys = "1–4 or R range · PgUp/PgDn scroll · ← back · Q leave" + } + r := monRanges[m.mon.rng] + var head strings.Builder + head.WriteString("\n ") + for i, x := range monRanges { + if i == m.mon.rng { + head.WriteString(fSelected.Render("["+x.label+"]") + " ") + } else { + head.WriteString(sDim.Render(" "+x.label+" ") + " ") + } + } + src := "from Prometheus" + switch { + case m.mon.err != nil: + src = sBad.Render("the control plane: " + m.mon.err.Error()) + case !m.mon.res.Configured: + src = sWarn.Render("from what this console has seen since it opened") + + sDim.Render(" — set monitoring.prometheus in server.yaml for history") + case len(m.mon.res.Errors) > 0: + for _, e := range m.mon.res.Errors { + src = sWarn.Render("Prometheus: " + truncate(e, 80)) + break + } + } + head.WriteString(" " + src + "\n\n") + + w := (pw - 6) / 2 + ch := 5 + span := r.label + if !m.mon.res.Configured { + span = "since opened" + } + pct := func(id string) string { return fmt.Sprintf("%.0f%%", lastOf(m.series(id))) } + box := func(title, now string, s []chartSeries, top float64) string { + if len(s) == 0 { + empty := "no samples yet" + if m.mon.res.Configured { + empty = "no requests in the last " + span + } + return mdBold.Render(title) + "\n\n " + sDim.Render(empty) + } + return chartBox(title, now, s, w, ch, top, span) + } + netBox := func(title string, s []chartSeries) string { + if len(s) == 0 { + return mdBold.Render(title) + "\n\n " + sDim.Render("no node exporter answering: llmgw node add installs one") + } + return box(title, bps(lastOf(s)), s, 0) + } + var b strings.Builder + b.WriteString(indent(side(w, box("CPU", pct("cpu"), m.series("cpu"), 100), + box("Memory", pct("mem"), m.series("mem"), 100)), " ") + "\n\n") + if g := m.series("gpu"); len(g) > 0 || len(m.data.sum.GPUs) > 0 { + b.WriteString(indent(side(w, box("GPU busy", "", g, 100), box("GPU memory", "", m.series("vram"), 100)), " ") + "\n\n") + } + if m.mon.res.Configured { + rps := m.series("rps") + tok := append(named(m.series("tokin"), "prompt"), named(m.series("tokout"), "completion")...) + b.WriteString(indent(side(w, box("Requests", rate(lastOf(rps)), rps, 0), + box("Tokens", rate(lastOf(tok)), tok, 0)), " ") + "\n\n") + ttft := append(named(m.series("ttft50"), "p50"), named(m.series("ttft99"), "p99")...) + load := append(named(m.series("inflight"), "in flight"), named(m.series("queue"), "queued")...) + load = append(load, named(m.series("errs"), "errors/s")...) + if len(ttft) == 0 { + ttft = nil + } + b.WriteString(indent(side(w, box("Time to first token (s)", "", ttft, 0), + box("In flight · queued · errors", "", load, 0)), " ") + "\n\n") + net := append(named(m.series("netin"), "in"), named(m.series("netout"), "out")...) + disk := append(named(m.series("diskr"), "read"), named(m.series("diskw"), "written")...) + b.WriteString(indent(side(w, netBox("Network", net), netBox("Disk", disk)), " ") + "\n\n") + } else { + b.WriteString(" " + sDim.Render("Requests, tokens, latency, network and disk are drawn from Prometheus.") + "\n\n") + } + + // Each node: its CPU and memory now, and how they went. + cpuBy, memBy, netBy := byName(m.series("nodecpu")), byName(m.series("nodemem")), byName(m.series("nodenet")) + sw := max(10, (pw-50)/2) + b.WriteString(" " + mdBold.Render("Nodes") + sDim.Render(fmt.Sprintf("%*s", 25+sw, "CPU")) + + sDim.Render(fmt.Sprintf("%*s", 18+sw, "memory")) + "\n") + for _, nd := range m.data.nodes { + c, mm := 0.0, 0.0 + if nd.Load != nil { + c, mm = nd.Load.CPUPercent, nd.Load.MemPercent + } + netCol := "" + if v, ok := netBy[nd.ID]; ok { + netCol = fmt.Sprintf(" net %s %9s", sparkline(v, 12, 0), bps(last(v))) + } + fmt.Fprintf(&b, " %-16s %s %4.0f%% %s %s %4.0f%% %s%s\n", truncate(nd.ID, 16), + meter(c/100, 8), c, sparkline(cpuBy[nd.ID], sw, 100), meter(mm/100, 8), mm, sparkline(memBy[nd.ID], sw, 100), netCol) + } + // Each card: busy, memory, heat and power. + if gpus := m.series("gpu"); len(gpus) > 0 { + vram, temp, pow := byName(m.series("vram")), byName(m.series("gputemp")), byName(m.series("gpupow")) + b.WriteString("\n " + mdBold.Render("GPUs") + "\n") + for _, g := range gpus { + fmt.Fprintf(&b, " %-16s busy %s %4.0f%% memory %s %4.0f%% %3.0f°C %s %4.0f W\n", truncate(g.name, 16), + sparkline(g.values, sw/2, 100), last(g.values), sparkline(vram[g.name], sw/2, 100), last(vram[g.name]), + last(temp[g.name]), sparkline(temp[g.name], 8, 100), last(pow[g.name])) + } + } + // Each model's requests. + if rps := m.series("rpsby"); len(rps) > 0 { + b.WriteString("\n " + mdBold.Render("Models") + sDim.Render(" requests a second") + "\n") + for _, s := range rps { + fmt.Fprintf(&b, " %-30s %s %s\n", truncate(s.name, 30), sparkline(s.values, max(20, pw-50), 0), rate(last(s.values))) + } + } + return "Monitor", head.String() + b.String(), keys +} + +func named(s []chartSeries, name string) []chartSeries { + for i := range s { + s[i].name = name + } + return s +} + +func byName(s []chartSeries) map[string][]float64 { + out := map[string][]float64{} + for _, x := range s { + out[x.name] = x.values + } + return out +} + +func last(v []float64) float64 { + for i := len(v) - 1; i >= 0; i-- { + if !math.IsNaN(v[i]) { + return v[i] + } + } + return 0 +} + +// Charts on a node's and a model's own page: the same reading, kept apart so +// it does not disturb the monitor's. + +func detailQueries(kind, id string, window time.Duration, points int) (time.Duration, []adminapi.MetricsQuery) { + step := max(5*time.Second, (window / time.Duration(max(points, 1))).Round(time.Second)) + rw := promDur(max(time.Minute, 4*step).Round(time.Second)) + q := strconv.Quote(id) + card := func(expr string) string { return `label_join(` + expr + `, "card", ":", "node", "gpu")` } + if kind == "node" { + return step, []adminapi.MetricsQuery{ + {ID: "cpu", Expr: `llmgw_node_cpu_utilization_percent{node=` + q + `}`}, + {ID: "mem", Expr: `100 * llmgw_node_memory_used_bytes{node=` + q + `} / llmgw_node_memory_bytes{node=` + q + `}`}, + {ID: "gpu", Expr: card(`llmgw_gpu_utilization_percent{node=` + q + `}`), By: "card"}, + {ID: "vram", Expr: card(`100 * llmgw_gpu_memory_used_bytes{node=` + q + `} / llmgw_gpu_memory_total_bytes{node=` + q + `}`), By: "card"}, + {ID: "netin", Expr: `sum(rate(node_network_receive_bytes_total{` + netDevs + `,node=` + q + `}[` + rw + `]))`}, + {ID: "netout", Expr: `sum(rate(node_network_transmit_bytes_total{` + netDevs + `,node=` + q + `}[` + rw + `]))`}, + {ID: "diskr", Expr: `sum(rate(node_disk_read_bytes_total{` + diskDevs + `,node=` + q + `}[` + rw + `]))`}, + {ID: "diskw", Expr: `sum(rate(node_disk_written_bytes_total{` + diskDevs + `,node=` + q + `}[` + rw + `]))`}, + } + } + return step, []adminapi.MetricsQuery{ + {ID: "rps", Expr: `sum(rate(llmgw_requests_total{model=` + q + `}[` + rw + `]))`}, + {ID: "errs", Expr: `sum(rate(llmgw_requests_total{model=` + q + `,status=~"5..|429"}[` + rw + `]))`}, + {ID: "tok", Expr: `sum by (direction) (rate(llmgw_tokens_total{model=` + q + `,direction!="cached"}[` + rw + `]))`, By: "direction"}, + {ID: "ttft50", Expr: `histogram_quantile(0.5, sum by (le) (rate(llmgw_ttft_seconds_bucket{model=` + q + `}[` + rw + `])))`}, + {ID: "ttft99", Expr: `histogram_quantile(0.99, sum by (le) (rate(llmgw_ttft_seconds_bucket{model=` + q + `}[` + rw + `])))`}, + } +} + +// detailTarget is the node or model whose page is open, if any. +func (m consoleModel) detailTarget() (kind, id string) { + if !m.detail { + return "", "" + } + switch { + case m.menu == cNodes && m.row < len(m.data.nodes): + return "node", m.data.nodes[m.row].ID + case m.menu == cModels && m.row < len(m.data.models): + return "model", m.data.models[m.row].Name + } + return "", "" +} + +func (m consoleModel) detailKey() string { + kind, id := m.detailTarget() + return fmt.Sprintf("%s/%s/%d/%d", kind, id, m.mon.rng, m.monitorWidth()) +} + +func (m consoleModel) detailDue() bool { + if k, _ := m.detailTarget(); k == "" { + return false + } + return m.det.key != m.detailKey() || time.Since(m.det.at) >= monRanges[m.mon.rng].refresh +} + +type detailMsg monitorMsg + +func (m *consoleModel) fetchDetail() tea.Cmd { + kind, id := m.detailTarget() + if kind == "" { + return nil + } + r := monRanges[m.mon.rng] + step, qs := detailQueries(kind, id, r.window, (m.monitorWidth()/2-2)*2) + key, c := m.detailKey(), m.client + m.det.at = time.Now() + return func() tea.Msg { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + var res adminapi.MetricsResult + err := c.do(ctx, http.MethodPost, "/api/v1/metrics/range", + adminapi.MetricsRequest{Window: r.window, Step: step, Queries: qs}, &res) + return detailMsg{key: key, res: res, err: err} + } +} + +// detailCharts are the charts at the foot of a node's or model's page. +func (m consoleModel) detailCharts(pw int) string { + kind, id := m.detailTarget() + if kind == "" { + return "" + } + r := monRanges[m.mon.rng] + head := "\n " + mdBold.Render("Over the last "+r.label) + sDim.Render(" (1–4 or R another range)") + "\n\n" + if m.det.err != nil { + return head + " " + sBad.Render(m.det.err.Error()) + } + get := func(qid string) []chartSeries { + var out []chartSeries + for _, s := range m.det.res.Series[qid] { + cs := chartSeries{name: s.Name} + for _, p := range s.Points { + cs.values = append(cs.values, p.Value) + } + out = append(out, cs) + } + return out + } + if !m.det.res.Configured { + // Without Prometheus, a node's own samples from this console. + if kind != "node" { + return head + " " + sDim.Render("A model's traffic is drawn from Prometheus: set monitoring.prometheus in server.yaml.") + } + get = func(qid string) []chartSeries { + var out []chartSeries + for _, s := range m.mon.localSeries("node" + qid) { + if s.Name == id { + cs := chartSeries{name: s.Name} + for _, p := range s.Points { + cs.values = append(cs.values, p.Value) + } + out = append(out, cs) + } + } + return out + } + } + w := (pw - 6) / 2 + box := func(title, now string, s []chartSeries, top float64) string { + if len(s) == 0 { + return mdBold.Render(title) + "\n\n " + sDim.Render("no samples") + } + return chartBox(title, now, s, w, 5, top, r.label) + } + var b strings.Builder + b.WriteString(head) + if kind == "node" { + cpu, mem := get("cpu"), get("mem") + b.WriteString(indent(side(w, box("CPU", fmt.Sprintf("%.0f%%", lastOf(cpu)), cpu, 100), + box("Memory", fmt.Sprintf("%.0f%%", lastOf(mem)), mem, 100)), " ")) + if g := get("gpu"); len(g) > 0 { + b.WriteString("\n\n" + indent(side(w, box("GPU busy", "", g, 100), box("GPU memory", "", get("vram"), 100)), " ")) + } + if m.det.res.Configured { + net := append(named(get("netin"), "in"), named(get("netout"), "out")...) + disk := append(named(get("diskr"), "read"), named(get("diskw"), "written")...) + if len(net)+len(disk) > 0 { + b.WriteString("\n\n" + indent(side(w, box("Network", bps(lastOf(net)), net, 0), box("Disk", bps(lastOf(disk)), disk, 0)), " ")) + } else { + b.WriteString("\n\n " + sDim.Render("Network and disk come from the node's exporter, which is not answering: apt-get install prometheus-node-exporter")) + } + } + return b.String() + } + rps, tok := get("rps"), get("tok") + b.WriteString(indent(side(w, box("Requests", rate(lastOf(rps)), rps, 0), box("Tokens", rate(lastOf(tok)), tok, 0)), " ")) + ttft := append(named(get("ttft50"), "p50"), named(get("ttft99"), "p99")...) + b.WriteString("\n\n" + indent(side(w, box("Time to first token (s)", "", ttft, 0), + box("Errors", rate(lastOf(get("errs"))), get("errs"), 0)), " ")) + return b.String() +} diff --git a/internal/cli/zz_live_render_test.go b/internal/cli/zz_live_render_test.go new file mode 100644 index 00000000..4b280d60 --- /dev/null +++ b/internal/cli/zz_live_render_test.go @@ -0,0 +1,33 @@ +//go:build liverender + +package cli + +import ( + "context" + "fmt" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/promq" +) + +func TestLiveRender(t *testing.T) { + p := promq.New("http://10.0.0.146:9090", 10*time.Second) + m := newConsoleModel("127.0.0.1:1", "/nonexistent") + m.loaded, m.menu, m.inPanel, m.width, m.height = true, cMonitor, true, 170, 70 + m.mon.rng = 1 + step, qs := monitorQueries(time.Hour, (m.monitorWidth()/2-2)*2) + res := adminapi.MetricsResult{Configured: true, Series: map[string][]promq.Series{}} + for _, q := range qs { + s, err := p.RangeQuery(context.Background(), q.Expr, q.By, time.Hour, step) + if err != nil { + t.Logf("%s: %v", q.ID, err) + } + res.Series[q.ID] = s + } + m.mon.res = res + m.mon.key = m.monitorKey() + _, body, _ := m.monitorPanel(m.monitorWidth()) + fmt.Println(body) +} diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index ec37921a..910d5ec7 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -45,6 +45,8 @@ type API struct { storeStatus func() ([]StoreMember, error) // serve runs registry-to-replica jobs; nil without a mirror and a store. serve *modelserve.Server + // history is the Prometheus the console's charts read from. + history History models Models capacity Capacity @@ -135,6 +137,7 @@ func (a *API) Register(mux *http.ServeMux) { mux.HandleFunc("POST /api/v1/models/{name}/rollback", a.rollbackModel) mux.HandleFunc("POST /api/v1/models/{name}/origin", a.setOrigin) mux.HandleFunc("GET /api/v1/plan", a.planModel) + mux.HandleFunc("POST /api/v1/metrics/range", a.metricsBatch) mux.HandleFunc("GET /api/v1/requests/{id}", a.request) mux.HandleFunc("POST /api/v1/models/{name}/retry", a.retryModel) mux.HandleFunc("GET /api/v1/models/{name}/headroom", func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/control/adminapi/metrics.go b/internal/control/adminapi/metrics.go new file mode 100644 index 00000000..be4192da --- /dev/null +++ b/internal/control/adminapi/metrics.go @@ -0,0 +1,89 @@ +package adminapi + +import ( + "context" + "encoding/json" + "io" + "net/http" + "sync" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/promq" +) + +// History reads the cluster's metrics back from the Prometheus that scraped +// them: what the console's charts draw. +type History interface { + RangeQuery(ctx context.Context, expr, nameBy string, window, step time.Duration) ([]promq.Series, error) +} + +// SetHistory supplies the Prometheus to read from; nil leaves the charts to +// what the console saw itself. +func (a *API) SetHistory(h History) { a.history = h } + +// MetricsQuery is one line set of a chart: a PromQL expression, and the label +// naming each line. +type MetricsQuery struct { + ID string `json:"id"` + Expr string `json:"expr"` + By string `json:"by,omitempty"` +} + +// MetricsRequest asks for several queries over one window at once. +type MetricsRequest struct { + Window time.Duration `json:"window"` + Step time.Duration `json:"step"` + Queries []MetricsQuery `json:"queries"` +} + +// MetricsResult is each query's lines, by id; Configured is false when no +// Prometheus is set, and Errors says which queries failed and why. +type MetricsResult struct { + Configured bool `json:"configured"` + Series map[string][]promq.Series `json:"series"` + Errors map[string]string `json:"errors,omitempty"` +} + +// metricsBatch runs the queries side by side: a screen of charts is a dozen +// of them, and one after another they would take as long as the slowest +// dozen times. +func (a *API) metricsBatch(w http.ResponseWriter, r *http.Request) { + var req MetricsRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) + return + } + res := MetricsResult{Series: map[string][]promq.Series{}} + if a.history == nil || len(req.Queries) == 0 { + writeJSON(w, http.StatusOK, res) + return + } + res.Configured = true + if req.Window <= 0 { + req.Window = time.Hour + } + if len(req.Queries) > 40 { + req.Queries = req.Queries[:40] + } + var mu sync.Mutex + var wg sync.WaitGroup + for _, q := range req.Queries { + wg.Add(1) + go func(q MetricsQuery) { + defer wg.Done() + s, err := a.history.RangeQuery(r.Context(), q.Expr, q.By, req.Window, req.Step) + mu.Lock() + defer mu.Unlock() + if err != nil { + if res.Errors == nil { + res.Errors = map[string]string{} + } + res.Errors[q.ID] = err.Error() + return + } + res.Series[q.ID] = s + }(q) + } + wg.Wait() + writeJSON(w, http.StatusOK, res) +} diff --git a/internal/control/adminapi/promsd.go b/internal/control/adminapi/promsd.go new file mode 100644 index 00000000..4edfdaf3 --- /dev/null +++ b/internal/control/adminapi/promsd.go @@ -0,0 +1,73 @@ +package adminapi + +import ( + "net" + "net/http" + "os" + "sort" + "strings" +) + +// Prometheus service discovery: the machines of the cluster as scrape targets +// for their node exporters, which say what llmgw does not — network and disk +// traffic, filesystems, every CPU's time. Served by the control plane, which +// knows the fleet, so a node added is a node scraped with nothing edited; and +// labelled with the node's id in llmgw, so its charts can be put beside the +// cluster's own. + +// NodeExporterPort is where prometheus-node-exporter listens. +const NodeExporterPort = "9100" + +// SDGroup is one target group of Prometheus' HTTP service discovery. +type SDGroup struct { + Targets []string `json:"targets"` + Labels map[string]string `json:"labels"` +} + +// NodeExporterTargets is every machine of the cluster: the nodes, this +// control plane, as the request reached it, and the store's members. +func (a *API) NodeExporterTargets(r *http.Request) []SDGroup { + return a.nodeExporterTargetsWith(r, a.Nodes()) +} + +func (a *API) nodeExporterTargetsWith(r *http.Request, nodes []NodeView) []SDGroup { + seen := map[string]bool{} + var out []SDGroup + add := func(host, node, role string) { + if host == "" || seen[host] { + return + } + seen[host] = true + out = append(out, SDGroup{Targets: []string{net.JoinHostPort(host, NodeExporterPort)}, + Labels: map[string]string{"node": node, "role": role}}) + } + if host, _, err := net.SplitHostPort(r.Host); err == nil { + name, _ := os.Hostname() + add(host, name, "control-plane") + } + for _, n := range nodes { + host := n.MgmtAddr + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + add(host, n.ID, "node") + } + if a.storeStatus != nil { + if members, err := a.storeStatus(); err == nil { + for _, m := range members { + host := m.Host + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + add(strings.TrimSpace(host), m.Name, "store") + } + } + } + sort.Slice(out, func(i, j int) bool { return out[i].Labels["node"] < out[j].Labels["node"] }) + return out +} + +// PromSD serves the targets as Prometheus' http_sd_configs reads them. +func (a *API) PromSD(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, a.NodeExporterTargets(r)) +} diff --git a/internal/control/adminapi/promsd_test.go b/internal/control/adminapi/promsd_test.go new file mode 100644 index 00000000..9ac6e799 --- /dev/null +++ b/internal/control/adminapi/promsd_test.go @@ -0,0 +1,23 @@ +package adminapi + +import ( + "net/http/httptest" + "testing" +) + +func TestNodeExporterTargetsAreTheControlPlaneAndTheStore(t *testing.T) { + a := &API{} + a.SetStoreStatus(func() ([]StoreMember, error) { + return []StoreMember{{Name: "pg01", Host: "10.0.0.149"}, {Name: "ctl", Host: "10.0.0.140"}}, nil + }) + r := httptest.NewRequest("GET", "http://10.0.0.140:9091/prometheus/targets", nil) + got := a.nodeExporterTargetsWith(r, nil) + if len(got) != 2 { + t.Fatalf("%+v", got) + } + for _, g := range got { + if g.Targets[0] == "10.0.0.140:9100" && g.Labels["role"] != "control-plane" { + t.Errorf("the control plane twice: %+v", got) + } + } +} diff --git a/internal/onboard/onboard.go b/internal/onboard/onboard.go index 48b595f6..42173375 100644 --- a/internal/onboard/onboard.go +++ b/internal/onboard/onboard.go @@ -60,6 +60,9 @@ func ParseFabric(s string) (Fabric, error) { // Options say which machine, and what to tell its agent. type Options struct { + // NoNodeExporter leaves Prometheus' node exporter off the machine. + NoNodeExporter bool + // Addr is how to reach the machine over SSH, and User who to log in as. Addr, User string // Node is its id in the cluster. Empty means its hostname, which is what @@ -673,9 +676,41 @@ systemctl is-active llmgw-agent`, nil); err != nil { return fmt.Errorf("the agent did not start: %s; see journalctl -u llmgw-agent on the node", lastLine(out)) } r.emit(StepStart, Done, "llmgw-agent is running", "") + if !r.Opts.NoNodeExporter { + r.nodeExporter(ctx) + } return nil } +// nodeExporter installs Prometheus' node exporter from the distribution, for +// what llmgw does not measure: network and disk traffic, filesystems. Best +// effort: a node without it serves as well, so a failure is a warning. +func (r *Runner) nodeExporter(ctx context.Context) { + out, err := r.Remote.Run(ctx, `set -e +if systemctl is-active --quiet prometheus-node-exporter 2>/dev/null || systemctl is-active --quiet node_exporter 2>/dev/null; then + echo already +elif command -v apt-get >/dev/null 2>&1; then + sudo DEBIAN_FRONTEND=noninteractive apt-get install -y -qq prometheus-node-exporter >/dev/null + sudo systemctl enable --now prometheus-node-exporter >/dev/null 2>&1 + echo installed +elif command -v dnf >/dev/null 2>&1; then + sudo dnf install -y -q golang-github-prometheus-node-exporter >/dev/null + sudo systemctl enable --now node_exporter >/dev/null 2>&1 + echo installed +else + echo "no apt-get or dnf"; exit 1 +fi`, nil) + switch { + case err != nil: + r.emit(StepStart, Warn, "the Prometheus node exporter was not installed: "+lastLine(out)+ + "; the node serves without it, and its network and disk are not charted", "") + case strings.Contains(string(out), "already"): + r.emit(StepStart, Info, "the Prometheus node exporter was already running", "") + default: + r.emit(StepStart, Info, "the Prometheus node exporter is installed, on port 9100", "") + } +} + // --- helpers --- func lastLine(b []byte) string {