diff --git a/Cargo.lock b/Cargo.lock index 0be94ad1609..664f74de6f0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -400,6 +400,28 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "async-task" version = "4.7.1" @@ -1922,13 +1944,22 @@ version = "0.1.0" dependencies = [ "aws-config", "aws-sdk-s3", + "axum", "buzz-backend-kubernetes", "chrono", + "futures-util", + "http 1.4.2", + "http-body-util", + "k8s-openapi", + "kube", "metrics", + "metrics-exporter-prometheus", "proptest", "serde", "serde_json", "tokio", + "tokio-util", + "tower 0.5.3", "tracing", ] @@ -3371,6 +3402,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "educe" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d7bc049e1bd8cdeb31b68bbd586a9464ecf9f3944af3958a7a9d0f8b9799417" +dependencies = [ + "enum-ordinalize", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "either" version = "1.16.0" @@ -3444,6 +3487,26 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "enum-ordinalize" +version = "4.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89dd01549b09589510cf0647475075d12071456586d70f5c75c98ae2a5537677" +dependencies = [ + "enum-ordinalize-derive", +] + +[[package]] +name = "enum-ordinalize-derive" +version = "4.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a65863d15a4ce2888bd2f0f543cc963d3879c3a022c8ee43f6141d479a3ac815" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "enumflags2" version = "0.7.12" @@ -4139,6 +4202,8 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ + "allocator-api2", + "equivalent", "foldhash 0.1.5", ] @@ -4389,6 +4454,17 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "hostname" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "617aaa3557aef3810a6369d0a99fac8a080891b68bd9f9812a1eeda0c0730cbd" +dependencies = [ + "cfg-if 1.0.4", + "libc", + "windows-link 0.2.1", +] + [[package]] name = "http" version = "0.2.12" @@ -5245,6 +5321,18 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "json-patch" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7421438de105a0827e44fadd05377727847d717c80ce29a229f85fd04c427b72" +dependencies = [ + "jsonptr", + "serde", + "serde_json", + "thiserror 2.0.18", +] + [[package]] name = "jsonpath-rust" version = "0.7.5" @@ -5258,6 +5346,16 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "jsonptr" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5a3cc660ba5d72bce0b3bb295bf20847ccbb40fd423f3f05b61273672e561fe" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "jsonwebtoken" version = "10.4.0" @@ -5346,6 +5444,7 @@ dependencies = [ "k8s-openapi", "kube-client", "kube-core", + "kube-runtime", ] [[package]] @@ -5394,6 +5493,7 @@ dependencies = [ "derive_more", "form_urlencoded", "http 1.4.2", + "json-patch", "k8s-openapi", "serde", "serde-value", @@ -5401,6 +5501,33 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "kube-runtime" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6aea4de4b562c5cc89ab10300bb63474ae1fa57ff5a19275f2e26401a323e3fd" +dependencies = [ + "ahash", + "async-broadcast", + "async-stream", + "backon", + "educe", + "futures", + "hashbrown 0.15.5", + "hostname", + "json-patch", + "k8s-openapi", + "kube-client", + "parking_lot", + "pin-project", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "lab" version = "0.11.0" @@ -11200,6 +11327,7 @@ dependencies = [ "futures-sink", "futures-util", "pin-project-lite", + "slab", "tokio", ] diff --git a/crates/buzz-sandbox-reaper/Cargo.toml b/crates/buzz-sandbox-reaper/Cargo.toml index 7e700db090b..b6565f3ce6b 100644 --- a/crates/buzz-sandbox-reaper/Cargo.toml +++ b/crates/buzz-sandbox-reaper/Cargo.toml @@ -14,14 +14,24 @@ path = "src/main.rs" # never in buzz-backend-kubernetes, whose binary ships inside the desktop app. [dependencies] buzz-backend-kubernetes = { path = "../buzz-backend-kubernetes" } +kube = { workspace = true, features = ["runtime"] } +k8s-openapi = { workspace = true } tokio = { workspace = true } +tokio-util = { workspace = true } +futures-util = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } chrono = { workspace = true } +axum = { workspace = true } +http = "1" +http-body-util = "0.1" metrics = { workspace = true } +metrics-exporter-prometheus = { workspace = true } tracing = { workspace = true } aws-config = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } aws-sdk-s3 = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } [dev-dependencies] +tower = { workspace = true } +tokio = { workspace = true, features = ["test-util"] } proptest = { workspace = true } diff --git a/crates/buzz-sandbox-reaper/src/main.rs b/crates/buzz-sandbox-reaper/src/main.rs index 18d1c6a407b..8596b4ad3cb 100644 --- a/crates/buzz-sandbox-reaper/src/main.rs +++ b/crates/buzz-sandbox-reaper/src/main.rs @@ -8,5 +8,7 @@ mod config; mod evidence; mod plan; +mod reconcile; +mod server; fn main() {} diff --git a/crates/buzz-sandbox-reaper/src/reconcile.rs b/crates/buzz-sandbox-reaper/src/reconcile.rs new file mode 100644 index 00000000000..c8cda7d5a67 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/reconcile.rs @@ -0,0 +1,786 @@ +//! Reconciliation of one developer namespace. +//! +//! Each namespace has one loop task that owns its [`Memory`]: a periodic pass +//! over the Sandboxes, and one Pod watch whose only job is to capture +//! termination-message checkpoint receipts before a deleted Pod takes them +//! with it. +//! +//! Reads: Sandboxes (with the apiserver `Date` as the only clock) and Pods. +//! Writes: one CAS merge patch per tombstone, a precondition-guarded delete +//! per expired tombstone, and Events. Never Secrets, never `pods/exec`, never +//! a create or a `Running` write. A CAS conflict re-reads the Sandbox and +//! re-classifies it rather than retrying a stale decision. Every write is a +//! single apiserver request, so a pass cancelled at shutdown leaves nothing +//! half-done: the next process re-observes and re-decides. + +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::Arc; +use std::time::Duration; + +use buzz_backend_kubernetes::lifecycle::{ + annotation, classify, sandbox_resource, Container, Observation, PodObservation, State, + AGENT_CONTAINER, CHECKPOINT, CHECKPOINTING, CHECKPOINTING_DISABLED, ENDED_AT, ENDED_REASON, + GENERATION, INITIAL_POD, LABEL_BINDING_VERSION, LABEL_MANAGED_BY, LIFECYCLE, LIFECYCLE_ENDED, + MANAGED_BY, RECOVERED_FROM, RESTORE_CHECKPOINT, SANDBOX_BINDING_VERSION, +}; +use chrono::{DateTime, SecondsFormat, Utc}; +use futures_util::StreamExt; +use k8s_openapi::api::core::v1::{Event, EventSource, ObjectReference, Pod}; +use kube::api::{Api, DeleteParams, ListParams, Patch, PatchParams, PostParams, Preconditions}; +use kube::core::{DynamicObject, ObjectMeta}; +use kube::runtime::{watcher, WatchStreamExt}; +use kube::Client; +use serde_json::json; +use tokio_util::sync::CancellationToken; + +use crate::config::Namespace; +use crate::evidence::{self, CheckpointStore}; +use crate::plan::{needs_evidence, plan, Action, Evidence}; +use crate::server::Health; + +/// Re-classifications after a CAS conflict before giving up until next pass. +const MAX_ATTEMPTS: usize = 3; + +/// Bound on one pass. Each S3 probe may take 25s, so a brownout could +/// otherwise stretch a pass of several Sandboxes past the liveness window. +/// An expired pass fails (readiness reports it) and the next one starts over. +pub const PASS_DEADLINE: Duration = Duration::from_secs(60); + +// Passes start at least every max(interval, PASS_DEADLINE). The liveness +// window is at least max(3 * interval, MIN_WINDOW), so it always exceeds that +// gap as long as the deadline leaves room for one more pass within it. +const _: () = assert!(PASS_DEADLINE.as_secs() * 2 <= crate::server::MIN_WINDOW.as_secs()); + +/// Receipts retained per namespace. A developer namespace runs a handful of +/// sessions; beyond this the watch is being flooded and new receipts are +/// dropped (their sessions fall back to an S3 listing of their own prefix, +/// as after a restart). +const MAX_RECEIPTS: usize = 256; + +/// Bound on one Sandbox list response (see `list_sandboxes`). +const MAX_LIST_BYTES: usize = 64 * 1024 * 1024; + +/// The kubelet caps termination messages at 4 KiB; anything larger is not one. +const MAX_RECEIPT_BYTES: usize = 4096; + +/// Every `state` label value, so a state that empties is reported as 0. +pub const STATE_LABELS: [&str; 11] = [ + "ignored", + "binding", + "bind_abandoned", + "binding_incomplete", + "active", + "draining", + "ending", + "completed", + "replaced", + "lost", + "ended", +]; + +/// Label selector for objects the provider marks as Sandbox sessions. +fn selector() -> String { + format!("{LABEL_MANAGED_BY}={MANAGED_BY},{LABEL_BINDING_VERSION}={SANDBOX_BINDING_VERSION}") +} + +/// Per-namespace memory across passes. Losing it (a restart) restarts every +/// timer from the next observation, which only delays decisions. It also +/// loses the receipts of Pods deleted since; for those sessions the +/// checkpoint comes from an S3 listing of the session prefix instead (see +/// `own_evidence`), never from guessing `missing`. +#[derive(Default)] +pub struct Memory { + /// Sandbox uid → first apiserver time its bound Pod was seen absent. + pod_missing_since: HashMap>, + /// Sandbox uid → first apiserver time it was seen Completed. + completed_since: HashMap>, + /// Pod uid → the agent container's raw termination message. Untrusted + /// until [`evidence::from_receipt`] binds it to the session prefix and S3. + receipts: HashMap, + /// (Sandbox uid, Event reason) already reported. + reported: HashSet<(String, &'static str)>, +} + +impl Memory { + /// Record a terminated agent container's termination message. Fed by + /// the Pod watch and by every pass, so a Pod deleted between passes + /// (a stop, a replacement) still leaves its receipt behind. + pub fn observe_pod(&mut self, pod: &Pod) { + let Some(uid) = pod.metadata.uid.as_deref() else { + return; + }; + let message = pod + .status + .as_ref() + .and_then(|s| s.container_statuses.as_ref()) + .and_then(|containers| containers.iter().find(|c| c.name == AGENT_CONTAINER)) + .and_then(|c| c.state.as_ref()?.terminated.as_ref()) + .filter(|terminated| terminated.exit_code == 0) + .and_then(|terminated| terminated.message.as_deref()) + .filter(|m| !m.trim().is_empty() && m.len() <= MAX_RECEIPT_BYTES); + let Some(message) = message else { + return; + }; + if self.receipts.len() >= MAX_RECEIPTS && !self.receipts.contains_key(uid) { + tracing::warn!(pod_uid = uid, "receipt memory full; dropping receipt"); + return; + } + self.receipts.insert(uid.to_string(), message.to_string()); + } + + /// Forget everything about Sandboxes that no longer exist, and receipts + /// of Pods no live Sandbox is bound to. + fn retain(&mut self, live: &HashSet, bound_pods: &HashSet) { + self.pod_missing_since.retain(|uid, _| live.contains(uid)); + self.completed_since.retain(|uid, _| live.contains(uid)); + self.receipts.retain(|uid, _| bound_pods.contains(uid)); + self.reported.retain(|(uid, _)| live.contains(uid)); + } +} + +/// Counts from one pass, for metrics and logs. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct Summary { + pub states: BTreeMap<&'static str, u64>, + pub tombstoned: u64, + pub deleted: u64, + pub held: u64, + pub ignored: u64, + pub conflicts: u64, +} + +fn state_label(state: &State) -> &'static str { + match state { + State::Ignored(_) => "ignored", + State::Binding { abandoned: false } => "binding", + State::Binding { abandoned: true } => "bind_abandoned", + State::BindingIncomplete { .. } => "binding_incomplete", + State::Active => "active", + State::Draining => "draining", + State::Ending { .. } => "ending", + State::Completed { .. } => "completed", + State::Replaced => "replaced", + State::Lost { .. } => "lost", + State::Ended { .. } => "ended", + } +} + +enum ApplyError { + Conflict, + Other(String), +} + +fn reason_is(error: &kube::Error, reason: &str) -> bool { + matches!(error, kube::Error::Api(e) if e.reason == reason) +} + +/// Record one pass in metrics. +pub fn record_pass(namespace: &str, result: &Result) { + let outcome = if result.is_ok() { "ok" } else { "error" }; + metrics::counter!("buzz_sandbox_reaper_passes_total", + "namespace" => namespace.to_string(), "result" => outcome) + .increment(1); + match result { + Ok(summary) => { + for state in STATE_LABELS { + let count = summary.states.get(state).copied().unwrap_or_default(); + metrics::gauge!("buzz_sandbox_reaper_sessions", + "namespace" => namespace.to_string(), "state" => state) + .set(count as f64); + } + metrics::gauge!("buzz_sandbox_reaper_checkpoint_holds", + "namespace" => namespace.to_string()) + .set(summary.held as f64); + metrics::gauge!("buzz_sandbox_reaper_last_success_timestamp_seconds", + "namespace" => namespace.to_string()) + .set(chrono::Utc::now().timestamp() as f64); + } + Err(error) => tracing::warn!(namespace, %error, "reconcile pass failed"), + } +} + +/// The loop for one namespace: a pass every `interval`, and the Pod watch +/// feeding [`Memory::observe_pod`] in between. Returns only on `cancel`; a +/// pass in flight finishes first (within [`PASS_DEADLINE`]). +pub async fn run_namespace( + reconciler: Arc>, + name: String, + namespace: Namespace, + interval: Duration, + health: Arc, + cancel: CancellationToken, +) { + let pods: Api = Api::namespaced(reconciler.client.clone(), &name); + let mut receipts = watcher(pods, watcher::Config::default().labels(&pod_selector())) + .default_backoff() + .boxed(); + let mut memory = Memory::default(); + let mut ticker = tokio::time::interval(interval); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tokio::select! { + biased; + () = cancel.cancelled() => return, + _ = ticker.tick() => { + // Metrics, logs, and health record the outcome. + let _ = reconciler.pass_with_deadline(&name, &namespace, &mut memory, &health).await; + } + event = receipts.next() => match event { + Some(Ok(watcher::Event::Apply(pod) | watcher::Event::InitApply(pod) + | watcher::Event::Delete(pod))) => memory.observe_pod(&pod), + Some(Ok(watcher::Event::Init | watcher::Event::InitDone)) => {} + Some(Err(error)) => { + // The backoff re-establishes the watch; passes still run. + tracing::warn!(namespace = name, %error, "Pod watch failed; retrying"); + metrics::counter!("buzz_sandbox_reaper_watch_errors_total", + "namespace" => name.clone()).increment(1); + } + // The watcher stream never ends; if it did, passes carry on + // and receipts come from passes alone. + None => receipts = futures_util::stream::pending().boxed(), + }, + } + } +} + +fn pod_selector() -> String { + format!("{LABEL_MANAGED_BY}={MANAGED_BY},{LABEL_BINDING_VERSION}={SANDBOX_BINDING_VERSION}") +} + +/// Reconciler for every configured namespace. +pub struct Reconciler { + pub client: Client, + pub store: S, + pub dry_run: bool, +} + +impl Reconciler { + fn sandboxes(&self, namespace: &str) -> Api { + Api::namespaced_with(self.client.clone(), namespace, &sandbox_resource()) + } + + /// List managed Sandboxes and return the apiserver clock from the same + /// response (quorum read: resourceVersion unset). + /// + /// Neither this list nor the Pod list is paginated: decisions need the + /// complete set, and one developer namespace holds that developer's + /// sessions plus tombstones for 30 days, far below a page of objects. A + /// count limit would either truncate (unsafe) or stop all reaping, + /// including tombstone expiry, exactly when the namespace is largest. The + /// body read is still byte-bounded, failing the pass rather than reading + /// without limit. + async fn list_sandboxes( + &self, + namespace: &str, + ) -> Result<(Vec, Option>), String> { + let url = format!("/apis/agents.x-k8s.io/v1beta1/namespaces/{namespace}/sandboxes"); + let request = kube::core::Request::new(url) + .list(&ListParams::default().labels(&selector())) + .map_err(|e| format!("build Sandbox list: {e}"))?; + let (parts, body) = request.into_parts(); + let response = self + .client + .send(http::Request::from_parts(parts, body.into())) + .await + .map_err(|e| format!("list Sandboxes in {namespace}: {e}"))?; + if !response.status().is_success() { + return Err(format!( + "list Sandboxes in {namespace}: HTTP {}", + response.status() + )); + } + let now = response + .headers() + .get(http::header::DATE) + .and_then(|v| v.to_str().ok()) + .and_then(|v| DateTime::parse_from_rfc2822(v).ok()) + .map(|v| v.with_timezone(&Utc)); + let body = http_body_util::Limited::new(response.into_body(), MAX_LIST_BYTES); + let bytes = http_body_util::BodyExt::collect(body) + .await + .map_err(|e| format!("read Sandbox list: {e}"))? + .to_bytes(); + let list: kube::core::ObjectList = + serde_json::from_slice(&bytes).map_err(|e| format!("decode Sandbox list: {e}"))?; + Ok((list.items, now)) + } + + /// One loop pass: liveness is recorded when it starts, readiness when it + /// succeeds within [`PASS_DEADLINE`]. + pub async fn pass_with_deadline( + &self, + namespace: &str, + config: &Namespace, + memory: &mut Memory, + health: &Health, + ) -> Result { + health.started(namespace); + let result = tokio::time::timeout(PASS_DEADLINE, self.pass(namespace, config, memory)) + .await + .unwrap_or_else(|_| { + Err(format!( + "pass over {namespace} exceeded {}s; retry next pass", + PASS_DEADLINE.as_secs() + )) + }); + record_pass(namespace, &result); + health.record(namespace, result.is_ok()); + result + } + + /// One pass over `namespace`. + pub async fn pass( + &self, + namespace: &str, + config: &Namespace, + memory: &mut Memory, + ) -> Result { + let (sandboxes, now) = self.list_sandboxes(namespace).await?; + // The apiserver clock is the only clock: a skewed local clock must + // never age a session into a tombstone or a deletion. + let now = now.ok_or("apiserver response carried no usable Date; skipping pass")?; + let pods: Api = Api::namespaced(self.client.clone(), namespace); + let listed = pods + .list(&ListParams::default().labels(&pod_selector())) + .await + .map_err(|e| format!("list Pods in {namespace}: {e}"))? + .items; + for pod in &listed { + memory.observe_pod(pod); + } + let mut by_name: HashMap = listed + .into_iter() + .filter_map(|pod| Some((pod.metadata.name.clone()?, pod))) + .collect(); + let mut summary = Summary::default(); + let mut failures = 0; + let mut live = HashSet::new(); + let mut bound_pods = HashSet::new(); + for sandbox in sandboxes { + let Some(name) = sandbox.metadata.name.clone() else { + continue; + }; + live.extend(sandbox.metadata.uid.clone()); + bound_pods.extend(annotation(&sandbox, INITIAL_POD).map(str::to_string)); + let pod = by_name.remove(&name); + if let Err(error) = self + .reconcile_one(namespace, config, sandbox, pod, now, memory, &mut summary) + .await + { + failures += 1; + tracing::warn!(namespace, sandbox = name, %error, "reconcile failed; retrying next pass"); + metrics::counter!("buzz_sandbox_reaper_errors_total", "namespace" => namespace.to_string()) + .increment(1); + } + } + memory.retain(&live, &bound_pods); + if failures > 0 { + return Err(format!( + "{failures} Sandbox reconciliations failed in {namespace}; retry next pass" + )); + } + Ok(summary) + } + + #[expect( + clippy::too_many_arguments, + reason = "one call site per pass; a context struct would only rename the same values" + )] + async fn reconcile_one( + &self, + namespace: &str, + config: &Namespace, + mut sandbox: DynamicObject, + mut pod: Option, + now: DateTime, + memory: &mut Memory, + summary: &mut Summary, + ) -> Result<(), String> { + let name = sandbox.metadata.name.clone().unwrap_or_default(); + for _ in 0..MAX_ATTEMPTS { + let uid = sandbox.metadata.uid.clone().unwrap_or_default(); + let missing_since = if annotation(&sandbox, INITIAL_POD).is_some() && pod.is_none() { + Some(*memory.pod_missing_since.entry(uid.clone()).or_insert(now)) + } else { + memory.pod_missing_since.remove(&uid); + None + }; + let observation = Observation::from_objects(&sandbox, pod.as_ref(), now, missing_since); + let state = classify(&observation, &config.owner_pubkey); + let evidence = self + .evidence(&state, &sandbox, &observation, config, memory) + .await; + if evidence == Evidence::Unavailable { + return Err("checkpoint evidence is unavailable; defer reconciliation".into()); + } + let hold_since = match state { + State::Completed { .. } => { + *memory.completed_since.entry(uid.clone()).or_insert(now) + } + _ => now, + }; + let action = plan(&state, &evidence, now, hold_since); + match self + .apply(namespace, &sandbox, &state, &action, now, memory, summary) + .await + { + Ok(()) => { + *summary.states.entry(state_label(&state)).or_default() += 1; + return Ok(()); + } + Err(ApplyError::Other(error)) => return Err(error), + Err(ApplyError::Conflict) => { + summary.conflicts += 1; + metrics::counter!("buzz_sandbox_reaper_cas_conflicts_total", "namespace" => namespace.to_string()) + .increment(1); + let Some(fresh) = self + .sandboxes(namespace) + .get_opt(&name) + .await + .map_err(|e| format!("re-read Sandbox: {e}"))? + else { + return Ok(()); + }; + sandbox = fresh; + pod = Api::::namespaced(self.client.clone(), namespace) + .get_opt(&name) + .await + .map_err(|e| format!("re-read Pod: {e}"))?; + if let Some(pod) = &pod { + memory.observe_pod(pod); + } + } + } + } + Err("Sandbox kept changing; deferring to next pass".into()) + } + + /// Checkpoint evidence for a session about to be tombstoned: its own + /// receipt, or else the checkpoint a recovered session was restored from, + /// so a failed restore or stop never orphans the original work. + async fn evidence( + &self, + state: &State, + sandbox: &DynamicObject, + observation: &Observation, + config: &Namespace, + memory: &Memory, + ) -> Evidence { + if !needs_evidence(state) { + return Evidence::Missing; + } + if annotation(sandbox, CHECKPOINTING) == Some(CHECKPOINTING_DISABLED) { + return Evidence::Unconfigured; + } + let own = self + .own_evidence(state, sandbox, observation, config, memory) + .await; + match ( + own, + annotation(sandbox, RESTORE_CHECKPOINT), + annotation(sandbox, RECOVERED_FROM), + ) { + (Evidence::Missing, Some(restore), Some(_)) => { + evidence::inherited(&self.store, restore, &config.checkpoint_prefix).await + } + (own, ..) => own, + } + } + + /// The session's own checkpoint. The receipt is the bound (first) Pod's + /// termination message: read from the live Pod when it is still there, + /// otherwise from what the watch captured. The session prefix comes from + /// config and the Sandbox's (validated) generation, never the Pod. + /// + /// When the bound Pod is gone and memory holds no receipt for it, the + /// receipt is unknown, not absent: a reaper restart or a watch gap may + /// have lost it. The newest object S3 lists in the session prefix then + /// stands in for it ([`evidence::latest`]). A failed listing defers the + /// session; only an empty listing records `missing`. No grace period is + /// needed first: the Pod is gone, so its harness has stopped writing, and + /// S3 listings are strongly consistent with completed uploads. + async fn own_evidence( + &self, + state: &State, + sandbox: &DynamicObject, + observation: &Observation, + config: &Namespace, + memory: &Memory, + ) -> Evidence { + // Abandoned or never-started bindings never ran a harness. + if matches!( + state, + State::Binding { .. } | State::BindingIncomplete { .. } + ) { + return Evidence::Missing; + } + let Some(initial) = observation.initial_pod.as_deref() else { + return Evidence::Missing; + }; + let Some(session_prefix) = annotation(sandbox, GENERATION) + .and_then(|generation| evidence::session_prefix(&config.checkpoint_prefix, generation)) + else { + return Evidence::Missing; + }; + let message = match &observation.pod { + Some(PodObservation { uid, container, .. }) if uid == initial => match container { + Container::Terminated { message, .. } => message.as_deref(), + _ => None, + }, + _ => match memory.receipts.get(initial) { + Some(message) => Some(message.as_str()), + None => return evidence::latest(&self.store, &session_prefix).await, + }, + }; + evidence::from_receipt(&self.store, message, &session_prefix).await + } + + #[expect( + clippy::too_many_arguments, + reason = "one call site per pass; a context struct would only rename the same values" + )] + async fn apply( + &self, + namespace: &str, + sandbox: &DynamicObject, + state: &State, + action: &Action, + now: DateTime, + memory: &mut Memory, + summary: &mut Summary, + ) -> Result<(), ApplyError> { + let name = sandbox.metadata.name.as_deref().unwrap_or_default(); + match action { + Action::None => Ok(()), + Action::Ignore(why) => { + summary.ignored += 1; + self.report( + namespace, + sandbox, + memory, + "SandboxIgnored", + "Warning", + &format!( + "buzz-sandbox-reaper leaves this Sandbox alone: {}", + why.as_str() + ), + now, + ) + .await; + Ok(()) + } + Action::Hold { until } => { + summary.held += 1; + self.report( + namespace, + sandbox, + memory, + "CheckpointMissing", + "Warning", + &format!( + "session completed without a verified checkpoint; holding until {}", + until.to_rfc3339_opts(SecondsFormat::Secs, true) + ), + now, + ) + .await; + Ok(()) + } + Action::Tombstone { reason, checkpoint } => { + tracing::info!( + namespace, + sandbox = name, + state = state_label(state), + reason = reason.as_str(), + checkpoint, + dry_run = self.dry_run, + "tombstoning session" + ); + if self.dry_run { + return Ok(()); + } + let patch = tombstone_patch(sandbox, reason.as_str(), checkpoint, now); + match self + .sandboxes(namespace) + .patch(name, &PatchParams::default(), &Patch::Merge(&patch)) + .await + { + Ok(_) => {} + Err(e) if reason_is(&e, "Conflict") => return Err(ApplyError::Conflict), + Err(e) if reason_is(&e, "NotFound") => return Ok(()), + Err(e) => return Err(ApplyError::Other(format!("tombstone Sandbox: {e}"))), + } + summary.tombstoned += 1; + metrics::counter!("buzz_sandbox_reaper_tombstones_total", + "namespace" => namespace.to_string(), "reason" => reason.as_str()) + .increment(1); + self.report( + namespace, + sandbox, + memory, + "SessionEnded", + "Normal", + &format!( + "session ended ({}); checkpoint: {checkpoint}", + reason.as_str() + ), + now, + ) + .await; + Ok(()) + } + Action::Delete => { + tracing::info!( + namespace, + sandbox = name, + dry_run = self.dry_run, + "deleting expired tombstone" + ); + if self.dry_run { + return Ok(()); + } + let params = DeleteParams { + preconditions: Some(Preconditions { + uid: sandbox.metadata.uid.clone(), + resource_version: sandbox.metadata.resource_version.clone(), + }), + ..Default::default() + }; + match self.sandboxes(namespace).delete(name, ¶ms).await { + Ok(_) => {} + Err(e) if reason_is(&e, "Conflict") => return Err(ApplyError::Conflict), + Err(e) if reason_is(&e, "NotFound") => return Ok(()), + Err(e) => return Err(ApplyError::Other(format!("delete tombstone: {e}"))), + } + summary.deleted += 1; + metrics::counter!("buzz_sandbox_reaper_tombstones_deleted_total", + "namespace" => namespace.to_string()) + .increment(1); + Ok(()) + } + } + } + + /// Emit one Event per (Sandbox, reason). Advisory: a failure is logged and + /// retried on the next pass because it is not recorded as reported. + #[expect( + clippy::too_many_arguments, + reason = "one call site per pass; a context struct would only rename the same values" + )] + async fn report( + &self, + namespace: &str, + sandbox: &DynamicObject, + memory: &mut Memory, + reason: &'static str, + kind: &str, + message: &str, + now: DateTime, + ) { + let uid = sandbox.metadata.uid.clone().unwrap_or_default(); + if self.dry_run || memory.reported.contains(&(uid.clone(), reason)) { + return; + } + let name = sandbox.metadata.name.clone().unwrap_or_default(); + let time = k8s_openapi::apimachinery::pkg::apis::meta::v1::Time(now); + let event = Event { + metadata: ObjectMeta { + generate_name: Some(format!("{name}.")), + namespace: Some(namespace.to_string()), + ..Default::default() + }, + involved_object: ObjectReference { + api_version: Some("agents.x-k8s.io/v1beta1".into()), + kind: Some("Sandbox".into()), + name: Some(name), + namespace: Some(namespace.to_string()), + uid: Some(uid.clone()), + ..Default::default() + }, + reason: Some(reason.into()), + message: Some(message.into()), + type_: Some(kind.into()), + source: Some(EventSource { + component: Some("buzz-sandbox-reaper".into()), + ..Default::default() + }), + reporting_component: Some("buzz-sandbox-reaper".into()), + reporting_instance: Some("buzz-sandbox-reaper".into()), + first_timestamp: Some(time.clone()), + last_timestamp: Some(time), + count: Some(1), + ..Default::default() + }; + let events: Api = Api::namespaced(self.client.clone(), namespace); + match events.create(&PostParams::default(), &event).await { + Ok(_) => { + memory.reported.insert((uid, reason)); + } + Err(error) => tracing::warn!(namespace, reason, %error, "could not record Event"), + } + } +} + +/// The one write that ends a session: a JSON merge patch guarded by the +/// observed resourceVersion. It only ever sets `Suspended`. +pub fn tombstone_patch( + sandbox: &DynamicObject, + reason: &str, + checkpoint: &str, + now: DateTime, +) -> serde_json::Value { + json!({ + "metadata": { + "resourceVersion": sandbox.metadata.resource_version, + "annotations": { + (LIFECYCLE): LIFECYCLE_ENDED, + (ENDED_REASON): reason, + (ENDED_AT): now.to_rfc3339_opts(SecondsFormat::Secs, true), + (CHECKPOINT): checkpoint, + }, + }, + "spec": {"operatingMode": "Suspended"}, + }) +} + +#[cfg(test)] +mod deadline_tests { + use super::*; + use crate::evidence::tests::FakeStore; + use crate::server::MIN_WINDOW; + + /// A pass whose dependency never answers ends at the deadline as a + /// failure. Liveness was recorded when it started; readiness was not. + #[tokio::test(start_paused = true)] + async fn a_hung_pass_fails_at_the_deadline_and_stays_live() { + let hung = tower::service_fn(|_: http::Request| { + std::future::pending::< + Result, std::convert::Infallible>, + >() + }); + let reconciler = Reconciler { + client: Client::new(hung, "default"), + store: FakeStore::default(), + dry_run: false, + }; + let config = Namespace { + developer: "dev".into(), + owner_pubkey: "a".repeat(64), + checkpoint_prefix: "dev/".into(), + }; + let health = Health::new(["ns".to_string()], Duration::from_secs(30)); + // Past the startup grace, liveness depends on recorded progress. + tokio::time::advance(MIN_WINDOW * 2).await; + let started = tokio::time::Instant::now(); + let result = tokio::time::timeout( + PASS_DEADLINE * 2, + reconciler.pass_with_deadline("ns", &config, &mut Memory::default(), &health), + ) + .await + .expect("PASS_DEADLINE bounds the pass"); + assert!(result.unwrap_err().contains("exceeded")); + assert_eq!(started.elapsed(), PASS_DEADLINE); + assert!(health.live()); + assert!(!health.ready()); + } +} diff --git a/crates/buzz-sandbox-reaper/src/server.rs b/crates/buzz-sandbox-reaper/src/server.rs new file mode 100644 index 00000000000..dc3b98c2030 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/server.rs @@ -0,0 +1,195 @@ +//! The reaper's only listener: `/metrics`, `/healthz`, `/readyz`. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use axum::extract::State; +use axum::http::StatusCode; +use axum::routing::get; +use axum::Router; +use metrics_exporter_prometheus::PrometheusHandle; +use tokio::time::Instant; + +/// Shortest probe window, whatever the pass interval. A loop starts a pass +/// at least every `max(interval, PASS_DEADLINE)`, which this window (at +/// least three intervals) always exceeds; see `reconcile::PASS_DEADLINE`. +pub const MIN_WINDOW: Duration = Duration::from_secs(120); + +/// Per-namespace loop progress. Liveness tracks pass *starts*, so a slow +/// dependency (S3, the apiserver) fails only readiness; restarting the +/// process for it would only discard the reaper's memory. +pub struct Health { + started: Instant, + /// A loop that has not attempted a pass within this window is wedged. + window: Duration, + namespaces: Mutex>, +} + +#[derive(Clone, Copy, Default)] +struct Progress { + attempted: Option, + succeeded: Option, +} + +impl Health { + /// `interval` is the pass interval; both probes allow three missed passes. + pub fn new(namespaces: impl IntoIterator, interval: Duration) -> Self { + Self { + started: Instant::now(), + window: (interval * 3).max(MIN_WINDOW), + namespaces: Mutex::new( + namespaces + .into_iter() + .map(|ns| (ns, Progress::default())) + .collect(), + ), + } + } + + /// Record that a pass started: the loop is alive. + pub fn started(&self, namespace: &str) { + if let Ok(mut namespaces) = self.namespaces.lock() { + namespaces + .entry(namespace.to_string()) + .or_default() + .attempted = Some(Instant::now()); + } + } + + /// Record one finished pass. Only [`Health::started`] proves liveness. + pub fn record(&self, namespace: &str, ok: bool) { + if !ok { + return; + } + if let Ok(mut namespaces) = self.namespaces.lock() { + namespaces + .entry(namespace.to_string()) + .or_default() + .succeeded = Some(Instant::now()); + } + } + + fn check(&self, pick: impl Fn(&Progress) -> Option) -> bool { + let Ok(namespaces) = self.namespaces.lock() else { + return false; + }; + let fresh = |at: Option| at.is_some_and(|at| at.elapsed() <= self.window); + namespaces + .values() + .all(|progress| fresh(pick(progress)) || self.started.elapsed() <= self.window) + } + + /// Every namespace loop is still running passes. + pub fn live(&self) -> bool { + self.check(|p| p.attempted) + } + + /// Every namespace had a successful pass recently. + pub fn ready(&self) -> bool { + let Ok(namespaces) = self.namespaces.lock() else { + return false; + }; + namespaces + .values() + .all(|p| p.succeeded.is_some_and(|at| at.elapsed() <= self.window)) + } +} + +#[derive(Clone)] +struct App { + health: Arc, + metrics: PrometheusHandle, +} + +/// The router served on `--listen`. +pub fn router(health: Arc, metrics: PrometheusHandle) -> Router { + Router::new() + .route( + "/metrics", + get(|State(app): State| async move { app.metrics.render() }), + ) + .route( + "/healthz", + get(|State(app): State| async move { + if app.health.live() { + (StatusCode::OK, "ok") + } else { + (StatusCode::SERVICE_UNAVAILABLE, "reconcile loop stalled") + } + }), + ) + .route( + "/readyz", + get(|State(app): State| async move { + if app.health.ready() { + (StatusCode::OK, "ok") + } else { + (StatusCode::SERVICE_UNAVAILABLE, "no recent successful pass") + } + }), + ) + .with_state(App { health, metrics }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn readiness_needs_a_success_in_every_namespace() { + let health = Health::new(["a".to_string(), "b".to_string()], Duration::from_secs(30)); + assert!(health.live(), "a fresh process is live within its window"); + assert!(!health.ready()); + health.record("a", true); + assert!(!health.ready(), "namespace b has not succeeded"); + health.record("b", false); + assert!(!health.ready()); + health.record("b", true); + assert!(health.ready()); + } + + /// A pass stuck on S3 for longer than the window keeps the process live + /// (it started recently) but not ready (nothing succeeded recently). + #[tokio::test(start_paused = true)] + async fn a_slow_dependency_fails_readiness_not_liveness() { + let health = Health::new(["a".to_string()], Duration::from_secs(30)); + health.record("a", true); + tokio::time::advance(MIN_WINDOW).await; + health.started("a"); + tokio::time::advance(Duration::from_secs(1)).await; + assert!(health.live()); + assert!(!health.ready()); + // A loop that stops starting passes is not live. + tokio::time::advance(MIN_WINDOW).await; + assert!(!health.live()); + } + + #[tokio::test] + async fn serves_metrics_and_probes() { + use tower::ServiceExt; + let recorder = metrics_exporter_prometheus::PrometheusBuilder::new().build_recorder(); + let handle = recorder.handle(); + let health = Arc::new(Health::new(["a".to_string()], Duration::from_secs(30))); + let app = router(Arc::clone(&health), handle); + let status = |path: &'static str| { + let app = app.clone(); + async move { + app.oneshot( + axum::http::Request::get(path) + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap() + .status() + } + }; + assert_eq!(status("/metrics").await, StatusCode::OK); + assert_eq!(status("/healthz").await, StatusCode::OK); + assert_eq!(status("/readyz").await, StatusCode::SERVICE_UNAVAILABLE); + health.record("a", true); + assert_eq!(status("/readyz").await, StatusCode::OK); + assert_eq!(status("/other").await, StatusCode::NOT_FOUND); + } +} diff --git a/docs/remote-agents.md b/docs/remote-agents.md index 8b79c48c386..57479b21eda 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -1569,6 +1569,21 @@ waits for a later pass. It never records `missing` because S3 failed. `timeout` or `other` for an outage, `access_denied` or `not_found_bucket` for a misconfigured role or bucket. +The reaper keeps observed receipts in memory only. A reaper restart, or a +watch gap while a stop or replacement deletes the Pod, can lose a receipt. +When the bound Pod is gone and no receipt for it was observed, the reaper +therefore treats the receipt as unknown, not as absent. It records the newest +object directly under the session's own `/` +prefix, by S3 `LastModified`, and records `missing` only when that listing is +empty. The prefix again comes from the configuration and the Sandbox, never +from the Pod. All sessions of one developer share one IAM role, so a +compromised session of the same developer could write into that prefix; it +cannot reach another developer's prefix. + +Each pass has a 60-second deadline. Liveness (`/healthz`) records when each +pass starts, so slow S3 or apiserver responses fail only readiness +(`/readyz`) and do not restart the reaper. + A recovered session that has no verified checkpoint of its own (for example, its restore failed) keeps the checkpoint it was recovered from. The reaper takes the key from the provider-written `restore-checkpoint` annotation,