src/bin/korp-collectors.rs
//! `korp-collectors` — **the thin collector runner**.
//!
//! # Why this binary exists
//!
//! The only way to run a collector used to be `korp collector run <name>`: a **278 MB**
//! eframe GUI binary out of a **965-crate** graph, which is also what the Kubernetes
//! render put in every pod's `command`. The *code* has been free of korp for a long time
//! — this crate depends on neither korp nor facett — but the *running* of it was not.
//! A foreign project could write a collector against [`korp_collectors::daemon::Collector`]
//! and then had nothing to start it with.
//!
//! This is that thing. It holds the registry, one seam set, and a bounded shutdown.
//! It is what a [`SystemdRuntime`](korp_collectors::runtime_systemd::SystemdRuntime) unit
//! execs, and what a pod `command` should be.
//!
//! # The two declarations it reads
//!
//! Both come from the substrate, neither is invented here:
//!
//! * **config** — `$COLLECTOR_CONFIG_JSON` (a systemd `Environment=`), or `--config
//! <file>` (a Kubernetes ConfigMap mount). Absent ⇒ `{}`, said out loud.
//! * **data path** — `$DATA_DIR` (what Skidbladnir's layout declares and what its
//! uninstall erases), or `--data-dir`. Absent ⇒ refuses, because a collector that
//! picks its own data directory is a collector whose uninstall cannot find it.
//!
//! # Bounded shutdown
//!
//! `SIGTERM`/`SIGINT` trip the collector's cooperative
//! [`Shutdown`](korp_collectors::daemon::Shutdown) and then this process **waits for
//! `run()` to return** before exiting. It does not detach and it does not exit 0 on a
//! timeout: an unbounded stop is how znippy-zoomies `49c7bc6` got a `SIGTERM` that
//! logged *"stop latency 1–15 ms"* and then died `status=11/SEGV` three seconds later,
//! with a background sorter still running as the runtime was torn down. A stop that has
//! not finished is reported as a stop that has not finished (exit 75), never as success.
use std::sync::Arc;
use std::time::{Duration, Instant};
/// How long a stop may take before this process says so and exits non-zero.
const STOP_DEADLINE: Duration = Duration::from_secs(120);
/// EX_TEMPFAIL — the stop ran out of time. Distinct from a run failure (1) and from a
/// clean stop (0), so a supervisor can tell "it would not let go" from "it broke".
const EXIT_STOP_TIMED_OUT: i32 = 75;
fn main() {
let args: Vec<String> = std::env::args().skip(1).collect();
// --version / -V — print the build stamp and EXIT before any daemon boot.
if korp_collectors::handle_version_flag(&args) {
std::process::exit(0);
}
match args.first().map(String::as_str) {
Some("--help") | Some("-h") | None => usage(),
Some("list") => std::process::exit(cmd_list()),
Some("run") => std::process::exit(cmd_run(&args[1..])),
Some("serve") => std::process::exit(cmd_serve(&args[1..])),
Some("status") => std::process::exit(cmd_status(&args[1..])),
Some("install") => std::process::exit(cmd_install(&args[1..], false)),
Some("uninstall") => std::process::exit(cmd_install(&args[1..], true)),
Some(other) => {
eprintln!("korp-collectors: unknown argument `{other}` (try --help)");
std::process::exit(2);
}
}
}
fn usage() {
println!("{}", korp_collectors::version_string());
println!();
println!("The thin collector runner — the registry, the seams, and a bounded stop.");
println!("Its collectors are korp-free: this crate depends on neither korp nor facett.");
println!();
println!("USAGE:");
println!(" korp-collectors install --config F [--user U] [--scope user|system] [--unit N]");
println!(" install the SERVICE through Skidbladnir");
println!(
" korp-collectors uninstall --config F [--user U] [--scope user|system] [--unit N]"
);
println!(
" stop, disable, remove the unit AND erase its data"
);
println!(" korp-collectors serve [--config F] [--data-dir D]");
println!(
" the SERVICE: nurse every declared collector"
);
println!(" korp-collectors status [--data-dir D] [--json]");
println!(
" read the live register — works from ANOTHER process"
);
println!(" korp-collectors run <name> [--once] [--config F] [--data-dir D]");
println!(" one collector, in the foreground");
println!(" korp-collectors list every registered collector");
println!(" korp-collectors --version | -V <name> v<version> (<git-hash>)");
println!();
println!("ENVIRONMENT (what a substrate declares; nothing here is invented):");
println!(" COLLECTOR_CONFIG_JSON ONE collector's config JSON (systemd Environment=)");
println!(" COLLECTOR_NURSERY_JSON the whole roster for `serve` (systemd Environment=)");
println!(" DATA_DIR the unit's data path (Skidbladnir layout)");
if cfg!(not(feature = "daemon")) {
println!();
println!("NOTE: this build has no `daemon` feature — `run`/`list` are not compiled in.");
}
if cfg!(not(feature = "service")) {
println!();
println!("NOTE: this build has no `service` feature — `serve` is not compiled in.");
}
if cfg!(not(feature = "systemd")) {
println!();
println!(
"NOTE: this build has no `systemd` feature — `install`/`uninstall` are not compiled in."
);
}
}
// ─────────────────────────────────────────────────────────────────────────────
// serve — the service
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(not(feature = "service"))]
fn cmd_serve(_args: &[String]) -> i32 {
eprintln!("korp-collectors: `serve` needs --features service (this build has none)");
2
}
/// **Nurse every declared collector.** The roster, the data path and what the service needs
/// from the host all arrive as declarations; this verb wires them to the nursery and gets
/// out of the way.
///
/// It does not decide what to do when the work dies — [`Nursery::report`] hands that to
/// `skidbladnir::readiness::report_work_status`, which is the one place in the house that
/// decides, so a collector service and every other managed service behave the same way.
#[cfg(feature = "service")]
fn cmd_serve(args: &[String]) -> i32 {
use korp_collectors::daemon::LogSink;
use korp_collectors::nursery::{Nursery, NurseryConfig};
let mut config_file: Option<String> = None;
let mut data_dir: Option<String> = None;
let mut it = args.iter();
while let Some(a) = it.next() {
match a.as_str() {
"--config" => config_file = it.next().cloned(),
"--data-dir" => data_dir = it.next().cloned(),
other => {
eprintln!("korp-collectors serve: unknown argument `{other}`");
return 2;
}
}
}
if let Some(d) = &data_dir {
// The flag wins over the environment, and it wins by being the environment the
// config reader looks at — so there is one precedence rule, in one place.
// SAFETY: single-threaded; no collector has been started yet.
unsafe { std::env::set_var("DATA_DIR", d) };
}
let cfg = match NurseryConfig::from_env(config_file.as_deref()) {
Ok(c) => c,
Err(e) => {
eprintln!("korp-collectors serve: {e:#}");
return 2;
}
};
eprintln!("{}", korp_collectors::version_string());
let log = LogSink::new(move |m| eprintln!("[{}] {m}", chrono_ish()));
let nursery = match Nursery::new(cfg, log) {
Ok(n) => n,
Err(e) => {
eprintln!("korp-collectors serve: {e:#}");
return 2;
}
};
eprintln!(
"korp-collectors serve: register {}",
nursery.register().path().display()
);
// SIGTERM/SIGINT are handled inside `Nursery::run`, on the runtime that owns the
// collectors — not here. A handler installed on this thread would need its own runtime
// or its own signal implementation, and a second way to stop the same collectors is how
// one of the two ends up not being wired.
let outcomes = match nursery.run() {
Ok(o) => o,
Err(e) => {
eprintln!("korp-collectors serve: {e:#}");
return 1;
}
};
for (name, o) in &outcomes {
eprintln!("collector {name}: {o:?}");
}
// Skidbladnir decides. If every collector's work is dead it exits non-zero from inside
// this call, which is the signal `Restart=on-failure` acts on — the alternative is a
// process that keeps listening while `systemctl is-active` says `active`.
nursery.report(&outcomes);
0
}
// ─────────────────────────────────────────────────────────────────────────────
// status — read the register, from any process
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(not(feature = "daemon"))]
fn cmd_status(_args: &[String]) -> i32 {
eprintln!("korp-collectors: `status` needs --features daemon (this build has none)");
2
}
/// **Read the live state of every collector some OTHER process is nursing.**
///
/// This verb is the evidence that the register works: it holds no supervisor, starts no
/// collector, and opens the register read-only with no lock. Whatever it prints, it learned
/// from a file another process wrote.
///
/// Exit codes are the verdict, not decoration. `0` every row is live; `3` at least one row
/// is a fossil from a dead writer; `4` there is no register at all. A status verb that
/// exits 0 over a dead service is the kind of green that can never be red.
#[cfg(feature = "daemon")]
fn cmd_status(args: &[String]) -> i32 {
use korp_collectors::register::RegisterReader;
let mut data_dir: Option<String> = None;
let mut as_json = false;
let mut it = args.iter();
while let Some(a) = it.next() {
match a.as_str() {
"--data-dir" => data_dir = it.next().cloned(),
"--json" => as_json = true,
other => {
eprintln!("korp-collectors status: unknown argument `{other}`");
return 2;
}
}
}
let Some(dir) = data_dir.or_else(|| std::env::var("DATA_DIR").ok()) else {
eprintln!(
"korp-collectors status: no data path declared (--data-dir or $DATA_DIR). The \
register lives in the service's declared data directory; guessing one would read \
a different service's answer"
);
return 2;
};
let dir = std::path::PathBuf::from(dir);
let reader = match RegisterReader::open(&dir) {
Ok(Some(r)) => r,
Ok(None) => {
eprintln!(
"korp-collectors status: no register in {} — no collector service has ever run \
there",
dir.display()
);
return 4;
}
Err(e) => {
eprintln!("korp-collectors status: {e:#}");
return 1;
}
};
let views = match reader.all() {
Ok(v) => v,
Err(e) => {
eprintln!("korp-collectors status: {e:#}");
return 1;
}
};
if as_json {
let rows: Vec<serde_json::Value> = views
.iter()
.map(|v| {
serde_json::json!({
"collector": v.row.collector,
"enabled": v.row.enabled,
"phase": v.row.phase,
"phase_label": v.row.phase.label(),
"healthy": v.row.healthy,
"facet": v.row.facet,
"freshness": v.freshness.kind(),
"freshness_why": v.freshness.why(),
"writer_pid": v.row.writer.pid,
"substrate": v.row.writer.substrate,
"unit": v.row.writer.unit,
"updated_unix": v.row.updated_unix,
"seq": v.row.seq,
})
})
.collect();
println!(
"{}",
serde_json::to_string_pretty(&serde_json::json!({
"register": reader.path().display().to_string(),
"reader_pid": std::process::id(),
"collectors": rows,
}))
.unwrap_or_else(|_| "{}".into())
);
} else {
println!(
"register {} (read by pid {}, no lock held)",
reader.path().display(),
std::process::id()
);
if views.is_empty() {
println!("(no collectors published yet)");
}
for v in &views {
println!("{}", v.row.line(&v.freshness));
}
}
if views.iter().any(|v| !v.freshness.is_live()) {
return 3;
}
0
}
#[cfg(not(feature = "daemon"))]
fn cmd_list() -> i32 {
eprintln!("korp-collectors: `list` needs --features daemon (this build has none)");
2
}
#[cfg(not(feature = "daemon"))]
fn cmd_run(_args: &[String]) -> i32 {
eprintln!("korp-collectors: `run` needs --features daemon (this build has none)");
2
}
/// Every registered collector, one per line — the same registry `run` looks up in, so
/// this cannot list a name `run` would then reject.
#[cfg(feature = "daemon")]
fn cmd_list() -> i32 {
let reg = korp_collectors::daemon::daemon_registry();
for c in ® {
let size = c
.durable_bytes()
.map(|b| korp_collectors::runtime::format_binary_quantity(b))
.unwrap_or_else(|| "-".into());
println!("{:<20} {:>8} {}", c.name(), size, c.title());
}
if reg.is_empty() {
eprintln!("korp-collectors: the registry is EMPTY in this build — check the features");
return 1;
}
0
}
/// Run one registered collector to completion, or until a bounded stop.
#[cfg(feature = "daemon")]
fn cmd_run(args: &[String]) -> i32 {
use korp_collectors::daemon::{CollectorCtx, LogSink, PhaseSink};
let mut name = None;
let mut once = false;
let mut config_file: Option<String> = None;
let mut data_dir: Option<String> = None;
let mut it = args.iter();
while let Some(a) = it.next() {
match a.as_str() {
"--once" => once = true,
"--config" => config_file = it.next().cloned(),
"--data-dir" => data_dir = it.next().cloned(),
other if other.starts_with('-') => {
eprintln!("korp-collectors run: unknown flag `{other}`");
return 2;
}
other => name = Some(other.to_string()),
}
}
let Some(name) = name.or_else(|| std::env::var("COLLECTOR_NAME").ok()) else {
eprintln!("korp-collectors run: no collector named (argv or $COLLECTOR_NAME)");
return 2;
};
// ── config: declared by the substrate, never guessed ──────────────────────────
let config = match load_config(config_file.as_deref()) {
Ok(v) => v,
Err(e) => {
eprintln!("korp-collectors run {name}: {e:#}");
return 2;
}
};
// ── data path: declared by the substrate, never guessed ───────────────────────
let Some(dir) = data_dir.or_else(|| std::env::var("DATA_DIR").ok()) else {
eprintln!(
"korp-collectors run {name}: no data path declared (--data-dir or $DATA_DIR). \
Refusing to pick one: a data directory this process invents is a directory \
the installer's uninstall will never find."
);
return 2;
};
let disk_dir = std::path::PathBuf::from(&dir).join(&name);
if let Err(e) = std::fs::create_dir_all(&disk_dir) {
eprintln!(
"korp-collectors run {name}: cannot create {}: {e}",
disk_dir.display()
);
return 1;
}
let Some(daemon) = korp_collectors::daemon::daemon_with_config(&name, true, config.clone())
else {
eprintln!(
"korp-collectors run: no collector named `{name}` in this build's registry \
(`korp-collectors list` shows what is compiled in)"
);
return 2;
};
eprintln!("{}", korp_collectors::version_string());
eprintln!("collector {name}: data {}", disk_dir.display());
// Seams. The SINK is deliberately in-memory here: where collected rows are written
// is a separate decision that has not been made, and a runner that silently picks a
// warehouse would be making it. Everything else is real.
let mut ctx = CollectorCtx::in_memory(&name, disk_dir, config);
ctx.fetch = korp_collectors::daemon::Fetcher::http();
ctx.log = LogSink::new(move |m| eprintln!("[{}] {m}", chrono_ish()));
let phase_name = name.clone();
ctx.phase = PhaseSink::new(move |p| eprintln!("[phase] {phase_name}: {}", p.label()));
let shutdown = ctx.shutdown.clone();
let rt = match tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
{
Ok(rt) => rt,
Err(e) => {
eprintln!("korp-collectors run {name}: tokio runtime: {e}");
return 1;
}
};
// ── the bounded stop ──────────────────────────────────────────────────────────
// SIGTERM/SIGINT trip the cooperative shutdown; the collector's own `run` loop
// observes it and returns. We then WAIT for that return.
#[cfg(unix)]
{
let sd = shutdown.clone();
rt.spawn(async move {
use tokio::signal::unix::{signal, SignalKind};
let mut term = match signal(SignalKind::terminate()) {
Ok(s) => s,
Err(e) => {
eprintln!("korp-collectors: cannot listen for SIGTERM: {e}");
return;
}
};
let mut int = match signal(SignalKind::interrupt()) {
Ok(s) => s,
Err(_) => return,
};
tokio::select! {
_ = term.recv() => eprintln!("korp-collectors: SIGTERM — stopping, waiting for the collector to finish"),
_ = int.recv() => eprintln!("korp-collectors: SIGINT — stopping, waiting for the collector to finish"),
}
sd.stop();
});
}
if once {
// --once: hold until Online, then trip the same cooperative stop (never a
// process::exit from under a running collector).
let sd = shutdown.clone();
let watch = Arc::clone(&daemon);
rt.spawn(async move {
loop {
if watch.phase().is_online() {
eprintln!("korp-collectors: online — --once stopping");
sd.stop();
return;
}
tokio::time::sleep(Duration::from_millis(250)).await;
}
});
}
let started = Instant::now();
rt.block_on(daemon.clone().run(ctx));
let ran = started.elapsed();
// `run` has returned, so the collector's own work is done. What is NOT proven by
// that alone is that everything it spawned is joined — which is why the runtime is
// shut down with a deadline rather than dropped (a dropped multi-thread runtime
// detaches its workers, and a detached thread that happens to finish looks exactly
// like a joined one).
let stop_started = Instant::now();
rt.shutdown_timeout(STOP_DEADLINE);
let stopping = stop_started.elapsed();
if stopping >= STOP_DEADLINE {
eprintln!(
"korp-collectors {name}: ran {ran:?}, then the runtime did NOT drain within \
{STOP_DEADLINE:?} — something it started is still going. Reporting a timed-out \
stop, not a clean one."
);
return EXIT_STOP_TIMED_OUT;
}
eprintln!("korp-collectors {name}: ran {ran:?}, stopped cleanly in {stopping:?}");
0
}
/// The unit config: `--config <file>` wins, else `$COLLECTOR_CONFIG_JSON`, else `{}`.
#[cfg(feature = "daemon")]
fn load_config(file: Option<&str>) -> anyhow::Result<serde_json::Value> {
use anyhow::Context;
if let Some(f) = file {
let s = std::fs::read_to_string(f).with_context(|| format!("read --config {f}"))?;
return serde_json::from_str(&s).with_context(|| format!("parse --config {f} as JSON"));
}
match std::env::var("COLLECTOR_CONFIG_JSON") {
Ok(s) if !s.trim().is_empty() => {
serde_json::from_str(&s).context("parse $COLLECTOR_CONFIG_JSON as JSON")
}
_ => {
eprintln!("korp-collectors: no config declared — running with `{{}}`");
Ok(serde_json::json!({}))
}
}
}
/// Seconds since process start — a log prefix that needs no time crate.
#[cfg(feature = "daemon")]
fn chrono_ish() -> String {
use std::sync::OnceLock;
static T0: OnceLock<Instant> = OnceLock::new();
format!(
"{:>7.1}s",
T0.get_or_init(Instant::now).elapsed().as_secs_f32()
)
}
// ─────────────────────────────────────────────────────────────────────────────
// install / uninstall — through Skidbladnir, which owns service lifecycle
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(not(feature = "systemd"))]
fn cmd_install(_args: &[String], _remove: bool) -> i32 {
eprintln!(
"korp-collectors: `install`/`uninstall` need --features systemd (this build has none)"
);
2
}
/// **Install (or uninstall) this service as a systemd unit.**
///
/// The whole verb is a translation: it reads the roster the operator declared, builds a
/// `UnitSpec::nursery` from it, and hands that to
/// [`SystemdRuntime`](korp_collectors::runtime_systemd::SystemdRuntime), which routes every
/// byte through Skidbladnir. Nothing here renders unit text, creates an account, chooses a
/// data path or sweeps one — a lifecycle verb that owned any of those would be a second
/// answer to a question Skidbladnir already answers, and the cost of that is two verbs
/// rendering two different services for the same product.
///
/// `--scope system` needs root, because writing `/etc/systemd/system` and creating a
/// service account do. `--scope user` needs none, and is the honest default for a
/// developer trying the service out.
#[cfg(feature = "systemd")]
fn cmd_install(args: &[String], remove: bool) -> i32 {
use korp_collectors::runtime::{Runtime, UnitSpec};
use korp_collectors::runtime_systemd::SystemdRuntime;
use skidbladnir::systemd::Scope;
let verb = if remove { "uninstall" } else { "install" };
let mut config_file: Option<String> = None;
let mut user: Option<String> = None;
let mut scope = Scope::User;
let mut unit = "collectors".to_string();
let mut program: Option<String> = None;
let mut it = args.iter();
while let Some(a) = it.next() {
match a.as_str() {
"--config" => config_file = it.next().cloned(),
"--user" => user = it.next().cloned(),
"--unit" => {
if let Some(u) = it.next() {
unit = u.clone();
}
}
"--program" => program = it.next().cloned(),
"--scope" => match it.next().map(String::as_str) {
Some("system") => scope = Scope::System,
Some("user") => scope = Scope::User,
other => {
eprintln!(
"korp-collectors {verb}: --scope takes `user` or `system`, got {other:?}"
);
return 2;
}
},
other => {
eprintln!("korp-collectors {verb}: unknown argument `{other}`");
return 2;
}
}
}
// The roster. `install` needs it because the unit carries it; `uninstall` needs it
// because the spec it removes must be the spec that was installed — an uninstall
// built from a guess erases a unit nobody installed and reports success.
let roster = match korp_collectors::nursery::NurseryConfig::from_env(config_file.as_deref()) {
Ok(c) => match serde_json::to_value(&c) {
Ok(v) => v,
Err(e) => {
eprintln!("korp-collectors {verb}: {e}");
return 2;
}
},
Err(e) => {
eprintln!("korp-collectors {verb}: {e:#}");
return 2;
}
};
// The program installed is THIS binary unless one is named. A lifecycle verb installs
// an already-built artifact; it never produces one.
let program = match program
.map(std::path::PathBuf::from)
.or_else(|| std::env::current_exe().ok())
{
Some(p) => p,
None => {
eprintln!(
"korp-collectors {verb}: cannot determine the program to install (pass --program)"
);
return 2;
}
};
let user = user.unwrap_or_else(|| match scope {
// A system unit runs as a service account, and a default that is the CALLER would
// install a system service running as a login user — the shape that put
// 1 605 675 tiles in the wrong home.
Scope::System => "_korpcoll".to_string(),
Scope::User => std::env::var("USER").unwrap_or_else(|_| "unknown".into()),
});
let rt = SystemdRuntime::new(&user, &program, scope);
let spec = UnitSpec::nursery(&unit, roster);
// A system-scope install of a NEW service account has to create it before anything can
// say where its data lives — the layout reads that account's passwd home. `start` does
// this too; doing it here as well is what lets the lines below print the real paths
// instead of failing to resolve them.
if !remove {
if let Err(e) = rt.provision_account() {
eprintln!("korp-collectors {verb}: {e:#}");
return 2;
}
}
let layout = match rt.layout() {
Ok(l) => l,
Err(e) => {
eprintln!("korp-collectors {verb}: {e:#}");
return 2;
}
};
println!("{}", korp_collectors::version_string());
println!(
"korp-collectors {verb}: unit {} scope {scope:?} user {user}",
rt.unit_file(&unit)
);
println!(" program path {}", layout.software_dir.display());
println!(
" data path {}",
layout.product_data(&rt.service_name(&unit)).display()
);
if remove {
match rt.uninstall(&spec) {
Ok(o) => {
println!("korp-collectors uninstall: removed {:?}", o.removed);
0
}
Err(e) => {
eprintln!("korp-collectors uninstall: {e:#}");
1
}
}
} else {
match rt.start(&spec) {
Ok(h) => {
println!(
"korp-collectors install: {} installed, enabled and started",
h.id
);
// Read the status back from systemd, never from the fact that start returned
// Ok. A lifecycle verb that prints success off its own return code is exactly
// the hollow green this house has been bitten by.
match rt.status(&h) {
Ok(st) if st.healthy => {
println!("korp-collectors install: systemd says {}", st.phase);
0
}
Ok(st) => {
eprintln!(
"korp-collectors install: the unit was written but systemd says {} — \
reporting that, not success",
st.phase
);
1
}
Err(e) => {
eprintln!("korp-collectors install: cannot read the unit back: {e:#}");
1
}
}
}
Err(e) => {
eprintln!("korp-collectors install: {e:#}");
1
}
}
}
}