crates/knut-pipelines/src/viewer.rs
//! Rendering the dataflow graph and run events.
//!
//! Two halves, both dependency-light and independent of the gRPC client:
//!
//! * **Static**: [`dump_text`] / [`dump_json`] render a fetched
//! [`DataflowGraph`] as a human-readable tree or as JSON.
//! * **Live**: the [`PipelineEventSink`] trait is where a run's
//! [`PipelineRunEvent`]s flow as they arrive. [`StdoutEventSink`] is the
//! seed implementation; an egui/ratatui/web live viewer is just another
//! sink (see `docs/pipelines-viewer.md`).
use std::io::{self, Write};
use crate::model::{DataflowGraph, EventLevel, PipelineRunEvent};
use crate::progress::{NodeStatus, ViewerState};
/// Serialize a graph to pretty JSON.
pub fn dump_json(graph: &DataflowGraph) -> crate::Result<String> {
Ok(serde_json::to_string_pretty(graph)?)
}
/// Render a graph as an indented text outline: each dataset node with its
/// type and the flows feeding it, followed by the topological order when the
/// graph is acyclic.
///
/// This is intentionally a pure `String` builder (no IO) so it is trivially
/// testable and reusable by any frontend.
pub fn dump_text(graph: &DataflowGraph) -> String {
use std::fmt::Write as _;
let mut s = String::new();
let id = graph.graph_id.as_deref().unwrap_or("<unassigned>");
let _ = writeln!(s, "Dataflow graph {id}");
if let Some(c) = &graph.default_catalog {
let _ = writeln!(s, " default catalog: {c}");
}
if let Some(d) = &graph.default_database {
let _ = writeln!(s, " default database: {d}");
}
let _ = writeln!(
s,
" {} dataset(s), {} flow(s)",
graph.datasets.len(),
graph.flows.len()
);
for d in &graph.datasets {
let _ = write!(s, " [{}] {}", d.output_type.label(), d.name);
if let Some(fmt) = &d.format {
let _ = write!(s, " (format={fmt})");
}
let _ = writeln!(s);
if let Some(c) = &d.comment {
let _ = writeln!(s, " # {c}");
}
for f in graph.flows_into(&d.name) {
let once = if f.once { " [once]" } else { "" };
if f.reads.is_empty() {
let _ = writeln!(s, " <- flow {}{once} (external source)", f.name);
} else {
let _ = writeln!(
s,
" <- flow {}{once} reads: {}",
f.name,
f.reads.join(", ")
);
}
}
}
match graph.topo_order() {
Some(order) if !order.is_empty() => {
let _ = writeln!(s, " topological order: {}", order.join(" -> "));
}
Some(_) => {}
None => {
let _ = writeln!(s, " topological order: <cycle detected>");
}
}
s
}
/// Render a full [`ViewerState`] — the graph annotated with live run status,
/// the latest Spark job progress, and a tail of the event log — as text. This
/// is the CLI counterpart of the GUI: the same state, drawn with characters.
///
/// `event_tail` caps how many of the most recent events to show (0 = none).
pub fn dump_state(state: &ViewerState, event_tail: usize) -> String {
use std::fmt::Write as _;
let mut s = String::new();
// The DAG with a status glyph per node.
let id = state.graph.graph_id.as_deref().unwrap_or("<unassigned>");
let (idle, running, done, failed) = state.status_counts();
let _ = writeln!(
s,
"Dataflow graph {id} — {} dataset(s), {} flow(s) [{done} done · {running} running · {failed} failed · {idle} idle]",
state.graph.datasets.len(),
state.graph.flows.len(),
);
for name in state.graph.topo_order().unwrap_or_else(|| {
state
.graph
.datasets
.iter()
.map(|d| d.name.clone())
.collect()
}) {
let Some(d) = state.graph.dataset(&name) else {
continue;
};
let st = state.status_of(&name);
let rows = match state.rows_written.get(&name) {
Some(n) => format!(", {n} rows"),
None => String::new(),
};
let _ = writeln!(
s,
" {} [{}] {} ({}{rows})",
status_glyph(st),
d.output_type.label(),
name,
st.label()
);
for f in state.graph.flows_into(&name) {
if f.reads.is_empty() {
let _ = writeln!(s, " <- {} (external source)", f.name);
} else {
let _ = writeln!(s, " <- {} reads: {}", f.name, f.reads.join(", "));
}
}
}
// Latest Spark job progress.
if let Some(p) = &state.progress {
let pct = (p.fraction() * 100.0).round() as u32;
let _ = writeln!(s, " job progress: {pct}% — {}", p.summary());
for stage in &p.stages {
let mark = if stage.done { "✓" } else { " " };
let _ = writeln!(
s,
" [{mark}] stage {} — {}/{} tasks",
stage.stage_id, stage.num_completed_tasks, stage.num_tasks,
);
}
}
// Event tail.
if event_tail > 0 && !state.events.is_empty() {
let start = state.events.len().saturating_sub(event_tail);
let _ = writeln!(s, " events (last {}):", state.events.len() - start);
for ev in &state.events[start..] {
let _ = writeln!(s, " {}", StdoutEventSink::<io::Stdout>::format_line(ev));
}
}
s
}
/// A single-character status indicator for a node.
fn status_glyph(status: NodeStatus) -> char {
match status {
NodeStatus::Idle => '·',
NodeStatus::Running => '▶',
NodeStatus::Done => '✓',
NodeStatus::Failed => '✗',
}
}
/// Somewhere [`PipelineRunEvent`]s are delivered as a run progresses.
///
/// The client streams events off the wire and pushes each into a sink; the
/// sink decides how to present it (print, append to a TUI table, push over a
/// websocket, …). Implementors should be cheap per event so the live stream
/// stays responsive.
pub trait PipelineEventSink {
/// Handle one event. Errors are surfaced to the caller, which may choose
/// to stop the stream.
fn on_event(&mut self, event: &PipelineRunEvent) -> crate::Result<()>;
/// Called once when the run's event stream ends (normally or via error).
/// Default: nothing.
fn on_finish(&mut self) -> crate::Result<()> {
Ok(())
}
}
/// The seed live viewer: prints each event as one line to a writer.
///
/// Format: `LEVEL [timestamp] message (element)`. Generic over the writer
/// so tests can capture into a `Vec<u8>` and a real run can use stdout via
/// [`StdoutEventSink::new`].
pub struct StdoutEventSink<W: Write> {
out: W,
count: usize,
}
impl StdoutEventSink<io::Stdout> {
/// A sink writing to the process's stdout.
pub fn new() -> Self {
StdoutEventSink {
out: io::stdout(),
count: 0,
}
}
}
impl Default for StdoutEventSink<io::Stdout> {
fn default() -> Self {
Self::new()
}
}
impl<W: Write> StdoutEventSink<W> {
/// A sink writing to an arbitrary writer (tests, files, pipes).
pub fn to_writer(out: W) -> Self {
StdoutEventSink { out, count: 0 }
}
/// Number of events seen so far.
pub fn count(&self) -> usize {
self.count
}
/// Format one event into a single line (no trailing newline). Exposed so
/// other sinks can reuse the canonical one-line form.
pub fn format_line(event: &PipelineRunEvent) -> String {
let level = match event.level {
EventLevel::Info => "INFO ",
EventLevel::Warn => "WARN ",
EventLevel::Error => "ERROR",
};
let ts = event.timestamp.as_deref().unwrap_or("-");
let mut line = format!("{level} [{ts}] {}", event.message);
if let Some(e) = &event.element {
use std::fmt::Write as _;
// Write straight into the buffer; no intermediate `format!` String.
let _ = write!(line, " ({e})");
}
line
}
}
impl<W: Write> PipelineEventSink for StdoutEventSink<W> {
fn on_event(&mut self, event: &PipelineRunEvent) -> crate::Result<()> {
self.count += 1;
let r = writeln!(self.out, "{}", Self::format_line(event));
crate::functional_status(
"knut-pipelines/event_sink",
"on_event",
r.is_ok(),
&event.message,
);
r?;
Ok(())
}
fn on_finish(&mut self) -> crate::Result<()> {
writeln!(self.out, "-- run stream ended ({} event(s)) --", self.count)?;
self.out.flush()?;
Ok(())
}
}