crates/knut-bifrost/tests/common/rig.rs
//! **The container rig**: MinIO + Apache Doris 4.1 (one FE, one BE) through
//! draupnir's one OCI engine, the BE's storage on this host's NVMe.
//!
//! Built for loki (the bench host); runs anywhere with a rootless podman socket.
//!
//! * Host networking for all three, so the FE's 307 names a BE address this
//! process can reach (the one hang the Doris sink is built to survive), and
//! no port mapping sits in the measured path.
//! * The FE/BE images run their own compose entrypoints (`init_fe.sh` /
//! `entry_point.sh`), configured by env (`FE_SERVERS`, `FE_ID`, `BE_ADDR`) —
//! draupnir sets a container's Cmd, not its Entrypoint, so the image's own
//! entrypoint always runs — with a conf the rig writes — the image's own conf plus
//! `priority_networks = 127.0.0.1/32` and, for the BE,
//! `storage_root_path = …,medium:ssd` on a bind-mounted NVMe directory — and
//! registers the BE over the FE's HTTP SQL API. No mysql client, no shell.
//! * `SKIP_CHECK_ULIMIT=true` on the BE: `start_be.sh` refuses a host whose
//! `vm.max_map_count` is below 2 000 000 (loki: 1 048 576; raising it needs
//! root). Fine for a rig, wrong for production — said here so nobody copies it.
//!
//! Containers are left running between tests (reuse is by health probe) and
//! taken down by the `rig_down` test.
use std::path::{Path, PathBuf};
use std::time::Duration;
use draupnir::container::{ContainerBoot, ContainerControl};
use draupnir::{BootSpec, Machine, NetMode};
use super::DorisSql;
pub const MINIO_IMAGE: &str = "quay.io/minio/minio:RELEASE.2025-04-08T15-41-24Z";
pub const DORIS_VERSION: &str = "4.1.4";
pub const MINIO_PORT: u16 = 19000;
pub const MINIO_USER: &str = "rig";
pub const MINIO_PASSWORD: &str = "rig-secret-123";
pub const BUCKET: &str = "lake";
pub const FE_HTTP: &str = "http://127.0.0.1:8030";
const NAMES: [&str; 3] = ["knut-rig-minio", "knut-rig-doris-fe", "knut-rig-doris-be"];
pub struct Rig {
pub dir: PathBuf,
pub s3_endpoint: String,
}
fn boot() -> ContainerBoot {
ContainerBoot::new()
}
/// The rig's data root: `$RIG_DIR`, else `~/t14s-runs/knut-doris/rig`.
pub fn rig_dir() -> PathBuf {
std::env::var_os("RIG_DIR").map(PathBuf::from).unwrap_or_else(|| {
PathBuf::from(std::env::var("HOME").expect("HOME")).join("t14s-runs/knut-doris/rig")
})
}
async fn http_ok(url: &str) -> bool {
tls_roots(reqwest::Client::builder()).build().expect("reqwest client")
.get(url)
.timeout(Duration::from_secs(3))
.send()
.await
.map(|r| r.status().is_success())
.unwrap_or(false)
}
async fn blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> T {
tokio::task::spawn_blocking(f).await.expect("blocking task")
}
/// The image's own conf file, plus `extra` lines (later keys win in Doris confs).
async fn conf_from_image(image: &str, path_in_image: &str, dest: &Path, extra: &str) {
let tmp = dest.with_extension("extract");
std::fs::create_dir_all(&tmp).unwrap();
let (img, p, t) = (image.to_string(), path_in_image.to_string(), tmp.clone());
blocking(move || boot().extract_path(&img, &p, &t)).await.expect("extract the image's conf");
let base = std::fs::read_to_string(tmp.join(Path::new(path_in_image).file_name().unwrap())).unwrap();
// JDK 17's cgroup-v2 probe NPEs ("anyController is null") in a rootless
// container with no delegated controllers, and BDB JE dies on it
// (JVMSystemUtils) before the FE ever answers — measured on loki
// 2026-09-26. The JVM's container detection is switched off; the heap is
// the conf's own -Xmx either way.
let base: String = base
.lines()
.map(|l| match l.strip_prefix("JAVA_OPTS_FOR_JDK_17=\"").and_then(|r| r.strip_suffix('"')) {
Some(opts) => format!("JAVA_OPTS_FOR_JDK_17=\"{opts} -XX:-UseContainerSupport\"\n"),
None => format!("{l}\n"),
})
.collect();
std::fs::write(dest, format!("{base}\n# ── knut container rig ──\n{extra}\n")).unwrap();
}
fn spec(name: &str, image: &str, cmd: &[&str], env: &[(&str, &str)], binds: Vec<String>) -> (BootSpec, Vec<String>) {
let mut s = BootSpec::container(name, image).with_cmd(cmd.iter().map(|s| s.to_string()));
for (k, v) in env {
s.env.insert((*k).into(), (*v).into());
}
s.net = NetMode::Host;
(s, binds)
}
impl Rig {
/// Bring the rig up (or find it up): MinIO with bucket `lake`, a Doris FE
/// answering SQL, and one BE alive on NVMe storage.
pub async fn up() -> Rig {
let dir = rig_dir();
for d in ["minio", "fe-meta", "fe-log", "be-storage", "be-log", "conf"] {
std::fs::create_dir_all(dir.join(d)).unwrap();
}
let s3_endpoint = format!("http://127.0.0.1:{MINIO_PORT}");
// ── MinIO ──
if !http_ok(&format!("{s3_endpoint}/minio/health/live")).await {
let (s, b) = spec(
NAMES[0],
MINIO_IMAGE,
&["server", "/data", "--address", &format!(":{MINIO_PORT}"), "--console-address", ":19001"],
&[("MINIO_ROOT_USER", MINIO_USER), ("MINIO_ROOT_PASSWORD", MINIO_PASSWORD)],
vec![format!("{}:/data", dir.join("minio").display())],
);
blocking(move || boot().boot_with_binds(&s, &b)).await.expect("boot minio");
let live = format!("http://127.0.0.1:{MINIO_PORT}/minio/health/live");
wait_for(|| http_ok(&live), "minio", 120).await;
}
let ep = s3_endpoint.clone();
blocking(move || {
let c = aegir::Client::new(ep, "us-east-1", MINIO_USER, MINIO_PASSWORD);
let _ = c.create_bucket(BUCKET); // exists already = fine
})
.await;
// ── Doris FE ──
let sql = DorisSql::new(FE_HTTP, "root", "");
if !http_ok(&format!("{FE_HTTP}/api/bootstrap")).await {
let img = format!("docker.io/apache/doris:fe-{DORIS_VERSION}");
let conf = dir.join("conf/fe.conf");
conf_from_image(&img, "/opt/apache-doris/fe/conf/fe.conf", &conf, "priority_networks = 127.0.0.1/32").await;
let (s, b) = spec(
NAMES[1],
&img,
&[],
&[("FE_SERVERS", "fe1:127.0.0.1:9010"), ("FE_ID", "1")],
vec![
format!("{}:/opt/apache-doris/fe/conf/fe.conf", conf.display()),
format!("{}:/opt/apache-doris/fe/doris-meta", dir.join("fe-meta").display()),
format!("{}:/opt/apache-doris/fe/log", dir.join("fe-log").display()),
],
);
blocking(move || boot().boot_with_binds(&s, &b)).await.expect("boot doris fe");
}
sql.wait_ready(Duration::from_secs(300)).await;
// ── Doris BE, storage on NVMe, marked SSD ──
if !http_ok("http://127.0.0.1:8040/api/health").await {
let img = format!("docker.io/apache/doris:be-{DORIS_VERSION}");
let conf = dir.join("conf/be.conf");
conf_from_image(
&img,
"/opt/apache-doris/be/conf/be.conf",
&conf,
"priority_networks = 127.0.0.1/32\nstorage_root_path = /opt/apache-doris/be/storage,medium:ssd",
)
.await;
let (s, b) = spec(
NAMES[2],
&img,
&[],
&[("FE_SERVERS", "fe1:127.0.0.1:9010"), ("BE_ADDR", "127.0.0.1:9050"), ("SKIP_CHECK_ULIMIT", "true")],
vec![
format!("{}:/opt/apache-doris/be/conf/be.conf", conf.display()),
format!("{}:/opt/apache-doris/be/storage", dir.join("be-storage").display()),
format!("{}:/opt/apache-doris/be/log", dir.join("be-log").display()),
],
);
blocking(move || boot().boot_with_binds(&s, &b)).await.expect("boot doris be");
wait_for(|| http_ok("http://127.0.0.1:8040/api/health"), "doris be", 300).await;
}
let backends = sql.exec("SHOW BACKENDS").await;
if backends.is_empty() {
sql.exec("ALTER SYSTEM ADD BACKEND \"127.0.0.1:9050\"").await;
}
let t0 = std::time::Instant::now();
loop {
let rows = sql.exec("SHOW BACKENDS").await;
if rows.iter().any(|r| r.iter().any(|c| c.as_str() == Some("true"))) {
break;
}
assert!(t0.elapsed() < Duration::from_secs(300), "the BE never came alive: {rows:?}");
tokio::time::sleep(Duration::from_secs(3)).await;
}
Rig { dir, s3_endpoint }
}
pub fn s3(&self) -> skade::S3Config {
skade::S3Config {
endpoint: self.s3_endpoint.clone(),
region: "us-east-1".into(),
bucket: BUCKET.into(),
access_key_id: MINIO_USER.into(),
secret_access_key: MINIO_PASSWORD.into(),
path_style: true,
tls: None,
}
}
/// The env the Doris sink and `DorisSql::from_env` read.
pub fn export_env(&self) {
// SAFETY: tests set these before spawning anything that reads them.
unsafe {
std::env::set_var("DORIS_USER", "root");
std::env::set_var("DORIS_PASSWORD", "");
std::env::set_var("MINIO_ACCESS_KEY", MINIO_USER);
std::env::set_var("MINIO_SECRET_KEY", MINIO_PASSWORD);
}
}
/// Stop and remove the three containers. Data directories stay.
pub async fn down() {
for n in NAMES {
let m = Machine::started(format!("draupnir-{n}"), &BootSpec::container(n, "unused"));
blocking(move || boot().stop(&m)).await;
}
}
}
async fn wait_for<F, Fut>(mut probe: F, what: &str, secs: u64)
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = bool>,
{
let t0 = std::time::Instant::now();
while !probe().await {
assert!(t0.elapsed() < Duration::from_secs(secs), "{what} never became healthy");
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots(b: reqwest::ClientBuilder) -> reqwest::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}