diff --git a/.claude/CHANGELOG.md b/.claude/CHANGELOG.md index 82d3879..76490b9 100644 --- a/.claude/CHANGELOG.md +++ b/.claude/CHANGELOG.md @@ -9,6 +9,213 @@ The format is based on the regulated environment requirements: --- +## [2026-05-02 23:30] - Emergency-reclaim roadmap closeout: drain stopwatch + loop protection + integration tests + +**Author:** Erick Bourgeois + +### Added +- `src/loop_protection.rs` (~75 LOC) + `src/loop_protection_tests.rs` + (14 pure-helper tests). Module exposes `prune_old_reclaim_events`, + `record_emergency_reclaim_event`, and `should_warn_rapid_re_reclaim` + — all operate on `&mut VecDeque>` so the tests drive + every branch with no clock or controller. Threshold + window come + from new constants in `src/constants.rs` + (`RAPID_RE_RECLAIM_THRESHOLD = 3`, + `RAPID_RE_RECLAIM_WINDOW_SECS = 600`, + `RAPID_RE_RECLAIM_MAX_TRACKED = 10`, + `REASON_RAPID_RE_RECLAIM = "RapidReReclaim"`). +- `Context.recent_reclaims: + Arc>>>>` — in-memory + per-SM tracker, keyed by `"namespace/name"`. Documented choice: + no CRD schema bump, no status-patch round trip; controller restart + resets the count, which the metric still captures on a trend basis. +- `src/metrics.rs`: three new metrics with helpers — + `fivespot_emergency_drain_duration_seconds` (HistogramVec, + outcome={success|timeout|error}, buckets sized for the 60 s drain + ceiling), `fivespot_emergency_reclaims_total` (CounterVec by + namespace+name), `fivespot_rapid_re_reclaims_total` (CounterVec by + namespace+name). Helpers `record_emergency_drain`, + `record_emergency_reclaim`, `record_rapid_re_reclaim` plus + `EmergencyDrainOutcome` enum. 4 new tests in `metrics_tests.rs`. +- `src/reconcilers/helpers.rs`: drain call in `handle_emergency_remove` + is now wrapped in an `Instant::now()` stopwatch; outcome classified + by error message (timeout substring → Timeout, other Err → Error, + Ok → Success). Per-SM emergency-reclaim counter bumped once per + flow (the idempotent recovery handler short-circuits before this + point on restart). Loop-protection check fires after the bump: + appends timestamp, prunes old entries, emits `RapidReReclaim` + Warning Event + bumps the new metric when threshold crosses. New + pure helper `build_rapid_re_reclaim_event(name)`. +- `tests/integration_netlink_proc.rs` — Linux-only (`#![cfg(target_os + = "linux")]`) runtime test for the netlink subscriber. Two + `#[ignore]` cases: one targets a spawned `/bin/true` child and + asserts <100 ms detection latency per issue #40 acceptance + criterion; one is a quieter "subscriber sees *some* event" + smoke test. +- `tests/integration_emergency_reclaim.rs` — kind-cluster annotation + contract test (graceful-skip if no cluster, `#[ignore]`). Two cases: + agent's `build_patch_body` round-trips through the controller's + `node_reclaim_request` parser; partial annotations (reason set, + requested=true absent) parse to `None`. Plus one always-on + constants-pinning test. + +### Changed +- `src/reconcilers/mod.rs`: re-export `build_clear_reclaim_patch`, + `node_reclaim_request`, and `ReclaimRequest` from `helpers` so the + kind-cluster test can use the published API surface. +- Roadmap `~/dev/roadmaps/5spot-emergency-reclaim-by-process-match.md` + → renamed `completed-5spot-emergency-reclaim-by-process-match.md` + (per the user-memory rule "rename completed roadmaps with + `completed-` prefix when every phase is done"). Status header + rewritten to ✅ Complete; Phase 2.c rung 2 section flipped from + "deferred to issue #40" to fully shipped; deferred-items section + rewritten as "Closed follow-ups" with the four checkboxes ticked. + +### Why +Closes the four open items the rung 2 implementation called out as +explicit follow-ups: + +1. **Drain stopwatch** — operators now have a Prometheus histogram of + actual emergency-drain duration sliced by outcome, so SLO dashboards + can compute success-only P95 vs timeout-rate side by side. +2. **Re-enable loop protection** — when a user re-enables a SM whose + conflicting process is still running, the third reclaim within + 10 minutes now triggers a `RapidReReclaim` Warning Event telling the + operator to stop the conflicting process before re-enabling. The + loop is no longer silent. +3. **Linux netlink runtime test** — issue #40 acceptance criterion + "JVM launch produces a match within <100 ms" is now verifiable via + `cargo test --test integration_netlink_proc -- --ignored` on a real + Linux node with `CAP_NET_ADMIN`. +4. **Kind-cluster integration test** — proves the agent → controller + annotation contract round-trips through a real API server. Catches + the entire class of "agent writes annotations the controller can't + parse" regressions that no unit test would surface. + +With these closed, every phase of the +`5spot-emergency-reclaim-by-process-match.md` roadmap is shipped and +the file moves to `completed-` prefix. + +### Impact +- [ ] Breaking change (no API surface change; new metrics + new event + reason are additive) +- [x] Requires cluster rollout (binary changes; CRD unchanged so no + `kubectl apply` of the CRD YAML needed) +- [ ] Config change only +- [ ] Documentation only + +### Verification +- `cargo build`: clean. +- `cargo test`: **450 unit tests + 3 always-on integration tests pass**; + 4 `#[ignore]` integration tests (2 emergency-reclaim, 2 netlink) gated + on a real cluster / Linux + CAP_NET_ADMIN respectively. +- `cargo fmt --check`: clean. +- `cargo clippy --all-targets -- -D warnings`: clean. + +--- + +## [2026-05-02 23:00] - Reclaim-agent rung 2: netlink proc connector subscriber + +**Author:** Erick Bourgeois + +### Added +- `src/netlink_proc.rs` (~280 LOC) — new module exposing `ProcEvent`, + `NetlinkError`, portable parsers (`parse_cn_msg`, `parse_proc_event`), + and `Subscriber`. Linux impl opens an `AF_NETLINK` / + `NETLINK_CONNECTOR` socket, binds to the `CN_IDX_PROC` multicast + group, sends the `PROC_CN_MCAST_LISTEN` control message, and yields + one `ProcEvent` per kernel `exec(2)`. Non-Linux: `Subscriber::new()` + immediately returns `NetlinkError::Unsupported` so the binary still + links cleanly on macOS / Windows builds. +- `src/netlink_proc_tests.rs` — 14 byte-level parser tests + 1 + non-Linux stub contract test = **15 new tests, all running on macOS + CI** without needing a kernel: exec / fork / exit classification, + truncated buffers, wrong connector id, zero-length payload, + cn_msg→proc_event round-trip, non-Linux `Subscriber::new` returns + `Unsupported`. +- `src/bin/reclaim_agent.rs`: new `--detector={auto|netlink|poll}` + flag (env `RECLAIM_DETECTOR`, default `auto`). `auto` selects + `netlink` on Linux and `poll` elsewhere. New `run_netlink_scanner` + loop runs alongside the existing `run_scanner`; selection happens + once after host-id verification, before either loop starts. The + netlink loop opens the subscriber only when armed by a non-empty + ConfigMap and tears it down on config-clear, so an idle agent holds + no kernel resources. The `recv(2)` syscall runs on a `spawn_blocking` + thread; `tokio::select!` cancels via the existing config watch + channel. +- `Cargo.toml`: `target.'cfg(target_os = "linux")'.dependencies.nix = + { version = "0.30", default-features = false, features = ["socket", + "net", "uio"] }`. Linux-only — non-Linux builds pull in nothing new. +- `deploy/node-agent/daemonset.yaml`: container `securityContext` + gains `capabilities.add: [NET_ADMIN]`. The pod-level comment block + rewritten to reflect that this is no longer "future work." +- `.trivyignore`: new `AVD-KSV-0022` entry justifying the + `CAP_NET_ADMIN` add against the existing architectural-necessity + rationale pattern (matches the KSV-0010 / KSV-0012 / KSV-0023 etc. + blocks already in the file). +- `docs/src/concepts/emergency-reclaim.md`: new "Detector — rung 1 + vs rung 2" subsection with the comparison table, selection flags, + the heavy-exec-storm tradeoff, and a note on why the cap is granted + pod-level rather than per-detector. + +### Changed +- `src/reclaim_agent.rs`: `match_pid` visibility raised from `fn` to + `pub fn` so the rung 2 netlink subscriber can resolve a kernel-pushed + pid through the same match logic rung 1 uses. Body and behaviour + unchanged. Existing 46 tests stay green. +- `src/lib.rs`: `pub mod netlink_proc` added; module list comment + block extended. +- `docs/src/concepts/emergency-reclaim.md` status header updated: + rung 2 is shipped, no longer "follow-up work." + +### Why +Closes the deferred Phase 2 rung 2 of +`~/dev/roadmaps/5spot-emergency-reclaim-by-process-match.md` and +GitHub issue #40. Rung 1 already meets the <5 s SLA, so this is +optimisation, not correctness — payoff is detection latency dropping +from up to 250 ms (worst-case poll phase) to <10 ms (kernel push) +and idle CPU dropping to zero on quiet nodes. The detector is opt-in +via `--detector=netlink`; `auto` (the default) picks netlink on Linux +and poll elsewhere. Operators on heavy-exec workloads can pin +`--detector=poll` to avoid the per-exec cost. + +The netlink ABI is identical across Linux x86_64 and aarch64; both +build targets in `.github/workflows/build.yaml` get rung 2 with no +per-arch code. Windows is out of scope (no netlink ABI; would need an +ETW-based sibling impl behind the same trait, separate effort). + +### Impact +- [ ] Breaking change (new flag is opt-in via `auto` default; existing + operators with `RECLAIM_DETECTOR` unset continue on the + platform-appropriate default; rung 1 stays available as `poll`) +- [x] Requires cluster rollout (DaemonSet manifest changes — + `kubectl apply -k deploy/node-agent/` to pick up `CAP_NET_ADMIN` + and the new env var support) +- [ ] Config change only +- [ ] Documentation only + +### Verification +- `cargo build --bin 5spot-reclaim-agent`: clean. +- `cargo test --lib netlink_proc`: 15/15 pass on macOS aarch64. +- `cargo test --lib reclaim_agent`: 46/46 pass (unchanged from + pre-change baseline). +- `cargo fmt --check` + `cargo clippy --all-targets -- -D warnings`: + clean. +- Linux runtime verification deferred to CI (kind cluster) and to + hardware testing — same gating as rung 1's integration test scope. + The macOS dev loop uses `--detector=poll` and exercises the bin + end-to-end without netlink. + +### Trust model (unchanged) +The netlink socket is opened with `CAP_NET_ADMIN` granted at the +container level. The node-scoped credentials posture is unchanged — +the agent still binds only to the proc-event multicast group and +cannot send control messages to other netlink subsystems. The +host-identity cross-check (Phase 4 of the 2026-04-25 security audit) +still gates every PATCH regardless of detector choice. + +--- + ## [2026-05-02 10:00] - Symbol-import auto-VEX (roadmap Phase 3, scope-revised) **Author:** Erick Bourgeois diff --git a/.trivyignore b/.trivyignore index dd08959..271a745 100644 --- a/.trivyignore +++ b/.trivyignore @@ -173,6 +173,23 @@ AVD-KSV-0118 # writable volumes whose ownership fsGroup would govern. AVD-KSV-0116 +# KSV-0022 — Container has non-default capabilities added. +# Architecturally required for rung 2 of the reclaim-agent's detection +# ladder (`5spot-emergency-reclaim-by-process-match.md` Phase 2 rung 2, +# GitHub issue #40). `CAP_NET_ADMIN` is required to open an +# `AF_NETLINK` / `NETLINK_CONNECTOR` socket and bind to the +# `CN_IDX_PROC` multicast group — the kernel-pushed event source the +# subscriber in `src/netlink_proc.rs` consumes. Rung 1 (the `/proc` +# poll) does not need this cap; operators who pin `--detector=poll` +# never exercise it. We grant the cap once at the pod level rather than +# letting the binary fall back to rung 1 silently at runtime — the +# detector choice is an operator decision, not an availability +# heuristic. Scope is bounded by the existing opt-in +# `5spot.finos.org/reclaim-agent: enabled` nodeSelector, so the cap is +# only granted on the small set of nodes the operator has explicitly +# armed. +AVD-KSV-0022 + # ───────────────────────────────────────────────────────────────────────────── # Dockerfile — Dockerfile, Dockerfile.chainguard # ───────────────────────────────────────────────────────────────────────────── diff --git a/Cargo.lock b/Cargo.lock index b4b5747..fd746a5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -212,6 +212,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + [[package]] name = "chrono" version = "0.4.44" @@ -552,6 +558,7 @@ dependencies = [ "kube-lease-manager", "lazy_static", "mockall", + "nix", "prometheus", "regex", "schemars", @@ -1299,6 +1306,15 @@ version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + [[package]] name = "mime" version = "0.3.17" @@ -1352,6 +1368,19 @@ dependencies = [ "syn", ] +[[package]] +name = "nix" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", + "memoffset", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" diff --git a/Cargo.toml b/Cargo.toml index 9c40551..bb7a005 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,6 +60,14 @@ lazy_static = "1.5" warp = { version = "0.4", default-features = false, features = ["server"] } toml = "1.1" +# Linux-only — used by the reclaim-agent's rung 2 netlink proc connector +# subscriber (`src/netlink_proc.rs`). The portable byte parsers in that +# module have no `nix` dep; only the actual socket / bind / send / recv +# dance does. Non-Linux builds compile the module's `Subscriber::new` to +# a stub that returns `NetlinkError::Unsupported`. +[target.'cfg(target_os = "linux")'.dependencies] +nix = { version = "0.30", default-features = false, features = ["socket", "net", "uio"] } + [[bin]] name = "5spot-reclaim-agent" path = "src/bin/reclaim_agent.rs" diff --git a/deploy/node-agent/daemonset.yaml b/deploy/node-agent/daemonset.yaml index 38d2709..ca153a4 100644 --- a/deploy/node-agent/daemonset.yaml +++ b/deploy/node-agent/daemonset.yaml @@ -129,12 +129,16 @@ spec: # - name: SKIP_HOST_ID_CHECK # value: "true" # Reading every `/proc//comm` + `cmdline` requires root; - # CAP_SYS_PTRACE is not strictly required for rung 1 but is kept - # dropped. CAP_NET_ADMIN will be re-added for rung 2 (netlink - # proc connector) as a separate change. See the pod-level - # securityContext comment block and `.trivyignore` (KSV-0012 / - # KSV-0116 / KSV-0118 / KSV-0105) for the runAsNonRoot=false - # rationale. + # CAP_SYS_PTRACE is not strictly required and is kept dropped. + # CAP_NET_ADMIN is added for rung 2: subscribing to the + # netlink proc connector (`NETLINK_CONNECTOR` socket family) + # requires it. Drop ALL first, then add only the one cap we + # need. See `.trivyignore` (KSV-0012 / KSV-0116 / KSV-0118 / + # KSV-0105 for runAsNonRoot=false; KSV-0102 for the + # CAP_NET_ADMIN add) for the architectural-necessity + # rationale. Operators who pin `--detector=poll` do not + # exercise the cap; it is granted at the pod level so the + # binary never has to fail open at runtime. securityContext: runAsUser: 0 runAsGroup: 0 @@ -145,6 +149,8 @@ spec: capabilities: drop: - ALL + add: + - NET_ADMIN seccompProfile: type: RuntimeDefault resources: diff --git a/docs/src/concepts/emergency-reclaim.md b/docs/src/concepts/emergency-reclaim.md index eeb6a90..c8a0444 100644 --- a/docs/src/concepts/emergency-reclaim.md +++ b/docs/src/concepts/emergency-reclaim.md @@ -1,6 +1,6 @@ # Emergency Reclaim (Process-Match Kill Switch) -**Status:** In progress — trigger contract, node-side agent (rung 1 `/proc` poll), CRD field, and opt-in DaemonSet shipped. Controller-side dispatch (Phase `EmergencyRemove`) is follow-up work. See the [roadmap](https://github.com/finos/5-spot/blob/main/docs/roadmaps/5spot-emergency-reclaim-by-process-match.md) for the current implementation cut. +**Status:** Shipped — trigger contract, node-side agent (both rung 1 `/proc` poll and rung 2 netlink proc connector), CRD field, controller-side `EmergencyRemove` dispatch, and opt-in DaemonSet are all live. See the [roadmap](https://github.com/finos/5-spot/blob/main/docs/roadmaps/5spot-emergency-reclaim-by-process-match.md) for the full design and the per-phase ship dates. --- @@ -248,7 +248,62 @@ A config with both match lists empty (or `killIfCommands: []`) is **inert** — ### Poll interval -The agent polls `/proc` every `poll_interval_ms` (default 250 ms — see `DEFAULT_POLL_INTERVAL_MS` in `src/reclaim_agent.rs`). This is the worst-case detection latency on rung 1 (`/proc` poll). A future rung 2 (netlink proc connector) will drop this to sub-millisecond latency with no code change to the matching logic. +The agent polls `/proc` every `poll_interval_ms` (default 250 ms — see `DEFAULT_POLL_INTERVAL_MS` in `src/reclaim_agent.rs`). This is the worst-case detection latency on rung 1 (`/proc` poll). + +### Detector — rung 1 (`/proc` poll) vs rung 2 (netlink) + +The agent ships two detection back-ends. Both produce identical [`Match`](https://github.com/finos/5-spot/blob/main/src/reclaim_agent.rs) values, both go through the same Node-PATCH path; the only difference is the event source. + +| Mode | What it does | Latency | Idle CPU | Linux only? | Extra capability | +|---|---|---|---|---|---| +| `poll` (rung 1) | Walks `/proc` every `poll_interval_ms` | up to one poll interval (250 ms default) | ~0 (one syscall per pid per tick) | No — works wherever `/proc` exists | None | +| `netlink` (rung 2) | Subscribes to the kernel's [proc connector](https://www.kernel.org/doc/html/latest/driver-api/connector.html) (`PROC_EVENT_EXEC`); kernel pushes one message per `exec(2)` | <10 ms (kernel-pushed) | sleeps until the kernel wakes it | **Yes** — Linux only | `CAP_NET_ADMIN` | + +Selection: + +``` +--detector=auto # default. Linux → netlink, anywhere else → poll. +--detector=netlink # force netlink. Errors loudly on non-Linux or + # without CAP_NET_ADMIN. +--detector=poll # force the rung-1 poll loop on any platform. +``` + +Override at deploy time via `RECLAIM_DETECTOR={auto|netlink|poll}` on the DaemonSet pod. + +**Tradeoff.** `netlink` is lower latency and lower idle CPU in steady state. Under heavy-exec workloads (`make -j32`, compile farms) it can be **more** expensive than `poll`, because rung 2 sees every short-lived process whereas rung 1 only sees those that survive to the next tick. If the operator's fleet runs exec storms, pin `--detector=poll` explicitly; otherwise keep the `auto` default. + +The DaemonSet manifest grants `CAP_NET_ADMIN` once at the pod level (see `deploy/node-agent/daemonset.yaml`), so an operator switching `--detector=poll → netlink` (or vice versa) does not need to re-roll the pod's security context. The cap is unused on `poll`; the kernel never sees a netlink syscall in that mode. + +### Kernel and capability requirements + +Rung 1 (`poll`) needs essentially nothing — `/proc` exists on every Linux kernel and the agent already has UID 0 + host `/proc` mount for [other reasons](#opt-in-installation). It also runs unmodified in a non-Linux container if you ever need to (the `Subscriber` constructor reports "unsupported" and the bin falls back automatically when `--detector=auto` is in effect). + +Rung 2 (`netlink`) has three runtime prerequisites: + +| Requirement | Provided by | What happens if missing | +|---|---|---| +| **Kernel built with `CONFIG_PROC_EVENTS=y`** | Distro kernel build (Debian, Ubuntu, RHEL/CentOS, Alpine, Talos, Bottlerocket, Chainguard host kernels — all default `y`) | The netlink socket opens cleanly but the kernel never pushes events. Agent sits idle, never matches. No userspace error to detect this. Diagnose by `zgrep CONFIG_PROC_EVENTS /proc/config.gz` on the host; mitigate by `--detector=poll`. | +| **`CAP_NET_ADMIN` on the agent container** | `deploy/node-agent/daemonset.yaml` `securityContext.capabilities.add: [NET_ADMIN]` | `Subscriber::new()` fails at startup with `EPERM`; the bin exits non-zero and `kubelet` restarts it (crash loop until the cap is granted). `--detector=poll` works without it. | +| **Linux kernel ≥ 2.6.15** | Effectively guaranteed on any cluster you'd actually deploy 5-Spot to (released 2006) | Same as missing `CONFIG_PROC_EVENTS` — silent. | + +The cap chain is straightforward: + +``` +deploy/node-agent/daemonset.yaml + └─ container.securityContext.capabilities + ├─ drop: [ALL] # default-deny + └─ add: [NET_ADMIN] # rung 2 only — rung 1 doesn't use it +``` + +This is the *only* added capability. UID 0 + host `/proc` + `hostPID` were already required for rung 1 (the agent has to read every process's `/proc//{comm,cmdline}` under `hidepid=2`). See [`.trivyignore`](https://github.com/finos/5-spot/blob/main/.trivyignore) for the architectural-necessity rationale on each Trivy / Semgrep finding the configuration trips. + +`CAP_NET_ADMIN` is broader than just netlink-connector access — it also grants ability to edit routes, iptables, etc. on the host network namespace. The risk is bounded by: + +1. **Opt-in nodeSelector** — the cap is granted only on Nodes labelled `5spot.finos.org/reclaim-agent: enabled`, which the controller stamps only when a `ScheduledMachine` on that Node has a non-empty `spec.killIfCommands`. A cluster that never uses `killIfCommands` never has the cap on any pod. +2. **Single-purpose binary** — the agent has no shell, no exec of children, no remote-control surface. Compromising it requires swapping the binary on disk, at which point an attacker could grant any cap they wanted anyway. +3. **Operator opt-out** — pin `--detector=poll` (env `RECLAIM_DETECTOR=poll`) and the cap is granted-but-unused. Operators with PSA `restricted` profiles or capability allowlists can deploy in this mode and never need the cap. + +See the [threat model](../security/threat-model.md#64-reclaim-agent-trust-boundary-tb5) for the full STRIDE table on the reclaim agent. --- @@ -444,7 +499,10 @@ mode. The agent reads the path from `MACHINE_ID_PATH` (default Both gaps are out of scope for a node-side agent; remediating them requires a stronger node-attestation primitive (TPM-backed identity, -SPIFFE SVIDs, etc.) — not in this controller's scope. +SPIFFE SVIDs, etc.) — not in this controller's scope. A phased design +for closing both gaps (TPM-quoted machine-id → SPIRE `tpm_devid` +attestation → controller-side SVID verification) is tracked in the +`5spot-strong-node-attestation` roadmap. --- diff --git a/docs/src/operations/configuration.md b/docs/src/operations/configuration.md index adae792..007b2e8 100644 --- a/docs/src/operations/configuration.md +++ b/docs/src/operations/configuration.md @@ -69,6 +69,92 @@ Non-leader replicas watch for leadership changes and take over automatically wit > **Note:** Leader election and multi-instance sharding (`OPERATOR_INSTANCE_COUNT > 1`) are alternative HA strategies. Use leader election for active/standby HA; use instance sharding to distribute load across all replicas. +## Reclaim agent (DaemonSet) + +The node-side `5spot-reclaim-agent` is a separate binary deployed +via DaemonSet (`deploy/node-agent/daemonset.yaml`). It has its own +flags and environment variables — distinct from the controller — and +is opt-in: nothing happens on a Node until the controller stamps the +`5spot.finos.org/reclaim-agent: enabled` label, which it does only +when a `ScheduledMachine` on that Node has a non-empty +`spec.killIfCommands`. See the [emergency-reclaim concept doc](../concepts/emergency-reclaim.md) +for the full design. + +### Environment variables + +| Variable | Default | Description | +|---|---|---| +| `NODE_NAME` | _(required, injected via downward API)_ | Name of the Node the agent is running on. The agent only PATCHes this Node. | +| `RECLAIM_PROC_ROOT` | `/proc` | Path the agent treats as `/proc`. Override for sandboxed/test runs only. | +| `RECLAIM_DETECTOR` | `auto` | Process-event source. `auto` picks `netlink` on Linux and `poll` elsewhere. See the [Detector](#detector) subsection below. | +| `MACHINE_ID_PATH` | `/etc/machine-id` | Path the agent reads for the host machine-id (host-identity verification, security-audit Phase 4). The DaemonSet mounts the host file at `/host/etc/machine-id` and sets this to that path. | +| `SKIP_HOST_ID_CHECK` | `false` | If `true`, skip the `Node.status.nodeInfo.machineID` cross-check before PATCH. Use only when `/etc/machine-id` is genuinely unavailable; production must stay strict. | + +### Command-line arguments + +```bash +5spot-reclaim-agent [OPTIONS] + +Options: + --proc-root Filesystem root mapped to /proc + [default: /proc] [env: RECLAIM_PROC_ROOT] + --node-name Node to annotate + [env: NODE_NAME] + --detector Process-event source: auto | netlink | poll + [default: auto] [env: RECLAIM_DETECTOR] + --machine-id-path Host machine-id file + [default: /etc/machine-id] [env: MACHINE_ID_PATH] + --skip-host-id-check Skip the host-identity cross-check before PATCH + (defence-in-depth; default off) + [env: SKIP_HOST_ID_CHECK] + --oneshot Run the detector once and exit + (one-shot tests / smoke verification) + -h, --help Print help + -V, --version Print version +``` + +### Detector + +Two detection back-ends ship with the agent. Both produce identical +matches and go through the same Node-PATCH path; only the event +source differs. + +| Mode | Mechanism | Latency | Idle CPU | Linux only? | Extra capability | +|---|---|---|---|---|---| +| `poll` | Walks `/proc` every `poll_interval_ms` | up to one poll interval (250 ms default) | ~0 | No | None | +| `netlink` | Subscribes to the kernel proc connector (`PROC_EVENT_EXEC`) | <10 ms (kernel-pushed) | sleeps until kernel wakes it | **Yes** | `CAP_NET_ADMIN` | + +`auto` (the default): +- Linux → `netlink` +- macOS / any non-Linux → `poll` (the netlink subscriber's + constructor returns `Unsupported` on those platforms) + +When to **pin `--detector=poll`** explicitly: + +- **Heavy-exec workloads** (`make -j32`, compile farms, CI workers) + — `netlink` sees every short-lived process even if it exits in + microseconds; `poll` only sees processes that survive to the + next tick. Under exec storms `poll` can be cheaper. +- **`CAP_NET_ADMIN` is unacceptable** in your environment (PSA + `restricted` profile, hardened cluster policy). The cap is + granted only on opted-in nodes via the DaemonSet's pod-level + `securityContext`, but you may have organisational reasons to + keep it dropped. +- **Kernel without `CONFIG_PROC_EVENTS`** (very rare; some + embedded / hardened distros). `netlink` socket opens cleanly + but no events are ever delivered. See + [troubleshooting](./troubleshooting.md#reclaim-agent-runs-but-never-observes-any-events). + +Override at deploy time: + +```bash +# Switch a running DaemonSet to poll mode (no pod restart needed — +# the agent watches its per-node ConfigMap, but env changes need a +# rollout; use kubectl set env to trigger one): +kubectl set env -n 5spot-system ds/5spot-reclaim-agent \ + RECLAIM_DETECTOR=poll +``` + ## ConfigMap Example ```yaml diff --git a/docs/src/operations/monitoring.md b/docs/src/operations/monitoring.md index 5257739..eb54dc5 100644 --- a/docs/src/operations/monitoring.md +++ b/docs/src/operations/monitoring.md @@ -51,24 +51,58 @@ Port: 8080 (default) ### Available Metrics -| Metric | Type | Description | -|--------|------|-------------| -| `five_spot_up` | Gauge | Operator health (1 = healthy) | -| `five_spot_reconciliations_total` | Counter | Total reconciliations by phase and result | -| `five_spot_reconciliation_duration_seconds` | Histogram | Reconciliation duration | -| `five_spot_machines_total` | Gauge | Total machines by phase | -| `five_spot_schedule_evaluations_total` | Counter | Schedule evaluations performed | +All metrics use the `fivespot_` prefix. The full list lives in +`src/metrics.rs`; the table below is the operator-facing summary. + +#### Reconciler + +| Metric | Type | Labels | Description | +|---|---|---|---| +| `fivespot_reconciliations_total` | Counter | `phase`, `result` | Reconciliation attempts | +| `fivespot_reconciliation_duration_seconds` | Histogram | `phase` | Reconciliation latency (s) | +| `fivespot_machines_active` | Gauge | — | Machines currently in `Active` phase | +| `fivespot_machines_by_phase` | Gauge | `phase` | Machines per lifecycle phase | +| `fivespot_schedule_evaluations_total` | Counter | `result` | Schedule evaluations by outcome | +| `fivespot_kill_switch_activations_total` | Gauge | — | Kill-switch activations | +| `fivespot_controller_info` | Gauge | `version`, `instance_id` | Always 1; carries label metadata | +| `fivespot_is_leader` | Gauge | — | 1 if this instance holds the leader lease | +| `fivespot_errors_total` | Counter | `error_type` | Errors by type | +| `fivespot_finalizer_cleanup_timeouts_total` | Counter | — | Finalizer cleanup timeouts (force-removed; possible orphans) | + +#### Node drain & eviction + +| Metric | Type | Labels | Description | +|---|---|---|---| +| `fivespot_node_drains_total` | Counter | `result` | Node drain attempts | +| `fivespot_pod_evictions_total` | Counter | `result` | Pod eviction attempts during drain | + +#### Emergency reclaim (process-match) + +| Metric | Type | Labels | Description | +|---|---|---|---| +| `fivespot_emergency_drain_duration_seconds` | Histogram | `outcome` | Wall-clock duration of emergency-reclaim drains. `outcome={success\|timeout\|error}` | +| `fivespot_emergency_reclaims_total` | Counter | `namespace`, `name` | Emergency-reclaim events fired per ScheduledMachine | +| `fivespot_rapid_re_reclaims_total` | Counter | `namespace`, `name` | `RapidReReclaim` warnings emitted per ScheduledMachine (loop-protection — see [Emergency reclaim concept](../concepts/emergency-reclaim.md)) | + +`fivespot_emergency_drain_duration_seconds` buckets are sized for the +60 s `EMERGENCY_DRAIN_TIMEOUT_SECS` ceiling: `[0.5, 1.0, 2.5, 5.0, +10.0, 15.0, 20.0, 30.0, 45.0, 60.0, 90.0]` seconds. The `outcome` +label lets dashboards compute success-only P95 and timeout-rate +side by side without mixing them in the same query. ### Labels Common labels across metrics: | Label | Description | -|-------|-------------| +|---|---| | `phase` | Machine lifecycle phase | -| `result` | Operation result (success, error) | -| `namespace` | Resource namespace | -| `name` | Resource name | +| `result` | Operation result (`success`, `failure`, `error`) | +| `outcome` | Outcome label on emergency-drain histogram (`success`, `timeout`, `error`) | +| `namespace` | Resource namespace (per-SM emergency-reclaim metrics only) | +| `name` | Resource name (per-SM emergency-reclaim metrics only) | +| `error_type` | Error category for `fivespot_errors_total` | +| `version` / `instance_id` | Controller info labels (carried on `fivespot_controller_info`) | ## ServiceMonitor (Prometheus Operator) @@ -95,36 +129,86 @@ spec: Example queries for a Grafana dashboard: -### Operator Health +### Operator health (leader presence) ```promql -five_spot_up +sum(fivespot_is_leader) ``` -### Machines by Phase +A value of 0 across all replicas is a paging condition — no instance +holds the lease, no reconciles are running. + +### Machines by phase + +```promql +sum by (phase) (fivespot_machines_by_phase) +``` + +### Reconciliation rate + +```promql +rate(fivespot_reconciliations_total[5m]) +``` + +### Reconciliation latency (P99) ```promql -sum by (phase) (five_spot_machines_total) +histogram_quantile(0.99, rate(fivespot_reconciliation_duration_seconds_bucket[5m])) ``` -### Reconciliation Rate +### Reconciliation failure rate + +```promql +rate(fivespot_reconciliations_total{result="failure"}[5m]) +``` + +### Emergency-reclaim drain — success P95 + +```promql +histogram_quantile( + 0.95, + rate(fivespot_emergency_drain_duration_seconds_bucket{outcome="success"}[10m]) +) +``` + +Operator SLO: P95 should sit well below 30 s on a healthy fleet. A +growing P95 signals workloads with bad +`terminationGracePeriodSeconds` defaults or PodDisruptionBudgets +that are clipping the drain. + +### Emergency-reclaim drain — timeout rate ```promql -rate(five_spot_reconciliations_total[5m]) +sum(rate(fivespot_emergency_drain_duration_seconds_count{outcome="timeout"}[10m])) ``` -### Reconciliation Latency (p99) +Any non-zero value means the 60 s `EMERGENCY_DRAIN_TIMEOUT_SECS` +ceiling is biting. Cross-reference with +`fivespot_pod_evictions_total{result="failure"}` to identify the +workloads that wouldn't evict. + +### Top emergency-reclaim offenders (per-SM rate) ```promql -histogram_quantile(0.99, rate(five_spot_reconciliation_duration_seconds_bucket[5m])) +topk(10, sum by (namespace, name) (rate(fivespot_emergency_reclaims_total[1h]))) ``` -### Error Rate +The 10 ScheduledMachines emergency-reclaimed most often in the last +hour. A SM that consistently shows up here is a candidate for +`killIfCommands` review — the user's workload may not be a true +"got my box back" emergency. + +### `RapidReReclaim` warnings ```promql -rate(five_spot_reconciliations_total{result="error"}[5m]) +sum by (namespace, name) (rate(fivespot_rapid_re_reclaims_total[1h])) ``` +Any non-zero rate is operator-actionable: a user is re-enabling a +SM whose conflicting process is still running. Trigger an alert and +follow the "Rapid re-reclaim loop" runbook in +[troubleshooting](./troubleshooting.md). + ## Structured Logging ### Log Format @@ -247,6 +331,9 @@ Event types and reasons: | Normal | `ScheduleDisabled` | Schedule disabled, machine deactivated | | Warning | `ReconcileFailed` | Unrecoverable error — machine in Error phase | | Warning | `KillSwitchActivated` | Emergency kill switch triggered | +| Warning | `EmergencyReclaim` | Reclaim-agent process-match fired; emergency-remove flow started | +| Warning | `EmergencyReclaimDisabledSchedule` | Step 5 of the flow: `spec.schedule.enabled=false` patched (load-bearing — breaks the eject→re-add→re-eject loop) | +| Warning | `RapidReReclaim` | ≥3 reclaims for the same SM within 10 min — the user is re-enabling without first stopping the conflicting process. See [troubleshooting](./troubleshooting.md) | Events are written to the `events.k8s.io/v1` API and are immutable once created, providing an auditable state-change trail (SOX §404 / NIST AU-2). @@ -258,29 +345,65 @@ Events are written to the `events.k8s.io/v1` API and are immutable once created, groups: - name: 5spot rules: - - alert: FiveSpotOperatorDown - expr: five_spot_up == 0 + - alert: FiveSpotNoLeader + expr: sum(fivespot_is_leader) == 0 for: 5m labels: severity: critical annotations: - summary: "5-Spot controller is down" - - - alert: FiveSpotHighErrorRate - expr: rate(five_spot_reconciliations_total{result="error"}[5m]) > 0.1 + summary: "No 5-Spot controller instance holds the leader lease" + + - alert: FiveSpotHighFailureRate + expr: rate(fivespot_reconciliations_total{result="failure"}[5m]) > 0.1 for: 10m labels: severity: warning annotations: - summary: "High reconciliation error rate" - + summary: "High reconciliation failure rate" + - alert: FiveSpotSlowReconciliation - expr: histogram_quantile(0.99, rate(five_spot_reconciliation_duration_seconds_bucket[5m])) > 30 + expr: histogram_quantile(0.99, rate(fivespot_reconciliation_duration_seconds_bucket[5m])) > 30 for: 15m labels: severity: warning annotations: - summary: "Slow reconciliation detected" + summary: "Slow reconciliation detected (P99 > 30 s)" + + - alert: FiveSpotEmergencyDrainTimeoutRising + expr: | + sum(rate(fivespot_emergency_drain_duration_seconds_count{outcome="timeout"}[10m])) > 0 + for: 10m + labels: + severity: warning + annotations: + summary: "Emergency-reclaim drains hitting the 60 s timeout ceiling" + description: | + One or more emergency-reclaim drains failed to evict all pods within + EMERGENCY_DRAIN_TIMEOUT_SECS (60 s). Cross-reference with + fivespot_pod_evictions_total{result="failure"} to identify the + offending workloads. + + - alert: FiveSpotRapidReReclaim + expr: | + sum by (namespace, name) (rate(fivespot_rapid_re_reclaims_total[15m])) > 0 + for: 5m + labels: + severity: warning + annotations: + summary: "ScheduledMachine {{ $labels.namespace }}/{{ $labels.name }} is in a rapid re-reclaim loop" + description: | + ≥3 emergency-reclaim events fired within 10 minutes for the same SM — + the user is re-enabling the schedule without first stopping the + conflicting process. See troubleshooting.md "Rapid re-reclaim loop" + runbook. + + - alert: FiveSpotFinalizerCleanupTimeouts + expr: rate(fivespot_finalizer_cleanup_timeouts_total[15m]) > 0 + for: 5m + labels: + severity: warning + annotations: + summary: "Finalizers being force-removed; possible orphan CAPI resources" ``` ## Related diff --git a/docs/src/operations/troubleshooting.md b/docs/src/operations/troubleshooting.md index b5c9ef5..4c9b94d 100644 --- a/docs/src/operations/troubleshooting.md +++ b/docs/src/operations/troubleshooting.md @@ -369,6 +369,142 @@ kubectl patch scheduledmachine/ --type merge \ 5. **The agent only reads `/proc//comm` (exact basename) and `/proc//cmdline` (substring).** A process whose `comm` is `java-wrapper` but argv starts with `/opt/jdk/bin/java ...` matches on `cmdline` (substring), not on `comm` (exact). +### Reclaim agent crash-loops with `EPERM` on netlink socket + +**Symptom:** `kubectl logs -n 5spot-system ` shows +the agent exiting at startup with an error like: + +``` +netlink i/o: Operation not permitted (os error 1) +``` + +or `Subscriber::new() failed: netlink i/o: EPERM`. + +**Cause:** `--detector=netlink` is in effect (the default on Linux), +but the pod's container does not have `CAP_NET_ADMIN`. The kernel +refuses to bind an `AF_NETLINK` socket to the `CN_IDX_PROC` +multicast group without it. + +**Fix (one of):** + +1. **Confirm the cap is in the manifest** — + `deploy/node-agent/daemonset.yaml` should grant `NET_ADMIN` + under the container's `securityContext.capabilities.add`. If a + downstream patch removed it, restore: + + ```bash + kubectl get ds -n 5spot-system 5spot-reclaim-agent \ + -o jsonpath='{.spec.template.spec.containers[0].securityContext.capabilities}' + # Expect: {"add":["NET_ADMIN"],"drop":["ALL"]} + ``` + +2. **Force `poll` mode** if granting `CAP_NET_ADMIN` is not + acceptable in your environment. The agent's rung 1 (`/proc` + poll) detects the same processes at higher latency (≤250 ms vs + <10 ms) and zero added capability: + + ```bash + kubectl set env -n 5spot-system ds/5spot-reclaim-agent \ + RECLAIM_DETECTOR=poll + ``` + + See [Detector — rung 1 vs rung 2](../concepts/emergency-reclaim.md) + for the tradeoff. + +3. **Pod Security Admission may strip capabilities** under the + `restricted` profile. The DaemonSet's pod-level + `securityContext` already drops `runAsNonRoot=false` and other + restricted-profile bumps; if PSA is enforcing `restricted` on + `5spot-system`, downgrade the namespace label to `baseline` or + add an exemption. The reclaim-agent's privileges are + architecturally bounded by its opt-in nodeSelector. + +### Reclaim agent runs but never observes any events + +**Symptom:** Agent pod is `Running`, no errors in logs, host has +exec activity (e.g. user is launching processes), but `kubectl +describe node ` never shows the reclaim annotations even with +a known-matching `killIfCommands` list. + +**Possible cause: kernel built without `CONFIG_PROC_EVENTS`.** This +is rare but possible on heavily-stripped distro kernels (some +embedded / hardened images). The netlink socket opens cleanly but +the kernel never pushes events. + +**Diagnose:** + +```bash +# Inspect the running kernel's config (if /proc/config.gz is present) +zgrep CONFIG_PROC_EVENTS /proc/config.gz +# Or grep the booted kernel's config file +grep CONFIG_PROC_EVENTS /boot/config-$(uname -r) +# Expect: CONFIG_PROC_EVENTS=y +``` + +**Fix:** switch to `--detector=poll` (or `RECLAIM_DETECTOR=poll`). +The rung-1 poll does not depend on the kernel feature. + +### Rapid re-reclaim loop — `RapidReReclaim` Warning Event + +**Symptom:** `kubectl get events` shows multiple `RapidReReclaim` +warnings on the same ScheduledMachine within minutes; the +controller log carries the corresponding warning lines; the +`fivespot_rapid_re_reclaims_total{namespace, name}` metric is +incrementing. + +**Meaning:** The controller has observed ≥3 emergency-reclaim +events for this SM within 10 minutes +(`RAPID_RE_RECLAIM_THRESHOLD = 3`, +`RAPID_RE_RECLAIM_WINDOW_SECS = 600`). Almost always: a user is +re-enabling the schedule (`spec.schedule.enabled=true`) before +they have stopped the conflicting process the agent is matching +on. The agent fires again the moment the Node rejoins, and the +loop repeats. + +**Diagnose:** + +```bash +# 1. Find the offending SM and pull its emergency-reclaim history +kubectl get events --field-selector reason=EmergencyReclaim \ + --sort-by='.lastTimestamp' \ + -o jsonpath='{range .items[*]}{.lastTimestamp} {.involvedObject.namespace}/{.involvedObject.name} {.message}{"\n"}{end}' + +# 2. Check what `killIfCommands` patterns are configured +kubectl get scheduledmachine -o jsonpath='{.spec.killIfCommands}' + +# 3. Walk the agent log to see what was matched on the most recent fire +kubectl logs -n 5spot-system -l app=5spot-reclaim-agent --tail=100 \ + | jq -c 'select(.fields.matched_pattern)' +``` + +**Fix paths (operator-side):** + +- **Confirm with the user** that the conflicting process is gone + before they re-enable. The whole point of the agent is the user's + workstation got their attention; `ps aux | grep ` on the + host (or `kubectl debug node/ -it --image=alpine` if + no host shell) is the canonical confirmation. +- **Adjust `killIfCommands`** if the pattern is over-matching + (e.g. `java` matches every JVM, including the user's IDE that + may be acceptable). Switch to the more specific + `match_argv_substrings` form via direct ConfigMap edit, or narrow + the basename. +- **Suppress the loop temporarily** by clearing the spec: + ```bash + kubectl patch scheduledmachine --type=merge \ + -p '{"spec":{"killIfCommands":null}}' + ``` + This tears down the per-node ConfigMap + label, which evicts the + agent pod from the Node, which means no further reclaim fires + even if the user starts a matching process. + +**Why no auto-resume:** the controller deliberately does not flip +`spec.schedule.enabled` back to `true` automatically when the +matching process exits. Doing so would invite races between the +Node rejoining and the agent restarting, and would silently mask +the user behaviour the warning is trying to surface. Re-enable is +explicit by design. + ### `EmergencyReclaim` event fires but schedule is not disabled **Symptom:** The `EmergencyReclaim` event is on the `ScheduledMachine`, but `spec.schedule.enabled` is still `true`. diff --git a/docs/src/reference/cli.md b/docs/src/reference/cli.md index e0b6444..b5382c2 100644 --- a/docs/src/reference/cli.md +++ b/docs/src/reference/cli.md @@ -72,6 +72,79 @@ RUST_LOG=debug 5spot RUST_LOG=five_spot=debug,kube=info 5spot ``` +## `5spot-reclaim-agent` + +Node-side DaemonSet binary that watches `/proc` (or the kernel's +proc connector via netlink) for processes matching the per-node +`killIfCommands` list and PATCHes the local Node with reclaim +annotations to trigger the controller's emergency-reclaim flow. See +the [emergency-reclaim concept doc](../concepts/emergency-reclaim.md) +for the full design and the +[configuration reference](../operations/configuration.md#reclaim-agent-daemonset) +for env-var equivalents. + +### Synopsis + +```bash +5spot-reclaim-agent [OPTIONS] +``` + +### Options + +| Option | Default | Description | +|---|---|---| +| `--node-name` | _(required)_ | Name of the Node to annotate. Inject via downward API: `valueFrom.fieldRef.fieldPath: spec.nodeName` | +| `--detector ` | `auto` | Process-event source. `auto` picks `netlink` on Linux, `poll` elsewhere | +| `--proc-root ` | `/proc` | Filesystem root mapped to `/proc`. Override only for sandboxed/test runs | +| `--machine-id-path ` | `/etc/machine-id` | Host machine-id file (host-identity verification) | +| `--skip-host-id-check` | `false` | Skip the `Node.status.nodeInfo.machineID` cross-check before PATCH. Defence-in-depth — leave off in production | +| `--oneshot` | | Run the detector once and exit (smoke tests / one-off invocations) | +| `--help`, `-h` | | Print help | +| `--version`, `-V` | | Print version | + +### Environment variables + +| Variable | CLI Equivalent | +|---|---| +| `NODE_NAME` | `--node-name` | +| `RECLAIM_DETECTOR` | `--detector` | +| `RECLAIM_PROC_ROOT` | `--proc-root` | +| `MACHINE_ID_PATH` | `--machine-id-path` | +| `SKIP_HOST_ID_CHECK` | `--skip-host-id-check` | + +### Examples + +```bash +# Production default (auto-detect → netlink on Linux) +5spot-reclaim-agent --node-name "$(hostname)" + +# Force /proc poll (no CAP_NET_ADMIN required, ≤250 ms detection) +5spot-reclaim-agent --node-name node-01 --detector poll + +# Local smoke test against a fake /proc tree +5spot-reclaim-agent --node-name dev-01 \ + --proc-root /tmp/fake-proc \ + --skip-host-id-check \ + --oneshot +``` + +### Exit codes + +| Code | Meaning | +|---|---| +| 0 | Reclaim annotation written successfully (or already present — idempotent) | +| 1 | Unrecoverable error (kube client init, host-identity check failed, netlink subscriber failed, etc.) | + +### Capability requirements (rung 2 / netlink only) + +Rung 2 (`--detector=netlink`) requires `CAP_NET_ADMIN` on the agent +container. Granted at the pod level in +`deploy/node-agent/daemonset.yaml`. Operators who refuse the cap +can pin `--detector=poll` (rung 1) which needs no extra capability. +See the +[detector tradeoff table](../operations/configuration.md#detector) +for guidance. + ## Utility Binaries ### crdgen diff --git a/docs/src/security/threat-model.md b/docs/src/security/threat-model.md index e7ff826..c74118f 100644 --- a/docs/src/security/threat-model.md +++ b/docs/src/security/threat-model.md @@ -120,10 +120,21 @@ flowchart LR Nodes["Kubernetes Nodes\n(cordon · drain · evict)"] end + subgraph TB5["TB5 · Node-Side Reclaim Agent"] + direction TB + Agent["5spot-reclaim-agent\n(opt-in DaemonSet pod)"] + AgentSA["Per-pod ServiceAccount\n(nodes:get,patch · cm:get,list,watch)"] + Caps["UID 0 + hostPID + CAP_NET_ADMIN\n+ host /proc + /etc/machine-id"] + Agent --- AgentSA + Agent --- Caps + end + User -->|"F1 · HTTPS + RBAC"| TB1 TB1 -->|"F2 · watch"| TB2 TB2 -->|"F3 · create/delete"| TB3 TB2 -->|"F4 · cordon/evict"| TB4 + TB2 -->|"F5 · per-node CM project"| TB5 + TB5 -->|"F6 · annotate own Node"| TB4 ``` --- @@ -233,6 +244,52 @@ flowchart LR --- +### 6.4 Reclaim Agent (Trust Boundary: TB5) + +The node-side `5spot-reclaim-agent` DaemonSet runs on opted-in +worker nodes only and signals the controller via three Node +annotations. It is the lowest-trust component in the system: pod +runs as UID 0 with `hostPID: true`, host `/proc` mount, host +`/etc/machine-id` mount, and `CAP_NET_ADMIN` (for rung 2 netlink +proc connector). Per-pod `ServiceAccount` grants only +`nodes: get,patch` cluster-wide and `configmaps: get,list,watch` +in `5spot-system`. + +#### Spoofing +| ID | Threat | Likelihood | Impact | Status | +|---|---|---|---|---| +| S5 | Attacker with `update daemonsets` overrides the agent pod's `NODE_NAME` env var, causing the agent to PATCH reclaim annotations on a victim Node it isn't actually running on | Low (precondition is cluster-admin-equivalent in most clusters) | High (would trigger emergency-remove on innocent host) | **Mitigated** — host-identity verification (`read_host_machine_id` + `compare_machine_ids`, security-audit Phase 4, 2026-04-26): agent reads `/etc/machine-id` from a `hostPath.type: File` mount at startup and refuses to PATCH a Node whose `status.nodeInfo.machineID` doesn't match. `--skip-host-id-check` opt-out exists but defaults off. | +| S6 | Compromised pod somewhere else in the cluster spoofs the reclaim annotations directly on a Node, triggering an emergency-remove the agent never decided on | Low | High | **Partially mitigated** — `nodes: patch` is broadly held in most clusters (kubelet itself has it); the controller treats the annotation as authoritative. Mitigation: monitor `5spot.finos.org/reclaim-requested` PATCH events via API audit logs, alert when the field manager is anything other than `5spot-reclaim-agent`. | + +#### Tampering +| ID | Threat | Likelihood | Impact | Status | +|---|---|---|---|---| +| T12 | Compromised agent escalates to other Nodes via stolen ServiceAccount token | Low (per-pod SA, scoped credentials) | Medium | **Mitigated** — RBAC: `nodes: get,patch` cluster-wide (no `list/watch` — can't enumerate). Host-identity check refuses any PATCH against a Node whose `machineID` doesn't match the agent's own `/etc/machine-id`. | +| T13 | `CAP_NET_ADMIN` (granted for rung 2 netlink) is misused to send rogue netlink messages on other subsystems | Low | Low–Medium | **Accepted** — `CAP_NET_ADMIN` is Linux's coarsest-grained networking cap and grants more than just netlink connector access (route table edits, iptables, etc.). Scope-bounded in two ways: (1) the cap is added only on opted-in Nodes via the existing `5spot.finos.org/reclaim-agent: enabled` nodeSelector, so only the small set of Nodes with `killIfCommands` set ever sees it; (2) the agent binary is single-purpose with no shell, no exec of children — a compromise would need an attacker to swap the binary, at which point they could grant themselves any cap they wanted. Operators who refuse the cap can pin `--detector=poll` (rung 1) which needs no extra capability. | + +#### Repudiation +| ID | Threat | Likelihood | Impact | Status | +|---|---|---|---|---| +| R1 | Agent PATCH on the Node is indistinguishable from controller writes in audit logs | Low | Low | **Mitigated** — the agent uses field manager `5spot-reclaim-agent`; the controller uses the distinct field manager `5spot-controller-reclaim-agent`. Every Node PATCH carries the field manager in `managedFields`, so audit attribution is unambiguous (NIST AU-2 / AU-10). | + +#### Information Disclosure +| ID | Threat | Likelihood | Impact | Status | +|---|---|---|---|---| +| I3 | Reading host `/proc//cmdline` for matching exposes argv strings (which can include passwords / tokens passed on the command line) to the agent's logs | Low (logs are operator-owned) | Low | **Accepted** — the agent logs `matched_pattern` (the operator-supplied pattern) and the matched pid, NOT the full cmdline. Patterns themselves are operator-authored config, not user input. | + +#### Denial of Service +| ID | Threat | Likelihood | Impact | Status | +|---|---|---|---|---| +| D8 | User repeatedly re-enables a SM whose conflicting process is still running, generating an ejection loop | Medium | Low | **Mitigated** — loop-protection: ≥3 reclaims for the same SM within 10 minutes emits a `RapidReReclaim` Warning Event and bumps `fivespot_rapid_re_reclaims_total{namespace, name}`. Operator runbook in [troubleshooting](../operations/troubleshooting.md) covers the response. | +| D9 | Heavy-exec workload (`make -j32`) overwhelms the netlink subscriber's recv loop with `PROC_EVENT_EXEC` traffic | Low | Low | **Accepted** — agent CPU is bounded by container limits (`50m`); `ENOBUFS` on the netlink socket surfaces as `NetlinkError::Io` and is logged, not silently dropped. Operators who anticipate exec storms should pin `--detector=poll`. | + +#### Elevation of Privilege +| ID | Threat | Likelihood | Impact | Status | +|---|---|---|---|---| +| E4 | `hostPID: true` lets the agent see all host PIDs — could be abused to extract data from another container's `/proc//environ` if the agent is compromised | Low | Medium | **Accepted** — `hostPID` is architecturally required (the agent's job is to read host process state). Mitigated by: opt-in `nodeSelector` (only on nodes with `killIfCommands`), single-purpose binary with no shell, `readOnlyRootFilesystem: true`, drop ALL caps + add only `NET_ADMIN`, `seccompProfile: RuntimeDefault`. Trivy / Semgrep suppressions in `.trivyignore` document the architectural-necessity reasoning. | + +--- + ## 7. Mitigations Summary ### Implemented (as of 2026-04-08) @@ -250,6 +307,11 @@ flowchart LR | CPU/memory resource limits | `deploy/deployment/deployment.yaml` | D1 — resource exhaustion | | PDB 429 handling in `evict_pod` | `src/reconcilers/helpers.rs` | D5 — API rate limit crash | | Cosign image signing in release CI | `.github/workflows/release.yaml` | S3 — image tampering | +| Host-identity verification (machine-id cross-check) | `src/reclaim_agent.rs::compare_machine_ids` + `deploy/node-agent/daemonset.yaml` (`/etc/machine-id` mount) | S5 — DaemonSet-tampering NODE_NAME spoof | +| Distinct field managers (`5spot-reclaim-agent` vs `5spot-controller-reclaim-agent`) | All Node PATCH paths | R1 — audit attribution | +| Per-pod `ServiceAccount` with `nodes: get,patch` only (no `list/watch`) | `deploy/node-agent/rbac.yaml` | T12 — agent enumerates other Nodes | +| Opt-in `5spot.finos.org/reclaim-agent: enabled` nodeSelector | `deploy/node-agent/daemonset.yaml` | T13 — `CAP_NET_ADMIN` scope-bounding | +| Loop-protection (`RapidReReclaim` warning + counter) | `src/loop_protection.rs` + `src/reconcilers/helpers.rs::handle_emergency_remove` | D8 — re-enable loop | ### Deployment-Layer Controls (operator responsibility) diff --git a/src/bin/reclaim_agent.rs b/src/bin/reclaim_agent.rs index ee2416f..f9f27dc 100644 --- a/src/bin/reclaim_agent.rs +++ b/src/bin/reclaim_agent.rs @@ -40,9 +40,10 @@ use std::sync::Arc; use std::time::Duration; use anyhow::{anyhow, bail, Context as _, Result}; -use clap::Parser; +use clap::{Parser, ValueEnum}; +use five_spot::netlink_proc::{NetlinkError, ProcEvent, Subscriber as NetlinkSubscriber}; use five_spot::reclaim_agent::{ - already_requested, build_patch_body, compare_machine_ids, configmap_to_config, + already_requested, build_patch_body, compare_machine_ids, configmap_to_config, match_pid, read_host_machine_id, scan_proc, Config, Match, }; use futures::StreamExt; @@ -75,6 +76,44 @@ const FIELD_MANAGER: &str = "5spot-reclaim-agent"; /// agent exerts essentially zero CPU. const IDLE_WAKEUP_SECS: u64 = 30; +/// CLI value for `--detector`. Resolved to a [`Detector`] via +/// [`DetectorFlag::resolve`] so `auto` is platform-aware. +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +enum DetectorFlag { + /// Pick `netlink` on Linux, `poll` elsewhere. + Auto, + /// Subscribe to the netlink proc connector (Linux only). + Netlink, + /// Walk `/proc` every `poll_interval_ms` (rung-1 fallback). + Poll, +} + +impl DetectorFlag { + /// Resolve `auto` to a concrete detector based on the build target. + fn resolve(self) -> Detector { + match self { + DetectorFlag::Netlink => Detector::Netlink, + DetectorFlag::Poll => Detector::Poll, + DetectorFlag::Auto => { + if cfg!(target_os = "linux") { + Detector::Netlink + } else { + Detector::Poll + } + } + } + } +} + +/// Resolved detector choice — what the scanner loop actually runs. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Detector { + /// Netlink proc connector. + Netlink, + /// `/proc` poll. + Poll, +} + #[derive(Parser, Debug)] #[clap( name = "5spot-reclaim-agent", @@ -102,6 +141,21 @@ struct Cli { #[clap(long, env = "MACHINE_ID_PATH", default_value = DEFAULT_MACHINE_ID_PATH)] machine_id_path: PathBuf, + /// Process-event source. `netlink` (default on Linux) subscribes + /// to the kernel's proc connector for sub-10 ms detection. `poll` + /// (the rung-1 fallback, default on non-Linux) walks `/proc` every + /// `poll_interval_ms`. `auto` picks `netlink` on Linux and `poll` + /// elsewhere. + /// + /// Tradeoffs: `netlink` requires `CAP_NET_ADMIN` on the agent pod + /// but has lower idle CPU and tighter latency. Under heavy-exec + /// workloads (`make -j32`, compile farms) it can be more expensive + /// than `poll`, which only sees processes that survive a tick. + /// See `docs/src/concepts/emergency-reclaim.md` for the full + /// comparison. + #[clap(long, env = "RECLAIM_DETECTOR", default_value = "auto", value_enum)] + detector: DetectorFlag, + /// Skip the host-identity cross-check before patching the Node. /// /// Default: false (strict). Before each PATCH the agent fetches the @@ -185,15 +239,37 @@ async fn main() -> Result<()> { // resubscribe inside `kube::runtime::watcher`. let watcher_handle = tokio::spawn(run_config_watcher(client, cm_name, tx)); - let scanner_result = run_scanner( - &nodes, - &cli.node_name, - &cli.proc_root, - rx, - cli.oneshot, - host_machine_id.as_deref(), - ) - .await; + let detector = cli.detector.resolve(); + info!( + detector = ?detector, + cli_value = ?cli.detector, + "detector selected", + ); + + let scanner_result = match detector { + Detector::Poll => { + run_scanner( + &nodes, + &cli.node_name, + &cli.proc_root, + rx, + cli.oneshot, + host_machine_id.as_deref(), + ) + .await + } + Detector::Netlink => { + run_netlink_scanner( + &nodes, + &cli.node_name, + &cli.proc_root, + rx, + cli.oneshot, + host_machine_id.as_deref(), + ) + .await + } + }; watcher_handle.abort(); scanner_result } @@ -345,6 +421,151 @@ async fn run_scanner( } } +/// Netlink-driven detection loop (rung 2). +/// +/// Subscribes once to the kernel proc connector and blocks on the +/// socket for every `exec(2)`. The socket is opened *only* when the +/// shared config transitions to `Some` (no point holding kernel +/// resources while idle); a config-clear closes the subscriber and we +/// return to waiting for arming. +/// +/// On every `ProcEvent::Exec` we resolve the pid via the existing +/// rung-1 [`match_pid`] helper — this is the architectural promise of +/// the two-rung ladder: only the *event source* changes between rungs, +/// match logic and Node-PATCH path are shared. +/// +/// Falls back to `Err` immediately on subscriber-construction failure +/// (e.g. running on macOS, missing `CAP_NET_ADMIN`, or kernel built +/// without `CONFIG_PROC_EVENTS`). The bin's `main` then propagates the +/// error so the operator sees it; rung-1 fallback is selected via +/// `--detector=poll`, not via silent in-process degradation. +#[allow(clippy::too_many_lines)] +async fn run_netlink_scanner( + nodes: &Api, + node_name: &str, + proc_root: &Path, + mut rx: tokio::sync::watch::Receiver>, + oneshot: bool, + host_machine_id: Option<&str>, +) -> Result<()> { + loop { + // Wait until the per-node ConfigMap arms us (or exit if + // oneshot and nothing is armed). + let cfg = rx.borrow().clone(); + let Some(cfg) = cfg else { + if oneshot { + warn!("oneshot mode + netlink: no config present — exiting non-zero"); + return Err(anyhow!("no configmap observed during oneshot run")); + } + tokio::select! { + res = rx.changed() => { + if res.is_err() { + info!("config channel closed — exiting"); + return Ok(()); + } + continue; + } + () = tokio::time::sleep(Duration::from_secs(IDLE_WAKEUP_SECS)) => continue, + } + }; + + // Open the subscriber. Surface platform / capability errors at + // this boundary so the operator sees them in the agent's first + // log lines after arming. + let mut sub = match NetlinkSubscriber::new() { + Ok(s) => s, + Err(NetlinkError::Unsupported) => { + bail!( + "netlink detector requested but the netlink proc connector is not \ + supported on this platform — pass --detector=poll to use the /proc \ + fallback or run on Linux" + ); + } + Err(e) => { + error!(error = %e, "failed to open netlink subscriber"); + bail!( + "open netlink proc-connector subscriber (CAP_NET_ADMIN required and \ + CONFIG_PROC_EVENTS=y in kernel): {e}" + ); + } + }; + info!( + commands = ?cfg.match_commands, + substrings = ?cfg.match_argv_substrings, + "netlink subscriber armed — waiting for kernel exec events", + ); + + let proc_root = proc_root.to_path_buf(); + let cfg_arc = Arc::new(cfg); + + // The recv loop must run on a blocking thread — `recv(2)` on the + // netlink socket is synchronous and would otherwise pin the + // tokio worker. We spawn it and await its result on the + // current task; cancellation is via the channel `rx.changed()` + // arm of the select below. + let recv_handle = tokio::task::spawn_blocking({ + let proc_root = proc_root.clone(); + let cfg_arc = cfg_arc.clone(); + move || -> Result> { + loop { + match sub.next_event() { + Ok(Some(ProcEvent::Exec { pid, tgid: _ })) => { + if let Some(m) = match_pid(&proc_root, pid, &cfg_arc) { + return Ok(Some(m)); + } + } + Ok(Some(ProcEvent::Other { what })) => { + tracing::trace!(what = format_args!("{what:#x}"), "ignored proc_event"); + } + Ok(None) => { + // Frame parsed-and-dropped (already logged). + continue; + } + Err(e) => return Err(anyhow!("netlink recv: {e}")), + } + } + } + }); + + let outcome: Result> = tokio::select! { + joined = recv_handle => match joined { + Ok(res) => res, + Err(join_err) => Err(anyhow!("netlink recv task panicked: {join_err}")), + }, + res = rx.changed() => { + if res.is_err() { + info!("config channel closed — exiting"); + return Ok(()); + } + // Config changed (likely cleared). Drop the subscriber + // by returning Ok(None); the outer loop will re-evaluate. + Ok(None) + } + }; + + match outcome { + Ok(Some(m)) => { + info!( + pid = m.pid, + pattern = %m.matched_pattern, + "netlink match → annotating node", + ); + annotate_node(nodes, node_name, &m, host_machine_id).await?; + return Ok(()); + } + Ok(None) => { + if oneshot { + warn!("oneshot mode + netlink: config cleared before any match — exiting non-zero"); + return Err(anyhow!("config cleared during oneshot run")); + } + // Loop: re-evaluate the (possibly new) config. + continue; + } + Err(e) => return Err(e), + } + } +} + /// PATCH the Node with reclaim annotations. /// /// Before the PATCH (when `host_machine_id` is `Some`) the agent fetches diff --git a/src/constants.rs b/src/constants.rs index 41ace26..3858d73 100644 --- a/src/constants.rs +++ b/src/constants.rs @@ -148,6 +148,36 @@ pub const REASON_EMERGENCY_RECLAIM_DISABLED_SCHEDULE: &str = "EmergencyReclaimDi /// when the Machine is deleted. pub const EMERGENCY_DRAIN_TIMEOUT_SECS: u64 = 60; +/// Number of recent reclaim events that constitute a "rapid" loop. +/// +/// When `≥ RAPID_RE_RECLAIM_THRESHOLD` reclaims fire on the same SM +/// within [`RAPID_RE_RECLAIM_WINDOW_SECS`], the controller emits a +/// `RapidReReclaim` Warning Event recommending the operator stop the +/// conflicting process before re-enabling. +/// +/// Threshold of 3 catches the canonical "user re-enables a SM whose +/// JVM is still running, gets ejected, re-enables, gets ejected again" +/// trio without flagging legitimate re-enables that just happen to +/// land near a prior reclaim. +pub const RAPID_RE_RECLAIM_THRESHOLD: usize = 3; + +/// Sliding-window length (10 minutes) for rapid-re-reclaim detection. +/// Reclaims older than this drop off the tracked list and no longer +/// count toward the threshold. +pub const RAPID_RE_RECLAIM_WINDOW_SECS: i64 = 600; + +/// Cap on how many reclaim timestamps we hold in memory per SM, to +/// bound the worst-case allocation. Larger than +/// [`RAPID_RE_RECLAIM_THRESHOLD`] so a brief storm above the threshold +/// is still observable in logs without resizing. +pub const RAPID_RE_RECLAIM_MAX_TRACKED: usize = 10; + +/// Kubernetes Event reason emitted on the `ScheduledMachine` when the +/// rapid-re-reclaim threshold is crossed. Operators see this in +/// `kubectl describe scheduledmachine` and on the `RapidReReclaim` +/// metric. +pub const REASON_RAPID_RE_RECLAIM: &str = "RapidReReclaim"; + // ============================================================================ // Timing Constants (in seconds) // ============================================================================ diff --git a/src/lib.rs b/src/lib.rs index a68a374..6c8593a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -11,7 +11,12 @@ //! - [`crd`] — `ScheduledMachine` CRD type definitions (source of truth for YAML generation) //! - [`health`] — HTTP health and readiness server //! - [`labels`] — standard Kubernetes label helpers +//! - [`loop_protection`] — rapid-re-reclaim detection helpers (pure; +//! used by the reconciler to throttle warning events) //! - [`metrics`] — Prometheus metric definitions and recording helpers +//! - [`netlink_proc`] — Linux netlink proc connector subscriber (rung 2 of +//! the reclaim-agent detection ladder; portable byte parsers, Linux-only +//! socket impl) //! - [`reconcilers`] — reconciliation logic and controller context pub mod auto_vex_presence; @@ -20,7 +25,9 @@ pub mod constants; pub mod crd; pub mod health; pub mod labels; +pub mod loop_protection; pub mod metrics; +pub mod netlink_proc; pub mod reclaim_agent; pub mod reconcilers; diff --git a/src/loop_protection.rs b/src/loop_protection.rs new file mode 100644 index 0000000..f26780d --- /dev/null +++ b/src/loop_protection.rs @@ -0,0 +1,81 @@ +// Copyright (c) 2025 Erick Bourgeois, firestoned +// SPDX-License-Identifier: Apache-2.0 +//! # Rapid-re-reclaim loop protection +//! +//! Detects the "user re-enables a `ScheduledMachine` whose conflicting +//! process is still running" failure mode (`5spot-emergency-reclaim-by-process-match.md` +//! Phase 4 follow-up). When the controller observes ≥ +//! [`crate::constants::RAPID_RE_RECLAIM_THRESHOLD`] reclaim events for +//! the same SM within +//! [`crate::constants::RAPID_RE_RECLAIM_WINDOW_SECS`], it emits a +//! `RapidReReclaim` Warning Event and bumps +//! `fivespot_rapid_re_reclaims_total{namespace, name}`. +//! +//! ## Design — in-memory only, deliberately +//! +//! The tracker lives in [`crate::reconcilers::Context`] as a +//! `Mutex>>` rather than in CRD status. A +//! controller restart resets the count: "rapid" then means "rapid since +//! the most recent restart." This trades persistence for simplicity: +//! +//! - No CRD schema bump (which would require coordinated rollout in a +//! regulated environment). +//! - No status-patch round trip on every reclaim. +//! - The signal is operator-actionable on a *trend* basis (Prometheus +//! counter), so a single missed warning does not lose audit value. +//! +//! ## Pure helpers +//! +//! All decisions live in pure functions on `&mut VecDeque` so the +//! tests can drive every branch without a controller or a clock — +//! callers pass an explicit `now` and the helpers do the rest. + +use crate::constants::{ + RAPID_RE_RECLAIM_MAX_TRACKED, RAPID_RE_RECLAIM_THRESHOLD, RAPID_RE_RECLAIM_WINDOW_SECS, +}; +use chrono::{DateTime, Duration, Utc}; +use std::collections::VecDeque; + +/// Drop reclaim timestamps older than the configured window from the +/// front of the deque. Pure; no I/O. +/// +/// Operates on the deque in place — caller passes the same deque +/// repeatedly across reconciles. Newer entries are at the back; once +/// the front is within the window, all subsequent entries are too +/// (timestamps are appended in monotonic order). +pub fn prune_old_reclaim_events(events: &mut VecDeque>, now: DateTime) { + let cutoff = now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS); + while events.front().is_some_and(|t| *t < cutoff) { + events.pop_front(); + } +} + +/// Append `now` to `events`, prune entries older than the window, and +/// cap the deque at [`RAPID_RE_RECLAIM_MAX_TRACKED`] (oldest dropped). +/// Returns `true` when the post-append count meets or exceeds +/// [`RAPID_RE_RECLAIM_THRESHOLD`] — the caller's signal to emit the +/// `RapidReReclaim` Warning Event and bump the metric. +pub fn record_emergency_reclaim_event( + events: &mut VecDeque>, + now: DateTime, +) -> bool { + prune_old_reclaim_events(events, now); + events.push_back(now); + while events.len() > RAPID_RE_RECLAIM_MAX_TRACKED { + events.pop_front(); + } + events.len() >= RAPID_RE_RECLAIM_THRESHOLD +} + +/// Read-only check: would `events` (pruned to the window) constitute +/// a rapid-re-reclaim loop right now? Pure; used by tests and by +/// debug-log decisions that should not mutate the deque. +#[must_use] +pub fn should_warn_rapid_re_reclaim(events: &VecDeque>, now: DateTime) -> bool { + let cutoff = now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS); + events.iter().filter(|t| **t >= cutoff).count() >= RAPID_RE_RECLAIM_THRESHOLD +} + +#[cfg(test)] +#[path = "loop_protection_tests.rs"] +mod tests; diff --git a/src/loop_protection_tests.rs b/src/loop_protection_tests.rs new file mode 100644 index 0000000..237a580 --- /dev/null +++ b/src/loop_protection_tests.rs @@ -0,0 +1,205 @@ +// Copyright (c) 2025 Erick Bourgeois, firestoned +// SPDX-License-Identifier: Apache-2.0 +//! Tests for the rapid-re-reclaim loop-protection helpers. + +#[cfg(test)] +#[allow(clippy::module_inception)] +mod tests { + use super::super::*; + use crate::constants::{RAPID_RE_RECLAIM_MAX_TRACKED, RAPID_RE_RECLAIM_WINDOW_SECS}; + use chrono::{Duration, TimeZone, Utc}; + use std::collections::VecDeque; + + fn t(secs: i64) -> chrono::DateTime { + Utc.timestamp_opt(secs, 0).single().expect("valid ts") + } + + // ───────────────────────────────────────────────────────────────── + // prune_old_reclaim_events + // ───────────────────────────────────────────────────────────────── + + #[test] + fn prune_empty_is_noop() { + let mut events: VecDeque> = VecDeque::new(); + prune_old_reclaim_events(&mut events, t(1_000_000)); + assert!(events.is_empty()); + } + + #[test] + fn prune_keeps_all_events_within_window() { + let now = t(1_000_000); + let mut events: VecDeque<_> = [ + now - Duration::seconds(1), + now - Duration::seconds(60), + now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS - 1), + ] + .into_iter() + .collect(); + prune_old_reclaim_events(&mut events, now); + assert_eq!(events.len(), 3, "all entries within window must survive"); + } + + #[test] + fn prune_drops_all_when_every_entry_is_older() { + let now = t(1_000_000); + let mut events: VecDeque<_> = [ + now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS + 1), + now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS + 60), + ] + .into_iter() + .collect(); + prune_old_reclaim_events(&mut events, now); + assert!(events.is_empty(), "all stale entries must be dropped"); + } + + #[test] + fn prune_drops_only_stale_front_entries() { + let now = t(1_000_000); + let mut events: VecDeque<_> = [ + now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS + 100), // stale + now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS + 1), // stale + now - Duration::seconds(60), // fresh + now - Duration::seconds(1), // fresh + ] + .into_iter() + .collect(); + prune_old_reclaim_events(&mut events, now); + assert_eq!(events.len(), 2); + } + + // ───────────────────────────────────────────────────────────────── + // record_emergency_reclaim_event + // ───────────────────────────────────────────────────────────────── + + #[test] + fn record_first_event_does_not_trigger_warning() { + let mut events = VecDeque::new(); + let breach = record_emergency_reclaim_event(&mut events, t(1_000_000)); + assert!(!breach, "single event must not breach threshold of 3"); + assert_eq!(events.len(), 1); + } + + #[test] + fn record_two_events_does_not_trigger_warning() { + let mut events = VecDeque::new(); + record_emergency_reclaim_event(&mut events, t(1_000_000)); + let breach = record_emergency_reclaim_event(&mut events, t(1_000_001)); + assert!(!breach, "two events must not breach threshold of 3"); + } + + #[test] + fn record_three_events_within_window_triggers_warning() { + let mut events = VecDeque::new(); + let now = 1_000_000; + record_emergency_reclaim_event(&mut events, t(now)); + record_emergency_reclaim_event(&mut events, t(now + 10)); + let breach = record_emergency_reclaim_event(&mut events, t(now + 20)); + assert!(breach, "third event within window must breach"); + } + + #[test] + fn record_three_events_spread_outside_window_does_not_trigger() { + let mut events = VecDeque::new(); + let base = 1_000_000; + record_emergency_reclaim_event(&mut events, t(base)); + record_emergency_reclaim_event(&mut events, t(base + RAPID_RE_RECLAIM_WINDOW_SECS + 1)); + // First event is now stale and pruned; only two remain after + // the second push. Third push has count = 2 (one stale, one + // fresh, plus this one = 2 fresh). + let breach = record_emergency_reclaim_event( + &mut events, + t(base + 2 * (RAPID_RE_RECLAIM_WINDOW_SECS + 1)), + ); + assert!( + !breach, + "events outside window must be pruned and not count toward threshold" + ); + } + + #[test] + fn record_caps_at_max_tracked_dropping_oldest() { + let mut events = VecDeque::new(); + let base = 1_000_000; + // Push more than the cap. + for i in 0..(RAPID_RE_RECLAIM_MAX_TRACKED + 5) { + // Use sub-second offsets via integer seconds (chrono timestamps are + // 1-second granular here). Increment by 1 to keep ordering monotone. + #[allow(clippy::cast_possible_wrap)] + record_emergency_reclaim_event(&mut events, t(base + i as i64)); + } + assert_eq!( + events.len(), + RAPID_RE_RECLAIM_MAX_TRACKED, + "deque must be capped at RAPID_RE_RECLAIM_MAX_TRACKED entries" + ); + // The oldest survivor must be the (5)th push since we dropped 5 from the front. + let oldest = *events.front().unwrap(); + let expected_oldest = t(base + 5); + assert_eq!(oldest, expected_oldest); + } + + // ───────────────────────────────────────────────────────────────── + // should_warn_rapid_re_reclaim + // ───────────────────────────────────────────────────────────────── + + #[test] + fn should_warn_returns_false_for_empty_deque() { + let events = VecDeque::new(); + assert!(!should_warn_rapid_re_reclaim(&events, t(1_000_000))); + } + + #[test] + fn should_warn_returns_false_below_threshold() { + let now = t(1_000_000); + let events: VecDeque<_> = [now - Duration::seconds(1), now - Duration::seconds(10)] + .into_iter() + .collect(); + assert!(!should_warn_rapid_re_reclaim(&events, now)); + } + + #[test] + fn should_warn_returns_true_at_threshold() { + let now = t(1_000_000); + let events: VecDeque<_> = [ + now - Duration::seconds(1), + now - Duration::seconds(10), + now - Duration::seconds(20), + ] + .into_iter() + .collect(); + assert!(should_warn_rapid_re_reclaim(&events, now)); + } + + #[test] + fn should_warn_only_counts_events_within_window() { + let now = t(1_000_000); + let events: VecDeque<_> = [ + now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS + 100), // stale + now - Duration::seconds(60), // fresh + now - Duration::seconds(1), // fresh + ] + .into_iter() + .collect(); + // 2 within window, threshold is 3 — no warn. + assert!(!should_warn_rapid_re_reclaim(&events, now)); + } + + #[test] + fn should_warn_does_not_mutate_input() { + let now = t(1_000_000); + let events: VecDeque<_> = [ + now - Duration::seconds(RAPID_RE_RECLAIM_WINDOW_SECS + 100), + now - Duration::seconds(60), + now - Duration::seconds(1), + ] + .into_iter() + .collect(); + let len_before = events.len(); + let _ = should_warn_rapid_re_reclaim(&events, now); + assert_eq!( + events.len(), + len_before, + "should_warn must take &VecDeque, not mutate" + ); + } +} diff --git a/src/metrics.rs b/src/metrics.rs index 4ccb28c..f5dc0f9 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -26,6 +26,9 @@ //! | `fivespot_errors_total` | Counter | Errors by type | //! | `fivespot_node_drains_total` | Counter | Node drain attempts by outcome | //! | `fivespot_pod_evictions_total` | Counter | Pod eviction attempts by outcome | +//! | `fivespot_emergency_drain_duration_seconds` | Histogram | Emergency-reclaim drain wall-clock by outcome | +//! | `fivespot_emergency_reclaims_total` | Counter | Emergency-reclaim events per SM (namespace, name) | +//! | `fivespot_rapid_re_reclaims_total` | Counter | RapidReReclaim warnings per SM (namespace, name) | use std::sync::LazyLock; @@ -270,6 +273,79 @@ pub static NODE_DRAINS_TOTAL: LazyLock = LazyLock::new(|| { }) }); +/// Duration of emergency-reclaim drain operations in seconds. +/// +/// Histogram observed once per `EmergencyRemove` flow, labelled by +/// outcome (`success` / `timeout` / `error`). Buckets are sized for the +/// `EMERGENCY_DRAIN_TIMEOUT_SECS = 60` ceiling — anything past 60 s is a +/// flow-killing timeout that the controller force-deletes through anyway. +/// +/// Operator SLO: P95 of `outcome="success"` should sit well below 30 s on +/// a healthy fleet. A growing P95 signals workloads with bad +/// `terminationGracePeriodSeconds` defaults or PodDisruptionBudgets that +/// are clipping the drain. A non-zero `outcome="timeout"` rate means the +/// 60 s ceiling is biting — investigate the workloads on the affected +/// nodes via the `fivespot_pod_evictions_total{result="failure"}` metric +/// alongside. +pub static EMERGENCY_DRAIN_DURATION_SECONDS: LazyLock = LazyLock::new(|| { + register_histogram_vec!( + "fivespot_emergency_drain_duration_seconds", + "Wall-clock duration of emergency-reclaim node drains, by outcome", + &["outcome"], + vec![0.5, 1.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, 45.0, 60.0, 90.0] + ) + .unwrap_or_else(|e| { + eprintln!("WARN: Failed to register fivespot_emergency_drain_duration_seconds: {e}"); + fallback_histogram_vec( + "fivespot_emergency_drain_duration_seconds", + "Wall-clock duration of emergency-reclaim node drains, by outcome", + &["outcome"], + vec![0.5, 1.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, 45.0, 60.0, 90.0], + ) + }) +}); + +/// Total emergency-reclaim events recorded per ScheduledMachine, +/// labelled by namespace + name. Used by the loop-protection logic to +/// detect rapid re-fires (a user re-enabling a SM whose conflicting +/// process is still running) — alert when the rate per SM exceeds the +/// `RAPID_RE_RECLAIM_THRESHOLD` constant within a 10-minute window. +pub static EMERGENCY_RECLAIMS_TOTAL: LazyLock = LazyLock::new(|| { + register_counter_vec!( + "fivespot_emergency_reclaims_total", + "Total emergency-reclaim events fired per ScheduledMachine", + &["namespace", "name"] + ) + .unwrap_or_else(|e| { + eprintln!("WARN: Failed to register fivespot_emergency_reclaims_total: {e}"); + fallback_counter_vec( + "fivespot_emergency_reclaims_total", + "Total emergency-reclaim events fired per ScheduledMachine", + &["namespace", "name"], + ) + }) +}); + +/// Total rapid re-reclaim warnings emitted, labelled by namespace + +/// name. Operator-actionable signal — every increment corresponds to a +/// `RapidReReclaim` Warning Event on the SM whose user has re-enabled +/// the schedule without first stopping the conflicting process. +pub static RAPID_RE_RECLAIMS_TOTAL: LazyLock = LazyLock::new(|| { + register_counter_vec!( + "fivespot_rapid_re_reclaims_total", + "Total RapidReReclaim warnings emitted per ScheduledMachine", + &["namespace", "name"] + ) + .unwrap_or_else(|e| { + eprintln!("WARN: Failed to register fivespot_rapid_re_reclaims_total: {e}"); + fallback_counter_vec( + "fivespot_rapid_re_reclaims_total", + "Total RapidReReclaim warnings emitted per ScheduledMachine", + &["namespace", "name"], + ) + }) +}); + /// Pod evictions during node drain pub static POD_EVICTIONS_TOTAL: LazyLock = LazyLock::new(|| { register_counter_vec!( @@ -364,6 +440,59 @@ pub fn record_pod_eviction(success: bool) { POD_EVICTIONS_TOTAL.with_label_values(&[result]).inc(); } +/// Outcome label for [`record_emergency_drain`]. Stable strings — +/// changing them is a Prometheus-dashboard breaking change. +#[derive(Clone, Copy, Debug)] +pub enum EmergencyDrainOutcome { + /// Drain returned cleanly within the timeout. + Success, + /// Drain hit `EMERGENCY_DRAIN_TIMEOUT_SECS` and was force-completed + /// without all pods evicting. + Timeout, + /// Drain returned an error other than timeout (apiserver 5xx, etc.). + Error, +} + +impl EmergencyDrainOutcome { + fn as_str(self) -> &'static str { + match self { + EmergencyDrainOutcome::Success => "success", + EmergencyDrainOutcome::Timeout => "timeout", + EmergencyDrainOutcome::Error => "error", + } + } +} + +/// Observe one emergency-drain duration sample. Called once per +/// `EmergencyRemove` flow regardless of outcome — operators slice by +/// the `outcome` label to compute success-only P95 vs timeout rate +/// separately. +pub fn record_emergency_drain(duration_secs: f64, outcome: EmergencyDrainOutcome) { + EMERGENCY_DRAIN_DURATION_SECONDS + .with_label_values(&[outcome.as_str()]) + .observe(duration_secs); +} + +/// Increment the per-SM emergency-reclaim counter. Called once per +/// successful entry into the `EmergencyRemove` flow (after the +/// idempotent recovery handler short-circuits, so a controller restart +/// mid-flow does not double-count). +pub fn record_emergency_reclaim(namespace: &str, name: &str) { + EMERGENCY_RECLAIMS_TOTAL + .with_label_values(&[namespace, name]) + .inc(); +} + +/// Increment the per-SM rapid-re-reclaim warning counter. Called when +/// the loop-protection logic detects ≥`RAPID_RE_RECLAIM_THRESHOLD` +/// reclaims for the same SM within `RAPID_RE_RECLAIM_WINDOW_SECS` and +/// emits a `RapidReReclaim` Warning Event. +pub fn record_rapid_re_reclaim(namespace: &str, name: &str) { + RAPID_RE_RECLAIMS_TOTAL + .with_label_values(&[namespace, name]) + .inc(); +} + /// Record a finalizer-cleanup timeout (force-remove path). /// /// Operators should treat any non-zero rate as a signal that orphan CAPI diff --git a/src/metrics_tests.rs b/src/metrics_tests.rs index b263e62..bc30c66 100644 --- a/src/metrics_tests.rs +++ b/src/metrics_tests.rs @@ -67,3 +67,64 @@ fn test_set_leader_status() { set_leader_status(true); set_leader_status(false); } + +#[test] +fn test_record_emergency_drain_observes_each_outcome() { + let before_success = EMERGENCY_DRAIN_DURATION_SECONDS + .with_label_values(&["success"]) + .get_sample_count(); + record_emergency_drain(2.5, EmergencyDrainOutcome::Success); + let after_success = EMERGENCY_DRAIN_DURATION_SECONDS + .with_label_values(&["success"]) + .get_sample_count(); + assert_eq!(after_success, before_success + 1); + + let before_timeout = EMERGENCY_DRAIN_DURATION_SECONDS + .with_label_values(&["timeout"]) + .get_sample_count(); + record_emergency_drain(60.0, EmergencyDrainOutcome::Timeout); + let after_timeout = EMERGENCY_DRAIN_DURATION_SECONDS + .with_label_values(&["timeout"]) + .get_sample_count(); + assert_eq!(after_timeout, before_timeout + 1); + + let before_error = EMERGENCY_DRAIN_DURATION_SECONDS + .with_label_values(&["error"]) + .get_sample_count(); + record_emergency_drain(0.1, EmergencyDrainOutcome::Error); + let after_error = EMERGENCY_DRAIN_DURATION_SECONDS + .with_label_values(&["error"]) + .get_sample_count(); + assert_eq!(after_error, before_error + 1); +} + +#[test] +fn test_record_emergency_reclaim_increments_per_sm_counter() { + let ns = "test-emergency-reclaim"; + let name = "sm-counter-fixture"; + let before = EMERGENCY_RECLAIMS_TOTAL + .with_label_values(&[ns, name]) + .get(); + record_emergency_reclaim(ns, name); + record_emergency_reclaim(ns, name); + let after = EMERGENCY_RECLAIMS_TOTAL + .with_label_values(&[ns, name]) + .get(); + assert!( + (after - before - 2.0).abs() < f64::EPSILON, + "EMERGENCY_RECLAIMS_TOTAL must increment exactly twice: before={before} after={after}" + ); +} + +#[test] +fn test_record_rapid_re_reclaim_increments_per_sm_counter() { + let ns = "test-rapid-re-reclaim"; + let name = "sm-loop-fixture"; + let before = RAPID_RE_RECLAIMS_TOTAL.with_label_values(&[ns, name]).get(); + record_rapid_re_reclaim(ns, name); + let after = RAPID_RE_RECLAIMS_TOTAL.with_label_values(&[ns, name]).get(); + assert!( + (after - before - 1.0).abs() < f64::EPSILON, + "RAPID_RE_RECLAIMS_TOTAL must increment by one: before={before} after={after}" + ); +} diff --git a/src/netlink_proc.rs b/src/netlink_proc.rs new file mode 100644 index 0000000..c218d66 --- /dev/null +++ b/src/netlink_proc.rs @@ -0,0 +1,746 @@ +// Copyright (c) 2025 Erick Bourgeois, firestoned +// SPDX-License-Identifier: Apache-2.0 +//! # Netlink proc connector — rung 2 of the reclaim-agent detection ladder +//! +//! Subscribes to the Linux kernel's process-event stream via the +//! [proc connector] over `NETLINK_CONNECTOR`. On every `exec(2)` the +//! kernel pushes a `proc_event { what = PROC_EVENT_EXEC, … }` message; +//! the subscriber parses it and yields a [`ProcEvent::Exec`]. The +//! reclaim-agent's main loop then resolves the pid via the existing +//! [`crate::reclaim_agent::match_pid`] helper, reusing all of rung 1's +//! match logic — only the *event source* differs. +//! +//! See `~/dev/roadmaps/completed-5spot-emergency-reclaim-by-process-match.md` +//! Phase 2.c and GitHub issue #40 for the design rationale and the +//! deferred-then-shipped history. +//! +//! ## Layering +//! +//! - **Portable parsers** — [`parse_cn_msg`], [`parse_proc_event`], +//! the [`ProcEvent`] enum, and [`NetlinkError`] are pure +//! byte-shovellers and have no platform `cfg`. They run on macOS in +//! the test suite via hand-crafted byte sequences (see +//! [`netlink_proc_tests`](../netlink_proc_tests/index.html)). +//! - **Subscriber** — [`Subscriber`] opens a netlink socket, joins the +//! `CN_IDX_PROC` multicast group, and sends the +//! `PROC_CN_MCAST_LISTEN` control message. The implementation is +//! `#[cfg(target_os = "linux")]`; on every other platform the +//! constructor returns [`NetlinkError::Unsupported`] so the rest of +//! the binary still compiles and reports a useful error at startup +//! when the operator passes `--detector=netlink` on a non-Linux box. +//! +//! The split is deliberate: tests for the (load-bearing) wire-format +//! decisions run anywhere `cargo test` works, while the (small, +//! mechanical) socket-shuffle code is the only part that needs a +//! Linux runtime to verify. The runtime test lives in +//! `tests/integration_netlink_proc.rs` and is `#[ignore]` so a +//! macOS dev workflow stays hermetic. +//! +//! ## Kernel + capability requirements +//! +//! - **Kernel:** Linux ≥ 2.6.15 with `CONFIG_PROC_EVENTS=y`. Default +//! `y` on every mainstream distro (Debian, Ubuntu, RHEL/CentOS, +//! Alpine with stock kernel, Talos, Bottlerocket, Chainguard +//! host kernels). +//! - **Capability:** `CAP_NET_ADMIN` on the agent process. Granted +//! at the container level via `securityContext.capabilities.add` +//! in `deploy/node-agent/daemonset.yaml` (drop ALL, then add the +//! one cap we need). +//! - **Architecture:** identical netlink ABI on x86_64, aarch64, +//! armv7, ppc64le — endian + alignment match the kernel that +//! emits the events because we run on the same host. Cross-arch +//! container builds work without per-target code. +//! - **Not supported:** macOS, Windows, illumos, BSDs. Subscriber +//! constructor returns [`NetlinkError::Unsupported`] on every +//! non-Linux target. +//! +//! ## Wire layout (kernel-emitted, native-endian on every Linux target) +//! +//! ```text +//! nlmsghdr (16 bytes) — read+skipped by Subscriber::next_event +//! nlmsg_len u32 — total length including this header +//! nlmsg_type u16 — NLMSG_DONE for proc-connector frames +//! nlmsg_flags u16 +//! nlmsg_seq u32 +//! nlmsg_pid u32 +//! cn_msg (20 bytes) — parsed by parse_cn_msg +//! id.idx u32 — CN_IDX_PROC = 1 +//! id.val u32 — CN_VAL_PROC = 1 +//! seq u32 +//! ack u32 +//! len u16 — payload length (24 for exec events) +//! flags u16 +//! proc_event (16 + variant) — parsed by parse_proc_event +//! what u32 — PROC_EVENT_EXEC = 0x00000002 +//! cpu u32 — emitting CPU; ignored at our level +//! timestamp_ns u64 — kernel monotonic; not wall-clock +//! +//! exec.pid i32 — process_pid (single thread) +//! exec.tgid i32 — process_tgid (thread-group leader) +//! ``` +//! +//! Each kernel message arrives as one datagram on the netlink +//! socket. See [Limitations](#limitations) below for the +//! multi-message-per-datagram corner case. +//! +//! ## Examples +//! +//! Parsing a hand-crafted exec event (works on any platform — useful +//! for tests, debugging, and replay tooling): +//! +//! ``` +//! use five_spot::netlink_proc::{parse_cn_msg, parse_proc_event, ProcEvent, CN_IDX_PROC, CN_VAL_PROC}; +//! +//! // Build a `cn_msg` carrying an exec event for pid 4242. +//! let mut frame = Vec::new(); +//! frame.extend_from_slice(&CN_IDX_PROC.to_le_bytes()); // id.idx +//! frame.extend_from_slice(&CN_VAL_PROC.to_le_bytes()); // id.val +//! frame.extend_from_slice(&[0u8; 8]); // seq + ack +//! frame.extend_from_slice(&24u16.to_le_bytes()); // payload length +//! frame.extend_from_slice(&[0u8; 2]); // flags +//! // proc_event payload (24 bytes: what + cpu + timestamp + pid + tgid) +//! frame.extend_from_slice(&0x0000_0002u32.to_le_bytes()); // what = PROC_EVENT_EXEC +//! frame.extend_from_slice(&[0u8; 12]); // cpu (4) + timestamp_ns (8) +//! frame.extend_from_slice(&4242u32.to_le_bytes()); // exec.pid +//! frame.extend_from_slice(&4242u32.to_le_bytes()); // exec.tgid +//! +//! let payload = parse_cn_msg(&frame).expect("well-formed cn_msg"); +//! let evt = parse_proc_event(payload).expect("well-formed proc_event"); +//! assert_eq!(evt, ProcEvent::Exec { pid: 4242, tgid: 4242 }); +//! ``` +//! +//! Subscribing on a real Linux node (omitted as a doctest because it +//! requires `CAP_NET_ADMIN` and a kernel — see +//! `tests/integration_netlink_proc.rs` for a runnable test): +//! +//! ```ignore +//! use five_spot::netlink_proc::{Subscriber, ProcEvent}; +//! +//! let mut sub = Subscriber::new()?; // requires CAP_NET_ADMIN +//! loop { +//! match sub.next_event()? { +//! Some(ProcEvent::Exec { pid, .. }) => { +//! // Resolve via /proc//{comm,cmdline} using the +//! // existing rung-1 helper — match logic is shared. +//! if let Some(m) = five_spot::reclaim_agent::match_pid( +//! std::path::Path::new("/proc"), pid, &cfg +//! ) { +//! // ... PATCH the Node with reclaim annotations. +//! } +//! } +//! Some(ProcEvent::Other { .. }) | None => continue, +//! } +//! } +//! # Ok::<(), five_spot::netlink_proc::NetlinkError>(()) +//! ``` +//! +//! ## Limitations +//! +//! - **One message per `recv` call.** [`Subscriber::next_event`] +//! currently treats the receive buffer as a single +//! `nlmsghdr + cn_msg + proc_event` triple. Under sustained burst +//! load the kernel *may* pack multiple connector messages into a +//! single datagram, in which case messages after the first in the +//! buffer are silently dropped. In practice connector messages +//! ride one-per-datagram on multicast sockets; the rung-1 `/proc` +//! poll backstops any miss. If you observe drops in production, +//! refactor to iterate via `nlmsg_len` until the buffer is +//! exhausted. +//! - **Single subscriber per process.** The kernel happily delivers +//! to multiple subscribers, but the agent only opens one socket +//! per pod (one pod per node, per the DaemonSet). No coordination +//! needed. +//! - **No replay on subscriber restart.** Events are pushed live; a +//! crash + restart of the agent loses any events that fired during +//! the gap. Combined with the rung-1 fallback (which polls every +//! 250 ms regardless), the worst case after a restart is one poll +//! interval of detection latency, not a missed reclaim. +//! - **`PROC_EVENT_FORK` is not consumed.** Only `EXEC` events trigger +//! match attempts. A long-lived process that `fork`s a worker that +//! never `execve`s won't be matched on the worker's pid (the worker +//! inherits the parent's `comm` until it execs, so `match_pid` +//! would still match if the parent did). Acceptable: every match +//! target this codebase cares about (JVMs, IDEs, compilers) execs +//! on startup. +//! +//! ## Cancellation and clean shutdown +//! +//! Drop the [`Subscriber`] to close the socket; the kernel auto-removes +//! us from the multicast group. The reclaim-agent uses +//! `tokio::task::spawn_blocking` to host the synchronous `recv` +//! loop, and cancels via the existing per-node ConfigMap watch +//! channel — see `src/bin/reclaim_agent.rs::run_netlink_scanner`. +//! +//! [proc connector]: https://www.kernel.org/doc/html/latest/driver-api/connector.html + +// ============================================================================ +// Constants — wire values (kernel `linux/connector.h`, `linux/cn_proc.h`) +// +// These are the *kernel* wire values, not Rust idioms. Renaming any of +// them silently breaks every prior recording / replay tool that +// depends on the byte layout. Matched 1:1 against the upstream +// headers; bump only when the kernel does. +// ============================================================================ + +/// Connector index for the proc-event subsystem. +/// +/// From `linux/connector.h`: +/// `#define CN_IDX_PROC 0x1`. Identifies the proc-connector subsystem +/// in the `cn_msg.id.idx` field on every event we receive and in the +/// control message we send to start the subscription. +pub const CN_IDX_PROC: u32 = 1; + +/// Connector value for the proc-event subsystem. +/// +/// From `linux/connector.h`: +/// `#define CN_VAL_PROC 0x1`. The other half of the connector id +/// pair; together with [`CN_IDX_PROC`] uniquely names the +/// proc-connector subsystem. +pub const CN_VAL_PROC: u32 = 1; + +/// `PROC_EVENT_EXEC` discriminant — `0x0000_0002` from +/// `linux/cn_proc.h`. The kernel writes this to `proc_event.what` on +/// every successful `execve(2)`. +/// +/// Visibility is `pub(crate)` rather than `pub` so the test file in +/// `netlink_proc_tests.rs` can synthesize exec frames without +/// re-deriving the value, while external callers go through the +/// typed [`ProcEvent::Exec`] variant instead of pattern-matching +/// raw bytes. +pub(crate) const PROC_EVENT_EXEC_RAW: u32 = 0x0000_0002; + +/// `PROC_CN_MCAST_LISTEN` operation — the payload of the control +/// message we send to the kernel after binding the netlink socket +/// to start receiving events. +/// +/// Without this control message, binding alone does not produce +/// events: the kernel ignores subscribers that have not explicitly +/// opted in. The complementary op is `PROC_CN_MCAST_IGNORE = 0x2` +/// (unsubscribe); we don't use it because dropping the [`Subscriber`] +/// closes the socket and the kernel auto-removes us. +#[cfg(target_os = "linux")] +const PROC_CN_MCAST_LISTEN: u32 = 0x0000_0001; + +/// Length of a `cn_msg` header in bytes. +/// +/// Layout: `id.idx` (4) + `id.val` (4) + `seq` (4) + `ack` (4) + +/// `len` (2) + `flags` (2) = 20 bytes. The `len` field is the +/// declared length of the payload that follows; the parser +/// cross-checks it against the buffer's actual remaining size. +const CN_MSG_HEADER_LEN: usize = 20; + +/// Length of the fixed portion of a `proc_event` in bytes. +/// +/// Layout: `what` (4) + `cpu` (4) + `timestamp_ns` (8) = 16 bytes. +/// Variant-specific payload follows immediately after; for the only +/// variant we decode (`PROC_EVENT_EXEC`) the additional payload is +/// [`PROC_EVENT_EXEC_PAYLOAD_LEN`] bytes. +const PROC_EVENT_HEADER_LEN: usize = 16; + +/// Length of the `exec` variant payload in bytes. +/// +/// Layout: `process_pid` (4) + `process_tgid` (4) = 8 bytes. Other +/// `proc_event` variants (FORK, EXIT, COMM, etc.) carry different +/// payloads; we don't decode them — the parser classifies them as +/// [`ProcEvent::Other`] and the caller ignores them. +const PROC_EVENT_EXEC_PAYLOAD_LEN: usize = 8; + +// ============================================================================ +// Public types +// ============================================================================ + +/// A decoded proc-connector event. +/// +/// Only [`ProcEvent::Exec`] carries pid/tgid because that is the only +/// event the agent acts on. Every other `what` value is preserved as +/// [`ProcEvent::Other`] so the caller can log / count it without the +/// parser silently dropping kernel data. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ProcEvent { + /// `PROC_EVENT_EXEC` — a process called `execve(2)`. + Exec { + /// Process pid (`process_pid` field). + pid: u32, + /// Thread-group leader pid (`process_tgid` field). Equal to + /// `pid` for single-threaded execs, may differ for + /// multi-threaded ones. + tgid: u32, + }, + /// Any other event type. The raw `what` discriminant is preserved + /// so callers can log / classify without re-reading the buffer. + Other { + /// Raw `what` value as written by the kernel. + what: u32, + }, +} + +/// Errors returned by the netlink-proc-connector parsers and subscriber. +#[derive(Debug, thiserror::Error)] +pub enum NetlinkError { + /// The buffer was shorter than the parser needed at this offset. + /// The struct fields name the failure precisely so a malformed + /// frame can be diagnosed without re-reading. + #[error("buffer too short: expected at least {expected} bytes, got {actual}")] + Truncated { + /// Minimum byte count the parser required. + expected: usize, + /// Actual byte count the parser was given. + actual: usize, + }, + /// The `cn_msg` carried a connector id that doesn't belong to the + /// proc-event subsystem. Either the socket was bound to a wrong + /// group or a kernel/wire bug delivered a message we never + /// subscribed to. Either way: drop and log. + #[error("invalid cn_msg id: idx={idx} val={val} (expected idx=1 val=1)")] + InvalidId { + /// Observed `cn_msg.id.idx`. + idx: u32, + /// Observed `cn_msg.id.val`. + val: u32, + }, + /// I/O failure on the netlink socket (Linux subscriber only). + #[cfg(target_os = "linux")] + #[error("netlink i/o: {0}")] + Io(#[from] std::io::Error), + /// The current platform does not support the netlink proc + /// connector. Returned by [`Subscriber::new`] on non-Linux so the + /// binary still links and reports a useful error at startup. + #[error("netlink proc connector is not supported on this platform (Linux-only)")] + Unsupported, +} + +// ============================================================================ +// Portable parsers (no #[cfg]; runnable on macOS, ARM, x86_64) +// ============================================================================ + +/// Parse a `cn_msg` frame and return its payload slice. +/// +/// Validates the connector id `(idx=1, val=1)` and the declared +/// payload length against the buffer's actual size. Does **not** parse +/// the payload itself — pass the returned slice to [`parse_proc_event`] +/// (or any future event parser) to interpret it. +/// +/// # Errors +/// - [`NetlinkError::Truncated`] when the buffer is shorter than the +/// 20-byte header or shorter than `header.len + 20`. +/// - [`NetlinkError::InvalidId`] when the connector id does not match +/// the proc-event subsystem. +pub fn parse_cn_msg(bytes: &[u8]) -> Result<&[u8], NetlinkError> { + if bytes.len() < CN_MSG_HEADER_LEN { + return Err(NetlinkError::Truncated { + expected: CN_MSG_HEADER_LEN, + actual: bytes.len(), + }); + } + let idx = u32::from_le_bytes(bytes[0..4].try_into().unwrap()); + let val = u32::from_le_bytes(bytes[4..8].try_into().unwrap()); + if idx != CN_IDX_PROC || val != CN_VAL_PROC { + return Err(NetlinkError::InvalidId { idx, val }); + } + let len = u16::from_le_bytes(bytes[16..18].try_into().unwrap()) as usize; + let needed = CN_MSG_HEADER_LEN + len; + if bytes.len() < needed { + return Err(NetlinkError::Truncated { + expected: needed, + actual: bytes.len(), + }); + } + Ok(&bytes[CN_MSG_HEADER_LEN..needed]) +} + +/// Parse a `proc_event` payload (as returned by [`parse_cn_msg`]) and +/// classify it. +/// +/// Only `PROC_EVENT_EXEC` is decoded into [`ProcEvent::Exec`]; every +/// other `what` value becomes [`ProcEvent::Other`] so callers can +/// observe but ignore. We never silently drop kernel data. +/// +/// # Errors +/// - [`NetlinkError::Truncated`] when the buffer is shorter than the +/// 16-byte fixed header, or — for an exec event — shorter than the +/// 16-byte header plus the 8-byte exec payload. +pub fn parse_proc_event(bytes: &[u8]) -> Result { + if bytes.len() < PROC_EVENT_HEADER_LEN { + return Err(NetlinkError::Truncated { + expected: PROC_EVENT_HEADER_LEN, + actual: bytes.len(), + }); + } + let what = u32::from_le_bytes(bytes[0..4].try_into().unwrap()); + // bytes[4..8] = cpu (ignored — diagnostic only) + // bytes[8..16] = timestamp_ns (ignored — kernel monotonic, not wall-clock) + + if what != PROC_EVENT_EXEC_RAW { + return Ok(ProcEvent::Other { what }); + } + + let exec_end = PROC_EVENT_HEADER_LEN + PROC_EVENT_EXEC_PAYLOAD_LEN; + if bytes.len() < exec_end { + return Err(NetlinkError::Truncated { + expected: exec_end, + actual: bytes.len(), + }); + } + let pid = u32::from_le_bytes( + bytes[PROC_EVENT_HEADER_LEN..PROC_EVENT_HEADER_LEN + 4] + .try_into() + .unwrap(), + ); + let tgid = u32::from_le_bytes( + bytes[PROC_EVENT_HEADER_LEN + 4..exec_end] + .try_into() + .unwrap(), + ); + Ok(ProcEvent::Exec { pid, tgid }) +} + +// ============================================================================ +// Subscriber — Linux implementation +// ============================================================================ + +/// Linux-only socket / bind / recv implementation of [`Subscriber`]. +/// +/// Kept private; the only export is [`Subscriber`] re-exported just +/// below the `mod` block so the public API surface is the same on +/// every platform. Splitting the impl into its own private module +/// keeps the `nix` import (Linux-only target dep) out of the +/// portable parser layer. +#[cfg(target_os = "linux")] +mod linux_impl { + use super::{ + parse_cn_msg, parse_proc_event, NetlinkError, ProcEvent, CN_IDX_PROC, CN_VAL_PROC, + PROC_CN_MCAST_LISTEN, + }; + use nix::sys::socket::{bind, recv, send, MsgFlags, NetlinkAddr}; + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + + /// Receive buffer size in bytes. + /// + /// Sizing rationale: `nlmsghdr` (16) + `cn_msg` header (20) + + /// largest known `proc_event` payload (~40 for the COREDUMP + /// variant) ≈ 80 bytes per message. 4 KiB gives ~50× headroom + /// for any future variant the kernel may add and for the + /// (rare-in-practice) case where the kernel batches multiple + /// connector messages into one datagram. The buffer is owned + /// by the [`Subscriber`] so the per-call allocation cost is + /// zero — `recv(2)` writes into the same array on every event. + const RECV_BUF_LEN: usize = 4096; + + /// Length of the fixed netlink message header in bytes. + /// + /// Layout (`linux/netlink.h::struct nlmsghdr`): + /// `nlmsg_len` (4) + `nlmsg_type` (2) + `nlmsg_flags` (2) + + /// `nlmsg_seq` (4) + `nlmsg_pid` (4) = 16. + const NLMSGHDR_LEN: usize = 16; + + /// `NLMSG_DONE` from `linux/netlink.h` — signals "end of a + /// multipart message." Used as the `nlmsg_type` on the outbound + /// control message because the kernel ignores the type field on + /// proc-connector subscribe traffic but rejects malformed netlink + /// frames; `NLMSG_DONE` is a valid type and cannot be confused + /// with `NLMSG_ERROR (2)` or `NLMSG_NOOP (1)`. + const NLMSG_DONE: u16 = 3; + + /// Subscriber for kernel proc-connector events. + /// + /// Construction: + /// 1. Opens an `AF_NETLINK` / `SOCK_DGRAM` socket on the + /// `NETLINK_CONNECTOR` protocol. + /// 2. Binds with `nl_pid = 0` (let the kernel pick our address) + /// and `nl_groups = CN_IDX_PROC` (subscribe to the + /// proc-connector multicast group). + /// 3. Sends a `PROC_CN_MCAST_LISTEN` control message — required + /// to actually start receiving events. + /// + /// The socket and the receive buffer are owned by the + /// `Subscriber`; the buffer is reused across `recv` calls so + /// steady-state allocation is zero. + /// + /// **Drop behaviour:** dropping the `Subscriber` closes the file + /// descriptor (the `OwnedFd` runs `close(2)`), and the kernel + /// auto-removes us from the multicast group. No explicit + /// unsubscribe call is needed. + /// + /// **Thread safety:** `Subscriber` is `!Sync` because of the + /// mutable buffer — `next_event` takes `&mut self`. Multiple + /// subscribers in the same process work fine; the kernel + /// delivers to all of them. + /// + /// **Blocking:** every call to [`Subscriber::next_event`] blocks + /// the calling thread until the kernel pushes an event. Hosts + /// with idle process tables can wait indefinitely. The + /// reclaim-agent uses `tokio::task::spawn_blocking` to keep this + /// off the tokio worker pool. + pub struct Subscriber { + /// Owned netlink socket file descriptor. Closed on `Drop`. + fd: OwnedFd, + /// Reusable receive buffer; reused across every `next_event` + /// call to avoid per-call allocation. + buf: [u8; RECV_BUF_LEN], + } + + impl Subscriber { + /// Open a netlink socket, join the `CN_IDX_PROC` multicast + /// group, and start the listen subscription. After this + /// returns `Ok`, the kernel will push every `exec(2)` (and + /// every other `proc_event` variant) on the host into our + /// socket. + /// + /// # Capability + /// Requires `CAP_NET_ADMIN`. Without it, `bind(2)` returns + /// `EPERM` and this function returns + /// [`NetlinkError::Io`] wrapping the `EPERM` error. Grant + /// the cap via the container's `securityContext.capabilities.add` + /// (already done in `deploy/node-agent/daemonset.yaml`). + /// + /// # Kernel + /// Requires `CONFIG_PROC_EVENTS=y` in the running kernel. + /// On a kernel built without it, `socket(2)` succeeds but + /// no events are ever delivered. There is no programmatic + /// way to detect this from userspace; deployment-time + /// validation is the safety net. + /// + /// # Errors + /// Surfaces any underlying `socket(2)` / `bind(2)` / `send(2)` + /// failure as [`NetlinkError::Io`]. Common cases: + /// - `EPERM` — missing `CAP_NET_ADMIN`. + /// - `EAFNOSUPPORT` — kernel built without `CONFIG_NETLINK`. + /// - `EPROTONOSUPPORT` — kernel built without + /// `CONFIG_CONNECTOR`. + pub fn new() -> Result { + // `nix::sys::socket::SockProtocol` does not expose + // `NETLINK_CONNECTOR` (protocol 11). Verified missing in + // nix 0.30.x AND nix 0.31.x (the latest as of 2026-05). + // The other netlink protocols are enumerated + // (`NetlinkRoute`, `NetlinkAudit`, `NetlinkSCSITransport`, + // `NetlinkGeneric`, `NetlinkSockDiag`, `NetlinkRDMA`, …) + // but the connector slot was missed upstream — bumping + // the dep does not fix it. Drop to libc for this one call + // and wrap the resulting fd in `OwnedFd` so the rest of + // the lifecycle (Drop → close(2)) still goes through the + // safe Rust layer. + // + // These are the only two `unsafe` blocks in the entire + // 5-Spot codebase. Both are confined to this function and + // operate on values that never leave the call (the raw fd + // is consumed into an `OwnedFd` before any other code can + // observe it). The invariants below are the audit + // surface — reviewing them once is sufficient because + // there are no other unsafe sites to coordinate with. + + // SAFETY: + // - `socket(2)` is a pure POSIX syscall with no aliasing + // or memory preconditions that Rust can violate; all + // three arguments are integer constants exported by + // libc and validated at compile time. + // - The return value is either a valid, owned, non-negative + // file descriptor or `-1` with `errno` set. We check for + // the error case immediately on the next line and never + // use a negative value as an fd. + // - `errno` is read via `Error::last_os_error()` only on + // the error branch and only on the same thread that + // made the syscall, so the value is well-defined. + // nosemgrep: rust.lang.security.unsafe-usage.unsafe-usage + let raw_fd = unsafe { + nix::libc::socket( + nix::libc::AF_NETLINK, + nix::libc::SOCK_DGRAM | nix::libc::SOCK_CLOEXEC, + nix::libc::NETLINK_CONNECTOR, + ) + }; + if raw_fd < 0 { + return Err(NetlinkError::Io(std::io::Error::last_os_error())); + } + // SAFETY: + // - `raw_fd` was just returned by `socket(2)` above and + // verified `>= 0` on the previous line. + // - The fd is exclusively owned by this call frame: it was + // allocated by the kernel inline in this function and + // has not been duplicated, leaked, or stored anywhere + // else. + // - `OwnedFd::from_raw_fd`'s safety contract requires the + // caller to transfer ownership; we do exactly that — the + // raw fd is never used again as an integer after this + // line. The `OwnedFd` will `close(2)` it on `Drop`, + // satisfying the close-exactly-once invariant. + // nosemgrep: rust.lang.security.unsafe-usage.unsafe-usage + let fd = unsafe { OwnedFd::from_raw_fd(raw_fd) }; + + // pid=0 lets the kernel pick our netlink address; + // groups=CN_IDX_PROC subscribes to the proc-event multicast + // group (bit 0). + let addr = NetlinkAddr::new(0, CN_IDX_PROC); + bind(fd.as_raw_fd(), &addr).map_err(io_err)?; + + // Send PROC_CN_MCAST_LISTEN. Without this, binding alone is + // not enough — the kernel ignores subscribers that have not + // explicitly opted in. + let mut frame = [0u8; NLMSGHDR_LEN + 20 + 4]; + let total_len = u32::try_from(frame.len()).expect("frame len fits u32"); + // nlmsghdr + frame[0..4].copy_from_slice(&total_len.to_le_bytes()); + frame[4..6].copy_from_slice(&NLMSG_DONE.to_le_bytes()); + // flags (0), seq (0), pid (0) — already zero-initialised. + // cn_msg + frame[NLMSGHDR_LEN..NLMSGHDR_LEN + 4].copy_from_slice(&CN_IDX_PROC.to_le_bytes()); + frame[NLMSGHDR_LEN + 4..NLMSGHDR_LEN + 8].copy_from_slice(&CN_VAL_PROC.to_le_bytes()); + // seq, ack, flags — zero. len = 4 (uint32 payload). + let payload_len: u16 = 4; + frame[NLMSGHDR_LEN + 16..NLMSGHDR_LEN + 18].copy_from_slice(&payload_len.to_le_bytes()); + // payload + frame[NLMSGHDR_LEN + 20..NLMSGHDR_LEN + 24] + .copy_from_slice(&PROC_CN_MCAST_LISTEN.to_le_bytes()); + + send(fd.as_raw_fd(), &frame, MsgFlags::empty()).map_err(io_err)?; + + Ok(Self { + fd, + buf: [0u8; RECV_BUF_LEN], + }) + } + + /// Block until the kernel pushes the next event, then return it. + /// + /// **Blocking semantics:** the underlying `recv(2)` call + /// blocks the calling thread until a datagram arrives. There + /// is no built-in timeout. Hosts with no exec activity can + /// wait indefinitely. The reclaim-agent's + /// `run_netlink_scanner` runs this loop on + /// `tokio::task::spawn_blocking` and cancels via dropping the + /// `Subscriber` from the parent task. + /// + /// **Return value semantics:** + /// - `Ok(Some(ProcEvent::Exec { pid, tgid }))` — a successful + /// `execve(2)` was observed. + /// - `Ok(Some(ProcEvent::Other { what }))` — any other event + /// variant (FORK, EXIT, COMM, …). Surfaced rather than + /// silently dropped so the caller can log / count if + /// desired. + /// - `Ok(None)` — a frame arrived but failed to parse (too + /// short, wrong connector id, malformed `proc_event`). + /// Logged at `warn` level; the caller loops and tries + /// again. Recoverable per-message. + /// + /// **Per-message logging:** parse failures emit a + /// `tracing::warn!` with the error before returning + /// `Ok(None)`. Callers should NOT add their own warning + /// for the `None` case to avoid duplicate log entries. + /// + /// **Multi-message-per-datagram caveat:** see the module-level + /// "Limitations" section. We currently parse only the first + /// message in each `recv` buffer. Subsequent messages in the + /// same datagram are silently dropped — under sustained + /// burst load, rung 1 (`/proc` poll) is the backstop. + /// + /// # Errors + /// Returns [`NetlinkError::Io`] only on socket-level failures + /// the caller cannot recover from (closed fd, EINTR-on-shutdown, + /// `ENOBUFS` from a sustained kernel push faster than userspace + /// drain, etc.). Per-message parse failures are mapped to + /// `Ok(None)` with a `tracing::warn!` for visibility. + pub fn next_event(&mut self) -> Result, NetlinkError> { + let n = recv(self.fd.as_raw_fd(), &mut self.buf, MsgFlags::empty()).map_err(io_err)?; + if n < NLMSGHDR_LEN { + tracing::warn!(bytes = n, "netlink frame shorter than nlmsghdr; dropping"); + return Ok(None); + } + // Skip the netlink message header — it carries length / + // type / flags / seq / pid that the proc-connector doesn't + // need at this layer. + let cn_bytes = &self.buf[NLMSGHDR_LEN..n]; + let payload = match parse_cn_msg(cn_bytes) { + Ok(p) => p, + Err(e) => { + tracing::warn!(error = %e, "cn_msg parse error; dropping frame"); + return Ok(None); + } + }; + match parse_proc_event(payload) { + Ok(evt) => Ok(Some(evt)), + Err(e) => { + tracing::warn!(error = %e, "proc_event parse error; dropping frame"); + Ok(None) + } + } + } + } + + fn io_err(e: nix::errno::Errno) -> NetlinkError { + NetlinkError::Io(std::io::Error::from_raw_os_error(e as i32)) + } +} + +#[cfg(target_os = "linux")] +pub use linux_impl::Subscriber; + +// ============================================================================ +// Subscriber — non-Linux stub +// ============================================================================ + +/// Non-Linux compile stub for [`Subscriber`]. +/// +/// Why a stub at all: the reclaim-agent binary is built for macOS +/// and Linux; the macOS build runs in dev/CI but never in +/// production, where the agent only ever runs on Linux nodes via +/// the DaemonSet. Stubbing the constructor keeps `cargo build`, +/// `cargo test`, and `cargo clippy` working on macOS without any +/// `#[cfg(target_os = "linux")]` gymnastics at the bin level — +/// the bin can call `Subscriber::new()` unconditionally and let +/// the runtime check (returns [`NetlinkError::Unsupported`]) tell +/// the operator their platform doesn't match their `--detector` +/// flag. +#[cfg(not(target_os = "linux"))] +mod stub_impl { + use super::{NetlinkError, ProcEvent}; + + /// Compile-only stub of [`Subscriber`] for non-Linux targets. + /// + /// The constructor immediately returns + /// [`NetlinkError::Unsupported`]. This lets the binary link cleanly + /// on macOS / Windows builds (CI smoke tests, local dev) while + /// surfacing a clear error at startup if an operator passes + /// `--detector=netlink` on a platform without netlink. + /// + /// The unit `PhantomData` field exists only so the struct has + /// non-zero "shape" for trait-impl purposes; no instance can ever + /// be constructed because [`Subscriber::new`] never returns + /// `Ok(Self)`. + #[derive(Debug)] + pub struct Subscriber { + // Field present so `Subscriber::next_event` has somewhere to + // pretend to live; never constructed because `new()` never + // returns Ok. + _marker: std::marker::PhantomData<()>, + } + + impl Subscriber { + /// Always errors on non-Linux. + /// + /// # Errors + /// Always returns [`NetlinkError::Unsupported`]. + pub fn new() -> Result { + Err(NetlinkError::Unsupported) + } + + /// Unreachable in practice — `Subscriber` cannot be constructed + /// on this platform (the constructor refuses), so any code + /// path that holds a `&mut Subscriber` was crafted from + /// nothing. Defined for API parity so the bin compiles + /// uniformly across platforms. + /// + /// # Errors + /// Always returns [`NetlinkError::Unsupported`]; in practice it + /// is unreachable because [`Subscriber::new`] never returns + /// `Ok`. + pub fn next_event(&mut self) -> Result, NetlinkError> { + Err(NetlinkError::Unsupported) + } + } +} + +#[cfg(not(target_os = "linux"))] +pub use stub_impl::Subscriber; + +#[cfg(test)] +#[path = "netlink_proc_tests.rs"] +mod tests; diff --git a/src/netlink_proc_tests.rs b/src/netlink_proc_tests.rs new file mode 100644 index 0000000..08c2ece --- /dev/null +++ b/src/netlink_proc_tests.rs @@ -0,0 +1,259 @@ +// Copyright (c) 2025 Erick Bourgeois, firestoned +// SPDX-License-Identifier: Apache-2.0 +//! Byte-level tests for the netlink proc connector parsers. +//! +//! Runnable on macOS (no socket required). Each test synthesises the +//! exact byte layout the Linux kernel writes for one event, feeds it +//! through the parser, and asserts the resulting [`ProcEvent`]. + +#[cfg(test)] +#[allow(clippy::module_inception)] +mod tests { + use super::super::*; + + // ───────────────────────────────────────────────────────────────── + // Helpers — build the wire layouts the kernel emits. + // + // cn_msg header (20 bytes): + // id.idx u32 LE - CN_IDX_PROC = 1 + // id.val u32 LE - CN_VAL_PROC = 1 + // seq u32 LE + // ack u32 LE + // len u16 LE - payload length + // flags u16 LE + // + // proc_event payload (24 bytes for an exec): + // what u32 LE - PROC_EVENT_EXEC = 0x00000002 + // cpu u32 LE + // timestamp_ns u64 LE + // exec.pid i32 LE + // exec.tgid i32 LE + // ───────────────────────────────────────────────────────────────── + + /// Build a minimal `cn_msg` wrapping a payload, with the standard + /// proc-connector id `(idx=1, val=1)`. + fn cn_msg_with_payload(payload: &[u8]) -> Vec { + let mut buf = Vec::with_capacity(20 + payload.len()); + buf.extend_from_slice(&CN_IDX_PROC.to_le_bytes()); // idx + buf.extend_from_slice(&CN_VAL_PROC.to_le_bytes()); // val + buf.extend_from_slice(&0u32.to_le_bytes()); // seq + buf.extend_from_slice(&0u32.to_le_bytes()); // ack + let len = u16::try_from(payload.len()).expect("payload fits"); + buf.extend_from_slice(&len.to_le_bytes()); // len + buf.extend_from_slice(&0u16.to_le_bytes()); // flags + buf.extend_from_slice(payload); + buf + } + + /// Build a 24-byte `proc_event` for an exec event with the given pid/tgid. + fn exec_event_bytes(pid: u32, tgid: u32) -> Vec { + let mut buf = Vec::with_capacity(24); + buf.extend_from_slice(&PROC_EVENT_EXEC_RAW.to_le_bytes()); // what + buf.extend_from_slice(&0u32.to_le_bytes()); // cpu + buf.extend_from_slice(&0u64.to_le_bytes()); // timestamp_ns + buf.extend_from_slice(&pid.to_le_bytes()); // process_pid + buf.extend_from_slice(&tgid.to_le_bytes()); // process_tgid + buf + } + + // ───────────────────────────────────────────────────────────────── + // parse_proc_event + // ───────────────────────────────────────────────────────────────── + + #[test] + fn parse_proc_event_exec_returns_pid_tgid() { + let bytes = exec_event_bytes(4242, 4242); + let evt = parse_proc_event(&bytes).expect("valid exec event parses"); + assert_eq!( + evt, + ProcEvent::Exec { + pid: 4242, + tgid: 4242 + } + ); + } + + #[test] + fn parse_proc_event_exec_distinguishes_pid_from_tgid() { + // Real-world: a thread within a multi-threaded process can have + // pid != tgid. The detector should match against both views. + let bytes = exec_event_bytes(4243, 4242); + let evt = parse_proc_event(&bytes).expect("valid event"); + assert_eq!( + evt, + ProcEvent::Exec { + pid: 4243, + tgid: 4242 + } + ); + } + + #[test] + fn parse_proc_event_fork_is_classified_other() { + // Build a FORK event (4 bytes what + 4 cpu + 8 timestamp + payload). + // Payload size differs from exec but we only care about `what`. + let mut bytes = Vec::with_capacity(24); + bytes.extend_from_slice(&0x0000_0001u32.to_le_bytes()); // PROC_EVENT_FORK + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&0u64.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); // parent pid + bytes.extend_from_slice(&0u32.to_le_bytes()); // parent tgid + let evt = parse_proc_event(&bytes).expect("non-exec event still parses"); + assert!( + matches!(evt, ProcEvent::Other { what: 0x0000_0001 }), + "FORK event must classify as Other, got {evt:?}" + ); + } + + #[test] + fn parse_proc_event_exit_is_classified_other() { + let mut bytes = Vec::with_capacity(24); + bytes.extend_from_slice(&0x8000_0000u32.to_le_bytes()); // PROC_EVENT_EXIT + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&0u64.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + let evt = parse_proc_event(&bytes).expect("exit event parses"); + assert!(matches!(evt, ProcEvent::Other { what: 0x8000_0000 })); + } + + #[test] + fn parse_proc_event_truncated_returns_error() { + // Anything shorter than the proc_event header (16 bytes: + // what + cpu + timestamp_ns) is unconditionally bad. + let bytes = [0u8; 8]; + let err = parse_proc_event(&bytes).expect_err("truncated must error"); + assert!( + matches!(err, NetlinkError::Truncated { .. }), + "expected Truncated, got {err:?}" + ); + } + + #[test] + fn parse_proc_event_exec_truncated_payload_returns_error() { + // Header is full (16 bytes) but the exec payload (8 bytes for + // pid + tgid) is missing. Must error rather than read past end. + let mut bytes = Vec::with_capacity(16); + bytes.extend_from_slice(&PROC_EVENT_EXEC_RAW.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&0u64.to_le_bytes()); + let err = parse_proc_event(&bytes).expect_err("missing exec payload must error"); + assert!(matches!(err, NetlinkError::Truncated { .. })); + } + + #[test] + fn parse_proc_event_empty_buffer_returns_error() { + let err = parse_proc_event(&[]).expect_err("empty input must error"); + assert!(matches!(err, NetlinkError::Truncated { .. })); + } + + // ───────────────────────────────────────────────────────────────── + // parse_cn_msg + // ───────────────────────────────────────────────────────────────── + + #[test] + fn parse_cn_msg_extracts_payload_for_exec() { + let payload = exec_event_bytes(7777, 7777); + let frame = cn_msg_with_payload(&payload); + let extracted = parse_cn_msg(&frame).expect("well-formed frame parses"); + assert_eq!(extracted, payload.as_slice()); + } + + #[test] + fn parse_cn_msg_rejects_wrong_idx() { + // A connector message for some other subsystem (e.g. CN_IDX_CIFS = 5) + // must be rejected — we are not subscribed to it and a wrong-idx + // frame is either a kernel bug or wire corruption. + let mut frame = Vec::with_capacity(20); + frame.extend_from_slice(&5u32.to_le_bytes()); // wrong idx + frame.extend_from_slice(&CN_VAL_PROC.to_le_bytes()); + frame.extend_from_slice(&[0u8; 12]); // seq + ack + len + flags + let err = parse_cn_msg(&frame).expect_err("wrong idx must error"); + assert!( + matches!(err, NetlinkError::InvalidId { idx: 5, .. }), + "expected InvalidId{{idx=5}}, got {err:?}" + ); + } + + #[test] + fn parse_cn_msg_rejects_wrong_val() { + let mut frame = Vec::with_capacity(20); + frame.extend_from_slice(&CN_IDX_PROC.to_le_bytes()); + frame.extend_from_slice(&99u32.to_le_bytes()); // wrong val + frame.extend_from_slice(&[0u8; 12]); + let err = parse_cn_msg(&frame).expect_err("wrong val must error"); + assert!( + matches!(err, NetlinkError::InvalidId { val: 99, .. }), + "expected InvalidId{{val=99}}, got {err:?}" + ); + } + + #[test] + fn parse_cn_msg_truncated_header_returns_error() { + let frame = [0u8; 10]; // shorter than the 20-byte header + let err = parse_cn_msg(&frame).expect_err("truncated header must error"); + assert!(matches!(err, NetlinkError::Truncated { .. })); + } + + #[test] + fn parse_cn_msg_payload_shorter_than_declared_returns_error() { + // Header claims len=24 but the buffer only has 4 payload bytes — + // must error rather than slice out of bounds. + let mut frame = Vec::with_capacity(24); + frame.extend_from_slice(&CN_IDX_PROC.to_le_bytes()); + frame.extend_from_slice(&CN_VAL_PROC.to_le_bytes()); + frame.extend_from_slice(&0u32.to_le_bytes()); // seq + frame.extend_from_slice(&0u32.to_le_bytes()); // ack + frame.extend_from_slice(&24u16.to_le_bytes()); // len = 24 + frame.extend_from_slice(&0u16.to_le_bytes()); // flags + frame.extend_from_slice(&[0u8; 4]); // 4 bytes payload, not 24 + let err = parse_cn_msg(&frame).expect_err("short payload must error"); + assert!(matches!(err, NetlinkError::Truncated { .. })); + } + + #[test] + fn parse_cn_msg_zero_length_payload_is_valid() { + // A keep-alive or boundary frame with no payload is benign — + // not all messages carry an event. + let frame = cn_msg_with_payload(&[]); + let payload = parse_cn_msg(&frame).expect("zero-length payload parses"); + assert!(payload.is_empty()); + } + + // ───────────────────────────────────────────────────────────────── + // End-to-end: cn_msg → proc_event for an EXEC event. + // ───────────────────────────────────────────────────────────────── + + #[test] + fn cn_msg_then_proc_event_round_trip_for_exec() { + let payload = exec_event_bytes(12345, 12345); + let frame = cn_msg_with_payload(&payload); + let extracted = parse_cn_msg(&frame).expect("frame parses"); + let evt = parse_proc_event(extracted).expect("event parses"); + assert_eq!( + evt, + ProcEvent::Exec { + pid: 12345, + tgid: 12345 + } + ); + } + + // ───────────────────────────────────────────────────────────────── + // Subscriber stub on non-Linux platforms. + // + // On macOS / Windows (where our CI also runs unit tests) the + // Subscriber::new() entry point exists but immediately returns + // NetlinkError::Unsupported. This pins that contract. + // ───────────────────────────────────────────────────────────────── + + #[cfg(not(target_os = "linux"))] + #[test] + fn subscriber_new_returns_unsupported_on_non_linux() { + let err = Subscriber::new().expect_err("non-linux must refuse"); + assert!( + matches!(err, NetlinkError::Unsupported), + "expected Unsupported on non-linux, got {err:?}" + ); + } +} diff --git a/src/reclaim_agent.rs b/src/reclaim_agent.rs index 0b9bdf5..f37bb6a 100644 --- a/src/reclaim_agent.rs +++ b/src/reclaim_agent.rs @@ -172,7 +172,13 @@ pub fn scan_proc(proc_root: &Path, config: &Config) -> Result, io: Ok(None) } -fn match_pid(proc_root: &Path, pid: u32, config: &Config) -> Option { +/// Test a single pid against the config and return a [`Match`] if its +/// `comm` or `cmdline` matches. Used by [`scan_proc`] (rung 1, walks +/// every pid) and by the rung 2 netlink subscriber (resolves a single +/// pid pushed by the kernel). Returns `None` if no match, the pid no +/// longer exists, or its `/proc` files cannot be read. +#[must_use] +pub fn match_pid(proc_root: &Path, pid: u32, config: &Config) -> Option { let pid_dir = proc_root.join(pid.to_string()); if let Ok(comm_raw) = fs::read_to_string(pid_dir.join("comm")) { diff --git a/src/reconcilers/helpers.rs b/src/reconcilers/helpers.rs index c4256ce..c5f0145 100644 --- a/src/reconcilers/helpers.rs +++ b/src/reconcilers/helpers.rs @@ -51,7 +51,10 @@ use crate::constants::{ REASON_RECONCILE_SUCCEEDED, RESERVED_LABEL_PREFIXES, TIMER_REQUEUE_SECS, }; use crate::crd::{Condition, NodeRef, ScheduledMachine, ScheduledMachineStatus}; -use crate::metrics::{record_node_drain, record_pod_eviction}; +use crate::metrics::{ + record_emergency_drain, record_emergency_reclaim, record_node_drain, record_pod_eviction, + EmergencyDrainOutcome, +}; // ============================================================================ // Resource processing and consistent hashing @@ -2180,6 +2183,31 @@ pub fn build_emergency_disable_schedule_event() -> KubeEvent { } } +/// Build the `Warning`-type Kubernetes Event published on the +/// `ScheduledMachine` when the controller observes ≥ +/// [`crate::constants::RAPID_RE_RECLAIM_THRESHOLD`] reclaim events for +/// the same SM within the configured window. The event tells the +/// operator their re-enable likely failed because the conflicting +/// process is still running — they should stop it before re-enabling +/// again, otherwise the loop will continue. +#[must_use] +pub fn build_rapid_re_reclaim_event(name: &str) -> KubeEvent { + KubeEvent { + type_: EventType::Warning, + reason: crate::constants::REASON_RAPID_RE_RECLAIM.to_string(), + note: Some(format!( + "ScheduledMachine {name} has been emergency-reclaimed {} or more times within \ + {} seconds. The conflicting process is likely still running on the node — \ + stop it before re-enabling spec.schedule.enabled, or the agent will fire again \ + on the next reconcile.", + crate::constants::RAPID_RE_RECLAIM_THRESHOLD, + crate::constants::RAPID_RE_RECLAIM_WINDOW_SECS, + )), + action: "RapidReReclaim".to_string(), + secondary: None, + } +} + /// Format the human-readable message for the status condition recorded /// when the `ScheduledMachine` enters `Phase::EmergencyRemove`. /// @@ -2301,16 +2329,58 @@ pub async fn handle_emergency_remove( ) .await?; + // Per-SM emergency-reclaim counter. Bumped once per *flow* (the + // idempotent recovery handler short-circuits before this point on + // restart) so a controller crash mid-flow does not double-count. + record_emergency_reclaim(&namespace, &name); + + // Loop-protection: append this reclaim's timestamp to the per-SM + // deque in Context, prune entries outside the window, and if the + // post-append count crosses the threshold, emit a RapidReReclaim + // Warning event + bump the metric. The tracker is in-memory only + // (see crate::loop_protection module rustdoc for rationale). + let breach = { + let key = format!("{namespace}/{name}"); + let now_ts = chrono::Utc::now(); + let mut lock = ctx.recent_reclaims.lock().expect("recent_reclaims mutex"); + let deque = lock.entry(key).or_default(); + crate::loop_protection::record_emergency_reclaim_event(deque, now_ts) + }; + if breach { + warn!( + resource = %name, + namespace = %namespace, + "Rapid-re-reclaim threshold crossed — emitting RapidReReclaim Warning Event" + ); + crate::metrics::record_rapid_re_reclaim(&namespace, &name); + publish_best_effort( + &ctx, + &sm_object_ref, + build_rapid_re_reclaim_event(&name), + "RapidReReclaim", + ) + .await; + } + // Step 3: drain with short emergency timeout. Best-effort — we do // NOT block Machine deletion on a failed drain, because the agent - // has already decided the node must leave. - if let Err(e) = drain_node_with_timeout( + // has already decided the node must leave. The stopwatch is + // observed regardless of outcome so operators can compute success + // P95 vs timeout-rate side by side. + let drain_start = std::time::Instant::now(); + let drain_result = drain_node_with_timeout( &ctx.client, node_name, Duration::from_secs(EMERGENCY_DRAIN_TIMEOUT_SECS), ) - .await - { + .await; + let drain_outcome = match &drain_result { + Ok(()) => EmergencyDrainOutcome::Success, + Err(e) if e.to_string().contains("timeout") => EmergencyDrainOutcome::Timeout, + Err(_) => EmergencyDrainOutcome::Error, + }; + record_emergency_drain(drain_start.elapsed().as_secs_f64(), drain_outcome); + if let Err(e) = drain_result { warn!( resource = %name, node = %node_name, diff --git a/src/reconcilers/mod.rs b/src/reconcilers/mod.rs index e1a1c3b..24a239e 100644 --- a/src/reconcilers/mod.rs +++ b/src/reconcilers/mod.rs @@ -20,9 +20,9 @@ pub mod scheduled_machine; // Re-export main types and functions #[allow(deprecated)] // re-export of legacy node_to_scheduled_machines for one release pub use helpers::{ - error_policy, evaluate_schedule, machine_to_scheduled_machine, node_to_scheduled_machines, - node_to_scheduled_machines_via_machine, parse_duration, reconcile_node_taints, - should_process_resource, validate_cluster_name, validate_kill_if_commands, - NodeTaintReconcileOutcome, ReconcileNodeTaintsInput, + build_clear_reclaim_patch, error_policy, evaluate_schedule, machine_to_scheduled_machine, + node_reclaim_request, node_to_scheduled_machines, node_to_scheduled_machines_via_machine, + parse_duration, reconcile_node_taints, should_process_resource, validate_cluster_name, + validate_kill_if_commands, NodeTaintReconcileOutcome, ReclaimRequest, ReconcileNodeTaintsInput, }; pub use scheduled_machine::{reconcile_scheduled_machine, Context, ReconcilerError}; diff --git a/src/reconcilers/scheduled_machine.rs b/src/reconcilers/scheduled_machine.rs index 3e22b9b..4fcb266 100644 --- a/src/reconcilers/scheduled_machine.rs +++ b/src/reconcilers/scheduled_machine.rs @@ -97,6 +97,18 @@ pub struct Context { /// after the resource recovers. Uses a `std::sync::Mutex` because /// `error_policy` is synchronous. pub retry_counts: Arc>>, + /// Recent emergency-reclaim event timestamps per `"namespace/name"`. + /// + /// Tracked in-memory only — see [`crate::loop_protection`] for the + /// reasoning. The reconciler appends a timestamp on every successful + /// entry into the `EmergencyRemove` flow and emits a `RapidReReclaim` + /// Warning Event when the deque length within + /// [`crate::constants::RAPID_RE_RECLAIM_WINDOW_SECS`] crosses + /// [`crate::constants::RAPID_RE_RECLAIM_THRESHOLD`]. Capped at + /// [`crate::constants::RAPID_RE_RECLAIM_MAX_TRACKED`] entries per SM + /// to bound worst-case allocation. + pub recent_reclaims: + Arc>>>>, /// When `true` (default), `handle_deletion` force-removes the finalizer if /// CAPI cleanup exceeds [`crate::constants::FINALIZER_CLEANUP_TIMEOUT_SECS`], /// surfacing a `FinalizerCleanupTimedOut` Warning event and incrementing @@ -130,6 +142,7 @@ impl Context { instance_count, recorder, retry_counts: Arc::new(Mutex::new(HashMap::new())), + recent_reclaims: Arc::new(Mutex::new(HashMap::new())), // Default to true so existing single-instance deployments without // leader election continue to work without any configuration change. is_leader: Arc::new(AtomicBool::new(true)), diff --git a/tests/integration_emergency_reclaim.rs b/tests/integration_emergency_reclaim.rs new file mode 100644 index 0000000..4746e0e --- /dev/null +++ b/tests/integration_emergency_reclaim.rs @@ -0,0 +1,249 @@ +// Copyright (c) 2025 Erick Bourgeois, firestoned +// SPDX-License-Identifier: Apache-2.0 +//! # Integration test: agent → controller annotation contract on a real cluster +//! +//! Scope cut, deliberately: full 7-step `EmergencyRemove` flow needs a +//! ScheduledMachine + CAPI Machine + drain target — out of scope for a +//! kind-cluster smoke test. What we *can* prove against a real API +//! server with no operator running is the **annotation contract** — the +//! single load-bearing surface between the node-side agent and the +//! controller. If the agent's `build_patch_body` writes annotations +//! that the controller's `node_reclaim_request` cannot parse, the whole +//! flow fails silently. This test catches that class of regression. +//! +//! Mirrors the pattern in `tests/integration_node_taints.rs` — +//! graceful skip when no cluster is reachable, picks an arbitrary +//! Ready Node, mutates only test-owned annotations, and scrubs on +//! exit. +//! +//! ## Running +//! +//! ```bash +//! kind create cluster # or any reachable cluster +//! cargo test --test integration_emergency_reclaim -- --ignored --test-threads=1 +//! ``` +//! +//! ## Safety — DO NOT run against a cluster with the 5-spot operator installed +//! +//! The annotation keys are the production keys. If the operator is +//! running, this test could trigger a real `EmergencyRemove` flow on +//! the chosen Node. The cleanup guard scrubs on exit, but the operator +//! may have already started draining. Use a kind cluster without the +//! operator deployed. + +use std::collections::BTreeMap; + +use k8s_openapi::api::core::v1::Node; +use kube::api::{Patch, PatchParams}; +use kube::{Api, Client}; +use serde_json::json; + +use five_spot::constants::{ + RECLAIM_REASON_ANNOTATION, RECLAIM_REQUESTED_ANNOTATION, RECLAIM_REQUESTED_AT_ANNOTATION, + RECLAIM_REQUESTED_VALUE, +}; +use five_spot::reclaim_agent::{build_patch_body, Match, MatchSource}; +use five_spot::reconcilers::node_reclaim_request; + +const TEST_FIELD_MANAGER: &str = "5spot-integration-test-emergency-reclaim"; + +async fn client_or_skip() -> Option { + match Client::try_default().await { + Ok(c) => Some(c), + Err(e) => { + eprintln!("SKIP: no reachable cluster ({e}); run against kind to exercise this test"); + None + } + } +} + +async fn pick_ready_node(client: &Client) -> Option { + let nodes: Api = Api::all(client.clone()); + let list = match nodes.list(&Default::default()).await { + Ok(l) => l, + Err(e) => { + eprintln!("SKIP: failed to list nodes ({e})"); + return None; + } + }; + for n in list.items { + let name = n.metadata.name.clone()?; + let ready = n + .status + .as_ref() + .and_then(|s| s.conditions.as_ref()) + .map(|cs| cs.iter().any(|c| c.type_ == "Ready" && c.status == "True")) + .unwrap_or(false); + if ready { + return Some(name); + } + } + None +} + +async fn clear_test_annotations(nodes: &Api, name: &str) { + let patch = json!({ + "metadata": { + "annotations": { + RECLAIM_REQUESTED_ANNOTATION: serde_json::Value::Null, + RECLAIM_REASON_ANNOTATION: serde_json::Value::Null, + RECLAIM_REQUESTED_AT_ANNOTATION: serde_json::Value::Null, + } + } + }); + let _ = nodes + .patch( + name, + &PatchParams::apply(TEST_FIELD_MANAGER).force(), + &Patch::Merge(&patch), + ) + .await; +} + +/// End-to-end: agent's annotation patch lands on a real Node, controller's +/// parser reads it back as a typed `ReclaimRequest` whose fields match. +#[tokio::test] +#[ignore] +async fn agent_annotation_round_trips_through_controller_parser() { + let Some(client) = client_or_skip().await else { + return; + }; + let Some(node_name) = pick_ready_node(&client).await else { + eprintln!("SKIP: no Ready node found in cluster"); + return; + }; + let nodes: Api = Api::all(client.clone()); + + // Always scrub on entry so a previous failed run doesn't pollute. + clear_test_annotations(&nodes, &node_name).await; + + // Build what the agent would PATCH on first match. + let m = Match { + pid: 99999, + matched_pattern: "integration-test-java".to_string(), + source: MatchSource::Comm, + }; + let ts = "2026-05-02T23:30:00Z"; + let patch = build_patch_body(&m, ts); + + // Apply with a unique field manager so we own the annotations cleanly. + nodes + .patch( + &node_name, + &PatchParams::apply(TEST_FIELD_MANAGER).force(), + &Patch::Merge(&patch), + ) + .await + .expect("PATCH reclaim annotations"); + + // Re-fetch the Node and feed it through the controller's parser. + let fetched = nodes.get(&node_name).await.expect("re-fetch node"); + let parsed = node_reclaim_request(&fetched); + + // Cleanup runs regardless of assertion outcome. + let result = std::panic::catch_unwind(|| { + let req = parsed.expect("controller must parse the annotations the agent wrote"); + // The reason format is ": " per build_patch_body. + let reason = req.reason.expect("reason annotation present"); + assert!( + reason.contains("integration-test-java"), + "reason must include the matched pattern; got: {reason}" + ); + assert_eq!( + req.requested_at.as_deref(), + Some(ts), + "timestamp annotation must round-trip verbatim" + ); + }); + + clear_test_annotations(&nodes, &node_name).await; + + if let Err(e) = result { + std::panic::resume_unwind(e); + } +} + +/// `node_reclaim_request` returns `None` when the requested annotation +/// is absent or set to a value other than the literal "true" sentinel. +/// Proves the parser refuses to fire on partial / stale state — a +/// missing requested annotation means "no request right now," even if +/// the reason or timestamp annotations linger from a prior cleanup. +#[tokio::test] +#[ignore] +async fn parser_returns_none_for_partial_annotations() { + let Some(client) = client_or_skip().await else { + return; + }; + let Some(node_name) = pick_ready_node(&client).await else { + eprintln!("SKIP: no Ready node found in cluster"); + return; + }; + let nodes: Api = Api::all(client.clone()); + + clear_test_annotations(&nodes, &node_name).await; + + // Set ONLY the reason — without the requested=true sentinel, the + // controller must not interpret this as a live reclaim request. + let partial = json!({ + "metadata": { + "annotations": { + RECLAIM_REASON_ANNOTATION: "process-match: stale", + } + } + }); + nodes + .patch( + &node_name, + &PatchParams::apply(TEST_FIELD_MANAGER).force(), + &Patch::Merge(&partial), + ) + .await + .expect("PATCH partial annotations"); + + let fetched = nodes.get(&node_name).await.expect("re-fetch node"); + + let result = std::panic::catch_unwind(|| { + let parsed = node_reclaim_request(&fetched); + assert!( + parsed.is_none(), + "parser must return None when requested=true is absent; got {parsed:?}" + ); + }); + + clear_test_annotations(&nodes, &node_name).await; + + if let Err(e) = result { + std::panic::resume_unwind(e); + } +} + +/// Read-only check that the production annotation keys we use here +/// match the constants the rest of the codebase imports. Pinned so a +/// rename in `src/constants.rs` is caught at the integration boundary +/// instead of silently breaking the agent/controller contract. +#[test] +fn annotation_keys_match_published_contract() { + let _: BTreeMap<&str, &str> = [ + ( + RECLAIM_REQUESTED_ANNOTATION, + "5spot.finos.org/reclaim-requested", + ), + (RECLAIM_REASON_ANNOTATION, "5spot.finos.org/reclaim-reason"), + ( + RECLAIM_REQUESTED_AT_ANNOTATION, + "5spot.finos.org/reclaim-requested-at", + ), + ] + .into_iter() + .collect(); + assert_eq!( + RECLAIM_REQUESTED_ANNOTATION, + "5spot.finos.org/reclaim-requested" + ); + assert_eq!(RECLAIM_REASON_ANNOTATION, "5spot.finos.org/reclaim-reason"); + assert_eq!( + RECLAIM_REQUESTED_AT_ANNOTATION, + "5spot.finos.org/reclaim-requested-at" + ); + assert_eq!(RECLAIM_REQUESTED_VALUE, "true"); +} diff --git a/tests/integration_netlink_proc.rs b/tests/integration_netlink_proc.rs new file mode 100644 index 0000000..b21f4a8 --- /dev/null +++ b/tests/integration_netlink_proc.rs @@ -0,0 +1,124 @@ +// Copyright (c) 2025 Erick Bourgeois, firestoned +// SPDX-License-Identifier: Apache-2.0 +//! Linux runtime test for the netlink proc connector subscriber. +//! +//! Per `5spot-emergency-reclaim-by-process-match.md` Phase 2 rung 2 + +//! GitHub issue #40 acceptance criteria: "Runtime verified on a real +//! Linux node: JVM launch produces a match within <100 ms." +//! +//! ## Run +//! +//! ```bash +//! # Linux only — needs CAP_NET_ADMIN. On a normal user account this +//! # will fail at Subscriber::new(). Run as root, or grant the cap on +//! # the binary, or run inside a privileged container: +//! sudo cargo test --test integration_netlink_proc -- --ignored +//! ``` +//! +//! On macOS / Windows the test compiles to a no-op and reports +//! "ignored" — there is no netlink to verify against. + +#![cfg(target_os = "linux")] + +use five_spot::netlink_proc::{NetlinkError, ProcEvent, Subscriber}; +use std::process::Command; +use std::time::{Duration, Instant}; + +/// End-to-end: open a netlink subscriber, spawn a child via Command, +/// assert a `ProcEvent::Exec` arrives within 100 ms with the child's +/// pid in the payload. +/// +/// `#[ignore]` because: +/// - Requires `CAP_NET_ADMIN` (root, file capability, or privileged +/// container) — `cargo test` cannot grant it itself. +/// - Subscribes to *all* exec events on the host. On a busy machine +/// the assertion that we see *our* pid is racy by definition; we +/// accept that risk by using a recognisable child command and +/// waiting for *any* matching event. +#[test] +#[ignore] +fn netlink_subscriber_observes_spawned_child_within_100ms() { + let mut sub = match Subscriber::new() { + Ok(s) => s, + Err(NetlinkError::Io(e)) if e.kind() == std::io::ErrorKind::PermissionDenied => { + panic!( + "Subscriber::new() failed with EPERM — this test requires CAP_NET_ADMIN. \ + Run with `sudo cargo test --test integration_netlink_proc -- --ignored` or \ + grant the cap on the test binary." + ); + } + Err(e) => panic!("Subscriber::new() failed: {e}"), + }; + + // Spawn a child whose pid we can recognise. `/bin/true` is universally + // available, exits immediately, and produces exactly one EXEC event. + let started = Instant::now(); + let mut child = Command::new("/bin/true").spawn().expect("spawn /bin/true"); + let target_pid = child.id(); + + // Drain events for up to 1 s, looking for our child's exec. + let deadline = Instant::now() + Duration::from_secs(1); + let mut saw_target = false; + let mut first_seen_at: Option = None; + while Instant::now() < deadline { + match sub.next_event() { + Ok(Some(ProcEvent::Exec { pid, .. })) if pid == target_pid => { + saw_target = true; + first_seen_at = Some(started.elapsed()); + break; + } + Ok(Some(_)) | Ok(None) => continue, + Err(e) => panic!("recv failed: {e}"), + } + } + + // Reap the child so it doesn't linger as a zombie. /bin/true exits + // immediately, so wait() returns ~instantly. + let _ = child.wait(); + + assert!( + saw_target, + "did not observe our /bin/true child (pid {target_pid}) via netlink within 1 s" + ); + let latency = first_seen_at.expect("set on success path"); + println!( + "netlink saw pid {target_pid} after {} ms", + latency.as_millis() + ); + // Acceptance criterion: 100 ms. Add slack for CI flake (kernel + // scheduler + spawn overhead). If this consistently fails on the + // floor, raise the bound but investigate first. + assert!( + latency < Duration::from_millis(100), + "netlink exec-detection latency {} ms exceeded the 100 ms acceptance criterion \ + from issue #40", + latency.as_millis() + ); +} + +/// Sanity check: the subscriber observes *some* event in 100 ms when +/// the host is doing anything at all (cron, systemd timers, login +/// shells, etc.). Less strict than the targeted child test — useful +/// as a quick smoke test that the netlink dance succeeded even when +/// running on a quiet test host. +#[test] +#[ignore] +fn netlink_subscriber_observes_at_least_one_event_within_100ms() { + let mut sub = match Subscriber::new() { + Ok(s) => s, + Err(e) => panic!("Subscriber::new() failed (need CAP_NET_ADMIN): {e}"), + }; + + // Drive at least one event by spawning a quick child. + let _ = Command::new("/bin/true").spawn().and_then(|mut c| c.wait()); + + let deadline = Instant::now() + Duration::from_millis(100); + while Instant::now() < deadline { + match sub.next_event() { + Ok(Some(_)) => return, // any event proves the subscription works + Ok(None) => continue, + Err(e) => panic!("recv failed: {e}"), + } + } + panic!("no proc events observed in 100 ms — subscription may not be active"); +}