From 71471de0a65bc9e8bdaad010cab0993cc75ea87d Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sat, 22 Aug 2026 21:17:42 +0400 Subject: [PATCH] fix(dashboard): improve control and hardware views Refine fleet and provisioning interactions, expand control-plane projections, and add CPU, memory, pressure, and storage telemetry for the local dashboard surfaces. --- .../src/observability/frame_collector.rs | 15 +- .../src/observability/orch_telemetry.rs | 39 ++- .../src/observability/provisioning_logs.rs | 2 +- apps/myelin/src/orchestration/control.rs | 59 ++-- .../myelin/src/orchestration/fleet_control.js | 61 +++- .../src/orchestration/provision_page.html | 39 ++- crates/dashboard/Cargo.toml | 3 + crates/dashboard/src/control.rs | 7 +- crates/dashboard/src/control_plane.rs | 208 ++++++++++++- crates/dashboard/src/control_plane_page.html | 286 +++++++++++++++--- crates/dashboard/src/hardware_view.rs | 99 +++++- crates/dashboard/src/server.rs | 2 +- crates/dashboard/src/swactor/actor_view.rs | 9 +- crates/telemetry/Cargo.toml | 5 + crates/telemetry/src/emit.rs | 3 +- crates/telemetry/src/endpoint.rs | 3 +- crates/telemetry/src/hardware/cpu.rs | 5 +- crates/telemetry/src/hardware/memory.rs | 151 +++++++++ crates/telemetry/src/hardware/mod.rs | 100 ++++++ crates/telemetry/src/hardware/pressure.rs | 95 ++++++ crates/telemetry/src/hardware/storage.rs | 123 ++++++++ crates/telemetry/src/mux.rs | 27 +- crates/telemetry/tests/t_telemetry.rs | 11 - crates/telemetry/tests/t_telemetry_realio.rs | 2 +- dash-fixes.md | 12 + 25 files changed, 1211 insertions(+), 155 deletions(-) create mode 100644 crates/telemetry/src/hardware/memory.rs create mode 100644 crates/telemetry/src/hardware/pressure.rs create mode 100644 crates/telemetry/src/hardware/storage.rs create mode 100644 dash-fixes.md diff --git a/apps/myelin/src/observability/frame_collector.rs b/apps/myelin/src/observability/frame_collector.rs index 95ef4e6..282bde0 100644 --- a/apps/myelin/src/observability/frame_collector.rs +++ b/apps/myelin/src/observability/frame_collector.rs @@ -7,6 +7,7 @@ //! only through these closures. use iroh::EndpointAddr; +use iroh_driver::telemetry_transport::PullCollectorConfig; use iroh_driver::{IrohDriver, PullCollectorHandle, TelemetryQuicHeader, spawn_pull_collector}; use parking_lot::Mutex; use std::collections::{BTreeMap, BTreeSet}; @@ -82,12 +83,14 @@ impl FrameCollector { flow_id[8..].copy_from_slice(&node_id.to_le_bytes()); let collector = spawn_pull_collector( engine, - endpoint, - peer, - flow_id, - Vec::new(), - SubscriptionRequest::all(), - Arc::clone(&self.pull_fanout), + PullCollectorConfig { + endpoint, + peer, + flow_id, + token: Vec::new(), + request: SubscriptionRequest::all(), + fanout: Arc::clone(&self.pull_fanout), + }, self.pull_header_tx.clone(), ); if let Some(previous) = self diff --git a/apps/myelin/src/observability/orch_telemetry.rs b/apps/myelin/src/observability/orch_telemetry.rs index 2a3cf91..e55a5e8 100644 --- a/apps/myelin/src/observability/orch_telemetry.rs +++ b/apps/myelin/src/observability/orch_telemetry.rs @@ -43,6 +43,15 @@ pub(crate) struct OrchTelemetry { descriptor: StreamDescriptor, } +pub(crate) struct BootstrapEmission<'a> { + pub(crate) dashboard: Option<&'a DashboardSupport>, + pub(crate) channel: &'a str, + pub(crate) run_id: u64, + pub(crate) node_id: u64, + pub(crate) phase: &'a str, + pub(crate) status: &'a str, + pub(crate) detail: Value, +} impl OrchTelemetry { pub(crate) fn new(run_id: u64, frame_log: Option<&Path>) -> Result { let stream = StreamId::new(NodeId::new("myelin-orchestrator"), Lifetime(run_id)); @@ -110,6 +119,10 @@ impl OrchTelemetry { id } + pub(crate) fn producer(&self) -> TelemetryProducer { + self.producer.clone() + } + pub(crate) fn emit_event( &mut self, dashboard: Option<&DashboardSupport>, @@ -140,27 +153,27 @@ impl OrchTelemetry { status: &str, detail: Value, ) { - self.emit_bootstrap_to_channel( + self.emit_bootstrap_to_channel(BootstrapEmission { dashboard, - MYELIN_ORCH_BOOTSTRAP, + channel: MYELIN_ORCH_BOOTSTRAP, run_id, node_id, phase, status, detail, - ); + }); } - pub(crate) fn emit_bootstrap_to_channel( - &mut self, - dashboard: Option<&DashboardSupport>, - channel: &str, - run_id: u64, - node_id: u64, - phase: &str, - status: &str, - detail: Value, - ) { + pub(crate) fn emit_bootstrap_to_channel(&mut self, emission: BootstrapEmission<'_>) { + let BootstrapEmission { + dashboard, + channel, + run_id, + node_id, + phase, + status, + detail, + } = emission; let benchmark = benchmark::stamp("myelin-orchestrator"); let payload = serde_json::to_vec(&json!({ "schema_version": benchmark["schema_version"].clone(), diff --git a/apps/myelin/src/observability/provisioning_logs.rs b/apps/myelin/src/observability/provisioning_logs.rs index 490cd7b..e42d1b0 100644 --- a/apps/myelin/src/observability/provisioning_logs.rs +++ b/apps/myelin/src/observability/provisioning_logs.rs @@ -10,7 +10,7 @@ use crate::provisioning::{ }; pub(crate) fn node_stream_id(run_id: u64, node_id: u64) -> StreamId { - StreamId::new(NodeId::new(&node_id.to_string()), Lifetime(run_id)) + StreamId::new(NodeId::new(node_id.to_string()), Lifetime(run_id)) } #[derive(Clone)] diff --git a/apps/myelin/src/orchestration/control.rs b/apps/myelin/src/orchestration/control.rs index b85a8e6..68ccfb7 100644 --- a/apps/myelin/src/orchestration/control.rs +++ b/apps/myelin/src/orchestration/control.rs @@ -538,7 +538,7 @@ async fn submit_node_job( ManualControlMsg::Query { reply_to } }) { Ok(response_rx) => response_rx, - Err(response) => return response, + Err(response) => return *response, }; let model = match response_rx.await { Ok(ManualControlReply::Status(model)) => model, @@ -689,7 +689,7 @@ async fn request_reply( ) -> Response { let response_rx = match begin_request_reply(state, timeout, build) { Ok(response_rx) => response_rx, - Err(response) => return response, + Err(response) => return *response, }; match response_rx.await { @@ -718,7 +718,7 @@ fn begin_request_reply( state: &ControlHttpState, timeout: Duration, build: impl FnOnce(ActorAddress) -> ManualControlMsg, -) -> Result, Response> { +) -> Result, Box> { let (response_tx, response_rx) = tokio::sync::oneshot::channel(); let response_tx = Arc::new(Mutex::new(Some(response_tx))); let reply_to = state @@ -730,26 +730,30 @@ fn begin_request_reply( timeout, }) .map_err(|error| { - ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ErrorResponse { - error: format!("create control reply observer: {error}"), - }), + Box::new( + ( + StatusCode::SERVICE_UNAVAILABLE, + Json(ErrorResponse { + error: format!("create control reply observer: {error}"), + }), + ) + .into_response(), ) - .into_response() })?; if let Err(error) = state .runtime .send_to(state.orchestrator, OrchestratorMsg::Manual(build(reply_to))) { let _ = state.runtime.stop_actor(reply_to); - return Err(( - StatusCode::SERVICE_UNAVAILABLE, - Json(ErrorResponse { - error: format!("orchestrator control actor unavailable: {error}"), - }), - ) - .into_response()); + return Err(Box::new( + ( + StatusCode::SERVICE_UNAVAILABLE, + Json(ErrorResponse { + error: format!("orchestrator control actor unavailable: {error}"), + }), + ) + .into_response(), + )); } Ok(response_rx) } @@ -998,7 +1002,7 @@ mod properties { status, }) }; - let reply = |result: Result<_, Response>| match result { + let reply = |result: Result<_, Box>| match result { Ok(receiver) => PendingHttpObservation::Reply { index, action: action.clone(), @@ -1191,9 +1195,10 @@ mod properties { HttpAction::from_raw(kind, command_slot, value) }) .collect::>(); - let mut config = RuntimeConfig::default(); - config.worker_count = 1; - let parts = RuntimeParts::new(config); + let parts = RuntimeParts::new(RuntimeConfig { + worker_count: 1, + ..RuntimeConfig::default() + }); let runtime = parts.runtime().clone(); let backend = SteppingBackend::new(); let engine = @@ -1294,9 +1299,10 @@ mod properties { fn generated_duplicate_control_replies_deliver_first_once_and_remove_observer( replies in prop::collection::vec(any::(), 0..=16) ) { - let mut config = RuntimeConfig::default(); - config.worker_count = 1; - let parts = RuntimeParts::new(config); + let parts = RuntimeParts::new(RuntimeConfig { + worker_count: 1, + ..RuntimeConfig::default() + }); let runtime = parts.runtime().clone(); let backend = SteppingBackend::new(); let engine = @@ -1380,9 +1386,10 @@ mod properties { #[test] fn reply_observer_disappearance_returns_a_bounded_terminal_http_response() { - let mut config = RuntimeConfig::default(); - config.worker_count = 1; - let parts = RuntimeParts::new(config); + let parts = RuntimeParts::new(RuntimeConfig { + worker_count: 1, + ..RuntimeConfig::default() + }); let runtime = parts.runtime().clone(); let backend = SteppingBackend::new(); let engine = diff --git a/apps/myelin/src/orchestration/fleet_control.js b/apps/myelin/src/orchestration/fleet_control.js index 27bc478..3eaece4 100644 --- a/apps/myelin/src/orchestration/fleet_control.js +++ b/apps/myelin/src/orchestration/fleet_control.js @@ -1,14 +1,50 @@ (() => { const CONTROL_ID = 'myelin-fleet-control'; + const CONFIRM_ID = 'myelin-confirm-dialog'; + const CONFIRM_STYLE_ID = 'myelin-confirm-dialog-style'; const selectedJobs = new Map(); - async function syncControl() { - const nodeView = document.querySelector('.node-view[data-node]'); - if (!nodeView) return; - const rawNodeId = nodeView.getAttribute('data-node') || ''; - if (!/^\d+$/.test(rawNodeId)) return; - const logicalNodeId = Number(rawNodeId); + function confirmKill(logicalNodeId) { + let dialog = document.getElementById(CONFIRM_ID); + if (!dialog) { + const style = document.createElement('style'); + style.id = CONFIRM_STYLE_ID; + style.textContent = ` + .myelin-confirm { width:min(440px,calc(100vw - 32px));padding:0;color:var(--text);background:var(--panel);border:1px solid var(--bad);border-radius:var(--r);box-shadow:0 18px 60px rgba(0,0,0,.55) } + .myelin-confirm::backdrop { background:rgba(0,6,12,.78) } + .myelin-confirm form { display:grid;gap:14px;padding:18px } + .myelin-confirm h2,.myelin-confirm p { margin:0 } + .myelin-confirm h2 { color:var(--bad) } + .myelin-confirm-actions { display:flex;justify-content:flex-end;gap:8px } + .myelin-confirm button { padding:6px 12px;background:transparent;color:var(--text);border:1px solid var(--border);border-radius:var(--r);cursor:pointer;font:600 13px var(--mono) } + .myelin-confirm button[value="confirm"] { color:var(--danger-ink);background:var(--danger-fill);border-color:var(--danger-border) } + `; + document.head.append(style); + dialog = document.createElement('dialog'); + dialog.id = CONFIRM_ID; + dialog.className = 'myelin-confirm'; + dialog.setAttribute('aria-labelledby', 'myelin-confirm-title'); + dialog.setAttribute('aria-describedby', 'myelin-confirm-message'); + dialog.innerHTML = `
+

