crates/knut-bifrost/src/doris_cli.rs
//! `bifrost doris …` — an Iceberg table's commits into an Apache Doris table,
//! as a command, as a systemd service and as a Kubernetes workload.
//!
//! Feature `service` (`cli` + `iceberg-doris` + skidbladnir).
//!
//! ```text
//! bifrost doris run <source> <doris> [--state F] [--once] [--interval S]
//! bifrost doris unit <source> <doris> [--system] # print the unit `install` writes
//! bifrost doris install <source> <doris> [--system] # write + enable + start (skidbladnir)
//! bifrost doris uninstall [--name N] [--system] [--purge]
//! bifrost doris start|stop|status [--name N] [--system]
//! bifrost doris k8s <source> <doris> --image I [--namespace NS] --out DIR
//! ```
//!
//! # One spec, two packages
//!
//! [`service_spec`] builds ONE `skidbladnir::service::ServiceSpec` from the
//! parsed arguments. `install` hands it to skidbladnir's systemd `Lifecycle`;
//! `k8s` hands the same spec to skidbladnir's `KubernetesRenderer` (a
//! StatefulSet with a volume for the cursor, a Secret for the credentials). The
//! command line the service runs is rendered from the parsed arguments, never
//! copied from argv, and both packages run the same one — a test holds them
//! byte-identical.
//!
//! # Secrets never go on the command line
//!
//! `MINIO_ACCESS_KEY`, `MINIO_SECRET_KEY`, `DORIS_USER`, `DORIS_PASSWORD` (and
//! optionally `BIFROST_S3_CA_PEM`, the CA as PEM text, and the catalog's
//! `BIFROST_CATALOG_CREDENTIAL` / `BIFROST_CATALOG_TOKEN`) come from the
//! environment only: flags land in `/proc/<pid>/cmdline`. At install time they
//! are read from the installer's environment and routed to skidbladnir's secret
//! channel — a 0600 `EnvironmentFile=` for systemd, a Secret for Kubernetes.
use std::path::{Path, PathBuf};
use std::time::Duration;
use clap::{Args, Subcommand, ValueEnum};
use crate::delivery::{DeliveryOptions, IcebergDelivery, IcebergLocator, bucket_of, s3_ca, s3_config};
use crate::error::{BifrostError, Result};
use crate::sink::doris::{DorisConfig, DorisSink, LoadFormat, LoadMode, ReqwestTransport};
use crate::state::SyncState;
/// The secrets a delivery reads from its environment. Never flags.
pub const SECRET_ENV: [&str; 7] = [
"MINIO_ACCESS_KEY",
"MINIO_SECRET_KEY",
"DORIS_USER",
"DORIS_PASSWORD",
"BIFROST_S3_CA_PEM",
crate::catalog_args::CREDENTIAL_ENV,
crate::catalog_args::TOKEN_ENV,
];
pub use crate::service_paths::{IMAGE_EXEC, K8S_STATE_DIR, installed_state_dir};
#[derive(Subcommand, Debug, Clone)]
pub enum DorisCommand {
/// Deliver: poll the table and load every new commit into Doris.
Run(RunArgs),
/// Print the systemd unit `install` would write.
Unit(ServiceArgs),
/// Write, enable and start the systemd unit (skidbladnir).
Install(ServiceArgs),
/// Stop, disable and remove the unit. The cursor stays unless `--purge`.
Uninstall(LifecycleArgs),
/// Start the installed unit.
Start(LifecycleArgs),
/// Stop the installed unit.
Stop(LifecycleArgs),
/// Print `active` or `inactive`.
Status(LifecycleArgs),
/// Render the Kubernetes package (StatefulSet + Secret + volume) into `--out`.
K8s(K8sArgs),
}
/// Where the Iceberg table is. Exactly one of `--warehouse`, `--s3-table`,
/// `--rest-uri`, `--grpc-uri`.
#[derive(Args, Debug, Clone, Default, PartialEq, Eq)]
pub struct SourceArgs {
/// An embedded skade warehouse directory (needs `--table`).
#[arg(long, value_name = "DIR", conflicts_with_all = ["s3_table", "rest_uri", "grpc_uri"])]
pub warehouse: Option<PathBuf>,
/// A table root on S3/MinIO, read with no catalog, e.g.
/// `s3://red/warehouse/bench/smoke10m`.
#[arg(long, value_name = "S3_URL", conflicts_with_all = ["rest_uri", "grpc_uri"])]
pub s3_table: Option<String>,
/// An Iceberg REST catalog, e.g. `https://minio:9000/_iceberg` (needs `--table`);
/// `--catalog` names a vendor preset.
#[arg(long, value_name = "URL", conflicts_with = "grpc_uri")]
pub rest_uri: Option<String>,
/// The REST `warehouse` (AIStor: the bucket).
#[arg(long)]
pub rest_warehouse: Option<String>,
/// Sign REST requests with SigV4 for this service (AIStor: `s3tables`).
#[arg(long, value_name = "SERVICE")]
pub sigv4: Option<String>,
/// The table: `ns.name` for REST, `name` or `ns.name` for a warehouse.
#[arg(long)]
pub table: Option<String>,
/// S3/MinIO endpoint for the table's files, e.g. `https://minio:9000`.
#[arg(long, value_name = "URL")]
pub s3_endpoint: Option<String>,
/// S3 region (MinIO ignores it; SigV4 needs one).
#[arg(long, default_value = "us-east-1")]
pub s3_region: String,
/// Bucket of the table's files. Defaults to the `--s3-table` bucket, then
/// to `--rest-warehouse`.
#[arg(long)]
pub s3_bucket: Option<String>,
/// CA certificate (PEM file) to verify an `https://` S3/REST endpoint with.
/// Alternatively `BIFROST_S3_CA_PEM` holds the PEM text itself.
#[arg(long, value_name = "PEM")]
pub s3_ca: Option<PathBuf>,
/// Which catalog `--rest-uri` is, and the `--grpc-uri` catalog.
#[command(flatten)]
pub catalog: crate::catalog_args::CatalogArgs,
}
/// Where the rows go.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct DorisArgs {
/// Doris FE HTTP base, e.g. `http://doris-fe:8030`.
#[arg(long, value_name = "URL")]
pub doris_fe: String,
/// PUT straight at this BE instead of following the FE's redirect.
#[arg(long, value_name = "URL")]
pub doris_be: Option<String>,
/// Destination database.
#[arg(long)]
pub doris_db: String,
/// Destination table.
#[arg(long)]
pub doris_table: String,
/// `upsert` (unique-key table, deletes applied) or `append` (duplicate-key
/// table, a delete is an error).
#[arg(long, value_enum, default_value_t = ModeArg::Upsert)]
pub mode: ModeArg,
/// Stream Load body format.
#[arg(long, value_enum, default_value_t = FormatArg::Json)]
pub format: FormatArg,
/// Chunks of one window loaded at once (each its own label).
#[arg(long, default_value_t = 4)]
pub parallel_loads: usize,
/// Body budget per Stream Load, MiB.
#[arg(long, default_value_t = 64)]
pub chunk_mb: usize,
/// The destination's key columns (comma-separated), for a source table that
/// declares no Iceberg identifier fields (a plain Spark table).
#[arg(long = "key", value_delimiter = ',', value_name = "COL")]
pub key: Vec<String>,
/// Do not version rows by the Iceberg sequence number (a table created
/// without `function_column.sequence_type`).
#[arg(long)]
pub no_sequence: bool,
/// Load without Stream Load two-phase commit (chunks become visible one by one).
#[arg(long)]
pub no_two_phase_commit: bool,
}
#[derive(ValueEnum, Debug, Clone, Copy, PartialEq, Eq)]
pub enum ModeArg {
Upsert,
Append,
}
#[derive(ValueEnum, Debug, Clone, Copy, PartialEq, Eq)]
pub enum FormatArg {
Json,
Arrow,
}
impl ModeArg {
fn as_str(self) -> &'static str {
match self {
ModeArg::Upsert => "upsert",
ModeArg::Append => "append",
}
}
}
impl FormatArg {
fn as_str(self) -> &'static str {
match self {
FormatArg::Json => "json",
FormatArg::Arrow => "arrow",
}
}
}
/// `run`: the source, the destination and the loop.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct RunArgs {
#[command(flatten)]
pub source: SourceArgs,
#[command(flatten)]
pub doris: DorisArgs,
/// The cursor file (per-table last delivered snapshot).
#[arg(long, value_name = "FILE", default_value = "bifrost-doris-cursor.json")]
pub state: PathBuf,
/// One round, then exit.
#[arg(long)]
pub once: bool,
/// Seconds between rounds while healthy (near-live: a round that finds no
/// new snapshot reads one metadata file).
#[arg(long, default_value_t = 5)]
pub interval: u64,
/// Cap (seconds) for the failure backoff.
#[arg(long, default_value_t = 300)]
pub max_backoff: u64,
/// At most this many one-commit windows per round.
#[arg(long, default_value_t = 64)]
pub max_windows: usize,
}
/// `unit` / `install`: what to run, and as which service.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct ServiceArgs {
#[command(flatten)]
pub run: RunArgs,
#[command(flatten)]
pub life: LifecycleArgs,
}
/// Which installed service a lifecycle verb acts on.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct LifecycleArgs {
/// Service name (one per delivered table).
#[arg(long, default_value = "bifrost-doris")]
pub name: String,
/// System scope (its own `bifrost` user); default is the user scope.
#[arg(long)]
pub system: bool,
/// With `uninstall`: also erase the cursor.
#[arg(long)]
pub purge: bool,
}
/// `k8s`: the Kubernetes package.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct K8sArgs {
#[command(flatten)]
pub run: RunArgs,
/// Workload name.
#[arg(long, default_value = "bifrost-doris")]
pub name: String,
/// The image carrying `/usr/local/bin/bifrost`.
#[arg(long)]
pub image: String,
#[arg(long, default_value = "default")]
pub namespace: String,
/// Size of the cursor volume.
#[arg(long, default_value = "1Gi")]
pub state_size: String,
/// Directory the manifests are written into (the Secret file is 0600).
#[arg(long, value_name = "DIR")]
pub out: PathBuf,
}
// ── run ─────────────────────────────────────────────────────────────────────
fn env(name: &str) -> Option<String> {
std::env::var(name).ok().filter(|v| !v.is_empty())
}
impl SourceArgs {
fn s3(&self, state_dir: &Path) -> Result<skade::S3Config> {
let endpoint = self
.s3_endpoint
.clone()
.ok_or_else(|| BifrostError::Other("--s3-endpoint is required for --s3-table / --rest-uri".into()))?;
let bucket = self
.s3_bucket
.clone()
.or_else(|| self.s3_table.as_deref().and_then(bucket_of).map(str::to_string))
.or_else(|| self.rest_warehouse.clone())
.ok_or_else(|| BifrostError::Other("--s3-bucket is required (no bucket in --s3-table or --rest-warehouse)".into()))?;
// `--s3-ca`, else `BIFROST_S3_CA_PEM` written beside the cursor.
let ca = s3_ca(self.s3_ca.as_deref(), state_dir)?;
s3_config(endpoint, self.s3_region.clone(), bucket, ca)
}
/// The table, found the way the flags say.
pub fn locator(&self, state_dir: &Path) -> Result<IcebergLocator> {
self.catalog.check_unused(self.rest_uri.is_some()).map_err(BifrostError::Other)?;
if let Some(dir) = &self.warehouse {
let table = self.table.clone().ok_or_else(|| BifrostError::Other("--warehouse needs --table".into()))?;
return Ok(IcebergLocator::Warehouse { dir: dir.clone(), table });
}
if let Some(loc) = &self.s3_table {
return Ok(IcebergLocator::S3(skade::connect::S3Table { s3: self.s3(state_dir)?, location: loc.clone() }));
}
if let Some(uri) = &self.rest_uri {
let table = self.table.clone().ok_or_else(|| BifrostError::Other("--rest-uri needs --table ns.name".into()))?;
let catalog = self
.catalog
.rest(uri, self.rest_warehouse.clone(), self.sigv4.clone(), self.s3(state_dir)?, true)
.map_err(BifrostError::Other)?;
return Ok(IcebergLocator::catalog(catalog, table));
}
if let Some(uri) = &self.catalog.grpc_uri {
let table = self.table.clone().ok_or_else(|| BifrostError::Other("--grpc-uri needs --table ns.name".into()))?;
#[cfg(feature = "skade-grpc")]
{
// The data: through S3 when `--s3-endpoint` names one, else the local filesystem.
let data = if self.s3_endpoint.is_some() { Some(self.s3(state_dir)?) } else { None };
return Ok(IcebergLocator::catalog(self.catalog.grpc(uri, data, true), table));
}
#[cfg(not(feature = "skade-grpc"))]
return Err(BifrostError::Other(format!(
"--grpc-uri {uri} (table {table}): this build has no gRPC catalog client - rebuild with --features skade-grpc"
)));
}
Err(BifrostError::Other("name the table: --warehouse DIR, --s3-table s3://…, --rest-uri URL, or --grpc-uri URL".into()))
}
/// The flags that recreate this source — no secrets among them.
fn to_args(&self) -> Vec<String> {
let mut a = Vec::new();
let mut push = |k: &str, v: Option<String>| {
if let Some(v) = v {
a.push(format!("--{k}"));
a.push(v);
}
};
push("warehouse", self.warehouse.as_ref().map(|p| p.display().to_string()));
push("s3-table", self.s3_table.clone());
push("rest-uri", self.rest_uri.clone());
push("rest-warehouse", self.rest_warehouse.clone());
push("sigv4", self.sigv4.clone());
push("table", self.table.clone());
push("s3-endpoint", self.s3_endpoint.clone());
push("s3-region", Some(self.s3_region.clone()));
push("s3-bucket", self.s3_bucket.clone());
push("s3-ca", self.s3_ca.as_ref().map(|p| p.display().to_string()));
a.extend(self.catalog.to_args());
a
}
}
impl DorisArgs {
fn config(&self) -> Result<DorisConfig> {
let mut cfg = DorisConfig::from_env(&self.doris_fe, &self.doris_db, &self.doris_table)?
.with_mode(match self.mode {
ModeArg::Upsert => LoadMode::Upsert,
ModeArg::Append => LoadMode::Append,
})
.with_max_chunk_bytes(self.chunk_mb.max(1) * 1024 * 1024)
.with_max_parallel_loads(self.parallel_loads)
.with_key_columns(self.key.clone())
.with_sequence(!self.no_sequence)
.with_two_phase_commit(!self.no_two_phase_commit);
cfg.format = match self.format {
FormatArg::Json => LoadFormat::Json,
FormatArg::Arrow => LoadFormat::Arrow,
};
if let Some(be) = &self.doris_be {
cfg = cfg.with_be_url(be);
}
Ok(cfg)
}
fn destination(&self) -> String {
format!("doris:{}.{}", self.doris_db, self.doris_table)
}
fn to_args(&self) -> Vec<String> {
let mut a = vec![
"--doris-fe".into(),
self.doris_fe.clone(),
"--doris-db".into(),
self.doris_db.clone(),
"--doris-table".into(),
self.doris_table.clone(),
"--mode".into(),
self.mode.as_str().into(),
"--format".into(),
self.format.as_str().into(),
"--chunk-mb".into(),
self.chunk_mb.to_string(),
"--parallel-loads".into(),
self.parallel_loads.to_string(),
];
if let Some(be) = &self.doris_be {
a.extend(["--doris-be".into(), be.clone()]);
}
if !self.key.is_empty() {
a.extend(["--key".into(), self.key.join(",")]);
}
if self.no_sequence {
a.push("--no-sequence".into());
}
if self.no_two_phase_commit {
a.push("--no-two-phase-commit".into());
}
a
}
}
impl RunArgs {
/// `doris run …` with these arguments and the cursor at `state` — the one
/// command line both packages run.
pub fn command_args(&self, state: &Path) -> Vec<String> {
let mut a = vec!["doris".to_string(), "run".to_string()];
a.extend(self.source.to_args());
a.extend(self.doris.to_args());
a.extend([
"--state".into(),
state.display().to_string(),
"--interval".into(),
self.interval.to_string(),
"--max-backoff".into(),
self.max_backoff.to_string(),
"--max-windows".into(),
self.max_windows.to_string(),
]);
a
}
fn state_dir(&self) -> PathBuf {
self.state.parent().filter(|p| !p.as_os_str().is_empty()).map(Path::to_path_buf).unwrap_or_else(|| PathBuf::from("."))
}
}
/// `doris run`: deliver until SIGTERM/ctrl-c (or one round with `--once`).
pub async fn run(args: &RunArgs) -> Result<()> {
let locator = args.source.locator(&args.state_dir())?;
let cfg = args.doris.config()?;
let transport = ReqwestTransport::new(&cfg)?;
let sink = DorisSink::new(cfg, transport);
let state = SyncState::open(&args.state)?;
let opts = DeliveryOptions { max_windows_per_round: args.max_windows.max(1), key_columns: args.doris.key.clone() };
let mut d = IcebergDelivery::new(locator, &args.doris.destination(), sink, state, opts);
tracing::info!(target: "bifrost::delivery", event = "start", table = %d.cursor_key(), cursor = ?d.cursor(), "delivering");
if args.once {
let o = d.round().await?;
tracing::info!(
target: "bifrost::delivery",
event = "round",
table = %d.cursor_key(),
windows = o.windows,
loaded = o.rows_loaded,
deleted = o.rows_deleted,
duplicate = o.rows_duplicate,
at = ?o.at_snapshot,
behind = o.behind,
read_ms = o.read.as_millis() as u64,
load_ms = o.load.as_millis() as u64,
"delivered",
);
return Ok(());
}
d.run(Duration::from_secs(args.interval.max(1)), Duration::from_secs(args.max_backoff.max(1)), shutdown_signal())
.await
}
/// SIGTERM or ctrl-c. The loop finishes the round in flight first.
async fn shutdown_signal() {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
match signal(SignalKind::terminate()) {
Ok(mut term) => {
tokio::select! {
_ = term.recv() => {}
_ = tokio::signal::ctrl_c() => {}
}
}
Err(_) => {
let _ = tokio::signal::ctrl_c().await;
}
}
}
#[cfg(not(unix))]
{
let _ = tokio::signal::ctrl_c().await;
}
}
// ── the one service spec ────────────────────────────────────────────────────
/// The ONE spec both packages are rendered from. `exec` is the binary path the
/// target runs, `state` the cursor file there; secrets are copied from the
/// installer's environment into the secret channel, never into the unit.
pub fn service_spec(
name: &str,
exec: &str,
run: &RunArgs,
state: &Path,
system: bool,
) -> skidbladnir::service::ServiceSpec {
use skidbladnir::service::{AgentMode, ServiceSpec};
use skidbladnir::systemd::{RestartPolicy, Scope};
let mut spec = ServiceSpec::new(name, "bifrost: Iceberg commits into Apache Doris", exec)
.scope(if system { Scope::System } else { Scope::User })
.restart(RestartPolicy::OnFailure)
.agent(AgentMode::Fat)
.env("RUST_LOG", "info");
if let Some(dir) = state.parent() {
spec = spec.working_dir(dir.to_string_lossy());
}
if system {
spec = spec.user("bifrost");
}
for a in run.command_args(state) {
spec = spec.arg(a);
}
for k in SECRET_ENV {
if let Some(v) = env(k) {
spec = spec.secret_env(k, v);
}
}
spec
}
/// The systemd lifecycle for `name`, spec built from `run` when given.
fn lifecycle(run: Option<&RunArgs>, life: &LifecycleArgs) -> Result<skidbladnir::service::Lifecycle> {
let exec = std::env::current_exe()
.map(|p| p.to_string_lossy().into_owned())
.unwrap_or_else(|_| "/usr/local/bin/bifrost".into());
let state_dir = installed_state_dir(&life.name, life.system);
let state = state_dir.join("cursor.json");
let mut spec = match run {
Some(r) => service_spec(&life.name, &exec, r, &state, life.system),
None => skidbladnir::service::ServiceSpec::new(&life.name, "bifrost: Iceberg commits into Apache Doris", &exec).scope(
if life.system { skidbladnir::systemd::Scope::System } else { skidbladnir::systemd::Scope::User },
),
};
// skidbladnir's uninstall erases a declared data root (its law: data dies).
// The cursor is what lets a reinstall resume without re-loading, so it is
// declared only when the operator asked for `--purge`.
if life.purge {
let root = skidbladnir::uninstall::DataRoot::from_env("BIFROST_DORIS_DATA", state_dir)
.map_err(|e| BifrostError::Other(format!("state dir: {e}")))?;
spec = spec.data_root(&root);
}
Ok(skidbladnir::service::Lifecycle::new(spec))
}
fn sk(e: anyhow::Error) -> BifrostError {
BifrostError::Other(format!("skidbladnir: {e:#}"))
}
/// The Kubernetes package: the same spec as `install`, run from the image's
/// binary with its cursor on a volume.
pub fn k8s_files(args: &K8sArgs) -> Vec<skidbladnir::service::RenderedFile> {
use skidbladnir::service::{ServiceRenderer as _, Storage};
let state = Path::new(K8S_STATE_DIR).join("cursor.json");
let spec = service_spec(&args.name, IMAGE_EXEC, &args.run, &state, false)
.image(&args.image)
.namespace(&args.namespace)
.replicas(1)
.stateful(true)
.storage(Storage::new(&args.state_size, K8S_STATE_DIR));
skidbladnir::k8s::KubernetesRenderer.render(&spec).files
}
/// Dispatch `bifrost doris <verb>`.
pub async fn dispatch(cmd: DorisCommand) -> Result<()> {
match cmd {
DorisCommand::Run(a) => run(&a).await,
DorisCommand::Unit(a) => {
let lc = lifecycle(Some(&a.run), &a.life)?;
print!("{}", crate::service_paths::unit_text(&lc, &a.life.name, a.life.system));
Ok(())
}
DorisCommand::Install(a) => {
let lc = lifecycle(Some(&a.run), &a.life)?;
crate::service_paths::install(&lc, &a.life.name, a.life.system).map_err(sk)?;
lc.enable().map_err(sk)?;
lc.start().map_err(sk)?;
println!("bifrost doris: INSTALLED and started ({})", lc.unit());
Ok(())
}
DorisCommand::Uninstall(l) => {
let r = lifecycle(None, &l)?.uninstall().map_err(sk)?;
let env = crate::service_paths::remove_env_file(&l.name, l.system)?;
println!(
"bifrost doris: unit removed = {}, env file removed = {env}, erased {} path(s)",
r.unit_removed,
r.erased.len()
);
Ok(())
}
DorisCommand::Start(l) => lifecycle(None, &l)?.start().map_err(sk),
DorisCommand::Stop(l) => lifecycle(None, &l)?.stop().map_err(sk),
DorisCommand::Status(l) => {
println!("{}", if lifecycle(None, &l)?.status().map_err(sk)? { "active" } else { "inactive" });
Ok(())
}
DorisCommand::K8s(a) => {
std::fs::create_dir_all(&a.out)?;
for f in k8s_files(&a) {
let file = a.out.join(Path::new(&f.path).file_name().unwrap_or(f.path.as_ref()));
std::fs::write(&file, &f.contents)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&file, std::fs::Permissions::from_mode(f.mode))?;
}
println!("{}", file.display());
}
Ok(())
}
}
}
#[cfg(test)]
mod tests {
use clap::Parser;
use super::*;
#[derive(Parser)]
struct T {
#[command(subcommand)]
cmd: DorisCommand,
}
fn parse(s: &str) -> DorisCommand {
T::try_parse_from(std::iter::once("t").chain(s.split_whitespace())).unwrap().cmd
}
const AISTOR: &str = "--rest-uri https://minio:9000/_iceberg --rest-warehouse red --sigv4 s3tables \
--table bench.people --s3-endpoint https://minio:9000 \
--doris-fe http://fe:8030 --doris-db lake --doris-table people";
#[test]
fn the_command_line_round_trips_through_its_own_rendering() {
let DorisCommand::Run(a) = parse(&format!("run {AISTOR} --state /s/c.json --interval 5")) else { panic!() };
let rendered = a.command_args(Path::new("/s/c.json"));
assert_eq!(&rendered[..2], ["doris", "run"]);
let DorisCommand::Run(b) = T::try_parse_from(std::iter::once("t".to_string()).chain(rendered.into_iter().skip(1)))
.unwrap()
.cmd
else {
panic!()
};
assert_eq!(a, b, "what the service runs parses back to what was asked for");
}
#[test]
fn two_sources_at_once_is_refused_by_clap() {
let r = T::try_parse_from(
"t run --warehouse /w --s3-table s3://b/x/y --doris-fe f --doris-db d --doris-table t".split_whitespace(),
);
assert!(r.is_err());
}
#[test]
fn systemd_and_k8s_run_the_same_command_and_no_secret_is_on_it() {
// SAFETY of the test: the env var is set only in this process.
unsafe { std::env::set_var("DORIS_PASSWORD", "hunter2-not-on-argv") };
let DorisCommand::K8s(k) = parse(&format!("k8s {AISTOR} --image reg/bifrost:1 --out /tmp/x")) else { panic!() };
let state = Path::new(K8S_STATE_DIR).join("cursor.json");
let spec = service_spec("bifrost-doris", IMAGE_EXEC, &k.run, &state, false);
let unit = spec.render_systemd_unit();
let files = k8s_files(&k);
let workload = files.iter().find(|f| f.contents.contains("kind: StatefulSet")).expect("a StatefulSet");
for a in k.run.command_args(&state) {
assert!(workload.contents.contains(&a), "k8s args carry {a}");
assert!(unit.contains(&a), "the unit carries {a}");
}
assert!(!unit.contains("hunter2"), "a secret never lands in the 0644 unit");
assert!(!workload.contents.contains("hunter2"), "a secret never lands in the workload");
let secret = files.iter().find(|f| f.contents.contains("kind: Secret")).expect("a Secret for the credentials");
assert_eq!(secret.mode, 0o600);
assert!(workload.contents.contains(K8S_STATE_DIR), "the cursor lives on the volume");
}
#[test]
fn bucket_comes_from_the_table_url() {
assert_eq!(bucket_of("s3://red/warehouse/bench/t"), Some("red"));
assert_eq!(bucket_of("s3a://b/x"), Some("b"));
assert_eq!(bucket_of("/local/x"), None);
}
}