From b95163823eafb56f7c21916c8f6055d7e21fac66 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Mon, 24 Aug 2026 01:28:38 +0400 Subject: [PATCH] feat(telemetry): add sequenced runtime dashboard signals Rate-limit actor census and activity telemetry while preserving immediate lifecycle transitions. Route managed-process output into Fleet node tails, bound explorer retention, aggregate SWIM probes, and make sampler health transition-based. Restore stable explorer synchronization and refine Fleet bulk controls. --- Cargo.lock | 1 + apps/myelin/src/node/worker_node_runtime.rs | 148 +++++++- apps/myelin/src/orchestration/app.rs | 5 + apps/myelin/src/orchestration/control.rs | 32 ++ .../src/orchestration/distribution_stack.rs | 33 +- .../myelin/src/orchestration/fleet_control.js | 356 +++++++++++++----- .../src/orchestration/manual_control.rs | 23 ++ crates/dashboard/AGENTS.md | 2 +- crates/dashboard/README.md | 11 +- crates/dashboard/src/control_plane.rs | 316 ++++++++++++++-- crates/dashboard/src/control_plane_page.html | 35 +- crates/dashboard/src/live_explorer.rs | 167 ++++++-- crates/dashboard/src/live_explorer_page.html | 38 +- crates/dashboard/src/swactor/actor_view.rs | 250 +++++++++++- crates/distribution/src/swim/telemetry.rs | 110 ++++-- crates/distribution/src/telemetry.rs | 23 ++ crates/distribution/tests/telemetry.rs | 24 +- crates/job-runner/Cargo.toml | 1 + crates/job-runner/src/node.rs | 18 +- .../process/SWACTOR_MANAGED_PROCESS_SPEC.md | 31 +- crates/process/src/actor.rs | 10 + crates/process/src/lifecycle.rs | 94 +++-- crates/process/src/message.rs | 2 + crates/process/src/supervisor.rs | 96 ++++- crates/process/tests/public_api_stage2.rs | 51 ++- crates/telemetry/TELEMETRY_SPEC.md | 30 +- crates/telemetry/src/endpoint.rs | 286 ++++++++++++-- .../telemetry/tests/t_telemetry_endpoint.rs | 18 +- src/delivery.rs | 3 +- src/stats.rs | 44 ++- src/worker.rs | 146 +++++-- xtask/src/demo/provider.rs | 1 + 32 files changed, 2017 insertions(+), 388 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e4c37d6..2b70208 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4368,6 +4368,7 @@ dependencies = [ "swactor-process", "swactor-transport", "tar", + "telemetry", "tempfile", "tokio", ] diff --git a/apps/myelin/src/node/worker_node_runtime.rs b/apps/myelin/src/node/worker_node_runtime.rs index 2595886..6c8eff7 100644 --- a/apps/myelin/src/node/worker_node_runtime.rs +++ b/apps/myelin/src/node/worker_node_runtime.rs @@ -58,7 +58,7 @@ use parking_lot::Mutex; use serde_json::{Value, json}; use swactor::actor::{ActorAddress, ActorInterface}; use swactor::runtime::{Ctx, ExternalSender, Inbox, Runtime}; -use swactor::stats::{ActorSnapshot, StatsHook}; +use swactor::stats::{ActorSnapshot, StatsHook, StatsSnapshotKind}; use swactor_engine::{ActorCompletion, Engine, EngineHandle, TokioBackend, TokioConfig}; use swactor_job_runner::{NodeJobActor, register_job_codecs}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; @@ -209,7 +209,7 @@ struct InspectableStatsHook { } impl StatsHook for InspectableStatsHook { - fn on_tick(&self, worker_id: usize, snapshots: &[ActorSnapshot]) { + fn on_snapshot(&self, worker_id: usize, snapshots: &[ActorSnapshot], kind: StatsSnapshotKind) { self.inspector.latest.lock().insert( worker_id, snapshots @@ -222,7 +222,7 @@ impl StatsHook for InspectableStatsHook { }) .collect(), ); - self.inner.on_tick(worker_id, snapshots); + self.inner.on_snapshot(worker_id, snapshots, kind); } } @@ -599,6 +599,13 @@ impl SamplerHealthContext { } } +#[derive(Clone, Copy, PartialEq, Eq)] +enum SamplerHealthState { + Waiting, + Ready, + Failed, +} + fn sampler_health_payload( context: SamplerHealthContext, sampler: &str, @@ -681,18 +688,33 @@ fn submit_sampler_sample_health( producer: &TelemetryProducer, health_channel: ChannelId, context: SamplerHealthContext, - sampler: &str, - sample_channel: &str, + identity: (&str, &str), seq: u64, error: Option<&str>, + state: &Mutex, ) { - let (status, detail) = match error { - Some(error) => ( + let (sampler, sample_channel) = identity; + let mut state = state.lock(); + let (status, detail, next) = match (*state, error) { + (SamplerHealthState::Waiting, None) => ( + "ready", + json!({"state":"sample_observed","sample_seq":seq}), + SamplerHealthState::Ready, + ), + (SamplerHealthState::Failed, None) => ( + "recovered", + json!({"state":"sample_observed","sample_seq":seq}), + SamplerHealthState::Ready, + ), + (SamplerHealthState::Waiting | SamplerHealthState::Ready, Some(error)) => ( "failed", json!({"state":"error","sample_seq":seq,"error":error}), + SamplerHealthState::Failed, ), - None => ("ready", json!({"state":"sample_observed","sample_seq":seq})), + (SamplerHealthState::Ready, None) | (SamplerHealthState::Failed, Some(_)) => return, }; + *state = next; + drop(state); submit_sampler_health( producer, health_channel, @@ -738,6 +760,7 @@ fn spawn_blocking_sampler_task( error_of, } = config; let started_producer = producer.clone(); + let health_state = Arc::new(Mutex::new(SamplerHealthState::Waiting)); telemetry::hardware::spawn_blocking_sampler( engine, interval, @@ -758,10 +781,10 @@ fn spawn_blocking_sampler_task( &producer, health_channel, health_context, - sampler, - sample_channel, + (sampler, sample_channel), seq, error_of(&sample), + &health_state, ); producer.submit_record(channel, &sample); }, @@ -2001,6 +2024,7 @@ fn run() -> Result<(), String> { .spawn( NodeJobActor::unbound(workdir, stack.runtime.create_sender()) .with_actor_timers(engine.handle()) + .with_process_telemetry(telemetry.producer.clone()) .with_route_registrar(job_route_registrar) .with_data_plane(Arc::new(data_plane.clone())), ) @@ -2769,6 +2793,12 @@ fn emit_swim_telemetry( .producer .submit_record(telemetry.channels.swim_probes, &record); } + if let Some(summary) = stack.drain_swim_probe_summary() { + let record = stack.swim_probe_summary_record(summary, local_phase); + telemetry + .producer + .submit_record(telemetry.channels.swim_probes, &record); + } } #[derive(Clone, Copy)] @@ -5378,6 +5408,104 @@ mod control_flow_properties { const DRIVE_PER_ACTION: usize = 16; const FINAL_DRIVE_BUDGET: usize = 256; + #[test] + fn sampler_health_emits_only_state_transitions() { + let endpoint = TelemetryEndpoint::with_descriptor( + StreamDescriptor { + stream: StreamId::new(NodeId::new("sampler-health-test"), Lifetime(1)), + label: None, + origin: StreamOrigin::RemoteNode, + }, + 16, + 16, + ); + let producer = endpoint.producer(); + let health_channel = producer.register_channel( + NODE_SAMPLER_CHANNEL, + ChannelContent::JsonRecord { + schema: Some(NODE_SAMPLER_CHANNEL.to_owned()), + }, + ); + let subscription = endpoint.subscribe_all("sampler-health-transitions"); + let context = SamplerHealthContext { + run_id: 1, + node_id: 2, + stage_index: 0, + }; + let state = Mutex::new(SamplerHealthState::Waiting); + + submit_sampler_started( + &producer, + health_channel, + context, + "cpu", + "host.cpu", + Duration::from_secs(1), + ); + submit_sampler_sample_health( + &producer, + health_channel, + context, + ("cpu", "host.cpu"), + 0, + None, + &state, + ); + submit_sampler_sample_health( + &producer, + health_channel, + context, + ("cpu", "host.cpu"), + 1, + None, + &state, + ); + submit_sampler_sample_health( + &producer, + health_channel, + context, + ("cpu", "host.cpu"), + 2, + Some("unavailable"), + &state, + ); + submit_sampler_sample_health( + &producer, + health_channel, + context, + ("cpu", "host.cpu"), + 3, + Some("still unavailable"), + &state, + ); + submit_sampler_sample_health( + &producer, + health_channel, + context, + ("cpu", "host.cpu"), + 4, + None, + &state, + ); + endpoint.tick(); + + let statuses = subscription + .drain_available() + .into_iter() + .filter_map(|event| match event { + TelemetryEvent::Frame(delivery) if delivery.channel.channel == health_channel => { + serde_json::from_slice::(&delivery.payload) + .ok() + .and_then(|value| value["status"].as_str().map(str::to_owned)) + } + _ => None, + }) + .collect::>(); + assert_eq!( + statuses, + ["started", "waiting", "ready", "failed", "recovered"] + ); + } #[test] fn hardware_sampler_runs_as_engine_task_not_actor() { diff --git a/apps/myelin/src/orchestration/app.rs b/apps/myelin/src/orchestration/app.rs index 3207bfe..eba3284 100644 --- a/apps/myelin/src/orchestration/app.rs +++ b/apps/myelin/src/orchestration/app.rs @@ -2354,6 +2354,7 @@ fn serve_cluster_flush_reply(reply: ManualControlReply) -> Option None, } @@ -2788,6 +2789,10 @@ fn emit_swim_probe_events( let record = stack.swim_probe_event_record(event, local_phase); orch_telemetry.emit_record(dashboard, &record); } + if let Some(summary) = stack.drain_swim_probe_summary() { + let record = stack.swim_probe_summary_record(summary, local_phase); + orch_telemetry.emit_record(dashboard, &record); + } } pub(crate) fn env_optional(name: &str) -> Option { diff --git a/apps/myelin/src/orchestration/control.rs b/apps/myelin/src/orchestration/control.rs index 68ccfb7..e73d206 100644 --- a/apps/myelin/src/orchestration/control.rs +++ b/apps/myelin/src/orchestration/control.rs @@ -429,6 +429,7 @@ pub(crate) fn plugin( let routes = Router::new() .route(FLEET_CONTROL_SCRIPT_URL, get(fleet_control_script)) .route("/api/control/status", get(status)) + .route("/api/control/fleet", get(fleet_status)) .route("/api/control/actors", get(actor_stats)) .route("/api/control/provision", post(provision)) .route("/api/control/kill", post(kill)) @@ -645,6 +646,37 @@ async fn status(State(state): State) -> Response { .await } +async fn fleet_status(State(state): State) -> Response { + let response_rx = match begin_request_reply(&state, CONTROL_REPLY_TIMEOUT, |reply_to| { + ManualControlMsg::QueryFleet { reply_to } + }) { + Ok(response_rx) => response_rx, + Err(response) => return *response, + }; + match response_rx.await { + Ok(ManualControlReply::FleetStatus(model)) => { + Json(ManualControlReply::FleetStatus(model)).into_response() + } + Ok(ManualControlReply::Rejected(error)) => { + (StatusCode::CONFLICT, Json(ErrorResponse { error })).into_response() + } + Ok(_) => ( + StatusCode::SERVICE_UNAVAILABLE, + Json(ErrorResponse { + error: "manual control returned an unexpected Fleet status reply".to_owned(), + }), + ) + .into_response(), + Err(error) => ( + StatusCode::SERVICE_UNAVAILABLE, + Json(ErrorResponse { + error: format!("control reply observer stopped: {error}"), + }), + ) + .into_response(), + } +} + async fn actor_stats(State(state): State) -> impl IntoResponse { Json(state.runtime.stats()) } diff --git a/apps/myelin/src/orchestration/distribution_stack.rs b/apps/myelin/src/orchestration/distribution_stack.rs index d03003c..7942839 100644 --- a/apps/myelin/src/orchestration/distribution_stack.rs +++ b/apps/myelin/src/orchestration/distribution_stack.rs @@ -29,9 +29,10 @@ use distribution::registry_actor::{RegistryActor, RegistryIn, RegistryView}; use distribution::swim::actor::{MembershipChanged, SwimActor, SwimIn}; use distribution::swim::member_list::MemberList; use distribution::swim::probe::SwimConfig; -use distribution::swim::telemetry::{ObservedProbeEvent, ObservedTransition, SwimTelemetry}; -use distribution::telemetry::MembershipTransition; -use distribution::telemetry::SwimProbeEvent; +use distribution::swim::telemetry::{ + ObservedProbeEvent, ObservedProbeSummary, ObservedTransition, SwimTelemetry, +}; +use distribution::telemetry::{MembershipTransition, SwimProbeEvent, SwimProbeSummary}; use distribution::transport_bridge::{ Outbox, OutboxPeerDirectory, OutboxRouteBinder, RelayMirror, RouteView, RouteViewTransport, }; @@ -314,6 +315,10 @@ impl DistributionRuntimeStack { self.swim_telemetry.drain_probe_events() } + pub(crate) fn drain_swim_probe_summary(&self) -> Option { + self.swim_telemetry.drain_probe_summary() + } + pub(crate) fn swim_recent_probe_targets(&self) -> Vec { self.swim_telemetry .recent_targets() @@ -353,6 +358,28 @@ impl DistributionRuntimeStack { lifeguard_enabled: config.lifeguard.is_some(), } } + + pub(crate) fn swim_probe_summary_record( + &self, + summary: ObservedProbeSummary, + local_phase: &str, + ) -> SwimProbeSummary { + SwimProbeSummary { + event: "summary".to_owned(), + interval_ms: summary.interval_ms, + sent: summary.sent, + acked: summary.acked, + direct_sent: summary.direct_sent, + indirect_sent: summary.indirect_sent, + rtt_samples: summary.rtt_samples, + rtt_ms_min: summary.rtt_ms_min, + rtt_ms_p50: summary.rtt_ms_p50, + rtt_ms_p95: summary.rtt_ms_p95, + rtt_ms_max: summary.rtt_ms_max, + local_phase: local_phase.to_owned(), + probe_interval_ms: duration_ms_u64(self.swim_config.probe_interval), + } + } pub(crate) fn membership_transition( &self, transition: &ObservedTransition, diff --git a/apps/myelin/src/orchestration/fleet_control.js b/apps/myelin/src/orchestration/fleet_control.js index 3eaece4..7f7492e 100644 --- a/apps/myelin/src/orchestration/fleet_control.js +++ b/apps/myelin/src/orchestration/fleet_control.js @@ -1,42 +1,61 @@ (() => { - const CONTROL_ID = 'myelin-fleet-control'; + const NODE_CONTROL_ID = 'myelin-fleet-control'; + const BULK_CONTROL_ID = 'myelin-fleet-bulk-control'; const CONFIRM_ID = 'myelin-confirm-dialog'; - const CONFIRM_STYLE_ID = 'myelin-confirm-dialog-style'; + const STYLE_ID = 'myelin-fleet-control-style'; const selectedJobs = new Map(); + const selectedNodes = new Set(); + let lastModel = null; + let syncing = false; + const page = document.getElementById('page'); + if (!page) return; - function confirmKill(logicalNodeId) { + function installUi() { + if (document.getElementById(STYLE_ID)) return; + const style = document.createElement('style'); + style.id = STYLE_ID; + style.textContent = ` + .myelin-control { display:flex;align-items:center;flex-wrap:wrap;gap:10px;margin:0 0 12px } + .myelin-control button { border:1px solid var(--cyan);background:transparent;color:var(--cyan);border-radius:var(--r);padding:6px 10px;font:600 12px var(--mono);cursor:pointer } + .myelin-control button.danger { border-color:var(--bad);color:var(--bad) } + .myelin-control button:disabled { cursor:not-allowed;opacity:.55 } + .myelin-control-message { color:var(--muted);font:12px var(--mono) } + .myelin-bulk-control { width:fit-content;margin:0 0 12px auto;padding:6px 8px;border:1px solid var(--divider);background:transparent;border-radius:var(--r) } + .node-card[data-bulk-selected="true"] { border-color:var(--bad);background:var(--selected);box-shadow:inset 3px 0 0 var(--bad) } + .node-card[data-bulk-killable="true"] { user-select:none } + .myelin-confirm { width:min(440px,calc(100vw - 32px));padding:0;color:var(--text);background:var(--panel);border:1px solid var(--bad);border-radius:var(--r);box-shadow:0 18px 60px rgba(0,0,0,.55) } + .myelin-confirm::backdrop { background:rgba(0,6,12,.78) } + .myelin-confirm form { display:grid;gap:14px;padding:18px } + .myelin-confirm h2,.myelin-confirm p { margin:0 } + .myelin-confirm h2 { color:var(--bad) } + .myelin-confirm-actions { display:flex;justify-content:flex-end;gap:8px } + .myelin-confirm button { padding:6px 12px;background:transparent;color:var(--text);border:1px solid var(--border);border-radius:var(--r);cursor:pointer;font:600 13px var(--mono) } + .myelin-confirm button[value="confirm"] { color:var(--danger-ink);background:var(--danger-fill);border-color:var(--danger-border) } + `; + document.head.append(style); + } + + function confirmTermination(title, message, action) { let dialog = document.getElementById(CONFIRM_ID); if (!dialog) { - const style = document.createElement('style'); - style.id = CONFIRM_STYLE_ID; - style.textContent = ` - .myelin-confirm { width:min(440px,calc(100vw - 32px));padding:0;color:var(--text);background:var(--panel);border:1px solid var(--bad);border-radius:var(--r);box-shadow:0 18px 60px rgba(0,0,0,.55) } - .myelin-confirm::backdrop { background:rgba(0,6,12,.78) } - .myelin-confirm form { display:grid;gap:14px;padding:18px } - .myelin-confirm h2,.myelin-confirm p { margin:0 } - .myelin-confirm h2 { color:var(--bad) } - .myelin-confirm-actions { display:flex;justify-content:flex-end;gap:8px } - .myelin-confirm button { padding:6px 12px;background:transparent;color:var(--text);border:1px solid var(--border);border-radius:var(--r);cursor:pointer;font:600 13px var(--mono) } - .myelin-confirm button[value="confirm"] { color:var(--danger-ink);background:var(--danger-fill);border-color:var(--danger-border) } - `; - document.head.append(style); dialog = document.createElement('dialog'); dialog.id = CONFIRM_ID; dialog.className = 'myelin-confirm'; dialog.setAttribute('aria-labelledby', 'myelin-confirm-title'); dialog.setAttribute('aria-describedby', 'myelin-confirm-message'); dialog.innerHTML = `
-