Terminate managed node?

+

+
+ + +
+
`; + document.body.append(dialog); + } + dialog.querySelector('#myelin-confirm-message').textContent = + `Terminate managed node ${logicalNodeId}? Vast.ai contracts are destroyed and billing stops.`; + dialog.returnValue = 'cancel'; + return new Promise(resolve => { + dialog.addEventListener('close', () => resolve(dialog.returnValue === 'confirm'), { once: true }); + dialog.showModal(); + }); + } + async function syncControl() { let model; try { const response = await fetch('/api/control/status', { cache: 'no-store' }); @@ -18,6 +54,17 @@ } catch (_) { return; } + window.dispatchEvent(new CustomEvent('dashboard-hardware-source', { + detail: { + source: model?.provider?.provisioning_mode === 'mock' ? 'orchestrator' : 'node', + }, + })); + + const nodeView = document.querySelector('.node-view[data-node]'); + if (!nodeView) return; + const rawNodeId = nodeView.getAttribute('data-node') || ''; + if (!/^\d+$/.test(rawNodeId)) return; + const logicalNodeId = Number(rawNodeId); const node = model?.nodes?.find(candidate => candidate.logical_node_id === logicalNodeId); if (!node) return; @@ -130,7 +177,7 @@ }; killButton.onclick = async () => { - if (!window.confirm(`Kill managed node ${logicalNodeId}? Vast.ai contracts are destroyed and billing stops.`)) return; + if (!await confirmKill(logicalNodeId)) return; killButton.disabled = true; message.textContent = 'Submitting kill…'; const commandId = `fleet-kill-${globalThis.crypto?.randomUUID?.() || Date.now()}`; diff --git a/apps/myelin/src/orchestration/provision_page.html b/apps/myelin/src/orchestration/provision_page.html index d6ed5a5..79e71dc 100644 --- a/apps/myelin/src/orchestration/provision_page.html +++ b/apps/myelin/src/orchestration/provision_page.html @@ -92,6 +92,19 @@ .sort-button[data-direction="asc"]::before { top: 3px; border-bottom: 5px solid currentColor; } .sort-button[data-direction="desc"]::before { bottom: 3px; border-top: 5px solid currentColor; } .sort-button:hover:not(:disabled), .sort-button[aria-pressed="true"] { color: var(--cyan); border-color: var(--cyan); background: var(--panel-hover); } + .offers-heading { display: flex; align-items: center; justify-content: space-between; gap: 12px; flex-wrap: wrap; margin-bottom: 12px; } + .offers-heading h2 { margin: 0; } + .offer-actions { display: flex; align-items: center; gap: 10px; flex-wrap: wrap; } + .offer-selection { margin: 0; font: 12px var(--mono); } + dialog.confirm-dialog { + width: min(440px, calc(100vw - 32px)); padding: 0; color: var(--text); + background: var(--panel); border: 1px solid var(--amber); border-radius: var(--r); + box-shadow: 0 18px 60px rgba(0, 0, 0, .55); + } + dialog.confirm-dialog::backdrop { background: rgba(0, 6, 12, .78); } + .confirm-form { display: grid; gap: 14px; padding: 18px; } + .confirm-form h2, .confirm-form p { margin: 0; } + .confirm-actions { display: flex; justify-content: flex-end; gap: 8px; } @media (max-width: 680px) { #search-form { grid-template-columns: 1fr; } } @media (prefers-reduced-motion: reduce) { * { transition-duration: .01ms !important; } } @@ -144,12 +157,27 @@
-

Offers

+
+

Offers

+
+

Selected: none (0/8)

+ +
+
+

SelectOfferHostGPURAMCCVerifiedReliabilityDownUpLocation$/hr
-

Selected exact offer IDs: none (0/8)

-
+ +
+

Confirm action

+

+
+ + +
+
+ diff --git a/crates/dashboard/Cargo.toml b/crates/dashboard/Cargo.toml index 3ae7410..ba17555 100644 --- a/crates/dashboard/Cargo.toml +++ b/crates/dashboard/Cargo.toml @@ -23,3 +23,6 @@ demo-control = [] [dev-dependencies] proptest = "1" tower = { version = "0.5", features = ["util"] } + +[lints] +workspace = true diff --git a/crates/dashboard/src/control.rs b/crates/dashboard/src/control.rs index 03c8a5d..1b58ca3 100644 --- a/crates/dashboard/src/control.rs +++ b/crates/dashboard/src/control.rs @@ -187,9 +187,10 @@ mod properties { concurrent in any::(), destination_disappears in any::(), ) { - let mut config = RuntimeConfig::default(); - config.worker_count = 1; - let parts = RuntimeParts::new(config); + let parts = RuntimeParts::new(RuntimeConfig { + worker_count: 1, + ..RuntimeConfig::default() + }); let runtime = parts.runtime().clone(); let backend = SteppingBackend::new(); let _engine = diff --git a/crates/dashboard/src/control_plane.rs b/crates/dashboard/src/control_plane.rs index 11075c0..f86af33 100644 --- a/crates/dashboard/src/control_plane.rs +++ b/crates/dashboard/src/control_plane.rs @@ -236,7 +236,9 @@ struct NodeCard { errors: Vec, cpu: Option, gpu: Option, + memory: Option, net: Option, + storage: Option, process: Option, history: Vec, actor_summary: ActorSummarySnapshot, @@ -329,15 +331,9 @@ fn stream_key(stream: &StreamEvent) -> String { fn node_card(node: &FusedNode, now: Instant) -> NodeCard { let summary = node.hardware.summary(); let totals = node.actors.totals(); - let mut roster: Vec = node.actors.actors.values().map(roster_row).collect(); - // Busiest actors first; ties fall back to address for stable rendering. - roster.sort_by(|left, right| { - right - .msg_per_sec - .partial_cmp(&left.msg_per_sec) - .map_or(std::cmp::Ordering::Equal, |order| order) - .then_with(|| left.address.cmp(&right.address)) - }); + // Address-keyed map order is the stable default. Volatile telemetry must + // not move a row out from under the pointer; the page offers explicit sorts. + let roster: Vec = node.actors.actors.values().map(roster_row).collect(); NodeCard { stream: StreamKeySnapshot { key: stream_key(&node.stream), @@ -352,7 +348,9 @@ fn node_card(node: &FusedNode, now: Instant) -> NodeCard { errors: node.hardware.errors(), cpu: node.hardware.cpu.as_ref().map(CpuSnapshot::from), gpu: node.hardware.gpu.as_ref().map(GpuSnapshot::from), + memory: node.hardware.memory.clone(), net: node.hardware.net.clone(), + storage: node.hardware.storage.clone(), process: node.hardware.process.clone(), history: node .hardware @@ -362,11 +360,16 @@ fn node_card(node: &FusedNode, now: Instant) -> NodeCard { ms_ago: duration_ms(now.duration_since(sample.at)), sample_unix_ms: sample.sample_unix_ms, cpu_total_percent: sample.cpu_total_percent, + cpu_cores_percent: sample.cpu_cores_percent.clone(), gpu_max_percent: sample.gpu_max_percent, gpu_memory_used_mib: sample.gpu_memory_used_mib, gpu_memory_total_mib: sample.gpu_memory_total_mib, net_rx_bps: sample.net_rx_bps, net_tx_bps: sample.net_tx_bps, + memory_used_percent: sample.memory_used_percent, + memory_pressure_some_avg10: sample.memory_pressure_some_avg10, + storage_used_percent: sample.storage_used_percent, + io_pressure_some_avg10: sample.io_pressure_some_avg10, }) .collect(), actor_summary: ActorSummarySnapshot { @@ -667,6 +670,191 @@ mod tests { assert_eq!(still_live.first(), Some(&"orch")); } + #[test] + fn hardware_channels_fold_into_fleet_snapshot() { + let view = ControlPlaneView::default(); + let stream = StreamId::new(NodeId::new("worker"), Lifetime(1)); + let cpu = json!({ + "schema":"host.cpu.v1", + "seq":1, + "sample_unix_ms":1_000, + "query_elapsed_ms":1, + "host":{ + "logical_cpus":8, + "total_percent":42.5, + "idle_percent":57.5, + "iowait_percent":0.0, + "steal_percent":0.0, + "load1":1.0, + "load5":0.5, + "load15":0.25 + }, + "cores":[ + {"index":0,"total_percent":25.0,"idle_percent":75.0,"iowait_percent":0.0,"steal_percent":0.0}, + {"index":1,"total_percent":60.0,"idle_percent":40.0,"iowait_percent":0.0,"steal_percent":0.0} + ], + "processes":[], + "error":null + }); + let gpu = json!({ + "schema":"host.gpu.v1", + "seq":1, + "sample_unix_ms":1_000, + "query_elapsed_ms":2, + "gpus":[{ + "index":0, + "uuid":"gpu-0", + "name":"test gpu", + "memory_used_mib":512, + "memory_total_mib":4096, + "utilization_gpu_percent":71, + "utilization_memory_percent":12, + "temperature_c":55, + "power_draw_w":25.0 + }], + "processes":[], + "error":null + }); + let memory = json!({ + "schema":"host.memory.v1", + "seq":1, + "sample_unix_ms":2_000, + "query_elapsed_ms":1, + "total_bytes":16_000, + "available_bytes":4_000, + "used_bytes":12_000, + "cached_bytes":2_000, + "swap_total_bytes":8_000, + "swap_used_bytes":1_000, + "pressure":{ + "some_avg10":1.25, + "some_avg60":0.75, + "some_avg300":0.5, + "some_total_us":100, + "full_avg10":0.1, + "full_avg60":0.05, + "full_avg300":0.01, + "full_total_us":10 + }, + "error":null + }); + let net_sample = |seq, sample_unix_ms, rx_bytes, tx_bytes| { + json!({ + "schema":"host.net.v1", + "seq":seq, + "sample_unix_ms":sample_unix_ms, + "interfaces":[{ + "name":"eth0", + "rx_bytes":rx_bytes, + "tx_bytes":tx_bytes, + "rx_packets":10, + "tx_packets":10, + "rx_errors":0, + "tx_errors":0, + "rx_dropped":0, + "tx_dropped":0 + }], + "error":null + }) + }; + let storage = json!({ + "schema":"host.storage.v1", + "seq":1, + "sample_unix_ms":2_000, + "query_elapsed_ms":1, + "filesystems":[{ + "mount":"/", + "total_bytes":100_000, + "used_bytes":80_000, + "available_bytes":20_000, + "used_percent":80.0 + }], + "pressure":{ + "some_avg10":2.5, + "some_avg60":1.5, + "some_avg300":0.5, + "some_total_us":200, + "full_avg10":0.2, + "full_avg60":0.1, + "full_avg300":0.05, + "full_total_us":20 + }, + "error":null + }); + + for (position, channel, payload) in [ + (0, "host.cpu", cpu), + (1, "host.gpu", gpu), + (2, "host.memory", memory), + (3, "host.net", net_sample(0, 1_000, 1_000, 2_000)), + (4, "host.net", net_sample(1, 2_000, 2_000, 3_500)), + (5, "host.storage", storage), + ] { + ingest_json( + &view, + &stream, + position, + channel, + serde_json::to_vec(&payload).expect("hardware payload"), + ); + } + + let snapshot = view.snapshot_json(); + let node = &snapshot["live"][0]; + assert_eq!(node["cpu"]["host"]["total_percent"], json!(42.5)); + assert_eq!(node["gpu"]["gpus"][0]["utilization_gpu_percent"], json!(71)); + assert_eq!(node["cpu"]["cores"][1]["total_percent"], json!(60.0)); + assert_eq!(node["memory"]["used_bytes"], json!(12_000)); + assert_eq!(node["net"]["interfaces"][0]["rx_bps"], json!(1_000.0)); + assert_eq!(node["net"]["interfaces"][0]["tx_bps"], json!(1_500.0)); + assert_eq!( + node["storage"]["filesystems"][0]["used_percent"], + json!(80.0) + ); + assert_eq!(node["history"][0]["cpu_cores_percent"], json!([25.0, 60.0])); + assert_eq!( + node["history"][0]["memory_pressure_some_avg10"], + json!(1.25) + ); + assert_eq!(node["history"][0]["io_pressure_some_avg10"], json!(2.5)); + assert_eq!(node["last_sample_unix_ms"], json!(2_000)); + assert!(node["errors"].as_array().expect("errors").is_empty()); + } + + #[test] + fn roster_default_order_does_not_follow_volatile_throughput() { + let view = ControlPlaneView::default(); + let stream = StreamId::new(NodeId::new("node"), Lifetime(1)); + ingest_json( + &view, + &stream, + 0, + "runtime.actors", + actors_payload( + 0, + json!([ + { "address": "zz", "messages_processed": 100 }, + { "address": "aa", "messages_processed": 1 } + ]), + ), + ); + { + let mut state = view.state.write(); + let actors = &mut state.streams.get_mut("node#1").expect("node").actors.actors; + actors.get_mut("zz").expect("zz actor").msg_per_sec = 10_000.0; + actors.get_mut("aa").expect("aa actor").msg_per_sec = 1.0; + } + + let snapshot = view.snapshot_json(); + let addresses: Vec<&str> = snapshot["live"][0]["roster"] + .as_array() + .expect("roster") + .iter() + .map(|actor| actor["address"].as_str().expect("address")) + .collect(); + assert_eq!(addresses, vec!["aa", "zz"]); + } + #[test] fn newer_life_generation_evicts_superseded_stream() { let view = ControlPlaneView::default(); @@ -701,7 +889,7 @@ mod tests { fn stale_pool_is_hard_capped() { let view = ControlPlaneView::default(); for index in 0..(STALE_POOL_CAP as u64 + 5) { - let stream = StreamId::new(NodeId::new(&format!("old-{index}")), Lifetime(1)); + let stream = StreamId::new(NodeId::new(format!("old-{index}")), Lifetime(1)); ingest_json( &view, &stream, diff --git a/crates/dashboard/src/control_plane_page.html b/crates/dashboard/src/control_plane_page.html index 0a74cf5..c08e52f 100644 --- a/crates/dashboard/src/control_plane_page.html +++ b/crates/dashboard/src/control_plane_page.html @@ -87,6 +87,12 @@ table { width: 100%; border-collapse: collapse; font-family: var(--mono); font-size: 12px; } th, td { text-align: left; padding: 7px 8px; border-bottom: 1px solid var(--divider); font-variant-numeric: tabular-nums; } th { color: var(--amber); font-weight: 600; font-size: 10px; text-transform: uppercase; letter-spacing: .08em; } + .roster-sort-controls { display: inline-flex; gap: 2px; margin-left: 4px; vertical-align: middle; } + .roster-sort-button { position: relative; width: 14px; height: 14px; padding: 0; color: var(--muted); background: transparent; border: 1px solid transparent; border-radius: var(--r); font-size: 0; line-height: 0; cursor: pointer; } + .roster-sort-button::before { content: ""; position: absolute; left: 3px; width: 0; height: 0; border-left: 3px solid transparent; border-right: 3px solid transparent; } + .roster-sort-button[data-direction="asc"]::before { top: 3px; border-bottom: 5px solid currentColor; } + .roster-sort-button[data-direction="desc"]::before { bottom: 3px; border-top: 5px solid currentColor; } + .roster-sort-button:hover, .roster-sort-button[aria-pressed="true"] { color: var(--cyan); border-color: var(--cyan); background: var(--panel-hover); } tbody tr[data-addr] { cursor: pointer; transition: background var(--t); } tbody tr[data-addr]:hover { background: var(--row-hover); } tbody tr[data-selected="true"] { background: var(--selected); box-shadow: inset 2px 0 0 var(--selected-edge); } @@ -102,6 +108,33 @@ .empty { padding: 24px; border: 1px dashed var(--border); border-radius: var(--r); color: var(--muted); font-family: var(--mono); font-size: 12px; } .notice { font-size: 12px; color: var(--muted); margin: 6px 0; } canvas { width: 100%; height: 64px; background: var(--inset); border: 1px solid var(--border); border-radius: var(--r); } + .hardware-source { margin-bottom: 8px; } + .hardware-grid { display: grid; grid-template-columns: repeat(12, minmax(0, 1fr)); gap: 8px; } + .hw-card { min-width: 0; padding: 10px; background: var(--inset); border: 1px solid var(--divider); border-radius: var(--r); } + .hw-card.cpu { grid-column: span 7; } + .hw-card.memory { grid-column: span 5; } + .hw-card.storage, .hw-card.gpu, .hw-card.network { grid-column: span 4; } + .hw-card-head { display: flex; align-items: baseline; justify-content: space-between; gap: 8px; margin-bottom: 7px; } + .hw-card-title { color: var(--muted); font: 700 10px/1.2 var(--mono); text-transform: uppercase; letter-spacing: .09em; } + .hw-card-value { color: var(--text); font: 700 16px/1 var(--mono); font-variant-numeric: tabular-nums; } + .hw-card-subtitle { margin-top: 5px; color: var(--muted); font: 11px/1.35 var(--mono); } + .metric-pair { display: flex; justify-content: space-between; gap: 8px; margin-top: 6px; color: var(--muted); font: 11px/1.3 var(--mono); } + .metric-pair strong { color: var(--text); font-weight: 600; } + .pressure-value.ok { color: var(--ok); } .pressure-value.warn { color: var(--amber); } .pressure-value.bad { color: var(--bad); } + canvas.thread-graph { height: 72px; background: var(--bg); border-color: var(--divider); image-rendering: pixelated; } + .thread-strip { display: grid; grid-template-columns: repeat(auto-fit, minmax(26px, 1fr)); gap: 3px; margin-top: 5px; } + .thread-cell { min-width: 0; height: 20px; display: grid; place-items: center; border: 1px solid var(--divider); border-radius: var(--r); color: var(--text); font: 9px/1 var(--mono); font-variant-numeric: tabular-nums; } + .network-list { display: grid; gap: 6px; } + .network-row { display: grid; grid-template-columns: minmax(0, 1fr) auto; gap: 8px; font: 11px/1.25 var(--mono); } + .network-row .interface { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; color: var(--muted); } + .hw-empty { color: var(--muted); font: 11px/1.35 var(--mono); } + @media (max-width: 900px) { + .hw-card.cpu { grid-column: span 12; } + .hw-card.memory, .hw-card.storage, .hw-card.gpu, .hw-card.network { grid-column: span 6; } + } + @media (max-width: 600px) { + .hw-card.cpu, .hw-card.memory, .hw-card.storage, .hw-card.gpu, .hw-card.network { grid-column: span 12; } + } .err { color: var(--bad); } @media (prefers-reduced-motion: reduce) { * { transition-duration: 0.01ms !important; animation: none !important; } } @@ -134,6 +167,8 @@ const params = new URLSearchParams(window.location.search); let selectedStream = params.get('stream') || null; let selectedActor = params.get('actor') || null; let rosterFilter = ''; +let rosterSort = { key: 'address', direction: 'asc' }; +let hardwareSource = 'node'; let lastSnapshot = null; let detailTimer = null; // A1: last-rendered HTML per region. A poll that yields identical markup @@ -144,6 +179,14 @@ let lastRosterHtml = null; let lastMachineHtml = null; let lastDossierHtml = null; +window.addEventListener('dashboard-hardware-source', event => { + const next = event.detail?.source === 'orchestrator' ? 'orchestrator' : 'node'; + if (next === hardwareSource) return; + hardwareSource = next; + lastMachineHtml = null; + if (lastSnapshot) render(); +}); + // Receipt ages (dossier) render as empty spans carrying an absolute epoch — // stable across polls — and a 1 Hz pass rewrites their textContent. All other // ticking text (card seen/stats) is written per-poll by updateCardTexts. @@ -173,6 +216,17 @@ function fmtRate(value) { if (n >= 1e3) return (n / 1e3).toFixed(1) + ' k'; return n.toFixed(n < 10 ? 1 : 0); } +function fmtBytes(value) { + const n = Number(value); + if (!Number.isFinite(n)) return '—'; + const units = ['B', 'KiB', 'MiB', 'GiB', 'TiB']; + let scaled = Math.max(0, n), unit = 0; + while (scaled >= 1024 && unit < units.length - 1) { + scaled /= 1024; + unit += 1; + } + return scaled.toFixed(scaled < 10 && unit > 0 ? 1 : 0) + ' ' + units[unit]; +} function typeShort(name) { if (!name) return '—'; const parts = String(name).split('::'); @@ -291,6 +345,14 @@ function updateCardTexts(nodes) { } } +function hardwareNodeFor(node, live, stale) { + if (hardwareSource !== 'orchestrator' || node.stream.origin === 'orchestrator') { + return { node, mirrored: false }; + } + const orchestrator = live.concat(stale).find(candidate => candidate.stream.origin === 'orchestrator'); + return orchestrator ? { node: orchestrator, mirrored: true } : { node, mirrored: false }; +} + function renderNode(page, node, live, stale) { const summary = node.actor_summary || {}; const active = document.activeElement; @@ -333,7 +395,8 @@ function renderNode(page, node, live, stale) { if (actorCount) actorCount.textContent = fmt(summary.actors); const seen = page.querySelector('h2 [data-seen]'); if (seen) seen.textContent = node.live ? 'live' : 'stale ' + ago(node.last_seen_ms_ago) + ' ago'; - const machine = machineDetail(node); + const machineSource = hardwareNodeFor(node, live, stale); + const machine = machineDetail(machineSource.node, machineSource.mirrored); const machineSlot = document.getElementById('machine-detail'); if (machineSlot && machine !== lastMachineHtml) { machineSlot.innerHTML = machine; @@ -347,67 +410,192 @@ function renderNode(page, node, live, stale) { } } -function machineDetail(node) { - const cpu = node.cpu, gpu = node.gpu, net = node.net; - const rows = []; - const cpuTotal = cpu && cpu.host ? cpu.host.total_percent : null; - rows.push(`
CPU${bar(cpuTotal)}${cpuTotal == null ? '—' : fmt(cpuTotal, 1) + '%'}
`); - if (gpu && gpu.gpus) { - gpu.gpus.forEach((device, index) => { - const util = device.utilization_gpu_percent; - const used = device.memory_used_mib || 0, total = device.memory_total_mib || 0; - rows.push(`
GPU ${fmt(index)}${bar(util)}${util == null ? '—' : fmt(util) + '%'}
`); - if (total) rows.push(`
 ${bar(used, total)}${fmt(used)}/${fmt(total)} MiB
