crates/knut-pipelines/src/bin/knutpipe.rs
//! `knutpipe` — view Spark Declarative Pipelines and Spark Connect jobs.
//!
//! The crate's front door. It is built on the **default** feature set, so the
//! static `show` view always works with no gRPC and no GUI. The live `watch`
//! view needs `--features grpc`; the desktop `gui` view needs `--features gui`
//! (and `grpc` too, for a live connection). Whatever the GUI can display, a
//! CLI subcommand can also print — the GUI is never the only way in.
use std::process::ExitCode;
use clap::{Parser, Subcommand};
use knut_pipelines::{demo_graph, dump_json, dump_text};
#[derive(Parser)]
#[command(
name = "knutpipe",
about = "View Spark Declarative Pipelines + Spark Connect jobs — live and static",
// `v<pkg> (<git-sha>)`, stamped pure-fs by build.rs → `knutpipe v<ver> (<hash>)`.
version = knut_pipelines::VERSION_LINE
)]
struct Cli {
/// Emit structured JSON log lines (machine-readable) for the run events a
/// test/bencher drives, instead of pretty text.
#[arg(long, global = true)]
json_logs: bool,
#[command(subcommand)]
command: Command,
}
#[derive(Subcommand)]
enum Command {
/// Render a dataflow graph (from a JSON file, or the built-in demo) as
/// text or JSON. The static view; no server needed.
Show {
/// Path to a graph JSON file (as produced by `--json`). Omit to render
/// a small built-in demo pipeline.
path: Option<String>,
/// Emit JSON instead of the text outline.
#[arg(long)]
json: bool,
},
/// Connect to a Spark Connect server and stream a pipeline run (or a SQL
/// job) live to the terminal. Requires `--features grpc`.
Watch {
/// Spark Connect URL, e.g. `sc://localhost:15002`.
url: String,
/// Run an existing dataflow graph by id (a pipeline run).
#[arg(long, value_name = "GRAPH_ID")]
graph: Option<String>,
/// Instead of a pipeline, execute this Spark SQL and show job progress.
#[arg(long, value_name = "SQL")]
sql: Option<String>,
/// Full-refresh the whole graph (pipeline runs only).
#[arg(long)]
full_refresh: bool,
},
/// Launch the desktop viewer (egui + node-graph). Requires `--features gui`
/// (add `grpc` for a live connection). Optionally preload a graph JSON
/// and/or connect to a server.
Gui {
/// Path to a graph JSON file to preload.
path: Option<String>,
/// Spark Connect URL to connect to for a live run.
#[arg(long, value_name = "URL")]
connect: Option<String>,
},
/// Serve the web viewer (axum + SSE): facett's Rust `pipeline` DAG viewer,
/// compiled to wasm (the `facet-wrapped` bundle), fed live `PipelineModel`
/// frames over Server-Sent Events. Requires `--features web`. With NO
/// `--connect` it serves the built-in demo graph and replays a synthetic run
/// so the DAG animates without a Spark server; with `--connect` (needs
/// `grpc`) it streams a live run.
Web {
/// Path to a graph JSON file to serve. Omit for the built-in demo.
path: Option<String>,
/// Spark Connect URL for a live run (needs `--features grpc`).
#[arg(long, value_name = "URL")]
connect: Option<String>,
/// Address to bind the HTTP server to.
#[arg(long, default_value = "127.0.0.1:8787")]
addr: String,
/// Directory holding the facet-wrapped bundle: `index.html`,
/// `facet_wrapped.js`, `facet_wrapped_bg.wasm` (facett: `cargo xtask
/// wrap-wasm` + the loader page). Defaults to `$FACET_WRAPPED_DIST`. The
/// server starts either way; without a bundle the page says what is
/// missing and how to build it, and the data routes still answer.
#[arg(long, value_name = "DIR")]
facet_dist: Option<std::path::PathBuf>,
},
/// Define a Spark **Declarative Pipeline** from a spec and run it, streaming
/// its events into the viewer (terminal, or `--gui` for the node graph).
/// Requires `--features grpc`. The spec is a dataflow-graph JSON whose
/// flows carry a `query` (see `examples/pipeline.json`).
Pipeline {
/// Spark Connect URL, e.g. `sc://localhost:15002`.
url: String,
/// Path to the pipeline spec (dataflow graph JSON with flow queries).
#[arg(long)]
spec: String,
/// Pipeline storage root URI (required to materialise outputs), e.g.
/// `file:///tmp/knut_pipeline`.
#[arg(long, default_value = "file:///tmp/knut_pipeline")]
storage: String,
/// Show it in the desktop GUI instead of the terminal (needs `gui`).
#[arg(long)]
gui: bool,
},
}
fn main() -> ExitCode {
let cli = Cli::parse();
// Shared knut bootstrap (one copy for bifröst + knutpipe; see `knut-tracing`).
// The Spark Connect client logs connect/run/event/progress under the
// `knutpipe::*` targets; level from `RUST_LOG` (default `info`).
knut_tracing::init_tracing(cli.json_logs);
match run(cli) {
Ok(()) => ExitCode::SUCCESS,
Err(e) => {
eprintln!("error: {e}");
ExitCode::FAILURE
}
}
}
fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> {
match cli.command {
Command::Show { path, json } => cmd_show(path, json),
Command::Watch {
url,
graph,
sql,
full_refresh,
} => cmd_watch(url, graph, sql, full_refresh),
Command::Gui { path, connect } => cmd_gui(path, connect),
Command::Web {
path,
connect,
addr,
facet_dist,
} => cmd_web(path, connect, addr, facet_dist),
Command::Pipeline {
url,
spec,
storage,
gui,
} => cmd_pipeline(url, spec, storage, gui),
}
}
/// `show` — static render. Always available.
fn cmd_show(path: Option<String>, json: bool) -> Result<(), Box<dyn std::error::Error>> {
let mut graph: knut_pipelines::DataflowGraph = match path {
Some(p) => serde_json::from_str(&std::fs::read_to_string(&p)?)?,
None => demo_graph(),
};
// Complete the DAG's edges by walking each flow's defining relation (its
// SQL query) — so a spec that only carries queries still renders full
// edges, without a server to replay the definitions against.
graph.populate_reads();
if json {
println!("{}", dump_json(&graph)?);
} else {
print!("{}", dump_text(&graph));
}
knut_pipelines::functional_status(
"knut-pipelines/cli_show",
"render_dag",
true,
if json { "json" } else { "text" },
);
Ok(())
}
// ── watch (live) — gRPC feature ────────────────────────────────────────────
#[cfg(feature = "grpc")]
fn cmd_watch(
url: String,
graph: Option<String>,
sql: Option<String>,
full_refresh: bool,
) -> Result<(), Box<dyn std::error::Error>> {
use futures_util::StreamExt;
use knut_pipelines::{PipelineRun, PipelinesClient, ViewerState, ViewerUpdate};
use knut_pipelines::viewer;
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
rt.block_on(async move {
let mut client = PipelinesClient::connect(&url).await?;
knut_pipelines::functional_status("knut-pipelines/cli_watch", "connect", true, &url);
let mut state = ViewerState::default();
// Two live sources, one unified stream of ViewerUpdates.
let mut stream: std::pin::Pin<Box<dyn futures_core::Stream<Item = _> + Send>> =
if let Some(sql) = sql {
eprintln!("# executing SQL on {url}");
Box::pin(client.execute_sql(&sql).await?)
} else {
let run = PipelineRun {
graph_id: graph.clone(),
full_refresh_all: full_refresh,
..Default::default()
};
eprintln!("# starting run on {url} (graph {:?})", graph);
Box::pin(client.run_updates(&run).await?)
};
while let Some(item) = stream.next().await {
let update = item?;
// Echo each update as it arrives (the live log), then fold it in.
match &update {
ViewerUpdate::Event(ev) => {
println!(
"{}",
viewer::StdoutEventSink::<std::io::Stdout>::format_line(ev)
);
}
ViewerUpdate::Progress(p) => {
println!(
" [progress] {}% — {}",
(p.fraction() * 100.0).round() as u32,
p.summary()
);
}
ViewerUpdate::GraphReplaced(_) => {}
}
state.apply(update);
}
// Final consolidated state — the same picture the GUI would settle on.
println!("\n=== final state ===");
print!("{}", viewer::dump_state(&state, 0));
Ok::<(), Box<dyn std::error::Error>>(())
})
}
#[cfg(not(feature = "grpc"))]
fn cmd_watch(
_url: String,
_graph: Option<String>,
_sql: Option<String>,
_full_refresh: bool,
) -> Result<(), Box<dyn std::error::Error>> {
Err("`watch` needs the live client — rebuild with `--features grpc`".into())
}
// ── gui (desktop) — gui feature ────────────────────────────────────────────
#[cfg(feature = "gui")]
fn cmd_gui(
path: Option<String>,
connect: Option<String>,
) -> Result<(), Box<dyn std::error::Error>> {
use knut_pipelines::ViewerState;
let state = match path {
Some(p) => ViewerState::new(serde_json::from_str(&std::fs::read_to_string(&p)?)?),
None => ViewerState::new(demo_graph()),
};
knut_pipelines::gui::launch(state, connect)?;
Ok(())
}
#[cfg(not(feature = "gui"))]
fn cmd_gui(
_path: Option<String>,
_connect: Option<String>,
) -> Result<(), Box<dyn std::error::Error>> {
Err(
"`gui` needs the desktop viewer — rebuild with `--features gui` (add `grpc` for live)"
.into(),
)
}
// ── web (axum + SSE viewer) — web feature ────────────────────────────────────
#[cfg(feature = "web")]
fn cmd_web(
path: Option<String>,
connect: Option<String>,
addr: String,
facet_dist: Option<std::path::PathBuf>,
) -> Result<(), Box<dyn std::error::Error>> {
use knut_pipelines::DataflowGraph;
use knut_pipelines::web::{bundle_from_env, load_bundle};
let mut graph: DataflowGraph = match path {
Some(p) => serde_json::from_str(&std::fs::read_to_string(&p)?)?,
None => demo_graph(),
};
// Complete edges by walking flow relations, so the served DAG is whole.
graph.populate_reads();
#[cfg(not(feature = "grpc"))]
if connect.is_some() {
return Err("`web --connect` needs the live client — rebuild with `--features grpc` (or add `web,grpc`)".into());
}
let socket: std::net::SocketAddr = addr.parse()?;
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
// The viewer's bytes: an explicit dir wins, else `$FACET_WRAPPED_DIST`. A
// refusal is served, not swallowed — `serve_bundle` prints it and every page
// route answers with it.
let bundle = match facet_dist {
Some(dir) => load_bundle(dir),
None => bundle_from_env(),
};
knut_pipelines::functional_status("knut-pipelines/cli_web", "serve", true, &addr);
rt.block_on(knut_pipelines::web::serve_bundle(graph, connect, socket, bundle))?;
Ok(())
}
#[cfg(not(feature = "web"))]
fn cmd_web(
_path: Option<String>,
_connect: Option<String>,
_addr: String,
_facet_dist: Option<std::path::PathBuf>,
) -> Result<(), Box<dyn std::error::Error>> {
Err(
"`web` needs the web viewer — rebuild with `--features web` (add `grpc` for `--connect`)"
.into(),
)
}
// ── pipeline (define + run an SDP) — grpc (+ optional gui) ───────────────────
#[cfg(feature = "grpc")]
fn cmd_pipeline(
url: String,
spec: String,
storage: String,
gui: bool,
) -> Result<(), Box<dyn std::error::Error>> {
use knut_pipelines::{DataflowGraph, ViewerState};
let mut graph: DataflowGraph = serde_json::from_str(&std::fs::read_to_string(&spec)?)?;
graph.populate_reads();
graph.validate()?;
if gui {
#[cfg(feature = "gui")]
{
knut_pipelines::gui::launch_live_pipeline(
ViewerState::new(graph.clone()),
url,
graph,
Some(storage),
)?;
return Ok(());
}
#[cfg(not(feature = "gui"))]
return Err("`pipeline --gui` needs the desktop viewer — add `--features gui`".into());
}
// Terminal: define the graph, run it, stream the SDP events.
use futures_util::StreamExt;
use knut_pipelines::{PipelineRun, PipelinesClient, ViewerUpdate, viewer};
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
rt.block_on(async move {
let mut client = PipelinesClient::connect(&url).await?;
eprintln!(
"# defining pipeline ({} datasets, {} flows)",
graph.datasets.len(),
graph.flows.len()
);
let graph_id = client.define_graph(&graph).await?;
eprintln!("# graph {graph_id}; starting run");
let run = PipelineRun {
graph_id: Some(graph_id),
full_refresh_all: true,
storage: Some(storage),
..Default::default()
};
let mut state = ViewerState::new(graph);
let mut stream = client.run_updates(&run).await?;
while let Some(item) = stream.next().await {
let update = item?;
if let ViewerUpdate::Event(ev) = &update {
println!(
"{}",
viewer::StdoutEventSink::<std::io::Stdout>::format_line(ev)
);
}
state.apply(update);
}
println!("\n=== final state ===");
print!("{}", viewer::dump_state(&state, 0));
Ok::<(), Box<dyn std::error::Error>>(())
})
}
#[cfg(not(feature = "grpc"))]
fn cmd_pipeline(
_url: String,
_spec: String,
_storage: String,
_gui: bool,
) -> Result<(), Box<dyn std::error::Error>> {
Err("`pipeline` needs the live client — rebuild with `--features grpc`".into())
}