crates/knut-popviz/src/bin/knut-popload.rs
//! `knut-popload` — the synthetic Swedish population graph into FalkorDB, **headless**.
//!
//! The same load the egui `knut-popviz` app drives ([`knut_popviz::run_load`]), for a
//! server: no display, one progress line per phase change (or per JSON object with
//! `--json`), and an exit code — `0` when the load is done, `1` when it failed. It is
//! what a "Simulate" press runs on the FalkorDB host.
//!
//! ```text
//! knut-popload --url rediss://localhost:6379 --graph simulated --scale 0.01 --drop \
//! --tls-cert sim.crt --tls-key sim.key --tls-ca ca-chain.crt [--json]
//! ```
//!
//! * `--url` (default `redis://127.0.0.1:6379`), `--graph` (default `pop`)
//! * `--scale` — fraction of the 11-million-person model (default `0.001`)
//! * `--drop` — delete the graph first (the user needs `graph.delete` on it)
//! * `--ingest auto|bulk|unwind` — `auto`: `GRAPH.BULK` into an empty graph
//! * `--tls-cert` + `--tls-key` [`--tls-ca`] — mutual TLS; the CN is the ACL user
//! * `--target iceberg --warehouse DIR` — the population as Iceberg tables instead (one
//! per label, a `rel_<REL>` per relationship, every row `round` 0) in a local skade
//! warehouse; `--drop` wipes it first. Feature `iceberg`.
//! * `--rest-uri URL` instead of `--warehouse DIR` (both `--target iceberg` and
//! `--permutate`) — the same tables behind an Iceberg REST catalog: MinIO AIStor with
//! `--rest-warehouse WH --sigv4 s3tables --s3-endpoint URL`, tables in `--rest-namespace`
//! (default `main`). The keys come from `MINIO_ACCESS_KEY` / `MINIO_SECRET_KEY` only.
//! Feature `iceberg-rest`.
//! * `--permutate F [--round N] --warehouse DIR` — change a fraction `F` of the persons in
//! that warehouse as round `N` (moves, benefit flips, closed loans, new vehicles) — by
//! default the next round, read off the lake itself — written as Spark copy-on-write
//! commits; see `knut_popviz::permutate`. Feature `iceberg`.
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use knut_popviz::{IngestMode, LoadParams, Phase, PopvizState, Target, run_load};
fn main() {
let args: Vec<String> = std::env::args().collect();
if args.iter().any(|a| a == "--help" || a == "-h") {
println!("{}", HELP.trim());
return;
}
let opt = |k: &str, d: &str| args.windows(2).find(|w| w[0] == k).map(|w| w[1].clone()).unwrap_or_else(|| d.into());
let json = args.iter().any(|a| a == "--json");
if args.iter().any(|a| a == "--permutate") {
permutate(&opt, json);
return;
}
let scale: f64 = match opt("--scale", "0.001").parse() {
Ok(s) if s > 0.0 && s <= 1.0 => s,
_ => fail_usage("--scale is a fraction in (0, 1]"),
};
let mut params = LoadParams {
url: opt("--url", "redis://127.0.0.1:6379"),
graph: opt("--graph", "pop"),
scale,
drop: args.iter().any(|a| a == "--drop"),
target: match opt("--target", "falkor").as_str() {
"iceberg" => match lake(&opt) {
knut_popviz::lake::Lake::Rest(r) => Target::IcebergRest(r),
knut_popviz::lake::Lake::Dir(d) => Target::Iceberg { warehouse: d.display().to_string() },
},
"falkor" => Target::Falkor,
t => fail_usage(&format!("--target {t}: one of falkor, iceberg")),
},
ingest: IngestMode::from_str_lenient(&opt("--ingest", "auto")),
tls: None,
};
let (cert, key, ca) = (opt("--tls-cert", ""), opt("--tls-key", ""), opt("--tls-ca", ""));
if !cert.is_empty() || !key.is_empty() {
if cert.is_empty() || key.is_empty() {
fail_usage("--tls-cert and --tls-key go together");
}
let ca = (!ca.is_empty()).then(|| std::path::PathBuf::from(&ca));
match graphar_falkordb::FalkorTls::from_files(cert.as_ref(), key.as_ref(), ca.as_deref()) {
Ok(id) => params.tls = Some(id),
Err(e) => fail_usage(&e.to_string()),
}
}
let (graph, url) = match ¶ms.target {
Target::Iceberg { warehouse } => (format!("iceberg:{warehouse}"), "-".to_string()),
Target::IcebergRest(r) => (knut_popviz::lake::Lake::Rest(r.clone()).describe(), "-".to_string()),
Target::Falkor => (params.graph.clone(), params.url.clone()),
};
let state = Arc::new(Mutex::new(PopvizState::new(scale)));
let people = state.lock().unwrap().labels.first().map(|l| l.total).unwrap_or(0);
report(json, &serde_json::json!({ "event": "start", "graph": graph, "url": url, "scale": scale, "persons": people }));
// `wake` is called on every progress step; print on a phase change, and at most
// every 5 s within one, so a journal holds the run without a line per batch.
let last = Arc::new(Mutex::new((Phase::Idle, Instant::now())));
let (st, lw) = (state.clone(), last.clone());
let t0 = Instant::now();
run_load(params, state.clone(), move || {
let s = st.lock().unwrap();
let mut l = lw.lock().unwrap();
if s.phase != l.0 || l.1.elapsed() >= Duration::from_secs(5) {
*l = (s.phase, Instant::now());
let (done, total): (usize, usize) = s.labels.iter().map(|x| (x.loaded, x.total)).fold((0, 0), |a, b| (a.0 + b.0, a.1 + b.1));
let (edone, etotal): (usize, usize) = s.rels.iter().map(|x| (x.loaded, x.total)).fold((0, 0), |a, b| (a.0 + b.0, a.1 + b.1));
report(json, &serde_json::json!({
"event": "progress", "phase": s.phase.label(),
"nodes": done, "nodes_total": total, "edges": edone, "edges_total": etotal,
}));
}
});
let s = state.lock().unwrap();
let nodes: usize = s.labels.iter().map(|l| l.loaded).sum();
let edges: usize = s.rels.iter().map(|r| r.loaded).sum();
let secs = t0.elapsed().as_secs_f64();
match s.phase {
Phase::Done => {
report(json, &serde_json::json!({ "event": "done", "graph": graph, "nodes": nodes, "edges": edges, "seconds": secs }));
}
_ => {
let why = s.error.clone().unwrap_or_else(|| format!("ended in phase `{}`", s.phase.label()));
report(json, &serde_json::json!({ "event": "failed", "graph": graph, "error": why, "seconds": secs }));
std::process::exit(1);
}
}
}
/// Where `--target iceberg` and `--permutate` write: `--rest-uri` (a catalog), else
/// `--warehouse DIR`. Names the missing flag rather than guessing.
fn lake(opt: &dyn Fn(&str, &str) -> String) -> knut_popviz::lake::Lake {
let uri = opt("--rest-uri", "");
if uri.is_empty() {
let dir = opt("--warehouse", "");
if dir.is_empty() {
fail_usage("name the lake: --warehouse DIR, or --rest-uri URL");
}
return knut_popviz::lake::Lake::Dir(dir.into());
}
let some = |k: &str| Some(opt(k, "")).filter(|v| !v.is_empty());
let s3_endpoint = some("--s3-endpoint").unwrap_or_else(|| fail_usage("--rest-uri needs --s3-endpoint URL"));
knut_popviz::lake::Lake::Rest(knut_popviz::lake::RestLake {
uri,
warehouse: some("--rest-warehouse"),
namespace: opt("--rest-namespace", "main"),
s3_endpoint,
s3_region: opt("--s3-region", "us-east-1"),
s3_bucket: some("--s3-bucket"),
s3_ca: some("--s3-ca").map(Into::into),
sigv4: some("--sigv4"),
})
}
/// `--permutate F --round N`, on `--warehouse DIR` or `--rest-uri URL`.
fn permutate(opt: &dyn Fn(&str, &str) -> String, json: bool) {
let fraction: f64 = opt("--permutate", "").parse().unwrap_or_else(|_| fail_usage("--permutate is a fraction in [0, 1]"));
// No --round: the next one (0 asks permutate_lake for it).
let round: i64 = opt("--round", "0").parse().unwrap_or_else(|_| fail_usage("--round N is an integer > 0"));
if round < 0 {
fail_usage("--round N is an integer > 0 (leave it out for the next round)");
}
let lake = lake(opt);
#[cfg(feature = "iceberg")]
{
let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("tokio runtime");
match rt.block_on(knut_popviz::permutate::permutate_lake(&lake, fraction, round)) {
Ok(r) => {
let v = r.json();
if json {
println!("{v}");
} else {
println!(
"knut-popload: round {} - {} persons changed: {} moved, {} benefit flips, {} loans closed, {} vehicles bought, in {:.1} s",
r.round, r.persons_changed, r.addresses_moved, r.bistand_flipped, r.loans_deleted, r.vehicles_added, r.seconds
);
}
}
Err(e) => {
report(json, &serde_json::json!({ "event": "failed", "graph": lake.describe(), "error": e, "seconds": 0.0 }));
std::process::exit(1);
}
}
}
#[cfg(not(feature = "iceberg"))]
{
let _ = (fraction, round, json, lake);
fail_usage("--permutate needs a build with --features iceberg");
}
}
fn report(json: bool, v: &serde_json::Value) {
if json {
println!("{v}");
return;
}
let get = |k: &str| v.get(k).map(|x| x.to_string().trim_matches('"').to_string()).unwrap_or_default();
match get("event").as_str() {
"start" => println!("knut-popload: {} persons (scale {}) into graph `{}` at {}", get("persons"), get("scale"), get("graph"), get("url")),
"progress" => println!(
" {:<17} nodes {}/{} edges {}/{}",
get("phase"), get("nodes"), get("nodes_total"), get("edges"), get("edges_total")
),
"done" => println!("knut-popload: done - {} nodes, {} edges into `{}` in {:.1} s", get("nodes"), get("edges"), get("graph"), v["seconds"].as_f64().unwrap_or(0.0)),
_ => eprintln!("knut-popload: FAILED - {} (graph `{}`)", get("error"), get("graph")),
}
}
fn fail_usage(why: &str) -> ! {
eprintln!("knut-popload: {why}\n\n{}", HELP.trim());
std::process::exit(2);
}
const HELP: &str = "
knut-popload - the synthetic Swedish population graph into FalkorDB, headless
--url URL redis:// or rediss:// (default redis://127.0.0.1:6379)
--graph NAME default pop
--scale F fraction of the 11-million-person model, (0,1] (default 0.001)
--drop delete the graph first
--ingest MODE auto|bulk|unwind (default auto: GRAPH.BULK into an empty graph)
--tls-cert PEM --tls-key PEM [--tls-ca PEM] mutual TLS; the certificate CN is the user
--target T falkor (default) or iceberg
--warehouse DIR the skade warehouse of --target iceberg and --permutate
--rest-uri URL ...or an Iceberg REST catalog instead (AIStor: https://minio:9000/_iceberg)
--rest-warehouse W the catalog's warehouse (AIStor: the warehouse bucket)
--rest-namespace N the namespace of the tables (default main)
--sigv4 SERVICE sign catalog calls with SigV4 (AIStor: s3tables)
--s3-endpoint URL the data files' S3 endpoint (required with --rest-uri)
--s3-region R default us-east-1
--s3-bucket B default: --rest-warehouse
--s3-ca PEM a private CA for the https:// catalog and S3 endpoint
keys: MINIO_ACCESS_KEY / MINIO_SECRET_KEY in the environment only
--permutate F change a fraction F of the persons in --warehouse as --round N
--round N the round a permutate stamps on what it changed (> 0; default: the
next one, one above the highest round in the lake)
--json one JSON object per line
";