Terminate managed node?

+

- +
`; document.body.append(dialog); } - dialog.querySelector('#myelin-confirm-message').textContent = - `Terminate managed node ${logicalNodeId}? Vast.ai contracts are destroyed and billing stops.`; + dialog.querySelector('#myelin-confirm-title').textContent = title; + dialog.querySelector('#myelin-confirm-message').textContent = message; + dialog.querySelector('[value="confirm"]').textContent = action; dialog.returnValue = 'cancel'; return new Promise(resolve => { dialog.addEventListener('close', () => resolve(dialog.returnValue === 'confirm'), { once: true }); @@ -44,28 +63,152 @@ }); } - async function syncControl() { - let model; - try { - const response = await fetch('/api/control/status', { cache: 'no-store' }); - if (!response.ok) return; - const payload = await response.json(); - model = payload.Status; - } catch (_) { + function killable(node) { + return node && !['kill_requested', 'stopping', 'stopped', 'orphan'].includes(node.phase); + } + + function managedNodes(model) { + return Array.isArray(model?.nodes) ? model.nodes : []; + } + + function commandId(logicalNodeId) { + return `fleet-kill-${logicalNodeId}-${globalThis.crypto?.randomUUID?.() || Date.now()}`; + } + + async function submitKills(logicalNodeIds) { + return Promise.all(logicalNodeIds.map(async logicalNodeId => { + try { + const response = await fetch(`/api/control/nodes/${logicalNodeId}/kill`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ command_id: commandId(logicalNodeId) }), + }); + if (!response.ok) { + const body = await response.json().catch(() => ({})); + throw new Error(body.error || `HTTP ${response.status}`); + } + return { logicalNodeId, error: null }; + } catch (error) { + return { logicalNodeId, error: error.message }; + } + })); + } + + function resultMessage(results) { + const failed = results.filter(result => result.error); + if (!failed.length) return `Termination accepted for ${results.length} node${results.length === 1 ? '' : 's'}`; + const accepted = results.length - failed.length; + const failures = failed.map(result => `${result.logicalNodeId}: ${result.error}`).join('; '); + return `${accepted} accepted; ${failed.length} failed — ${failures}`; + } + + function syncFleetControl(model) { + const page = document.getElementById('page'); + const cards = [...page.querySelectorAll('a.node-card[data-node]')]; + if (!cards.length) { + document.getElementById(BULK_CONTROL_ID)?.remove(); return; } - window.dispatchEvent(new CustomEvent('dashboard-hardware-source', { - detail: { - source: model?.provider?.provisioning_mode === 'mock' ? 'orchestrator' : 'node', - }, - })); + const byId = new Map(managedNodes(model).map(node => [node.logical_node_id, node])); + for (const logicalNodeId of [...selectedNodes]) { + if (!killable(byId.get(logicalNodeId))) selectedNodes.delete(logicalNodeId); + } + for (const card of cards) { + const logicalNodeId = Number(card.dataset.node); + const selectable = card.dataset.origin !== 'orchestrator' + && Number.isSafeInteger(logicalNodeId) + && killable(byId.get(logicalNodeId)); + card.dataset.bulkKillable = String(selectable); + card.dataset.bulkSelected = String(selectable && selectedNodes.has(logicalNodeId)); + card.setAttribute('aria-selected', String(selectable && selectedNodes.has(logicalNodeId))); + card.title = selectable ? 'Open node; Shift+click to select for termination' : ''; + } - const nodeView = document.querySelector('.node-view[data-node]'); - if (!nodeView) return; - const rawNodeId = nodeView.getAttribute('data-node') || ''; - if (!/^\d+$/.test(rawNodeId)) return; - const logicalNodeId = Number(rawNodeId); - const node = model?.nodes?.find(candidate => candidate.logical_node_id === logicalNodeId); + const count = selectedNodes.size; + let control = document.getElementById(BULK_CONTROL_ID); + if (!count) { + control?.remove(); + return; + } + if (!control) { + control = document.createElement('div'); + control.id = BULK_CONTROL_ID; + control.className = 'myelin-control myelin-bulk-control'; + control.innerHTML = ` + + `; + page.querySelector('.grid').before(control); + control.querySelector('[data-action="kill-selected"]').onclick = killSelected; + } + control.querySelector('[data-selection]').textContent = + `${count} worker node${count === 1 ? '' : 's'} selected`; + const button = control.querySelector('[data-action="kill-selected"]'); + button.disabled = false; + button.textContent = `Terminate ${count} selected`; + } + + + async function killSelected() { + const control = document.getElementById(BULK_CONTROL_ID); + const button = control?.querySelector('[data-action="kill-selected"]'); + const message = control?.querySelector('[data-message]'); + const byId = new Map(managedNodes(lastModel).map(node => [node.logical_node_id, node])); + const ids = [...selectedNodes].filter(logicalNodeId => killable(byId.get(logicalNodeId))); + if (!ids.length) return; + const label = ids.join(', '); + if (!await confirmTermination( + `Terminate ${ids.length} managed node${ids.length === 1 ? '' : 's'}?`, + `Terminate managed node${ids.length === 1 ? '' : 's'} ${label}? Vast.ai contracts are destroyed and billing stops.`, + `Terminate ${ids.length}`, + )) return; + button.disabled = true; + message.textContent = 'Submitting termination requests…'; + const results = await submitKills(ids); + for (const result of results) { + if (!result.error) selectedNodes.delete(result.logicalNodeId); + } + message.textContent = resultMessage(results); + syncFleetControl(lastModel); + } + + function ensureOrchestratorControl(nodeView, model) { + const heading = nodeView.querySelector('section.panel > h2'); + if (!heading) return; + let control = document.getElementById(NODE_CONTROL_ID); + if (!control) { + control = document.createElement('div'); + control.id = NODE_CONTROL_ID; + control.className = 'myelin-control'; + control.innerHTML = ` + `; + heading.after(control); + control.querySelector('[data-action="shutdown-all"]').onclick = shutdownAll; + } + const count = managedNodes(model).filter(killable).length; + const button = control.querySelector('[data-action="shutdown-all"]'); + button.disabled = count === 0; + button.textContent = count ? `Shutdown all (${count})` : 'No managed nodes to shut down'; + } + + async function shutdownAll() { + const control = document.getElementById(NODE_CONTROL_ID); + const button = control?.querySelector('[data-action="shutdown-all"]'); + const message = control?.querySelector('[data-message]'); + const ids = managedNodes(lastModel).filter(killable).map(node => node.logical_node_id); + if (!ids.length) return; + if (!await confirmTermination( + `Shut down all ${ids.length} managed node${ids.length === 1 ? '' : 's'}?`, + `This terminates managed nodes ${ids.join(', ')}. Vast.ai contracts are destroyed and billing stops.`, + 'Shutdown all', + )) return; + button.disabled = true; + message.textContent = 'Submitting termination requests…'; + const results = await submitKills(ids); + message.textContent = resultMessage(results); + } + + async function syncManagedNodeControl(nodeView, model, logicalNodeId) { + const node = managedNodes(model).find(candidate => candidate.logical_node_id === logicalNodeId); if (!node) return; let job = { state: 'idle', message: 'No job submitted' }; @@ -76,34 +219,16 @@ const heading = nodeView.querySelector('section.panel > h2'); if (!heading) return; - let control = document.getElementById(CONTROL_ID); + let control = document.getElementById(NODE_CONTROL_ID); if (!control) { control = document.createElement('div'); - control.id = CONTROL_ID; - control.style.cssText = 'display:flex;align-items:center;flex-wrap:wrap;gap:10px;margin:0 0 12px'; - - const fileInput = document.createElement('input'); - fileInput.type = 'file'; - fileInput.accept = '.toml,text/plain,application/toml'; - fileInput.dataset.jobFile = ''; - fileInput.style.cssText = 'max-width:260px;color:var(--muted);font:12px ui-monospace,SFMono-Regular,Menlo,Consolas,monospace'; - - const jobButton = document.createElement('button'); - jobButton.type = 'button'; - jobButton.dataset.action = 'job'; - jobButton.textContent = 'Submit job'; - jobButton.style.cssText = 'border:1px solid #5cd5ff;background:transparent;color:#5cd5ff;border-radius:3px;padding:6px 10px;font:600 12px ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;cursor:pointer'; - - const killButton = document.createElement('button'); - killButton.type = 'button'; - killButton.dataset.action = 'kill'; - killButton.style.cssText = 'border:1px solid #ef4444;background:transparent;color:#ef4444;border-radius:3px;padding:6px 10px;font:600 12px ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;cursor:pointer'; - - const message = document.createElement('span'); - message.dataset.message = ''; - message.style.cssText = 'font:12px ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;color:var(--muted)'; - - control.append(fileInput, jobButton, killButton, message); + control.id = NODE_CONTROL_ID; + control.className = 'myelin-control'; + control.innerHTML = ` + + + `; + control.querySelector('[data-job-file]').style.cssText = 'max-width:260px;color:var(--muted);font:12px var(--mono)'; heading.after(control); } @@ -124,17 +249,11 @@ killButton.hidden = terminal; killButton.disabled = pending; killButton.textContent = pending ? 'Kill requested' : 'Kill'; - killButton.style.opacity = pending ? '.55' : '1'; - if (job.state !== 'idle') { - message.textContent = job.message; - } else if (selectedJob) { - message.textContent = `Selected ${selectedJob.name}`; - } else if (terminal) { - message.textContent = `managed node ${logicalNodeId}: ${node.phase}`; - } else { - message.textContent = 'Select a job TOML file'; - } + if (job.state !== 'idle') message.textContent = job.message; + else if (selectedJob) message.textContent = `Selected ${selectedJob.name}`; + else if (terminal) message.textContent = `managed node ${logicalNodeId}: ${node.phase}`; + else message.textContent = 'Select a job TOML file'; fileInput.onchange = async () => { const file = fileInput.files?.[0]; @@ -145,8 +264,7 @@ return; } try { - const text = await file.text(); - selectedJobs.set(logicalNodeId, { name: file.name, text }); + selectedJobs.set(logicalNodeId, { name: file.name, text: await file.text() }); jobButton.disabled = jobActive; message.textContent = `Selected ${file.name}`; } catch (error) { @@ -177,30 +295,76 @@ }; killButton.onclick = async () => { - if (!await confirmKill(logicalNodeId)) return; + if (!await confirmTermination( + 'Terminate managed node?', + `Terminate managed node ${logicalNodeId}? Vast.ai contracts are destroyed and billing stops.`, + 'Terminate node', + )) return; killButton.disabled = true; message.textContent = 'Submitting kill…'; - const commandId = `fleet-kill-${globalThis.crypto?.randomUUID?.() || Date.now()}`; - try { - const response = await fetch(`/api/control/nodes/${logicalNodeId}/kill`, { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify({ command_id: commandId }), - }); - if (!response.ok) { - const body = await response.json().catch(() => ({})); - throw new Error(body.error || `HTTP ${response.status}`); - } - message.textContent = 'Kill accepted'; - } catch (error) { - killButton.disabled = false; - message.textContent = error.message; - } + const [result] = await submitKills([logicalNodeId]); + message.textContent = result.error ? result.error : 'Kill accepted'; + if (result.error) killButton.disabled = false; }; } - const page = document.getElementById('page'); - if (page) new MutationObserver(syncControl).observe(page, { childList: true, subtree: true }); + async function syncControl() { + if (syncing) return; + syncing = true; + try { + let model; + try { + const response = await fetch('/api/control/fleet', { cache: 'no-store' }); + if (!response.ok) return; + model = (await response.json()).FleetStatus; + } catch (_) { + return; + } + lastModel = model; + window.dispatchEvent(new CustomEvent('dashboard-hardware-source', { + detail: { source: model?.provider?.provisioning_mode === 'mock' ? 'orchestrator' : 'node' }, + })); + + const nodeView = document.querySelector('.node-view[data-node]'); + if (!nodeView) { + syncFleetControl(model); + return; + } + if (nodeView.dataset.origin === 'orchestrator') { + ensureOrchestratorControl(nodeView, model); + return; + } + const rawNodeId = nodeView.dataset.node || ''; + if (/^\d+$/.test(rawNodeId)) await syncManagedNodeControl(nodeView, model, Number(rawNodeId)); + } finally { + syncing = false; + } + } + + installUi(); + page.addEventListener('click', event => { + const card = event.target.closest('a.node-card[data-node]'); + if (!card || !page.contains(card) || !event.shiftKey) return; + const logicalNodeId = Number(card.dataset.node); + const node = managedNodes(lastModel).find(candidate => candidate.logical_node_id === logicalNodeId); + const selectable = card.dataset.origin !== 'orchestrator' + && Number.isSafeInteger(logicalNodeId) + && killable(node); + if (!selectable) return; + event.preventDefault(); + if (selectedNodes.has(logicalNodeId)) selectedNodes.delete(logicalNodeId); + else selectedNodes.add(logicalNodeId); + syncFleetControl(lastModel); + }); + new MutationObserver(mutations => { + const dashboardChanged = mutations.some(mutation => { + const element = mutation.target.nodeType === Node.ELEMENT_NODE + ? mutation.target + : mutation.target.parentElement; + return !element?.closest(`#${NODE_CONTROL_ID}, #${BULK_CONTROL_ID}`); + }); + if (dashboardChanged) syncControl(); + }).observe(page, { childList: true, subtree: true }); syncControl(); setInterval(syncControl, 1000); })(); diff --git a/apps/myelin/src/orchestration/manual_control.rs b/apps/myelin/src/orchestration/manual_control.rs index c6f7c14..b882acd 100644 --- a/apps/myelin/src/orchestration/manual_control.rs +++ b/apps/myelin/src/orchestration/manual_control.rs @@ -262,6 +262,12 @@ pub(crate) struct ManualReadModel { pub nodes: Vec, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct FleetReadModel { + pub provider: ProviderReadiness, + pub nodes: Vec, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub(crate) enum EffectKind { Create, @@ -430,6 +436,13 @@ impl ManualControl { } } + pub(crate) fn fleet_read_model(&self) -> FleetReadModel { + FleetReadModel { + provider: self.provider.clone(), + nodes: self.snapshot.nodes.clone(), + } + } + pub(crate) fn request_provision( &mut self, request: ProvisionRequest, @@ -1297,6 +1310,9 @@ pub(crate) enum ManualControlMsg { Query { reply_to: ActorAddress, }, + QueryFleet { + reply_to: ActorAddress, + }, Flush { reply_to: ActorAddress, }, @@ -1329,6 +1345,7 @@ pub(crate) enum ManualControlReply { Accepted(CommandRecord), Provider(ProviderReadiness), Status(ManualReadModel), + FleetStatus(FleetReadModel), Offers(Vec), Rejoined(RejoinBinding), Flushed, @@ -1702,6 +1719,12 @@ impl ManualActorControl { ManualControlMsg::Query { reply_to } => { let _ = ctx.send(reply_to, ManualControlReply::Status(self.core.read_model())); } + ManualControlMsg::QueryFleet { reply_to } => { + let _ = ctx.send( + reply_to, + ManualControlReply::FleetStatus(self.core.fleet_read_model()), + ); + } ManualControlMsg::PersistenceFinished { generation, error } => { if self.persistence_in_flight != Some(generation) { return; diff --git a/crates/dashboard/AGENTS.md b/crates/dashboard/AGENTS.md index 6a09e7a..c4baae6 100644 --- a/crates/dashboard/AGENTS.md +++ b/crates/dashboard/AGENTS.md @@ -8,4 +8,4 @@ Keep this crate read-only with respect to observed programs. - It must not send control signals to observed runtimes. - It must not require changes outside `crates/dashboard` for dashboard-only work. -Main built-in view: the fused control plane at `/` and `/view/fleet` (node cards with machine + actor rollup, per-node roster, per-actor dossier via `/api/view/fleet/detail`), backed by `host.*`, `proc.