crates/knut-pipelines/tests/integration_spark_pipelines.rs
//! The pipeline viewer's **Spark Declarative Pipelines** path against a real
//! Spark Connect server.
//!
//! Boots a Spark 4.x Connect server (the stock `apache/spark:4.1.x` image ships
//! the `spark-pipelines` jar, so the `spark.pipelines` gRPC handler is live),
//! points [`PipelinesClient`] at it, loads the demo dataflow graph from
//! `examples/pipeline.json`, defines it (`CreateDataflowGraph` + `DefineOutput`
//! + `DefineFlow{WriteRelationFlowDetails}`), starts a full-refresh run, and
//! asserts a stream of [`PipelineRunEvent`]s comes back and the run reaches
//! `COMPLETED` — the exact path `knutpipe pipeline …` / the live GUI drive.
//!
//! This complements `integration_spark_connect.rs` (which proves the SQL
//! job-progress path); here the *pipelines* define/run contract is the subject.
//!
//! ## Prerequisites
//! - Podman socket: `systemctl --user start podman.socket`
//! - Image (override with `KNUT_SPARK_IMAGE`): defaults to the blessed GA
//! `docker.io/apache/spark:4.1.2-scala2.13-java21-python3-ubuntu` (pairs with
//! `iceberg-spark-runtime-4.1_2.13:1.11.0`); set `KNUT_SPARK_IMAGE` to
//! `…:4.2.0-…` for the 4.2.0 GA line. Either requires no build; a
//! locally built `localhost/knut-spark-iceberg` also works.
//!
//! ## Run
//! ```bash
//! cargo test -p knut-pipelines --features grpc --test integration_spark_pipelines \
//! -- --ignored --nocapture
//! ```
#![cfg(feature = "grpc")]
use std::time::Duration;
use knut_pipelines::{
DataflowGraph, EventLevel, PipelineEventSink, PipelineRun, PipelineRunEvent, PipelinesClient,
};
use testcontainers::{GenericImage, ImageExt, core::ContainerPort, runners::AsyncRunner};
/// Collects a pipeline run's events so the test can assert it actually ran.
#[derive(Default)]
struct CollectingSink {
events: Vec<PipelineRunEvent>,
}
impl PipelineEventSink for CollectingSink {
fn on_event(&mut self, event: &PipelineRunEvent) -> knut_pipelines::Result<()> {
self.events.push(event.clone());
Ok(())
}
}
fn ensure_docker_host() {
// Zero-shell: resolve the rootless podman socket from XDG_RUNTIME_DIR
// (which is `/run/user/<uid>`) rather than shelling out to `id -u`.
if std::env::var("DOCKER_HOST").is_err() {
let runtime = std::env::var("XDG_RUNTIME_DIR").unwrap_or_else(|_| "/run/user/1000".into());
let sock = format!("{runtime}/podman/podman.sock");
if std::path::Path::new(&sock).exists() {
unsafe { std::env::set_var("DOCKER_HOST", format!("unix://{sock}")) };
}
}
}
#[tokio::test]
#[ignore = "needs podman + a Spark Connect image with the pipelines jar; run with --ignored"]
async fn knutpipe_pipelines_run_streams_events() {
ensure_docker_host();
// ── setup: Spark Connect container ──────────────────────────────────────
// The stock apache/spark image daemonizes start-connect-server.sh, so keep
// it foreground (SPARK_NO_DAEMONIZE) and bind the connect port. This same
// command is a no-op override for the knut-spark-iceberg image (it just
// drops the inert `lake` catalog conf; the demo uses `spark_catalog`).
let image = std::env::var("KNUT_SPARK_IMAGE")
.unwrap_or_else(|_| "docker.io/apache/spark:4.1.2-scala2.13-java21-python3-ubuntu".into());
let (name, tag) = image.rsplit_once(':').unwrap_or((image.as_str(), "latest"));
let spark = GenericImage::new(name, tag)
.with_exposed_port(ContainerPort::Tcp(15002))
.with_startup_timeout(Duration::from_secs(300))
.with_env_var("SPARK_NO_DAEMONIZE", "1")
.with_cmd([
"/opt/spark/sbin/start-connect-server.sh",
"--conf",
"spark.connect.grpc.binding.port=15002",
])
.start()
.await
.expect("start spark connect (pull apache/spark:4.1.2 or set KNUT_SPARK_IMAGE)");
let port = spark.get_host_port_ipv4(15002).await.unwrap();
let url = format!("sc://127.0.0.1:{port}");
// ── wait until Spark Connect answers a real query ───────────────────────
// (the channel connects before the JVM is ready, so probe with SQL)
let mut client = {
let mut attempt = 0;
loop {
let ok = match PipelinesClient::connect(&url).await {
Ok(mut c) => {
use futures_util::StreamExt;
match c.execute_sql("SELECT 1").await {
Ok(mut s) => {
// Drain the probe stream; a clean end means ready.
let mut ready = true;
while let Some(item) = s.next().await {
if item.is_err() {
ready = false;
break;
}
}
if ready { Some(c) } else { None }
}
Err(_) => None,
}
}
Err(_) => None,
};
match ok {
Some(c) => break c,
None if attempt < 60 => {
attempt += 1;
tokio::time::sleep(Duration::from_secs(3)).await;
}
None => panic!("Spark Connect never became ready after 180s"),
}
}
};
// ── define the demo dataflow graph from examples/pipeline.json ──────────
let spec = concat!(env!("CARGO_MANIFEST_DIR"), "/examples/pipeline.json");
let graph: DataflowGraph =
serde_json::from_str(&std::fs::read_to_string(spec).expect("read pipeline.json"))
.expect("parse pipeline.json");
graph.validate().expect("demo graph validates");
assert_eq!(graph.datasets.len(), 2, "bronze + silver datasets");
assert_eq!(graph.flows.len(), 2, "bronze + silver flows");
let graph_id = client
.define_graph(&graph)
.await
.expect("define the demo dataflow graph (CreateDataflowGraph/DefineOutput/DefineFlow)");
assert!(!graph_id.is_empty(), "server assigned a graph id");
eprintln!("defined dataflow graph {graph_id}");
// ── start a full-refresh run, collect the streamed events ───────────────
let run = PipelineRun {
graph_id: Some(graph_id),
full_refresh_all: true,
storage: Some("file:///tmp/knut_pipeline_itest".into()),
..Default::default()
};
let mut sink = CollectingSink::default();
client
.start_run(&run, &mut sink)
.await
.expect("pipeline run streams to completion");
eprintln!("run streamed {} event(s):", sink.events.len());
for ev in &sink.events {
eprintln!(" [{:?}] {}", ev.level, ev.message);
}
// A real SDP run emits per-flow lifecycle events and a final run-complete.
assert!(
!sink.events.is_empty(),
"a real pipelines run emits at least one event"
);
assert!(
sink.events
.iter()
.any(|e| e.message.contains("knut_bronze")),
"the bronze flow appears in the event stream"
);
let completed = sink
.events
.iter()
.filter(|e| e.level != EventLevel::Error)
.any(|e| e.message.to_ascii_uppercase().contains("COMPLETED"));
assert!(completed, "the run reports a COMPLETED milestone");
// teardown: container drops here.
}