diff options
| author | tv <tv@krebsco.de> | 2026-10-30 17:42:26 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-10-30 17:42:26 +0100 |
| commit | f65ecfdf313658a3979a2d0add53e342ce4b97df (patch) | |
| tree | 57320539da53c39c3d0d37870f0f65ba33ca7fc8 /src | |
| parent | b224eee76a9314992ec7a83321803e11a0b0b37f (diff) | |
add dashboard
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/dashboard/dashboard.html | 422 | ||||
| -rw-r--r-- | src/api/dashboard/mod.rs | 394 | ||||
| -rw-r--r-- | src/api/mod.rs | 7 |
3 files changed, 823 insertions, 0 deletions
diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html new file mode 100644 index 0000000..c87341f --- /dev/null +++ b/src/api/dashboard/dashboard.html @@ -0,0 +1,422 @@ +<!doctype html> +<html lang="en"> +<head> +<meta charset="utf-8"> +<meta name="viewport" content="width=device-width, initial-scale=1"> +<title>continuum</title> +<style> + :root { + --bg: #fafaf9; --fg: #1c1917; --muted: #78716c; --line: #e7e5e4; --panel: #ffffff; + --ok: #15803d; --warn: #c2410c; --bad: #b91c1c; --intended: #a8a29e; --self: #1d4ed8; + } + @media (prefers-color-scheme: dark) { + :root { + --bg: #1c1917; --fg: #f5f5f4; --muted: #a8a29e; --line: #44403c; --panel: #292524; + --ok: #4ade80; --warn: #fb923c; --bad: #f87171; --intended: #57534e; --self: #60a5fa; + } + } + * { box-sizing: border-box; } + html { background: var(--bg); } + /* The window's height, unless it is way too small: then the page scrolls. */ + body { margin: 0; background: var(--bg); color: var(--fg); + font: 14px/1.4 system-ui, sans-serif; + height: 100vh; min-height: 480px; display: flex; flex-direction: column; } + header { display: flex; gap: 1em; align-items: baseline; padding: 12px 16px; + border-bottom: 1px solid var(--line); flex-wrap: wrap; } + header h1 { font-size: 16px; margin: 0; } + #connection { margin-left: auto; color: var(--muted); } + #connection.live { color: var(--ok); } + #connection.down { color: var(--bad); } + /* Swarm on the left, the full height; peers (as tall as their rows, up to 8) above + the figures (the rest) on the right. */ + main { flex: 1; min-height: 0; display: grid; gap: 16px; padding: 16px; + grid-template-columns: minmax(0, 3fr) minmax(0, 2fr); + grid-template-rows: auto minmax(0, 1fr); + grid-template-areas: "map peers" "map figures"; } + @media (max-width: 900px) { + body { height: auto; } + main { grid-template-columns: minmax(0, 1fr); grid-template-rows: 70vh auto 70vh; + grid-template-areas: "map" "peers" "figures"; } + } + section { background: var(--panel); border: 1px solid var(--line); border-radius: 6px; + padding: 12px; min-width: 0; min-height: 0; display: flex; flex-direction: column; } + section h2 { font-size: 13px; text-transform: uppercase; letter-spacing: .04em; + color: var(--muted); margin: 0 0 8px; flex: none; } + section h2 .what { text-transform: none; letter-spacing: 0; color: var(--fg); } + #map { grid-area: map; } + #peers-box { grid-area: peers; } + #figures-box { grid-area: figures; } + svg { flex: 1; min-height: 0; width: 100%; display: block; cursor: default; } + svg circle { cursor: pointer; } + .legend { color: var(--muted); font-size: 12px; display: flex; gap: 1em; flex-wrap: wrap; + flex: none; } + .legend span::before { content: ""; display: inline-block; width: 18px; height: 0; + vertical-align: middle; margin-right: 4px; border-top: 2px solid var(--colour, currentColor); } + .legend .intended::before { border-top-style: dashed; --colour: var(--intended); } + .legend .direct::before { --colour: var(--muted); } + .legend .relayed::before { border-top-style: dashed; --colour: var(--muted); } + .legend .dormant::before { border-top-style: dotted; --colour: var(--bad); } + .scroll { overflow: auto; min-height: 0; } + #figures-box .scroll { flex: 1; } + table { border-collapse: collapse; width: 100%; font-variant-numeric: tabular-nums; } + th, td { text-align: left; padding: 3px 8px 3px 0; border-bottom: 1px solid var(--line); + white-space: nowrap; } + th { color: var(--muted); font-weight: 500; position: sticky; top: 0; + background: var(--panel); } + #peers tr[data-peer] { cursor: pointer; } + #peers tr.focused td { background: var(--line); } + .id { font-family: ui-monospace, monospace; font-size: 12px; } + .active { color: var(--ok); } .dormant { color: var(--bad); } .unknown { color: var(--muted); } + details { margin-left: 1em; } + summary { cursor: pointer; } + .tree { font-family: ui-monospace, monospace; font-size: 12px; } + .tree > details { margin-left: 0; } + .key { color: var(--muted); } +</style> +</head> +<body> +<header> + <h1>continuum</h1> + <span id="node" class="id"></span> + <span id="connection">connecting…</span> +</header> +<main> + <section id="map"> + <h2>Swarm</h2> + <div class="legend" id="legend"></div> + <svg id="graph"></svg> + </section> + <section id="peers-box"> + <h2>Peers</h2> + <div class="scroll" id="peers-scroll"><table id="peers"></table></div> + </section> + <section id="figures-box"> + <h2 id="figures-title">Figures</h2> + <div class="scroll"><div id="stats" class="tree"></div></div> + </section> +</main> +<script> +"use strict"; +// The state as the node published it, and when it did (the last event). +let state = null; +let stateAt = 0; + +// Fields that grow by themselves (`GROWING` and `is_age` in the node's dashboard). +const GROWING = new Set(["uptimeSecs", "dormantForSecs", "observedSecs", + "lastFromNeighbourSecs", "oldestQuarantineSecs"]); +const isAge = key => key === "secsAgo" || key.endsWith("SecsAgo") || GROWING.has(key); +const elapsed = () => Math.floor((Date.now() - stateAt) / 1000); +const aged = (key, value) => + isAge(key) && typeof value === "number" ? value + elapsed() : value; + +// RFC 7386. +function mergePatch(target, patch) { + if (patch === null || typeof patch !== "object" || Array.isArray(patch)) return patch; + const result = target !== null && typeof target === "object" && !Array.isArray(target) + ? target : {}; + for (const [key, value] of Object.entries(patch)) { + if (value === null) delete result[key]; + else result[key] = mergePatch(result[key], value); + } + return result; +} + +function el(tag, attrs, ...children) { + const node = document.createElement(tag); + for (const [key, value] of Object.entries(attrs || {})) node.setAttribute(key, value); + for (const child of children) { + node.append(child instanceof Node ? child : document.createTextNode(String(child))); + } + return node; +} +const SVG = "http://www.w3.org/2000/svg"; +function svg(tag, attrs) { + const node = document.createElementNS(SVG, tag); + for (const [key, value] of Object.entries(attrs || {})) node.setAttribute(key, value); + return node; +} + +const short = id => id.length > 14 ? id.slice(0, 6) + "…" + id.slice(-6) : id; +function name(id) { + const member = state.graph.members[id]; + return (member && member.alias) || (state.peers.peers[id] || {}).alias || short(id); +} +function age(secs) { + if (secs === null || secs === undefined) return ""; + if (secs < 120) return secs + " s"; + if (secs < 7200) return Math.floor(secs / 60) + " min"; + if (secs < 172800) return Math.floor(secs / 3600) + " h"; + return Math.floor(secs / 86400) + " d"; +} + +// The node in focus: its figures are shown. null: this node's (`/admin/stats`). +let focused = null; +function focus(id) { + focused = id === null || id === state.node.id ? null : id; + render(); +} + +// --- Peers --- +function renderPeers() { + const table = document.getElementById("peers"); + const rows = [el("tr", {}, ...["peer", "status", "seen", "via", "rtt", "neighbour", "same epoch"] + .map(h => el("th", {}, h)))]; + for (const [id, peer] of Object.entries(state.peers.peers)) { + const destination = peer.destination; + const path = destination && destination.paths && destination.paths[0]; + const rtt = path && typeof path.rttMs === "number" ? path.rttMs.toFixed(1) + " ms" : ""; + const via = destination + ? destination.address + (destination.connection ? " (" + destination.connection + ")" : "") + : ""; + const row = el("tr", { "data-peer": id, class: id === focused ? "focused" : "" }, + el("td", { title: id }, peer.alias || el("span", { class: "id" }, short(id))), + el("td", { class: peer.status }, peer.status + + (peer.status === "dormant" ? " " + age(aged("dormantForSecs", peer.dormantForSecs)) : "")), + el("td", {}, age(aged("lastSeenSecsAgo", peer.lastSeenSecsAgo))), + el("td", { class: "id" }, via), + el("td", {}, rtt), + el("td", {}, peer.neighbour ? "yes" : ""), + el("td", {}, peer.sameEpoch === null || peer.sameEpoch === undefined + ? "" : peer.sameEpoch ? "yes" : "no")); + row.addEventListener("click", () => focus(id === focused ? null : id)); + rows.push(row); + } + table.replaceChildren(...rows); + // As tall as the rows, up to 8 of them (and the header); then it scrolls. + const scroll = document.getElementById("peers-scroll"); + scroll.style.maxHeight = rows.length > 9 + ? (rows[8].offsetTop + rows[8].offsetHeight) + "px" : ""; +} + +// --- Figures: a tree, its open branches kept across renders --- +const open = new Set(["stats", "peer", "links"]); +function tree(key, value, path) { + if (value !== null && typeof value === "object") { + const details = el("details", {}); + if (open.has(path)) details.open = true; + details.addEventListener("toggle", () => { + if (details.open) open.add(path); else open.delete(path); + }); + const entries = Array.isArray(value) ? value.map((v, i) => [String(i), v]) : Object.entries(value); + details.append(el("summary", {}, el("span", { class: "key" }, key), + Array.isArray(value) ? " [" + value.length + "]" : "")); + for (const [k, v] of entries) details.append(tree(k, v, path + "." + k)); + return details; + } + return el("div", {}, el("span", { class: "key" }, key + ": "), + JSON.stringify(aged(key, value))); +} +function renderStats() { + const title = document.getElementById("figures-title"); + const what = el("span", { class: "what" }); + let shown; + if (focused === null) { + what.textContent = name(state.node.id) + " (this node)"; + shown = [tree("stats", state.stats, "stats")]; + } else { + what.textContent = name(focused); + what.title = focused; + const peer = state.peers.peers[focused]; + const report = state.graph.measured[focused]; + shown = []; + if (peer) shown.push(tree("peer", peer, "peer")); + else shown.push(el("div", { class: "key" }, "Not a peer of this node.")); + if (report) shown.push(tree("links", report, "links")); + } + title.replaceChildren(focused === null ? "Figures: " : "Peer: ", what); + document.getElementById("stats").replaceChildren(...shown); +} + +// --- Swarm map: a small force layout, positions kept across renders --- +const positions = new Map(); +const transportColour = new Map([["mycelium", "#2563eb"], ["libp2p", "#9333ea"], + ["tor", "#0891b2"]]); +const SPARE = ["#ca8a04", "#db2777", "#65a30d"]; +function colourOf(transport) { + if (!transport) return "var(--muted)"; + if (!transportColour.has(transport)) { + transportColour.set(transport, SPARE[(transportColour.size - 3) % SPARE.length]); + } + return transportColour.get(transport); +} + +function renderLegend(measured) { + const transports = new Set(); + for (const { link } of measured.values()) transports.add(link.transport || ""); + const items = [el("span", { class: "intended" }, "intended (member list)")]; + for (const transport of [...transports].sort()) { + const item = el("span", {}, transport || "unknown transport"); + item.style.setProperty("--colour", colourOf(transport || null)); + items.push(item); + } + items.push(el("span", { class: "direct" }, "direct"), + el("span", { class: "relayed" }, "relayed"), + el("span", { class: "dormant" }, "dormant")); + document.getElementById("legend").replaceChildren(...items); +} + +function edges() { + const intended = new Map(); + for (const [a, neighbours] of Object.entries(state.graph.intended)) { + for (const b of neighbours) { + const key = a < b ? a + " " + b : b + " " + a; + intended.set(key, [a, b]); + } + } + // One measured edge per pair: the fresher report wins. + const measured = new Map(); + for (const [a, report] of Object.entries(state.graph.measured)) { + for (const link of report.links) { + const key = a < link.peer ? a + " " + link.peer : link.peer + " " + a; + const known = measured.get(key); + if (!known || report.secsAgo < known.secsAgo) { + measured.set(key, { a, b: link.peer, link, secsAgo: report.secsAgo }); + } + } + } + return { intended, measured }; +} + +function layout(nodes, pairs, width, height) { + for (const id of nodes) { + if (!positions.has(id)) { + positions.set(id, { x: width / 2 + (Math.random() - .5) * width / 2, + y: height / 2 + (Math.random() - .5) * height / 2 }); + } + } + const ideal = Math.min(width, height) / Math.max(2, Math.sqrt(nodes.length) * 1.5); + for (let round = 0; round < 120; round++) { + const force = new Map(nodes.map(id => [id, { x: 0, y: 0 }])); + for (let i = 0; i < nodes.length; i++) { + for (let j = i + 1; j < nodes.length; j++) { + const p = positions.get(nodes[i]), q = positions.get(nodes[j]); + const dx = p.x - q.x, dy = p.y - q.y, d2 = Math.max(dx * dx + dy * dy, 1); + const push = ideal * ideal / d2; + force.get(nodes[i]).x += dx * push / 10; force.get(nodes[i]).y += dy * push / 10; + force.get(nodes[j]).x -= dx * push / 10; force.get(nodes[j]).y -= dy * push / 10; + } + } + for (const [a, b] of pairs) { + const p = positions.get(a), q = positions.get(b); + if (!p || !q) continue; + const dx = q.x - p.x, dy = q.y - p.y, d = Math.max(Math.hypot(dx, dy), 1); + const pull = (d - ideal) / d / 4; + force.get(a).x += dx * pull; force.get(a).y += dy * pull; + force.get(b).x -= dx * pull; force.get(b).y -= dy * pull; + } + for (const id of nodes) { + const p = positions.get(id), f = force.get(id); + p.x = Math.min(width - 40, Math.max(40, p.x + Math.max(-10, Math.min(10, f.x)))); + p.y = Math.min(height - 20, Math.max(20, p.y + Math.max(-10, Math.min(10, f.y)))); + } + } +} + +let lastShape = ""; +function renderGraph() { + const root = document.getElementById("graph"); + const width = root.clientWidth || 800, height = root.clientHeight || 560; + root.setAttribute("viewBox", "0 0 " + width + " " + height); + const nodes = Object.keys(state.graph.members).sort(); + const { intended, measured } = edges(); + renderLegend(measured); + const pairs = [...intended.values(), ...[...measured.values()].map(m => [m.a, m.b])]; + // Laid out anew only when who's there or who links to whom changed. + const shape = nodes.join(",") + "|" + [...intended.keys(), ...measured.keys()].sort().join(","); + if (shape !== lastShape) { + for (const id of [...positions.keys()]) if (!nodes.includes(id)) positions.delete(id); + layout(nodes, pairs, width, height); + lastShape = shape; + } + const children = []; + for (const [key, [a, b]] of intended) { + if (measured.has(key)) continue; + const p = positions.get(a), q = positions.get(b); + children.push(svg("line", { x1: p.x, y1: p.y, x2: q.x, y2: q.y, + stroke: "var(--intended)", "stroke-dasharray": "4 4", "stroke-width": 1 })); + } + for (const { a, b, link, secsAgo } of measured.values()) { + const p = positions.get(a), q = positions.get(b); + if (!p || !q) continue; + const line = svg("line", { x1: p.x, y1: p.y, x2: q.x, y2: q.y, + stroke: link.dormant ? "var(--bad)" : colourOf(link.transport), "stroke-width": 2 }); + // Dotted where dormant, dashed where relayed, solid where direct. + if (link.dormant) { + line.setAttribute("stroke-dasharray", "1 4"); + line.setAttribute("stroke-linecap", "round"); + } else if (link.relayed) { + line.setAttribute("stroke-dasharray", "7 4"); + } + const title = svg("title"); + title.textContent = name(a) + " – " + name(b) + (link.transport ? " via " + link.transport : "") + + (link.relayed ? ", relayed" : "") + (link.dormant ? ", dormant" : "") + + (typeof link.rttMs === "number" ? ", " + link.rttMs.toFixed(1) + " ms" : "") + + ", measured " + age(secsAgo + elapsed()) + " ago"; + line.append(title); + children.push(line); + if (typeof link.rttMs === "number") { + const label = svg("text", { x: (p.x + q.x) / 2, y: (p.y + q.y) / 2 - 3, + "text-anchor": "middle", "font-size": 10, fill: "var(--muted)" }); + label.textContent = link.rttMs < 10 ? link.rttMs.toFixed(1) : Math.round(link.rttMs); + children.push(label); + } + } + for (const id of nodes) { + const p = positions.get(id), member = state.graph.members[id]; + const reported = id in state.graph.measured; + const shown = focused === null ? member.self : id === focused; + if (shown) { + children.push(svg("circle", { cx: p.x, cy: p.y, r: 13, fill: "none", + stroke: "var(--self)", "stroke-width": 2, "stroke-opacity": .5 })); + } + const circle = svg("circle", { cx: p.x, cy: p.y, r: member.self ? 8 : 6, + fill: member.self ? "var(--self)" : reported ? "var(--fg)" : "var(--panel)", + stroke: member.self ? "var(--self)" : "var(--fg)", "stroke-width": 1.5 }); + const title = svg("title"); + title.textContent = id + (reported ? "" : " (no links reported)"); + circle.append(title); + circle.addEventListener("click", event => { event.stopPropagation(); focus(id); }); + const label = svg("text", { x: p.x + 10, y: p.y + 4, "font-size": 12, fill: "var(--fg)" }); + label.textContent = name(id); + label.style.cursor = "pointer"; + label.addEventListener("click", event => { event.stopPropagation(); focus(id); }); + children.push(circle, label); + } + root.replaceChildren(...children); +} + +function render() { + if (!state) return; + document.getElementById("node").textContent = + (state.node.alias ? state.node.alias + " · " : "") + state.node.id; + renderPeers(); + renderStats(); + renderGraph(); +} + +// A click beside the nodes: back to this node. +document.getElementById("graph").addEventListener("click", () => focus(null)); + +// --- Events --- +const connection = document.getElementById("connection"); +const events = new EventSource("dashboard/events"); +events.addEventListener("open", () => { + connection.textContent = "live"; connection.className = "live"; +}); +events.addEventListener("error", () => { + // EventSource reconnects by itself; the node sends a snapshot first again. + connection.textContent = "reconnecting…"; connection.className = "down"; +}); +events.addEventListener("snapshot", event => { + state = JSON.parse(event.data); stateAt = Date.now(); render(); +}); +events.addEventListener("patch", event => { + if (!state) return; + state = mergePatch(state, JSON.parse(event.data)); stateAt = Date.now(); render(); +}); +// Ages count on between events; nothing else changes without one. +setInterval(render, 1000); +window.addEventListener("resize", () => { lastShape = ""; render(); }); +</script> +</body> +</html> diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs new file mode 100644 index 0000000..adaed10 --- /dev/null +++ b/src/api/dashboard/mod.rs @@ -0,0 +1,394 @@ +//! The dashboard (`--features dashboard`): a page showing the node's figures +//! (`/admin/stats`), its peers (`/peers`) and a map of the swarm, kept up to date +//! through Server-Sent Events. +//! +//! One producer per node serves every client: while at least one is connected (or +//! asked for the state within `LINGER`), it builds the snapshot every `TICK` and, if +//! something changed, publishes it, and a JSON merge patch (RFC 7386) from the one +//! before to every client connected. Nothing changed, nothing is sent (but SSE's +//! keep-alive comments). Ages (`…SecsAgo`, `uptimeSecs` and the like, `GROWING`) grow by +//! themselves: they alone changing is no change, and the page counts them on. +//! +//! The map: the graph the member list makes (`topology::intended_graph`, as this node +//! sees it), and the links every member measures, asked for with LINKS every +//! `LINKS_EVERY` while anyone looks (`swarm::links`), the same answers for all clients. + +use std::collections::{BTreeMap, BTreeSet}; +use std::convert::Infallible; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use axum::http::{HeaderValue, header::CONTENT_TYPE}; +use axum::response::sse::{Event, KeepAlive, Sse}; +use axum::response::{Html, IntoResponse, Response}; +use axum::routing::get; +use axum::{Extension, Router}; +use serde_json::{Map, Value as JsonValue, json}; +use tokio::sync::{broadcast, watch}; +use tokio::time::Instant; + +use crate::locks::MutexExt; +use crate::state::AppState; +use crate::swarm::{links, members, topology}; +use crate::transport::Link; + +/// How often the snapshot is built and compared, while anyone looks. +const TICK: Duration = Duration::from_secs(2); +/// How often members are asked for their links, and the graph recomputed. +const LINKS_EVERY: Duration = Duration::from_secs(30); +/// A member's links older than this are left out (it didn't answer since). +const LINKS_KEPT_FOR: Duration = Duration::from_secs(90); +/// The producer runs on this long after the last request for the state. +const LINGER: Duration = Duration::from_secs(60); +/// Members asked for their links at most, each round. +const MAX_ASKED: usize = 512; + +/// A snapshot or a patch as published: its number, and its JSON. +struct Published { + seq: u64, + json: String, +} + +/// The producer and what it published, shared by all clients. +struct Hub { + state: AppState, + /// The latest snapshot; None while the producer hasn't made one (since it started). + latest: watch::Sender<Option<Arc<Published>>>, + /// Each patch, to the clients connected. + patches: broadcast::Sender<Arc<Published>>, + control: Mutex<Control>, +} + +struct Control { + running: bool, + /// The producer stops once no client is connected and this is past. + wanted_until: Instant, +} + +impl Hub { + fn new(state: AppState) -> Arc<Self> { + let (latest, _) = watch::channel(None); + let (patches, _) = broadcast::channel(64); + Arc::new(Self { + state, + latest, + patches, + control: Mutex::new(Control { + running: false, + wanted_until: Instant::now(), + }), + }) + } + + /// Someone looks: the producer runs (started if it wasn't), for `LINGER` at least. + fn want(self: &Arc<Self>) { + let mut control = self.control.lock_or_recover(); + control.wanted_until = Instant::now() + LINGER; + if !control.running { + control.running = true; + // What it published before it stopped is stale. + self.latest.send_replace(None); + tokio::spawn(Arc::clone(self).run()); + } + } + + /// The latest snapshot, once there is one. + async fn current(&self) -> Arc<Published> { + let mut latest = self.latest.subscribe(); + loop { + // The borrow (a lock guard, not Send) ends here, before any later await. + let published = latest + .wait_for(Option::is_some) + .await + .ok() + .and_then(|current| current.clone()); + if let Some(published) = published { + return published; + } + // The sender lives as long as the hub: not reached. + tokio::time::sleep(TICK).await; + } + } + + async fn run(self: Arc<Self>) { + let mut last: Option<JsonValue> = None; + let mut seq = 0u64; + let mut graph = Graph::default(); + let mut graph_due = Instant::now(); + loop { + if Instant::now() >= graph_due { + graph.refresh(&self.state).await; + graph_due = Instant::now() + LINKS_EVERY; + } + let snapshot = build(&self.state, &graph).await; + let changed = last + .as_ref() + .is_none_or(|old| without_ages(old) != without_ages(&snapshot)); + if changed { + seq += 1; + if let Some(old) = &last + && let Some(patch) = merge_patch(old, &snapshot) + { + let _ = self.patches.send(Arc::new(Published { + seq, + json: patch.to_string(), + })); + } + self.latest.send_replace(Some(Arc::new(Published { + seq, + json: snapshot.to_string(), + }))); + last = Some(snapshot); + } + tokio::time::sleep(TICK).await; + { + let mut control = self.control.lock_or_recover(); + if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until { + control.running = false; + return; + } + } + } + } +} + +/// The intended graph, recomputed every `LINKS_EVERY` (it costs about `members²`), when +/// members are asked for their links too. +#[derive(Default)] +struct Graph { + intended: BTreeMap<String, BTreeSet<String>>, +} + +impl Graph { + async fn refresh(&mut self, state: &AppState) { + self.intended = topology::intended_graph(state).unwrap_or_default(); + let members: Vec<String> = { + let replica = state.doc.read().await; + members::members(&replica.doc) + }; + for member in members + .into_iter() + .filter(|member| member != state.local_id()) + .take(MAX_ASKED) + { + let state = state.clone(); + tokio::spawn(async move { links::ask(&state, &member).await }); + } + } +} + +/// The snapshot: the node, its figures, its peers, and the map. +async fn build(state: &AppState, graph: &Graph) -> JsonValue { + let stats = serde_json::to_value(super::stats::collect(state).await).unwrap_or_default(); + let peers = serde_json::to_value(super::peers::peer_list(state).await).unwrap_or_default(); + let aliases = { + let replica = state.doc.read().await; + members::aliases(&replica.doc) + }; + let local = state.local_id(); + let intended: Map<String, JsonValue> = graph + .intended + .iter() + .map(|(member, neighbours)| (member.clone(), json!(neighbours))) + .collect(); + let mut measured = Map::new(); + measured.insert( + local.to_owned(), + json!({ "secsAgo": 0, "links": links_json(&links::own_links(state).await) }), + ); + for (member, age, member_links) in state.links.replies() { + if age <= LINKS_KEPT_FOR { + measured.insert( + member, + json!({ "secsAgo": age.as_secs(), "links": links_json(&member_links) }), + ); + } + } + // Everyone on the map: in the member list's graph, or measured. + let mut members_json = Map::new(); + for member in graph.intended.keys().chain(measured.keys()) { + members_json.insert( + member.clone(), + json!({ "alias": aliases.get(member), "self": member == local }), + ); + } + json!({ + "node": { "id": local, "alias": state.alias }, + "stats": stats, + "peers": peers, + "graph": { "members": members_json, "intended": intended, "measured": measured }, + }) +} + +fn links_json(links: &[Link]) -> JsonValue { + links + .iter() + .map(|link| { + json!({ + "peer": link.peer, + "transport": link.transport, + "relayed": link.relayed, + "rttMs": link.rtt_us.map(|us| f64::from(us) / 1000.0), + "dormant": link.dormant, + }) + }) + .collect() +} + +/// Fields that grow by themselves besides those named `…SecsAgo` (the page knows +/// them too). +const GROWING: &[&str] = &[ + "uptimeSecs", + "dormantForSecs", + "observedSecs", + "lastFromNeighbourSecs", + "oldestQuarantineSecs", +]; + +/// Whether the field `key` is an age, which grows by itself. +fn is_age(key: &str) -> bool { + key == "secsAgo" || key.ends_with("SecsAgo") || GROWING.contains(&key) +} + +/// `value` without its ages, to tell whether anything else changed. +fn without_ages(value: &JsonValue) -> JsonValue { + match value { + JsonValue::Object(object) => JsonValue::Object( + object + .iter() + .filter(|(key, _)| !is_age(key)) + .map(|(key, value)| (key.clone(), without_ages(value))) + .collect(), + ), + JsonValue::Array(items) => JsonValue::Array(items.iter().map(without_ages).collect()), + other => other.clone(), + } +} + +/// The JSON merge patch (RFC 7386) from `old` to `new`, if they differ: objects field +/// by field (a field gone is null), anything else replaced whole. A null in `new` reads +/// as the field gone, as a merge patch can't tell them apart. +fn merge_patch(old: &JsonValue, new: &JsonValue) -> Option<JsonValue> { + match (old, new) { + (JsonValue::Object(old), JsonValue::Object(new)) => { + let mut patch = Map::new(); + for key in old.keys() { + if !new.contains_key(key) { + patch.insert(key.clone(), JsonValue::Null); + } + } + for (key, value) in new { + match old.get(key) { + Some(previous) => { + if let Some(change) = merge_patch(previous, value) { + patch.insert(key.clone(), change); + } + } + None => { + patch.insert(key.clone(), value.clone()); + } + } + } + (!patch.is_empty()).then_some(JsonValue::Object(patch)) + } + (old, new) if old == new => None, + (_, new) => Some(new.clone()), + } +} + +/// The dashboard's routes (authenticated, as the rest of `/admin`), with their hub. +pub(crate) fn router(state: AppState) -> Router<AppState> { + Router::new() + .route("/admin/dashboard", get(page)) + .route("/admin/dashboard/state", get(current_state)) + .route("/admin/dashboard/events", get(events)) + .layer(Extension(Hub::new(state))) +} + +/// `GET /admin/dashboard`: the page. +async fn page() -> Html<&'static str> { + Html(include_str!("dashboard.html")) +} + +/// `GET /admin/dashboard/state`: the latest snapshot, for a client that wants it whole. +async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response { + hub.want(); + let current = hub.current().await; + ( + [(CONTENT_TYPE, HeaderValue::from_static("application/json"))], + current.json.clone(), + ) + .into_response() +} + +/// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a +/// `patch` whenever something changed (each with its number as the event's ID; a +/// client that fell behind gets a `snapshot` again). +async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse { + // Subscribed before the snapshot is read: no patch after it is missed. + let mut patches = hub.patches.subscribe(); + hub.want(); + let first = hub.current().await; + let stream = async_stream::stream! { + let mut seen = first.seq; + yield Ok::<_, Infallible>(snapshot_event(&first)); + loop { + match patches.recv().await { + Ok(patch) if patch.seq <= seen => {} + Ok(patch) => { + seen = patch.seq; + yield Ok(Event::default() + .event("patch") + .id(patch.seq.to_string()) + .data(patch.json.clone())); + } + Err(broadcast::error::RecvError::Lagged(_)) => { + let current = hub.current().await; + seen = current.seq; + yield Ok(snapshot_event(¤t)); + } + Err(broadcast::error::RecvError::Closed) => break, + } + } + }; + Sse::new(stream).keep_alive(KeepAlive::default()) +} + +fn snapshot_event(published: &Published) -> Event { + Event::default() + .event("snapshot") + .id(published.seq.to_string()) + .data(published.json.clone()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn merge_patches() { + let old = json!({ "a": 1, "b": { "c": 2, "d": [1, 2] }, "gone": true }); + let new = json!({ "a": 1, "b": { "c": 3, "d": [1, 2] }, "added": "x" }); + assert_eq!( + merge_patch(&old, &new), + Some(json!({ "b": { "c": 3 }, "gone": null, "added": "x" })) + ); + assert_eq!(merge_patch(&old, &old), None); + assert_eq!( + merge_patch(&json!([1]), &json!([1, 2])), + Some(json!([1, 2])) + ); + } + + #[test] + fn ages_alone_are_no_change() { + let old = + json!({ "uptimeSecs": 1, "peers": { "p": { "lastSeenSecsAgo": 3, "rttMs": 2 } } }); + let older = + json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 2 } } }); + assert_eq!(without_ages(&old), without_ages(&older)); + let moved = + json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 5 } } }); + assert_ne!(without_ages(&old), without_ages(&moved)); + } +} diff --git a/src/api/mod.rs b/src/api/mod.rs index 2cb0415..fd05669 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,6 +1,8 @@ //! HTTP API: router, the API description and shared response helpers. pub(crate) mod auth; +#[cfg(feature = "dashboard")] +mod dashboard; mod debug; mod document; mod health; @@ -55,6 +57,11 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> .route("/admin/config", get(get_config)) .merge(crate::remake::router::<AppState>()); + #[cfg(feature = "dashboard")] + { + api = api.merge(dashboard::router(state.clone())); + } + if debug_api { warn!("Debug API enabled"); api = api.route( |
