The full switch: Monitors and Disks read only the sensor API; /api/metrics retired

- stock Monitors modules fed by one sensor stream, the catalog and one batch of histories (GET /api/sensors/history?ids=…), through sampleAdapter.ts - the stream closes while the page is hidden and refills its curves on return - Disks tab live rates from the stream - new sensors: hamlab.sample_interval, ups.last_event - stream: asking for about the sampling rate means every sample - NUT runtime rounded at the source Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-19 07:52:14 -07:00
parent f01175ceca
commit 4ddf10be51
13 changed files with 283 additions and 173 deletions
@@ -3,7 +3,8 @@ import { api, DashPrefs, Sample, Series, Temp } from './api'
 import Sparkline from './Sparkline'
 import { Loading, Spinner, toast } from './ui'
 import { fmtMs, useRefreshMs } from './prefs'
-import { useSensorStream } from './sensors'
+import { fetchCatalog, fetchHistories, useSensorStream } from './sensors'
+import { Catalog, MONITOR_HISTORY, MONITOR_SHAPE, Values, sampleOf, seriesOf, stale } from './sampleAdapter'
 import { Widget, WidgetData, WidgetSpec, shapeOf } from './widgets'
 import WidgetPicker from './WidgetPicker'

@@ -272,8 +273,6 @@ export default function Dashboard() {
  const hist = useRef<Series | null>(null)
  const saveTimer = useRef<number | undefined>(undefined)
  const refreshMs = useRefreshMs()
-  const intervalRef = useRef(DEFAULT_INTERVAL)
-  intervalRef.current = refreshMs

  // Persist layout changes server-side (debounced) so every device sees the same dashboard.
  const update = useCallback((fn: (l: Layout) => Layout) => {
@@ -295,45 +294,52 @@ export default function Dashboard() {

  useEffect(() => {
    let stop = false
-    let timer: number | undefined
    api.dashPrefs().then((p) => { if (!stop) { setLayout(normalize(p)); setLoaded(true) } }).catch(() => setErr('could not load layout'))
-    const seed = async () => {
-      try { hist.current = await api.metricsHistory(); setLatest(await api.metrics()); setErr('') }
-      catch (e) { setErr(e instanceof Error ? e.message : 'failed') }
-    }
-    const tick = async () => {
-      if (stop) return
-      if (document.visibilityState === 'visible') {
-        try {
-          const s = await api.metrics()
-          const h = hist.current
-          if (h) {
-            push(h.t, s.t); push(h.cpu, s.cpu); push(h.mem, s.mem.used)
-            const n = h.t.length
-            for (const x of s.net) {
-              const v = (h.net[x.iface] ??= { rx: [], tx: [] })
-              pad(v.rx, n - 1); pad(v.tx, n - 1); push(v.rx, x.rx_bps); push(v.tx, x.tx_bps)
-            }
-            for (const d of s.disks) {
-              const v = (h.disk[d.name] ??= { r: [], w: [], u: [] })
-              pad(v.r, n - 1); pad(v.w, n - 1); pad(v.u, n - 1); push(v.r, d.read_bps); push(v.w, d.write_bps); push(v.u, d.util_pct)
-            }
-            for (const t of s.temps) {
-              const v = (h.temp[`${t.chip}/${t.label}`] ??= [])
-              pad(v, n - 1); push(v, t.c)
-            }
-          }
-          setLatest(s); setErr('')
-        } catch (e) { setErr(e instanceof Error ? e.message : 'failed') }
-      }
-      timer = window.setTimeout(tick, intervalRef.current)
-    }
-    seed().then(() => { timer = window.setTimeout(tick, intervalRef.current) })
-    const onVis = () => { if (document.visibilityState === 'visible' && !hist.current) seed() }
-    document.addEventListener('visibilitychange', onVis)
-    return () => { stop = true; clearTimeout(timer); document.removeEventListener('visibilitychange', onVis) }
+    return () => { stop = true }
  }, [])

+  // Everything on this page comes from the sensor API: the catalog for names, labels and
+  // thresholds; one batch of histories to seed the curves; one stream for the values.
+  const catalog = useRef<Catalog>(new Map())
+  const refetching = useRef(false)
+  const stock = useSensorStream<{ v: Values }>(MONITOR_SHAPE, { full: false, everyMs: refreshMs })
+  const seed = useCallback(async () => {
+    try {
+      catalog.current = new Map((await fetchCatalog()).map((x) => [x.id, x]))
+      hist.current = seriesOf(await fetchHistories(MONITOR_HISTORY), catalog.current)
+      setErr('')
+    } catch (e) { setErr(e instanceof Error ? e.message : 'failed') }
+  }, [])
+  useEffect(() => { seed() }, [seed, stock.resumed]) // on load, and again after the page was hidden: the curves have a gap to refill
+  useEffect(() => {
+    const v = stock.data?.v
+    if (!v || catalog.current.size === 0) return
+    if (stale(v, catalog.current) && !refetching.current) { // something appeared (a disk, an interface): learn its labels
+      refetching.current = true
+      fetchCatalog().then((c) => { catalog.current = new Map(c.map((x) => [x.id, x])) }).catch(() => {}).finally(() => window.setTimeout(() => { refetching.current = false }, 10000))
+    }
+    const s = sampleOf(v, catalog.current)
+    const h = hist.current
+    if (h) {
+      push(h.t, s.t); push(h.cpu, s.cpu); push(h.mem, s.mem.used)
+      const n = h.t.length
+      for (const x of s.net) {
+        const c = (h.net[x.iface] ??= { rx: [], tx: [] })
+        pad(c.rx, n - 1); pad(c.tx, n - 1); push(c.rx, x.rx_bps); push(c.tx, x.tx_bps)
+      }
+      for (const d of s.disks) {
+        const c = (h.disk[d.name] ??= { r: [], w: [], u: [] })
+        pad(c.r, n - 1); pad(c.w, n - 1); pad(c.u, n - 1); push(c.r, d.read_bps); push(c.w, d.write_bps); push(c.u, d.util_pct)
+      }
+      for (const t of s.temps) {
+        const c = (h.temp[`${t.chip}/${t.label}`] ??= [])
+        pad(c, n - 1); push(c, t.c)
+      }
+    }
+    setLatest(s)
+  }, [stock.data])
+  useEffect(() => { if (stock.error && !latest) setErr(stock.error) }, [stock.error, latest])
+
  // The sampler on this machine follows the global rate (clamped to what it supports).
  useEffect(() => {
    if (!loaded) return
@@ -1,9 +1,10 @@
-import { useCallback, useEffect, useState } from 'react'
-import { api, Sample, SmartInfo, StorageDevice, ZFSInfo, ZFSSnapshot } from './api'
+import { useCallback, useEffect, useMemo, useState } from 'react'
+import { api, SmartInfo, StorageDevice, ZFSInfo, ZFSSnapshot } from './api'
 import { fmtBytes, fmtRate } from './Dashboard'
 import TaskOutput from './TaskOutput'
 import { Btn, Loading, Spinner, toast } from './ui'
 import { useRefreshMs } from './prefs'
+import { Sensor, useSensorStream } from './sensors'

 const pct = (v: number) => `${Math.round(v)}%`
 const KIND: Record<string, string> = { nvme: 'NVMe', ssd: 'SSD', hdd: 'HDD', usb: 'USB', emmc: 'eMMC', sd: 'SD', other: 'disk' }
@@ -183,7 +184,6 @@ export default function Disks() {
  const [devs, setDevs] = useState<StorageDevice[] | null>(null)
  const [zfs, setZfs] = useState<ZFSInfo | null>(null)
  const [hasSmart, setHasSmart] = useState(true)
-  const [live, setLive] = useState<Sample | null>(null)
  const [err, setErr] = useState('')
  const [task, setTask] = useState<string | null>(null)
  const [allSmart, setAllSmart] = useState<boolean>(() => localStorage.getItem('hamlab.disks.smart') === '1')
@@ -196,13 +196,20 @@ export default function Disks() {
    } catch (e) { setErr(e instanceof Error ? e.message : 'failed') }
  }, [])
  useEffect(() => { load() }, [load])
-  useEffect(() => {
-    let stop = false
-    const tick = async () => { if (!stop && document.visibilityState === 'visible') { try { setLive(await api.metrics()) } catch { /* ignore */ } } }
-    tick()
-    const t = setInterval(tick, rate)
-    return () => { stop = true; clearInterval(t) }
-  }, [rate])
+  // Live per-device rates come from the sensor stream, grouped by the kernel name in each sensor's labels.
+  const stream = useSensorStream<{ d: Record<string, Sensor> }>({ d: 'disk.*' }, { everyMs: rate })
+  const live = useMemo(() => {
+    const by: Record<string, { name: string; read_bps: number; write_bps: number; util_pct: number }> = {}
+    for (const x of Object.values(stream.data?.d ?? {})) {
+      const name = x.labels?.device
+      if (!name || typeof x.value !== 'number') continue
+      const row = (by[name] ??= { name, read_bps: 0, write_bps: 0, util_pct: 0 })
+      if (x.id.endsWith('.read')) row.read_bps = x.value
+      else if (x.id.endsWith('.write')) row.write_bps = x.value
+      else if (x.id.endsWith('.util')) row.util_pct = x.value
+    }
+    return { disks: Object.values(by) }
+  }, [stream.data])

  if (!devs) return <div className="dash">{err ? <div className="error">{err}</div> : <Loading label="Scanning disks…" />}</div>
  const liveFor = (dev: StorageDevice) => {
@@ -18,7 +18,7 @@ export default function Embed() {
  const spec: WidgetSpec = { id: 'w', type, sensor, title: p.get('title') ?? undefined }
  const src: Source = { machine: p.get('machine') || null, token: p.get('token') ?? undefined }
  const every = Number(p.get('every')) || undefined
-  const live = useSensorStream<{ w: WidgetData }>(sensor ? { w: sensor } : null, src, every)
+  const live = useSensorStream<{ w: WidgetData }>(sensor ? { w: sensor } : null, { src, everyMs: every })
  return (
    <div className={'embed' + (p.get('bare') === '1' ? ' bare' : '')}>
      {!sensor ? <div className="error">embed needs ?sensor=</div> : (
@@ -168,8 +168,6 @@ export const api = {
  serviceAction: (unit: string, action: string) => req<void>('POST', `/api/services/${encodeURIComponent(unit)}/${action}`),
  dashPrefs: () => req<DashPrefs>('GET', '/api/prefs/dashboard'),
  saveDashPrefs: (p: DashPrefs) => req<void>('PUT', '/api/prefs/dashboard', p),
-  metrics: () => req<Sample>('GET', '/api/metrics'),
-  metricsHistory: () => req<Series>('GET', '/api/metrics/history'),
  login: (password: string) => req<void>('POST', '/api/login', { password }, true),
  logout: () => req<void>('POST', '/api/logout', undefined, true),
  meHere: () => req<Me>('GET', '/api/me', undefined, true),
@@ -0,0 +1,71 @@
+// The stock Monitors modules were written against one Sample object. They now read the
+// sensor API like everything else; this file is the seam — it rebuilds that object from
+// sensor values (the live stream) and sensor metadata (the catalog: labels, thresholds).
+import { Sample, Series } from './api'
+import { Sensor } from './sensors'
+
+export type Values = Record<string, number | boolean | string>
+export type Catalog = Map<string, Sensor>
+
+// Everything the modules draw, as one stream query.
+export const MONITOR_SHAPE = { v: '*' }
+// The curves they keep: CPU, memory, per-NIC and per-disk rates, every temperature.
+export const MONITOR_HISTORY = 'cpu.usage,mem.used,net.*.rx,net.*.tx,disk.*.read,disk.*.write,disk.*.util,temp.*'
+
+const num = (v: Values, id: string) => (typeof v[id] === 'number' ? (v[id] as number) : 0)
+const str = (v: Values, id: string) => (typeof v[id] === 'string' ? (v[id] as string) : '')
+
+// Groups ids like net.<x>.rx by the real name in the catalog's labels, keeping catalog order.
+function groups(v: Values, cat: Catalog, prefix: string, label: string): Map<string, string> {
+  const out = new Map<string, string>() // real name → id stem ("net.enp3s0")
+  for (const id of Object.keys(v)) {
+    if (!id.startsWith(prefix)) continue
+    const real = cat.get(id)?.labels?.[label]
+    if (real && !out.has(real)) out.set(real, id.slice(0, id.lastIndexOf('.')))
+  }
+  return new Map([...out].sort((a, b) => a[0].localeCompare(b[0]))) // by real name: "/" leads the filesystems
+}
+
+// Every id in the values must be known to the catalog, or the catalog is stale (a disk
+// or interface appeared) and wants refetching.
+export const stale = (v: Values, cat: Catalog) => Object.keys(v).some((id) => !cat.has(id))
+
+export function sampleOf(v: Values, cat: Catalog): Sample {
+  const cores = Object.keys(v).map((id) => /^cpu\.core\.(\d+)\.usage$/.exec(id)).filter(Boolean).map((m) => [Number(m![1]), v[m![0]] as number] as const).sort((a, b) => a[0] - b[0]).map((x) => x[1])
+  const mem = (p: string) => ({ total: num(v, p + '.total'), used: num(v, p + '.used'), avail: num(v, p + '.avail'), free: num(v, p + '.free'), cached: num(v, p + '.cached'), buffers: num(v, p + '.buffers') })
+  const hw = (prefix: string) => Object.keys(v).filter((id) => id.startsWith(prefix) && cat.get(id)?.labels?.chip !== undefined).map((id) => ({ id, s: cat.get(id)! }))
+    .sort((a, b) => a.s.labels!.chip.localeCompare(b.s.labels!.chip) || a.s.labels!.label.localeCompare(b.s.labels!.label))
+  return {
+    t: Date.now(), interval_ms: num(v, 'hamlab.sample_interval'),
+    cpu: num(v, 'cpu.usage'), cores, load: [num(v, 'load.1'), num(v, 'load.5'), num(v, 'load.15')],
+    mem: mem('mem'), swap: mem('swap'), uptime: num(v, 'uptime'),
+    temps: hw('temp.').map(({ id, s }) => ({ chip: s.labels!.chip, label: s.labels!.label, c: num(v, id), max: s.warn, crit: s.crit })),
+    fans: hw('fan.').map(({ id, s }) => ({ chip: s.labels!.chip, label: s.labels!.label, rpm: num(v, id) })),
+    net: [...groups(v, cat, 'net.', 'iface')].map(([iface, p]) => ({ iface, up: v[p + '.up'] === true, rx_bps: num(v, p + '.rx'), tx_bps: num(v, p + '.tx'), rx_total: num(v, p + '.rx_total'), tx_total: num(v, p + '.tx_total') })),
+    disks: [...groups(v, cat, 'disk.', 'device')].map(([name, p]) => ({ name, model: cat.get(p + '.size')?.labels?.model || undefined, size: num(v, p + '.size'), read_bps: num(v, p + '.read'), write_bps: num(v, p + '.write'), util_pct: num(v, p + '.util'), temp: num(v, p + '.temp') || undefined })),
+    fs: [...groups(v, cat, 'fs.', 'mount')].map(([mount, p]) => ({ mount, type: cat.get(p + '.size')?.labels?.fstype ?? '', size: num(v, p + '.size'), used: num(v, p + '.used') })),
+    ups: v['ups.battery'] === undefined ? null : {
+      driver: cat.get('ups.last_event')?.labels?.driver ?? '', model: cat.get('ups.battery')?.labels?.model ?? '', state: str(v, 'ups.state'), source: str(v, 'ups.source'),
+      on_battery: v['ups.on_battery'] === true, in_v: num(v, 'ups.in_v'), out_v: num(v, 'ups.out_v'), battery_pct: num(v, 'ups.battery'), runtime_min: num(v, 'ups.runtime'),
+      load_w: num(v, 'ups.load'), load_pct: num(v, 'ups.load_pct'), last_event: str(v, 'ups.last_event') || undefined,
+    },
+  }
+}
+
+// The modules' history: one shared time axis (the CPU's), every other curve right-aligned
+// to it and zero-padded on the left where the sensor is younger than the ring.
+export function seriesOf(hist: { id: string; t: number[]; v: number[] }[], cat: Catalog): Series {
+  const by = new Map(hist.map((h) => [h.id, h]))
+  const t = by.get('cpu.usage')?.t ?? []
+  const n = t.length
+  const fit = (id: string) => { const v = (by.get(id)?.v ?? []).slice(-n); return [...new Array(Math.max(0, n - v.length)).fill(0), ...v] as number[] }
+  const out: Series = { t: [...t], cpu: fit('cpu.usage'), mem: fit('mem.used'), net: {}, disk: {}, temp: {} }
+  for (const h of hist) {
+    const l = cat.get(h.id)?.labels
+    const stem = h.id.slice(0, h.id.lastIndexOf('.'))
+    if (h.id.startsWith('net.') && l?.iface) out.net[l.iface] ??= { rx: fit(stem + '.rx'), tx: fit(stem + '.tx') }
+    else if (h.id.startsWith('disk.') && l?.device) out.disk[l.device] ??= { r: fit(stem + '.read'), w: fit(stem + '.write'), u: fit(stem + '.util') }
+    else if (h.id.startsWith('temp.') && l?.chip !== undefined) out.temp[`${l.chip}/${l.label}`] = fit(h.id)
+  }
+  return out
+}
@@ -32,24 +32,45 @@ export async function fetchHistory(id: string, src?: Source): Promise<{ t: numbe
  return r.ok ? r.json() : null
 }

-// useSensorStream keeps a shaped query answered: the object comes back filled with whole
-// sensors and is replaced on every sample. One stream serves any number of widgets.
-export function useSensorStream<T = unknown>(shape: Shape | null, src?: Source, everyMs?: number): { data: T | null; error: string } {
+export async function fetchHistories(ids: string, src?: Source): Promise<{ id: string; unit?: string; t: number[]; v: number[] }[]> {
+  const r = await fetch(url(src, '/api/sensors/history', { ids }), { credentials: 'same-origin' })
+  if (!r.ok) throw new Error(r.statusText)
+  return (await r.json()).series
+}
+
+export interface StreamOpts { src?: Source; everyMs?: number; full?: boolean }
+
+// useSensorStream keeps a shaped query answered: the object comes back filled (with whole
+// sensors unless full is false) and is replaced on every sample. One stream serves any
+// number of widgets. It closes while the page is hidden — a phone in a pocket shouldn't
+// hold a stream open — and `resumed` counts the times it came back, so a caller that
+// keeps history knows there is a gap to refill.
+export function useSensorStream<T = unknown>(shape: Shape | null, opts: StreamOpts = {}): { data: T | null; error: string; resumed: number } {
  const [data, setData] = useState<T | null>(null)
  const [error, setError] = useState('')
+  const [resumed, setResumed] = useState(0)
+  const [visible, setVisible] = useState(() => document.visibilityState === 'visible')
  const key = shape === null ? '' : JSON.stringify(shape)
-  const srcKey = `${src?.machine ?? '·'}|${src?.token ?? ''}|${everyMs ?? ''}`
+  const { src, everyMs, full = true } = opts
+  const srcKey = `${src?.machine ?? '·'}|${src?.token ?? ''}|${everyMs ?? ''}|${full}`
  const srcRef = useRef(src); srcRef.current = src
+  useEffect(() => {
+    const onVis = () => { const v = document.visibilityState === 'visible'; setVisible(v); if (v) setResumed((n) => n + 1) }
+    document.addEventListener('visibilitychange', onVis)
+    return () => document.removeEventListener('visibilitychange', onVis)
+  }, [])
  useEffect(() => {
    if (!key) { setData(null); return }
-    const params: Record<string, string> = { q: key, full: '1' }
+    if (!visible) return
+    const params: Record<string, string> = { q: key }
+    if (full) params.full = '1'
    if (everyMs) params.every = String(everyMs)
    const es = new EventSource(url(srcRef.current, '/api/sensors/stream', params))
    es.onmessage = (e) => { try { setData(JSON.parse(e.data)); setError('') } catch { /* skip a bad frame */ } }
    es.onerror = () => setError('reconnecting…') // EventSource retries by itself
    return () => es.close()
-  }, [key, srcKey]) // eslint-disable-line react-hooks/exhaustive-deps
-  return { data, error }
+  }, [key, srcKey, visible]) // eslint-disable-line react-hooks/exhaustive-deps
+  return { data, error, resumed }
 }

 // ---- formatting, by unit ----