From 0defac66e61010048e3fa30380e37df79c263234 Mon Sep 17 00:00:00 2001 From: Sheng Kun Chang Date: Tue, 30 Jun 2026 03:33:39 +0800 Subject: [PATCH] feat(dev): ship runner module logs to dispatch (log pipeline) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Slice 3 of the module log pipeline — the dev runner's write path. The in-container `dev --all` runner already line-reads each module's stdout/stderr (spawn_forwarder); now it also parses each line and ships it to dispatch's ingest, which buffers it into the per-module ring the developer console reads. - log_shipper: parse_line turns a line into {ts,level,msg,app_id} — extracting the SDK's structured JSON (slog) fields, falling back to the stream (stderr→warn, stdout→info) for non-JSON lines. A per-module background thread batches entries (cap 200 / ~1s) and POSTs them to {MS_DISPATCH_URL}/internal/modules/{id}/logs with X-MS-Service-Secret == MS_INTERNAL_SECRET (the host-minted per-session secret the ingest binds to). Best-effort: a failed or unconfigured ship never blocks the module. - app_id rides each entry so the app-side console can filter a shared dev module's logs per app (dispatch ignores the field until the Go Entry.AppID follow-up). Verified: cargo check + cargo test (5 log_shipper tests) green; cargo fmt clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/commands/dev/log_shipper.rs | 214 ++++++++++++++++++++++++++++++++ src/commands/dev/mod.rs | 35 +++++- 2 files changed, 246 insertions(+), 3 deletions(-) create mode 100644 src/commands/dev/log_shipper.rs diff --git a/src/commands/dev/log_shipper.rs b/src/commands/dev/log_shipper.rs new file mode 100644 index 0000000..fd68cbe --- /dev/null +++ b/src/commands/dev/log_shipper.rs @@ -0,0 +1,214 @@ +//! Dev log shipper: the in-container `dev --all` runner taps each module's +//! stdout/stderr (see `spawn_forwarder`) and POSTs batched lines to dispatch's +//! ingest endpoint, which buffers them into a per-module Redis ring that the +//! developer console's Logcat reads. Best-effort throughout — a failed or +//! unconfigured ship never blocks the module. Deployed modules use CloudWatch, +//! not this path. + +use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender}; +use std::thread; +use std::time::Duration; + +use serde::Serialize; + +const FLUSH_INTERVAL: Duration = Duration::from_millis(1000); +const MAX_BATCH: usize = 200; + +/// One log line in the {ts,level,msg} shape the web Logcat renders, plus the +/// trusted app the line was emitted under (for the app-side console's per-app +/// filtering of a shared dev module's logs). +#[derive(Serialize, Clone, Debug, PartialEq)] +pub struct LogEntry { + pub ts: String, + pub level: String, + pub msg: String, + /// From the SDK's structured log (`app_id`); empty for non-JSON lines. + /// Omitted from the wire when empty. + #[serde(skip_serializing_if = "String::is_empty")] + pub app_id: String, +} + +#[derive(Serialize)] +struct Batch<'a> { + entries: &'a [LogEntry], +} + +/// The channel the forwarder pushes parsed lines into. +pub type LogSink = Sender; + +/// parse_line turns a raw output line into a LogEntry. The SDK emits structured +/// JSON (slog) — extract its level/msg/time/app_id; any other line (a third-party +/// print, a panic, the SDK's own stderr diagnostics) falls back to the stream: +/// stderr → warn, stdout → info, with the raw line as the message. +pub fn parse_line(raw: &str, is_stderr: bool) -> LogEntry { + if let Ok(serde_json::Value::Object(obj)) = serde_json::from_str::(raw) { + if obj.contains_key("level") { + let level = obj + .get("level") + .and_then(|v| v.as_str()) + .map(normalize_level) + .unwrap_or_else(|| default_level(is_stderr)); + let msg = obj + .get("msg") + .or_else(|| obj.get("message")) + .and_then(|v| v.as_str()) + .unwrap_or(raw) + .to_string(); + let ts = obj + .get("time") + .and_then(|v| v.as_str()) + .map(short_time) + .unwrap_or_default(); + let app_id = obj + .get("app_id") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + return LogEntry { + ts, + level, + msg, + app_id, + }; + } + } + LogEntry { + ts: String::new(), + level: default_level(is_stderr), + msg: raw.to_string(), + app_id: String::new(), + } +} + +fn default_level(is_stderr: bool) -> String { + if is_stderr { "warn" } else { "info" }.to_string() +} + +fn normalize_level(s: &str) -> String { + match s.to_ascii_lowercase().as_str() { + "warn" | "warning" => "warn", + "error" | "fatal" => "error", + _ => "info", + } + .to_string() +} + +/// short_time pulls HH:MM:SS out of an RFC3339/slog time string, best-effort. +/// "2026-06-30T12:48:21.123Z" → "12:48:21"; a non-RFC3339 value passes through. +fn short_time(t: &str) -> String { + match t.split_once('T') { + Some((_, rest)) => rest.chars().take(8).collect(), + None => t.to_string(), + } +} + +/// spawn starts a per-module shipper thread that batches entries from the +/// returned sink and POSTs them to dispatch. Returns None (shipping disabled) +/// when no dispatch URL or secret is configured — dev logs then just stay in the +/// terminal, unchanged. +pub fn spawn( + client: reqwest::blocking::Client, + base_url: String, + module_id: String, + secret: String, +) -> Option { + if base_url.is_empty() || secret.is_empty() { + return None; + } + let url = format!( + "{}/internal/modules/{}/logs", + base_url.trim_end_matches('/'), + module_id + ); + let (tx, rx) = mpsc::channel::(); + thread::spawn(move || run(rx, client, url, secret)); + Some(tx) +} + +fn run(rx: Receiver, client: reqwest::blocking::Client, url: String, secret: String) { + let mut buf: Vec = Vec::new(); + loop { + match rx.recv_timeout(FLUSH_INTERVAL) { + Ok(entry) => { + buf.push(entry); + while buf.len() < MAX_BATCH { + match rx.try_recv() { + Ok(e) => buf.push(e), + Err(_) => break, + } + } + if buf.len() >= MAX_BATCH { + flush(&client, &url, &secret, &mut buf); + } + } + Err(RecvTimeoutError::Timeout) => flush(&client, &url, &secret, &mut buf), + Err(RecvTimeoutError::Disconnected) => { + flush(&client, &url, &secret, &mut buf); + return; + } + } + } +} + +fn flush(client: &reqwest::blocking::Client, url: &str, secret: &str, buf: &mut Vec) { + if buf.is_empty() { + return; + } + // Best-effort: a transport failure drops the batch rather than blocking the + // module or growing the buffer unboundedly. + let _ = client + .post(url) + .header("X-MS-Service-Secret", secret) + .json(&Batch { entries: buf }) + .send(); + buf.clear(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_json_line() { + let raw = r#"{"time":"2026-06-30T12:48:21.123Z","level":"WARN","msg":"slow query","app_id":"app-7"}"#; + let e = parse_line(raw, false); + assert_eq!(e.level, "warn"); + assert_eq!(e.msg, "slow query"); + assert_eq!(e.ts, "12:48:21"); + assert_eq!(e.app_id, "app-7"); + } + + #[test] + fn parse_json_message_alias_and_error() { + let e = parse_line(r#"{"level":"error","message":"boom"}"#, false); + assert_eq!(e.level, "error"); + assert_eq!(e.msg, "boom"); + assert_eq!(e.ts, ""); + assert_eq!(e.app_id, ""); + } + + #[test] + fn parse_non_json_falls_back_to_stream() { + let out = parse_line("plain stdout line", false); + assert_eq!(out.level, "info"); + assert_eq!(out.msg, "plain stdout line"); + + let err = parse_line("go: cannot find module", true); + assert_eq!(err.level, "warn"); + assert_eq!(err.msg, "go: cannot find module"); + } + + #[test] + fn normalize_levels() { + assert_eq!(normalize_level("DEBUG"), "info"); + assert_eq!(normalize_level("Warning"), "warn"); + assert_eq!(normalize_level("FATAL"), "error"); + } + + #[test] + fn spawn_disabled_without_config() { + let c = reqwest::blocking::Client::new(); + assert!(spawn(c.clone(), String::new(), "m1".into(), "s".into()).is_none()); + assert!(spawn(c, "http://x".into(), "m1".into(), String::new()).is_none()); + } +} diff --git a/src/commands/dev/mod.rs b/src/commands/dev/mod.rs index 642a2fa..0ca21bc 100644 --- a/src/commands/dev/mod.rs +++ b/src/commands/dev/mod.rs @@ -42,6 +42,7 @@ use super::{ }; use crate::{api, credentials, http}; +mod log_shipper; pub(crate) mod module_meta; mod tunnel; mod workspace; @@ -254,6 +255,15 @@ fn run_inner(root: &Path, _args: &DevArgs) -> Result<()> { "postgres://mirrorstack:mirrorstack@postgres:5432/ms_app_modules?sslmode=disable".into() }); + // Dev log shipping: tap each module's stdout/stderr and POST batches to + // dispatch's ingest, which the developer console reads. Reuses the runner's + // MS_DISPATCH_URL (container → dispatch) and MS_INTERNAL_SECRET (== the live + // tunnel session's secret, the ingest's auth). Either unset → shipping is + // disabled and logs just stay in the terminal. + let log_dispatch_url = std::env::var("MS_DISPATCH_URL").unwrap_or_default(); + let log_secret = std::env::var("MS_INTERNAL_SECRET").unwrap_or_default(); + let log_client = http::client(Duration::from_secs(5)).ok(); + eprintln!( "{} starting {} {}", ok_mark(), @@ -299,8 +309,17 @@ fn run_inner(root: &Path, _args: &DevArgs) -> Result<()> { let stdout = child.stdout.take().unwrap(); let stderr = child.stderr.take().unwrap(); let label: &'static str = Box::leak(slug.clone().into_boxed_str()); - spawn_forwarder(stdout, label, false); - spawn_forwarder(stderr, label, true); + let sink = match (&log_client, module_meta::read_module_id(&m.abs_dir)) { + (Some(client), Ok(module_id)) => log_shipper::spawn( + client.clone(), + log_dispatch_url.clone(), + module_id, + log_secret.clone(), + ), + _ => None, + }; + spawn_forwarder(stdout, label, false, sink.clone()); + spawn_forwarder(stderr, label, true, sink); eprintln!(" {} {}", style("✓").green(), slug); children.push((slug, child)); @@ -360,7 +379,12 @@ fn run_inner(root: &Path, _args: &DevArgs) -> Result<()> { Ok(()) } -fn spawn_forwarder(reader: R, label: &'static str, is_stderr: bool) { +fn spawn_forwarder( + reader: R, + label: &'static str, + is_stderr: bool, + sink: Option, +) { thread::spawn(move || { let prefix = line_prefix(label); let mut buf = BufReader::new(reader); @@ -371,6 +395,11 @@ fn spawn_forwarder(reader: R, label: &'static str, is_ Ok(0) => return, Ok(_) => { let trimmed = line.trim_end_matches(['\n', '\r']); + // Ship to the developer console (best-effort), then mirror to + // the terminal as before. + if let Some(s) = &sink { + let _ = s.send(log_shipper::parse_line(trimmed, is_stderr)); + } if is_stderr { eprintln!("{prefix} {trimmed}"); } else {