diff --git a/Cargo.toml b/Cargo.toml index 0740cde..1b766a1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,8 +9,6 @@ members = [ "crates/transport", "crates/distribution", "crates/iroh-driver", - "crates/datastream", - "crates/data-plane", "crates/dashboard", "apps/myelin", "xtask", @@ -23,9 +21,6 @@ default-members = [ "crates/provisioning", "crates/transport", "crates/distribution", - "crates/iroh-driver", - "crates/datastream", - "crates/data-plane", "crates/dashboard", "apps/myelin", "tools/vastai", diff --git a/apps/myelin/src/chat/runtime.rs b/apps/myelin/src/chat/runtime.rs index 6139979..2c18dbe 100644 --- a/apps/myelin/src/chat/runtime.rs +++ b/apps/myelin/src/chat/runtime.rs @@ -12,9 +12,10 @@ use std::thread; use std::time::{Duration, Instant}; use datastream::{ - ChannelContent, ChannelId, DatastreamEndpoint, DatastreamProducer, Frame, Lifetime, NodeId, + ChannelContent, ChannelId, DatastreamEndpoint, DatastreamProducer, Lifetime, NodeId, StreamDescriptor, StreamId, StreamOrigin, }; +use datastream::frame::Frame; use serde::Deserialize; use serde_json::{Value, json}; #[cfg(target_os = "linux")] diff --git a/apps/myelin/src/node/worker_node_runtime.rs b/apps/myelin/src/node/worker_node_runtime.rs index db426c5..5c7ba82 100644 --- a/apps/myelin/src/node/worker_node_runtime.rs +++ b/apps/myelin/src/node/worker_node_runtime.rs @@ -17,10 +17,11 @@ use std::thread; use std::time::{Duration, Instant}; use datastream::{ - ChannelContent, ChannelId, DATASTREAM_PUBLISHER_NAME, DatastreamEndpoint, DatastreamEvent, + ChannelContent, ChannelId, DATASTREAM_PUBLISHER_NAME, DatastreamEndpoint, DatastreamProducer, DatastreamPublisherActor, DatastreamSubscribe, DatastreamSubscription, Lifetime, NodeId, Record, StreamDescriptor, StreamId, StreamOrigin, }; +use datastream::frame::DatastreamEvent; use crate::codecs::register_myelin_actor_codecs; use crate::gguf_shard::{StageShardPlan, materialize_stage_shard_http, validate_stage_shard_cache}; diff --git a/apps/myelin/src/observability/frame_archive.rs b/apps/myelin/src/observability/frame_archive.rs index 9aa60ad..1e82f83 100644 --- a/apps/myelin/src/observability/frame_archive.rs +++ b/apps/myelin/src/observability/frame_archive.rs @@ -2,7 +2,7 @@ use std::fs::{self, File, OpenOptions}; use std::io::{BufRead, BufReader, Write}; use std::path::{Path, PathBuf}; -use datastream::{Frame, StreamId}; +use datastream::frame::{Frame, StreamId}; use serde_json::json; use crate::observability::benchmark; diff --git a/apps/myelin/src/observability/frame_collector.rs b/apps/myelin/src/observability/frame_collector.rs new file mode 100644 index 0000000..0846f9b --- /dev/null +++ b/apps/myelin/src/observability/frame_collector.rs @@ -0,0 +1,310 @@ +//! Frame collection and load-progress extraction for the control loop. +//! +//! [`FrameCollector`] wraps the mpsc channel that buffers datastream frames +//! drained from the iroh driver. It exposes two drain methods that forward +//! queued frames to sinks (dashboard/archive) via closures; the control loop +//! never names [`Frame`] or [`DatastreamEvent`] directly — frames reach sinks +//! only through these closures. Load-progress extraction +//! ([`StageLoadProgress`]) is co-located here because it is the one legitimate +//! read of telemetry content for control decisions (weight-load liveness). + +use datastream::frame::{ChannelRef, DatastreamEvent, Frame, StreamId}; +use iroh_driver::IrohDriver; +use serde_json::Value; +use std::collections::BTreeMap; +use std::sync::mpsc; +use std::time::Instant; + +use crate::observability::orch_datastream::DashboardSupport; + +/// A datastream frame queued from a remote node, with its resolved channel name. +#[derive(Clone, Debug)] +struct CollectedDatastreamFrame { + stream: StreamId, + channel_name: String, + frame: Frame, +} + +/// Per-node load progress distilled from telemetry frames (control input). +#[derive(Clone, Debug, Default)] +pub(crate) struct StageLoadProgress { + pub(crate) node_id: u64, + pub(crate) stage_index: Option, + pub(crate) phase: Option, + pub(crate) bytes_done: Option, + pub(crate) bytes_total: Option, + pub(crate) last_progress: Option, + pub(crate) last_worker_event: Option, + pub(crate) failure_reason: Option, + pub(crate) host_gpu_samples: u64, +} + +impl StageLoadProgress { + pub(crate) fn to_json(&self) -> Value { + serde_json::json!({ + "node_id": self.node_id, + "stage_index": self.stage_index, + "phase": self.phase.as_deref().unwrap_or("unknown"), + "bytes_done": self.bytes_done, + "bytes_total": self.bytes_total, + "last_progress_age_ms": self.last_progress.map(|at| at.elapsed().as_millis()), + "last_worker_event": self.last_worker_event, + "failure_reason": self.failure_reason, + "host_gpu_samples": self.host_gpu_samples, + "host_gpu_missing": self.host_gpu_samples == 0, + }) + } +} + +/// Buffers datastream frames drained from the driver and forwards them to sinks. +pub(crate) struct FrameCollector { + tx: mpsc::Sender, + rx: mpsc::Receiver, +} + +impl FrameCollector { + pub(crate) fn new() -> Self { + let (tx, rx) = mpsc::channel(); + Self { tx, rx } + } + + /// Drain iroh datastream connections into the internal queue. + pub(crate) fn pump(&self, driver: &IrohDriver) { + drain_datastream_connections(driver, &self.tx); + } + + /// Drain queued frames, forwarding each via the closure. No progress extraction. + pub(crate) fn drain(&self, forward: F) + where + F: FnMut(&StreamId, &str, &Frame), + { + let mut forward = forward; + while let Ok(collected) = self.rx.try_recv() { + forward(&collected.stream, &collected.channel_name, &collected.frame); + } + } + + /// Drain queued frames, extracting load progress and forwarding each via the closure. + pub(crate) fn drain_with_progress( + &self, + progress: &mut BTreeMap, + forward: F, + ) where + F: FnMut(&StreamId, &str, &Frame), + { + let mut forward = forward; + let now = Instant::now(); + while let Ok(collected) = self.rx.try_recv() { + update_load_progress_from_frame(progress, &collected, now); + forward(&collected.stream, &collected.channel_name, &collected.frame); + } + } +} + +fn update_load_progress_from_frame( + progress: &mut BTreeMap, + collected: &CollectedDatastreamFrame, + now: Instant, +) { + let stream_node_id = collected.stream.node.as_str().parse::().ok(); + if collected.channel_name == "host.gpu" { + if let Some(node_id) = stream_node_id { + let entry = progress + .entry(node_id) + .or_insert_with(|| StageLoadProgress { + node_id, + ..StageLoadProgress::default() + }); + entry.host_gpu_samples = entry.host_gpu_samples.saturating_add(1); + } + return; + } + + let Ok(value) = serde_json::from_slice::(&collected.frame.payload) else { + return; + }; + if value.get("type").and_then(Value::as_str) == Some("NodeEvent") { + update_load_progress_from_node_event(progress, &value, now); + return; + } + if collected.channel_name == "myelin.worker.weights" { + let Some(node_id) = stream_node_id else { + return; + }; + update_load_progress_from_worker_event(progress, node_id, None, &value, now); + } +} + +fn update_load_progress_from_node_event( + progress: &mut BTreeMap, + value: &Value, + now: Instant, +) { + let Some(node_id) = numeric_json_field(value, "node_id") else { + return; + }; + let stage_index = + numeric_json_field(value, "stage_index").and_then(|stage| u32::try_from(stage).ok()); + let phase = value.get("phase").and_then(Value::as_str); + let status = value.get("status").and_then(Value::as_str); + let detail = value.get("detail").unwrap_or(&Value::Null); + if phase == Some("load_weights") { + let load_phase = match status { + Some("started") => Some("loading_weights"), + Some("ready") => Some("weights_loaded"), + Some("failed") => Some("failed"), + _ => None, + }; + if let Some(load_phase) = load_phase { + let entry = progress + .entry(node_id) + .or_insert_with(|| StageLoadProgress { + node_id, + ..StageLoadProgress::default() + }); + entry.stage_index = stage_index.or(entry.stage_index); + entry.phase = Some(load_phase.to_owned()); + entry.last_progress = Some(now); + if status == Some("failed") { + entry.failure_reason = detail + .get("error") + .and_then(Value::as_str) + .map(str::to_owned); + } + } + } + if let Some(worker_event) = detail.get("event") { + update_load_progress_from_worker_event(progress, node_id, stage_index, worker_event, now); + } +} + +fn update_load_progress_from_worker_event( + progress: &mut BTreeMap, + node_id: u64, + stage_index: Option, + event: &Value, + now: Instant, +) { + let Some(event_type) = event.get("type").and_then(Value::as_str) else { + return; + }; + let Some(phase) = load_phase_for_worker_event(event_type) else { + return; + }; + let entry = progress + .entry(node_id) + .or_insert_with(|| StageLoadProgress { + node_id, + ..StageLoadProgress::default() + }); + entry.stage_index = stage_index.or(entry.stage_index); + entry.phase = Some(phase.to_owned()); + entry.last_worker_event = Some(event_type.to_owned()); + entry.last_progress = Some(now); + if let Some(bytes_done) = + numeric_json_field(event, "bytes_done").or_else(|| numeric_json_field(event, "bytes")) + { + entry.bytes_done = Some(bytes_done); + } + if let Some(bytes_total) = numeric_json_field(event, "bytes_total") { + entry.bytes_total = Some(bytes_total); + } +} + +fn numeric_json_field(value: &Value, field: &str) -> Option { + value + .get(field) + .and_then(|value| value.as_u64().or_else(|| value.as_str()?.parse().ok())) +} + +fn load_phase_for_worker_event(event_type: &str) -> Option<&'static str> { + match event_type { + "GgufDownloadStarted" | "GgufDownloadProgress" => Some("prefetching_model"), + "GgufCacheReady" => Some("cache_ready"), + "StageShardFetchStarted" + | "StageShardRangeFetchStarted" + | "StageShardRangeFetchReady" + | "StageShardTensorFetchStarted" + | "StageShardTensorFetchReady" => Some("fetching_stage_shard"), + "StageShardCacheReady" => Some("stage_shard_cache_ready"), + "StageShardReady" => Some("stage_shard_ready"), + "StageShardFetchFailed" => Some("failed"), + "PipelineStageFromGgufStarted" => Some("constructing_stage"), + "PipelineStageFromGgufReady" => Some("stage_constructed"), + "TokenizerBuildStarted" => Some("building_tokenizer"), + "TokenizerBuildReady" => Some("tokenizer_ready"), + "WeightsLoaded" => Some("weights_loaded"), + "WorkerFatal" => Some("failed"), + _ => None, + } +} + +fn drain_datastream_connections( + driver: &IrohDriver, + frame_tx: &mpsc::Sender, +) { + for read in driver.drain_datastream_reads() { + let mut channels = read + .header + .channels + .iter() + .map(|descriptor| { + ( + ChannelRef { + stream: descriptor.stream.clone(), + channel: descriptor.id, + }, + descriptor.name.clone(), + ) + }) + .collect::>(); + for event in read.events { + match event { + DatastreamEvent::ChannelDeclared(descriptor) => { + channels.insert( + ChannelRef { + stream: descriptor.stream.clone(), + channel: descriptor.id, + }, + descriptor.name, + ); + } + DatastreamEvent::Frame(delivery) => { + let channel_name = channels + .get(&delivery.channel) + .cloned() + .unwrap_or_else(|| format!("channel#{}", delivery.channel.channel.0)); + let frame = Frame::new( + delivery.channel.channel, + delivery.position, + delivery.payload, + ); + if frame_tx + .send(CollectedDatastreamFrame { + stream: delivery.channel.stream, + channel_name, + frame, + }) + .is_err() + { + return; + } + } + DatastreamEvent::StreamDeclared(_) | DatastreamEvent::StreamEnded(_) => {} + } + } + } +} + +/// Publish a frame to the dashboard, if one is attached. Used by the producer +/// flush path as well as the control-loop drain closures. +pub(crate) fn ingest_dashboard_frame( + dashboard: Option<&DashboardSupport>, + stream: &StreamId, + channel: &str, + frame: &Frame, +) { + if let Some(dashboard) = dashboard { + dashboard.publish_frame(stream, channel, frame); + } +} diff --git a/apps/myelin/src/observability/mod.rs b/apps/myelin/src/observability/mod.rs index 403fc68..b8b6798 100644 --- a/apps/myelin/src/observability/mod.rs +++ b/apps/myelin/src/observability/mod.rs @@ -3,7 +3,9 @@ pub(crate) mod benchmark; #[cfg(feature = "dashboard")] pub(crate) mod dashboard_view; +pub(crate) mod frame_collector; pub(crate) mod frame_archive; pub(crate) mod lifecycle; +pub(crate) mod orch_datastream; pub(crate) mod provisioning_logs; pub(crate) mod telemetry; diff --git a/apps/myelin/src/observability/orch_datastream.rs b/apps/myelin/src/observability/orch_datastream.rs new file mode 100644 index 0000000..93029ba --- /dev/null +++ b/apps/myelin/src/observability/orch_datastream.rs @@ -0,0 +1,338 @@ +//! Orchestrator-owned datastream producer plus dashboard sink support. +//! +//! [`OrchDatastream`] owns the producer-side endpoint that emits bootstrap, +//! prompt, provisioning, and SWIM telemetry. [`DashboardSupport`] adapts the +//! optional live dashboard. Both are consumed by the control loop in +//! `orchestration::app`; frame-bearing read paths live in `frame_collector`. + +use datastream::frame::{Frame, StreamId}; +use datastream::{ + ChannelContent, ChannelId, DatastreamEndpoint, DatastreamProducer, Lifetime, NodeId, Record, + StreamDescriptor, StreamOrigin, +}; +use serde_json::{Value, json}; +use std::collections::BTreeMap; +use std::path::Path; +use std::sync::Arc; + +use crate::observability::benchmark; +use crate::observability::frame_archive::FrameArchive; +use crate::observability::frame_collector::ingest_dashboard_frame; +#[cfg(feature = "dashboard")] +use crate::observability::dashboard_view::MyelinClusterDashboardView; +use crate::observability::telemetry::{ + MYELIN_PROVISIONING_EVENTS, MyelinProvisionEventRecord, MyelinProvisionLogRecord, + myelin_provision_log_channel, +}; +#[cfg(feature = "dashboard")] +use crate::orchestration::app::env_optional; +use crate::provisioning::{ProvisionEvent, ProvisionLogLine}; +use distribution::telemetry::{MembershipTransition, SwimProbeEvent}; +use swactor_engine::EngineHandle; +use swactor::stats::StatsHook; + +pub(crate) const MYELIN_ORCH_BOOTSTRAP: &str = "myelin.orch.bootstrap"; +pub(crate) const MYELIN_ORCH_PROMPT: &str = "myelin.orch.prompt"; +pub(crate) const MYELIN_SWIM_MEMBERSHIP: &str = "myelin.swim.membership"; +pub(crate) const MYELIN_STAGE_ROUTE: &str = "myelin.orch.stage_route"; + +pub(crate) struct OrchDatastream { + stream: StreamId, + endpoint: DatastreamEndpoint, + producer: DatastreamProducer, + channels: BTreeMap, + channel_names: BTreeMap, + archive: Option, +} + +impl OrchDatastream { + pub(crate) fn new(run_id: u64, frame_log: Option<&Path>) -> Result { + let stream = StreamId::new(NodeId::new("myelin-orchestrator"), Lifetime(run_id)); + let endpoint = DatastreamEndpoint::with_descriptor( + StreamDescriptor { + stream: stream.clone(), + label: Some("myelin orchestrator".to_owned()), + origin: StreamOrigin::Orchestrator, + }, + 4096, + 1024, + ); + let producer = endpoint.producer(); + let mut out = Self { + stream, + endpoint, + producer, + channels: BTreeMap::new(), + channel_names: BTreeMap::new(), + archive: frame_log + .map(|p| FrameArchive::open_with_label(p, "datastream frame log")) + .transpose()?, + }; + for name in [ + MYELIN_PROVISIONING_EVENTS, + MYELIN_ORCH_BOOTSTRAP, + MYELIN_ORCH_PROMPT, + MYELIN_SWIM_MEMBERSHIP, + MYELIN_STAGE_ROUTE, + ] { + out.channel_by_name(name); + } + out.record_channel::(); + out.record_channel::(); + Ok(out) + } + + pub(crate) fn channel_by_name(&mut self, name: &str) -> ChannelId { + if let Some(id) = self.channels.get(name).copied() { + return id; + } + let id = self.producer.register_channel( + name, + ChannelContent::JsonRecord { + schema: Some(name.to_owned()), + }, + ); + self.channels.insert(name.to_owned(), id); + self.channel_names.insert(id, name.to_owned()); + id + } + + pub(crate) fn record_channel(&mut self) -> ChannelId { + if let Some(id) = self.channels.get(R::CHANNEL).copied() { + return id; + } + let id = self.producer.register_record::(); + self.channels.insert(R::CHANNEL.to_owned(), id); + self.channel_names.insert(id, R::CHANNEL.to_owned()); + id + } + + pub(crate) fn emit_event( + &mut self, + dashboard: Option<&DashboardSupport>, + event: ProvisionEvent, + ) { + let payload = serde_json::to_vec(&MyelinProvisionEventRecord::new(event)) + .expect("serialize provisioning event"); + self.emit_bytes(dashboard, MYELIN_PROVISIONING_EVENTS, payload); + } + + pub(crate) fn emit_log( + &mut self, + dashboard: Option<&DashboardSupport>, + line: ProvisionLogLine, + ) { + let channel = myelin_provision_log_channel(line.node_id, line.stream); + let payload = serde_json::to_vec(&MyelinProvisionLogRecord::new(line)) + .expect("serialize provision log"); + self.emit_bytes(dashboard, &channel, payload); + } + + pub(crate) fn emit_bootstrap( + &mut self, + dashboard: Option<&DashboardSupport>, + run_id: u64, + node_id: u64, + phase: &str, + status: &str, + detail: Value, + ) { + self.emit_bootstrap_to_channel( + dashboard, + MYELIN_ORCH_BOOTSTRAP, + run_id, + node_id, + phase, + status, + detail, + ); + } + + pub(crate) fn emit_bootstrap_to_channel( + &mut self, + dashboard: Option<&DashboardSupport>, + channel: &str, + run_id: u64, + node_id: u64, + phase: &str, + status: &str, + detail: Value, + ) { + let benchmark = benchmark::stamp("myelin-orchestrator"); + let payload = serde_json::to_vec(&json!({ + "schema_version": benchmark["schema_version"].clone(), + "type":"OrchBootstrap", + "event_type":"OrchBootstrap", + "event_name":phase, + "phase":phase, + "status":status, + "run_id":run_id, + "node_id":node_id, + "producer_component":benchmark["producer_component"].clone(), + "producer_instance_id":benchmark["producer_instance_id"].clone(), + "producer_process_id":benchmark["producer_process_id"].clone(), + "producer_sequence":benchmark["producer_sequence"].clone(), + "wall_clock_unix_ms":benchmark["wall_clock_unix_ms"].clone(), + "monotonic_ms":benchmark["monotonic_ms"].clone(), + "clock_source":benchmark["clock_source"].clone(), + "span_id":format!("myelin-orchestrator:{run_id}:{}:{phase}", benchmark["producer_sequence"]), + "parent_span_id":Value::Null, + "benchmark":benchmark, + "detail":detail, + })) + .expect("serialize orch bootstrap event"); + self.emit_bytes(dashboard, channel, payload); + } + + pub(crate) fn emit_prompt( + &mut self, + dashboard: Option<&DashboardSupport>, + run_id: u64, + node_id: u64, + request_id: u64, + phase: &str, + status: &str, + detail: Value, + ) { + let benchmark = benchmark::stamp("myelin-orchestrator"); + let payload = serde_json::to_vec(&json!({ + "schema_version": benchmark["schema_version"].clone(), + "type":"OrchPromptEvent", + "event_type":"OrchPromptEvent", + "event_name":phase, + "phase":phase, + "status":status, + "run_id":run_id, + "node_id":node_id, + "request_id":request_id, + "producer_component":benchmark["producer_component"].clone(), + "producer_instance_id":benchmark["producer_instance_id"].clone(), + "producer_process_id":benchmark["producer_process_id"].clone(), + "producer_sequence":benchmark["producer_sequence"].clone(), + "wall_clock_unix_ms":benchmark["wall_clock_unix_ms"].clone(), + "monotonic_ms":benchmark["monotonic_ms"].clone(), + "clock_source":benchmark["clock_source"].clone(), + "span_id":format!("myelin-orchestrator:{run_id}:{request_id}:{}:{phase}", benchmark["producer_sequence"]), + "parent_span_id":format!("request:{request_id}"), + "benchmark":benchmark, + "detail":detail, + })) + .expect("serialize orch prompt event"); + self.emit_bytes(dashboard, MYELIN_ORCH_PROMPT, payload); + } + + pub(crate) fn emit_record( + &mut self, + dashboard: Option<&DashboardSupport>, + record: &R, + ) { + let id = self.record_channel::(); + self.producer.submit_record(id, record); + self.flush(dashboard, "orchestrator"); + } + + pub(crate) fn emit_bytes( + &mut self, + dashboard: Option<&DashboardSupport>, + channel: &str, + payload: Vec, + ) { + self.emit_bytes_from(dashboard, channel, payload, "orchestrator"); + } + + pub(crate) fn emit_bytes_from( + &mut self, + dashboard: Option<&DashboardSupport>, + channel: &str, + payload: Vec, + source: &str, + ) { + let id = self.channel_by_name(channel); + self.producer.submit_bytes(id, payload); + self.flush(dashboard, source); + } + + pub(crate) fn flush(&mut self, dashboard: Option<&DashboardSupport>, source: &str) { + let stream = self.stream.clone(); + for frame in self.endpoint.mux().drain() { + let channel = self + .channel_names + .get(&frame.channel) + .cloned() + .unwrap_or_else(|| format!("channel#{}", frame.channel.0)); + ingest_dashboard_frame(dashboard, &stream, &channel, &frame); + self.archive_frame(source, &stream, &channel, &frame); + } + } + + pub(crate) fn archive_frame( + &mut self, + source: &str, + stream: &StreamId, + channel: &str, + frame: &Frame, + ) { + if let Some(archive) = &mut self.archive { + let _ = archive.record(source, stream, channel, frame); + } + } + + /// Attach the datastream stats hook on `channel` so actor snapshots flow to it. + pub(crate) fn stats_hook_on(&self, channel: ChannelId) -> Arc { + self.producer.stats_hook_on(channel) + } +} + +#[cfg(feature = "dashboard")] +pub(crate) struct DashboardSupport { + handle: dashboard::DashboardHandle, +} + +#[cfg(feature = "dashboard")] +impl DashboardSupport { + pub(crate) fn start(enabled: bool, engine: &EngineHandle) -> Result, String> { + if !enabled { + return Ok(None); + } + let mut config = dashboard::DashboardConfig::default(); + if let Some(port) = env_optional("MYELIN_DASHBOARD_PORT") { + config.port = port + .parse::() + .map_err(|e| format!("invalid MYELIN_DASHBOARD_PORT={port:?}: {e}"))?; + } + let handle = dashboard::DashboardHandle::new(config); + handle.register_view(Arc::new(MyelinClusterDashboardView::new())); + engine.spawn(handle.http_server()); + Ok(Some(Self { handle })) + } + + pub(crate) fn publish_frame(&self, stream: &StreamId, channel: &str, frame: &Frame) { + self.handle.publish(dashboard::FrameEvent { + stream: dashboard::StreamEvent { + node: stream.node.as_str().to_string(), + life: stream.life.0, + }, + channel: channel.to_owned(), + position: frame.position.0, + payload: frame.payload.clone(), + }); + } +} + +#[cfg(not(feature = "dashboard"))] +pub(crate) struct DashboardSupport; + +#[cfg(not(feature = "dashboard"))] +impl DashboardSupport { + pub(crate) fn start(enabled: bool, _engine: &EngineHandle) -> Result, String> { + if enabled { + return Err( + "MYELIN_DASHBOARD requires building myelin-system with feature dashboard" + .to_owned(), + ); + } + Ok(None) + } + + pub(crate) fn publish_frame(&self, _stream: &StreamId, _channel: &str, _frame: &Frame) {} +} diff --git a/apps/myelin/src/orchestration/app.rs b/apps/myelin/src/orchestration/app.rs index 5059ff6..fcffe56 100644 --- a/apps/myelin/src/orchestration/app.rs +++ b/apps/myelin/src/orchestration/app.rs @@ -16,18 +16,15 @@ use crate::node_actor::{ NodeAgentMsg, StageEdgeKindWire, StageInboundEdgeWire, StageObjectSpecWire, StageOutboundEdgeWire, StageProvisionWire, StageRingSpecWire, }; -#[cfg(feature = "dashboard")] -use crate::observability::dashboard_view::MyelinClusterDashboardView; -use crate::observability::{benchmark, frame_archive::FrameArchive}; +use crate::observability::frame_collector::{FrameCollector, StageLoadProgress}; +use crate::observability::orch_datastream::{ + DashboardSupport, OrchDatastream, MYELIN_STAGE_ROUTE, MYELIN_SWIM_MEMBERSHIP, +}; use crate::orchestration::actor::{OrchestratorActor, OrchestratorMsg, OrchestratorReport}; use crate::orchestration::config::{DEFAULT_CONFIG_PATH, TomlConfigOverlay}; use crate::gguf_shard::{StageShardPlan, plan_stage_shard}; use crate::node_provisioning::{ProviderKind, provider_kind}; -use crate::observability::telemetry::{ - MYELIN_PROVISIONING_EVENTS, MyelinProvisionEventRecord, MyelinProvisionLogRecord, - myelin_provision_log_channel, -}; use crate::orchestration::cluster_reconciler::{ProvisionedClusterGuard, ReconcilerNodeBinding}; use crate::orchestration::distribution_stack::{DistributionRuntimeStack, duration_ms_u64}; use crate::orchestration::provider_adapters::relay::{ @@ -54,14 +51,9 @@ use ::provisioning::{ RunId as ClusterRunId, RunNodeGroupSpec, SwactorId, SwarmJoinTemplate, }; use data_plane::object_record as ingress; -use datastream::{ - ChannelContent, ChannelId, ChannelRef, DatastreamEndpoint, DatastreamEvent, DatastreamProducer, - DatastreamPublisherMsg, DatastreamSubscribe, Frame, Lifetime, NodeId, Record, StreamDescriptor, - StreamId, StreamOrigin, SubscriptionRequest, -}; +use datastream::{DatastreamPublisherMsg, DatastreamSubscribe, SubscriptionRequest}; use distribution::node::DistributedNodeConfig; use distribution::swim::telemetry::ObservedTransition; -use distribution::telemetry::{MembershipTransition, SwimProbeEvent}; use distribution::types::{MemberState, NodeId as DistNodeId}; use iroh::EndpointAddr; use iroh_driver::{ @@ -87,10 +79,6 @@ const PUMP_INTERVAL: Duration = Duration::from_millis(10); const RUNTIME_READY_ACK_RETRY_INTERVAL: Duration = Duration::from_millis(250); const STAGE_PROVISION_ACTIVE_RESEND_AFTER: Duration = Duration::from_secs(60); const PIPELINE_PROMPT_WAIT_LOG_INTERVAL: Duration = Duration::from_secs(15); -const MYELIN_ORCH_BOOTSTRAP: &str = "myelin.orch.bootstrap"; -const MYELIN_ORCH_PROMPT: &str = "myelin.orch.prompt"; -const MYELIN_SWIM_MEMBERSHIP: &str = "myelin.swim.membership"; -const MYELIN_STAGE_ROUTE: &str = "myelin.orch.stage_route"; const DATASTREAM_FRAME_LOG_ENV: &str = "MYELIN_DATASTREAM_FRAME_LOG"; pub(crate) fn run_with_options( @@ -222,7 +210,7 @@ where }; let actors_channel = orch_datastream.channel_by_name("runtime.actors"); - let orch_stats_hook = orch_datastream.producer.stats_hook_on(actors_channel); + let orch_stats_hook = orch_datastream.stats_hook_on(actors_channel); // Build the core swactor runtime parts, clone the routing handle needed by // integrations, then hand the workers to the engine. The engine owns both @@ -348,7 +336,7 @@ where json!({"transport":"iroh","routes":"attached","protocol_ticker":"engine-hosted"}), ); - let (frame_tx, frame_rx) = mpsc::channel::(); + let collector = FrameCollector::new(); bootstrap( &mut orch_datastream, None, @@ -490,8 +478,7 @@ where driver: &mut driver, stack: &stack, obs_rx: &obs_rx, - frame_rx: &frame_rx, - frame_tx: &frame_tx, + collector: &collector, orchestrator_reports: &orchestrator_reports, stop_rx: &stop_rx, dashboard: dashboard.as_ref(), @@ -570,8 +557,7 @@ where driver: &mut driver, stack: &stack, obs_rx: &obs_rx, - frame_rx: &frame_rx, - frame_tx: &frame_tx, + collector: &collector, orchestrator_reports: &orchestrator_reports, stop_rx: &stop_rx, dashboard: dashboard.as_ref(), @@ -2053,8 +2039,7 @@ struct RuntimeReadyAckLoop<'a> { driver: &'a mut IrohDriver, stack: &'a DistributionRuntimeStack, obs_rx: &'a mpsc::Receiver, - frame_rx: &'a mpsc::Receiver, - frame_tx: &'a mpsc::Sender, + collector: &'a FrameCollector, orchestrator_reports: &'a swactor::runtime::Inbox, stop_rx: &'a mpsc::Receiver<()>, dashboard: Option<&'a DashboardSupport>, @@ -2078,8 +2063,7 @@ fn wait_for_runtime_ready_acks( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -2127,7 +2111,7 @@ fn wait_for_runtime_ready_acks( }) { return Ok(false); } - pump(driver, frame_tx); + collector.pump(driver); drain_orch_stdio_capture( orch_stdio_rx, orch_datastream, @@ -2143,7 +2127,12 @@ fn wait_for_runtime_ready_acks( while let Ok(observation) = obs_rx.try_recv() { emit_plugin_observation(orch_datastream, dashboard, provider, &observation); } - drain_frames(frame_rx, dashboard, orch_datastream); + collector.drain(|stream, channel, frame| { + if let Some(d) = dashboard { + d.publish_frame(stream, channel, frame); + } + orch_datastream.archive_frame("node", stream, channel, frame); + }); while let Some(report) = orchestrator_reports.try_recv() { let OrchestratorReport::NodeRuntimeReadyAck { run_id: ack_run_id, @@ -2340,8 +2329,7 @@ fn start_and_provision_workers( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -2437,8 +2425,13 @@ fn start_and_provision_workers( if provisioned_nodes.awaiting_runtime() || provisioned_nodes.is_converged() { break; } - pump(driver, frame_tx); - drain_frames(frame_rx, dashboard, orch_datastream); + collector.pump(driver); + collector.drain(|stream, channel, frame| { + if let Some(d) = dashboard { + d.publish_frame(stream, channel, frame); + } + orch_datastream.archive_frame("node", stream, channel, frame); + }); drain_orch_stdio_capture( orch_stdio_rx, orch_datastream, @@ -2487,8 +2480,7 @@ fn start_and_provision_workers( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -2533,8 +2525,7 @@ fn start_and_provision_workers( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -2576,8 +2567,13 @@ fn start_and_provision_workers( provisioned_nodes .poll(SystemTime::now()) .map_err(|error| format!("cluster convergence: {error}"))?; - pump(driver, frame_tx); - drain_frames(frame_rx, dashboard, orch_datastream); + collector.pump(driver); + collector.drain(|stream, channel, frame| { + if let Some(d) = dashboard { + d.publish_frame(stream, channel, frame); + } + orch_datastream.archive_frame("node", stream, channel, frame); + }); while let Ok(observation) = obs_rx.try_recv() { emit_plugin_observation(orch_datastream, dashboard, &config.provider, &observation); } @@ -2612,8 +2608,7 @@ fn start_and_provision_workers( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -2636,8 +2631,7 @@ fn start_and_provision_workers( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -2967,8 +2961,7 @@ fn wait_for_runtime_readies( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -2988,7 +2981,7 @@ fn wait_for_runtime_readies( cluster.current_attempt(*node_id) == Some(::provisioning::NodeAttemptId(ready.readiness_id)) }); - pump(driver, frame_tx); + collector.pump(driver); emit_swim_transitions( orch_datastream, dashboard, @@ -2997,7 +2990,12 @@ fn wait_for_runtime_readies( stack, ); emit_swim_probe_events(orch_datastream, dashboard, stack, "runtime_ready_wait"); - drain_frames(frame_rx, dashboard, orch_datastream); + collector.drain(|stream, channel, frame| { + if let Some(d) = dashboard { + d.publish_frame(stream, channel, frame); + } + orch_datastream.archive_frame("node", stream, channel, frame); + }); drain_orch_stdio_capture( orch_stdio_rx, orch_datastream, @@ -3072,8 +3070,7 @@ fn wait_for_weights_loaded_count( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -3096,7 +3093,7 @@ fn wait_for_weights_loaded_count( let mut stage_last_sends = BTreeMap::::new(); let mut load_progress = BTreeMap::::new(); loop { - pump(driver, frame_tx); + collector.pump(driver); emit_swim_transitions(orch_datastream, dashboard, run_id, node_id, stack); emit_swim_probe_events(orch_datastream, dashboard, stack, "weights_loaded_wait"); drain_orch_stdio_capture(orch_stdio_rx, orch_datastream, dashboard, run_id, node_id); @@ -3124,7 +3121,7 @@ fn wait_for_weights_loaded_count( let mut provision = PipelineStageProvision { driver: &mut *driver, stack, - frame_tx, + collector, dashboard, orch_datastream: &mut *orch_datastream, run_id, @@ -3194,7 +3191,12 @@ fn wait_for_weights_loaded_count( | PluginObservation::StderrLine { .. } => {} } } - drain_frames_with_load_progress(frame_rx, dashboard, orch_datastream, &mut load_progress); + collector.drain_with_progress(&mut load_progress, |stream, channel, frame| { + if let Some(d) = dashboard { + d.publish_frame(stream, channel, frame); + } + orch_datastream.archive_frame("node", stream, channel, frame); + }); while let Some(report) = orchestrator_reports.try_recv() { match report { OrchestratorReport::WeightsReady { @@ -3230,7 +3232,7 @@ fn wait_for_weights_loaded_count( struct PipelineStageProvision<'a> { driver: &'a mut IrohDriver, stack: &'a DistributionRuntimeStack, - frame_tx: &'a mpsc::Sender, + collector: &'a FrameCollector, dashboard: Option<&'a DashboardSupport>, orch_datastream: &'a mut OrchDatastream, run_id: u64, @@ -3428,7 +3430,7 @@ fn send_pipeline_stage_provision( ctx.pipeline_coordinator, ctx.stage_shard_plans, )?; - pump(ctx.driver, ctx.frame_tx); + ctx.collector.pump(ctx.driver); Ok(()) } @@ -3442,43 +3444,6 @@ struct ActivePrompt { events: mpsc::Sender, } -#[derive(Clone, Debug)] -struct CollectedDatastreamFrame { - stream: StreamId, - channel_name: String, - frame: Frame, -} - -#[derive(Clone, Debug, Default)] -struct StageLoadProgress { - node_id: u64, - stage_index: Option, - phase: Option, - bytes_done: Option, - bytes_total: Option, - last_progress: Option, - last_worker_event: Option, - failure_reason: Option, - host_gpu_samples: u64, -} - -impl StageLoadProgress { - fn to_json(&self) -> Value { - json!({ - "node_id": self.node_id, - "stage_index": self.stage_index, - "phase": self.phase.as_deref().unwrap_or("unknown"), - "bytes_done": self.bytes_done, - "bytes_total": self.bytes_total, - "last_progress_age_ms": self.last_progress.map(|at| at.elapsed().as_millis()), - "last_worker_event": self.last_worker_event, - "failure_reason": self.failure_reason, - "host_gpu_samples": self.host_gpu_samples, - "host_gpu_missing": self.host_gpu_samples == 0, - }) - } -} - fn stage_load_phase_is_active(phase: Option<&str>) -> bool { matches!( phase, @@ -3556,425 +3521,6 @@ fn stage_load_liveness_detail( }) } -fn update_load_progress_from_frame( - progress: &mut BTreeMap, - collected: &CollectedDatastreamFrame, - now: Instant, -) { - let stream_node_id = collected.stream.node.as_str().parse::().ok(); - if collected.channel_name == "host.gpu" { - if let Some(node_id) = stream_node_id { - let entry = progress - .entry(node_id) - .or_insert_with(|| StageLoadProgress { - node_id, - ..StageLoadProgress::default() - }); - entry.host_gpu_samples = entry.host_gpu_samples.saturating_add(1); - } - return; - } - - let Ok(value) = serde_json::from_slice::(&collected.frame.payload) else { - return; - }; - if value.get("type").and_then(Value::as_str) == Some("NodeEvent") { - update_load_progress_from_node_event(progress, &value, now); - return; - } - if collected.channel_name == "myelin.worker.weights" { - let Some(node_id) = stream_node_id else { - return; - }; - update_load_progress_from_worker_event(progress, node_id, None, &value, now); - } -} - -fn update_load_progress_from_node_event( - progress: &mut BTreeMap, - value: &Value, - now: Instant, -) { - let Some(node_id) = numeric_json_field(value, "node_id") else { - return; - }; - let stage_index = - numeric_json_field(value, "stage_index").and_then(|stage| u32::try_from(stage).ok()); - let phase = value.get("phase").and_then(Value::as_str); - let status = value.get("status").and_then(Value::as_str); - let detail = value.get("detail").unwrap_or(&Value::Null); - if phase == Some("load_weights") { - let load_phase = match status { - Some("started") => Some("loading_weights"), - Some("ready") => Some("weights_loaded"), - Some("failed") => Some("failed"), - _ => None, - }; - if let Some(load_phase) = load_phase { - let entry = progress - .entry(node_id) - .or_insert_with(|| StageLoadProgress { - node_id, - ..StageLoadProgress::default() - }); - entry.stage_index = stage_index.or(entry.stage_index); - entry.phase = Some(load_phase.to_owned()); - entry.last_progress = Some(now); - if status == Some("failed") { - entry.failure_reason = detail - .get("error") - .and_then(Value::as_str) - .map(str::to_owned); - } - } - } - if let Some(worker_event) = detail.get("event") { - update_load_progress_from_worker_event(progress, node_id, stage_index, worker_event, now); - } -} - -fn update_load_progress_from_worker_event( - progress: &mut BTreeMap, - node_id: u64, - stage_index: Option, - event: &Value, - now: Instant, -) { - let Some(event_type) = event.get("type").and_then(Value::as_str) else { - return; - }; - let Some(phase) = load_phase_for_worker_event(event_type) else { - return; - }; - let entry = progress - .entry(node_id) - .or_insert_with(|| StageLoadProgress { - node_id, - ..StageLoadProgress::default() - }); - entry.stage_index = stage_index.or(entry.stage_index); - entry.phase = Some(phase.to_owned()); - entry.last_worker_event = Some(event_type.to_owned()); - entry.last_progress = Some(now); - if let Some(bytes_done) = - numeric_json_field(event, "bytes_done").or_else(|| numeric_json_field(event, "bytes")) - { - entry.bytes_done = Some(bytes_done); - } - if let Some(bytes_total) = numeric_json_field(event, "bytes_total") { - entry.bytes_total = Some(bytes_total); - } -} - -fn numeric_json_field(value: &Value, field: &str) -> Option { - value - .get(field) - .and_then(|value| value.as_u64().or_else(|| value.as_str()?.parse().ok())) -} - -fn load_phase_for_worker_event(event_type: &str) -> Option<&'static str> { - match event_type { - "GgufDownloadStarted" | "GgufDownloadProgress" => Some("prefetching_model"), - "GgufCacheReady" => Some("cache_ready"), - "StageShardFetchStarted" - | "StageShardRangeFetchStarted" - | "StageShardRangeFetchReady" - | "StageShardTensorFetchStarted" - | "StageShardTensorFetchReady" => Some("fetching_stage_shard"), - "StageShardCacheReady" => Some("stage_shard_cache_ready"), - "StageShardReady" => Some("stage_shard_ready"), - "StageShardFetchFailed" => Some("failed"), - "PipelineStageFromGgufStarted" => Some("constructing_stage"), - "PipelineStageFromGgufReady" => Some("stage_constructed"), - "TokenizerBuildStarted" => Some("building_tokenizer"), - "TokenizerBuildReady" => Some("tokenizer_ready"), - "WeightsLoaded" => Some("weights_loaded"), - "WorkerFatal" => Some("failed"), - _ => None, - } -} - -fn drain_datastream_connections( - driver: &IrohDriver, - frame_tx: &mpsc::Sender, -) { - for read in driver.drain_datastream_reads() { - let mut channels = read - .header - .channels - .iter() - .map(|descriptor| { - ( - ChannelRef { - stream: descriptor.stream.clone(), - channel: descriptor.id, - }, - descriptor.name.clone(), - ) - }) - .collect::>(); - for event in read.events { - match event { - DatastreamEvent::ChannelDeclared(descriptor) => { - channels.insert( - ChannelRef { - stream: descriptor.stream.clone(), - channel: descriptor.id, - }, - descriptor.name, - ); - } - DatastreamEvent::Frame(delivery) => { - let channel_name = channels - .get(&delivery.channel) - .cloned() - .unwrap_or_else(|| format!("channel#{}", delivery.channel.channel.0)); - let frame = Frame::new( - delivery.channel.channel, - delivery.position, - delivery.payload, - ); - if frame_tx - .send(CollectedDatastreamFrame { - stream: delivery.channel.stream, - channel_name, - frame, - }) - .is_err() - { - return; - } - } - DatastreamEvent::StreamDeclared(_) | DatastreamEvent::StreamEnded(_) => {} - } - } - } -} - -struct OrchDatastream { - stream: StreamId, - endpoint: DatastreamEndpoint, - producer: DatastreamProducer, - channels: BTreeMap, - channel_names: BTreeMap, - archive: Option, -} - -impl OrchDatastream { - fn new(run_id: u64, frame_log: Option<&Path>) -> Result { - let stream = StreamId::new(NodeId::new("myelin-orchestrator"), Lifetime(run_id)); - let endpoint = DatastreamEndpoint::with_descriptor( - StreamDescriptor { - stream: stream.clone(), - label: Some("myelin orchestrator".to_owned()), - origin: StreamOrigin::Orchestrator, - }, - 4096, - 1024, - ); - let producer = endpoint.producer(); - let mut out = Self { - stream, - endpoint, - producer, - channels: BTreeMap::new(), - channel_names: BTreeMap::new(), - archive: frame_log - .map(|p| FrameArchive::open_with_label(p, "datastream frame log")) - .transpose()?, - }; - for name in [ - MYELIN_PROVISIONING_EVENTS, - MYELIN_ORCH_BOOTSTRAP, - MYELIN_ORCH_PROMPT, - MYELIN_SWIM_MEMBERSHIP, - MYELIN_STAGE_ROUTE, - ] { - out.channel_by_name(name); - } - out.record_channel::(); - out.record_channel::(); - Ok(out) - } - - fn channel_by_name(&mut self, name: &str) -> ChannelId { - if let Some(id) = self.channels.get(name).copied() { - return id; - } - let id = self.producer.register_channel( - name, - ChannelContent::JsonRecord { - schema: Some(name.to_owned()), - }, - ); - self.channels.insert(name.to_owned(), id); - self.channel_names.insert(id, name.to_owned()); - id - } - - fn record_channel(&mut self) -> ChannelId { - if let Some(id) = self.channels.get(R::CHANNEL).copied() { - return id; - } - let id = self.producer.register_record::(); - self.channels.insert(R::CHANNEL.to_owned(), id); - self.channel_names.insert(id, R::CHANNEL.to_owned()); - id - } - - fn emit_event(&mut self, dashboard: Option<&DashboardSupport>, event: ProvisionEvent) { - let payload = serde_json::to_vec(&MyelinProvisionEventRecord::new(event)) - .expect("serialize provisioning event"); - self.emit_bytes(dashboard, MYELIN_PROVISIONING_EVENTS, payload); - } - - fn emit_log(&mut self, dashboard: Option<&DashboardSupport>, line: ProvisionLogLine) { - let channel = myelin_provision_log_channel(line.node_id, line.stream); - let payload = serde_json::to_vec(&MyelinProvisionLogRecord::new(line)) - .expect("serialize provision log"); - self.emit_bytes(dashboard, &channel, payload); - } - - fn emit_bootstrap( - &mut self, - dashboard: Option<&DashboardSupport>, - run_id: u64, - node_id: u64, - phase: &str, - status: &str, - detail: Value, - ) { - self.emit_bootstrap_to_channel( - dashboard, - MYELIN_ORCH_BOOTSTRAP, - run_id, - node_id, - phase, - status, - detail, - ); - } - - fn emit_bootstrap_to_channel( - &mut self, - dashboard: Option<&DashboardSupport>, - channel: &str, - run_id: u64, - node_id: u64, - phase: &str, - status: &str, - detail: Value, - ) { - let benchmark = benchmark::stamp("myelin-orchestrator"); - let payload = serde_json::to_vec(&json!({ - "schema_version": benchmark["schema_version"].clone(), - "type":"OrchBootstrap", - "event_type":"OrchBootstrap", - "event_name":phase, - "phase":phase, - "status":status, - "run_id":run_id, - "node_id":node_id, - "producer_component":benchmark["producer_component"].clone(), - "producer_instance_id":benchmark["producer_instance_id"].clone(), - "producer_process_id":benchmark["producer_process_id"].clone(), - "producer_sequence":benchmark["producer_sequence"].clone(), - "wall_clock_unix_ms":benchmark["wall_clock_unix_ms"].clone(), - "monotonic_ms":benchmark["monotonic_ms"].clone(), - "clock_source":benchmark["clock_source"].clone(), - "span_id":format!("myelin-orchestrator:{run_id}:{}:{phase}", benchmark["producer_sequence"]), - "parent_span_id":Value::Null, - "benchmark":benchmark, - "detail":detail, - })) - .expect("serialize orch bootstrap event"); - self.emit_bytes(dashboard, channel, payload); - } - - fn emit_prompt( - &mut self, - dashboard: Option<&DashboardSupport>, - run_id: u64, - node_id: u64, - request_id: u64, - phase: &str, - status: &str, - detail: Value, - ) { - let benchmark = benchmark::stamp("myelin-orchestrator"); - let payload = serde_json::to_vec(&json!({ - "schema_version": benchmark["schema_version"].clone(), - "type":"OrchPromptEvent", - "event_type":"OrchPromptEvent", - "event_name":phase, - "phase":phase, - "status":status, - "run_id":run_id, - "node_id":node_id, - "request_id":request_id, - "producer_component":benchmark["producer_component"].clone(), - "producer_instance_id":benchmark["producer_instance_id"].clone(), - "producer_process_id":benchmark["producer_process_id"].clone(), - "producer_sequence":benchmark["producer_sequence"].clone(), - "wall_clock_unix_ms":benchmark["wall_clock_unix_ms"].clone(), - "monotonic_ms":benchmark["monotonic_ms"].clone(), - "clock_source":benchmark["clock_source"].clone(), - "span_id":format!("myelin-orchestrator:{run_id}:{request_id}:{}:{phase}", benchmark["producer_sequence"]), - "parent_span_id":format!("request:{request_id}"), - "benchmark":benchmark, - "detail":detail, - })) - .expect("serialize orch prompt event"); - self.emit_bytes(dashboard, MYELIN_ORCH_PROMPT, payload); - } - - fn emit_record(&mut self, dashboard: Option<&DashboardSupport>, record: &R) { - let id = self.record_channel::(); - self.producer.submit_record(id, record); - self.flush(dashboard, "orchestrator"); - } - - fn emit_bytes( - &mut self, - dashboard: Option<&DashboardSupport>, - channel: &str, - payload: Vec, - ) { - self.emit_bytes_from(dashboard, channel, payload, "orchestrator"); - } - - fn emit_bytes_from( - &mut self, - dashboard: Option<&DashboardSupport>, - channel: &str, - payload: Vec, - source: &str, - ) { - let id = self.channel_by_name(channel); - self.producer.submit_bytes(id, payload); - self.flush(dashboard, source); - } - - fn flush(&mut self, dashboard: Option<&DashboardSupport>, source: &str) { - let stream = self.stream.clone(); - for frame in self.endpoint.mux().drain() { - let channel = self - .channel_names - .get(&frame.channel) - .cloned() - .unwrap_or_else(|| format!("channel#{}", frame.channel.0)); - ingest_dashboard_frame(dashboard, &stream, &channel, &frame); - self.archive_frame(source, &stream, &channel, &frame); - } - } - - fn archive_frame(&mut self, source: &str, stream: &StreamId, channel: &str, frame: &Frame) { - if let Some(archive) = &mut self.archive { - let _ = archive.record(source, stream, channel, frame); - } - } -} - struct OrchStdioCapture; struct OrchStdioLine { @@ -4066,60 +3612,6 @@ fn drain_orch_stdio_capture( } } -#[cfg(feature = "dashboard")] -struct DashboardSupport { - handle: dashboard::DashboardHandle, -} - -#[cfg(feature = "dashboard")] -impl DashboardSupport { - fn start(enabled: bool, engine: &EngineHandle) -> Result, String> { - if !enabled { - return Ok(None); - } - let mut config = dashboard::DashboardConfig::default(); - if let Some(port) = env_optional("MYELIN_DASHBOARD_PORT") { - config.port = port - .parse::() - .map_err(|e| format!("invalid MYELIN_DASHBOARD_PORT={port:?}: {e}"))?; - } - let handle = dashboard::DashboardHandle::new(config); - handle.register_view(Arc::new(MyelinClusterDashboardView::new())); - engine.spawn(handle.http_server()); - Ok(Some(Self { handle })) - } - - fn publish_frame(&self, stream: &StreamId, channel: &str, frame: &Frame) { - self.handle.publish(dashboard::FrameEvent { - stream: dashboard::StreamEvent { - node: stream.node.as_str().to_string(), - life: stream.life.0, - }, - channel: channel.to_owned(), - position: frame.position.0, - payload: frame.payload.clone(), - }); - } -} - -#[cfg(not(feature = "dashboard"))] -struct DashboardSupport; - -#[cfg(not(feature = "dashboard"))] -impl DashboardSupport { - fn start(enabled: bool, _engine: &EngineHandle) -> Result, String> { - if enabled { - return Err( - "MYELIN_DASHBOARD requires building myelin-system with feature dashboard" - .to_owned(), - ); - } - Ok(None) - } - - fn publish_frame(&self, _stream: &StreamId, _channel: &str, _frame: &Frame) {} -} - struct ChannelObservationSink { tx: Mutex>, } @@ -4254,8 +3746,7 @@ fn wait_for_weights_loaded(ctx: RuntimeReadyAckLoop<'_>, stage_index: u32) -> Re driver, stack: _, obs_rx, - frame_rx, - frame_tx, + collector, orchestrator_reports, stop_rx, dashboard, @@ -4267,7 +3758,7 @@ fn wait_for_weights_loaded(ctx: RuntimeReadyAckLoop<'_>, stage_index: u32) -> Re .. } = ctx; loop { - pump(driver, frame_tx); + collector.pump(driver); drain_orch_stdio_capture(orch_stdio_rx, orch_datastream, dashboard, run_id, node_id); if stop_requested(stop_rx) { return Err("shutdown requested while waiting for weights loaded".to_owned()); @@ -4275,7 +3766,12 @@ fn wait_for_weights_loaded(ctx: RuntimeReadyAckLoop<'_>, stage_index: u32) -> Re drain_observations_with_exit(obs_rx, dashboard, orch_datastream, provider, |_, status| { format!("node exited while loading weights: {status:?}") })?; - drain_frames(frame_rx, dashboard, orch_datastream); + collector.drain(|stream, channel, frame| { + if let Some(d) = dashboard { + d.publish_frame(stream, channel, frame); + } + orch_datastream.archive_frame("node", stream, channel, frame); + }); while let Some(report) = orchestrator_reports.try_recv() { match report { OrchestratorReport::WeightsReady { @@ -4914,8 +4410,7 @@ fn serve_prompts( driver, stack, obs_rx, - frame_rx, - frame_tx, + collector, stop_rx, dashboard, orch_datastream, @@ -4945,7 +4440,7 @@ fn serve_prompts( }; let mut active: Option = None; loop { - pump(driver, frame_tx); + collector.pump(driver); orch_datastream.flush(dashboard, "orchestrator"); if let Some(pipeline) = pipeline_runtime.as_mut() { pipeline.poll_driver(driver); @@ -4967,7 +4462,12 @@ fn serve_prompts( &provider, |_, status| format!("node exited: {status:?}"), )?; - drain_frames(frame_rx, dashboard, orch_datastream); + collector.drain(|stream, channel, frame| { + if let Some(d) = dashboard { + d.publish_frame(stream, channel, frame); + } + orch_datastream.archive_frame("node", stream, channel, frame); + }); drain_orch_stdio_capture(orch_stdio_rx, orch_datastream, dashboard, run_id, node_id); let swim_transitions = emit_swim_transitions(orch_datastream, dashboard, run_id, node_id, stack); @@ -4997,7 +4497,7 @@ fn serve_prompts( orchestrator_actor, OrchestratorMsg::ObserveOperatorStop { run_id }, ); - pump(driver, frame_tx); + collector.pump(driver); return Ok(()); } @@ -5299,58 +4799,6 @@ fn emit_plugin_observation( } } -fn drain_frames( - frame_rx: &mpsc::Receiver, - dashboard: Option<&DashboardSupport>, - orch_datastream: &mut OrchDatastream, -) { - while let Ok(collected) = frame_rx.try_recv() { - archive_collected_frame(collected, dashboard, orch_datastream); - } -} - -fn drain_frames_with_load_progress( - frame_rx: &mpsc::Receiver, - dashboard: Option<&DashboardSupport>, - orch_datastream: &mut OrchDatastream, - progress: &mut BTreeMap, -) { - while let Ok(collected) = frame_rx.try_recv() { - update_load_progress_from_frame(progress, &collected, Instant::now()); - archive_collected_frame(collected, dashboard, orch_datastream); - } -} - -fn archive_collected_frame( - collected: CollectedDatastreamFrame, - dashboard: Option<&DashboardSupport>, - orch_datastream: &mut OrchDatastream, -) { - ingest_dashboard_frame( - dashboard, - &collected.stream, - &collected.channel_name, - &collected.frame, - ); - orch_datastream.archive_frame( - "node", - &collected.stream, - &collected.channel_name, - &collected.frame, - ); -} - -fn ingest_dashboard_frame( - dashboard: Option<&DashboardSupport>, - stream: &StreamId, - channel: &str, - frame: &Frame, -) { - if let Some(dashboard) = dashboard { - dashboard.publish_frame(stream, channel, frame); - } -} - fn emit_swim_transitions( orch_datastream: &mut OrchDatastream, dashboard: Option<&DashboardSupport>, @@ -5404,15 +4852,7 @@ fn emit_swim_probe_events( } } -/// Drain iroh ingress/egress queues and datastream connections. Core -/// progression and protocol tick injection are owned by the engine (see -/// `spawn_protocol_ticker`); this only drains integration-owned queues -/// (ENGINE_SPEC.md). -fn pump(driver: &IrohDriver, frame_tx: &mpsc::Sender) { - drain_datastream_connections(driver, frame_tx); -} - -fn env_optional(name: &str) -> Option { +pub(crate) fn env_optional(name: &str) -> Option { std::env::var(name) .ok() .map(|value| value.trim().to_owned()) diff --git a/crates/dashboard/src/hardware_view.rs b/crates/dashboard/src/hardware_view.rs index 399a32c..6b86c75 100644 --- a/crates/dashboard/src/hardware_view.rs +++ b/crates/dashboard/src/hardware_view.rs @@ -9,7 +9,7 @@ use datastream::hardware::gpu::{ GpuDeviceSample, GpuProcessSample, HOST_GPU_CHANNEL, HostGpuSample, }; use datastream::hardware::net::{HOST_NET_CHANNEL, HostNetSample, NetInterfaceSample}; -use datastream::record::Record; +use datastream::Record; use parking_lot::RwLock; use serde::Serialize; use serde_json::{Value, json}; diff --git a/crates/datastream/src/emit.rs b/crates/datastream/src/emit.rs index 3c8c34a..aacac4d 100644 --- a/crates/datastream/src/emit.rs +++ b/crates/datastream/src/emit.rs @@ -7,9 +7,9 @@ use swactor::actor::ActorAddress; use swactor::process_observer::ProcessOutputObserver; use swactor::runtime::Runtime; -use super::frame::{ChannelId, Frame, Lifetime, NodeId, StreamId}; +use crate::frame::{ChannelId, Frame, Lifetime, NodeId, StreamId}; +use crate::record::Record; use super::mux::Mux; -use super::record::Record; use super::wire::{DatastreamFrame, encode_delivery}; /// Legacy sink for frames after mux drain has assigned positions. diff --git a/crates/datastream/src/lib.rs b/crates/datastream/src/lib.rs index b105a62..1109ab1 100644 --- a/crates/datastream/src/lib.rs +++ b/crates/datastream/src/lib.rs @@ -1,12 +1,29 @@ //! The per-node telemetry **datastream** (see `DATASTREAM_SPEC.md`). //! -//! A deliberately dumb pipe: producers dump bytes tagged by stream-local +//! A deliberately dumb pipe: producers dump bytes tagged with a stream-local //! channel id, a single per-node mux accepts those bytes and assigns canonical //! positions during drain, the endpoint broadcasts catalog-aware events to //! subscribers, ingest reconstructs streams by position, and views are //! read-time projections over stored frames. Nothing between a producer and a //! view interprets the payload. //! +//! ## Producer vs observer surface +//! +//! This crate has two surfaces: +//! +//! - **Producer** — re-exported at the crate root ([`DatastreamEndpoint`], +//! [`DatastreamProducer`], [`Record`], [`ChannelId`], [`StreamId`], …). +//! Everything control-plane and actor code needs to *emit* telemetry. +//! +//! - **Observer** — in submodules ([`frame::Frame`], [`frame::DatastreamEvent`], +//! [`store::Store`], [`views`], [`ingest::Consumer`]). Everything a sink +//! (dashboard, archive, transport) needs to *read* telemetry. +//! +//! The crate root deliberately does **not** re-export [`frame::Frame`] or +//! [`frame::DatastreamEvent`]. `use datastream::Frame` is a compile error; the +//! full path `datastream::frame::Frame` compiles but is banned in control-plane +//! modules by `cargo xtask check-telemetry-isolation`. +//! //! ```text //! producers (caller-owned records + text) //! │ bytes tagged by registered ChannelId @@ -46,17 +63,18 @@ pub mod transport; pub mod views; pub mod wire; +// ── Producer surface (re-exported at root; safe for control-plane code) ── + pub use endpoint::{ CatalogSnapshot, ChannelRegistrationError, DatastreamEndpoint, DatastreamProducer, DatastreamSnapshot, DatastreamSubscription, DeliveryFanout, EndpointTick, SubscriberSnapshot, - SubscriptionId, frame_event_to_delivery, + SubscriptionId, }; pub use frame::{ ChannelContent, ChannelContentKind, ChannelDescriptor, ChannelFilter, ChannelId, ChannelRef, - DatastreamEvent, Frame, FrameDelivery, Lifetime, NodeId, Position, SourceFilter, - StreamDescriptor, StreamId, StreamOrigin, SubscriptionRequest, + Lifetime, NodeId, Position, SourceFilter, StreamDescriptor, StreamId, StreamOrigin, + SubscriptionRequest, }; -pub use ingest::Consumer; pub use mux::Mux; pub use publisher_actor::{ DATASTREAM_PUBLISHER_NAME, DatastreamPublisherActor, DatastreamPublisherMsg, @@ -64,6 +82,11 @@ pub use publisher_actor::{ }; pub use record::{ChannelKind, ChannelRegistry, Record}; pub use sink_actor::{DATASTREAM_SINK_NAME, DatastreamSink}; -pub use store::{GapSpan, Store, StoredStream}; -pub use transport::{Delivery, Reorder, ScriptedTransport, StreamScript}; -pub use views::{Body, LogEntry, MergedFrame}; + +// ── Observer surface (in submodules; NOT re-exported at root) ── +// +// frame::Frame, frame::DatastreamEvent, frame::FrameDelivery, +// store::Store, ingest::Consumer, views::*, transport::Delivery +// +// Access these via their module paths (e.g. `datastream::frame::Frame`). +// Control-plane modules must not import them — enforced by CI. diff --git a/crates/datastream/src/mux.rs b/crates/datastream/src/mux.rs index 101c642..4e19a0f 100644 --- a/crates/datastream/src/mux.rs +++ b/crates/datastream/src/mux.rs @@ -7,7 +7,7 @@ use crossbeam_channel::{Receiver, Sender, TryRecvError, TrySendError, bounded}; use std::sync::atomic::{AtomicU64, Ordering}; -use super::frame::{ChannelId, Frame, Position, StreamId}; +use crate::frame::{ChannelId, Frame, Position, StreamId}; struct PendingFrame { channel: ChannelId, diff --git a/crates/datastream/src/sink_actor.rs b/crates/datastream/src/sink_actor.rs index d5d610a..734a208 100644 --- a/crates/datastream/src/sink_actor.rs +++ b/crates/datastream/src/sink_actor.rs @@ -14,7 +14,7 @@ use swactor::actor::ActorInterface; use swactor::runtime::Ctx; -use super::frame::{Frame, StreamId}; +use crate::frame::{Frame, StreamId}; use super::wire::{DatastreamFrame, decode_delivery}; /// Receives legacy [`DatastreamFrame`] cluster messages and folds each decoded diff --git a/crates/datastream/src/wire.rs b/crates/datastream/src/wire.rs index ad2e0f0..bae7e55 100644 --- a/crates/datastream/src/wire.rs +++ b/crates/datastream/src/wire.rs @@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize}; use swactor::Error; use swactor_transport::{Codec, CodecRegistry, NetworkMessage}; -use super::frame::{ChannelId, DatastreamEvent, Frame, Lifetime, NodeId, Position, StreamId}; +use crate::frame::{ChannelId, DatastreamEvent, Frame, Lifetime, NodeId, Position, StreamId}; /// Why a buffer could not be decoded as an envelope. #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/crates/datastream/tests/t_datastream.rs b/crates/datastream/tests/t_datastream.rs index 67830e1..5340309 100644 --- a/crates/datastream/tests/t_datastream.rs +++ b/crates/datastream/tests/t_datastream.rs @@ -1,14 +1,13 @@ use std::sync::Arc; use std::thread; +use datastream::frame::Frame; use datastream::ingest::Consumer; use datastream::mux::Mux; use datastream::transport::{Delivery, Reorder, ScriptedTransport, StreamScript}; use datastream::views::{self, Body, LogEntry}; use datastream::wire::{decode_delivery, encode_delivery}; -use datastream::{ - ChannelId, ChannelKind, ChannelRegistry, Frame, Lifetime, NodeId, Position, Record, StreamId, -}; +use datastream::{ChannelId, ChannelKind, ChannelRegistry, Lifetime, NodeId, Position, Record, StreamId}; use serde::{Deserialize, Serialize}; const RESOURCE_CHANNEL: ChannelId = ChannelId(1); diff --git a/crates/datastream/tests/t_datastream_endpoint.rs b/crates/datastream/tests/t_datastream_endpoint.rs index 3bafd54..9501535 100644 --- a/crates/datastream/tests/t_datastream_endpoint.rs +++ b/crates/datastream/tests/t_datastream_endpoint.rs @@ -2,9 +2,9 @@ use std::time::Duration; use datastream::{ ChannelContent, ChannelContentKind, ChannelFilter, ChannelId, DatastreamEndpoint, - DatastreamEvent, FrameDelivery, Lifetime, NodeId, Position, Record, SourceFilter, StreamId, - SubscriptionRequest, + Lifetime, NodeId, Position, Record, SourceFilter, StreamId, SubscriptionRequest, }; +use datastream::frame::{DatastreamEvent, FrameDelivery}; use serde_json::Value; use swactor::actor::ActorAddress; use swactor::stats::ActorSnapshot; diff --git a/crates/datastream/tests/t_datastream_realio.rs b/crates/datastream/tests/t_datastream_realio.rs index 8df84e5..f114b11 100644 --- a/crates/datastream/tests/t_datastream_realio.rs +++ b/crates/datastream/tests/t_datastream_realio.rs @@ -4,11 +4,12 @@ use std::collections::{HashMap, HashSet}; use std::net::UdpSocket; use std::time::Duration; +use datastream::frame::Frame; use datastream::ingest::Consumer; use datastream::mux::Mux; use datastream::transport::Delivery; use datastream::wire::{decode_delivery, encode_delivery}; -use datastream::{ChannelId, Frame, Lifetime, NodeId, Position, Record, StreamId}; +use datastream::{ChannelId, Lifetime, NodeId, Position, Record, StreamId}; use serde::{Deserialize, Serialize}; const RESOURCE_CHANNEL: ChannelId = ChannelId(1); diff --git a/crates/iroh-driver/src/datastream_transport.rs b/crates/iroh-driver/src/datastream_transport.rs index 4996567..63867aa 100644 --- a/crates/iroh-driver/src/datastream_transport.rs +++ b/crates/iroh-driver/src/datastream_transport.rs @@ -5,10 +5,9 @@ use std::sync::Arc; use std::time::Duration; use crossbeam_channel::TryRecvError; -use datastream::{ - ChannelDescriptor, ChannelId, ChannelRef, DatastreamEvent, DatastreamSnapshot, - DatastreamSubscription, FrameDelivery, Position, StreamDescriptor, -}; +use datastream::{DatastreamSnapshot, DatastreamSubscription}; +use datastream::frame::{ChannelDescriptor, ChannelId, ChannelRef, DatastreamEvent, FrameDelivery, Position, +StreamDescriptor,}; use iroh::endpoint::{Connection, RecvStream, SendStream}; use iroh::{Endpoint, EndpointAddr}; use swactor_engine::EngineHandle; diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 8fa5669..254acbe 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -266,6 +266,7 @@ COMMANDS: Run real cargo myelin-chat acceptance check and write benchmark artifacts. myelin-chat-compare Compare two benchmark summaries and report comparable deltas. + check-telemetry-isolation Verify no frame types appear in control-plane modules. test Run the basic non-binding test barrier: root crate plus each non-binding repository package with `cargo test -p`." ); @@ -6851,9 +6852,91 @@ mod tests { } } +/// Scan control-plane source files for forbidden telemetry frame type references. +/// The datastream is metrics/logging only; control decisions must never branch +/// on a frame. Frame types live in `datastream::frame::*` (not re-exported at +/// root) and must not appear in orchestration or other control modules. +fn check_telemetry_isolation() -> ExitCode { + /// Directories whose .rs files are control-plane: they must not touch + /// frame types or read-side modules. + const CONTROL_DIRS: &[&str] = &[ + "apps/myelin/src/orchestration", + "crates/distribution/src", + "crates/data-plane/src", + "crates/provisioning/src", + ]; + + /// Substrings that indicate a telemetry frame type or read-side module + /// has leaked into control code. `datastream::frame::` covers Frame, + /// DatastreamEvent, FrameDelivery, and every other frame-module type. + const FORBIDDEN: &[&str] = &[ + "datastream::frame::", + "datastream::store::", + "datastream::ingest::", + "datastream::views::", + "datastream::transport::", + "CollectedDatastreamFrame", + ]; + let mut violations = Vec::new(); + for dir in CONTROL_DIRS { + collect_rs_files(dir, &mut violations); + } + + let mut found = false; + for file in &violations { + let Ok(src) = std::fs::read_to_string(file) else { + continue; + }; + for (lineno, line) in src.lines().enumerate() { + for pat in FORBIDDEN { + if line.contains(pat) { + eprintln!( + "telemetry-isolation violation: {file}:{}: {}", + lineno + 1, + line.trim() + ); + found = true; + } + } + } + } + + if found { + eprintln!( + "\ntelemetry-isolation: control-plane code must not import frame types \ + or read-side modules. Use the datastream producer API (root re-exports) \ + for emitting telemetry, never `datastream::frame::*` for reading it." + ); + ExitCode::from(1) + } else { + println!("telemetry-isolation: OK — no frame types in control-plane modules."); + ExitCode::SUCCESS + } +} + +/// Recursively collect .rs file paths under `dir` into `out`. +fn collect_rs_files(dir: &str, out: &mut Vec) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + if let Some(s) = path.to_str() { + collect_rs_files(s, out); + } + } else if path.extension().is_some_and(|ext| ext == "rs") { + if let Some(s) = path.to_str() { + out.push(s.to_owned()); + } + } + } +} + fn main() -> ExitCode { let mut args = std::env::args().skip(1); match args.next().as_deref() { + Some("check-telemetry-isolation") => check_telemetry_isolation(), Some("test") if args.next().is_none() => run_tests(), Some("myelin-chat-check") => run_myelin_chat_check(args.collect()), Some("myelin-chat-compare") => run_myelin_chat_compare(args.collect()),