crates/knut-bifrost/src/falkordb_cli.rs
//! `bifrost falkordb …` — `bifrost watch` (an Iceberg lake streamed into a
//! FalkorDB graph) as a systemd service and as a Kubernetes workload, through
//! skidbladnir: the house's one system installer.
//!
//! Feature `service`.
//!
//! ```text
//! bifrost falkordb unit <watch flags> [--name N] [--system] # print the unit `install` writes
//! bifrost falkordb install <watch flags> [--name N] [--system] # write + enable + start
//! bifrost falkordb uninstall [--name N] [--system] [--purge]
//! bifrost falkordb start|stop|status [--name N] [--system]
//! bifrost falkordb k8s <watch flags> --image I [--namespace NS] --out DIR
//! ```
//!
//! `<watch flags>` are exactly `bifrost watch`'s: the lake (`--warehouse`,
//! `--s3-root`, `--rest-uri` with `--catalog`, `--grpc-uri`), `--mapping`,
//! `--falkor`, `--graph`, TLS, interval, metrics.
//!
//! # One spec, two packages
//!
//! [`service_spec`] builds ONE `skidbladnir::service::ServiceSpec`; `install`
//! hands it to skidbladnir's systemd `Lifecycle`, `k8s` to its
//! `KubernetesRenderer` (a StatefulSet, a volume for the cursor, a Secret for
//! the credentials). The command line is rendered from the parsed flags
//! ([`WatchArgs::to_args`]), never copied from argv. The cursor is always the
//! service's own (`<state dir>/cursor.json`), so `--state`, `--once` and
//! `--dry-run` are refused: a service polls forever into a real graph.
//!
//! # Secrets never go on the command line
//!
//! [`SECRET_ENV`] — the S3 keys, the S3 CA as PEM text, the FalkorDB URL when it
//! carries a password (`BIFROST_FALKOR_URL`), the catalog's OAuth2 credential
//! and token — are read from the installer's environment and routed to
//! skidbladnir's secret channel: a 0600 `EnvironmentFile=` for systemd, a
//! Secret for Kubernetes. A `--falkor` URL with credentials in it is refused.
//!
//! # Kubernetes and files
//!
//! skidbladnir renders no ConfigMap, so every file the watch reads — the
//! mapping, PEMs — must already be in the image: `k8s` takes their paths as
//! paths **inside the image** and refuses relative ones. The S3 CA can come as
//! PEM text through `BIFROST_S3_CA_PEM` instead.
use std::path::{Path, PathBuf};
use clap::{Args, Subcommand};
use crate::catalog_args::{CREDENTIAL_ENV, TOKEN_ENV};
use crate::service_paths::{IMAGE_EXEC, K8S_STATE_DIR, installed_state_dir};
use crate::error::{BifrostError, Result};
use crate::watch_cli::{FALKOR_URL_ENV, WatchArgs};
/// The secrets a watch reads from its environment. Never flags.
pub const SECRET_ENV: [&str; 6] = [
"MINIO_ACCESS_KEY",
"MINIO_SECRET_KEY",
"BIFROST_S3_CA_PEM",
FALKOR_URL_ENV,
CREDENTIAL_ENV,
TOKEN_ENV,
];
/// The default service / workload name.
pub const DEFAULT_NAME: &str = "bifrost-falkordb";
const DESCRIPTION: &str = "bifrost: Iceberg lake into a FalkorDB graph";
#[derive(Subcommand, Debug, Clone)]
pub enum FalkordbCommand {
/// 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),
}
/// `unit` / `install`: what to watch, and as which service.
#[derive(Args, Debug, Clone, PartialEq, Eq)]
pub struct ServiceArgs {
#[command(flatten)]
pub watch: WatchArgs,
#[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 lake → graph stream).
#[arg(long, default_value = DEFAULT_NAME)]
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 watch: WatchArgs,
/// Workload name.
#[arg(long, default_value = DEFAULT_NAME)]
pub name: String,
/// The image carrying `/usr/local/bin/bifrost` (and the mapping file).
#[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,
}
fn env(name: &str) -> Option<String> {
std::env::var(name).ok().filter(|v| !v.is_empty())
}
/// Whether a `redis://user:pass@host` URL carries credentials.
fn has_userinfo(url: &str) -> bool {
let rest = url.split_once("://").map_or(url, |(_, r)| r);
rest.split(['/', '?', '#']).next().is_some_and(|authority| authority.contains('@'))
}
/// The watch a service runs: refused flags checked, the cursor its own.
pub fn service_watch(w: &WatchArgs, state: &Path) -> Result<WatchArgs> {
let refuse = |what: &str| Err(BifrostError::Other(format!("bifrost falkordb: {what}")));
if w.once {
return refuse("--once makes no service (it would exit after one round) - use `bifrost watch --once`");
}
if w.dry_run {
return refuse("--dry-run makes no service (it writes no graph) - use `bifrost watch --dry-run`");
}
if w.state.is_some() {
return refuse("--state: an installed service keeps its cursor in its own state dir");
}
if let Some(url) = &w.falkor
&& has_userinfo(url)
{
return refuse(&format!(
"--falkor carries a password, and argv is world-readable - set {FALKOR_URL_ENV} instead (it goes to the secret channel)"
));
}
Ok(WatchArgs { state: Some(state.to_path_buf()), ..w.clone() })
}
/// The ONE spec both packages are rendered from. `exec` is the binary path the
/// target runs; `watch` the already-checked flags (see [`service_watch`]).
/// Secrets are copied from the installer's environment into the secret
/// channel, never into the unit.
pub fn service_spec(name: &str, exec: &str, watch: &WatchArgs, system: bool) -> skidbladnir::service::ServiceSpec {
use skidbladnir::service::{AgentMode, ServiceSpec};
use skidbladnir::systemd::{RestartPolicy, Scope};
let mut spec = ServiceSpec::new(name, DESCRIPTION, exec)
.scope(if system { Scope::System } else { Scope::User })
.restart(RestartPolicy::OnFailure)
.agent(AgentMode::Fat)
.env("RUST_LOG", "info");
if let Some(dir) = watch.state.as_deref().and_then(Path::parent) {
spec = spec.working_dir(dir.to_string_lossy());
}
if system {
spec = spec.user("bifrost");
}
for a in watch.to_args() {
spec = spec.arg(a);
}
for k in SECRET_ENV {
if let Some(v) = env(k) {
spec = spec.secret_env(k, v);
}
}
spec
}
fn scope(system: bool) -> skidbladnir::systemd::Scope {
if system { skidbladnir::systemd::Scope::System } else { skidbladnir::systemd::Scope::User }
}
/// The systemd lifecycle for `life.name`, its spec built from `watch` when given.
fn lifecycle(watch: Option<&WatchArgs>, life: &LifecycleArgs) -> Result<skidbladnir::service::Lifecycle> {
let exec = std::env::current_exe()
.map(|p| p.to_string_lossy().into_owned())
.unwrap_or_else(|_| IMAGE_EXEC.into());
let state_dir = installed_state_dir(&life.name, life.system);
let mut spec = match watch {
Some(w) => {
let cwd = std::env::current_dir()?;
let w = service_watch(&w.absolutized(&cwd), &state_dir.join("cursor.json"))?;
service_spec(&life.name, &exec, &w, life.system)
}
None => skidbladnir::service::ServiceSpec::new(&life.name, DESCRIPTION, &exec).scope(scope(life.system)),
};
// skidbladnir's uninstall erases a declared data root (its law: data dies).
// The cursor is what lets a reinstall resume without re-ingesting, so it is
// declared only when the operator asked for `--purge`.
if life.purge {
let root = skidbladnir::uninstall::DataRoot::from_env("BIFROST_FALKORDB_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 files a pod reads must be in the image: absolute paths only.
fn in_image(w: &WatchArgs) -> Result<()> {
let c = &w.lake.catalog;
let files = [
Some(&w.mapping.mapping),
w.lake.warehouse.as_ref(),
w.lake.s3_ca.as_ref(),
w.tls.tls_cert.as_ref(),
w.tls.tls_key.as_ref(),
w.tls.tls_ca.as_ref(),
c.grpc_ca.as_ref(),
c.grpc_cert.as_ref(),
c.grpc_key.as_ref(),
w.metrics_file.as_ref(),
];
for f in files.into_iter().flatten() {
if !f.is_absolute() {
return Err(BifrostError::Other(format!(
"bifrost falkordb k8s: {} is relative - name it as a path inside the image (skidbladnir mounts no ConfigMap)",
f.display()
)));
}
}
Ok(())
}
/// 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) -> Result<Vec<skidbladnir::service::RenderedFile>> {
use skidbladnir::service::{ServiceRenderer as _, Storage};
in_image(&args.watch)?;
let w = service_watch(&args.watch, &Path::new(K8S_STATE_DIR).join("cursor.json"))?;
let spec = service_spec(&args.name, IMAGE_EXEC, &w, false)
.image(&args.image)
.namespace(&args.namespace)
.replicas(1)
.stateful(true)
.storage(Storage::new(&args.state_size, K8S_STATE_DIR));
Ok(skidbladnir::k8s::KubernetesRenderer.render(&spec).files)
}
/// Dispatch `bifrost falkordb <verb>`.
pub fn dispatch(cmd: FalkordbCommand) -> Result<()> {
match cmd {
FalkordbCommand::Unit(a) => {
let lc = lifecycle(Some(&a.watch), &a.life)?;
print!("{}", crate::service_paths::unit_text(&lc, &a.life.name, a.life.system));
Ok(())
}
FalkordbCommand::Install(a) => {
let lc = lifecycle(Some(&a.watch), &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 falkordb: INSTALLED and started ({})", lc.unit());
Ok(())
}
FalkordbCommand::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 falkordb: unit removed = {}, env file removed = {env}, erased {} path(s)",
r.unit_removed,
r.erased.len()
);
Ok(())
}
FalkordbCommand::Start(l) => lifecycle(None, &l)?.start().map_err(sk),
FalkordbCommand::Stop(l) => lifecycle(None, &l)?.stop().map_err(sk),
FalkordbCommand::Status(l) => {
println!("{}", if lifecycle(None, &l)?.status().map_err(sk)? { "active" } else { "inactive" });
Ok(())
}
FalkordbCommand::K8s(a) => {
let files = k8s_files(&a)?;
std::fs::create_dir_all(&a.out)?;
for f in files {
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: FalkordbCommand,
}
fn parse(s: &str) -> std::result::Result<FalkordbCommand, clap::Error> {
T::try_parse_from(std::iter::once("t").chain(s.split_whitespace())).map(|t| t.cmd)
}
#[derive(Parser)]
struct W {
#[command(flatten)]
w: WatchArgs,
}
const LAKE: &str = "--rest-uri https://polaris:8181 --catalog polaris --rest-warehouse cat \
--s3-endpoint https://minio:9000 --s3-bucket lake --mapping /etc/bifrost/m.yml \
--falkor rediss://falkor:6380 --graph social";
#[test]
fn the_unit_runs_watch_with_its_own_cursor_and_parses_back() {
let Ok(FalkordbCommand::Install(a)) = parse(&format!("install {LAKE} --system")) else { panic!() };
assert_eq!(a.life.name, DEFAULT_NAME);
let state = installed_state_dir(&a.life.name, true).join("cursor.json");
let w = service_watch(&a.watch, &state).unwrap();
let spec = service_spec(&a.life.name, "/usr/bin/bifrost", &w, true);
let unit = spec.render_systemd_unit();
let args = w.to_args();
assert_eq!(args[0], "watch");
for x in &args {
assert!(unit.contains(x.as_str()), "the unit carries {x}");
}
assert!(unit.contains("/var/lib/bifrost-falkordb/cursor.json"));
let back = W::try_parse_from(std::iter::once("t".to_string()).chain(args.into_iter().skip(1))).unwrap().w;
assert_eq!(back, w, "what the service runs parses back to what was asked for");
}
#[test]
fn once_dry_run_state_and_a_password_in_the_url_are_refused() {
for extra in ["--once", "--dry-run", "--state /tmp/s.json"] {
let Ok(FalkordbCommand::Unit(a)) = parse(&format!("unit {LAKE} {extra}")) else { panic!() };
assert!(service_watch(&a.watch, Path::new("/s/c.json")).is_err(), "{extra} refused");
}
let lake = LAKE.replace("rediss://falkor:6380", "redis://:pw@falkor:6379");
let Ok(FalkordbCommand::Unit(a)) = parse(&format!("unit {lake}")) else { panic!() };
let e = service_watch(&a.watch, Path::new("/s/c.json")).unwrap_err().to_string();
assert!(e.contains(FALKOR_URL_ENV) && !e.contains(":pw@"), "{e}");
}
#[test]
fn userinfo_is_found_only_in_the_authority() {
assert!(has_userinfo("redis://:pw@h:6379"));
assert!(has_userinfo("rediss://u:p@h"));
assert!(!has_userinfo("redis://h:6379"));
assert!(!has_userinfo("redis://h:6379/0?x=a@b"));
}
#[test]
fn systemd_and_k8s_run_the_same_command_and_no_secret_is_on_it() {
// SAFETY of the test: the env vars are set only in this process.
unsafe {
std::env::set_var(CREDENTIAL_ENV, "id:hunter2-not-on-argv");
std::env::set_var(FALKOR_URL_ENV, "redis://:hunter3@f:6379");
}
let Ok(FalkordbCommand::K8s(k)) = parse(&format!("k8s {LAKE} --image reg/bifrost:1 --out /tmp/x")) else {
panic!()
};
let state = Path::new(K8S_STATE_DIR).join("cursor.json");
let w = service_watch(&k.watch, &state).unwrap();
let unit = service_spec(&k.name, IMAGE_EXEC, &w, false).render_systemd_unit();
let files = k8s_files(&k).unwrap();
let workload = files.iter().find(|f| f.contents.contains("kind: StatefulSet")).expect("a StatefulSet");
for a in w.to_args() {
assert!(workload.contents.contains(&a), "k8s args carry {a}");
assert!(unit.contains(&a), "the unit carries {a}");
}
for secret in ["hunter2", "hunter3"] {
assert!(!unit.contains(secret), "a secret never lands in the 0644 unit");
assert!(!workload.contents.contains(secret), "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 k8s_refuses_a_file_that_cannot_be_in_the_image() {
let lake = LAKE.replace("/etc/bifrost/m.yml", "m.yml");
let Ok(FalkordbCommand::K8s(k)) = parse(&format!("k8s {lake} --image i --out /tmp/x")) else { panic!() };
assert!(k8s_files(&k).unwrap_err().to_string().contains("inside the image"));
}
#[test]
fn lifecycle_verbs_take_only_a_name() {
let Ok(FalkordbCommand::Uninstall(l)) = parse("uninstall --name g1 --system --purge") else { panic!() };
assert_eq!(l, LifecycleArgs { name: "g1".into(), system: true, purge: true });
assert!(parse("status --mapping m.yml").is_err());
}
}