diff --git a/.dockerignore b/.dockerignore index 4611add..3b36028 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,5 +1,6 @@ * !target/x86_64-unknown-linux-musl/release/swactor +!target/x86_64-unknown-linux-musl/release/swactor-datastream-collector !target/release/swactor-diag-collector !target/release/swactor-diag-postproc !examples/single-gpu-inference/target/release/gpu-node diff --git a/Dockerfile b/Dockerfile index 1e65f4a..5db55b3 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,3 +1,4 @@ FROM scratch COPY target/x86_64-unknown-linux-musl/release/swactor /swactor +COPY target/x86_64-unknown-linux-musl/release/swactor-datastream-collector /collector ENTRYPOINT ["/swactor"] diff --git a/crates/dashboard/Cargo.toml b/crates/dashboard/Cargo.toml index 67a68f5..19440dd 100644 --- a/crates/dashboard/Cargo.toml +++ b/crates/dashboard/Cargo.toml @@ -32,6 +32,11 @@ replay-viewer = ["dep:anyhow", "dep:distribution", "distribution/collector"] # SSE UI folding the vast.ai fleet telemetry and the orchestrator's # distribution snapshot server-side. live-collector = ["dep:distribution", "distribution/collector"] +# Datastream-backed dashboard: bind the demo's UDP telemetry sink, demux the +# per-node frames, and drive the live HTTP dashboard. Only needs the +# `distribution::datastream` module (not feature-gated there), so it stays free +# of the heavier collector/iroh stack. +datastream = ["dep:distribution"] [[bin]] name = "swactor-tui" @@ -43,6 +48,11 @@ name = "dashboard_collector" path = "src/bin/dashboard_collector.rs" required-features = ["live-collector"] +[[bin]] +name = "swactor-datastream-dashboard" +path = "src/bin/datastream_dashboard.rs" +required-features = ["datastream"] + [[example]] name = "replay_viewer" path = "examples/replay_viewer.rs" diff --git a/crates/dashboard/src/bin/datastream_dashboard.rs b/crates/dashboard/src/bin/datastream_dashboard.rs new file mode 100644 index 0000000..2188bcb --- /dev/null +++ b/crates/dashboard/src/bin/datastream_dashboard.rs @@ -0,0 +1,86 @@ +//! Datastream-backed dashboard: bind the demo cluster's UDP telemetry sink +//! (the same wire `swactor-datastream-collector` reads), demultiplex the +//! per-node frames, and serve the live HTTP/browser dashboard from them. +//! +//! Drop-in for the `collector` service in the datastream demo: same UDP bind, +//! but instead of printing frames it renders them in the existing dashboard UI. +//! +//! Usage: `swactor-datastream-dashboard [--bind HOST:PORT] [--port HTTP_PORT] [--node FILTER]` +//! --bind UDP sink to listen on (default `0.0.0.0:7700`) +//! --port HTTP dashboard port (default `9090`) +//! --node show the node whose id/region/role matches FILTER (default: first seen) + +use dashboard::datastream_source::run_datastream_ingest; +use dashboard::{start_dashboard, DashboardConfig}; +use tracing_subscriber::filter::LevelFilter; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::util::SubscriberInitExt; + +fn main() { + let args = Args::parse(); + + let dashboard = start_dashboard(DashboardConfig { + port: args.port, + ..Default::default() + }); + // Route process-output / membership tracing into the activity log panel. + // Cap at INFO so the panel shows our events, not tokio/mio TRACE internals. + tracing_subscriber::registry() + .with(dashboard.layer()) + .with(LevelFilter::INFO) + .init(); + dashboard.start_http_standalone(); + + eprintln!( + "datastream dashboard: serving on container port {} (UDP sink {}) — \ + browse via the published host port:\n \ + / Overview / Actors (single node)\n \ + /plugin/distribution SWIM connection graph (all nodes)\n \ + /plugin/vastai Fleet table (all nodes)", + args.port, args.bind + ); + + // Blocks forever, demuxing frames into the dashboard's pushed stats. + if let Err(e) = run_datastream_ingest(&args.bind, args.node.as_deref(), &dashboard) { + eprintln!("datastream dashboard: fatal: failed to bind {}: {e}", args.bind); + std::process::exit(1); + } +} + +struct Args { + bind: String, + port: u16, + node: Option, +} + +impl Args { + fn parse() -> Self { + let mut bind = std::env::var("SWACTOR_DATASTREAM_BIND").ok(); + let mut port: u16 = 9090; + let mut node: Option = None; + + let mut it = std::env::args().skip(1); + while let Some(arg) = it.next() { + match arg.as_str() { + "--bind" => bind = it.next(), + "--port" => port = it.next().and_then(|v| v.parse().ok()).unwrap_or(port), + "--node" => node = it.next(), + other => { + if let Some(v) = other.strip_prefix("--bind=") { + bind = Some(v.to_string()); + } else if let Some(v) = other.strip_prefix("--port=") { + port = v.parse().unwrap_or(port); + } else if let Some(v) = other.strip_prefix("--node=") { + node = Some(v.to_string()); + } + } + } + } + + Self { + bind: bind.unwrap_or_else(|| "0.0.0.0:7700".to_string()), + port, + node, + } + } +} diff --git a/crates/dashboard/src/datastream_source.rs b/crates/dashboard/src/datastream_source.rs new file mode 100644 index 0000000..a126653 --- /dev/null +++ b/crates/dashboard/src/datastream_source.rs @@ -0,0 +1,871 @@ +//! Datastream → dashboard adapter. +//! +//! Binds the UDP sink the demo cluster ships to (the same wire the dumb +//! `swactor-datastream-collector` reads), **demultiplexes** the per-node frames, +//! and drives the dashboard's *existing* views from them — no bespoke UI: +//! +//! * the single-node **Overview / Actors** page (`/`) via a synthesized +//! [`RuntimeStats`] (each datastream channel becomes one synthetic actor row); +//! * the canonical **Distribution** connection-graph page +//! (`/plugin/distribution`, [`crate::DISTRIBUTION_PAGE_HTML`]) via a +//! [`DistributionNodeSnapshot`] rebuilt from the selected node's membership — +//! so it renders the exact SWIM graph a live node shows; +//! * a cross-node **Fleet** table (`/plugin/vastai`) served in the same +//! dashboard chrome (nav bar + palette), not a separate app. +//! +//! Process output and membership transitions are emitted as `tracing` events so +//! they flow through the dashboard's activity-log path. +//! +//! A node is only shown while it is *live* (has shipped a frame within +//! [`NODE_TTL`]); a node that stops streaming drops out of every view, so a +//! restarted/departed node leaves no ghost in the graph or the fleet table. + +use std::collections::{HashMap, HashSet}; +use std::net::UdpSocket; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use distribution::datastream::catalog::{ + self, IdentityRecord, LifecycleCost, MembershipTransition, Record, ResourceSample, + RuntimeStats as DsRuntimeStats, TransportInternals, +}; +use distribution::datastream::wire::decode_delivery; +use distribution::snapshot::{DistributionNodeSnapshot, MemberInfo}; + +use swactor::actor::ActorAddress; +use swactor::stats::{ActorInfo, RuntimeStats, TickTiming, WorkerInfo}; + +use crate::plugin::{DashboardPlugin, PluginResponse}; +use crate::{DashboardHandle, DISTRIBUTION_PAGE_HTML}; + +/// A node counts as live — shown in the graph and fleet table — if it has +/// streamed a frame within this window. Nodes ship resource/runtime/transport +/// samples every ~1s, so a node silent past this has left the cluster; its +/// `models` entry lingers but is filtered out of every view (no ghost nodes). +const NODE_TTL: Duration = Duration::from_secs(8); + +/// Per-node accumulator: the latest value seen on each typed channel plus +/// running membership/process state. One of these drives the display. +#[derive(Default)] +struct DatastreamModel { + identity: Option, + resource: Option, + runtime: Option, + transport: Option, + lifecycle: Option, + /// peer node-id → latest liveness state. + membership: HashMap, + /// last membership transition, formatted for display. + last_transition: Option, + /// proc label → (line count, last line). + procs: HashMap, + first_seen: Option, + last_seen: Option, +} + +/// A log line to surface through the dashboard's activity path. +enum LogEvent { + Info(String), + Warn(String), +} + +impl DatastreamModel { + /// Fold one frame's channel/payload into the model, returning any activity + /// log events it produced (process output, membership transitions). + fn update(&mut self, channel: &str, payload: &[u8]) -> Vec { + let now = Instant::now(); + self.first_seen.get_or_insert(now); + self.last_seen = Some(now); + let mut events = Vec::new(); + + match channel { + catalog::IDENTITY => { + if let Ok(r) = IdentityRecord::decode(payload) { + self.identity = Some(r); + } + } + catalog::HOST_RESOURCE => { + if let Ok(r) = ResourceSample::decode(payload) { + self.resource = Some(r); + } + } + catalog::RUNTIME_STATS => { + if let Ok(r) = DsRuntimeStats::decode(payload) { + self.runtime = Some(r); + } + } + catalog::TRANSPORT_INTERNALS => { + if let Ok(r) = TransportInternals::decode(payload) { + self.transport = Some(r); + } + } + catalog::LIFECYCLE_COST => { + if let Ok(r) = LifecycleCost::decode(payload) { + self.lifecycle = Some(r); + } + } + catalog::MEMBERSHIP => { + if let Ok(t) = MembershipTransition::decode(payload) { + // Display the short id; key the membership map by the full + // id so the views can resolve it to a friendly label. + let line = format!("{}: {} → {}", short_id(&t.peer), t.from, t.to); + self.membership.insert(t.peer.clone(), t.to.clone()); + self.last_transition = Some(line.clone()); + events.push(LogEvent::Info(format!("membership {line}"))); + } + } + // Raw-text process output: `proc.