src/runtime.rs
//! **The `Runtime` substrate trait** (feature `daemon`) — the interface that decides
//! *WHERE* a collector daemon (or, later, a knut pipeline) runs, decoupled from its
//! *logic*. Per [`../.nornir/runtime-aortas-design-2026-07-29.md`](../.nornir/runtime-aortas-design-2026-07-29.md).
//!
//! A collector's behaviour is **substrate-agnostic** ([`Collector`](crate::daemon::Collector)
//! against a [`CollectorCtx`](crate::daemon::CollectorCtx)); this trait is the seam that
//! spawns it on one of two substrates:
//!
//! * **[`RuntimeKind::Jera`]** — the nordisk substrate. *Metal* today: an in-process
//! jera DAEMON job on one host (the shipped `osm-cut` daemon). *Distributed* later
//! ([`RuntimeKind::JeraDistributed`], DESIGN-ONLY here): a worker registry that pulls
//! work — **not built**, a stubbed path with a `TODO`.
//! * **[`RuntimeKind::Kube`]** — full Kubernetes via kube-rs (SSA, zero-shell).
//!
//! ## Leaf discipline
//! This is **only the interface**. korp-collectors depends on neither korp nor jera nor
//! kube, so the trait pulls **no new deps** — it is pure `serde` + `anyhow` (both already
//! under `daemon`). The concrete [`Runtime`] implementations (`JeraRuntime`, `KubeRuntime`)
//! live in **korp**, which owns jera / aegir / skidbladnir. A [`UnitSpec`] is the
//! backend-neutral description of a runnable (a collector daemon now, a pipeline later)
//! that both implementations consume.
use std::path::PathBuf;
use serde::{Deserialize, Serialize};
/// Which substrate a [`UnitSpec`] runs on (design §"The `Runtime` trait — 2 substrates").
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum RuntimeKind {
/// The nordisk substrate — metal (in-process jera DAEMON job) now. Default.
#[default]
Jera,
/// jera grown to a worker registry + capability scheduling. **DESIGN-ONLY** — a
/// stubbed path with a `TODO`, deliberately NOT built (design §"Distributed").
JeraDistributed,
/// Full Kubernetes via kube-rs (server-side apply, zero-shell).
Kube,
/// **A systemd unit of its own** — its own cgroup, its own `Restart=`, its own
/// service account, installed and torn down **through Skidbladnir**
/// ([`crate::runtime_systemd::SystemdRuntime`], feature `systemd`).
///
/// This is the substrate that lets a collector be *nursed from outside the
/// process*: on [`Jera`](RuntimeKind::Jera) a collector is a thread inside its
/// host server, so an OOM in one collector takes every other collector — and the
/// host — with it.
///
/// **It is a choice, never a consequence.** Nothing in this crate infers a
/// substrate from a unit's size or runtime; [`UnitSpec::placement_advice`] may
/// *warn*, and that is the whole of its authority.
Systemd,
}
impl RuntimeKind {
/// Stable machine key (config JSON · `$KORP_STATE` · tests).
pub fn key(self) -> &'static str {
match self {
RuntimeKind::Jera => "jera",
RuntimeKind::JeraDistributed => "jera_distributed",
RuntimeKind::Kube => "kube",
RuntimeKind::Systemd => "systemd",
}
}
/// Parse a stable key back (round-trips [`key`](Self::key)); unknown ⇒ `None`.
pub fn from_key(s: &str) -> Option<Self> {
Some(match s.trim() {
"jera" | "" => RuntimeKind::Jera,
"jera_distributed" => RuntimeKind::JeraDistributed,
"kube" => RuntimeKind::Kube,
"systemd" => RuntimeKind::Systemd,
_ => return None,
})
}
}
/// What KIND of runnable a [`UnitSpec`] is — decides the k8s object a
/// [`RuntimeKind::Kube`] backend renders (Daemon → Deployment/StatefulSet, Cron →
/// CronJob, OneShot → Job) and how a [`RuntimeKind::Jera`] backend supervises it.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum UnitKind {
/// A never-ending daemon that strives to hold ONLINE (a collector; the shipped
/// `osm-cut`). Kube → Deployment (StatefulSet when it needs a durable dir).
Daemon,
/// A scheduled run on a cron expression. Kube → CronJob.
Cron {
/// A standard 5-field cron schedule (`"0 * * * *"`).
schedule: String,
},
/// A run-once-to-completion job. Kube → Job.
OneShot,
}
impl UnitKind {
/// Stable machine key (tests · the Kube object-name suffix).
pub fn key(&self) -> &'static str {
match self {
UnitKind::Daemon => "daemon",
UnitKind::Cron { .. } => "cron",
UnitKind::OneShot => "oneshot",
}
}
}
/// HOW the unit's code is entered — in-process (the Jera-metal path, a registered
/// [`Collector`](crate::daemon::Collector) run in a thread) or as a container image +
/// argv (the Kube / jera-distributed path).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "entry", rename_all = "snake_case")]
pub enum Entry {
/// In-process: run the registered collector of this name (Jera-metal). This is the
/// zero-infra path the shipped `osm-cut` daemon takes.
InProcess {
/// The registered [`Collector::name`](crate::daemon::CollectorConfig::name).
collector: String,
},
/// **One unit that nurses MANY collectors** — the standalone collector service
/// ([`crate::nursery`]).
///
/// This is the entry a foreign project installs. [`Entry::InProcess`] gives one unit
/// per collector, which is right when a collector needs its own cgroup and its own
/// `Restart=` — the 176 GB tiler does. It is wrong for twenty collectors that each
/// fetch 30 MB an hour: twenty units, twenty accounts and twenty registers to read,
/// for work that fits in one process. Both shapes exist because both are real, and the
/// unit declares which it is rather than a runtime guessing from a size.
Nursery {
/// The whole roster, serialised into the unit's environment as
/// [`crate::nursery::NURSERY_CONFIG_ENV`].
roster: serde_json::Value,
},
/// A container image + argv (Kube, or a jera-distributed worker running a container).
/// For a collector the argv is `korp collector run <name>` (design §Kube).
Image {
/// The OCI image reference.
image: String,
/// The container argv.
argv: Vec<String>,
},
}
impl Entry {
/// The in-process entry for the collector `name` (the Jera-metal default).
pub fn in_process(name: impl Into<String>) -> Self {
Entry::InProcess {
collector: name.into(),
}
}
/// The nursery entry for a whole roster — one unit, many collectors.
pub fn nursery(roster: serde_json::Value) -> Self {
Entry::Nursery { roster }
}
/// A container entry running `korp collector run <name>` from `image` — the shape a
/// [`RuntimeKind::Kube`] backend renders into a pod command.
pub fn collector_image(image: impl Into<String>, name: &str) -> Self {
Entry::Image {
image: image.into(),
argv: vec!["korp".into(), "collector".into(), "run".into(), name.into()],
}
}
}
/// The unit's storage needs — a durable on-disk dir (⇒ a PVC / StatefulSet on Kube) and
/// an S3 bucket for servable artifacts (⇒ the rustfs Service on Kube).
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct StorageSpec {
/// A durable per-unit directory (`~/.korp/collectors/<name>/` on metal). Its presence
/// makes a Kube [`UnitKind::Daemon`] a StatefulSet + PVC rather than a Deployment.
#[serde(default)]
pub durable_dir: Option<PathBuf>,
/// **The durable working set this unit DECLARES it needs**, as a Kubernetes
/// quantity (`"176Gi"`). `None` means *the unit declared nothing* — which is the
/// honest state, and the only correct value when nobody knows.
///
/// It used to be filled in as `"20Gi"` for every collector by the caller that
/// built the spec, which made `osm-cut` — 176 366 896 594 B of planet — read as
/// a 20 GiB unit. **A field that lies is worse than no field**, because it looks
/// like information: a PVC sized from it is silently 8× too small, and an operator
/// reading it has no way to tell a real declaration from a filled-in default.
/// Set it from the collector's own [`durable_bytes`](crate::daemon::CollectorConfig::durable_bytes)
/// declaration via [`UnitSpec::durable_bytes`], or leave it `None`.
///
/// It **never selects a substrate**. See [`UnitSpec::placement_advice`].
#[serde(default)]
pub durable_size: Option<String>,
/// The S3 bucket for servable artifacts (`nordisk-maps`) — backed by embedded rustfs
/// on metal, the rustfs Service on Kube.
#[serde(default)]
pub s3_bucket: Option<String>,
}
/// Coarse resource requests (Kube container `resources`; advisory on metal).
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Resources {
/// CPU request in millicores (`1000` = one core). `None` ⇒ unset.
#[serde(default)]
pub cpu_millis: Option<u32>,
/// Memory request in MiB. `None` ⇒ unset.
#[serde(default)]
pub memory_mb: Option<u32>,
}
/// **The backend-neutral description of a runnable** (design §"UnitSpec") — a collector
/// daemon now, a knut pipeline later. Both [`Runtime`] implementations consume it; the
/// Kube backend renders it to manifests, the Jera backend spawns it as a DAEMON job.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct UnitSpec {
/// The unit name (the collector name; the k8s object name).
pub name: String,
/// Daemon / Cron / OneShot.
pub kind: UnitKind,
/// In-process fn vs image + argv.
pub entry: Entry,
/// The unit's own config (the collector's persisted `params`) — a ConfigMap on Kube,
/// the [`CollectorCtx::config`](crate::daemon::CollectorCtx::config) on metal.
#[serde(default)]
pub config: serde_json::Value,
/// Durable dir + S3 bucket.
#[serde(default)]
pub storage: StorageSpec,
/// CPU / memory requests.
#[serde(default)]
pub resources: Resources,
/// Public (non-secret) environment — inline env on Kube.
#[serde(default)]
pub env: Vec<(String, String)>,
/// Secret environment (creds) — a Secret on Kube, never inline.
#[serde(default)]
pub secret_env: Vec<(String, String)>,
/// The container image for the Kube render when [`Entry`] is in-process (the metal
/// path has no image). `None` ⇒ the Kube backend falls back to `korp:latest`.
#[serde(default)]
pub image: Option<String>,
/// The k8s namespace for the Kube render. `None` ⇒ `default`.
#[serde(default)]
pub namespace: Option<String>,
}
impl UnitSpec {
/// A **collector daemon** unit: in-process entry, a durable dir + S3 bucket, no cron.
/// The shape the shipped `osm-cut` daemon is expressed as for [`RuntimeKind::Jera`].
pub fn collector_daemon(name: impl Into<String>, config: serde_json::Value) -> Self {
let name = name.into();
UnitSpec {
entry: Entry::in_process(&name),
name,
kind: UnitKind::Daemon,
config,
storage: StorageSpec::default(),
resources: Resources::default(),
env: Vec::new(),
secret_env: Vec::new(),
image: None,
namespace: None,
}
}
/// **A NURSERY unit: one service, many collectors.** The shape a foreign project
/// installs — `korp-collectors serve` with the roster in its environment.
///
/// `name` is the unit's own name (`collectors` ⇒ `collector-collectors.service`), not a
/// collector's; the roster names the collectors.
pub fn nursery(name: impl Into<String>, roster: serde_json::Value) -> Self {
UnitSpec {
name: name.into(),
kind: UnitKind::Daemon,
entry: Entry::nursery(roster),
config: serde_json::Value::Null,
storage: StorageSpec::default(),
resources: Resources::default(),
env: Vec::new(),
secret_env: Vec::new(),
image: None,
namespace: None,
}
}
/// Set the durable per-unit dir (Kube ⇒ StatefulSet + PVC).
pub fn durable_dir(mut self, dir: impl Into<PathBuf>) -> Self {
self.storage.durable_dir = Some(dir.into());
self
}
/// Set the PVC size for the durable dir on Kube (e.g. `"20Gi"`).
pub fn durable_size(mut self, size: impl Into<String>) -> Self {
self.storage.durable_size = Some(size.into());
self
}
/// Set the servable-artifact S3 bucket.
pub fn s3_bucket(mut self, bucket: impl Into<String>) -> Self {
self.storage.s3_bucket = Some(bucket.into());
self
}
/// Set the container image used by the Kube render.
pub fn image(mut self, image: impl Into<String>) -> Self {
self.image = Some(image.into());
self
}
/// Set the k8s namespace used by the Kube render.
pub fn namespace(mut self, ns: impl Into<String>) -> Self {
self.namespace = Some(ns.into());
self
}
/// Add a public env var.
pub fn env(mut self, k: impl Into<String>, v: impl Into<String>) -> Self {
self.env.push((k.into(), v.into()));
self
}
/// Add a secret env var (a Kube Secret; never inline).
pub fn secret_env(mut self, k: impl Into<String>, v: impl Into<String>) -> Self {
self.secret_env.push((k.into(), v.into()));
self
}
/// **Declare the durable working set in BYTES** — the honest twin of
/// [`durable_size`](Self::durable_size), which takes a pre-formatted string and so
/// invites a made-up one. Renders the nearest exact binary unit (`Gi`/`Mi`/`Ki`),
/// rounding **up** so a rendered PVC is never smaller than what was declared.
pub fn durable_bytes(mut self, bytes: u64) -> Self {
self.storage.durable_size = Some(format_binary_quantity(bytes));
self
}
/// Does this unit need a durable dir (⇒ Kube StatefulSet not Deployment)?
pub fn is_stateful(&self) -> bool {
self.storage.durable_dir.is_some()
}
/// **Everything worth saying out loud about running this unit on `kind` — and
/// nothing more.** Returns advice, never a decision: the programmer picks the
/// substrate, and this function's whole authority is to name a mismatch they may
/// then ignore. An empty vec means "nothing to say", NOT "approved".
///
/// Deliberately NOT a threshold table that routes a unit to a substrate. A rule
/// that sends a 176 GiB unit to systemd looks like rigour and is really a policy
/// nobody voted for, applied at the moment somebody is trying to do something
/// unusual on purpose.
pub fn placement_advice(&self, kind: RuntimeKind) -> Vec<String> {
let mut out = Vec::new();
let declared = self
.storage
.durable_size
.as_deref()
.and_then(parse_binary_quantity);
const GIB: u64 = 1024 * 1024 * 1024;
if self.is_stateful() && self.storage.durable_size.is_none() {
out.push(format!(
"unit '{}' has a durable dir but declares no size — a Kube render must \
then guess a PVC size, and nobody can tell an unsized unit from a small one",
self.name
));
}
if let Some(bytes) = declared {
if kind == RuntimeKind::Kube && bytes > 100 * GIB {
out.push(format!(
"unit '{}' declares {} of durable storage on Kube: a PVC that large \
pins the pod to one node, and an evicted pod restarts the step from zero",
self.name,
format_binary_quantity(bytes)
));
}
if kind == RuntimeKind::Jera && bytes > 10 * GIB {
out.push(format!(
"unit '{}' declares {} of durable storage on the in-process substrate: \
it shares its host's process, so an OOM here takes every other collector \
and the host with it",
self.name,
format_binary_quantity(bytes)
));
}
}
// The nursery re-introduces, on purpose, the weakness the in-process substrate has:
// one process for many collectors. That is the right trade for twenty collectors
// fetching megabytes, and the wrong one for a collector that can exhaust the host —
// so it is SAID, at the point a spec is built, and never enforced.
if let Entry::Nursery { roster } = &self.entry {
let n = roster
.get("collectors")
.and_then(|c| c.as_array())
.map(|a| a.len())
.unwrap_or(0);
if let Some(bytes) = declared {
if bytes > 10 * GIB {
out.push(format!(
"unit '{}' is a NURSERY of {n} collector(s) declaring {} of durable storage: they share one process, so an OOM in the big one takes the other {} down with it. A collector that can exhaust the host wants a unit of its own (Entry::in_process)",
self.name,
format_binary_quantity(bytes),
n.saturating_sub(1)
));
}
}
if n > 1 && kind == RuntimeKind::Kube {
out.push(format!(
"unit '{}' is a NURSERY of {n} collector(s) on Kube: one pod supervising many collectors gives up the per-collector restart and per-collector limits that are the reason to be on Kube at all. One unit per collector (Entry::in_process) is the shape that substrate rewards",
self.name
));
}
}
if matches!(self.kind, UnitKind::Cron { .. }) && kind == RuntimeKind::Jera {
out.push(format!(
"unit '{}' is a Cron unit on the in-process substrate, which has no \
scheduler — its run loop must sleep for itself",
self.name
));
}
out
}
}
/// Render `bytes` as the nearest EXACT binary Kubernetes quantity, rounding up so a
/// PVC sized from it is never smaller than what was declared (`176366896594` ⇒ `"165Gi"`).
pub fn format_binary_quantity(bytes: u64) -> String {
const KI: u64 = 1024;
const MI: u64 = KI * 1024;
const GI: u64 = MI * 1024;
let ceil = |n: u64, d: u64| n.div_ceil(d);
if bytes >= GI {
format!("{}Gi", ceil(bytes, GI))
} else if bytes >= MI {
format!("{}Mi", ceil(bytes, MI))
} else if bytes >= KI {
format!("{}Ki", ceil(bytes, KI))
} else {
format!("{bytes}")
}
}
/// Parse a binary Kubernetes quantity (`"20Gi"`, `"512Mi"`, `"1024"`) back to bytes.
/// Decimal SI suffixes (`G`, `M`) and anything unparseable ⇒ `None` — a reader that
/// cannot be sure what a string meant must say so rather than assume.
pub fn parse_binary_quantity(s: &str) -> Option<u64> {
let s = s.trim();
let (num, mult) = if let Some(n) = s.strip_suffix("Gi") {
(n, 1024u64 * 1024 * 1024)
} else if let Some(n) = s.strip_suffix("Mi") {
(n, 1024 * 1024)
} else if let Some(n) = s.strip_suffix("Ki") {
(n, 1024)
} else {
(s, 1)
};
num.trim()
.parse::<u64>()
.ok()
.and_then(|v| v.checked_mul(mult))
}
/// An opaque handle to a started unit — enough for [`Runtime::stop`] / [`Runtime::status`]
/// to find it again (the jera job / the k8s object).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Handle {
/// Which substrate started it.
pub runtime: RuntimeKind,
/// The unit name.
pub name: String,
/// The substrate-specific id — the jera `job_id`, or the k8s `<namespace>/<kind>/<name>`.
pub id: String,
}
impl Handle {
/// A handle for `name` on `runtime` with substrate id `id`.
pub fn new(runtime: RuntimeKind, name: impl Into<String>, id: impl Into<String>) -> Self {
Handle {
runtime,
name: name.into(),
id: id.into(),
}
}
}
/// The live status of a started unit — the phase label, a health flag, and recent logs
/// (design §"RunStatus{phase,healthy,logs}").
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunStatus {
/// A short phase label ("online", "cutting sweden 50%", "rolling out", "failed: …").
pub phase: String,
/// Green only when the unit holds its steady/ready state.
pub healthy: bool,
/// Recent log lines (best-effort; may be empty for a substrate that does not surface them).
#[serde(default)]
pub logs: Vec<String>,
}
impl RunStatus {
/// A healthy (green) status with `phase`.
pub fn healthy(phase: impl Into<String>) -> Self {
RunStatus {
phase: phase.into(),
healthy: true,
logs: Vec::new(),
}
}
/// A not-yet-healthy status with `phase` (a working state or a failure).
pub fn unhealthy(phase: impl Into<String>) -> Self {
RunStatus {
phase: phase.into(),
healthy: false,
logs: Vec::new(),
}
}
}
/// **The substrate trait** (design §"The `Runtime` trait"). A collector daemon (and, later,
/// a knut pipeline) is spawned through this — Jera-metal now, Jera-distributed / Kube-ready
/// — never hard-wired to in-process. Object-safe (`&self` methods, no generics), so a
/// `Box<dyn Runtime>` selects the substrate at runtime from the persisted per-collector
/// choice.
pub trait Runtime: Send + Sync {
/// Which substrate this is.
fn kind(&self) -> RuntimeKind;
/// Start (or resume) `spec`, returning a [`Handle`] to it. Idempotent where the
/// substrate allows (a jera daemon already running resumes; a Kube apply is SSA).
fn start(&self, spec: &UnitSpec) -> anyhow::Result<Handle>;
/// Stop the unit `h` refers to. Idempotent (a no-op when it is already stopped).
fn stop(&self, h: &Handle) -> anyhow::Result<()>;
/// The live [`RunStatus`] of the unit `h` refers to.
fn status(&self, h: &Handle) -> anyhow::Result<RunStatus>;
/// **Reconcile** the running set to `desired`: start every unit in `desired` that is
/// not running, and stop everything running that is not in `desired`. This is the one
/// call the Settings toggle / the matrix ▶Run route through.
fn reconcile(&self, desired: &[UnitSpec]) -> anyhow::Result<()>;
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
/// **A declared size is carried, an undeclared one stays undeclared.** The bug
/// this replaces was a caller stamping `"20Gi"` onto every unit, so this asserts
/// BOTH directions: the declaration survives round-trip, and absence stays absence.
#[test]
fn durable_size_is_declared_or_absent_never_invented() {
// The measured planet, `.nornir/collector-runtime-design-2026-08-04.md` §6.1.
const PLANET: u64 = 176_366_896_594;
let declared = UnitSpec::collector_daemon("osm-cut", serde_json::json!({}))
.durable_dir("/var/lib/x")
.durable_bytes(PLANET);
let q = declared.storage.durable_size.as_deref().unwrap();
assert_eq!(q, "165Gi", "176 366 896 594 B rounded UP to whole GiB");
let back = parse_binary_quantity(q).unwrap();
assert!(
back >= PLANET,
"a PVC sized from the quantity is never SMALLER than declared: {back} < {PLANET}"
);
assert!(
back - PLANET < 1024 * 1024 * 1024,
"and never more than a GiB larger"
);
let undeclared = UnitSpec::collector_daemon("dcat", serde_json::json!({}));
assert!(
undeclared.storage.durable_size.is_none(),
"nothing declared ⇒ nothing stored"
);
}
/// Quantities round-trip, and an unparseable one is `None` rather than a guess.
#[test]
fn binary_quantities_round_trip_and_refuse_what_they_cannot_read() {
for b in [
0u64,
1,
1023,
1024,
5 * 1024 * 1024,
20 * 1024 * 1024 * 1024,
] {
let q = format_binary_quantity(b);
assert_eq!(parse_binary_quantity(&q), Some(b), "{b} -> {q} -> ?");
}
assert_eq!(parse_binary_quantity("20Gi"), Some(20 * 1024 * 1024 * 1024));
// Decimal SI is a DIFFERENT number; refusing beats silently reading it as binary.
assert_eq!(
parse_binary_quantity("20G"),
None,
"decimal SI is not binary — say so"
);
assert_eq!(parse_binary_quantity("lots"), None);
assert_eq!(parse_binary_quantity(""), None);
}
/// **Advice advises. It never selects.** Every arm here returns strings; there is
/// no code path in this crate that turns a size into a `RuntimeKind`, and the
/// assertions below pin that: the SAME spec is legal on every substrate.
#[test]
fn placement_advice_warns_and_never_chooses() {
const GIB: u64 = 1024 * 1024 * 1024;
let big = UnitSpec::collector_daemon("osm-cut", serde_json::json!({}))
.durable_dir("/var/lib/x")
.durable_bytes(176 * GIB);
// Kube: says the PVC pins the pod. Jera: says an OOM takes the host.
let on_kube = big.placement_advice(RuntimeKind::Kube);
assert!(
on_kube.iter().any(|w| w.contains("pins the pod")),
"{on_kube:?}"
);
let on_jera = big.placement_advice(RuntimeKind::Jera);
assert!(
on_jera
.iter()
.any(|w| w.contains("takes every other collector")),
"{on_jera:?}"
);
// systemd: the substrate that isolates it — nothing to say.
assert!(
big.placement_advice(RuntimeKind::Systemd).is_empty(),
"no warning is not approval, but there is nothing to warn about"
);
// A small unit draws no size warning ANYWHERE — so the warnings above came from
// the declared size, not from the substrate.
let small = UnitSpec::collector_daemon("dcat", serde_json::json!({}))
.durable_dir("/var/lib/d")
.durable_bytes(64 * 1024 * 1024);
for k in [RuntimeKind::Jera, RuntimeKind::Kube, RuntimeKind::Systemd] {
assert!(
small.placement_advice(k).is_empty(),
"{k:?}: {:?}",
small.placement_advice(k)
);
}
// A durable dir with NO declaration is itself the thing worth saying.
let unsized_unit =
UnitSpec::collector_daemon("mystery", serde_json::json!({})).durable_dir("/var/lib/m");
assert!(
unsized_unit
.placement_advice(RuntimeKind::Kube)
.iter()
.any(|w| w.contains("declares no size")),
"an undeclared size is reported, not defaulted"
);
// A Cron unit on a substrate with no scheduler.
let cron = UnitSpec {
kind: UnitKind::Cron {
schedule: "0 * * * *".into(),
},
..UnitSpec::collector_daemon("nordpool", serde_json::json!({}))
};
assert!(
cron.placement_advice(RuntimeKind::Jera)
.iter()
.any(|w| w.contains("no scheduler"))
);
assert!(
cron.placement_advice(RuntimeKind::Kube).is_empty(),
"Kube renders a CronJob — nothing to warn about"
);
}
#[test]
fn runtime_kind_keys_round_trip() {
for k in [
RuntimeKind::Jera,
RuntimeKind::JeraDistributed,
RuntimeKind::Kube,
] {
assert_eq!(RuntimeKind::from_key(k.key()), Some(k));
}
assert_eq!(
RuntimeKind::default(),
RuntimeKind::Jera,
"default substrate is jera"
);
assert_eq!(
RuntimeKind::from_key(""),
Some(RuntimeKind::Jera),
"blank ⇒ jera"
);
assert_eq!(RuntimeKind::from_key("nope"), None);
}
#[test]
fn unit_spec_builder_and_serde_round_trip() {
let spec =
UnitSpec::collector_daemon("osm-cut", serde_json::json!({ "regions": ["sweden"] }))
.durable_dir("/home/x/.korp/collectors/osm-cut")
.durable_size("20Gi")
.s3_bucket("nordisk-maps")
.image("korp:latest")
.namespace("korp")
.env("KORP_S3_ENDPOINT", "http://rustfs:9000")
.secret_env("KORP_S3_SECRET", "minioadmin");
assert_eq!(spec.kind, UnitKind::Daemon);
assert_eq!(
spec.entry,
Entry::InProcess {
collector: "osm-cut".into()
}
);
assert!(spec.is_stateful(), "a durable dir makes it stateful");
assert_eq!(spec.storage.s3_bucket.as_deref(), Some("nordisk-maps"));
// Serde round-trips (persisted / inspected).
let s = serde_json::to_string(&spec).unwrap();
let back: UnitSpec = serde_json::from_str(&s).unwrap();
assert_eq!(back, spec);
}
#[test]
fn unit_kind_keys_and_cron_carries_schedule() {
assert_eq!(UnitKind::Daemon.key(), "daemon");
assert_eq!(UnitKind::OneShot.key(), "oneshot");
let cron = UnitKind::Cron {
schedule: "0 * * * *".into(),
};
assert_eq!(cron.key(), "cron");
let back: UnitKind = serde_json::from_str(&serde_json::to_string(&cron).unwrap()).unwrap();
assert_eq!(back, cron);
}
#[test]
fn collector_image_entry_is_korp_collector_run() {
let e = Entry::collector_image("korp:latest", "osm-cut");
assert_eq!(
e,
Entry::Image {
image: "korp:latest".into(),
argv: vec![
"korp".into(),
"collector".into(),
"run".into(),
"osm-cut".into()
],
}
);
}
/// **The trait is object-safe and drives a `Box<dyn Runtime>`.** A mock runtime records
/// its reconcile set — the exact shape korp's Settings toggle uses to select a
/// substrate at runtime from the persisted per-collector choice.
#[test]
fn runtime_is_object_safe_via_a_mock() {
struct MockRuntime {
running: Mutex<Vec<String>>,
}
impl Runtime for MockRuntime {
fn kind(&self) -> RuntimeKind {
RuntimeKind::Jera
}
fn start(&self, spec: &UnitSpec) -> anyhow::Result<Handle> {
self.running.lock().unwrap().push(spec.name.clone());
Ok(Handle::new(RuntimeKind::Jera, &spec.name, &spec.name))
}
fn stop(&self, h: &Handle) -> anyhow::Result<()> {
self.running.lock().unwrap().retain(|n| n != &h.name);
Ok(())
}
fn status(&self, h: &Handle) -> anyhow::Result<RunStatus> {
let running = self.running.lock().unwrap().iter().any(|n| n == &h.name);
Ok(if running {
RunStatus::healthy("online")
} else {
RunStatus::unhealthy("off")
})
}
fn reconcile(&self, desired: &[UnitSpec]) -> anyhow::Result<()> {
let want: Vec<String> = desired.iter().map(|s| s.name.clone()).collect();
self.running.lock().unwrap().retain(|n| want.contains(n));
for s in desired {
self.start(s)?;
}
Ok(())
}
}
let rt: Box<dyn Runtime> = Box::new(MockRuntime {
running: Mutex::new(Vec::new()),
});
assert_eq!(rt.kind(), RuntimeKind::Jera);
let spec = UnitSpec::collector_daemon("osm-cut", serde_json::json!({}));
let h = rt.start(&spec).unwrap();
assert!(rt.status(&h).unwrap().healthy);
// reconcile to the empty set stops it.
rt.reconcile(&[]).unwrap();
assert!(
!rt.status(&h).unwrap().healthy,
"reconcile(&[]) stops everything"
);
}
/// **A nursery's placement advice names the trade it makes, and never enforces it.**
///
/// The nursery deliberately re-introduces the shared-process weakness: one process, many
/// collectors. That is right for twenty collectors fetching megabytes and wrong for one
/// that can exhaust the host, so the mismatch is SAID at the point a spec is built — and
/// `placement_advice` returns strings, which no code path can act on.
#[test]
fn a_nursery_is_advised_about_the_process_it_shares_and_never_stopped() {
let roster = serde_json::json!({
"collectors": [{ "name": "osm-cut" }, { "name": "dcat" }, { "name": "pxweb" }]
});
let big = UnitSpec::nursery("collectors", roster.clone())
.durable_dir("/home/_c/collectors/data")
.durable_bytes(176_366_896_594);
let advice = big.placement_advice(RuntimeKind::Systemd);
assert!(
advice
.iter()
.any(|a| a.contains("share one process") && a.contains("165Gi")),
"a large nursery must be told what it is sharing: {advice:?}"
);
assert!(
advice.iter().any(|a| a.contains("Entry::in_process")),
"the advice must name the shape that fixes it: {advice:?}"
);
// …and a SMALL nursery is not lectured. An advisor that always speaks is noise, and
// noise is how the one line that mattered gets skipped.
let small = UnitSpec::nursery("collectors", roster.clone())
.durable_dir("/home/_c/collectors/data")
.durable_bytes(2 * 1024 * 1024 * 1024);
assert!(
small
.placement_advice(RuntimeKind::Systemd)
.iter()
.all(|a| !a.contains("share one process")),
"a 2 GiB nursery needs no OOM warning: {:?}",
small.placement_advice(RuntimeKind::Systemd)
);
// On Kube, many-in-one gives up exactly what that substrate is for.
let kube = big.placement_advice(RuntimeKind::Kube);
assert!(
kube.iter().any(|a| a.contains("per-collector restart")),
"a Kube nursery must be told what it gives up: {kube:?}"
);
// A single-collector nursery on Kube gives nothing up, so it is not warned.
let one = UnitSpec::nursery(
"collectors",
serde_json::json!({ "collectors": [{ "name": "dcat" }] }),
);
assert!(
one.placement_advice(RuntimeKind::Kube)
.iter()
.all(|a| !a.contains("per-collector restart")),
"a one-collector nursery on Kube is not the shape being warned about"
);
// ★ The whole point: advice is advice. Nothing here selects a substrate.
for kind in [RuntimeKind::Jera, RuntimeKind::Kube, RuntimeKind::Systemd] {
let _: Vec<String> = big.placement_advice(kind);
}
}
}