bench/src/containers.rs
//! Bring up / tear down the bench's competitor backends — **RustFS** (S3),
//! **Nessie** (RocksDB + Iceberg REST), **Polaris** (Quarkus + Iceberg REST) —
//! in **pure Rust, zero-shell**. Every container operation goes over the
//! podman/Docker **API socket** via **`draupnir::container::ContainerBoot`** — the
//! ONE bollard REST engine of the constellation, the same engine **`jera`** drives
//! (jera has no engine of its own; it delegates to draupnir). Container specs are
//! built with **`jera::container::ContainerSpec`** (the sanctioned spec/builder) and
//! mapped onto draupnir's `BootSpec` via [`ContainerSpec::to_draupnir`]. **No
//! `Command::new("podman"|"docker")` anywhere** — the old `containers/*.sh` port
//! shelled out; this does not.
//!
//! Every operational gotcha we hit bringing these up is still encoded here as a
//! **pre-check** (refuse early with a clear message) or a **pre-check-fix** (repair
//! automatically), so a fresh machine gets a working stack from
//! `bench-containers up all`:
//!
//! - **engine detection** — gone: we connect to the podman/Docker **API socket**
//! (draupnir honours `DOCKER_HOST`, else the rootless
//! `$XDG_RUNTIME_DIR/podman/podman.sock`). A boot fails with a clear error if no
//! socket is reachable — never a CLI probe, never a `--version` subprocess.
//! - **stale container** — draupnir force-removes any same-named container before it
//! (re)creates one (its `boot` does a `remove_container(force)` first), so a
//! restart is clean without a separate `rm -f`.
//! - **RustFS named volume** — a named volume (`nornir_bench_rustfs_data`) sidesteps
//! the UID-10001 ownership a bind mount would impose. The engine auto-creates a
//! named volume referenced in a mount on first use, so no explicit `volume create`
//! is needed (the old shell `volume create` is dropped).
//! - **Nessie → RustFS gateway** — Nessie reaches the host-published S3 endpoint
//! via `host.containers.internal`; the bench client uses `localhost`.
//! - **Nessie needs S3** — if RustFS isn't up, start it first (dependency).
//! - **Nessie takes no OAuth** — its client must NOT be handed `credential=…`
//! (it answers `501` on `/v1/oauth/tokens`); that's a bench-run env concern,
//! flagged in [`up_all`]'s closing notes.
//! - **Polaris UID mapping** — Polaris runs as container uid 10000; without a
//! userns map it writes the bind-mounted warehouse as a subuid the bench client
//! can't write alongside (first a `503`, then `EACCES`). **FLAGGED GAP:** neither
//! `jera::container::ContainerSpec` nor draupnir's create body carries a `userns`
//! knob today, so the `--userns=keep-id:uid=10000,gid=10001` fix cannot be
//! expressed zero-shell yet. See [`polaris_up`].
//! - **Polaris stale warehouse** — removed with `std::fs::remove_dir_all` (a dir we
//! own). A prior *non-userns* run can leave subuid-owned files the host user can't
//! unlink; the old code shelled `<engine> unshare rm` for that — a pure-Rust
//! violation we refuse. See [`clean_polaris_warehouse`].
//! - **Polaris feature flags** — FILE storage needs three quoted-key feature flags
//! passed as `-D` (JVM) props, carried here as the `JAVA_OPTS_APPEND` env var.
//! - **Polaris bootstrap** — fetch an OAuth token and create the FILE catalog (plain
//! HTTP via `ureq`, unchanged).
use std::path::Path;
use std::time::{Duration, Instant};
use anyhow::{anyhow, bail, Context, Result};
use draupnir::container::{ContainerBoot, ContainerControl, ContainerState};
use draupnir::Machine;
use jera::container::{BindMount, ContainerSpec};
const RUSTFS: &str = "nornir-bench-rustfs";
const NESSIE: &str = "nornir-bench-nessie";
/// Nessie with NO `nessie.catalog.*` config — Nessie API v2 only, the way a
/// Spark-fed production Nessie runs. Its own name and host port so it can sit
/// beside the REST-configured one.
const NESSIE_PLAIN: &str = "nornir-bench-nessie-plain";
const NESSIE_PLAIN_PORT: u16 = 19121;
const POLARIS: &str = "nornir-bench-polaris";
const POLARIS_WH: &str = "/tmp/nornir_bench_wh";
const SPARK: &str = "nornir-bench-spark";
// Images the specs run. Kept as constants so a name-addressed teardown/status
// (a *separate* `bench-containers` invocation) rebuilds the same spec identity.
const RUSTFS_IMAGE: &str = "docker.io/rustfs/rustfs:latest";
/// Pinned to the production Nessie (0.105.3, java 21, ROCKSDB version store).
const NESSIE_IMAGE: &str = "ghcr.io/projectnessie/nessie:0.105.3";
const POLARIS_IMAGE: &str = "docker.io/apache/polaris:latest";
/// Built from `knut/containers/spark-iceberg/Containerfile` (Spark 4.2 + Iceberg
/// 1.11, Hadoop `lake` catalog). Build it once with the engine of your choice.
const SPARK_IMAGE: &str = "knut-spark-iceberg";
/// Host path that MUST match the container's catalog warehouse — the Hadoop
/// catalog writes absolute `/tmp/warehouse/...` paths into the metadata, so skade
/// reads the copied-out tree at the same absolute path.
const SPARK_WH: &str = "/tmp/warehouse";
// ---- engine handle + name-addressed lifecycle over the ONE bollard engine ---
/// A handle to the ONE zero-shell OCI engine (draupnir's `ContainerBoot` — the
/// engine `jera` itself drives). Connects to the podman/Docker socket lazily on
/// first use; every op fails with a clear error if no socket is reachable.
fn engine() -> ContainerBoot {
ContainerBoot::new()
}
/// Rebuild a draupnir [`Machine`] handle for the container named `name` so a
/// *separate* CLI invocation (`status` / `down`) can address a container **this
/// process did not boot**. draupnir prefixes every container it boots with
/// `draupnir-` (see `draupnir::container::ContainerBoot`), so the engine-side name
/// is `draupnir-<name>`; only that id (and the container backend tag) is read by
/// `container_state`/`stop`, so the image is immaterial here.
fn machine(name: &str) -> Machine {
let spec = ContainerSpec::image("").with_name(name).to_draupnir();
Machine::started(format!("draupnir-{name}"), &spec)
}
/// Boot `spec` **detached** on the ONE OCI engine and leave it running (create +
/// start over the socket; draupnir force-removes any stale same-named container
/// first). The spec's bind mounts ([`ContainerSpec::bind_strings`]) are threaded
/// through so named volumes / host binds land as `-v` equivalents.
fn boot_detached(eng: &ContainerBoot, spec: &ContainerSpec) -> Result<()> {
let bootspec = spec.to_draupnir();
let binds = spec.bind_strings();
eng.boot_with_binds(&bootspec, &binds)
.map_err(|e| anyhow!("boot container `{}` ({}): {e}", spec.name, spec.image))?;
Ok(())
}
/// Is the named container currently running? (engine `container_state`, no `ps`).
fn is_running(eng: &ContainerBoot, name: &str) -> bool {
matches!(eng.container_state(&machine(name)), ContainerState::Running)
}
/// Stop + remove the named container (idempotent, best-effort) — the zero-shell
/// replacement for `<engine> rm -f <name>`.
fn remove(eng: &ContainerBoot, name: &str) {
eng.stop(&machine(name));
}
// ---- small HTTP helpers (health polling + Polaris bootstrap) ----------------
/// ureq 3 (2026-10-02) with ureq 2's defaults: no proxy from the environment and
/// at most 5 redirects.
fn http() -> ureq::Agent {
ureq::Agent::config_builder()
.timeout_global(Some(Duration::from_secs(3)))
.proxy(None)
.max_redirects(5)
.build()
.new_agent()
}
/// A backend is "up" once its endpoint answers at all (any non-5xx — an
/// unauthenticated 4xx still proves the port is serving). ureq surfaces a 4xx/5xx
/// as `Err(StatusCode)`, so a 4xx (e.g. RustFS's `403` on `/`) counts as up.
fn http_up(url: &str) -> bool {
match http().get(url).call() {
Ok(_) => true,
Err(ureq::Error::StatusCode(code)) => code < 500,
Err(_) => false, // transport error (connection refused / DNS) ⇒ down
}
}
fn poll(label: &str, secs: u64, ready: impl Fn() -> bool) -> Result<()> {
for i in 0..secs {
if ready() {
eprintln!("✅ {label} ready after {i}s");
return Ok(());
}
std::thread::sleep(Duration::from_secs(1));
}
bail!("timeout waiting for {label} after {secs}s")
}
// ---- RustFS (shared S3 warehouse) ------------------------------------------
/// Host port the RustFS S3 endpoint is published on — configurable via
/// `BENCH_RUSTFS_PORT` (default `9000`) so it can dodge a host-port clash (e.g. a
/// Njord MinIO `kubectl port-forward` already holding :9000). The container's
/// internal port stays 9000; only the host publish + the S3-client / Nessie-gateway
/// endpoints follow this. When you override it, also set
/// `BENCH_S3_ENDPOINT=http://localhost:<port>` for the bench client.
pub fn rustfs_port() -> String {
std::env::var("BENCH_RUSTFS_PORT").unwrap_or_else(|_| "9000".into())
}
pub fn rustfs_up() -> Result<()> {
let eng = engine();
let port: u16 = rustfs_port().parse().context("BENCH_RUSTFS_PORT")?;
let console: u16 = std::env::var("BENCH_RUSTFS_CONSOLE_PORT")
.unwrap_or_else(|_| "9001".into())
.parse()
.context("BENCH_RUSTFS_CONSOLE_PORT")?;
eprintln!("🚀 RustFS S3 on :{port} (console :{console})");
// pre-check-fix: a named volume avoids the UID-10001 ownership a bind mount
// needs. The engine auto-creates it on first mount (no `volume create`).
let mut spec = ContainerSpec::image(RUSTFS_IMAGE)
.with_name(RUSTFS)
// host `port` -> container 9000 (distinct when BENCH_RUSTFS_PORT != 9000).
.with_port_map(port, 9000)
.with_port_map(console, 9001);
spec.cmd = vec!["/data".into()];
spec.env.insert("RUSTFS_ACCESS_KEY".into(), "rustfsadmin".into());
spec.env.insert("RUSTFS_SECRET_KEY".into(), "rustfsadmin".into());
spec.env.insert("RUSTFS_CONSOLE_ENABLE".into(), "true".into());
spec.mounts = vec![BindMount::rw("nornir_bench_rustfs_data", "/data")];
boot_detached(&eng, &spec)?;
let health = format!("http://localhost:{port}/");
poll("RustFS", 60, || http_up(&health))?;
eprintln!(" S3: http://localhost:{port} (rustfsadmin/rustfsadmin); the bench client creates the `warehouse` bucket on first use");
Ok(())
}
// ---- Nessie (RocksDB + Iceberg REST over the shared S3) ---------------------
pub fn nessie_up() -> Result<()> {
let eng = engine();
// pre-check-fix: Nessie's warehouse is the shared S3 — bring RustFS up first.
if !is_running(&eng, RUSTFS) {
eprintln!("⚙ RustFS not running — starting it first (Nessie's warehouse is the shared S3)");
rustfs_up()?;
}
eprintln!("🚀 Nessie (RocksDB) on :19120, warehouse s3://warehouse/wh");
// Nessie (in the container) reaches the host-published S3 via the gateway —
// follow BENCH_RUSTFS_PORT so it tracks a relocated RustFS (see rustfs_port).
let nessie_s3_endpoint = format!("http://host.containers.internal:{}", rustfs_port());
let mut spec = ContainerSpec::image(NESSIE_IMAGE).with_name(NESSIE);
spec.ports = vec![19120];
for (k, v) in [
("nessie.version.store.type", "ROCKSDB"),
("nessie.version.store.persist.rocks.database-path", "/tmp/rocksdb"),
("nessie.catalog.default-warehouse", "warehouse"),
("nessie.catalog.warehouses.warehouse.location", "s3://warehouse/wh"),
("nessie.catalog.service.s3.default-options.region", "us-east-1"),
("nessie.catalog.service.s3.default-options.endpoint", &nessie_s3_endpoint),
("nessie.catalog.service.s3.default-options.path-style-access", "true"),
("nessie.catalog.service.s3.default-options.auth-type", "STATIC"),
(
"nessie.catalog.service.s3.default-options.access-key",
"urn:nessie-secret:quarkus:nessie.catalog.secrets.access-key",
),
("nessie.catalog.secrets.access-key.name", "rustfsadmin"),
("nessie.catalog.secrets.access-key.secret", "rustfsadmin"),
] {
spec.env.insert(k.to_string(), v.to_string());
}
boot_detached(&eng, &spec)?;
poll("Nessie", 120, || http_up("http://localhost:19120/iceberg/v1/config"))?;
eprintln!(" REST: http://localhost:19120/iceberg (no OAuth — do NOT set BENCH_REST_PROPS for the Nessie client)");
Ok(())
}
/// Nessie with the version store only — no `nessie.catalog.*` warehouse/S3
/// config, so its Iceberg REST side is unconfigured and API v2 is the door.
/// Mirrors a production Nessie that Spark writes through. Host :19121.
pub fn nessie_plain_up() -> Result<()> {
let eng = engine();
eprintln!("🚀 Nessie (RocksDB, API v2 only, no catalog config) on :{NESSIE_PLAIN_PORT}");
let mut spec = ContainerSpec::image(NESSIE_IMAGE)
.with_name(NESSIE_PLAIN)
.with_port_map(NESSIE_PLAIN_PORT, 19120);
for (k, v) in [
("nessie.version.store.type", "ROCKSDB"),
("nessie.version.store.persist.rocks.database-path", "/tmp/rocksdb"),
// Production runs with authentication disabled; say so, don't inherit it.
("nessie.server.authentication.enabled", "false"),
] {
spec.env.insert(k.to_string(), v.to_string());
}
boot_detached(&eng, &spec)?;
let cfg = format!("http://localhost:{NESSIE_PLAIN_PORT}/api/v2/config");
poll("Nessie (plain)", 120, || http_up(&cfg))?;
eprintln!(" API v2: http://localhost:{NESSIE_PLAIN_PORT}/api/v2");
Ok(())
}
// ---- Polaris (Quarkus + Iceberg REST, FILE warehouse) ----------------------
/// pre-check-fix: remove a stale warehouse (a directory we own) with
/// `std::fs::remove_dir_all` — **zero-shell**.
///
/// **FLAGGED GAP (was `podman unshare rm -rf`):** after a *non-userns* Polaris run,
/// the warehouse files are owned by a subuid the host user cannot unlink. The old
/// code shelled `<engine> unshare rm -rf` to delete them inside the engine's user
/// namespace — a pure-Rust violation. There is **no pure-Rust way** to remove a
/// file owned by a subuid we do not map (it needs the kernel user-namespace the
/// engine sets up), and we refuse to shell out. So we delete what we own and, if a
/// subuid-owned remnant blocks us, fail with a clear message rather than a broken
/// shell fallback. NOTE: once Polaris boots with its `--userns=keep-id` map again
/// (see the FLAG in [`polaris_up`]), the warehouse is host-owned and this succeeds.
fn clean_polaris_warehouse() -> Result<()> {
if !Path::new(POLARIS_WH).exists() {
return Ok(());
}
match std::fs::remove_dir_all(POLARIS_WH) {
Ok(()) => Ok(()),
Err(e) => bail!(
"cannot remove stale Polaris warehouse {POLARIS_WH}: {e}\n \
(likely subuid-owned from a prior non-userns run; removing it needs the \
engine user-namespace. We refuse to shell out to `podman unshare rm`. \
Delete it yourself, e.g. rootless `podman unshare rm -rf {POLARIS_WH}`, \
then retry.)"
),
}
}
pub fn polaris_up() -> Result<()> {
let eng = engine();
clean_polaris_warehouse()?;
std::fs::create_dir_all(POLARIS_WH).with_context(|| format!("create {POLARIS_WH}"))?;
eprintln!("🚀 Polaris on :8181 (mgmt :8182), FILE warehouse {POLARIS_WH}");
// FLAGGED GAP — the userns UID fix cannot be expressed zero-shell yet.
// The shell port ran Polaris with `--userns=keep-id:uid=10000,gid=10001` so the
// container's uid-10000 `polaris` user writes the bind-mounted warehouse as the
// HOST user (else a 503 on metadata, then EACCES on the data dir, and the client
// can't read the FILE catalog). Neither `jera::container::ContainerSpec` nor
// draupnir's create body carries a `userns` knob today (bollard's
// `HostConfig.userns_mode` is not wired through). The clean fix is a one-field
// addition — `ContainerSpec.userns: Option<String>` → `draupnir::BootSpec` →
// `HostConfig.userns_mode` — kept zero-shell. Until that lands, Polaris boots
// WITHOUT the map: FILE-catalog writes may be subuid-owned and the bench client
// may hit EACCES. We refuse to shell `podman run --userns=…`, so we flag it.
eprintln!(
"⚠️ FLAG: Polaris booting WITHOUT `--userns=keep-id:uid=10000,gid=10001` — \
jera/draupnir expose no `userns` knob yet, and we will not shell out. The \
FILE warehouse under {POLARIS_WH} may be written as a subuid the bench \
client cannot write alongside (503/EACCES). Wire `ContainerSpec.userns` → \
draupnir `HostConfig.userns_mode` to restore the fix zero-shell."
);
let mut spec = ContainerSpec::image(POLARIS_IMAGE).with_name(POLARIS);
spec.ports = vec![8181, 8182];
// The FILE warehouse bind. NOTE: the shell port used a `:z` SELinux relabel
// suffix (`{POLARIS_WH}:{POLARIS_WH}:z`); `jera::BindMount` renders only
// `host:container[:ro]`, so `:z` is dropped — on an SELinux-enforcing host add
// the relabel out-of-band or extend `BindMount` with a mount-options field.
spec.mounts = vec![BindMount::rw(POLARIS_WH, POLARIS_WH)];
for (k, v) in [
("POLARIS_BOOTSTRAP_CREDENTIALS", "POLARIS,root,s3cr3t"),
("polaris.persistence.type", "in-memory"),
("polaris.readiness.ignore-severe-issues", "true"),
// For the (attempted) S3 catalog: feed RustFS creds via the AWS provider
// chain — Polaris vends *subscoped* creds and ignores the catalog's static
// creds for that step, so without these it fails at credential loading.
("AWS_ACCESS_KEY_ID", "rustfsadmin"),
("AWS_SECRET_ACCESS_KEY", "rustfsadmin"),
("AWS_REGION", "us-east-1"),
// Storage gated by feature flags whose keys carry a quoted segment; passed
// as JVM -D props so the quotes survive. FILE works end-to-end; S3 is
// enabled + SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION clears the STS wall, but
// the vended S3FileIO still 301s against RustFS — so the S3 catalog is
// best-effort, FILE is the one the bench relies on.
(
"JAVA_OPTS_APPEND",
r#"-Dpolaris.features."SUPPORTED_CATALOG_STORAGE_TYPES"=["FILE","S3"] -Dpolaris.features."ALLOW_INSECURE_STORAGE_TYPES"=true -Dpolaris.features."ALLOW_SPECIFYING_FILE_IO_IMPL"=true -Dpolaris.features."SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION"=true"#,
),
] {
spec.env.insert(k.to_string(), v.to_string());
}
boot_detached(&eng, &spec)?;
poll("Polaris", 120, || http_up("http://localhost:8182/q/health"))?;
polaris_bootstrap()?;
eprintln!(" REST: http://localhost:8181/api/catalog (client creds root/s3cr3t via BENCH_POLARIS_CRED)");
Ok(())
}
/// OAuth token → create the FILE-backed `warehouse` catalog (idempotent).
fn polaris_bootstrap() -> Result<()> {
eprintln!("🔑 Polaris OAuth token + catalog bootstrap…");
let agent = http();
let resp = agent
.post("http://localhost:8181/api/catalog/v1/oauth/tokens")
.send_form([
("grant_type", "client_credentials"),
("client_id", "root"),
("client_secret", "s3cr3t"),
("scope", "PRINCIPAL_ROLE:ALL"),
])
.context("polaris oauth token request")?;
let token = resp
.into_body()
.read_json::<serde_json::Value>()?
.get("access_token")
.and_then(|v| v.as_str())
.map(str::to_string)
.ok_or_else(|| anyhow!("polaris token response had no access_token"))?;
// The FILE catalog the bench relies on (works end-to-end).
let file_cat = serde_json::json!({
"catalog": {
"name": "warehouse",
"type": "INTERNAL",
"properties": { "default-base-location": format!("file://{POLARIS_WH}") },
"storageConfigInfo": {
"storageType": "FILE",
"allowedLocations": [format!("file://{POLARIS_WH}")]
}
}
});
create_catalog(&agent, &token, "warehouse", file_cat)?;
// Best-effort S3 catalog over RustFS. Cleared the STS wall (skip-subscoping +
// AWS env), but the vended S3FileIO 301s on the data write (path-style not
// applied) — so this may exist yet not serve writes. Kept so the slot is ready
// if a future Polaris/RustFS combo honors path-style on vended creds.
let s3_cat = serde_json::json!({
"catalog": {
"name": "wh_s3",
"type": "INTERNAL",
"properties": {
"default-base-location": "s3://warehouse/pol",
"s3.endpoint": "http://host.containers.internal:9000",
"s3.path-style-access": "true",
"s3.access-key-id": "rustfsadmin",
"s3.secret-access-key": "rustfsadmin",
"s3.region": "us-east-1"
},
"storageConfigInfo": {
"storageType": "S3",
"allowedLocations": ["s3://warehouse/pol"],
"roleArn": "arn:aws:iam::000000000000:role/polaris",
"endpoint": "http://host.containers.internal:9000",
"pathStyleAccess": true,
"region": "us-east-1"
}
}
});
create_catalog(&agent, &token, "wh_s3 (best-effort S3)", s3_cat)?;
Ok(())
}
fn create_catalog(agent: &ureq::Agent, token: &str, label: &str, body: serde_json::Value) -> Result<()> {
match agent
.post("http://localhost:8181/api/management/v1/catalogs")
.header("Authorization", &format!("Bearer {token}"))
.send_json(body)
{
Ok(_) => eprintln!("✅ catalog '{label}' created"),
Err(ureq::Error::StatusCode(409)) => eprintln!("✅ catalog '{label}' already exists"),
Err(ureq::Error::StatusCode(c)) => eprintln!("⚠️ catalog '{label}' create returned HTTP {c}"),
Err(e) => return Err(anyhow!("polaris catalog '{label}' create: {e}")),
}
Ok(())
}
// ---- Spark + Iceberg (the external "skade vs Spark" baseline) --------------
/// Bring up Spark+Iceberg and have **Spark** write a `rows`-row Iceberg table
/// (`lake.db.t`, Hadoop catalog). Prints Spark's write time.
///
/// **FLAGGED GAP — the host-side warehouse copy cannot be done zero-shell yet.**
/// The shell port finished with `<engine> cp <spark>:/tmp/warehouse {SPARK_WH}` so
/// skade could read the SAME table from the host. draupnir's only copy-out
/// (`ContainerBoot::extract_path`) creates a **throwaway** container from the IMAGE
/// and downloads from that — it cannot see data written at RUNTIME into this
/// running container's writable layer (exactly Spark's freshly-written table). A
/// zero-shell copy needs a `download_from_container(<running name>, path)` op on
/// draupnir's engine (bollard already exposes it; draupnir wires it only to the
/// throwaway path). Until draupnir grows that, we run the write over the socket
/// (`exec`, zero-shell) but refuse to shell `podman cp` for the copy — so this
/// returns an honest error at the copy step rather than a broken fake.
pub fn spark_up(rows: u64) -> Result<()> {
let eng = engine();
eprintln!("🚀 Spark 4.2 + Iceberg 1.11 (Hadoop `lake` catalog → {SPARK_WH})");
// Detached: the Connect-server CMD (the image default) keeps it alive so we can
// `exec spark-sql`.
let spec = ContainerSpec::image(SPARK_IMAGE).with_name(SPARK);
boot_detached(&eng, &spec)?;
std::thread::sleep(Duration::from_secs(6)); // JVM/connect warmup
let sql = format!(
"CREATE NAMESPACE IF NOT EXISTS lake.db; \
CREATE TABLE lake.db.t USING iceberg AS \
SELECT id, CAST(id AS STRING) AS name FROM range({rows});"
);
let argv = [
"/opt/spark/bin/spark-sql",
"--conf",
"spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
"--conf",
"spark.sql.catalog.lake=org.apache.iceberg.spark.SparkCatalog",
"--conf",
"spark.sql.catalog.lake.type=hadoop",
"--conf",
"spark.sql.catalog.lake.warehouse=/tmp/warehouse",
"-e",
&sql,
];
let t0 = Instant::now();
// Zero-shell `exec` over the podman/Docker `/exec` REST op (draupnir's engine).
let out = eng
.exec(&machine(SPARK), &argv)
.map_err(|e| anyhow!("spark-sql exec: {e}"))?;
if out.exit_code != Some(0) {
bail!(
"spark-sql exited {:?}:\n{}\n{}",
out.exit_code,
out.stdout.join("\n"),
out.stderr.join("\n")
);
}
let secs = t0.elapsed().as_secs_f64();
eprintln!(" Spark wrote {rows} rows in {secs:.2}s ({:.1} Mrows/s)", rows as f64 / secs / 1e6);
// FLAGGED GAP: copy the warehouse OUT of the RUNNING container to the host.
let _ = std::fs::remove_dir_all(SPARK_WH);
bail!(
"Spark wrote lake.db.t inside the container, but copying the warehouse to \
{SPARK_WH} needs a zero-shell 'copy from a RUNNING container' op that \
draupnir does not expose yet (its extract_path is IMAGE-based, so it can't \
see runtime-written data). Refusing to shell `podman cp`. Add a \
`download_from_container(<running name>, path)` op to draupnir's engine \
(bollard already supports it) to finish the skade-vs-Spark baseline."
);
}
pub fn spark_down() -> Result<()> {
let eng = engine();
remove(&eng, SPARK);
eprintln!("⏹ down {SPARK}");
Ok(())
}
// ---- orchestration ---------------------------------------------------------
pub fn up_all() -> Result<()> {
rustfs_up()?;
nessie_up()?;
polaris_up()?;
eprintln!("\n✅ stack up. Run the cross-catalog bench (one bencher at a time):");
eprintln!(" TPCH_COMPARE_SF=0.5 nornir bench run nornir-catalog");
eprintln!(" (do NOT set BENCH_REST_PROPS — Nessie has no OAuth; Polaris uses BENCH_POLARIS_CRED, default root:s3cr3t)");
Ok(())
}
pub fn down(service: &str) -> Result<()> {
let eng = engine();
let name = match service {
"rustfs" => RUSTFS,
"nessie" => NESSIE,
"nessie-plain" => NESSIE_PLAIN,
"polaris" => POLARIS,
other => bail!("unknown service `{other}` (rustfs|nessie|nessie-plain|polaris|all)"),
};
remove(&eng, name);
eprintln!("⏹ down {name}");
Ok(())
}
pub fn down_all() -> Result<()> {
let eng = engine();
for n in [RUSTFS, NESSIE, POLARIS] {
remove(&eng, n);
}
eprintln!("⏹ down all bench containers");
Ok(())
}
pub fn status() -> Result<()> {
let eng = engine();
for (svc, name, probe) in [
("rustfs", RUSTFS, "http://localhost:9000/"),
("nessie", NESSIE, "http://localhost:19120/iceberg/v1/config"),
("polaris", POLARIS, "http://localhost:8182/q/health"),
] {
let state = if !is_running(&eng, name) {
"stopped"
} else if http_up(probe) {
"ready"
} else {
"starting"
};
eprintln!(" {svc:<8} {state}");
}
Ok(())
}