diff --git a/Cargo.lock b/Cargo.lock index 1a32b483..58efe35b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3959,6 +3959,7 @@ dependencies = [ "metrics", "metrics-exporter-prometheus", "metrics-process", + "serde_json", "tokio", "tower", "tracing", diff --git a/crates/lance-context-metrics/Cargo.toml b/crates/lance-context-metrics/Cargo.toml index 43a5dfac..86374308 100644 --- a/crates/lance-context-metrics/Cargo.toml +++ b/crates/lance-context-metrics/Cargo.toml @@ -10,6 +10,7 @@ axum = { version = "0.8", features = ["json"] } metrics = "0.24" metrics-exporter-prometheus = { version = "0.16", default-features = false } metrics-process = "2.4" +serde_json = "1" tokio = { version = "1", features = ["time"] } tracing = "0.1" diff --git a/crates/lance-context-metrics/README.md b/crates/lance-context-metrics/README.md new file mode 100644 index 00000000..83bd1793 --- /dev/null +++ b/crates/lance-context-metrics/README.md @@ -0,0 +1,23 @@ +# Runtime resource observations + +The shared router exposes `GET /api/v1/internal/runtime-resources?nonce=TOKEN` +on both master and worker HTTP ports. It reads the current container's cgroup-v2 +`memory.current`, `memory.max`, and `memory.events` directly, without opening +stores, acquiring merge locks, running commands or contacting other services. +Each kernel file is limited to 8 KiB. Responses use `Cache-Control: no-store` and +echo an optional alphanumeric/hyphen nonce (up to 64 bytes). + +A successful response contains schema version 1, sample time, memory usage/limit, +and the raw `oom` and `oom_kill` counters. An unlimited limit is JSON `null`, not +an invented memory budget. Missing, unsupported, oversized or malformed cgroup +observations return HTTP 503; no previous successful observation is cached. +Memory pressure or prior OOM is reported as data, not hidden behind a generic +HTTP success/health flag. This endpoint does not restart writers or infer stalled +merge progress. + +Recovery callers must bind the request to the actual Pod IP/container identity, +check that identity before and after reading, verify their fresh nonce and reject +stale, failed, missing or unlimited-budget observations. Existing readiness, +expected-worker configuration and OOM/memory admission rules still apply. The +endpoint supplies evidence; it does not authorize takeover or replace native +ownership/fencing. diff --git a/crates/lance-context-metrics/src/lib.rs b/crates/lance-context-metrics/src/lib.rs index 5b565344..3f7f0b71 100644 --- a/crates/lance-context-metrics/src/lib.rs +++ b/crates/lance-context-metrics/src/lib.rs @@ -22,6 +22,8 @@ use axum::{ use metrics_exporter_prometheus::{Matcher, PrometheusBuilder, PrometheusHandle}; use metrics_process::Collector; +mod runtime_resources; + /// Explicit histogram buckets (upper bounds, seconds) for request-scale latency /// metrics — anything matching the `_duration_seconds` suffix. /// @@ -311,11 +313,12 @@ fn describe_metrics() { ); } -/// Router exposing `GET /metrics` (relative to wherever it is nested/merged). +/// Process metrics and fresh, read-only cgroup observations for recovery admission. pub fn metrics_router(handle: MetricsHandle) -> Router { Router::new() .route("/metrics", get(render)) .with_state(handle) + .merge(runtime_resources::router()) } async fn render(State(handle): State) -> impl IntoResponse { @@ -391,6 +394,18 @@ mod tests { metrics::gauge!("master_experiments").set(3.0); let app = metrics_router(handle); + let invalid_nonce = app + .clone() + .oneshot( + Request::builder() + .uri("/api/v1/internal/runtime-resources?nonce=invalid%20nonce") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(invalid_nonce.status(), StatusCode::BAD_REQUEST); + assert_eq!(invalid_nonce.headers()["cache-control"], "no-store"); let resp = app .oneshot( Request::builder() diff --git a/crates/lance-context-metrics/src/runtime_resources.rs b/crates/lance-context-metrics/src/runtime_resources.rs new file mode 100644 index 00000000..baab2ba2 --- /dev/null +++ b/crates/lance-context-metrics/src/runtime_resources.rs @@ -0,0 +1,227 @@ +//! Fresh cgroup-v2 observations for recovery admission; never a writer watchdog. + +use std::collections::{BTreeMap, HashMap}; +use std::fs::File; +use std::io::{self, Read}; +use std::path::{Path, PathBuf}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use axum::extract::{Query, State}; +use axum::http::{header, StatusCode}; +use axum::response::{IntoResponse, Response}; +use axum::{routing::get, Json, Router}; +use serde_json::json; + +const MAX_FILE_BYTES: u64 = 8192; + +#[derive(Debug, PartialEq)] +struct Snapshot { + current: u64, + limit: Option, + oom: u64, + oom_kill: u64, +} + +fn invalid() -> io::Error { + io::Error::new( + io::ErrorKind::InvalidData, + "invalid cgroup memory observation", + ) +} + +fn bounded(root: &Path, name: &str) -> io::Result { + let mut text = String::new(); + File::open(root.join(name))? + .take(MAX_FILE_BYTES + 1) + .read_to_string(&mut text)?; + if text.len() as u64 > MAX_FILE_BYTES { + return Err(invalid()); + } + Ok(text) +} + +fn snapshot(root: &Path) -> io::Result { + let current = bounded(root, "memory.current")? + .trim() + .parse::() + .map_err(|_| invalid())?; + let maximum = bounded(root, "memory.max")?; + let limit = match maximum.trim() { + "max" => None, + value => { + let limit = value.parse::().map_err(|_| invalid())?; + if limit == 0 { + return Err(invalid()); + } + Some(limit) + } + }; + let events = bounded(root, "memory.events")?; + let mut counters = BTreeMap::new(); + for line in events.lines() { + let mut fields = line.split_whitespace(); + let name = fields.next().ok_or_else(invalid)?; + let count = fields + .next() + .ok_or_else(invalid)? + .parse::() + .map_err(|_| invalid())?; + if fields.next().is_some() || counters.insert(name, count).is_some() { + return Err(invalid()); + } + } + Ok(Snapshot { + current, + limit, + oom: *counters.get("oom").ok_or_else(invalid)?, + oom_kill: *counters.get("oom_kill").ok_or_else(invalid)?, + }) +} + +pub(super) fn router() -> Router { + Router::new() + .route("/api/v1/internal/runtime-resources", get(observe)) + .with_state(PathBuf::from("/sys/fs/cgroup")) +} + +async fn observe( + State(root): State, + Query(query): Query>, +) -> Response { + let nonce = query.get("nonce").map(String::as_str).unwrap_or(""); + if nonce.len() > 64 + || !nonce + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-') + { + return ( + StatusCode::BAD_REQUEST, + [(header::CACHE_CONTROL, "no-store")], + Json(json!({"error": "invalid observation nonce"})), + ) + .into_response(); + } + // These three small kernel pseudo-files do not access a table, acquire a + // store lock, scrape peers, or run an external command. No prior sample is + // cached: missing/unsupported cgroups cannot become a healthy zero sample. + let value = match snapshot(&root) { + Ok(value) => value, + Err(_) => { + return ( + StatusCode::SERVICE_UNAVAILABLE, + [(header::CACHE_CONTROL, "no-store")], + Json(json!({"error": "cgroup v2 memory observation unavailable", "nonce": nonce})), + ) + .into_response(); + } + }; + let sampled_at = match SystemTime::now().duration_since(UNIX_EPOCH) { + Ok(time) => time.as_millis(), + Err(_) => return StatusCode::SERVICE_UNAVAILABLE.into_response(), + }; + ( + [(header::CACHE_CONTROL, "no-store")], + Json(json!({ + "schema": 1, + "nonce": nonce, + "sampled_at_unix_millis": sampled_at, + "cgroup_version": 2, + "memory_current_bytes": value.current, + "memory_max_bytes": value.limit, + "oom": value.oom, + "oom_kill": value.oom_kill, + })), + ) + .into_response() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + + struct Fixture(PathBuf); + + impl Fixture { + fn new() -> Self { + let path = std::env::temp_dir().join(format!( + "lance-runtime-resources-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + fs::create_dir(&path).unwrap(); + fs::write(path.join("memory.current"), "100\n").unwrap(); + fs::write(path.join("memory.max"), "1000\n").unwrap(); + fs::write( + path.join("memory.events"), + "low 0\nhigh 0\nmax 0\noom 2\noom_kill 1\n", + ) + .unwrap(); + Self(path) + } + } + + impl Drop for Fixture { + fn drop(&mut self) { + fs::remove_dir_all(&self.0).unwrap(); + } + } + + #[test] + fn preserves_oom_evidence_and_unlimited_is_not_a_finite_budget() { + let fixture = Fixture::new(); + assert_eq!( + snapshot(&fixture.0).unwrap(), + Snapshot { + current: 100, + limit: Some(1000), + oom: 2, + oom_kill: 1 + } + ); + fs::write(fixture.0.join("memory.max"), "max\n").unwrap(); + assert_eq!(snapshot(&fixture.0).unwrap().limit, None); + } + + #[test] + fn partial_duplicate_or_oversized_observations_fail_closed() { + let fixture = Fixture::new(); + for events in [ + "oom 0\n", + "oom 0\noom 1\noom_kill 0\n", + "oom invalid\noom_kill 0\n", + ] { + fs::write(fixture.0.join("memory.events"), events).unwrap(); + assert!(snapshot(&fixture.0).is_err()); + } + fs::write(fixture.0.join("memory.current"), "1".repeat(8193)).unwrap(); + assert!(snapshot(&fixture.0).is_err()); + } + + #[tokio::test] + async fn each_request_reads_fresh_values_and_failure_cannot_serve_old_success() { + let fixture = Fixture::new(); + let query = || Query(HashMap::from([("nonce".to_owned(), "read-123".to_owned())])); + for current in [100, 900] { + fs::write(fixture.0.join("memory.current"), current.to_string()).unwrap(); + let response = observe(State(fixture.0.clone()), query()).await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()[header::CACHE_CONTROL], "no-store"); + let bytes = axum::body::to_bytes(response.into_body(), 4096) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(value["nonce"], "read-123"); + assert_eq!(value["memory_current_bytes"], current); + assert_eq!(value["oom_kill"], 1); + } + fs::remove_file(fixture.0.join("memory.events")).unwrap(); + assert_eq!( + observe(State(fixture.0.clone()), query()).await.status(), + StatusCode::SERVICE_UNAVAILABLE + ); + } +}