crates/knut-pipelines/src/server.rs
//! `knut_pipelines::server` — bring a local Spark **Connect + SDP pipelines**
//! server up/down **entirely from Rust**, over podman's REST API (feature
//! `server`).
//!
//! Zero-shell charter: this module never shells out. There is no `sh`, no
//! `podman`/`docker` CLI, no [`std::process::Command`] anywhere in the bring-up
//! path. The container lifecycle (pull, create, start, readiness, stop) is routed
//! THROUGH **draupnir** — the shared boot/provisioning library that owns the ONE
//! bollard/podman REST engine for the whole constellation. knut keeps only the
//! Spark-specific policy (image, start args, the `spark.pipelines` ready line, the
//! rootless-socket auto-start); it does NOT roll a second container engine.
//!
//! This is the reusable library half of the `knut-spark-server` binary: the CLI
//! is a thin wrapper over [`SparkServer`]. Downstream crates (e.g. korp's demo
//! orchestrator) bring a Spark Connect server up from Rust with:
//!
//! ```no_run
//! # async fn f() -> knut_pipelines::server::Result<()> {
//! use knut_pipelines::server::SparkServer;
//! let server = SparkServer::connect()?;
//! let url = server.up().await?; // e.g. "sc://localhost:15002"
//! // … drive a pipeline run against `url` …
//! server.stop().await?;
//! # Ok(()) }
//! ```
//!
//! Image + start args + ready line mirror the `integration_spark_pipelines`
//! test: the stock `apache/spark:4.1.x` image (blessed default `4.1.2`, or
//! `4.2.0` GA via `KNUT_SPARK_IMAGE`) already ships the `spark-connect`
//! + `spark-pipelines` jars, so `start-connect-server.sh` exposes the
//! `spark.pipelines` gRPC handler with no build.
//!
//! **Container name.** draupnir prefixes its instances with `draupnir-`, so the
//! managed container is `draupnir-knut-spark-connect`. `stop`/`status` address it
//! by that stable name (a REST call by name), so they work across separate CLI
//! invocations exactly like before — only the external name string changed.
//! **Logs.** draupnir streams a container's logs via an in-process follow task, so
//! `logs` follows what THIS process booted; a standalone `logs` invocation on an
//! already-running server prints what the shared engine has buffered (see the
//! blocker note in `workspace_nordisk/bencher_agents_problems.md`).
use std::time::Duration;
use draupnir::container::{ContainerBoot, ContainerControl, ContainerState, Readiness};
use draupnir::{Boot, BootSpec, Machine};
/// Container name we manage by default. draupnir prefixes `draupnir-`, so the real
/// container is `draupnir-knut-spark-connect`.
pub const DEFAULT_NAME: &str = "knut-spark-connect";
/// The log line the Spark Connect server prints once the gRPC port is bound.
pub const READY_LINE: &str = "Spark Connect server started at";
/// Blessed GA pair — Spark **4.1.2** paired with `iceberg-spark-runtime-4.1_2.13:1.11.0`
/// (the Iceberg jar is baked by `containers/spark-iceberg/Containerfile`; this tag
/// selects the matching stock Spark GA, which already ships the `spark-connect` +
/// `spark-pipelines` jars). This is the default image the pipelines server boots.
pub const DEFAULT_IMAGE_41: &str = "docker.io/apache/spark:4.1.2-scala2.13-java21-python3-ubuntu";
/// Spark **4.2.0 GA** (released 2026-07-14) — kept as a named alternate you select
/// with `KNUT_SPARK_IMAGE`. Ships the same `spark-pipelines` jar, but Iceberg 1.11.0
/// publishes no `iceberg-spark-runtime-4.2` yet, so the Iceberg-paired **4.1.2** stays
/// the blessed default; use this for the SDP/pipelines path where no Iceberg jar is
/// needed.
pub const IMAGE_42: &str = "docker.io/apache/spark:4.2.0-scala2.13-java21-python3-ubuntu";
/// Default image — the blessed 4.1.2 GA pair ([`DEFAULT_IMAGE_41`]). Override with
/// `KNUT_SPARK_IMAGE` (e.g. to [`IMAGE_42`]).
pub const DEFAULT_IMAGE: &str = DEFAULT_IMAGE_41;
/// Default published gRPC port for the Connect server.
pub const DEFAULT_PORT: &str = "15002";
/// Errors from the container lifecycle are boxed and `Send + Sync` so they cross
/// `.await` points and thread boundaries in a downstream async orchestrator.
pub type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
/// What [`SparkServer::status`] reports about the managed container.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ServerStatus {
/// The container is running; carries the `sc://…` client URL.
Running { url: String },
/// The container exists but is not running.
Stopped,
/// No such container (run [`SparkServer::up`] to start one).
Absent,
}
/// A pure-Rust manager for the Spark Connect + SDP pipelines container, routed
/// through draupnir's ONE OCI engine.
///
/// Construct with [`SparkServer::connect`] (honours `KNUT_SPARK_IMAGE`, `PORT`
/// and `DOCKER_HOST` env vars); override with the builder setters. All lifecycle
/// verbs ([`up`](Self::up)/[`stop`](Self::stop)/[`logs`](Self::logs)/
/// [`status`](Self::status)) go through draupnir over the podman/Docker HTTP API —
/// no shell, no second engine.
pub struct SparkServer {
/// The shared draupnir OCI engine (lazily connects to the podman socket on first
/// use). Cloneable so it can move into `spawn_blocking` (draupnir is synchronous
/// with its own runtime, so it is driven off the async executor's worker threads).
backend: ContainerBoot,
name: String,
image: String,
port: String,
}
impl SparkServer {
/// Ensure the podman API socket is up (auto-starting the rootless
/// `podman.socket` unit over D-Bus if needed) and build a manager with defaults
/// (image = `KNUT_SPARK_IMAGE` or [`DEFAULT_IMAGE`]; port = `PORT` or
/// [`DEFAULT_PORT`]; socket = `DOCKER_HOST` or the rootless podman socket).
///
/// If the socket is absent and cannot be started we do **not** shell out: we
/// print the one-time prerequisite hint to stderr and return an error.
pub fn connect() -> Result<Self> {
if let Err(e) = ensure_podman_socket() {
let url = socket_url();
let path = url.strip_prefix("unix://").unwrap_or(&url);
eprintln!("error: podman API socket unavailable at {path}: {e}");
eprintln!("hint: enable it once with: systemctl --user enable --now podman.socket");
eprintln!(" (or point DOCKER_HOST at a running Docker/podman socket)");
return Err("podman API socket unavailable".into());
}
Ok(Self {
backend: ContainerBoot::new(),
name: DEFAULT_NAME.to_string(),
image: env_image(),
port: env_port(),
})
}
/// Override the managed container name (default [`DEFAULT_NAME`]).
pub fn with_name(mut self, name: impl Into<String>) -> Self {
self.name = name.into();
self
}
/// Override the image (default [`DEFAULT_IMAGE`] / `KNUT_SPARK_IMAGE`).
pub fn with_image(mut self, image: impl Into<String>) -> Self {
self.image = image.into();
self
}
/// Override the published gRPC port (default [`DEFAULT_PORT`] / `PORT`).
pub fn with_port(mut self, port: impl Into<String>) -> Self {
self.port = port.into();
self
}
/// The Spark Connect client URL for the published port, e.g.
/// `sc://localhost:15002`.
pub fn url(&self) -> String {
format!("sc://localhost:{}", self.port)
}
/// The published/bound gRPC port as a number (default 15002 if `PORT` is junk).
fn port_num(&self) -> u16 {
self.port.parse().unwrap_or(15002)
}
/// The draupnir spec for the Connect server: the image, the stable name, the
/// `spark.pipelines` start command (`SPARK_NO_DAEMONIZE` in the foreground), and
/// the published gRPC port. The container binds the same port it is published on
/// (draupnir maps host↔container 1:1), so the client `url()` stays correct for
/// any `PORT`.
fn boot_spec(&self) -> BootSpec {
let port = self.port_num();
let mut d = BootSpec::container(self.name.clone(), self.image.clone()).with_cmd([
"/opt/spark/sbin/start-connect-server.sh".to_string(),
"--conf".to_string(),
format!("spark.connect.grpc.binding.port={port}"),
]);
d.env
.insert("SPARK_NO_DAEMONIZE".to_string(), "1".to_string());
d.ports = vec![port];
d
}
/// A [`Machine`] handle addressing the managed container by its stable draupnir
/// name — so `stop`/`status` reach the exact container across separate process
/// invocations (they are REST calls by name, no in-process boot state needed).
fn machine(&self) -> Machine {
Machine::started(format!("draupnir-{}", self.name), &self.boot_spec())
}
/// `up`: through draupnir — remove any stale same-named container, pull the image
/// if absent, create + start the server detached (publishing the gRPC port), and
/// wait until it logs the ready line. Returns the `sc://…` client URL; the server
/// stays running (call [`stop`](Self::stop) to tear it down).
pub async fn up(&self) -> Result<String> {
let backend = self.backend.clone();
let spec = self.boot_spec();
let url = self.url();
eprintln!(
"# starting {} ({}) on port {} ...",
self.name, self.image, self.port
);
eprintln!("# pulling if absent + waiting for the JVM to bind the gRPC port ...");
// draupnir is synchronous (it owns its own tokio runtime), so drive it on a
// blocking thread — never nested inside this async executor's worker.
let joined = tokio::task::spawn_blocking(move || -> std::result::Result<(), String> {
let machine = backend
.boot(&spec)
.map_err(|e| format!("container boot: {e}"))?;
backend
.wait_ready(
&machine,
&Readiness::LogMatch(READY_LINE.to_string()),
Duration::from_secs(240),
)
.map_err(|e| format!("wait for ready: {e}"))?;
Ok(())
})
.await
.map_err(|e| format!("spawn_blocking join: {e}"))?;
joined.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })?;
Ok(url)
}
/// `stop`: stop + remove the container through draupnir (idempotent, best-effort).
pub async fn stop(&self) -> Result<()> {
let backend = self.backend.clone();
let machine = self.machine();
tokio::task::spawn_blocking(move || backend.stop(&machine))
.await
.map_err(|e| format!("spawn_blocking join: {e}"))?;
Ok(())
}
/// `logs`: stream the container logs (follow) to stdout, via draupnir's log
/// drain. The shared engine follows logs in-process, so this surfaces what the
/// engine has buffered for the managed container; it loops until the container
/// is no longer running.
pub async fn logs(&self) -> Result<()> {
let backend = self.backend.clone();
let machine = self.machine();
tokio::task::spawn_blocking(move || {
loop {
for line in backend.drain_logs(&machine) {
println!("{line}");
}
if !matches!(backend.container_state(&machine), ContainerState::Running) {
break;
}
std::thread::sleep(Duration::from_millis(500));
}
})
.await
.map_err(|e| format!("spawn_blocking join: {e}"))?;
Ok(())
}
/// `status`: report whether the managed container is running / stopped /
/// absent (and, if running, its client URL) — via draupnir by name.
pub async fn status(&self) -> Result<ServerStatus> {
let backend = self.backend.clone();
let machine = self.machine();
let url = self.url();
let state = tokio::task::spawn_blocking(move || backend.container_state(&machine))
.await
.map_err(|e| format!("spawn_blocking join: {e}"))?;
Ok(match state {
ContainerState::Running => ServerStatus::Running { url },
ContainerState::Exited(_) => ServerStatus::Stopped,
ContainerState::Gone => ServerStatus::Absent,
})
}
}
fn env_image() -> String {
std::env::var("KNUT_SPARK_IMAGE").unwrap_or_else(|_| DEFAULT_IMAGE.into())
}
fn env_port() -> String {
std::env::var("PORT").unwrap_or_else(|_| DEFAULT_PORT.into())
}
/// Resolve the podman API socket URL: honour `DOCKER_HOST`, else the rootless
/// user socket under `XDG_RUNTIME_DIR` (falling back to `/run/user/1000`).
fn socket_url() -> String {
if let Ok(h) = std::env::var("DOCKER_HOST") {
return h;
}
let xdg = std::env::var("XDG_RUNTIME_DIR").unwrap_or_else(|_| "/run/user/1000".into());
format!("unix://{xdg}/podman/podman.sock")
}
/// Ensure the rootless podman API socket exists, **starting the `podman.socket`
/// systemd user unit over D-Bus** if it's missing — pure Rust, zero-shell (no
/// `systemctl`, no podman CLI; LAW 7). A no-op when the socket is already there.
///
/// This is what lets the demo boot podman *itself*: nothing has to be enabled by
/// hand beforehand. It needs a user session bus (present in an interactive
/// login); with none, it returns an error and the caller falls back to the
/// manual `systemctl --user enable --now podman.socket` hint. draupnir's own engine
/// only *checks* the socket (it never shells out to start it), so knut owns this
/// auto-start step and runs it before handing the lifecycle to draupnir.
pub fn ensure_podman_socket() -> Result<()> {
let url = socket_url();
let path = url.strip_prefix("unix://").unwrap_or(&url).to_string();
// Probe by CONNECTING, not by `exists()`: a stopped `podman.socket` leaves
// a stale socket *file* behind that refuses connections, so a bare
// existence check would wrongly conclude it's up (→ later "Connect" error).
if socket_connectable(&path) {
return Ok(());
}
start_podman_socket_unit()?;
// Wait until the socket actually ACCEPTS a connection (StartUnit is async;
// socket-activation binds the listener when the job completes).
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
while std::time::Instant::now() < deadline {
if socket_connectable(&path) {
return Ok(());
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
Err(format!("started podman.socket but {path} is not accepting connections after 5s").into())
}
/// True iff something is actually listening on the unix socket at `path`. A
/// stale socket file can linger after `stop` with no listener, so this connects
/// rather than just checking the path exists.
fn socket_connectable(path: &str) -> bool {
std::os::unix::net::UnixStream::connect(path).is_ok()
}
/// Ask the systemd **user** manager to start `podman.socket` over the session
/// bus — the D-Bus equivalent of `systemctl --user start podman.socket`, with no
/// process exec.
fn start_podman_socket_unit() -> Result<()> {
let conn = zbus::blocking::Connection::session()
.map_err(|e| format!("no user session D-Bus to auto-start podman.socket: {e}"))?;
let proxy = zbus::blocking::Proxy::new(
&conn,
"org.freedesktop.systemd1",
"/org/freedesktop/systemd1",
"org.freedesktop.systemd1.Manager",
)
.map_err(|e| format!("systemd1 D-Bus proxy: {e}"))?;
// StartUnit(name, mode) -> job object path; "replace" is the standard mode.
let _job: zbus::zvariant::OwnedObjectPath = proxy
.call("StartUnit", &("podman.socket", "replace"))
.map_err(|e| format!("StartUnit podman.socket: {e}"))?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
/// The draupnir spec knut hands the shared engine carries the Spark image, the
/// stable name, `SPARK_NO_DAEMONIZE`, the `spark.pipelines` start command bound
/// to the published port, and that port published — so a boot through draupnir
/// brings up the exact Connect server knut's own bollard path used to. Pure
/// mapping, no daemon.
#[test]
fn boot_spec_maps_the_connect_server_onto_draupnir() {
let server = SparkServer {
backend: ContainerBoot::new(),
name: DEFAULT_NAME.to_string(),
image: DEFAULT_IMAGE.to_string(),
port: DEFAULT_PORT.to_string(),
};
let d = server.boot_spec();
assert_eq!(d.backend, draupnir::Backend::Container);
assert_eq!(
d.image,
draupnir::ImageSource::OciImage(DEFAULT_IMAGE.into())
);
assert_eq!(d.name, DEFAULT_NAME);
assert_eq!(
d.env.get("SPARK_NO_DAEMONIZE").map(String::as_str),
Some("1")
);
assert_eq!(d.ports, vec![15002]);
assert!(
d.cmd
.iter()
.any(|a| a == "spark.connect.grpc.binding.port=15002"),
"cmd binds the published port: {:?}",
d.cmd
);
assert!(d.cmd.iter().any(|a| a.ends_with("start-connect-server.sh")));
d.validate().expect("mapped spec is valid");
}
/// A custom `PORT` binds AND publishes that port (draupnir maps host↔container
/// 1:1), and the client URL follows it — so `sc://localhost:<PORT>` stays correct.
#[test]
fn custom_port_is_bound_published_and_reflected_in_the_url() {
let server = SparkServer {
backend: ContainerBoot::new(),
name: DEFAULT_NAME.to_string(),
image: DEFAULT_IMAGE.to_string(),
port: "15055".to_string(),
};
assert_eq!(server.url(), "sc://localhost:15055");
let d = server.boot_spec();
assert_eq!(d.ports, vec![15055]);
assert!(
d.cmd
.iter()
.any(|a| a == "spark.connect.grpc.binding.port=15055")
);
}
/// The blessed default is the **4.1.2 GA** pair (Spark 4.1.2 +
/// iceberg-spark-runtime-4.1_2.13:1.11.0), NOT 4.2 — 4.2.0 GA stays available
/// as a named const you select with `KNUT_SPARK_IMAGE` (Iceberg has no 4.2
/// runtime yet). This goes RED if a release/edit silently re-defaults the
/// server off the Iceberg-paired 4.1.2.
#[test]
fn default_image_is_the_blessed_41_ga_not_42() {
assert_eq!(
DEFAULT_IMAGE, DEFAULT_IMAGE_41,
"default must be the 4.1 GA pair"
);
assert!(
DEFAULT_IMAGE_41.contains("4.1.2") && !DEFAULT_IMAGE_41.contains("preview"),
"blessed default is Spark 4.1.2 GA: {DEFAULT_IMAGE_41}"
);
assert!(
IMAGE_42.contains("4.2.0") && !IMAGE_42.contains("preview"),
"4.2.0 GA kept as the named alternate: {IMAGE_42}"
);
// With no override, resolution falls through to the blessed 4.1 GA.
if std::env::var_os("KNUT_SPARK_IMAGE").is_none() {
assert_eq!(env_image(), DEFAULT_IMAGE_41);
}
}
/// `stop`/`status` address the container by its stable draupnir name, so they
/// reach the exact instance across separate CLI invocations.
#[test]
fn managed_machine_uses_the_stable_draupnir_prefixed_name() {
let server = SparkServer {
backend: ContainerBoot::new(),
name: DEFAULT_NAME.to_string(),
image: DEFAULT_IMAGE.to_string(),
port: DEFAULT_PORT.to_string(),
};
assert_eq!(server.machine().id, "draupnir-knut-spark-connect");
}
}