diff --git a/.deploy/deploy.example.toml b/.deploy/deploy.example.toml index 39404aa..0317946 100644 --- a/.deploy/deploy.example.toml +++ b/.deploy/deploy.example.toml @@ -3,7 +3,6 @@ # and fill in your machine details. [defaults] -dashboard_port = 9090 relay_port = 3340 # image = "swactor" # Required for --docker mode # container = "swactor" # Required for --docker mode diff --git a/Cargo.lock b/Cargo.lock index 1088bcd..df2d336 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -267,58 +267,6 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" -[[package]] -name = "axum" -version = "0.8.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" -dependencies = [ - "axum-core", - "bytes", - "form_urlencoded", - "futures-util", - "http", - "http-body", - "http-body-util", - "hyper", - "hyper-util", - "itoa", - "matchit", - "memchr", - "mime", - "percent-encoding", - "pin-project-lite", - "serde_core", - "serde_json", - "serde_path_to_error", - "serde_urlencoded", - "sync_wrapper", - "tokio", - "tower", - "tower-layer", - "tower-service", - "tracing", -] - -[[package]] -name = "axum-core" -version = "0.5.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" -dependencies = [ - "bytes", - "futures-core", - "http", - "http-body", - "http-body-util", - "mime", - "pin-project-lite", - "sync_wrapper", - "tower-layer", - "tower-service", - "tracing", -] - [[package]] name = "backon" version = "1.6.0" @@ -968,22 +916,6 @@ dependencies = [ "syn", ] -[[package]] -name = "dashboard" -version = "0.1.0" -dependencies = [ - "axum", - "crossbeam-queue", - "datastream", - "distribution", - "libc", - "serde", - "serde_json", - "swactor", - "tokio", - "tokio-stream", -] - [[package]] name = "dashmap" version = "6.2.1" @@ -2642,12 +2574,6 @@ dependencies = [ "regex-automata", ] -[[package]] -name = "matchit" -version = "0.8.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" - [[package]] name = "md5" version = "0.7.0" @@ -2723,7 +2649,6 @@ dependencies = [ name = "mvp-system" version = "0.1.0" dependencies = [ - "dashboard", "datastream", "distribution", "iroh", @@ -4661,17 +4586,6 @@ dependencies = [ "zmij", ] -[[package]] -name = "serde_path_to_error" -version = "0.1.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" -dependencies = [ - "itoa", - "serde", - "serde_core", -] - [[package]] name = "serde_spanned" version = "1.1.1" @@ -5432,7 +5346,6 @@ dependencies = [ "tokio", "tower-layer", "tower-service", - "tracing", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index fd2b0aa..f110cb8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,6 @@ members = [ ".", "crates/bindings/python", "crates/bindings/wasm-runtime", - "crates/dashboard", "crates/process", "crates/transport", "crates/distribution", diff --git a/apps/pipeline-parallel-inference/Cargo.lock b/apps/pipeline-parallel-inference/Cargo.lock index 0f89a26..48ca61b 100644 --- a/apps/pipeline-parallel-inference/Cargo.lock +++ b/apps/pipeline-parallel-inference/Cargo.lock @@ -150,58 +150,6 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" -[[package]] -name = "axum" -version = "0.8.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" -dependencies = [ - "axum-core", - "bytes", - "form_urlencoded", - "futures-util", - "http", - "http-body", - "http-body-util", - "hyper", - "hyper-util", - "itoa", - "matchit", - "memchr", - "mime", - "percent-encoding", - "pin-project-lite", - "serde_core", - "serde_json", - "serde_path_to_error", - "serde_urlencoded", - "sync_wrapper", - "tokio", - "tower", - "tower-layer", - "tower-service", - "tracing", -] - -[[package]] -name = "axum-core" -version = "0.5.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" -dependencies = [ - "bytes", - "futures-core", - "http", - "http-body", - "http-body-util", - "mime", - "pin-project-lite", - "sync_wrapper", - "tower-layer", - "tower-service", - "tracing", -] - [[package]] name = "backon" version = "1.6.0" @@ -690,22 +638,6 @@ dependencies = [ "syn", ] -[[package]] -name = "dashboard" -version = "0.1.0" -dependencies = [ - "axum", - "crossbeam-queue", - "datastream", - "distribution", - "libc", - "serde", - "serde_json", - "swactor", - "tokio", - "tokio-stream", -] - [[package]] name = "dashmap" version = "6.2.1" @@ -2279,12 +2211,6 @@ dependencies = [ "regex-automata", ] -[[package]] -name = "matchit" -version = "0.8.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" - [[package]] name = "md5" version = "0.7.0" @@ -2994,7 +2920,6 @@ name = "pipeline-parallel-inference" version = "0.1.0" dependencies = [ "base64", - "dashboard", "datastream", "distribution", "futures-util", @@ -3898,17 +3823,6 @@ dependencies = [ "zmij", ] -[[package]] -name = "serde_path_to_error" -version = "0.1.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" -dependencies = [ - "itoa", - "serde", - "serde_core", -] - [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -4216,6 +4130,7 @@ dependencies = [ "crossbeam-queue", "crossbeam-utils", "getrandom 0.2.17", + "parking_lot", "serde", ] @@ -4576,7 +4491,6 @@ dependencies = [ "tokio", "tower-layer", "tower-service", - "tracing", ] [[package]] diff --git a/apps/pipeline-parallel-inference/Cargo.toml b/apps/pipeline-parallel-inference/Cargo.toml index dea33ba..23416ce 100644 --- a/apps/pipeline-parallel-inference/Cargo.toml +++ b/apps/pipeline-parallel-inference/Cargo.toml @@ -5,7 +5,6 @@ name = "pipeline-parallel-inference" version = "0.1.0" edition = "2024" publish = false - [dependencies] swactor = { path = "../..", features = ["transport", "serde", "std"] } swactor-transport = { path = "../../crates/transport" } @@ -20,9 +19,6 @@ tokio = { version = "1", features = ["full"] } distribution = { path = "../../crates/distribution" } iroh-driver = { path = "../../crates/iroh-driver" } datastream = { path = "../../crates/datastream" } -# Live runtime dashboard (HTTP overview/actors/topology pages, served on -# localhost when PP_DASHBOARD is set). -dashboard = { path = "../../crates/dashboard" } iroh = "0.98" urlencoding = "2" base64 = "0.22" diff --git a/apps/pipeline-parallel-inference/scripts/demo-fleet.sh b/apps/pipeline-parallel-inference/scripts/demo-fleet.sh index 6c791e5..61a0fd7 100755 --- a/apps/pipeline-parallel-inference/scripts/demo-fleet.sh +++ b/apps/pipeline-parallel-inference/scripts/demo-fleet.sh @@ -1,22 +1,11 @@ #!/usr/bin/env bash -# demo-fleet.sh — one-command local mock of a vast.ai fleet, watchable live. +# demo-fleet.sh — one-command local mock of a vast.ai fleet. # # Brings up, from a single command, a self-contained demo of the production -# topology with NO off-box collector: the orchestrator runs locally, hosts the -# FULL swactor dashboard, AND hosts the fleet view in-process. Each stage ships -# its telemetry over the datastream (identity + host.resource frames) to the -# orchestrator's in-process FleetView sink; the orchestrator's Fleet tab renders -# it beside its own live actor / topology / distribution views. -# -# - pp-orchestrator on the HOST in --seed mode (PP_DASHBOARD on), spawning N -# pp-worker containers (one per stage) via docker-gpu-node.sh, each on -# --network host, and serving the full dashboard at http://127.0.0.1:9095/. -# It binds a UDP datastream sink (PP_FLEET_SINK) and folds every stage's -# frames into the live Fleet table. -# - Each stage pp-worker emits its fleet frames (~every few seconds) to that -# sink, so the Fleet tab animates in real time. -# - PP_HOLD=1 keeps the cluster up after the first drive, so the stages keep -# streaming and the dashboard stays live for inspection. +# topology with NO off-box collector. The orchestrator runs locally in seed +# mode and spawns N pp-worker containers via docker-gpu-node.sh on +# --network host. PP_HOLD=1 keeps the cluster up after the first drive for +# manual inspection. # # Ctrl+C (or any exit) tears everything down: stage containers, orchestrator, # and all temp files. @@ -34,10 +23,6 @@ # GPU gauges populate once a real GPU source is wired. # PP_PROMPT inference prompt (default: "fleet demo") # PP_MAX_TOKENS decode token cap (default: 4) -# PP_BIND_HOST dashboard + fleet-sink bind host (default: 127.0.0.1) -# PP_DASHBOARD_PORT orchestrator dashboard HTTP port (default: 9095) -# PP_FLEET_PORT orchestrator fleet UDP sink port (default: 9096) -# PP_NO_OPEN if set, don't try to open the dashboard in a browser # PP_STAGE_NETWORK docker network for stages (default: host) set -euo pipefail @@ -46,14 +31,6 @@ IMAGE="${PP_DIAG_IMAGE:-swactor-pp-gpu:latest}" BASE_IMAGE="${PP_BASE_IMAGE:-swactor-pp-base:cuda12.6}" PROMPT="${PP_PROMPT:-fleet demo}" MAX_TOKENS="${PP_MAX_TOKENS:-4}" -BIND_HOST="${PP_BIND_HOST:-127.0.0.1}" -DASH_PORT="${PP_DASHBOARD_PORT:-9095}" -FLEET_PORT="${PP_FLEET_PORT:-9096}" -CONTAINER_PREFIX="demo-fleet-stage" -RUN_ID="demo-fleet-$(date +%s)" -# The full swactor dashboard — including the in-process Fleet tab — is served by -# the orchestrator at "/". -DASH_URL="http://${BIND_HOST}:${DASH_PORT}/" if ! [[ "$NUM_STAGES" =~ ^[0-9]+$ ]] || [ "$NUM_STAGES" -lt 2 ]; then echo "demo-fleet: N must be an integer >= 2 (seed mode needs >=2 stages), got '$NUM_STAGES'" >&2 @@ -65,16 +42,6 @@ fi if ! docker info >/dev/null 2>&1; then echo "demo-fleet: docker daemon unreachable" >&2; exit 2 fi -# Fail loudly on a clash for the orchestrator dashboard port. Its HTTP server is -# spawned on the driver's tokio runtime and `.expect()`s its bind; a collision -# panics that task silently and the run carries on with no dashboard. Catch it -# here so the user can pick a free one. -if (exec 3<>"/dev/tcp/${BIND_HOST}/${DASH_PORT}") 2>/dev/null; then - exec 3>&- 3<&- - echo "demo-fleet: dashboard port ${DASH_PORT} is already in use." \ - "Pick a free one: PP_DASHBOARD_PORT=9097 $0 ${NUM_STAGES}" >&2 - exit 2 -fi SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" CRATE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" @@ -137,25 +104,18 @@ cleanup() { trap cleanup EXIT trap 'exit 130' INT TERM -# ── Step 3: orchestrator (hosts the full dashboard + fleet, holds cluster) ─ +# ── Step 3: orchestrator (holds cluster) ────────────────────────────────── # stdin is the FIFO; we hold its write end open on fd 3 so hold_open() never # sees EOF and the cluster stays up until we tear down. exec 3<>"$FIFO" echo "demo-fleet: launching orchestrator + ${NUM_STAGES} stage containers (run_id=$RUN_ID)" # The orchestrator (and, via inheritance, docker-gpu-node.sh) read these from -# the environment. PP_HOLD keeps the cluster up; PP_DASHBOARD makes the -# orchestrator host the full swactor dashboard locally; PP_FLEET_SINK is the -# UDP address the orchestrator binds for its in-process FleetView and that each -# stage ships its datastream frames to (reachable from the --network host stage -# containers via the shared loopback). +# the environment. PP_HOLD keeps the cluster up after the first drive. export PP_HOLD=1 export PP_WORKER_STUB=1 export PP_DEV=CPU export PP_IMAGE="$IMAGE" export PP_CONTAINER_PREFIX="$CONTAINER_PREFIX" -export PP_DASHBOARD=1 -export PP_DASHBOARD_PORT="$DASH_PORT" -export PP_FLEET_SINK="${BIND_HOST}:${FLEET_PORT}" [ -n "${PP_GPUS:-}" ] && export PP_GPUS [ -n "${PP_STAGE_NETWORK:-}" ] && export PP_STAGE_NETWORK "$ORCHESTRATOR_BIN" \ @@ -168,32 +128,7 @@ export PP_FLEET_SINK="${BIND_HOST}:${FLEET_PORT}" <"$FIFO" >"$ORCH_LOG" 2>&1 & ORCH_PID=$! -# ── Step 4: wait for the orchestrator's dashboard to bind, then announce + open -WAITED=0 -until (echo > "/dev/tcp/${BIND_HOST}/${DASH_PORT}") >/dev/null 2>&1; do - if ! kill -0 "$ORCH_PID" >/dev/null 2>&1; then - echo "demo-fleet: orchestrator exited before its dashboard came up." >&2 - tail -n 40 "$ORCH_LOG" >&2 || true - exit 1 - fi - WAITED=$((WAITED + 1)) - [ "$WAITED" -ge 30 ] && { echo "demo-fleet: orchestrator dashboard did not bind :${DASH_PORT} in 30s" >&2; tail -n 40 "$ORCH_LOG" >&2; exit 1; } - sleep 1 -done -echo -echo " ┌─────────────────────────────────────────────────────────────┐" -echo " │ Full swactor dashboard: $DASH_URL" -echo " │ (overview / actors / topology / distribution / netmap / fleet)" -echo " │ Fleet datastream sink: ${BIND_HOST}:${FLEET_PORT} (UDP, in-process)" -echo " └─────────────────────────────────────────────────────────────┘" -echo -if [ -z "${PP_NO_OPEN:-}" ]; then - if command -v xdg-open >/dev/null 2>&1; then (xdg-open "$DASH_URL" >/dev/null 2>&1 &) || true - elif command -v open >/dev/null 2>&1; then (open "$DASH_URL" >/dev/null 2>&1 &) || true - fi -fi - -# ── Step 5: wait until the cluster is converged + held open ──────────────── +# ── Step 4: wait until the cluster is converged + held open ──────────────── echo "demo-fleet: waiting for the cluster to converge (first inference drive)…" WAITED=0 until grep -q "holding cluster open" "$ORCH_LOG" 2>/dev/null; do @@ -208,10 +143,10 @@ until grep -q "holding cluster open" "$ORCH_LOG" 2>/dev/null; do sleep 1 done + RUNNING=$(docker ps -q --filter "name=^${CONTAINER_PREFIX}-[0-9]+$" | wc -l | tr -d ' ') echo -echo "demo-fleet: ✅ fleet up — ${RUNNING}/${NUM_STAGES} stage containers streaming real metrics." -echo "demo-fleet: watch live at $DASH_URL" +echo "demo-fleet: fleet up — ${RUNNING}/${NUM_STAGES} stage containers are running." echo "demo-fleet: press Ctrl+C to tear everything down." echo diff --git a/apps/pipeline-parallel-inference/scripts/docker-dashboard-e2e.sh b/apps/pipeline-parallel-inference/scripts/docker-dashboard-e2e.sh deleted file mode 100755 index 40faec3..0000000 --- a/apps/pipeline-parallel-inference/scripts/docker-dashboard-e2e.sh +++ /dev/null @@ -1,148 +0,0 @@ -#!/usr/bin/env bash -# docker-dashboard-e2e.sh — the docker-e2e run, held open under the live -# swactor dashboard, one dashboard PER STAGE. -# -# Brings up `N` stub-mode `pp-worker` containers on localhost and drives -# one InferenceRequest through them, exactly like `docker-e2e.sh` — but each -# stage serves the live swactor dashboard (PP_STAGE_DASHBOARD) and the -# orchestrator HOLDS after the drive (PP_HOLD). The stage containers run on -# `--network host`, so each stage's dashboard is reachable on the host at -# http://localhost: (BASE default 9100) -# i.e. stage 0 → 9100, stage 1 → 9101, … Each board shows that stage's -# StageActor + bridge actors and live message activity as tokens flow. -# -# This targets the STAGE runtimes deliberately: the orchestrator's own -# runtime is near-empty (it sends one request and waits), so there is nothing -# to see there — the actors that do the work live inside the stage processes. -# -# The cluster stays up until you press Enter in this terminal, at which point -# the orchestrator unwinds and tears everything down. -# -# Usage: -# apps/pipeline-parallel-inference/scripts/docker-dashboard-e2e.sh [N] -# -# Environment overrides: -# PP_STAGE_DASHBOARD_PORT_BASE base port; stage K serves BASE+K (default 9100) -# PP_IMAGE code image tag (default: swactor-pp-gpu:latest) -# PP_BASE_IMAGE base image tag (default: swactor-pp-base:cuda12.6) -# PP_CONTAINER_PREFIX container name prefix (default: pp-stage) -# PP_MAX_TOKENS max decode tokens (default: 4) -# PP_PROMPT inference prompt (default: "Say hello") -# PP_SKIP_BUILD skip cargo build (use existing target/) -# PP_SKIP_IMAGE_BUILD skip docker image build (use existing tag) -set -euo pipefail - -NUM_STAGES="${1:-3}" -PREFIX="${PP_CONTAINER_PREFIX:-pp-stage}" -MAX_TOKENS="${PP_MAX_TOKENS:-4}" -PROMPT="${PP_PROMPT:-Say hello}" -PORT_BASE="${PP_STAGE_DASHBOARD_PORT_BASE:-9100}" -ORCH_PORT="${PP_DASHBOARD_PORT:-9099}" -IMAGE="${PP_IMAGE:-swactor-pp-gpu:latest}" -BASE_IMAGE="${PP_BASE_IMAGE:-swactor-pp-base:cuda12.6}" - -if ! [[ "$NUM_STAGES" =~ ^[0-9]+$ ]] || [ "$NUM_STAGES" -lt 2 ]; then - echo "docker-dashboard-e2e: NUM_STAGES must be an integer >= 2, got '$NUM_STAGES'" >&2 - exit 2 -fi - -if ! command -v docker >/dev/null 2>&1; then - echo "docker-dashboard-e2e: docker not on PATH" >&2 - exit 2 -fi -if ! docker info >/dev/null 2>&1; then - echo "docker-dashboard-e2e: docker daemon unreachable" >&2 - exit 2 -fi - -# Fail loudly on a port clash for ANY stage port. The dashboard's HTTP server -# is spawned on the driver's tokio runtime and `.expect()`s its bind; a -# collision panics that task silently and the stage keeps running, so the -# browser just shows whatever already owns the port. Catch it here instead. -if (exec 3<>"/dev/tcp/127.0.0.1/${ORCH_PORT}") 2>/dev/null; then - exec 3>&- 3<&- - echo "docker-dashboard-e2e: orchestrator port ${ORCH_PORT} is already in use." \ - "Pick a free one: PP_DASHBOARD_PORT=9098 $0 ${NUM_STAGES}" >&2 - exit 2 -fi -for ((k = 0; k < NUM_STAGES; k++)); do - p=$((PORT_BASE + k)) - if (exec 3<>"/dev/tcp/127.0.0.1/${p}") 2>/dev/null; then - exec 3>&- 3<&- - echo "docker-dashboard-e2e: port ${p} (stage ${k}) is already in use." \ - "Pick a free base: PP_STAGE_DASHBOARD_PORT_BASE=9200 $0 ${NUM_STAGES}" >&2 - exit 2 - fi -done - -SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" -CRATE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" -WORKSPACE_DIR="$(cd "$CRATE_DIR/../.." && pwd)" -ORCHESTRATOR_BIN="$CRATE_DIR/target/release/pp-orchestrator" -WORKER_BIN="$CRATE_DIR/target/release/pp-worker" -WORKER_PY="$CRATE_DIR/pp_tinygrad_worker.py" - -# Step 1: build the release artifacts the docker image packages. -if [ -z "${PP_SKIP_BUILD:-}" ]; then - echo "docker-dashboard-e2e: building pp-worker + pp-orchestrator (release)" - cargo build --manifest-path "$CRATE_DIR/Cargo.toml" --release \ - --bin pp-worker --bin pp-orchestrator -fi -for f in "$ORCHESTRATOR_BIN" "$WORKER_BIN" "$WORKER_PY"; do - [ -f "$f" ] || { echo "docker-dashboard-e2e: missing $f" >&2; exit 1; } -done - -# Step 2: build the layered image (heavy CUDA base, then thin code layer). -# The stage dashboard lives in the pp-worker binary baked into this image, -# so a stale image without it will show nothing — rebuild unless you know the -# current image already carries the dashboard-enabled binary. -if [ -z "${PP_SKIP_IMAGE_BUILD:-}" ]; then - echo "docker-dashboard-e2e: building $BASE_IMAGE (base)" - docker build -f "$CRATE_DIR/Dockerfile.base" -t "$BASE_IMAGE" "$WORKSPACE_DIR" - echo "docker-dashboard-e2e: building $IMAGE (code)" - docker build -f "$CRATE_DIR/Dockerfile" \ - --build-arg "BASE_IMAGE=$BASE_IMAGE" -t "$IMAGE" "$WORKSPACE_DIR" -fi - -# Step 3: clean up stage containers from prior runs, and on exit (the -# orchestrator's ChainGuard kills its own children, but a Ctrl-C mid-run can -# leave strays). -cleanup_containers() { - local ids - ids=$(docker ps -aq --filter "name=^${PREFIX}-[0-9]+$" || true) - if [ -n "$ids" ]; then - # shellcheck disable=SC2086 - docker rm -f $ids >/dev/null 2>&1 || true - fi -} -cleanup_containers -trap cleanup_containers EXIT - -echo "docker-dashboard-e2e: dashboards will come up at:" -echo " orchestrator: http://localhost:${ORCH_PORT} (overview / actors / topology / distribution)" -for ((k = 0; k < NUM_STAGES; k++)); do - echo " stage ${k}: http://localhost:$((PORT_BASE + k))" -done - -# Step 4: drive pp-orchestrator with the docker shim. The orchestrator serves its -# own dashboard (PP_DASHBOARD) — including the live SWIM distribution graph and -# message tallies — and each stage serves its own (PP_STAGE_DASHBOARD). PP_HOLD -# makes the orchestrator block at the end, ticking the driver so the -# distribution view keeps updating. stdin/stdout stay on this terminal so the -# hold can read your Enter. -PP_WORKER_STUB=1 \ -PP_IMAGE="$IMAGE" \ -PP_CONTAINER_PREFIX="$PREFIX" \ -PP_DEV=CPU \ -PP_HOLD=1 \ -PP_DASHBOARD=1 \ -PP_DASHBOARD_PORT="$ORCH_PORT" \ -PP_STAGE_DASHBOARD=1 \ -PP_STAGE_DASHBOARD_PORT_BASE="$PORT_BASE" \ -"$ORCHESTRATOR_BIN" \ - --seed \ - --num-stages "$NUM_STAGES" \ - --gpu-node "$SCRIPT_DIR/docker-gpu-node.sh" \ - --worker "$WORKER_PY" \ - --prompt "$PROMPT" \ - --max-tokens "$MAX_TOKENS" diff --git a/apps/pipeline-parallel-inference/scripts/docker-gpu-node.sh b/apps/pipeline-parallel-inference/scripts/docker-gpu-node.sh index 409ed42..3cea65d 100755 --- a/apps/pipeline-parallel-inference/scripts/docker-gpu-node.sh +++ b/apps/pipeline-parallel-inference/scripts/docker-gpu-node.sh @@ -102,8 +102,6 @@ exec docker run --rm --init \ -e FIRST_PEER_DIRECT \ -e PP_BOOT_DELAY_STAGE \ -e PP_BOOT_DELAY_SECS \ - -e PP_STAGE_DASHBOARD \ - -e PP_STAGE_DASHBOARD_PORT_BASE \ -e PP_FLEET_SINK \ -e DEV="$DEV" \ -e WORKER_SCRIPT=/usr/local/share/pp_tinygrad_worker.py \ diff --git a/apps/pipeline-parallel-inference/src/bin/pp_orchestrator.rs b/apps/pipeline-parallel-inference/src/bin/pp_orchestrator.rs index 341573a..8b44253 100644 --- a/apps/pipeline-parallel-inference/src/bin/pp_orchestrator.rs +++ b/apps/pipeline-parallel-inference/src/bin/pp_orchestrator.rs @@ -29,31 +29,27 @@ //! 5. Kills any spawned child processes and (on `--vastai`) destroys all //! rented instances regardless of success or failure. + use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; -use datastream::{DATASTREAM_SINK_NAME, DatastreamSink}; +use datastream::transport::Delivery; +use datastream::{Consumer, DATASTREAM_SINK_NAME, DatastreamSink}; use distribution::node::DistributedNodeConfig; -use distribution::snapshot::DistributionNodeSnapshot; use distribution::registry::RegistryConfig; use distribution::swim::probe::SwimConfig; use iroh::{PublicKey, RelayMode, SecretKey}; use iroh_driver::IrohDriverConfig; - use swactor::actor::ActorAddress; use swactor::runtime::Inbox; use pipeline_parallel_inference::cluster::ClusterNode; - -use dashboard::datastream_source::{FleetView, distribution_cache_plugin, fleet_cache_plugin}; -use dashboard::{DashboardConfig, start_dashboard}; - use pipeline_parallel_inference::iroh_transport::{ ACTOR_ALPN, ActorMessagePump, IrohActorTransport, }; @@ -66,7 +62,6 @@ use pipeline_parallel_inference::orchestrator::{ use pipeline_parallel_inference::topology::{ENTRY_NAME, stage_name}; const ORCHESTRATOR_NAME: &str = "pp-orchestrator"; -type SharedSnapshot = Arc>>; fn node_config() -> DistributedNodeConfig { DistributedNodeConfig { @@ -380,50 +375,13 @@ fn check_child_death(guard: &mut ChainGuard) -> Result<(), String> { Ok(()) } -/// Register the Fleet tab and spawn the orchestrator-hosted datastream -/// consumer behind it: the [`DatastreamSink`] actor, which every node's -/// `ClusterFrameSink` resolves (under [`DATASTREAM_SINK_NAME`]) and ships its -/// telemetry to over the regular swactor transport (no dedicated channel). The -/// actor folds each delivery into an in-process `FleetView` and caches the -/// fleet JSON the tab serves. Returns the sink's address so the boot-phase -/// provisioner can ship the orchestrator's own (and each booting node's) frames -/// to it in-process. -/// -/// The actor is spawned here, but the [`DATASTREAM_SINK_NAME`] cluster-name -/// registration is deliberately NOT done here — see [`register_fleet_sink_name`] -/// and the [`ORCHESTRATOR_NAME`] registration: a name published before the -/// cluster is non-empty sizes its SWIM dissemination budget for a one-node -/// cluster and exhausts it before any stage can observe the entry via piggyback -/// gossip, so the stages never resolve the sink and the Fleet tab stays empty. -/// The caller must register the returned address post-convergence. -fn wire_fleet_sink( - cluster: &ClusterNode, - handle: Arc, -) -> Option { - let fleet_cache: Arc>> = Arc::new(Mutex::new(None)); - let dist_cache: Arc>> = Arc::new(Mutex::new(None)); - handle.register_plugin(fleet_cache_plugin(Arc::clone(&fleet_cache))); - handle.register_plugin(distribution_cache_plugin(Arc::clone(&dist_cache))); - - // Fold every received delivery into a FleetView, caching datastream-derived - // fleet/distribution JSON and pushing synthesized stats/activity into the - // dashboard. This keeps the dashboard off the local swactor runtime. - let mut view = FleetView::new(None); - let fleet_cache = Arc::clone(&fleet_cache); - let dist_cache = Arc::clone(&dist_cache); - let handle_for_updates = Arc::clone(&handle); +/// Spawn the local datastream consumer and return its actor address. The fold +/// keeps the raw stream in a datastream store without coupling this binary to +/// any presentation layer. +fn wire_datastream_sink(cluster: &ClusterNode) -> Option { + let mut consumer = Consumer::new(); let sink = DatastreamSink::new(move |stream, frame| { - let update = view.ingest(&stream, &frame); - *fleet_cache.lock().unwrap() = Some(update.fleet_json); - if let Some(dist_json) = update.dist_json { - *dist_cache.lock().unwrap() = Some(dist_json); - } - if let Some(stats) = update.stats { - handle_for_updates.set_stats(stats); - } - for (is_warn, message) in update.logs { - handle_for_updates.push_activity(is_warn, message); - } + let _ = consumer.accept(Delivery { stream, frame }); }); match cluster.rt.spawn(sink) { Ok(addr) => { @@ -437,20 +395,16 @@ fn wire_fleet_sink( } } -/// Publish the datastream-sink actor under [`DATASTREAM_SINK_NAME`] so the -/// stages can resolve it and ship their fleet telemetry. MUST be called -/// post-convergence (see [`wire_fleet_sink`]): registering it earlier would -/// size the SWIM dissemination budget for a one-node cluster and the entry -/// would exhaust its budget before any stage could observe it via piggyback -/// gossip — leaving the Fleet tab empty. Mirrors the [`ORCHESTRATOR_NAME`] -/// post-convergence registration. -fn register_fleet_sink_name(cluster: &ClusterNode, addr: Option) { +/// Publish the datastream sink after convergence so the name-dissemination +/// budget is sized for the real cluster, matching the orchestrator inbox. +fn register_datastream_sink_name(cluster: &ClusterNode, addr: Option) { if let Some(addr) = addr { cluster.register_name(DATASTREAM_SINK_NAME, addr); eprintln!("pp-orchestrator: datastream-sink registered -> {addr:?}"); } } + fn run_seed(args: &Args) -> i32 { let gpu_node_bin = resolve_gpu_node_path(args); if !gpu_node_bin.exists() { @@ -469,21 +423,6 @@ fn run_seed(args: &Args) -> i32 { return 1; } - // Build the dashboard handle up front. Runtime stats/distribution panels are - // fed only by datastream updates folded through `wire_fleet_sink`. - let dashboard = if std::env::var_os("PP_DASHBOARD").is_some() { - let port: u16 = std::env::var("PP_DASHBOARD_PORT") - .ok() - .and_then(|s| s.trim().parse().ok()) - .unwrap_or(9090); - let handle = Arc::new(start_dashboard(DashboardConfig { - port, - ..Default::default() - })); - Some((handle, port)) - } else { - None - }; let mut cluster = match ClusterNode::new( IrohDriverConfig { @@ -504,9 +443,6 @@ fn run_seed(args: &Args) -> i32 { } }; - // Keep the existing snapshot cache only for hold/prompt loop refresh paths; - // dashboard Distribution is fed from datastream-derived FleetView updates. - let want_dist = std::env::var_os("PP_DASHBOARD").is_some(); let my_id = cluster.node_id(); let my_hex: String = my_id.0.iter().map(|b| format!("{:02x}", b)).collect(); @@ -520,6 +456,7 @@ fn run_seed(args: &Args) -> i32 { "pp-orchestrator (--seed --num-stages {n}): orchestrator node {my_hex}, direct={direct:?}", n = args.num_stages, ); + let seed_datastream_sink_addr = wire_datastream_sink(&cluster); // Run inside a labelled block so every failure point can `break` // with both an exit code and a stable exit-reason string; the @@ -529,34 +466,6 @@ fn run_seed(args: &Args) -> i32 { let rt = Arc::clone(&cluster.rt); let router = Arc::clone(&cluster.transport_router); - // Shared distribution snapshot cell for hold-loop refresh paths. The - // dashboard's Distribution plugin is datastream-backed via FleetView. - let dist_cached: Option = if want_dist { - Some(Arc::new(Mutex::new(Some(cluster.snapshot())))) - } else { - None - }; - - // Datastream-sink actor address, set when the dashboard wires the Fleet - // tab. Its cluster-name publish is deferred to post-convergence (see - // `register_fleet_sink_name`). - let mut seed_fleet_sink_addr: Option = None; - - // When the in-process dashboard is on, register only datastream-backed - // plugins. FleetView updates fill both Fleet and Distribution caches. - if let Some((handle, port)) = &dashboard { - // Fleet tab ("Fleet" nav): the orchestrator-hosted datastream - // consumer. Each stage resolves the `datastream-sink` actor and - // ships its identity + host.resource frames over the swactor - // transport; the actor folds them into a live FleetView and serves - // the cross-node telemetry table beside the orchestrator's own views. - seed_fleet_sink_addr = wire_fleet_sink(&cluster, Arc::clone(handle)); - handle.start_http(cluster.driver.tokio_handle()); - eprintln!( - "pp-orchestrator: live dashboard on http://localhost:{port} \ - (overview / actors / topology / distribution / fleet)" - ); - } let response_inbox = match rt.new_inbox::() { Ok(i) => i, @@ -612,9 +521,7 @@ fn run_seed(args: &Args) -> i32 { // via SWIM piggyback gossip. Doing it post-convergence gives the registry // a budget sized for the real cluster. cluster.register_name(ORCHESTRATOR_NAME, inbox_addr); - // Publish the datastream-sink now, for the same budget reason: stages - // resolve this name to ship their fleet telemetry to the Fleet tab. - register_fleet_sink_name(&cluster, seed_fleet_sink_addr); + register_datastream_sink_name(&cluster, seed_datastream_sink_addr); eprintln!("pp-orchestrator: registered {ORCHESTRATOR_NAME} -> {inbox_addr:?}"); if let Err(e) = conv_res { eprintln!("pp-orchestrator: {e}"); @@ -752,14 +659,9 @@ fn run_seed(args: &Args) -> i32 { println!("{text}"); println!("============================================"); // The ChainGuard is still in scope here, so the stage - // containers stay up while we hold — letting the dashboard - // show a live, converged cluster rather than a torn-down one. - if dashboard.is_some() || std::env::var_os("PP_HOLD").is_some() { - hold_open( - &mut cluster, - dist_cached.as_ref(), - dashboard.as_ref().map(|(_, p)| *p), - ); + // containers stay up while we hold. + if std::env::var_os("PP_HOLD").is_some() { + hold_open(&mut cluster); } (0, "ok") } @@ -890,25 +792,11 @@ impl AwaitError { /// `dead`. When that happens, a `pp_drive_dead_member` diagnostic /// event is emitted identifying the stage and the dead member's /// `node_id_short` before returning [`AwaitError::ForwardPathDead`]. -/// Block the orchestrator after a successful drive so the live dashboard — -/// and the stage containers, whose `ChainGuard` is still in scope — stay up -/// for inspection. Returns when the operator presses Enter or closes stdin -/// (Ctrl-D), at which point the run unwinds and tears the cluster down. /// Hold the cluster open after a successful drive. The `ChainGuard` is still /// in scope (containers stay up), and we keep ticking the driver so SWIM stays -/// converged. The optional snapshot cache is retained for non-dashboard hold-loop -/// refresh paths; dashboard panels are updated through datastream FleetView. -/// Returns when the operator presses Enter or closes stdin (Ctrl-D). -fn hold_open(cluster: &mut ClusterNode, dist_cached: Option<&SharedSnapshot>, port: Option) { - match port { - Some(p) => eprintln!( - "pp-orchestrator: holding cluster open — orchestrator dashboard at \ - http://localhost:{p}. Press Enter (or Ctrl-D) to tear down." - ), - None => eprintln!( - "pp-orchestrator: holding cluster open. Press Enter (or Ctrl-D) to tear down." - ), - } +/// converged. Returns when the operator presses Enter or closes stdin (Ctrl-D). +fn hold_open(cluster: &mut ClusterNode) { + eprintln!("pp-orchestrator: holding cluster open. Press Enter (or Ctrl-D) to tear down."); // Read stdin on a side thread so the main thread can keep pumping the // driver; a blocking read here would freeze SWIM and the live snapshot. let stop = Arc::new(AtomicBool::new(false)); @@ -920,16 +808,10 @@ fn hold_open(cluster: &mut ClusterNode, dist_cached: Option<&SharedSnapshot>, po stop.store(true, Ordering::SeqCst); }); } - // Drain inbound ACTOR_ALPN traffic so fleet `DatastreamFrame`s from the - // stages reach the `datastream-sink` actor while the cluster is held open; - // `pump_once` only services SWIM/protocol gossip, not app messages. let msg_pump = ActorMessagePump::new(); while !stop.load(Ordering::SeqCst) { cluster.pump_once(); msg_pump.pump(&cluster.driver, &cluster.codecs, &cluster.rt); - if let Some(cached) = dist_cached { - *cached.lock().unwrap() = Some(cluster.snapshot()); - } std::thread::sleep(Duration::from_millis(200)); } } @@ -1084,13 +966,12 @@ fn drive_once( result } -/// Live multi-prompt loop for vast.ai mode (dashboard or `--hold`). Keeps the -/// cluster converged and the dashboard SSE fed while the operator drives more -/// prompts. A stdin-reader side thread feeds prompt lines so the main thread -/// can keep pumping the driver; each non-empty line drives one more inference at -/// the next `drive_seq`. A blank line, `quit`, or EOF (Ctrl-D) ends the loop, -/// after which the caller's finalize + teardown tail runs. -#[allow(clippy::too_many_arguments)] +/// Live multi-prompt loop for vast.ai `--hold` mode. Keeps the cluster +/// converged while the operator drives more prompts. A stdin-reader side thread +/// feeds prompt lines so the main thread can keep pumping the driver; each +/// non-empty line drives one more inference at the next `drive_seq`. A blank +/// line, `quit`, or EOF (Ctrl-D) ends the loop, after which the caller's +/// finalize + teardown tail runs. fn prompt_loop( cluster: &mut ClusterNode, response_inbox: &Inbox, @@ -1100,19 +981,11 @@ fn prompt_loop( label: &str, max_tokens: u32, first_drive_seq: u32, - dist_cached: Option<&SharedSnapshot>, - port: Option, ) { - match port { - Some(p) => eprintln!( - "pp-orchestrator: cluster live — dashboard at http://localhost:{p}. \ - Type a prompt + Enter to drive again; blank line / Ctrl-D / `quit` to tear down." - ), - None => eprintln!( - "pp-orchestrator: cluster live. Type a prompt + Enter to drive again; \ - blank line / Ctrl-D / `quit` to tear down." - ), - } + eprintln!( + "pp-orchestrator: cluster live. Type a prompt + Enter to drive again; \ + blank line / Ctrl-D / `quit` to tear down." + ); // Read prompts on a side thread so the main thread keeps pumping the driver; // a blocking stdin read here would freeze SWIM and the live snapshot. The @@ -1139,16 +1012,13 @@ fn prompt_loop( }); } - // Drain inbound ACTOR_ALPN traffic between drives so fleet telemetry keeps - // flowing into the `datastream-sink` while the operator is idle at the prompt. + // Drain inbound ACTOR_ALPN traffic between drives while the operator is + // idle at the prompt. let msg_pump = ActorMessagePump::new(); let mut drive_seq = first_drive_seq; while !stop.load(Ordering::SeqCst) { cluster.pump_once(); msg_pump.pump(&cluster.driver, &cluster.codecs, &cluster.rt); - if let Some(cached) = dist_cached { - *cached.lock().unwrap() = Some(cluster.snapshot()); - } // Drive any prompts that arrived since the last tick. drive_once pumps // the driver itself while awaiting each response. while let Ok(prompt) = rx.try_recv() { @@ -1427,21 +1297,6 @@ fn run_vastai(args: &Args) -> i32 { let num_stages = cluster.num_stages; let label = cluster.label.clone(); - // Build the dashboard handle up front. Runtime stats/distribution panels are - // fed only by datastream updates folded through `wire_fleet_sink`. - let dashboard = if std::env::var_os("PP_DASHBOARD").is_some() { - let port: u16 = std::env::var("PP_DASHBOARD_PORT") - .ok() - .and_then(|s| s.trim().parse().ok()) - .unwrap_or(9090); - let handle = Arc::new(start_dashboard(DashboardConfig { - port, - ..Default::default() - })); - Some((handle, port)) - } else { - None - }; let mut cluster_node = match ClusterNode::new( IrohDriverConfig { @@ -1462,10 +1317,6 @@ fn run_vastai(args: &Args) -> i32 { } }; - // The Distribution tab renders from the orchestrator's own SWIM view; its - // snapshot cell exists whenever the in-process dashboard is on. - let want_dist = std::env::var_os("PP_DASHBOARD").is_some(); - // Per-cluster drive counter, emitted on pp_drive_start / pp_drive_end so // the bundle reader can slice the interleaved event stream by attempt. // One-shot and --hold each drive exactly once per process, so this is 1. @@ -1511,47 +1362,9 @@ fn run_vastai(args: &Args) -> i32 { eprintln!("pp-orchestrator: no relay URL after 20s — vastai mode usually requires one"); } - // Per-stage host telemetry rides the swactor cluster transport: each rented - // stage resolves the `datastream-sink` actor and ships its frames there, - // and the orchestrator folds the booting node's SSH output into the same - // sink. `fleet_sink_addr` is filled when the dashboard wiring spawns the - // sink below; the provisioner ships boot frames to it in-process. - let mut fleet_sink_addr: Option = None; - - // ── Live dashboard + runtime — started BEFORE the lease ────────────── - // The lease + image-load phase is the slow, failure-prone part the operator - // most needs to watch, so the HTTP server binds here (start_http) rather - // than after convergence. Plugins populate as the cluster comes up. The - // ClusterNode already owns the runtime + codec + router (shared with the - // protocol actors); the dashboard handle was built up front above. let rt = Arc::clone(&cluster_node.rt); let router = Arc::clone(&cluster_node.transport_router); - // Shared distribution snapshot cell for prompt-loop refresh paths. The - // dashboard's Distribution plugin is datastream-backed via FleetView; this - // cache remains for existing loop bookkeeping and is skipped when off. - let dist_cached: Option = if want_dist { - Some(Arc::new(Mutex::new(Some(cluster_node.snapshot())))) - } else { - None - }; - - // When the in-process dashboard is on, register only datastream-backed - // plugins. FleetView updates fill both Fleet and Distribution caches. - if let Some((handle, port)) = &dashboard { - // Fleet tab: the orchestrator-hosted datastream consumer. Each rented - // stage resolves the `datastream-sink` actor and ships its telemetry - // over the swactor transport; the actor folds them into a live FleetView - // and serves the cross-node table. The returned address also receives - // each booting node's SSH output (boot phase) and the orchestrator's own - // frames, in-process. - fleet_sink_addr = wire_fleet_sink(&cluster_node, Arc::clone(handle)); - handle.start_http(cluster_node.driver.tokio_handle()); - eprintln!( - "pp-orchestrator: live dashboard on http://localhost:{port} \ - (overview / actors / topology / distribution / fleet)" - ); - } // ── Acquire the running cluster ────────────────────────────────── // Lease N fresh instances and (on --hold) persist the handle. @@ -1611,11 +1424,7 @@ fn run_vastai(args: &Args) -> i32 { num_stages, ); // Run the lease in 200ms slices instead of one blocking call, so the - // main thread can pump SWIM and refresh the dashboard snapshot while - // instances come up. Otherwise the membership/topology/net-map panels - // freeze at the empty startup snapshot for the entire (multi-minute, - // CDI-retrying) lease — even though stages are already joining SWIM — - // which defeats the point of binding the dashboard before the lease. + // main thread can pump SWIM while instances come up. let lease_result = { let mut lease_fut = Box::pin(pipeline_parallel_inference::vastai::lease_chain( &http, @@ -1646,9 +1455,6 @@ fn run_vastai(args: &Args) -> i32 { Ok(res) => break res, Err(_elapsed) => { cluster_node.pump_once(); - if let Some(cached) = dist_cached.as_ref() { - *cached.lock().unwrap() = Some(cluster_node.snapshot()); - } } } } @@ -1693,62 +1499,7 @@ fn run_vastai(args: &Args) -> i32 { }; eprintln!("pp-orchestrator: cluster contracts {contract_ids:?}"); - // ── Boot-phase telemetry (best-effort, opt-in) ─────────────────────── - // With a deploy SSH key configured (PP_DEPLOY_KEY) and the fleet sink live, - // SSH into each rented node and stream pp-worker's boot log onto the - // orchestrator's own datastream (proc.boot..*) until the node's - // swactor telemetry takes over the cluster transport. No deploy key → - // skipped; the container entrypoint still launches the worker, so the run is - // unchanged. This is the "ssh signal until the node runs swactor" half. - use pipeline_parallel_inference::{provision, vastai}; - if let (Some(sink_addr), Some(key_file)) = (fleet_sink_addr, provision::deploy_key_path()) { - match tokio_rt.block_on(vastai::list_instances_by_label( - &http, base_url, &api_key, &label, - )) { - Ok(list) => { - provision::install_boot_telemetry(&rt, &my_hex, 0, sink_addr); - match rt.spawn(provision::ProvisionActor::new( - rt.create_sender(), - cluster_node.driver.tokio_handle(), - )) { - Ok(prov_addr) => { - for (stage, &cid) in contract_ids.iter().enumerate() { - let Some(inst) = list.iter().find(|i| i.contract_id == cid) else { - continue; - }; - let host = if !inst.ssh_host.is_empty() { - inst.ssh_host.clone() - } else { - inst.public_ipaddr.clone() - }; - if host.is_empty() || inst.ssh_port == 0 { - eprintln!( - "pp-orchestrator: stage {stage} has no SSH endpoint yet; boot tail skipped" - ); - continue; - } - let _ = rt.send_to( - prov_addr, - provision::ProvisionMsg::TailStage { - stage: stage as u32, - ssh: provision::SshTarget { - host, - port: inst.ssh_port, - username: "root".to_string(), - key_file: key_file.clone(), - }, - }, - ); - } - } - Err(e) => eprintln!("pp-orchestrator: could not spawn ProvisionActor: {e}"), - } - } - Err(e) => eprintln!( - "pp-orchestrator: boot telemetry skipped (SSH endpoint discovery failed: {e})" - ), - } - } + let datastream_sink_addr = wire_datastream_sink(&cluster_node); // Drive the run inside a labelled block returning `(code, reason)` so // every failure point can name the reason it bailed; the orchestrator's @@ -1780,9 +1531,7 @@ fn run_vastai(args: &Args) -> i32 { "pp-orchestrator: waiting for SWIM convergence ({} alive peers, {}s budget)...", num_stages, orch_converge_secs, ); - // Drain inbound ACTOR_ALPN throughout convergence + pipeline wiring so - // each stage's fleet `DatastreamFrame`s reach the `datastream-sink` as it - // joins (pump_once only services SWIM/protocol gossip, not app messages). + // Drain inbound ACTOR_ALPN throughout convergence + pipeline wiring. let fleet_pump = ActorMessagePump::new(); let conv_res = await_convergence( num_stages as usize, @@ -1792,11 +1541,6 @@ fn run_vastai(args: &Args) -> i32 { cluster_node.pump_once(); fleet_pump.pump(&cluster_node.driver, &cluster_node.codecs, &cluster_node.rt); let snap = cluster_node.snapshot(); - // Keep the dashboard membership/topology panels live as peers - // join during convergence. - if let Some(cached) = dist_cached.as_ref() { - *cached.lock().unwrap() = Some(snap.clone()); - } snap.members.iter().filter(|m| m.state == "alive").count() }, ); @@ -1806,10 +1550,8 @@ fn run_vastai(args: &Args) -> i32 { } cluster_node.register_name(ORCHESTRATOR_NAME, inbox_addr); + register_datastream_sink_name(&cluster_node, datastream_sink_addr); eprintln!("pp-orchestrator: registered {ORCHESTRATOR_NAME} -> {inbox_addr:?}"); - // Publish the datastream-sink now, for the same budget reason: rented - // stages resolve this name to ship their fleet telemetry to the Fleet tab. - register_fleet_sink_name(&cluster_node, fleet_sink_addr); // Spec §4.5 + §4.6: gate the drive on (a) every pp-stage-K // resolvable and (b) pp-entry resolvable. Both are proxies for @@ -1828,10 +1570,7 @@ fn run_vastai(args: &Args) -> i32 { let (stage0_addr, stage0_node_id) = loop { cluster_node.pump_once(); fleet_pump.pump(&cluster_node.driver, &cluster_node.codecs, &cluster_node.rt); - // Keep the dashboard panels live while the pipeline wires up. - if let Some(cached) = dist_cached.as_ref() { - *cached.lock().unwrap() = Some(cluster_node.snapshot()); - } + // Keep app messages flowing while the pipeline wires up. for k in 0..num_stages { if roster_hex[k as usize].is_some() { continue; @@ -1920,13 +1659,9 @@ fn run_vastai(args: &Args) -> i32 { Err(e) => (1, e.exit_reason()), }; - // Live multi-prompt loop, gated on the dashboard or --hold. A plain - // one-shot (neither set) keeps today's single-drive-then-exit behaviour. - // Modeled on hold_open: a stdin-reader side thread feeds prompt lines - // while the main thread pumps the driver (~200 ms) and refreshes the - // distribution snapshot between prompts. Dashboard panels are fed by - // datastream FleetView updates; EOF / blank line / `quit` leaves the loop. - if std::env::var_os("PP_DASHBOARD").is_some() || args.hold { + // Live multi-prompt loop for held clusters. A plain one-shot keeps + // today's single-drive-then-exit behaviour. + if args.hold { prompt_loop( &mut cluster_node, &response_inbox, @@ -1936,8 +1671,6 @@ fn run_vastai(args: &Args) -> i32 { &label, args.max_tokens, drive_seq, - dist_cached.as_ref(), - dashboard.as_ref().map(|(_, p)| *p), ); } diff --git a/crates/dashboard/AGENTS.md b/crates/dashboard/AGENTS.md index 9d8175b..5454bc1 100644 --- a/crates/dashboard/AGENTS.md +++ b/crates/dashboard/AGENTS.md @@ -1,39 +1,11 @@ -# Datastream Dashboard — Agent Interface +# Dashboard crate contract -## Data Flow +Keep this crate read-only with respect to observed programs. -The dashboard is an HTTP/SSE consumer of datastream-derived models. Agents should inspect the browser endpoints and JSON plugin endpoints. +- It may ingest datastream frames. +- It may retain bounded raw-frame and view state for HTML/API rendering. +- It may host universal swactor runtime views. +- It must not send control signals to observed runtimes. +- It must not require changes outside `crates/dashboard` for dashboard-only work. -`datastream_source::FleetView` is the canonical fold from delivered frames to dashboard models. It produces: - -- fleet JSON for the node/fleet page -- distribution JSON for the distribution page -- `RuntimeStats` for the selected node overview and actors table -- activity messages for the event stream - -## HTTP Pages - -- `GET /` — selected node overview -- `GET /actors` — selected node actor rows -- `GET /topology` — topology derived from the latest selected-node stats -- `GET /plugin/distribution` — distribution graph and peer/cache state -- `GET /plugin/vastai` — fleet view -- `GET /events` — server-sent events for stats, activity, history, and plugin updates - -## JSON Endpoints - -- `GET /api/stats` — latest selected-node `RuntimeStats`, or `{}` before the first selected-node frame -- `GET /api/topology` — topology derived from latest stats, or `{}` -- `GET /api/history` — in-memory worker history -- `GET /api/logs` — retained activity events, optionally filtered by query params -- `GET /api/plugin/vastai` — current fleet JSON cache -- `GET /api/plugin/distribution` — current distribution JSON cache - -The same plugin names are used on the SSE stream for incremental browser -updates. Distribution page buttons post to `/api/plugin/distribution/rejoin` -and `/api/plugin/distribution/clear_status`; the datastream-backed plugin -acknowledges them as read-only no-ops. - -## Extension Rule - -Extend the dashboard through plugins backed by datastream-folded caches. A producer may add telemetry records, a fold may update shared JSON, and a plugin may serve that JSON/page over HTTP and SSE. +Main built-in view: `/view/swactor/workers`, backed by `runtime.stats`, `runtime.workers`, and `runtime.actors` frames when present. diff --git a/crates/dashboard/Cargo.lock b/crates/dashboard/Cargo.lock new file mode 100644 index 0000000..c403b5a --- /dev/null +++ b/crates/dashboard/Cargo.lock @@ -0,0 +1,849 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "base64ct" +version = "1.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" + +[[package]] +name = "bitflags" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" + +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + +[[package]] +name = "bytes" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "const-oid" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crossbeam-queue" +version = "0.3.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "curve25519-dalek" +version = "4.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" +dependencies = [ + "cfg-if", + "cpufeatures", + "curve25519-dalek-derive", + "digest", + "fiat-crypto", + "rustc_version", + "subtle", + "zeroize", +] + +[[package]] +name = "curve25519-dalek-derive" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "dashboard" +version = "0.1.0" +dependencies = [ + "axum", + "datastream", + "parking_lot", + "serde", + "serde_json", + "swactor", + "tokio", + "tokio-stream", +] + +[[package]] +name = "datastream" +version = "0.1.0" +dependencies = [ + "libc", + "serde", + "serde_json", + "swactor", + "swactor-transport", +] + +[[package]] +name = "der" +version = "0.7.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" +dependencies = [ + "const-oid", + "zeroize", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + +[[package]] +name = "ed25519" +version = "2.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53" +dependencies = [ + "pkcs8", + "serde", + "signature", +] + +[[package]] +name = "ed25519-dalek" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9" +dependencies = [ + "curve25519-dalek", + "ed25519", + "rand_core", + "serde", + "sha2", + "subtle", + "zeroize", +] + +[[package]] +name = "fiat-crypto" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "http" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "hyper" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", + "tower-service", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.186" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + +[[package]] +name = "memchr" +version = "2.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" + +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + +[[package]] +name = "mio" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "pkcs8" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" +dependencies = [ + "der", + "spki", +] + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + +[[package]] +name = "signature" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" +dependencies = [ + "rand_core", +] + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" + +[[package]] +name = "socket2" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "spki" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" +dependencies = [ + "base64ct", + "der", +] + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + +[[package]] +name = "swactor" +version = "0.1.0" +dependencies = [ + "crossbeam-queue", + "crossbeam-utils", + "getrandom", + "parking_lot", + "serde", +] + +[[package]] +name = "swactor-transport" +version = "0.1.0" +dependencies = [ + "ed25519-dalek", + "rand_core", + "serde", + "serde_json", + "swactor", +] + +[[package]] +name = "syn" +version = "2.0.118" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + +[[package]] +name = "tokio" +version = "1.52.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe" +dependencies = [ + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-stream" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "log", + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + +[[package]] +name = "zmij" +version = "1.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" diff --git a/crates/dashboard/Cargo.toml b/crates/dashboard/Cargo.toml index 4233fd6..6e07886 100644 --- a/crates/dashboard/Cargo.toml +++ b/crates/dashboard/Cargo.toml @@ -1,19 +1,16 @@ +[workspace] + [package] name = "dashboard" version = "0.1.0" edition = "2024" [dependencies] +axum = "0.8" +datastream = { path = "../datastream" } swactor = { path = "../..", features = ["serde"] } +parking_lot = "0.12" serde = { version = "1", features = ["derive"] } serde_json = "1" -axum = "0.8" -tokio = { version = "1", features = ["net", "rt-multi-thread", "sync", "time"] } +tokio = { version = "1", features = ["net", "rt-multi-thread", "sync"] } tokio-stream = "0.1" -crossbeam-queue = "0.3.12" -datastream = { path = "../datastream" } -distribution = { path = "../distribution" } - -[target.'cfg(target_os = "linux")'.dependencies] -libc = "0.2" - diff --git a/crates/dashboard/README.md b/crates/dashboard/README.md index 4c1783f..fc6176f 100644 --- a/crates/dashboard/README.md +++ b/crates/dashboard/README.md @@ -1,46 +1,16 @@ # dashboard -Datastream-only HTTP dashboard for visualizing swactor-derived telemetry in a browser. The dashboard consumes folded datastream records and serves live pages over HTTP/SSE. +Read-only HTML/SSE dashboard over incoming datastream frames. -## Features +The crate owns the Axum server, bounded raw frame window, and view registry. Component crates can keep their own view implementations beside their code and register them through `DashboardHandle::register_view`. The built-in swactor worker page is hosted here because worker/actor/message processing is universal to swactor programs. -| Feature | Default | Description | -|---------|---------|-------------| -| `distribution` | yes | `/distribution` page with SWIM membership, gossip directory routes, peer auth, and location cache data derived from datastream frames | -| Fleet view | yes | `/vastai` page showing nodes folded by `datastream_source::FleetView` | +## Routes -## HTTP Dashboard +- `GET /` — dashboard index +- `GET /events` — raw incoming frames as SSE +- `GET /api/frames` — recent raw frame window +- `GET /api/views` — registered view metadata +- `GET /view/swactor/workers` — built-in worker page +- `GET /api/view/swactor/workers` — worker page JSON snapshot -The dashboard is embedded by an application that owns a datastream sink. The -sink folds delivered frames through `datastream_source::FleetView`, then pushes -the resulting stats, activity lines, and cache-backed plugin JSON into the -dashboard handle. - -Pages: -- `http://localhost:9090/` — live overview from the selected datastream node -- `http://localhost:9090/actors` — actor table reconstructed from actor telemetry records -- `http://localhost:9090/plugin/distribution` — SWIM membership, gossip directory routes, peer auth, and cache entries -- `http://localhost:9090/plugin/vastai` — fleet/node view fed by the shared fleet cache - -The dashboard model is folded by `datastream_source::FleetView`. Producers -publish telemetry records to datastream channels; the dashboard sink folds those -records into cached JSON, pushes activity messages, and updates the HTTP/SSE -views. - -## Public API - -Embed the dashboard by constructing `DashboardConfig` and calling -`start_dashboard(config)`. The returned handle owns the HTTP server state and -supports externally pushed stats, activity messages, history access, plugin -registration, landing page overrides, extra routers, and shutdown. - -Plugins are the extension boundary. New dashboard surfaces should register a -`DashboardPlugin` or use a cache-backed plugin such as -`fleet_cache_plugin(cache)` / `distribution_cache_plugin(cache)`, then feed it -from datastream-derived JSON caches. - -## Pipeline app - -`apps/pipeline-parallel-inference` enables the dashboard with `PP_DASHBOARD=1`. -Its orchestrator hosts the HTTP server, spawns the `datastream-sink` actor, and -feeds every dashboard view from `FleetView` updates. +All state is derived from observed frames. The dashboard sends no control signals back to producers. diff --git a/crates/dashboard/src/bin/swactor_dummy_node.rs b/crates/dashboard/src/bin/swactor_dummy_node.rs new file mode 100644 index 0000000..20de4fb --- /dev/null +++ b/crates/dashboard/src/bin/swactor_dummy_node.rs @@ -0,0 +1,400 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +use dashboard::swactor::{RUNTIME_ACTORS, RUNTIME_STATS, RUNTIME_WORKERS}; +use dashboard::{DashboardConfig, DashboardHandle, start_dashboard}; +use datastream::frame::{ChannelId, Frame, Lifetime, NodeId, Position, StreamId}; +use parking_lot::Mutex; +use serde::Serialize; +use serde_json::{Value, json}; +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; +use swactor::config::RuntimeConfig; +use swactor::runtime::Runtime; +use swactor::stats::{ActorSnapshot, StatsHook}; + +const NODE_ID: &str = "dashboard-swactor-dummy"; +const WORKER_ACTORS: usize = 12; +const PUBLISH_INTERVAL: Duration = Duration::from_millis(250); +const PULSE_INTERVAL: Duration = Duration::from_millis(25); +const WORK_ITEM_DELAY: Duration = Duration::from_micros(200); + +#[derive(Clone)] +struct PulseTick { + seq: u64, +} + +#[derive(Clone)] +struct WorkItem { + seq: u64, + route: u32, + hops_left: u8, +} + +#[derive(Clone)] +enum RouterMsg { + Configure { workers: Vec }, + Beat { seq: u64 }, + Complete { worker: u32, seq: u64, route: u32 }, +} + +struct PulseActor { + router: ActorAddress, +} + +impl ActorInterface for PulseActor { + type Incoming = PulseTick; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: PulseTick) { + let _ = ctx.send(self.router, RouterMsg::Beat { seq: msg.seq }); + } +} + +struct RouterActor { + workers: Vec, + next: usize, + completed: u64, +} + +impl RouterActor { + fn new() -> Self { + Self { + workers: Vec::new(), + next: 0, + completed: 0, + } + } +} + +impl ActorInterface for RouterActor { + type Incoming = RouterMsg; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: RouterMsg) { + match msg { + RouterMsg::Configure { workers } => { + self.workers = workers; + self.next = 0; + } + RouterMsg::Beat { seq } => { + if self.workers.is_empty() { + return; + } + + let burst = 48 + (seq as usize % 32); + for route in 0..burst { + let target = self.workers[self.next % self.workers.len()]; + self.next = self.next.wrapping_add(1); + let _ = ctx.send( + target, + WorkItem { + seq, + route: route as u32, + hops_left: 1 + ((seq + route as u64) % 3) as u8, + }, + ); + } + } + RouterMsg::Complete { worker, seq, route } => { + self.completed = self.completed.wrapping_add(1); + + if self.completed % 7 == 0 && !self.workers.is_empty() { + let target = + self.workers[(worker as usize + route as usize) % self.workers.len()]; + let _ = ctx.send( + target, + WorkItem { + seq, + route: route.wrapping_add(1000), + hops_left: 1, + }, + ); + } + } + } + } +} + +struct WorkerActor { + id: u32, + router: ActorAddress, +} + +impl ActorInterface for WorkerActor { + type Incoming = WorkItem; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: WorkItem) { + if msg.hops_left > 0 { + let _ = ctx.send( + ctx.self_addr(), + WorkItem { + seq: msg.seq, + route: msg.route, + hops_left: msg.hops_left - 1, + }, + ); + return; + } + + thread::sleep(WORK_ITEM_DELAY); + let _ = ctx.send( + self.router, + RouterMsg::Complete { + worker: self.id, + seq: msg.seq, + route: msg.route, + }, + ); + } +} + +struct QueuedSinkActor; + +impl ActorInterface for QueuedSinkActor { + type Incoming = WorkItem; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, _msg: WorkItem) { + ctx.suspend_self(); + } +} + +#[derive(Default)] +struct DashboardStatsHook { + actors: Mutex>, +} + +#[derive(Clone)] +struct ActorDetail { + worker_id: usize, + mailbox_depth: usize, + last_msg_type: Option, + messages_processed: u64, + poisoned: bool, + message_type_counts: Vec<(String, u64)>, +} + +impl StatsHook for DashboardStatsHook { + fn on_tick(&self, worker_id: usize, snapshots: &[ActorSnapshot]) { + let mut actors = self.actors.lock(); + for snapshot in snapshots { + actors.insert( + snapshot.address, + ActorDetail { + worker_id, + mailbox_depth: snapshot.mailbox_depth, + last_msg_type: snapshot.last_msg_type.map(str::to_owned), + messages_processed: snapshot.messages_processed, + poisoned: snapshot.poisoned, + message_type_counts: snapshot + .message_type_counts + .iter() + .map(|(name, count)| ((*name).to_owned(), *count)) + .collect(), + }, + ); + } + } +} + +impl DashboardStatsHook { + fn snapshot( + &self, + live_workers: &[(ActorAddress, usize)], + names: &HashMap, + ) -> Vec { + let actors = self.actors.lock(); + live_workers + .iter() + .map(|(address, worker_id)| { + let detail = actors.get(address); + ActorDetailFrame { + address: address.to_string(), + name: names.get(address).cloned(), + worker_id: detail.map_or(*worker_id, |detail| detail.worker_id), + mailbox_depth: detail.map_or(0, |detail| detail.mailbox_depth), + last_msg_type: detail.and_then(|detail| detail.last_msg_type.clone()), + messages_processed: detail.map_or(0, |detail| detail.messages_processed), + poisoned: detail.is_some_and(|detail| detail.poisoned), + message_type_counts: detail + .map(|detail| detail.message_type_counts.clone()) + .unwrap_or_default(), + } + }) + .collect() + } +} + +#[derive(Serialize)] +struct ActorDetailFrame { + address: String, + name: Option, + worker_id: usize, + mailbox_depth: usize, + last_msg_type: Option, + messages_processed: u64, + poisoned: bool, + message_type_counts: Vec<(String, u64)>, +} + +fn main() { + let dashboard = start_dashboard(DashboardConfig::default()); + dashboard.start_http_standalone(); + + let mut runtime = Runtime::new(RuntimeConfig { + num_threads: 4, + max_actors: 128, + channel_buffer_size: 4096, + actor_message_budget: 8, + }); + let stats_hook = Arc::new(DashboardStatsHook::default()); + runtime.set_stats_hook(stats_hook.clone()); + + let router = runtime.spawn(RouterActor::new()).expect("spawn router"); + let mut names = HashMap::new(); + names.insert(router, "router".to_owned()); + + let mut workers = Vec::with_capacity(WORKER_ACTORS); + for id in 0..WORKER_ACTORS { + let address = runtime + .spawn(WorkerActor { + id: id as u32, + router, + }) + .expect("spawn worker actor"); + names.insert(address, format!("worker-{id}")); + workers.push(address); + } + + let pulse = runtime.spawn(PulseActor { router }).expect("spawn pulse"); + names.insert(pulse, "pulse".to_owned()); + let queued_sink = runtime.spawn(QueuedSinkActor).expect("spawn queued sink"); + names.insert(queued_sink, "queued-sink".to_owned()); + runtime + .send_to( + router, + RouterMsg::Configure { + workers: workers.clone(), + }, + ) + .expect("configure router"); + + let runtime = runtime.run().expect("start swactor runtime"); + let stream = StreamId::new(NodeId::new(NODE_ID), Lifetime(1)); + let mut position = 0_u64; + let mut seq = 0_u64; + let mut ticks_until_publish = 0_u8; + + println!( + "dashboard listening at http://127.0.0.1:{}/view/swactor/workers", + DashboardConfig::default().port + ); + println!( + "dummy node {NODE_ID} running {} swactor actors", + names.len() + ); + + loop { + let _ = runtime.runtime.send_to(pulse, PulseTick { seq }); + if seq % 2 == 0 { + let _ = runtime.runtime.send_to( + queued_sink, + WorkItem { + seq, + route: u32::MAX, + hops_left: 0, + }, + ); + } + seq = seq.wrapping_add(1); + + if ticks_until_publish == 0 { + publish_runtime_snapshot( + &dashboard, + &stream, + &mut position, + &runtime.runtime, + &stats_hook, + &names, + ); + ticks_until_publish = (PUBLISH_INTERVAL.as_millis() / PULSE_INTERVAL.as_millis()) as u8; + } + ticks_until_publish = ticks_until_publish.saturating_sub(1); + + thread::sleep(PULSE_INTERVAL); + } +} + +fn publish_runtime_snapshot( + dashboard: &DashboardHandle, + stream: &StreamId, + position: &mut u64, + runtime: &Runtime, + stats_hook: &DashboardStatsHook, + names: &HashMap, +) { + let stats = runtime.stats(); + let actor_details = stats_hook.snapshot(&stats.actors, names); + let actors: Vec = stats + .actors + .iter() + .map(|(address, worker_id)| json!([address.to_string(), worker_id])) + .collect(); + let workers = serde_json::to_value(&stats.workers).expect("serialize worker stats"); + let actor_details = serde_json::to_value(actor_details).expect("serialize actor stats"); + let tick_timings = serde_json::to_value(&stats.tick_timings).expect("serialize tick timings"); + let total_mailbox_depth: usize = stats + .workers + .iter() + .map(|worker| worker.mailbox_depth) + .sum(); + + publish_json( + dashboard, + stream, + position, + RUNTIME_STATS, + json!({ + "num_workers": stats.num_workers, + "uptime_ms": stats.uptime_ms, + "actors_live": stats.actors.len(), + "mailbox_depth": total_mailbox_depth, + "actors": actors, + "workers": workers, + "actor_details": actor_details, + "tick_timings": tick_timings, + }), + ); + + publish_json( + dashboard, + stream, + position, + RUNTIME_WORKERS, + json!({ "workers": stats.workers }), + ); + + publish_json( + dashboard, + stream, + position, + RUNTIME_ACTORS, + json!({ "actors": actor_details }), + ); +} + +fn publish_json( + dashboard: &DashboardHandle, + stream: &StreamId, + position: &mut u64, + channel: &str, + value: Value, +) { + let payload = serde_json::to_vec(&value).expect("serialize dashboard frame"); + let frame = Frame::new(ChannelId::new(channel), Position(*position), payload); + dashboard.ingest(stream, &frame); + *position = position.wrapping_add(1); +} diff --git a/crates/dashboard/src/datastream_source.rs b/crates/dashboard/src/datastream_source.rs deleted file mode 100644 index c17310a..0000000 --- a/crates/dashboard/src/datastream_source.rs +++ /dev/null @@ -1,1008 +0,0 @@ -//! Datastream → dashboard adapter. -//! -//! Binds the UDP sink used by demo clusters, **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 -//! distribution-page JSON rebuilt from the selected node's membership and -//! `dist.state` — 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 dashboard activity -//! events by the datastream host. -//! -//! 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::sync::{Arc, Mutex}; -use std::time::{Duration, Instant}; - -use datastream::Record; -use datastream::frame::{Frame, StreamId}; -use datastream::health::{DATASTREAM_HEALTH, DatastreamHealth}; -use distribution::telemetry::{ - DIST_STATE, DistributionState, MEMBERSHIP, MembershipTransition, TRANSPORT_INTERNALS, - TransportInternals, -}; - -use crate::telemetry::{ - ActorRec, ActorRuntimeDetail, HOST_RESOURCE, IDENTITY, IdentityRecord, RUNTIME_ACTORS, - RUNTIME_STATS, RUNTIME_WORKERS, ResourceSample, RuntimeStats as DsRuntimeStats, WorkerCounters, -}; - -use swactor::actor::ActorAddress; -use swactor::stats::{ActorInfo, RuntimeStats, TickTiming, WorkerInfo}; - -use crate::plugin::{DashboardPlugin, PluginResponse}; - -/// 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, - /// Consolidated distribution-subsystem state (cache/registry/directory/...). - dist_state: Option, - /// Per-actor runtime detail (the real actor table). - actor_detail: Option, - /// Aggregated worker-runtime counters (routing/error tallies + tick timing). - worker_counters: Option, - /// Datastream self-health (mux assigned/dropped + loss rate). - datastream_health: Option, - /// peer node-id → latest liveness state. - membership: HashMap, - /// peer node-id → cause of its most recent liveness transition (the SWIM - /// observer's reason string; the value-add of W3/W4 over the state-diff). - membership_reason: HashMap, - /// last membership transition, formatted for display (with its cause). - 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 { - IDENTITY => { - if let Ok(r) = IdentityRecord::decode(payload) { - self.identity = Some(r); - } - } - HOST_RESOURCE => { - if let Ok(r) = ResourceSample::decode(payload) { - self.resource = Some(r); - } - } - RUNTIME_STATS => { - if let Ok(r) = DsRuntimeStats::decode(payload) { - self.runtime = Some(r); - } - } - TRANSPORT_INTERNALS => { - if let Ok(r) = TransportInternals::decode(payload) { - self.transport = Some(r); - } - } - DIST_STATE => { - if let Ok(r) = DistributionState::decode(payload) { - self.dist_state = Some(r); - } - } - RUNTIME_ACTORS => { - if let Ok(r) = ActorRuntimeDetail::decode(payload) { - self.actor_detail = Some(r); - } - } - RUNTIME_WORKERS => { - if let Ok(r) = WorkerCounters::decode(payload) { - self.worker_counters = Some(r); - } - } - DATASTREAM_HEALTH => { - if let Ok(r) = DatastreamHealth::decode(payload) { - self.datastream_health = Some(r); - } - } - 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. Carry - // the observer's cause string through to the display — it is - // the whole value-add of the production observer over the old - // state-diff (which only ever knew *that* a peer changed). - let line = if t.reason.is_empty() { - format!("{}: {} → {}", short_id(&t.peer), t.from, t.to) - } else { - format!( - "{}: {} → {} ({})", - short_id(&t.peer), - t.from, - t.to, - t.reason - ) - }; - self.membership.insert(t.peer.clone(), t.to.clone()); - if !t.reason.is_empty() { - self.membership_reason - .insert(t.peer.clone(), t.reason.clone()); - } - self.last_transition = Some(line.clone()); - events.push(LogEvent::Info(format!("membership {line}"))); - } - } - // Raw-text process output: `proc.