From 8391a6df59d566ffd7de01560c262d1afdd8d43d Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 7 Nov 2026 22:35:09 +0100 Subject: dashboard: subscribed nodes' messages in the console Each tick's messages of a subscribed node, listed in the console with this node's own, from its own stream. Auth plan M3 (live data from every reachable node). --- docs/access.md | 4 +- docs/dashboard.md | 14 +++- docs/openapi.yaml | 12 +++- src/api/cors.rs | 6 +- src/api/dashboard/dashboard.html | 147 +++++++++++++++++++++++++++------------ src/api/dashboard/mod.rs | 62 ++++++++++------- src/transport/traffic.rs | 28 ++++++-- 7 files changed, 184 insertions(+), 89 deletions(-) diff --git a/docs/access.md b/docs/access.md index e8f9893..961fc72 100644 --- a/docs/access.md +++ b/docs/access.md @@ -106,8 +106,8 @@ curl -H "Authorization: Bearer ct1.…" http://other-node:3000/store `GET /auth/whoami` says who a request's user is, how it proved it, and what it may do. Pages of other origins, the dashboards of other members, may call a node with a token: -its own links (`/admin/dashboard/links/own`, and its events) and traffic -(`/admin/dashboard/traffic/own/events`), `/auth/refresh` and `/auth/whoami`, nothing +its own links (`/admin/dashboard/links/own`, and its events) and messages +(`/admin/dashboard/messages/own/events`), `/auth/refresh` and `/auth/whoami`, nothing else. A node allows the origins of the dashboards its member list names (`--dashboard-url`) and those of `--cors-origins`; without credentials (no cookie or password crosses origins: the page sends its token). diff --git a/docs/dashboard.md b/docs/dashboard.md index 45a2248..c063c78 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -134,9 +134,10 @@ right after their legend makes all dots larger or smaller, and the `speed` slide it makes them run faster or slower (0.1 to 0.9 map units per millisecond, 0.3 at first: a link of 200 in about 0.7 s); both are kept in the browser. A picture of the traffic, not a record of it. A subscribed node's messages with its peers run -along its links too, from its own counts (read from it with the page's token, as its -links are): those with this node are drawn from this node's counts, and between two -subscribed nodes from what the sender sent. Links between other members show none. +along its links too, from its messages of each 2 s (read from it with the page's token, +as its links are): those with this node are drawn from this node's counts, and between +two subscribed nodes from what the sender sent, so each message once. Links between +other members show none. `messages: on` after the sliders hides them (`messages: off` shows them again; the legend and the sliders greyed meanwhile); the choice is kept in the browser. @@ -322,6 +323,13 @@ PINGs with, are listed too: under the node at the address they went to, where it known, else under the address. Not listed: WHO and ME, which stay in the layers below the node (they are counted, [`measurements.md`](measurements.md), Traffic). +A subscribed node's messages with its peers are listed too, as they come from it (every +2 s): named by both ends, the receiver first (`b ← a`), each once, as on the map (those +with this node only from this node, between two subscribed nodes from the sender); its +tooltip names the node that logged it. Shown in the figures box, it says when, both +ends, its type and size: the message itself is written out only on the node that +logged it. + A click on a message shows it in the figures box: when, with whom, its type and size, and the message written out (its bytes as sizes). A click on it again, or on anything on the map or in the peers, goes back. diff --git a/docs/openapi.yaml b/docs/openapi.yaml index a5f7dcb..20cbf80 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1050,11 +1050,17 @@ paths: '401': $ref: '#/components/responses/Unauthorized' - /admin/dashboard/traffic/own/events: + /admin/dashboard/messages/own/events: get: tags: [Admin] - summary: This node's traffic with each peer, every 2 s - description: Server-sent events, `traffic` with what this node sent to and received from each peer by node ID, by kind (messages and bytes counted since it started, as `/peers` has them), every 2 s; `offline` when the node stops. For a page of another member's dashboard subscribed to this node, to draw its messages on the map (CORS, with a token). + summary: This node's messages, every 2 s + description: | + Server-sent events, `messages` with the messages this node sent and received in + the last 2 s (none too), each its number, time, way, peer, type, kind and size, + not written out; `offline` when the node stops. For a page of another member's + dashboard subscribed to this node: its console lists them, its map runs them + along this node's links (CORS, with a token). The node keeps its message log + while one is open. responses: '200': description: The events. diff --git a/src/api/cors.rs b/src/api/cors.rs index 2d5fdc3..406b7be 100644 --- a/src/api/cors.rs +++ b/src/api/cors.rs @@ -1,6 +1,6 @@ //! Pages of other origins reading this node: the dashboards of other members, which a -//! page subscribes to for their own links and traffic (`/admin/dashboard/links/own`, -//! `/admin/dashboard/traffic/own/events`), refreshing its +//! page subscribes to for their own links and messages (`/admin/dashboard/links/own`, +//! `/admin/dashboard/messages/own/events`), refreshing its //! token on any node (`/auth/refresh`, `/auth/whoami`). Only those endpoints, only for //! the origins of the dashboards the member list names and those of `--cors-origins`, //! and without credentials: a page sends its token itself (`Authorization: Bearer`), @@ -22,7 +22,7 @@ fn open(path: &str) -> bool { path, "/admin/dashboard/links/own" | "/admin/dashboard/links/own/events" - | "/admin/dashboard/traffic/own/events" + | "/admin/dashboard/messages/own/events" | "/auth/refresh" | "/auth/whoami" ) diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index efbf892..3f02a64 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -774,11 +774,25 @@ function renderStats() { // Drawn once: a message doesn't change, and drawing it anew would scroll it back up. if (stats.dataset.message === String(message.seq)) return; stats.dataset.message = String(message.seq); - what.textContent = message.type + " " + WAYS[message.way] + " " + name(message.peer); + what.textContent = message.node + ? message.type + " " + name(message.to) + " ← " + name(message.from) + : message.type + " " + WAYS[message.way] + " " + name(message.peer); const field = (key, value) => el("div", { class: "leaf" }, el("span", { class: "key" }, key + ": "), value); title.replaceChildren("Message: ", what); retitle(title, title.textContent); + // A subscribed node's: logged there, the message itself not read from it. + if (message.node) { + document.getElementById("stats").replaceChildren( + field("at", clockTime(new Date(message.at))), + field("from", message.from), + field("to", message.to), + field("logged by", message.node), + field("type", message.type), + field("bytes", String(message.bytes)), + el("div", { class: "note" }, "Written out only on the node that logged it (its console).")); + return; + } document.getElementById("stats").replaceChildren( field("at", clockTime(new Date(message.at))), field(message.way === "received" ? "from" : "to", message.peer), @@ -1224,7 +1238,8 @@ function renderGraph() { const marking = new Map(); for (const [message, how] of [[pointedMessage, "pointed"], [shownMessage, "shown"]]) { if (message === null) continue; - for (const key of [state.node.id, message.peer, pairKey(state.node.id, message.peer)]) { + const [a, b] = message.node ? [message.from, message.to] : [state.node.id, message.peer]; + for (const key of [a, b, pairKey(a, b)]) { marking.set(key, how); } } @@ -2292,9 +2307,12 @@ function follow(id, delay = 2000) { const member = state && state.graph.members[id]; if (!member || !member.dashboard) return fail("it publishes no dashboard (--dashboard-url)"); if (!narrowToken) return fail(tokenWhy); + // Its endpoints, beside its page (`…/admin/dashboard` → `…/admin/dashboard/…`). + const base = member.dashboard.split("#")[0].replace(/\/+$/, ""); + const at = path => new URL(path, base).href; let url; try { - url = new URL("dashboard/links/own/events", member.dashboard.split("#")[0].replace(/\/+$/, "")).href; + url = at("dashboard/links/own/events"); } catch (_) { return fail("its dashboard's address isn't one"); } @@ -2310,25 +2328,19 @@ function follow(id, delay = 2000) { render(); } }).catch(error => fail(error.message || String(error))); - // Its traffic, for its messages on the map: while it goes, quietly; a failure of its - // own leaves the links be. - let trafficUrl; - try { - trafficUrl = new URL("dashboard/traffic/own/events", url).href; - } catch (_) { - return; - } - const followTraffic = (wait = 2000) => { + // Its messages, for the console and the map: while they come, quietly; a failure of + // their own leaves the links be. + const messagesUrl = at("dashboard/messages/own/events"); + const followMessages = (wait = 2000) => { if (controller.signal.aborted) return; - read(trafficUrl, controller.signal, () => { entry.trafficLive = true; wait = 2000; }, - (kind, data) => { if (kind === "traffic") remoteTrafficArrived(id, entry, JSON.parse(data)); }) + read(messagesUrl, controller.signal, () => { entry.messagesLive = true; wait = 2000; }, + (kind, data) => { if (kind === "messages") remoteMessagesArrived(id, entry, JSON.parse(data)); }) .catch(() => { - entry.trafficLive = false; - entry.traffic = null; - if (!controller.signal.aborted) setTimeout(() => followTraffic(Math.min(60000, wait * 2)), wait); + entry.messagesLive = false; + if (!controller.signal.aborted) setTimeout(() => followMessages(Math.min(60000, wait * 2)), wait); }); }; - followTraffic(); + followMessages(); } // Server-sent events from `url`, read with this page's narrow token (`EventSource` // can't send one): `opened` once answered, `event(kind, data)` for each; rejects when it @@ -2644,8 +2656,9 @@ graph.addEventListener("mousemove", event => { if (event.target === graph) hover graph.addEventListener("mouseleave", () => hover(null)); // --- Messages on the map: this node's traffic with each peer (`/peers`, `traffic`), -// and each subscribed node's with its peers (`remoteTrafficArrived`), as it grew since -// the last update, drawn as dots running along the link the way the messages went. Counts come with each update (every 2 s), so the dots are spread +// as it grew since the last update, and each subscribed node's messages with its peers +// (`remoteMessagesArrived`), drawn as dots running along the link the way the messages +// went. Counts come with each update (every 2 s), so the dots are spread // over the next 2 s, a few at most per link and way: a picture of the traffic, not a // record of it. All at one speed (`flowSpeed`), so over a longer link they arrive // later. Kept above everything else on the map (`flows`, after what `show` places). --- @@ -2839,24 +2852,49 @@ function dotsNow(gap) { !calm.matches && !document.hidden && !document.getElementById("map").classList.contains("collapsed"); } -// A subscribed node's traffic (`followTraffic`), as `trafficArrived` this node's: its -// messages with this node are this node's to draw, and what it received from another -// subscribed node that one's (as what it sent). -function remoteTrafficArrived(id, entry, traffic) { +// A subscribed node's messages (`follow`), those of a tick: listed in the console and +// run along its links, as this node's own. Its messages with this node are this node's +// to show, and what it received from another subscribed node that one's (as what it +// sent): each message once. Each named by both ends (`from`, `to`) and the node that +// logged it (`node`), its number and exchange made its own (`node:seq`). +function remoteMessagesArrived(id, entry, list) { if (!state) return; - const current = new Map(); - for (const [peer, counts] of Object.entries(traffic)) { - current.set(peer, { sent: messages(counts.sent), received: messages(counts.received) }); - } - const wall = Date.now(), gap = wall - (entry.trafficAt || 0); - entry.trafficAt = wall; - if (entry.traffic && dotsNow(gap)) { - const self = state.node.id; - addFlows(id, current, entry.traffic, (way, peer) => peer === self || - (way === "received" && peer !== id && subscriptions.has(peer) && - (subscribed.get(peer) || {}).trafficLive)); + const self = state.node.id; + const wall = Date.now(), gap = wall - (entry.messagesAt || 0); + entry.messagesAt = wall; + const kept = []; + for (const message of list) { + if (message.way === "refused" || message.peer === self) continue; + if (message.way === "received" && message.peer !== id && subscriptions.has(message.peer) && + (subscribed.get(message.peer) || {}).messagesLive) continue; + const [from, to] = message.way === "sent" ? [id, message.peer] : [message.peer, id]; + kept.push({ ...message, seq: id + ":" + message.seq, node: id, from, to, + related: message.related ? id + ":" + message.related : null }); + } + if (kept.length === 0) return; + if (dotsNow(gap)) { + const now = performance.now(); + const links = new Map(); + for (const message of kept) { + if (!linksShown && isLinks(message.counted)) continue; + const key = message.from + " " + message.to; + const link = links.get(key) || { from: message.from, to: message.to, kinds: {}, total: 0 }; + const kind = link.kinds[message.counted] || (link.kinds[message.counted] = { messages: 0, bytes: 0 }); + kind.messages += 1; + kind.bytes += message.bytes; + link.total += 1; + links.set(key, link); + } + for (const { from, to, kinds, total } of links.values()) { + const shown = dotsFor(kinds, total); + shown.forEach((dot, i) => { + flows.push({ from, to, start: now + i * FLOW_SPREAD_MS / shown.length, + kind: dot.kind, r: dotRadius(dot.bytes) }); + }); + } + if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows); } - entry.traffic = current; + messagesArrived(kept); } // --- The console's rates: messages and bytes per second sent and received, per peer @@ -2977,15 +3015,27 @@ function peerOf(key) { return a === self ? b : b === self ? a : null; } // A console line's tint for the node or link pointed at on the map, or in focus. -function peerClasses(peer) { - return (peer === peerOf(focused) ? " peer-focused" : "") + - (peer === peerOf(hovered) ? " peer-hovered" : ""); +// Whether a message (its line's `data-…`: `peer` for this node's, `from` and `to` for a +// subscribed node's) is one with the node or over the link `key` stands for. +function about(data, key) { + if (key === null || !state) return false; + if (!data.peer) { + if (!data.from) return false; + return isPair(key) ? key === pairKey(data.from, data.to) + : key !== state.node.id && (data.from === key || data.to === key); + } + const peer = peerOf(key); + return peer !== null && data.peer === peer; +} +function peerClasses(message) { + const data = { peer: message.node ? "" : message.peer, from: message.from, to: message.to }; + return (about(data, focused) ? " peer-focused" : "") + + (about(data, hovered) ? " peer-hovered" : ""); } function markPeerMessages() { - const focusedPeer = peerOf(focused), hoveredPeer = peerOf(hovered); for (const line of document.getElementById("console").children) { - line.classList.toggle("peer-focused", focusedPeer !== null && line.dataset.peer === focusedPeer); - line.classList.toggle("peer-hovered", hoveredPeer !== null && line.dataset.peer === hoveredPeer); + line.classList.toggle("peer-focused", about(line.dataset, focused)); + line.classList.toggle("peer-hovered", about(line.dataset, hovered)); } filterConsole(); } @@ -3023,16 +3073,21 @@ function filterConsole() { } // A line: when, which way, with whom, the type (in its kind's colour) and the size. function consoleLine(message) { + // A subscribed node's (`remoteMessagesArrived`): its receiver, and whom from. + const between = message.node ? name(message.to) + " ← " + name(message.from) + " " + : WAYS[message.way] + " " + name(message.peer) + " "; const line = el("div", { "data-seq": String(message.seq), "data-related": message.related || "", - "data-peer": message.peer, + "data-peer": message.node ? "" : message.peer, + "data-from": message.from || "", "data-to": message.to || "", class: (shownMessage && shownMessage.seq === message.seq ? "focused" : "") + (isLinks(message.counted) ? " links-kind" : "") + (relatedMarked !== null && message.related === relatedMarked ? " related" : "") + - peerClasses(message.peer), - title: message.way === "refused" ? "refused: the transport wouldn't send it" : "" }, + peerClasses(message), + title: message.way === "refused" ? "refused: the transport wouldn't send it" + : message.node ? "logged by " + name(message.node) + ", subscribed to" : "" }, el("span", { class: "when" }, clockTime(new Date(message.at)) + " "), - WAYS[message.way] + " " + name(message.peer) + " ", + between, el("span", { style: "color: " + kindColour(message.counted) }, message.type), " " + size(message.bytes)); // On press, as the peers' rows: lines come and go under the pointer. diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index 1d7dbf7..7d940c9 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -540,8 +540,8 @@ pub(crate) fn router(state: AppState) -> Router { .route("/admin/dashboard/links/own", get(own_links_now)) .route("/admin/dashboard/links/own/events", get(own_links_events)) .route( - "/admin/dashboard/traffic/own/events", - get(own_traffic_events), + "/admin/dashboard/messages/own/events", + get(own_messages_events), ) .layer(Extension(Hub::new(state, closing))) } @@ -598,31 +598,28 @@ async fn own_links_events(Extension(hub): Extension>) -> impl IntoRespo ) } -/// This node's traffic with each peer: by node ID, what it sent and received, by kind -/// (messages and bytes counted since it started, as `/peers` has them). -async fn own_traffic(state: &AppState) -> JsonValue { - let peers = state.peers_snapshot().await; - let traffic: Map = peers - .iter() - .map(|(peer, _)| { - let counted = state - .traffic - .node(peer, &state.transport.identities_of(peer)); - ( - peer.clone(), - json!({ "sent": counted.sent, "received": counted.received }), - ) - }) - .collect(); - JsonValue::Object(traffic) +/// Stops looking at the message log when dropped (`own_messages_events`). +struct Watching(Arc); + +impl Drop for Watching { + fn drop(&mut self) { + self.0.state.traffic.log.watch(false); + } } -/// `GET /admin/dashboard/traffic/own/events`: server-sent events, `traffic` with this -/// node's traffic with each peer (`own_traffic`) every `TICK`, for a page of another -/// member subscribed to this node to draw its messages on the map, as a page of this -/// node draws its own from `/peers`; `offline` when the node stops, which ends it. -async fn own_traffic_events(Extension(hub): Extension>) -> impl IntoResponse { +/// `GET /admin/dashboard/messages/own/events`: server-sent events, `messages` with the +/// messages this node sent and received since the last (every `TICK`, those of the +/// tick, none too, as the page's own `messages` events have them, without each message +/// written out), for a page of another member subscribed to this node: its console lists them, +/// and its map runs them along this node's links. `offline` when the node stops, which +/// ends it. The message log is kept while it is open. +async fn own_messages_events(Extension(hub): Extension>) -> impl IntoResponse { + let log = &hub.state.traffic.log; + log.watch(true); + let watching = Watching(Arc::clone(&hub)); + let mut logged = log.latest(); let stream = async_stream::stream! { + let _watching = watching; let mut every = tokio::time::interval(TICK); loop { tokio::select! { @@ -632,8 +629,21 @@ async fn own_traffic_events(Extension(hub): Extension>) -> impl IntoRes break; } } - let traffic = own_traffic(&hub.state).await; - yield Ok(Event::default().event("traffic").data(traffic.to_string())); + // Every tick, none too: the page times its dots by them. + let mut messages = hub.state.traffic.log.after(logged, MESSAGES_AT_MOST); + if let Some(last) = messages.last() { + logged = last.seq; + by_node(&hub.state, &mut messages).await; + } + let messages: Vec = messages + .into_iter() + .filter_map(|message| { + let mut json = serde_json::to_value(message).ok()?; + json.as_object_mut()?.remove("text"); + Some(json) + }) + .collect(); + yield Ok(Event::default().event("messages").data(JsonValue::Array(messages).to_string())); } }; ( diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 291a76b..8220041 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -9,7 +9,7 @@ //! sent (the firewall plays a lossy network) but not as received. use std::collections::{HashMap, VecDeque}; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, MutexGuard}; use anyhow::Result; @@ -156,7 +156,8 @@ const LOGGED_TEXT: usize = 8 * 1024; /// handed. #[derive(Debug, Default)] pub(crate) struct MessageLog { - watching: AtomicBool, + /// How many look (`watch`). + watching: AtomicUsize, inner: Mutex, } @@ -230,14 +231,29 @@ fn related(peer: &str, message: &ParsedMessage) -> Option { // Read by the dashboard only. #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] impl MessageLog { - /// Keeps messages from now on (`true`), or no more, forgetting those kept. + /// One more looks (`true`): messages are kept from now on; or one less (`false`): + /// once none looks, no more, and those kept are forgotten. pub(crate) fn watch(&self, on: bool) { - self.watching.store(on, Ordering::Relaxed); - if !on { + if on { + self.watching.fetch_add(1, Ordering::Relaxed); + return; + } + let was = self + .watching + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| { + Some(n.saturating_sub(1)) + }) + .unwrap_or(0); + if was <= 1 { self.inner.lock_or_recover().entries.clear(); } } + /// The latest message's number (`Logged::seq`): those after it are new. + pub(crate) fn latest(&self) -> u64 { + self.inner.lock_or_recover().next + } + pub(crate) fn note( &self, way: &'static str, @@ -245,7 +261,7 @@ impl MessageLog { message: &ParsedMessage, bytes: usize, ) { - if !self.watching.load(Ordering::Relaxed) { + if self.watching.load(Ordering::Relaxed) == 0 { return; } let at = std::time::SystemTime::now() -- cgit v1.3.1