`); - }); - } - if (net && net.interfaces) { - net.interfaces.slice(0, 4).forEach(nic => { - rows.push(`
${esc(nic.name.slice(0, 12))}${fmtRate(nic.rx_bps)}↓ ${fmtRate(nic.tx_bps)}↑
`); - }); - } - if (node.history && node.history.length) drawMachineHistory(node.history); - const errors = (node.errors || []).map(e => `
${esc(e)}
`).join(''); - return `
${rows.join('')}
${errors}`; +function percentOf(used, total) { + const numerator = Number(used), denominator = Number(total); + return Number.isFinite(numerator) && Number.isFinite(denominator) && denominator > 0 + ? numerator * 100 / denominator + : null; } -function drawMachineHistory(history) { +function pressureTone(value, warn = 1, bad = 5) { + const pressure = Number(value); + if (!Number.isFinite(pressure)) return ''; + return pressure >= bad ? 'bad' : pressure >= warn ? 'warn' : 'ok'; +} + +function machineDetail(node, mirrored = false) { + const cpu = node.cpu, gpu = node.gpu, memory = node.memory, net = node.net, storage = node.storage; + const cpuHost = cpu?.host; + const cpuTotal = cpuHost?.total_percent; + const cores = [...(cpu?.cores || [])].sort((left, right) => left.index - right.index); + const threadHeight = Math.min(92, Math.max(42, cores.length * 4)); + drawThreadGraph(node.history || [], cores.map(core => core.total_percent)); + const threadCells = cores.map(core => { + const value = core.total_percent; + const load = value == null ? 0 : Math.max(0, Math.min(100, Number(value))); + const label = value == null ? '—' : fmt(value); + return `${label}`; + }).join(''); + const load = [cpuHost?.load1, cpuHost?.load5, cpuHost?.load15].map(value => fmt(value, 2)).join(' / '); + const cpuLabel = cpuTotal == null ? '—' : fmt(cpuTotal, 1) + '%'; + + const memoryPercent = percentOf(memory?.used_bytes, memory?.total_bytes); + const memoryPressure = memory?.pressure?.some_avg10; + const swapPercent = percentOf(memory?.swap_used_bytes, memory?.swap_total_bytes); + const memoryLabel = memoryPercent == null ? '—' : fmt(memoryPercent, 1) + '%'; + + const filesystem = storage?.filesystems?.[0]; + const storagePercent = filesystem?.used_percent; + const ioPressure = storage?.pressure?.some_avg10; + const storageLabel = storagePercent == null ? '—' : fmt(storagePercent, 1) + '%'; + + const gpuDevices = gpu?.gpus || []; + const gpuMax = gpuDevices.reduce((maximum, device) => { + const value = Number(device.utilization_gpu_percent); + return Number.isFinite(value) ? Math.max(maximum, value) : maximum; + }, 0); + const gpuRows = gpuDevices.map((device, index) => { + const utilization = device.utilization_gpu_percent; + const memoryPercent = percentOf(device.memory_used_mib, device.memory_total_mib); + return `
GPU ${fmt(index)}${utilization == null ? '—' : fmt(utilization) + '%'}
+ ${bar(utilization)} +
Memory${device.memory_total_mib ? fmt(device.memory_used_mib) + ' / ' + fmt(device.memory_total_mib) + ' MiB' : '—'}
+ ${device.memory_total_mib ? bar(memoryPercent) : ''}`; + }).join(''); + + const networkRows = (net?.interfaces || []).slice(0, 4).map(nic => ` +
+ ${esc(nic.name)} + ${fmtRate(nic.rx_bps)}↓  ${fmtRate(nic.tx_bps)}↑ +
`).join(''); + + const source = mirrored ? '
Mock node · local orchestrator hardware
' : ''; + const errors = (node.errors || []).map(error => `
${esc(error)}
`).join(''); + return `${source}
+
+
CPU · ${fmt(cores.length)} threads${cpuLabel}
+ +
${threadCells || 'Waiting for per-thread samples'}
+
Load 1 / 5 / 15 min  ${load}
+
+
+
Memory${memoryLabel}
+ ${bar(memoryPercent)} +
Used${fmtBytes(memory?.used_bytes)} / ${fmtBytes(memory?.total_bytes)}
+
Available${fmtBytes(memory?.available_bytes)}
+
Swap${swapPercent == null ? '—' : fmt(swapPercent, 1) + '%'}
+
PSI some · 10s${memoryPressure == null ? '—' : fmt(memoryPressure, 2) + '%'}
+
+
+
Storage · ${esc(filesystem?.mount || '/')}${storageLabel}
+ ${bar(storagePercent)} +
Used${fmtBytes(filesystem?.used_bytes)} / ${fmtBytes(filesystem?.total_bytes)}
+
Available${fmtBytes(filesystem?.available_bytes)}
+
I/O PSI · 10s${ioPressure == null ? '—' : fmt(ioPressure, 2) + '%'}
+
+
+
GPU · ${fmt(gpuDevices.length)} devices${gpuDevices.length ? fmt(gpuMax) + '%' : '—'}
+ ${gpuRows || '
No GPU telemetry
'} +
+
+
Network${fmt(net?.interfaces?.length || 0)}
+
${networkRows || '
No interface telemetry
'}
+
+
${errors}`; +} + +function drawThreadGraph(history, currentCores) { requestAnimationFrame(() => { - const canvas = document.getElementById('machine-spark'); + const canvas = document.getElementById('thread-graph'); if (!canvas) return; + const ratio = Math.max(1, window.devicePixelRatio || 1); + const rect = canvas.getBoundingClientRect(); + const width = Math.max(1, Math.round(rect.width)); + const height = Math.max(1, Math.round(rect.height)); + const pixelWidth = Math.round(width * ratio), pixelHeight = Math.round(height * ratio); + if (canvas.width !== pixelWidth || canvas.height !== pixelHeight) { + canvas.width = pixelWidth; + canvas.height = pixelHeight; + } const ctx = canvas.getContext('2d'); - ctx.clearRect(0, 0, canvas.width, canvas.height); - const points = history.map(h => ({ cpu: h.cpu_total_percent, gpu: h.gpu_max_percent })); - drawLine(ctx, points.map(p => p.gpu), '#fbbf24'); - drawLine(ctx, points.map(p => p.cpu), '#34d399'); + ctx.setTransform(ratio, 0, 0, ratio, 0, 0); + const styles = getComputedStyle(document.documentElement); + const background = styles.getPropertyValue('--bg').trim() || '#00060c'; + const divider = styles.getPropertyValue('--divider').trim() || '#0d2c4a'; + const foreground = styles.getPropertyValue('--ok').trim() || '#00d400'; + ctx.globalAlpha = 1; + ctx.fillStyle = background; + ctx.fillRect(0, 0, width, height); + + const threadCount = Math.max(currentCores.length, ...history.map(point => point.cpu_cores_percent?.length || 0)); + if (!threadCount) return; + const rowHeight = height / threadCount; + ctx.fillStyle = divider; + for (let row = 1; row < threadCount; row += 1) { + ctx.fillRect(0, Math.floor(row * rowHeight), width, 1); + } + const slot = 2; + const points = history + .map(point => point.cpu_cores_percent || []) + .slice(-Math.floor(width / slot)); + if (!points.length) points.push(currentCores); + ctx.fillStyle = foreground; + points.forEach((values, pointIndex) => { + const x = width - (points.length - pointIndex) * slot; + values.forEach((value, threadIndex) => { + if (value == null || !Number.isFinite(Number(value))) return; + const utilization = Math.max(0, Math.min(100, Number(value))); + ctx.globalAlpha = 0.10 + utilization * 0.009; + const y = Math.ceil(threadIndex * rowHeight); + ctx.fillRect(x, y, slot, Math.max(1, Math.floor(rowHeight) - 1)); + }); + }); + ctx.globalAlpha = 1; }); } -function drawLine(ctx, values, color) { - const valid = values.filter(v => v != null); - if (valid.length < 2) return; - const max = Math.max(100, ...valid); - const step = ctx.canvas.width / (values.length - 1 || 1); - ctx.strokeStyle = color; - ctx.lineWidth = 1.5; - ctx.beginPath(); - let started = false; - values.forEach((value, index) => { - if (value == null) return; - const x = index * step; - const y = ctx.canvas.height - (value / max) * (ctx.canvas.height - 6) - 3; - if (!started) { ctx.moveTo(x, y); started = true; } else ctx.lineTo(x, y); - }); - ctx.stroke(); +function rosterSortControls(key, label) { + return ` + + + `; +} + +function compareRosterValues(left, right) { + const leftMissing = left == null || left === ''; + const rightMissing = right == null || right === ''; + if (leftMissing || rightMissing) return leftMissing === rightMissing ? 0 : leftMissing ? 1 : -1; + if (typeof left === 'number' && typeof right === 'number') return left - right; + if (typeof left === 'boolean' && typeof right === 'boolean') return Number(left) - Number(right); + return String(left).localeCompare(String(right), undefined, { numeric: true, sensitivity: 'base' }); +} + +function isLegacyHardwareSampler(actor) { + const actorType = String(actor.actor_type || ''); + return (actorType.includes('BlockingSamplerActor') && actorType.includes('telemetry::hardware')) + || /Host(?:Cpu|Gpu|Net)SamplerActor/.test(actorType); } function renderRoster(node) { const wrap = document.getElementById('roster-wrap'); if (!wrap) return; - let rows = node.roster || []; + let rows = [...(node.roster || [])]; + if (hardwareSource === 'orchestrator') { + // Mock containers may come from an older local image. Current samplers are + // engine tasks; do not leak legacy sampler actors into the simulated roster. + rows = rows.filter(actor => !isLegacyHardwareSampler(actor)); + const actorCount = document.querySelector('[data-actor-count]'); + if (actorCount) actorCount.textContent = fmt(rows.length); + } if (rosterFilter) { const q = rosterFilter.toLowerCase(); rows = rows.filter(a => `${a.name || ''} ${a.actor_type || ''} ${a.address} ${a.worker_id ?? ''}`.toLowerCase().includes(q)); } + rows.sort((left, right) => { + const compared = compareRosterValues(left[rosterSort.key], right[rosterSort.key]); + return (rosterSort.direction === 'asc' ? compared : -compared) + || String(left.address).localeCompare(String(right.address)); + }); const capped = rows.slice(0, ROSTER_RENDER_CAP); const notice = rows.length > capped.length ? `
showing ${fmt(capped.length)} of ${fmt(rows.length)} — refine the filter to see more
` @@ -417,7 +605,7 @@ function renderRoster(node) { html = '
No actors on this stream (or none match the filter).
'; } else { html = notice + ` - + ${capped.map(a => ` @@ -552,6 +740,12 @@ pageEl.addEventListener('click', e => { render(); return; } + const sortButton = e.target.closest('.roster-sort-button'); + if (sortButton) { + rosterSort = { key: sortButton.dataset.sort, direction: sortButton.dataset.direction }; + render(); + return; + } const rosterRow = e.target.closest('tr[data-addr]'); if (rosterRow) { selectedActor = rosterRow.getAttribute('data-addr'); diff --git a/crates/dashboard/src/hardware_view.rs b/crates/dashboard/src/hardware_view.rs index 4d61f11..d7c2d19 100644 --- a/crates/dashboard/src/hardware_view.rs +++ b/crates/dashboard/src/hardware_view.rs @@ -8,7 +8,9 @@ use telemetry::hardware::cpu::{ use telemetry::hardware::gpu::{ GpuDeviceSample, GpuProcessSample, HOST_GPU_CHANNEL, HostGpuSample, }; +use telemetry::hardware::memory::{HOST_MEMORY_CHANNEL, HostMemorySample}; use telemetry::hardware::net::{HOST_NET_CHANNEL, HostNetSample, NetInterfaceSample}; +use telemetry::hardware::storage::{HOST_STORAGE_CHANNEL, HostStorageSample}; use serde::Serialize; @@ -27,7 +29,9 @@ pub(crate) struct NodeHardwareState { decode_errors: BTreeMap<&'static str, String>, pub(crate) cpu: Option, pub(crate) gpu: Option, + pub(crate) memory: Option, pub(crate) net: Option, + pub(crate) storage: Option, pub(crate) process: Option, pub(crate) history: VecDeque, } @@ -39,7 +43,9 @@ impl NodeHardwareState { decode_errors: BTreeMap::new(), cpu: None, gpu: None, + memory: None, net: None, + storage: None, process: None, history: VecDeque::with_capacity(HISTORY_CAP), } @@ -64,6 +70,14 @@ impl NodeHardwareState { } Err(error) => self.store_decode_error(HOST_GPU_CHANNEL, error), }, + HOST_MEMORY_CHANNEL => match HostMemorySample::decode(payload) { + Ok(sample) => { + self.memory = Some(sample); + self.decode_errors.remove(HOST_MEMORY_CHANNEL); + self.update_history(now); + } + Err(error) => self.store_decode_error(HOST_MEMORY_CHANNEL, error), + }, HOST_NET_CHANNEL => match HostNetSample::decode(payload) { Ok(sample) => { self.net = Some(NetSnapshot::from_sample(sample, self.net.as_ref())); @@ -72,6 +86,14 @@ impl NodeHardwareState { } Err(error) => self.store_decode_error(HOST_NET_CHANNEL, error), }, + HOST_STORAGE_CHANNEL => match HostStorageSample::decode(payload) { + Ok(sample) => { + self.storage = Some(sample); + self.decode_errors.remove(HOST_STORAGE_CHANNEL); + self.update_history(now); + } + Err(error) => self.store_decode_error(HOST_STORAGE_CHANNEL, error), + }, _ => { if channel.starts_with("proc.") && channel.ends_with(".lifecycle") { self.process = decode_process_snapshot(payload); @@ -99,6 +121,35 @@ impl NodeHardwareState { .as_ref() .and_then(|sample| sample.host.as_ref()) .and_then(|host| host.total_percent); + let cpu_cores_percent = self + .cpu + .as_ref() + .map(|sample| { + sample + .cores + .iter() + .map(|core| core.total_percent) + .collect::>() + }) + .unwrap_or_default(); + let memory_used_percent = self.memory.as_ref().and_then(|sample| { + Some(sample.used_bytes? as f64 * 100.0 / sample.total_bytes?.max(1) as f64) + }); + let memory_pressure_some_avg10 = self + .memory + .as_ref() + .and_then(|sample| sample.pressure.as_ref()) + .map(|pressure| pressure.some_avg10); + let storage_used_percent = self + .storage + .as_ref() + .and_then(|sample| sample.filesystems.first()) + .and_then(|filesystem| filesystem.used_percent); + let io_pressure_some_avg10 = self + .storage + .as_ref() + .and_then(|sample| sample.pressure.as_ref()) + .map(|pressure| pressure.some_avg10); let mut gpu_max_percent = None; let mut gpu_memory_used_mib = 0_u64; @@ -128,17 +179,24 @@ impl NodeHardwareState { sample_unix_ms: [ self.cpu.as_ref().map(|sample| sample.sample_unix_ms), self.gpu.as_ref().map(|sample| sample.sample_unix_ms), + self.memory.as_ref().map(|sample| sample.sample_unix_ms), self.net.as_ref().map(|sample| sample.sample_unix_ms), + self.storage.as_ref().map(|sample| sample.sample_unix_ms), ] .into_iter() .flatten() .max(), cpu_total_percent, + cpu_cores_percent, gpu_max_percent, gpu_memory_used_mib, gpu_memory_total_mib, net_rx_bps, net_tx_bps, + memory_used_percent, + memory_pressure_some_avg10, + storage_used_percent, + io_pressure_some_avg10, } } @@ -149,11 +207,16 @@ impl NodeHardwareState { { last.sample_unix_ms = summary.sample_unix_ms; last.cpu_total_percent = summary.cpu_total_percent; + last.cpu_cores_percent = summary.cpu_cores_percent.clone(); last.gpu_max_percent = summary.gpu_max_percent; last.gpu_memory_used_mib = summary.gpu_memory_used_mib; last.gpu_memory_total_mib = summary.gpu_memory_total_mib; last.net_rx_bps = summary.net_rx_bps; last.net_tx_bps = summary.net_tx_bps; + last.memory_used_percent = summary.memory_used_percent; + last.memory_pressure_some_avg10 = summary.memory_pressure_some_avg10; + last.storage_used_percent = summary.storage_used_percent; + last.io_pressure_some_avg10 = summary.io_pressure_some_avg10; return; } @@ -164,11 +227,16 @@ impl NodeHardwareState { at: now, sample_unix_ms: summary.sample_unix_ms, cpu_total_percent: summary.cpu_total_percent, + cpu_cores_percent: summary.cpu_cores_percent, gpu_max_percent: summary.gpu_max_percent, gpu_memory_used_mib: summary.gpu_memory_used_mib, gpu_memory_total_mib: summary.gpu_memory_total_mib, net_rx_bps: summary.net_rx_bps, net_tx_bps: summary.net_tx_bps, + memory_used_percent: summary.memory_used_percent, + memory_pressure_some_avg10: summary.memory_pressure_some_avg10, + storage_used_percent: summary.storage_used_percent, + io_pressure_some_avg10: summary.io_pressure_some_avg10, }); } @@ -180,9 +248,23 @@ impl NodeHardwareState { if let Some(error) = self.gpu.as_ref().and_then(|sample| sample.error.as_ref()) { errors.push(format!("{HOST_GPU_CHANNEL}: {error}")); } + if let Some(error) = self + .memory + .as_ref() + .and_then(|sample| sample.error.as_ref()) + { + errors.push(format!("{HOST_MEMORY_CHANNEL}: {error}")); + } if let Some(error) = self.net.as_ref().and_then(|sample| sample.error.as_ref()) { errors.push(format!("{HOST_NET_CHANNEL}: {error}")); } + if let Some(error) = self + .storage + .as_ref() + .and_then(|sample| sample.error.as_ref()) + { + errors.push(format!("{HOST_STORAGE_CHANNEL}: {error}")); + } errors } } @@ -308,26 +390,36 @@ impl NetInterfaceSnapshot { } } -#[derive(Clone, Copy)] +#[derive(Clone)] pub(crate) struct HardwareSummary { pub(crate) sample_unix_ms: Option, pub(crate) cpu_total_percent: Option, + pub(crate) cpu_cores_percent: Vec>, pub(crate) gpu_max_percent: Option, pub(crate) gpu_memory_used_mib: u64, pub(crate) gpu_memory_total_mib: u64, pub(crate) net_rx_bps: f64, pub(crate) net_tx_bps: f64, + pub(crate) memory_used_percent: Option, + pub(crate) memory_pressure_some_avg10: Option, + pub(crate) storage_used_percent: Option, + pub(crate) io_pressure_some_avg10: Option, } pub(crate) struct HardwareHistoryState { pub(crate) at: Instant, pub(crate) sample_unix_ms: Option, pub(crate) cpu_total_percent: Option, + pub(crate) cpu_cores_percent: Vec>, pub(crate) gpu_max_percent: Option, pub(crate) gpu_memory_used_mib: u64, pub(crate) gpu_memory_total_mib: u64, pub(crate) net_rx_bps: f64, pub(crate) net_tx_bps: f64, + pub(crate) memory_used_percent: Option, + pub(crate) memory_pressure_some_avg10: Option, + pub(crate) storage_used_percent: Option, + pub(crate) io_pressure_some_avg10: Option, } #[derive(Serialize)] @@ -383,11 +475,16 @@ pub(crate) struct HardwareHistorySnapshot { pub(crate) ms_ago: u64, pub(crate) sample_unix_ms: Option, pub(crate) cpu_total_percent: Option, + pub(crate) cpu_cores_percent: Vec>, pub(crate) gpu_max_percent: Option, pub(crate) gpu_memory_used_mib: u64, pub(crate) gpu_memory_total_mib: u64, pub(crate) net_rx_bps: f64, pub(crate) net_tx_bps: f64, + pub(crate) memory_used_percent: Option, + pub(crate) memory_pressure_some_avg10: Option, + pub(crate) storage_used_percent: Option, + pub(crate) io_pressure_some_avg10: Option, } pub(crate) fn duration_ms(duration: Duration) -> u64 { diff --git a/crates/dashboard/src/server.rs b/crates/dashboard/src/server.rs index a65eba2..0c5e97f 100644 --- a/crates/dashboard/src/server.rs +++ b/crates/dashboard/src/server.rs @@ -469,7 +469,7 @@ mod tests { } fn is_valid_for_route(&self) -> bool { - self.payload % 5 == 0 + self.payload.is_multiple_of(5) } } diff --git a/crates/dashboard/src/swactor/actor_view.rs b/crates/dashboard/src/swactor/actor_view.rs index 2dabfbf..9254ea9 100644 --- a/crates/dashboard/src/swactor/actor_view.rs +++ b/crates/dashboard/src/swactor/actor_view.rs @@ -20,6 +20,7 @@ //! out. Receipts are bounded and interval-spaced so noisy actors cannot flood //! the page. +use std::cmp::Reverse; use std::collections::{BTreeMap, VecDeque}; use std::time::{Duration, Instant}; @@ -172,8 +173,10 @@ impl RuntimeState { } pub(crate) fn totals(&self) -> Totals { - let mut totals = Totals::default(); - totals.actors = self.actors.len().min(u32::MAX as usize) as u32; + let mut totals = Totals { + actors: self.actors.len().min(u32::MAX as usize) as u32, + ..Totals::default() + }; for actor in self.actors.values() { totals.mailbox_depth = totals.mailbox_depth.saturating_add(actor.mailbox_depth); totals.msg_per_sec += actor.msg_per_sec; @@ -449,7 +452,7 @@ fn parse_message_type_counts(value: Option<&Value>) -> Option .iter() .filter_map(|(name, count)| value_to_u64(count).map(|count| (name.clone(), count))) .collect(); - out.sort_by(|a, b| b.1.cmp(&a.1)); + out.sort_by_key(|&(_, count)| Reverse(count)); return Some(out); } None diff --git a/crates/telemetry/Cargo.toml b/crates/telemetry/Cargo.toml index 8966225..eea19a7 100644 --- a/crates/telemetry/Cargo.toml +++ b/crates/telemetry/Cargo.toml @@ -7,6 +7,8 @@ license = "AGPL-3.0-only" [dependencies] swactor = { path = "../..", features = ["serde", "transport"] } swactor-transport = { path = "../transport" } +swactor-engine = { path = "../engine" } +futures-channel = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" iroh = "0.98" @@ -14,3 +16,6 @@ crossbeam-channel = "0.5" [target.'cfg(target_os = "linux")'.dependencies] libc = "0.2" + +[lints] +workspace = true diff --git a/crates/telemetry/src/emit.rs b/crates/telemetry/src/emit.rs index 8aff1d3..999675b 100644 --- a/crates/telemetry/src/emit.rs +++ b/crates/telemetry/src/emit.rs @@ -17,6 +17,7 @@ pub trait FrameSink: Send { /// Ship one positioned frame for `stream`. Best-effort: a sink may drop. fn ship(&mut self, stream: &StreamId, frame: &Frame); } +pub type ProcessChannelRouter = dyn Fn(&str, bool) -> ChannelId + Send + Sync; /// Static identity a node needs to build its mux. pub struct EmitterConfig { @@ -27,7 +28,7 @@ pub struct EmitterConfig { struct MuxProcObserver { mux: Arc, - channel_for: Arc ChannelId + Send + Sync>, + channel_for: Arc, } impl ProcessOutputObserver for MuxProcObserver { diff --git a/crates/telemetry/src/endpoint.rs b/crates/telemetry/src/endpoint.rs index cd439e2..9f3ad10 100644 --- a/crates/telemetry/src/endpoint.rs +++ b/crates/telemetry/src/endpoint.rs @@ -16,6 +16,7 @@ use serde::{Deserialize, Serialize}; use swactor::process_observer::ProcessOutputObserver; use swactor::stats::{ActorSnapshot, StatsHook}; +use crate::emit::ProcessChannelRouter; use crate::frame::{ ChannelContent, ChannelDescriptor, ChannelFilter, ChannelId, ChannelRef, FrameDelivery, SourceFilter, StreamDescriptor, StreamId, StreamOrigin, SubscriptionRequest, TelemetryEvent, @@ -721,7 +722,7 @@ fn register_channel( /// Legacy/custom process-output observer adapter that submits stdout/stderr chunks as frames. pub struct TelemetryProcessObserver { producer: TelemetryProducer, - channel_for: Arc ChannelId + Send + Sync>, + channel_for: Arc, } impl ProcessOutputObserver for TelemetryProcessObserver { diff --git a/crates/telemetry/src/hardware/cpu.rs b/crates/telemetry/src/hardware/cpu.rs index 8c8602e..01b2401 100644 --- a/crates/telemetry/src/hardware/cpu.rs +++ b/crates/telemetry/src/hardware/cpu.rs @@ -324,10 +324,9 @@ fn parse_cpu_snapshot(raw: &str) -> Option { } else if let Some(index) = label .strip_prefix("cpu") .and_then(|suffix| suffix.parse::().ok()) + && let Some(times) = parse_cpu_times(line) { - if let Some(times) = parse_cpu_times(line) { - cores.push(CpuCoreTimes { index, times }); - } + cores.push(CpuCoreTimes { index, times }); } } diff --git a/crates/telemetry/src/hardware/memory.rs b/crates/telemetry/src/hardware/memory.rs new file mode 100644 index 0000000..f9658e6 --- /dev/null +++ b/crates/telemetry/src/hardware/memory.rs @@ -0,0 +1,151 @@ +use std::collections::BTreeMap; +use std::fs; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; + +use super::pressure::{self, PressureSample}; +use crate::record::Record; + +pub const HOST_MEMORY_CHANNEL: &str = "host.memory"; +pub const MEMORY_SAMPLE_INTERVAL: Duration = Duration::from_secs(1); + +const SCHEMA: &str = "host.memory.v1"; +const KIB: u64 = 1024; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct HostMemorySample { + pub schema: String, + pub seq: u64, + pub sample_unix_ms: u64, + pub query_elapsed_ms: Option, + pub total_bytes: Option, + pub available_bytes: Option, + pub used_bytes: Option, + pub cached_bytes: Option, + pub swap_total_bytes: Option, + pub swap_used_bytes: Option, + pub pressure: Option, + pub error: Option, +} + +impl Record for HostMemorySample { + const CHANNEL: &'static str = HOST_MEMORY_CHANNEL; +} + +pub fn sample(seq: u64) -> HostMemorySample { + let started = Instant::now(); + let sample_unix_ms = unix_ms_now(); + let memory = match fs::read_to_string("/proc/meminfo") { + Ok(raw) => match parse_meminfo(&raw) { + Some(memory) => memory, + None => return HostMemorySample::error(seq, "parse /proc/meminfo"), + }, + Err(error) => return HostMemorySample::error(seq, format!("read /proc/meminfo: {error}")), + }; + HostMemorySample { + schema: SCHEMA.to_owned(), + seq, + sample_unix_ms, + query_elapsed_ms: Some(elapsed_ms(started)), + total_bytes: Some(memory.total_bytes), + available_bytes: Some(memory.available_bytes), + used_bytes: Some(memory.total_bytes.saturating_sub(memory.available_bytes)), + cached_bytes: Some(memory.cached_bytes), + swap_total_bytes: Some(memory.swap_total_bytes), + swap_used_bytes: Some( + memory + .swap_total_bytes + .saturating_sub(memory.swap_free_bytes), + ), + pressure: pressure::read("memory").ok(), + error: None, + } +} + +impl HostMemorySample { + fn error(seq: u64, error: impl Into) -> Self { + Self { + schema: SCHEMA.to_owned(), + seq, + sample_unix_ms: unix_ms_now(), + query_elapsed_ms: None, + total_bytes: None, + available_bytes: None, + used_bytes: None, + cached_bytes: None, + swap_total_bytes: None, + swap_used_bytes: None, + pressure: pressure::read("memory").ok(), + error: Some(error.into()), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct MemoryCounters { + total_bytes: u64, + available_bytes: u64, + cached_bytes: u64, + swap_total_bytes: u64, + swap_free_bytes: u64, +} + +fn parse_meminfo(raw: &str) -> Option { + let mut values = BTreeMap::new(); + for line in raw.lines() { + let (name, rest) = line.split_once(':')?; + let value_kib = rest.split_whitespace().next()?.parse::().ok()?; + values.insert(name, value_kib.saturating_mul(KIB)); + } + let cached_bytes = values + .get("Cached") + .copied() + .unwrap_or(0) + .saturating_add(values.get("SReclaimable").copied().unwrap_or(0)); + Some(MemoryCounters { + total_bytes: *values.get("MemTotal")?, + available_bytes: *values.get("MemAvailable")?, + cached_bytes, + swap_total_bytes: values.get("SwapTotal").copied().unwrap_or(0), + swap_free_bytes: values.get("SwapFree").copied().unwrap_or(0), + }) +} + +fn unix_ms_now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) +} + +fn elapsed_ms(started: Instant) -> u64 { + started.elapsed().as_millis().try_into().unwrap_or(u64::MAX) +} + +#[cfg(test)] +mod tests { + use super::{parse_meminfo, sample}; + + #[test] + fn derives_used_cached_and_swap_memory() { + let counters = parse_meminfo( + "MemTotal: 1000 kB\nMemAvailable: 400 kB\nCached: 100 kB\nSReclaimable: 20 kB\nSwapTotal: 200 kB\nSwapFree: 150 kB\n", + ) + .expect("memory counters"); + assert_eq!(counters.total_bytes, 1_024_000); + assert_eq!(counters.available_bytes, 409_600); + assert_eq!(counters.cached_bytes, 122_880); + assert_eq!(counters.swap_total_bytes - counters.swap_free_bytes, 51_200); + } + + #[test] + fn samples_live_memory() { + let sample = sample(7); + assert_eq!(sample.seq, 7); + assert!(sample.total_bytes.is_some_and(|total| total > 0)); + assert!(sample.error.is_none()); + } +} diff --git a/crates/telemetry/src/hardware/mod.rs b/crates/telemetry/src/hardware/mod.rs index a1b16c9..a65628e 100644 --- a/crates/telemetry/src/hardware/mod.rs +++ b/crates/telemetry/src/hardware/mod.rs @@ -1,3 +1,103 @@ pub mod cpu; pub mod gpu; +pub mod memory; pub mod net; +pub mod pressure; +pub mod storage; + +use std::time::Duration; + +use swactor_engine::EngineHandle; + +/// Run a stateful blocking sampler on the engine without creating an actor. +/// +/// Sampling never overlaps: the next interval is armed only after the previous +/// blocking sample has returned and `observed` has consumed its result. +pub fn spawn_blocking_sampler( + engine: EngineHandle, + period: Duration, + state: State, + sample: fn(State, u64) -> (State, Sample), + started: Started, + mut observed: Observed, +) where + State: Send + 'static, + Sample: Send + 'static, + Started: FnOnce() + Send + 'static, + Observed: FnMut(u64, Sample) + Send + 'static, +{ + let task_engine = engine.clone(); + engine.spawn(async move { + started(); + let blocking_work = task_engine.blocking_work_sender(); + let mut interval = task_engine.interval(period); + let mut state = state; + let mut seq = 0_u64; + + loop { + (&mut interval).await; + let current_state = state; + let (sample_tx, sample_rx) = futures_channel::oneshot::channel(); + let work = Box::new(move || { + let _ = sample_tx.send(sample(current_state, seq)); + }); + if blocking_work.submit(work).is_err() { + return; + } + let Ok((next_state, result)) = sample_rx.await else { + return; + }; + state = next_state; + observed(seq, result); + seq = seq.saturating_add(1); + } + }); +} + +#[cfg(test)] +mod tests { + use std::sync::mpsc; + use std::time::Duration; + + use swactor::config::RuntimeConfig; + use swactor::runtime::RuntimeParts; + use swactor_engine::{Engine, TokioBackend, TokioConfig}; + + use super::spawn_blocking_sampler; + + #[test] + fn blocking_sampler_runs_sequentially_with_monotonic_sequences() { + let engine = Engine::new( + RuntimeParts::new(RuntimeConfig::default()), + TokioBackend::new(TokioConfig::default()).expect("Tokio backend"), + ) + .expect("engine"); + let (observed_tx, observed_rx) = mpsc::channel(); + + spawn_blocking_sampler( + engine.handle(), + Duration::from_millis(1), + 0_u64, + |state, seq| (state + 1, (seq, state)), + || {}, + move |seq, result| { + observed_tx + .send((seq, result)) + .expect("observation receiver") + }, + ); + + assert_eq!( + observed_rx + .recv_timeout(Duration::from_secs(1)) + .expect("first sample"), + (0, (0, 0)), + ); + assert_eq!( + observed_rx + .recv_timeout(Duration::from_secs(1)) + .expect("second sample"), + (1, (1, 1)), + ); + } +} diff --git a/crates/telemetry/src/hardware/pressure.rs b/crates/telemetry/src/hardware/pressure.rs new file mode 100644 index 0000000..a805997 --- /dev/null +++ b/crates/telemetry/src/hardware/pressure.rs @@ -0,0 +1,95 @@ +use std::fs; + +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct PressureSample { + pub some_avg10: f64, + pub some_avg60: f64, + pub some_avg300: f64, + pub some_total_us: u64, + pub full_avg10: Option, + pub full_avg60: Option, + pub full_avg300: Option, + pub full_total_us: Option, +} + +pub(crate) fn read(resource: &str) -> Result { + let path = format!("/proc/pressure/{resource}"); + let raw = fs::read_to_string(&path).map_err(|error| format!("read {path}: {error}"))?; + parse(&raw).ok_or_else(|| format!("parse {path}")) +} + +fn parse(raw: &str) -> Option { + let some = parse_row(raw.lines().find(|line| line.starts_with("some "))?)?; + let full = raw + .lines() + .find(|line| line.starts_with("full ")) + .and_then(parse_row); + Some(PressureSample { + some_avg10: some.avg10, + some_avg60: some.avg60, + some_avg300: some.avg300, + some_total_us: some.total_us, + full_avg10: full.as_ref().map(|row| row.avg10), + full_avg60: full.as_ref().map(|row| row.avg60), + full_avg300: full.as_ref().map(|row| row.avg300), + full_total_us: full.map(|row| row.total_us), + }) +} + +#[derive(Debug, Clone, Copy)] +struct PressureRow { + avg10: f64, + avg60: f64, + avg300: f64, + total_us: u64, +} + +fn parse_row(line: &str) -> Option { + let mut avg10 = None; + let mut avg60 = None; + let mut avg300 = None; + let mut total_us = None; + for field in line.split_whitespace().skip(1) { + let (name, value) = field.split_once('=')?; + match name { + "avg10" => avg10 = value.parse().ok(), + "avg60" => avg60 = value.parse().ok(), + "avg300" => avg300 = value.parse().ok(), + "total" => total_us = value.parse().ok(), + _ => {} + } + } + Some(PressureRow { + avg10: avg10?, + avg60: avg60?, + avg300: avg300?, + total_us: total_us?, + }) +} + +#[cfg(test)] +mod tests { + use super::parse; + + #[test] + fn parses_some_and_full_pressure_rows() { + let sample = parse( + "some avg10=1.25 avg60=2.50 avg300=3.75 total=1234\nfull avg10=0.10 avg60=0.20 avg300=0.30 total=42\n", + ) + .expect("pressure sample"); + assert_eq!(sample.some_avg10, 1.25); + assert_eq!(sample.some_total_us, 1234); + assert_eq!(sample.full_avg10, Some(0.10)); + assert_eq!(sample.full_total_us, Some(42)); + } + + #[test] + fn accepts_cpu_pressure_without_full_row() { + let sample = + parse("some avg10=0.00 avg60=0.01 avg300=0.02 total=99\n").expect("pressure sample"); + assert_eq!(sample.some_total_us, 99); + assert_eq!(sample.full_avg10, None); + } +} diff --git a/crates/telemetry/src/hardware/storage.rs b/crates/telemetry/src/hardware/storage.rs new file mode 100644 index 0000000..3e15dea --- /dev/null +++ b/crates/telemetry/src/hardware/storage.rs @@ -0,0 +1,123 @@ +use std::ffi::CString; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; + +use super::pressure::{self, PressureSample}; +use crate::record::Record; + +pub const HOST_STORAGE_CHANNEL: &str = "host.storage"; +pub const STORAGE_SAMPLE_INTERVAL: Duration = Duration::from_secs(1); + +const SCHEMA: &str = "host.storage.v1"; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct HostStorageSample { + pub schema: String, + pub seq: u64, + pub sample_unix_ms: u64, + pub query_elapsed_ms: Option, + pub filesystems: Vec, + pub pressure: Option, + pub error: Option, +} + +impl Record for HostStorageSample { + const CHANNEL: &'static str = HOST_STORAGE_CHANNEL; +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct FilesystemSample { + pub mount: String, + pub total_bytes: u64, + pub used_bytes: u64, + pub available_bytes: u64, + pub used_percent: Option, +} + +pub fn sample(seq: u64) -> HostStorageSample { + let started = Instant::now(); + match read_filesystem("/") { + Ok(filesystem) => HostStorageSample { + schema: SCHEMA.to_owned(), + seq, + sample_unix_ms: unix_ms_now(), + query_elapsed_ms: Some(elapsed_ms(started)), + filesystems: vec![filesystem], + pressure: pressure::read("io").ok(), + error: None, + }, + Err(error) => HostStorageSample { + schema: SCHEMA.to_owned(), + seq, + sample_unix_ms: unix_ms_now(), + query_elapsed_ms: Some(elapsed_ms(started)), + filesystems: Vec::new(), + pressure: pressure::read("io").ok(), + error: Some(error), + }, + } +} + +#[cfg(target_os = "linux")] +fn read_filesystem(mount: &str) -> Result { + let path = CString::new(mount).map_err(|error| format!("filesystem path: {error}"))?; + let mut stats = std::mem::MaybeUninit::::uninit(); + // SAFETY: `path` is a live NUL-terminated string and `stats` points to writable storage. + let result = unsafe { libc::statvfs(path.as_ptr(), stats.as_mut_ptr()) }; + if result != 0 { + return Err(format!( + "statvfs {mount}: {}", + std::io::Error::last_os_error() + )); + } + // SAFETY: successful `statvfs` initialized the output structure. + let stats = unsafe { stats.assume_init() }; + let fragment_size = stats.f_frsize; + let total_bytes = stats.f_blocks.saturating_mul(fragment_size); + let free_bytes = stats.f_bfree.saturating_mul(fragment_size); + let available_bytes = stats.f_bavail.saturating_mul(fragment_size); + let used_bytes = total_bytes.saturating_sub(free_bytes); + let used_percent = (total_bytes > 0).then_some(used_bytes as f64 * 100.0 / total_bytes as f64); + Ok(FilesystemSample { + mount: mount.to_owned(), + total_bytes, + used_bytes, + available_bytes, + used_percent, + }) +} + +#[cfg(not(target_os = "linux"))] +fn read_filesystem(mount: &str) -> Result { + Err(format!("filesystem sampling unsupported for {mount}")) +} + +fn unix_ms_now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) +} + +fn elapsed_ms(started: Instant) -> u64 { + started.elapsed().as_millis().try_into().unwrap_or(u64::MAX) +} + +#[cfg(test)] +mod tests { + use super::sample; + + #[test] + fn samples_root_filesystem_capacity() { + let sample = sample(9); + assert_eq!(sample.seq, 9); + let root = sample.filesystems.first().expect("root filesystem"); + assert_eq!(root.mount, "/"); + assert!(root.total_bytes > 0); + assert!(root.used_bytes <= root.total_bytes); + assert!(sample.error.is_none()); + } +} diff --git a/crates/telemetry/src/mux.rs b/crates/telemetry/src/mux.rs index 4e19a0f..51bea9c 100644 --- a/crates/telemetry/src/mux.rs +++ b/crates/telemetry/src/mux.rs @@ -4,7 +4,7 @@ //! payloads into a bounded queue first, then assigns a single monotonic position //! sequence while draining accepted payloads. -use crossbeam_channel::{Receiver, Sender, TryRecvError, TrySendError, bounded}; +use crossbeam_channel::{Receiver, Sender, TrySendError, bounded}; use std::sync::atomic::{AtomicU64, Ordering}; use crate::frame::{ChannelId, Frame, Position, StreamId}; @@ -26,7 +26,7 @@ pub struct Mux { impl Mux { /// Create a mux for `stream` with a bounded outgoing queue. pub fn new(stream: StreamId, capacity: usize) -> Self { - let capacity = capacity.max(1).min(1_048_576); + let capacity = capacity.clamp(1, 1_048_576); let (tx, rx) = bounded(capacity); Mux { stream, @@ -61,20 +61,15 @@ impl Mux { /// Pull all currently queued frames in mux queue order. pub fn drain(&self) -> Vec { let mut frames = Vec::new(); - loop { - match self.rx.try_recv() { - Ok(pending) => { - // Position is consumed only after a pending frame has left - // the queue; failed submit never reaches this point. - let position = Position(self.next.fetch_add(1, Ordering::Relaxed)); - frames.push(Frame { - channel: pending.channel, - position, - payload: pending.payload, - }); - } - Err(TryRecvError::Empty) | Err(TryRecvError::Disconnected) => break, - } + while let Ok(pending) = self.rx.try_recv() { + // Position is consumed only after a pending frame has left + // the queue; failed submit never reaches this point. + let position = Position(self.next.fetch_add(1, Ordering::Relaxed)); + frames.push(Frame { + channel: pending.channel, + position, + payload: pending.payload, + }); } frames } diff --git a/crates/telemetry/tests/t_telemetry.rs b/crates/telemetry/tests/t_telemetry.rs index e92610c..69bd86b 100644 --- a/crates/telemetry/tests/t_telemetry.rs +++ b/crates/telemetry/tests/t_telemetry.rs @@ -27,17 +27,6 @@ impl Record for ResourceSample { const CHANNEL: &'static str = "host.resource"; } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -struct MembershipTransition { - peer: String, - from: String, - to: String, -} - -impl Record for MembershipTransition { - const CHANNEL: &'static str = "membership"; -} - fn stream() -> StreamId { StreamId::new(NodeId::new("node-alpha"), Lifetime(1)) } diff --git a/crates/telemetry/tests/t_telemetry_realio.rs b/crates/telemetry/tests/t_telemetry_realio.rs index 9bd5556..c633ffb 100644 --- a/crates/telemetry/tests/t_telemetry_realio.rs +++ b/crates/telemetry/tests/t_telemetry_realio.rs @@ -10,7 +10,7 @@ use telemetry::ingest::Consumer; use telemetry::mux::Mux; use telemetry::transport::Delivery; use telemetry::wire::{decode_delivery, encode_delivery}; -use telemetry::{ChannelId, Lifetime, NodeId, Position, Record, StreamId}; +use telemetry::{ChannelId, Lifetime, NodeId, Record, StreamId}; const RESOURCE_CHANNEL: ChannelId = ChannelId(1); const LOG_CHANNEL: ChannelId = ChannelId(2); diff --git a/dash-fixes.md b/dash-fixes.md new file mode 100644 index 0000000..0048cec --- /dev/null +++ b/dash-fixes.md @@ -0,0 +1,12 @@ +- Add connection map? +- Noisy telemetrics +- terminate cluster button +- popup should be ui-flavored not browser popup +- hardware stats seem broken +- button for selecting offers needs to be higher, so I dont have to scroll down. Maybe next to 'OFFERS' text. +- need some more information indicators for the remote nodes. Maybe stdout should be displayed on the Fleet focus view, so we can see activity? +- Actors and their stats are incredibly noisy, not sure its an easy fix. IFF the fix is obvious and easy, we need a way to fix the stats displayed for: mailbox (clears so fast its always 0), throughput (always too high-looking, ~4k msg/s but its processed maybe 2 or three) When focused, the throughput graph looks fine though. +- For telemetry specifically, a lot of this should be done by a task not an actor, actors are control flow, tasks are for continuous state-independant work (like already-negotiated streams using quic, and hardware stat samples on a fixed timer). +- Potential, future, instead of only 'Kill' we should have other buttons that match typical cloud and particularly vastai options? +- Actors on the Fleet single node focus panel need to not jump around like they do, it makes clicking them too difficult. +- We should add arrows, copy the vastai provisioning arrows, that allow you to sort the various actor bits quickly
ActorTypeStateMailboxMsg/sProcessedWorkerLast message
Actor${rosterSortControls('address', 'actor')}Type${rosterSortControls('actor_type', 'type')}State${rosterSortControls('poisoned', 'state')}Mailbox${rosterSortControls('mailbox_depth', 'mailbox')}Msg/s${rosterSortControls('msg_per_sec', 'throughput')}Processed${rosterSortControls('messages_processed', 'processed count')}Worker${rosterSortControls('worker_id', 'worker')}Last message${rosterSortControls('last_msg_type', 'last message')}
${esc(shortAddr(a.address))}${a.name ? `
${esc(a.name)}` : ''}		 ${typeShort(a.actor_type)}