From 88db35dcd6c51fb18b0ceec5abf6e8651967df05 Mon Sep 17 00:00:00 2001 From: plotarmordev <299844489+plotarmordev@users.noreply.github.com> Date: Thu, 10 Sep 2026 14:31:40 +0800 Subject: [PATCH] feat(grill-perf): add bounded opt-in provider snapshots --- crates/grill-perf/src/evidence.rs | 52 +- crates/grill-perf/src/main.rs | 1 + crates/grill-perf/src/metrics.rs | 593 ++++++++++++++ crates/grill-perf/src/model.rs | 4 + crates/grill-perf/src/run.rs | 60 +- crates/grill-perf/tests/cli.rs | 2 + crates/grill-perf/tests/support/metrics.rs | 859 +++++++++++++++++++++ docs/performance/CONTRACT.md | 79 ++ docs/performance/README.md | 56 ++ 9 files changed, 1699 insertions(+), 7 deletions(-) create mode 100644 crates/grill-perf/src/metrics.rs create mode 100644 crates/grill-perf/tests/support/metrics.rs diff --git a/crates/grill-perf/src/evidence.rs b/crates/grill-perf/src/evidence.rs index d0bbc4a..d2b3cf4 100644 --- a/crates/grill-perf/src/evidence.rs +++ b/crates/grill-perf/src/evidence.rs @@ -157,6 +157,7 @@ pub struct Loaded { pub waves: Vec>, pub states: Vec<&'static str>, pub history: crate::lifecycle::History, + pub metrics: Option, } pub fn load(root: &Path) -> Result { directory(root)?; @@ -169,6 +170,12 @@ pub fn load(root: &Path) -> Result { if let Some(deployment) = &plan.deployment { deployment.validate()?; } + if let Some(config) = &plan.metrics { + if plan.version != 2 { + return Err("metrics require execution-session provenance".into()); + } + config.validate(plan.waves.len(), plan.local_http)?; + } let expected_mechanism = (plan.workload.request.profile == Profile::VllmFixedV1) .then_some("declared-vllm-prefix-cache"); if plan.cache_mechanism.as_deref() != expected_mechanism @@ -209,6 +216,8 @@ pub fn load(root: &Path) -> Result { let plan_hash = digest(&plan_bytes); let mut waves = Vec::with_capacity(plan.waves.len()); let mut states = Vec::with_capacity(plan.waves.len()); + let mut metrics_budget = crate::metrics::Budget::default(); + let mut metrics_waves = Vec::new(); for spec in &plan.waves { let dir = wave_dir(root, spec.index); if !exists(&dir)? { @@ -255,6 +264,29 @@ pub fn load(root: &Path) -> Result { { return Err("wave lineage mismatch".into()); } + match (&plan.metrics, &wave.metrics) { + (Some(config), Some(reference)) => { + let summary = crate::metrics::load( + &dir, + reference, + config, + &plan_hash, + spec.index, + &mut metrics_budget, + )?; + if wave.attempts.iter().any(|a| { + a.timing + .dispatch_offset_us + .checked_add(a.timing.settle_us) + .is_none_or(|n| n > summary.receipt.measured_duration_us) + }) { + return Err("metrics measurement boundary contradicts wave settlement".into()); + } + metrics_waves.push(summary); + } + (None, None) => (), + _ => return Err("missing or undeclared metrics companion evidence".into()), + } for (lane, a) in wave.attempts.iter().enumerate() { if a.lane as usize != lane || a.response_bytes > workload.limits.response_bytes @@ -332,6 +364,11 @@ pub fn load(root: &Path) -> Result { } let history = crate::lifecycle::history(root, &plan, &plan_hash, &states, &waves)?; Ok(Loaded { + metrics: plan.metrics.as_ref().map(|config| crate::metrics::Summary { + config: config.clone(), + budget: metrics_budget, + waves: metrics_waves, + }), plan, waves, states, @@ -575,6 +612,12 @@ pub struct Comparison { pub reference: Option>, pub drift: Option>, pub changes: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub baseline_metrics: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub candidate_metrics: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub reference_metrics: Option, } fn change(a: Option, b: Option) -> Option { a.zip(b) @@ -678,7 +721,7 @@ fn complete_lane_observations(values: &[Vec>]) -> bool { pub fn compare(a: &Path, b: &Path, reference: Option<&Path>) -> Result { let left = load(a)?; let right = load(b)?; - let reference = reference + let mut reference = reference .map(|path| load(path).map_err(|e| format!("reference: {e}"))) .transpose()?; for (side, other) in @@ -689,9 +732,10 @@ pub fn compare(a: &Path, b: &Path, reference: Option<&Path>) -> Result) -> Result = baseline .iter() @@ -931,6 +976,9 @@ pub fn compare(a: &Path, b: &Path, reference: Option<&Path>) -> Result Self { + Self { + version: 1, + endpoint, + allowlist: NAMES.map(str::to_owned), + scope: "server-wide-full-labels-not-workload-attributed".into(), + cadence: "reserved-before-origin-and-after-all-lanes-settle".into(), + deadline_us: 2_000_000, + body_bytes: 1024 * 1024, + line_bytes: 64 * 1024, + series: 256, + labels_per_series: 16, + label_bytes_per_series: 4096, + run_budget_us: 30_000_000, + retained_raw_bytes: 16 * 1024 * 1024, + max_requests: waves * 2, + } + } + pub fn validate(&self, waves: usize, local_http: bool) -> Result<()> { + let url = wire::endpoint(&self.endpoint, local_http)?; + if *self != Self::new(url.to_string(), waves) { + return Err("unsupported metrics protocol or bounds".into()); + } + Ok(()) + } +} +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct Reference { + pub file: String, + pub sha256: String, + pub overhead_us: u64, +} +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Status { + Complete, + ParseError, + TransportError, + HttpError, + Unsupported, + Deadline, + BodyLimit, + SkippedBudget, +} +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Snapshot { + pub started_unix_ms: u64, + pub scrape_us: u64, + pub duration_us: u64, + pub charged_us: u64, + pub allowance_us: u64, + pub status: Status, + pub error: Option, + pub http_status: Option, + pub raw_bytes: usize, + pub raw_sha256: String, +} +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Receipt { + pub version: u32, + pub plan_sha256: String, + pub wave: u32, + pub before: Snapshot, + pub after: Snapshot, + pub measured_origin_unix_ms: u64, + pub measured_duration_us: u64, +} +#[derive(Clone, Copy, Debug, Default, Serialize)] +pub struct Budget { + pub requests: usize, + pub charged_us: u64, + pub duration_us: u64, + pub raw_bytes: usize, +} +impl Budget { + fn exhausted(&self, config: &Config) -> bool { + self.requests >= config.max_requests + || self.charged_us >= config.run_budget_us + || config.retained_raw_bytes.saturating_sub(self.raw_bytes) < config.body_bytes + } + fn retain(&mut self, config: &Config, snapshot: &Snapshot) -> Result<()> { + let skipped = snapshot.status == Status::SkippedBudget; + if skipped != self.exhausted(config) + || snapshot.charged_us != snapshot.duration_us + || snapshot.allowance_us + != if skipped { + 0 + } else { + config + .deadline_us + .min(config.run_budget_us.saturating_sub(self.charged_us)) + } + || (snapshot.status == Status::Deadline && snapshot.scrape_us < snapshot.allowance_us) + || snapshot.scrape_us > snapshot.duration_us + || snapshot + .http_status + .is_some_and(|s| !(100..=999).contains(&s)) + || (!skipped && (snapshot.duration_us == 0 || snapshot.scrape_us == 0)) + || (snapshot.raw_bytes > 0 && snapshot.http_status.is_none()) + || (snapshot.status == Status::Unsupported && snapshot.http_status != Some(200)) + || snapshot.raw_bytes > config.body_bytes + || (skipped + && (snapshot.duration_us != 0 + || snapshot.scrape_us != 0 + || snapshot.raw_bytes != 0 + || snapshot.http_status.is_some())) + || snapshot.error.as_ref().is_some_and(|s| s.len() > 256) + || (snapshot.status == Status::Complete) != snapshot.error.is_none() + { + return Err("invalid metrics snapshot budget or status".into()); + } + self.requests += usize::from(!skipped); + self.charged_us = self + .charged_us + .checked_add(snapshot.charged_us) + .ok_or("metrics budget overflow")?; + self.duration_us = self + .duration_us + .checked_add(snapshot.duration_us) + .ok_or("metrics duration overflow")?; + self.raw_bytes += snapshot.raw_bytes; + Ok(()) + } + pub fn finish_wave(&mut self, overhead_us: u64, snapshots_us: u64) -> Result<()> { + let publication_us = overhead_us + .checked_sub(snapshots_us) + .ok_or("metrics overhead contradicts scrape duration")?; + self.charged_us = self + .charged_us + .checked_add(publication_us) + .ok_or("metrics budget overflow")?; + self.duration_us = self + .duration_us + .checked_add(publication_us) + .ok_or("metrics duration overflow")?; + Ok(()) + } +} +pub fn unix_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis().min(u128::from(u64::MAX)) as u64) + .unwrap_or(0) +} + +pub async fn scrape( + client: &reqwest::Client, + config: &Config, + budget: &mut Budget, + dir: &Path, + side: &str, +) -> Result { + let start = Instant::now(); + let mut snapshot = Snapshot { + started_unix_ms: unix_ms(), + scrape_us: 0, + duration_us: 0, + charged_us: 0, + allowance_us: 0, + status: Status::SkippedBudget, + error: Some("whole-run telemetry budget exhausted".into()), + http_status: None, + raw_bytes: 0, + raw_sha256: String::new(), + }; + let mut raw = Vec::new(); + if !budget.exhausted(config) { + let allowance = config + .deadline_us + .min(config.run_budget_us - budget.charged_us); + snapshot.allowance_us = allowance; + let operation = async { + let mut response = client + .get(&config.endpoint) + .header("accept", "text/plain; version=0.0.4") + .header("accept-encoding", "identity") + .send() + .await + .map_err(|_| (Status::TransportError, "metrics request failed".to_owned()))?; + snapshot.http_status = Some(response.status().as_u16()); + let encoded = response + .headers() + .get_all("content-encoding") + .iter() + .any(|v| match v.to_str() { + Ok(s) => s + .split(',') + .any(|s| !s.trim().eq_ignore_ascii_case("identity")), + Err(_) => true, + }); + while let Some(chunk) = response.chunk().await.map_err(|_| { + ( + Status::TransportError, + "metrics body transfer failed".to_owned(), + ) + })? { + if wire::us(start) >= allowance { + return Err((Status::Deadline, "metrics scrape deadline expired".into())); + } + let remaining = config.body_bytes - raw.len(); + raw.extend_from_slice(&chunk[..chunk.len().min(remaining)]); + if chunk.len() > remaining { + return Err((Status::BodyLimit, "metrics body exceeds bound".into())); + } + } + if snapshot.http_status != Some(200) { + return Err(( + Status::HttpError, + "metrics HTTP status is not successful".into(), + )); + } + if encoded { + return Err(( + Status::Unsupported, + "metrics content encoding is unsupported".into(), + )); + } + parse(&raw, config).map_err(|e| (Status::ParseError, e))?; + if wire::us(start) >= allowance { + return Err((Status::Deadline, "metrics scrape deadline expired".into())); + } + Ok::<(), (Status, String)>(()) + }; + match tokio::time::timeout(Duration::from_micros(allowance), operation).await { + Ok(Ok(())) => { + snapshot.status = Status::Complete; + snapshot.error = None; + } + Ok(Err((status, error))) => { + snapshot.status = status; + snapshot.error = Some(error); + } + Err(_) => { + snapshot.status = Status::Deadline; + snapshot.error = Some("metrics scrape deadline expired".into()); + } + } + snapshot.scrape_us = wire::us(start).max(1); + } + snapshot.raw_bytes = raw.len(); + snapshot.raw_sha256 = evidence::digest(&raw); + evidence::write(&dir.join(format!("metrics-{side}.bin")), &raw)?; + if snapshot.status != Status::SkippedBudget { + snapshot.duration_us = wire::us(start).max(snapshot.scrape_us); + snapshot.charged_us = snapshot.duration_us; + } + budget.retain(config, &snapshot)?; + Ok(snapshot) +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize)] +pub struct Identity { + pub name: String, + pub labels: BTreeMap, +} +#[derive(Debug, Serialize)] +pub struct Series { + #[serde(flatten)] + pub identity: Identity, + pub value: f64, +} +fn identifier(text: &str, metric: bool) -> bool { + let mut bytes = text.bytes(); + bytes + .next() + .is_some_and(|b| b.is_ascii_alphabetic() || b == b'_' || (metric && b == b':')) + && bytes.all(|b| b.is_ascii_alphanumeric() || b == b'_' || (metric && b == b':')) +} +fn labels<'a>(mut text: &'a str, config: &Config) -> Result<(BTreeMap, &'a str)> { + let original = text.len(); + let mut labels = BTreeMap::new(); + loop { + text = text.trim_start_matches([' ', '\t']); + if let Some(rest) = text.strip_prefix('}') { + if original - rest.len() + 1 > config.label_bytes_per_series { + return Err("metrics label bytes exceed bound".into()); + } + return Ok((labels, rest)); + } + let (name, rest) = text.split_once('=').ok_or("invalid metrics label")?; + let name = name.trim_end_matches([' ', '\t']); + if !identifier(name, false) || labels.len() >= config.labels_per_series { + return Err("invalid or excessive metrics labels".into()); + } + text = rest + .trim_start_matches([' ', '\t']) + .strip_prefix('"') + .ok_or("metrics label must be quoted")?; + let mut value = String::new(); + let mut chars = text.char_indices(); + let end = loop { + let (index, c) = chars.next().ok_or("unterminated metrics label")?; + match c { + '"' => break index + 1, + '\\' => value.push(match chars.next().map(|(_, c)| c) { + Some('n') => '\n', + Some('"') => '"', + Some('\\') => '\\', + _ => return Err("invalid metrics label escape".into()), + }), + c if c.is_control() => return Err("invalid metrics label control".into()), + c => value.push(c), + } + if value.len() > config.label_bytes_per_series { + return Err("metrics label bytes exceed bound".into()); + } + }; + if labels.insert(name.to_owned(), value).is_some() { + return Err("duplicate metrics label".into()); + } + text = text[end..].trim_start_matches([' ', '\t']); + if let Some(rest) = text.strip_prefix(',') { + text = rest; + } else if !text.starts_with('}') { + return Err("invalid metrics label separator".into()); + } + } +} +pub fn parse(raw: &[u8], config: &Config) -> Result> { + if raw.len() > config.body_bytes { + return Err("metrics body exceeds bound".into()); + } + let text = std::str::from_utf8(raw).map_err(|_| "metrics body is not UTF-8")?; + let mut seen = BTreeSet::new(); + let mut series = Vec::new(); + for line in text.lines() { + if line.len() > config.line_bytes { + return Err("metrics line exceeds bound".into()); + } + let line = line.trim_matches([' ', '\t', '\r']); + if line.is_empty() || line.starts_with('#') { + continue; + } + let end = line.find(['{', ' ', '\t']).unwrap_or(line.len()); + let name = &line[..end]; + if !NAMES.contains(&name) { + continue; + } + if seen.len() >= config.series { + return Err("metrics selected series exceed bound".into()); + } + let (labels, value) = if let Some(rest) = line[end..].strip_prefix('{') { + labels(rest, config)? + } else { + (BTreeMap::new(), &line[end..]) + }; + if !value.starts_with([' ', '\t']) { + return Err("missing metrics sample separator".into()); + } + let mut fields = value.split_whitespace(); + let value = fields + .next() + .ok_or("missing metrics sample value")? + .parse::() + .map_err(|_| "invalid metrics number")?; + if !value.is_finite() { + return Err("nonfinite metrics number".into()); + } + if let Some(timestamp) = fields.next() { + timestamp + .parse::() + .map_err(|_| "invalid metrics timestamp")?; + } + if fields.next().is_some() { + return Err("unsupported metrics sample suffix".into()); + } + let identity = Identity { + name: name.into(), + labels, + }; + if !seen.insert(identity.clone()) { + return Err("duplicate metrics series".into()); + } + if NAMES[..2].contains(&name) && value < 0.0 { + return Err("negative metrics counter".into()); + } + series.push(Series { identity, value }); + } + series.sort_by(|a, b| a.identity.cmp(&b.identity)); + Ok(series) +} + +#[derive(Serialize)] +pub struct CounterDelta { + #[serde(flatten)] + pub identity: Identity, + pub before: Option, + pub after: Option, + pub delta: Option, + pub status: &'static str, +} +#[derive(Serialize)] +pub struct Acceptance { + pub labels: BTreeMap, + pub draft_delta: Option, + pub accepted_delta: Option, + pub ratio: Option, + pub status: &'static str, +} +#[derive(Serialize)] +pub struct WaveSummary { + pub receipt: Receipt, + pub overhead_us: u64, + pub before_series: Vec, + pub after_series: Vec, + pub counters: Vec, + pub acceptance: Vec, +} +#[derive(Serialize)] +pub struct Summary { + pub config: Config, + pub budget: Budget, + pub waves: Vec, +} +pub fn publish(dir: &Path, receipt: &Receipt) -> Result { + let bytes = serde_json::to_vec_pretty(receipt).map_err(|e| e.to_string())?; + if bytes.len() > RECEIPT_CAP { + return Err("metrics receipt exceeds bound".into()); + } + evidence::write(&dir.join("metrics.json"), &bytes)?; + evidence::sync(dir)?; + Ok(Reference { + file: "metrics.json".into(), + sha256: evidence::digest(&bytes), + overhead_us: 0, + }) +} +pub fn load( + dir: &Path, + reference: &Reference, + config: &Config, + plan_hash: &str, + wave: u32, + budget: &mut Budget, +) -> Result { + if reference.file != "metrics.json" { + return Err("invalid metrics companion path".into()); + } + let bytes = evidence::read(&dir.join("metrics.json"), RECEIPT_CAP)?; + if evidence::digest(&bytes) != reference.sha256 { + return Err("metrics companion hash mismatch".into()); + } + let receipt: Receipt = + serde_json::from_slice(&bytes).map_err(|e| format!("invalid metrics receipt: {e}"))?; + if receipt.version != 1 || receipt.plan_sha256 != plan_hash || receipt.wave != wave { + return Err("metrics companion lineage mismatch".into()); + } + let snapshots_us = receipt + .before + .duration_us + .checked_add(receipt.after.duration_us) + .ok_or("metrics duration overflow")?; + let mut parsed = Vec::with_capacity(2); + for (side, snapshot) in [("before", &receipt.before), ("after", &receipt.after)] { + budget.retain(config, snapshot)?; + let raw = evidence::read(&dir.join(format!("metrics-{side}.bin")), config.body_bytes)?; + if raw.len() != snapshot.raw_bytes || evidence::digest(&raw) != snapshot.raw_sha256 { + return Err("metrics raw evidence hash mismatch".into()); + } + let samples = parse(&raw, config); + match snapshot.status { + Status::Complete if snapshot.http_status == Some(200) => parsed.push(samples?), + Status::ParseError if snapshot.http_status == Some(200) && samples.is_err() => { + parsed.push(Vec::new()) + } + Status::Complete | Status::ParseError => { + return Err("metrics parser status contradicts raw evidence".into()); + } + Status::HttpError if snapshot.http_status.is_none_or(|s| s == 200) => { + return Err("invalid metrics HTTP failure".into()); + } + Status::BodyLimit if raw.len() != config.body_bytes => { + return Err("invalid metrics body limit".into()); + } + _ => parsed.push(Vec::new()), + } + } + budget.finish_wave(reference.overhead_us, snapshots_us)?; + let after_series = parsed.pop().ok_or("missing after metrics")?; + let before_series = parsed.pop().ok_or("missing before metrics")?; + let before: BTreeMap<_, _> = before_series + .iter() + .map(|s| (&s.identity, s.value)) + .collect(); + let after: BTreeMap<_, _> = after_series + .iter() + .map(|s| (&s.identity, s.value)) + .collect(); + let identities: BTreeSet<_> = before + .keys() + .chain(after.keys()) + .copied() + .filter(|id| NAMES[..2].contains(&id.name.as_str())) + .collect(); + let counters: Vec<_> = identities + .into_iter() + .map(|identity| { + let a = before.get(identity).copied(); + let b = after.get(identity).copied(); + let (delta, status) = match a.zip(b) { + Some((a, b)) if b < a => (None, "reset_observed"), + Some((a, b)) if (b - a).is_finite() => { + (Some(b - a), "nondecreasing_restart_unverified") + } + Some(_) => (None, "nonfinite_delta"), + None => (None, "missing"), + }; + CounterDelta { + identity: identity.clone(), + before: a, + after: b, + delta, + status, + } + }) + .collect(); + let mut pairs = BTreeMap::new(); + for counter in &counters { + let pair = pairs + .entry(&counter.identity.labels) + .or_insert((None, None)); + if counter.identity.name == NAMES[0] { + pair.0 = counter.delta; + } else { + pair.1 = counter.delta; + } + } + let acceptance = pairs + .into_iter() + .map(|(labels, (draft_delta, accepted_delta))| { + let (ratio, status) = match draft_delta.zip(accepted_delta) { + Some((draft, accepted)) if draft > 0.0 && accepted <= draft => ( + Some(accepted / draft), + "compatible_deltas_restart_unverified", + ), + Some((0.0, _)) => (None, "zero_denominator"), + Some(_) => (None, "inconsistent_counters"), + None => (None, "missing_or_reset"), + }; + Acceptance { + labels: labels.clone(), + draft_delta, + accepted_delta, + ratio, + status, + } + }) + .collect(); + Ok(WaveSummary { + receipt, + overhead_us: reference.overhead_us, + before_series, + after_series, + counters, + acceptance, + }) +} diff --git a/crates/grill-perf/src/model.rs b/crates/grill-perf/src/model.rs index dc49afa..b3730a5 100644 --- a/crates/grill-perf/src/model.rs +++ b/crates/grill-perf/src/model.rs @@ -361,6 +361,8 @@ pub struct Plan { pub started_unix_ms: u64, pub cache_namespace: Option, pub waves: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub metrics: Option, } #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] @@ -489,6 +491,8 @@ pub struct Wave { pub completion_tokens: Option, pub achieved_completion_tokens_per_second: Option, pub eligible: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub metrics: Option, } pub fn eligibility(a: &Attempt, r: &RequestSettings, phase: Phase) -> Vec { let mut errors = Vec::new(); diff --git a/crates/grill-perf/src/run.rs b/crates/grill-perf/src/run.rs index ca55776..10e70fa 100644 --- a/crates/grill-perf/src/run.rs +++ b/crates/grill-perf/src/run.rs @@ -1,4 +1,4 @@ -use crate::{evidence, lifecycle, model::*, wire}; +use crate::{evidence, lifecycle, metrics, model::*, wire}; use serde::{Deserialize, Serialize}; use std::io::Read; use std::path::PathBuf; @@ -43,6 +43,9 @@ pub struct Options { /// Optional JSON declarations of model revision, runtime, hardware and settings. #[arg(long)] pub deployment: Option, + /// Optional bounded server-wide diagnostics; can perturb between-wave cadence. + #[arg(long)] + pub metrics_url: Option, #[arg(long)] pub auth_env: Option, #[arg(long)] @@ -95,6 +98,14 @@ pub fn execute(o: &Options) -> Result { .max() .unwrap_or(1); let waves = workload.waves(); + let metrics = o + .metrics_url + .as_ref() + .map(|endpoint| { + wire::endpoint(endpoint, o.local_http) + .map(|url| metrics::Config::new(url.to_string(), waves.len())) + }) + .transpose()?; let normalized = serde_json::to_vec(&workload).map_err(|e| e.to_string())?; let cache_namespace = if workload.request.cache == Cache::Observe && !workload.salted() { None @@ -126,6 +137,7 @@ pub fn execute(o: &Options) -> Result { started_unix_ms: unix_ms(), cache_namespace, waves, + metrics, }; // Seed magnitude peaks at a corner of the trial/lane range and index 1023 is // the widest index, but lane 0 renders one digit narrower than lanes 10..63 in @@ -163,12 +175,20 @@ pub fn execute(o: &Options) -> Result { let plan_bytes = evidence::json(&o.out.join("plan.json"), &plan)?; evidence::sync(&o.out)?; let plan_hash = evidence::digest(&plan_bytes); - collect(&o.out, &plan, &plan_hash, 0, 0, o.json) + collect( + &o.out, + &plan, + &plan_hash, + 0, + 0, + o.json, + metrics::Budget::default(), + ) } pub fn resume(root: &std::path::Path, json: bool) -> Result { let _owner = lifecycle::ownership(root)?; - let loaded = evidence::load(root)?; + let mut loaded = evidence::load(root)?; if loaded.plan.version != 2 { return Err("legacy runs cannot be resumed; no execution-session provenance".into()); } @@ -177,6 +197,10 @@ pub fn resume(root: &std::path::Path, json: bool) -> Result { { return Err("continuation requires the original collector binary and version".into()); } + let metrics_budget = loaded + .metrics + .take() + .map_or(metrics::Budget::default(), |m| m.budget); let history = &loaded.history; if history.open || history.last_status.as_deref() != Some("paused") { return Err("continuation requires a settled cooperative pause; interrupted, failed or reserved/unsettled work is not replayed".into()); @@ -192,6 +216,7 @@ pub fn resume(root: &std::path::Path, json: bool) -> Result { history.count, history.next_wave, json, + metrics_budget, ) } @@ -202,6 +227,7 @@ fn collect( session: usize, first_wave: usize, json: bool, + mut metrics_budget: metrics::Budget, ) -> Result { let url = wire::endpoint(&plan.endpoint, plan.local_http)?; let auth = wire::credential(plan.auth_env.as_deref())?; @@ -243,6 +269,7 @@ fn collect( }; let operation = runtime.block_on(async { let client = wire::client(plan.local_http, plan.pool_max_idle_per_host)?; + let metrics_client = plan.metrics.as_ref().map(|_| wire::client(plan.local_http, 1)).transpose()?; let mut signal = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()).map_err(|_| "cannot subscribe to interrupt signal")?; let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).map_err(|_| "cannot subscribe to termination signal")?; for spec in &plan.waves[first_wave..] { @@ -263,9 +290,16 @@ fn collect( drop(admission); let reservation_publication_us = wire::us(reservation_start); let prepared: Vec<_> = reservation.requests.into_iter().map(|body| wire::request(&client, &url, auth.as_ref(), body, plan.workload.request.stream)).collect::>()?; + let telemetry_start = plan.metrics.as_ref().map(|_| Instant::now()); + let before = match (&plan.metrics, &metrics_client) { + (Some(config), Some(client)) => Some(metrics::scrape(client, config, &mut metrics_budget, &dir, "before").await?), + _ => None, + }; + let before_overhead_us = telemetry_start.map_or(0, wire::us); let (stop, cancellation) = watch::channel(interrupted()); let mut tasks = JoinSet::new(); - let preparation_us = wire::us(preparation); + let preparation_us = wire::us(preparation).saturating_sub(before_overhead_us); + let measured_origin_unix_ms = plan.metrics.as_ref().map(|_| metrics::unix_ms()); let origin = Instant::now(); for (lane, request) in prepared.into_iter().enumerate() { tasks.spawn(wire::collect(client.clone(), request, plan.workload.limits.clone(), plan.workload.request.stream, lane as u32, origin, cancellation.clone())); @@ -282,6 +316,22 @@ fn collect( } } } + let measured_duration_us = wire::us(origin); + let metrics = match (&plan.metrics, &metrics_client, before, measured_origin_unix_ms) { + (Some(config), Some(client), Some(before), Some(measured_origin_unix_ms)) => { + let after_start = Instant::now(); + let after = metrics::scrape(client, config, &mut metrics_budget, &dir, "after").await?; + let snapshots_us = before.duration_us + after.duration_us; + let mut reference = metrics::publish(&dir, &metrics::Receipt { + version: 1, plan_sha256: plan_hash.into(), wave: spec.index, + before, after, measured_origin_unix_ms, measured_duration_us, + })?; + reference.overhead_us = before_overhead_us + wire::us(after_start); + metrics_budget.finish_wave(reference.overhead_us, snapshots_us)?; + Some(reference) + } + _ => None, + }; // No hashing, filesystem writes or receipt publication while any peer is active. let publication = Instant::now(); settled.sort_by_key(|a| a.attempt.lane); @@ -298,7 +348,7 @@ fn collect( } let network_failed = attempts.iter().any(|a| a.status != Status::Complete); let (eligible, tokens, rate) = evidence::throughput(&attempts, last - first); - let wave = Wave { version: 1, plan_sha256: plan_hash.into(), reservation_sha256, spec: spec.clone(), attempts, elapsed_us: last - first, dispatch_spread_us: last_dispatch - first, preparation_us, reservation_publication_us, body_publication_us: wire::us(publication), completion_tokens: tokens, achieved_completion_tokens_per_second: rate, eligible }; + let wave = Wave { version: 1, plan_sha256: plan_hash.into(), reservation_sha256, spec: spec.clone(), attempts, elapsed_us: last - first, dispatch_spread_us: last_dispatch - first, preparation_us, reservation_publication_us, body_publication_us: wire::us(publication), completion_tokens: tokens, achieved_completion_tokens_per_second: rate, eligible, metrics }; evidence::publish(&dir, "wave.json", &wave)?; summary.wave_publication_us += wire::us(publication); summary.wave_preparation_us += preparation_us; diff --git a/crates/grill-perf/tests/cli.rs b/crates/grill-perf/tests/cli.rs index b76c75e..c79cef9 100644 --- a/crates/grill-perf/tests/cli.rs +++ b/crates/grill-perf/tests/cli.rs @@ -3770,3 +3770,5 @@ fn comparison_reference_lane_permutation_preserves_totals_but_is_ineligible() { false ); } +#[path = "support/metrics.rs"] +mod metrics_tests; diff --git a/crates/grill-perf/tests/support/metrics.rs b/crates/grill-perf/tests/support/metrics.rs new file mode 100644 index 0000000..e126c69 --- /dev/null +++ b/crates/grill-perf/tests/support/metrics.rs @@ -0,0 +1,859 @@ +use super::*; +use sha2::{Digest, Sha256}; + +const DRAFT: &str = "vllm:spec_decode_num_draft_tokens_total"; +const ACCEPTED: &str = "vllm:spec_decode_num_accepted_tokens_total"; +const RUNNING: &str = "vllm:num_requests_running"; + +struct MetricsServer { + endpoint: String, + count: Arc, + stop: Arc, + join: Option>, +} +impl MetricsServer { + fn new(handler: impl Fn(usize, &str) -> Vec + Send + Sync + 'static) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/metrics", listener.local_addr().unwrap()); + listener.set_nonblocking(true).unwrap(); + let count = Arc::new(AtomicUsize::new(0)); + let stop = Arc::new(AtomicBool::new(false)); + let (calls, stopped) = (count.clone(), stop.clone()); + let handler = Arc::new(handler); + let join = thread::spawn(move || { + let mut workers = Vec::new(); + while !stopped.load(Ordering::SeqCst) { + match listener.accept() { + Ok((mut stream, _)) => { + let (handler, calls) = (handler.clone(), calls.clone()); + workers.push(thread::spawn(move || { + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + stream + .set_write_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut bytes = Vec::new(); + let mut byte = [0]; + while !bytes.ends_with(b"\r\n\r\n") { + stream.read_exact(&mut byte).unwrap(); + bytes.push(byte[0]); + assert!(bytes.len() <= 64 * 1024); + } + let headers = String::from_utf8(bytes).unwrap(); + assert!(headers.starts_with("GET /metrics HTTP/1.1\r\n")); + assert!(!headers.to_ascii_lowercase().contains("authorization:")); + let response = handler(calls.fetch_add(1, Ordering::SeqCst), &headers); + // Bounds/deadline fixtures deliberately close before consuming the entity. + let _ = stream.write_all(&response); + })); + } + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(1)) + } + Err(e) => panic!("metrics fixture: {e}"), + } + } + for worker in workers { + worker.join().unwrap(); + } + }); + Self { + endpoint, + count, + stop, + join: Some(join), + } + } +} +impl Drop for MetricsServer { + fn drop(&mut self) { + self.stop.store(true, Ordering::SeqCst); + if let Err(error) = self.join.take().unwrap().join() + && !thread::panicking() + { + std::panic::resume_unwind(error); + } + } +} +fn response(body: &[u8]) -> Vec { + let mut bytes = format!("HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len()).into_bytes(); + bytes.extend_from_slice(body); + bytes +} +fn command(temp: &Temp, server: &Server, metrics: &str, name: &str, work: Value) -> Command { + let input = temp.path(&format!("{name}.json")); + fs::write(&input, serde_json::to_vec(&work).unwrap()).unwrap(); + let mut command = cli(); + command + .arg("run") + .arg(input) + .args([ + "--endpoint", + &server.endpoint, + "--metrics-url", + metrics, + "--model", + "fixture-model", + "--local-http", + "--json", + "--auth-env", + "GRILL_METRICS_TEST_KEY", + "--out", + ]) + .arg(temp.path(name)) + .env("GRILL_METRICS_TEST_KEY", "fixture-secret"); + command +} +fn compare(root: &Path) -> Output { + cli() + .arg("compare") + .arg(root) + .arg(root) + .arg("--json") + .output() + .unwrap() +} +fn report(root: &Path) -> Value { + let output = compare(root); + successful(&output); + serde_json::from_slice(&output.stdout).unwrap() +} +fn fast(mut stream: TcpStream, _: usize, _: Value) { + header(&mut stream, "text/event-stream"); + frame( + &mut stream, + json!({"choices":[{"index":0,"delta":{"content":"ok"}}]}), + ); + finish(&mut stream, Some(8), Some(0)); +} +fn padded() -> Vec { + let mut body = Vec::with_capacity(1024 * 1024); + while body.len() < 1024 * 1024 { + let n = (1024 * 1024 - body.len()).min(64 * 1024); + body.push(b'#'); + body.extend(std::iter::repeat_n(b'x', n - 2)); + body.push(b'\n'); + } + body +} +fn digest(bytes: &[u8]) -> String { + Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} +fn rewrite_receipt(root: &Path, edit: impl FnOnce(&mut Value)) { + let dir = root.join("wave-000000"); + let mut receipt = value(dir.join("metrics.json")); + edit(&mut receipt); + let bytes = serde_json::to_vec_pretty(&receipt).unwrap(); + fs::write(dir.join("metrics.json"), &bytes).unwrap(); + let mut wave = value(dir.join("wave.json")); + wave["metrics"]["sha256"] = json!(digest(&bytes)); + fs::write( + dir.join("wave.json"), + serde_json::to_vec_pretty(&wave).unwrap(), + ) + .unwrap(); +} + +#[test] +fn metrics_opt_out_has_no_calls_or_receipt_fields_and_loads_legacy_evidence() { + let temp = Temp::new(); + let server = Server::new(fast); + let metrics = MetricsServer::new(|_, _| panic!("opt-out must not scrape")); + successful(&run(&temp, &server, "off", &workload(1, 0, 1))); + let root = temp.path("off"); + let original = fs::read(root.join("wave-000000/wave.json")).unwrap(); + assert!(value(root.join("plan.json")).get("metrics").is_none()); + assert!(wave(&temp, "off", 0).get("metrics").is_none()); + assert!(!root.join("wave-000000/metrics-before.bin").exists()); + assert!(report(&root).get("baseline_metrics").is_none()); + assert_eq!( + fs::read(root.join("wave-000000/wave.json")).unwrap(), + original + ); + assert_eq!(metrics.count.load(Ordering::SeqCst), 0); + assert_eq!(server.count.load(Ordering::SeqCst), 1); + let legacy = temp.path("legacy"); + fs::create_dir(&legacy).unwrap(); + fs::create_dir(legacy.join("wave-000000")).unwrap(); + for path in ["workload.json", "wave-000000/response-0000.bin"] { + fs::copy(root.join(path), legacy.join(path)).unwrap(); + } + let mut plan = value(root.join("plan.json")); + plan["version"] = json!(1); + let bytes = serde_json::to_vec_pretty(&plan).unwrap(); + fs::write(legacy.join("plan.json"), &bytes).unwrap(); + let plan_hash = digest(&bytes); + let mut reservation = value(root.join("wave-000000/reservation.json")); + reservation["plan_sha256"] = json!(plan_hash); + let bytes = serde_json::to_vec_pretty(&reservation).unwrap(); + fs::write(legacy.join("wave-000000/reservation.json"), &bytes).unwrap(); + let mut receipt = wave(&temp, "off", 0); + receipt["plan_sha256"] = json!(plan_hash); + receipt["reservation_sha256"] = json!(digest(&bytes)); + let bytes = serde_json::to_vec_pretty(&receipt).unwrap(); + fs::write(legacy.join("wave-000000/wave.json"), &bytes).unwrap(); + assert!(report(&legacy).get("baseline_metrics").is_none()); + assert_eq!( + fs::read(legacy.join("wave-000000/wave.json")).unwrap(), + bytes + ); +} + +#[test] +fn metrics_scrapes_bracket_all_lanes_and_expose_label_matched_deltas_offline() { + let temp = Temp::new(); + let root = temp.path("on"); + let settled = Arc::new(AtomicUsize::new(0)); + let done = settled.clone(); + let dir = root.clone(); + let server = Server::new(move |mut stream, _, _| { + assert!(dir.join("wave-000000/metrics-before.bin").exists()); + header(&mut stream, "text/event-stream"); + frame( + &mut stream, + json!({"choices":[{"index":0,"delta":{"content":"ok"}}]}), + ); + done.fetch_add(1, Ordering::SeqCst); + finish(&mut stream, Some(8), Some(0)); + }); + let calls = server.count.clone(); + let dir = root.clone(); + let metrics = MetricsServer::new(move |index, _| { + assert!(dir.join("wave-000000/reservation.json").exists()); + assert!(!dir.join("wave-000000/wave.json").exists()); + if index == 0 { + assert_eq!(calls.load(Ordering::SeqCst), 0); + response(format!("{DRAFT}{{model=\"a\",engine=\"0\"}} 10\n{ACCEPTED}{{engine=\"0\",model=\"a\"}} 4\n{RUNNING}{{engine=\"0\"}} 2\n").as_bytes()) + } else { + assert_eq!(index, 1); + assert_eq!(settled.load(Ordering::SeqCst), 2); + response(format!("{DRAFT}{{engine=\"0\",model=\"a\"}} 18\n{ACCEPTED}{{model=\"a\",engine=\"0\"}} 10\n{RUNNING}{{engine=\"0\"}} 0\n").as_bytes()) + } + }); + successful( + &command(&temp, &server, &metrics.endpoint, "on", workload(2, 0, 1)) + .output() + .unwrap(), + ); + assert_eq!(metrics.count.load(Ordering::SeqCst), 2); + let output = report(&root); + let telemetry = &output["baseline_metrics"]["waves"][0]; + assert_eq!(telemetry["acceptance"][0]["ratio"], 0.75); + assert_eq!(telemetry["counters"].as_array().unwrap().len(), 2); + assert!( + telemetry["counters"] + .as_array() + .unwrap() + .iter() + .all(|v| v["name"] != RUNNING) + ); + assert_eq!(telemetry["before_series"][0]["value"], 2.0); + assert_eq!(telemetry["after_series"][0]["value"], 0.0); + assert!( + telemetry["receipt"]["measured_duration_us"] + .as_u64() + .unwrap() + >= wave(&temp, "on", 0)["elapsed_us"].as_u64().unwrap() + ); + assert_eq!(output["changes"][0]["eligible"], true); + assert_eq!( + output["baseline_metrics"]["config"]["endpoint"], + metrics.endpoint + ); + assert_eq!(metrics.count.load(Ordering::SeqCst), 2); +} + +#[test] +fn metrics_missing_reset_zero_and_unmatched_scopes_never_create_ratios() { + let temp = Temp::new(); + let server = Server::new(fast); + let metrics = MetricsServer::new(|index, _| { + let raw = if index == 0 { + format!( + "{DRAFT}{{model=\"reset\"}} 10\n{ACCEPTED}{{model=\"reset\"}} 5\n{DRAFT}{{model=\"zero\"}} 0\n{ACCEPTED}{{model=\"zero\"}} 0\n{DRAFT}{{model=\"only-draft\"}} 1\n{ACCEPTED}{{model=\"only-accepted\"}} 1\n{DRAFT}{{model=\"gone\"}} 1\n" + ) + } else { + format!( + "{DRAFT}{{model=\"reset\"}} 2\n{ACCEPTED}{{model=\"reset\"}} 1\n{DRAFT}{{model=\"zero\"}} 0\n{ACCEPTED}{{model=\"zero\"}} 0\n{DRAFT}{{model=\"only-draft\"}} 2\n{ACCEPTED}{{model=\"only-accepted\"}} 2\n" + ) + }; + response(raw.as_bytes()) + }); + successful( + &command(&temp, &server, &metrics.endpoint, "on", workload(1, 0, 1)) + .output() + .unwrap(), + ); + let output = report(&temp.path("on")); + let telemetry = &output["baseline_metrics"]["waves"][0]; + assert!( + telemetry["acceptance"] + .as_array() + .unwrap() + .iter() + .all(|v| v["ratio"].is_null()) + ); + let counters = telemetry["counters"].as_array().unwrap(); + assert!( + counters + .iter() + .filter(|v| v["labels"]["model"] == "reset") + .all(|v| v["delta"].is_null() && v["status"] == "reset_observed") + ); + assert!(counters.iter().any(|v| v["labels"]["model"] == "gone" + && v["delta"].is_null() + && v["status"] == "missing")); +} + +#[test] +fn metrics_parser_bounds_and_invalid_samples_remain_diagnostic_failures() { + let temp = Temp::new(); + let server = Server::new(fast); + let labels = (0..17) + .map(|i| format!("l{i}=\"v\"")) + .collect::>() + .join(","); + let samples = vec![ + format!("{DRAFT}{{a=\"x\",a=\"y\"}} 1\n"), + format!("{DRAFT}{{a=\"x\",b=\"y\"}} 1\n{DRAFT}{{b=\"y\",a=\"x\"}} 2\n"), + format!("{DRAFT} NaN\n"), + format!("{DRAFT} +Inf\n"), + format!("{DRAFT} -1\n"), + format!("{DRAFT}{{{labels}}} 1\n"), + format!("{DRAFT}{{a=\"{}\"}} 1\n", "x".repeat(4096)), + format!("#{}\n", "x".repeat(64 * 1024)), + (0..257) + .map(|i| format!("{RUNNING}{{lane=\"{i}\"}} 0\n")) + .collect(), + format!("{DRAFT}{{a=\"bad\\q\"}} 1\n"), + format!("{DRAFT} 1e999\n"), + ]; + let count = samples.len(); + let metrics = MetricsServer::new(move |index, _| response(samples[index / 2].as_bytes())); + successful( + &command( + &temp, + &server, + &metrics.endpoint, + "invalid", + workload(1, 0, count as u32), + ) + .output() + .unwrap(), + ); + let output = report(&temp.path("invalid")); + assert_eq!(output["changes"][0]["eligible"], true); + for wave in output["baseline_metrics"]["waves"].as_array().unwrap() { + assert_eq!(wave["receipt"]["before"]["status"], "parse_error"); + assert_eq!(wave["receipt"]["after"]["status"], "parse_error"); + assert!(wave["before_series"].as_array().unwrap().is_empty()); + assert!(wave["receipt"]["before"]["error"].is_string()); + } + assert_eq!(metrics.count.load(Ordering::SeqCst), count * 2); +} + +#[test] +fn metrics_raw_budget_skips_without_calls_and_survives_pause_resume() { + let temp = Temp::new(); + let (ready, releases) = std::sync::mpsc::sync_channel(1); + let server = Server::new(move |stream, index, request| { + if index == 7 { + let (release, wait) = std::sync::mpsc::sync_channel(0); + ready.send(release).unwrap(); + wait.recv_timeout(Duration::from_secs(30)).unwrap(); + } + fast(stream, index, request); + }); + let metrics = MetricsServer::new(|index, _| { + assert!(index < 16, "retained raw budget was reset"); + response(&padded()) + }); + let mut work = workload(1, 0, 9); + work["limits"]["total_ms"] = json!(15000); + work["limits"]["idle_ms"] = json!(12000); + let child = command(&temp, &server, &metrics.endpoint, "budget", work) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let release = releases.recv_timeout(Duration::from_secs(60)).unwrap(); + successful( + &cli() + .arg("pause") + .arg(temp.path("budget")) + .output() + .unwrap(), + ); + release.send(()).unwrap(); + assert_eq!(child.wait_with_output().unwrap().status.code(), Some(2)); + assert_eq!(metrics.count.load(Ordering::SeqCst), 16); + let old = fs::read(temp.path("budget/wave-000007/metrics.json")).unwrap(); + let resumed = cli() + .arg("resume") + .arg(temp.path("budget")) + .arg("--json") + .env("GRILL_METRICS_TEST_KEY", "fixture-secret") + .output() + .unwrap(); + assert_eq!( + resumed.status.code(), + Some(2), + "{}", + String::from_utf8_lossy(&resumed.stderr) + ); + assert_eq!(metrics.count.load(Ordering::SeqCst), 16); + assert_eq!(server.count.load(Ordering::SeqCst), 9); + assert_eq!( + fs::read(temp.path("budget/wave-000007/metrics.json")).unwrap(), + old + ); + let receipt = value(temp.path("budget/wave-000008/metrics.json")); + for side in ["before", "after"] { + assert_eq!(receipt[side]["status"], "skipped_budget"); + assert_eq!(receipt[side]["raw_bytes"], 0); + assert_eq!(receipt[side]["charged_us"], 0); + } + let output = compare(&temp.path("budget")); + assert_eq!(output.status.code(), Some(2)); + let output: Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!( + output["baseline_metrics"]["budget"]["raw_bytes"], + 16 * 1024 * 1024 + ); +} + +#[test] +fn metrics_failures_do_not_retry_redirect_or_change_performance_eligibility() { + let temp = Temp::new(); + let server = Server::new(fast); + let sink = MetricsServer::new(|_, _| panic!("metrics redirect followed")); + let location = sink.endpoint.clone(); + let metrics = MetricsServer::new(move |index, _| { + match index { + 0 => format!("HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").into_bytes(), + 1 => b"HTTP/1.1 503 Unavailable\r\nContent-Length: 4\r\nConnection: close\r\n\r\noops".to_vec(), + 2 => response(&vec![b'x'; 1024 * 1024 + 1]), + 3 => Vec::new(), + _ => panic!("unexpected retry"), + } + }); + successful( + &command( + &temp, + &server, + &metrics.endpoint, + "errors", + workload(1, 0, 2), + ) + .output() + .unwrap(), + ); + let output = report(&temp.path("errors")); + let waves = &output["baseline_metrics"]["waves"]; + assert_eq!(waves[0]["receipt"]["before"]["status"], "http_error"); + assert_eq!(waves[0]["receipt"]["after"]["status"], "http_error"); + assert_eq!(waves[1]["receipt"]["before"]["status"], "body_limit"); + assert_eq!(waves[1]["receipt"]["after"]["status"], "transport_error"); + assert_eq!(output["changes"][0]["eligible"], true); + assert_eq!(metrics.count.load(Ordering::SeqCst), 4); + assert_eq!(sink.count.load(Ordering::SeqCst), 0); +} + +#[test] +fn metrics_tamper_missing_companion_and_rehashed_malformed_raw_fail_closed() { + let temp = Temp::new(); + let server = Server::new(fast); + let metrics = MetricsServer::new(|_, _| response(format!("{DRAFT} 1\n").as_bytes())); + successful( + &command(&temp, &server, &metrics.endpoint, "on", workload(1, 0, 1)) + .output() + .unwrap(), + ); + let root = temp.path("on"); + let raw = root.join("wave-000000/metrics-before.bin"); + let original = fs::read(&raw).unwrap(); + fs::write(&raw, b"tampered").unwrap(); + assert_eq!(compare(&root).status.code(), Some(1)); + fs::write(&raw, &original).unwrap(); + let path = root.join("wave-000000/metrics.json"); + let original = fs::read(&path).unwrap(); + fs::remove_file(&path).unwrap(); + assert_eq!(compare(&root).status.code(), Some(1)); + fs::write(&path, original).unwrap(); + let invalid = format!("{DRAFT} NaN\n").into_bytes(); + fs::write(&raw, &invalid).unwrap(); + rewrite_receipt(&root, |receipt| { + receipt["before"]["raw_bytes"] = json!(invalid.len()); + receipt["before"]["raw_sha256"] = json!(digest(&invalid)); + }); + assert_eq!(compare(&root).status.code(), Some(1)); + assert_eq!(metrics.count.load(Ordering::SeqCst), 2); +} + +#[test] +fn metrics_partial_publication_is_unsettled_and_never_resumed() { + let temp = Temp::new(); + let (ready, seen) = std::sync::mpsc::sync_channel(1); + let server = Server::new(move |stream, index, request| { + let (release, wait) = std::sync::mpsc::sync_channel(0); + ready.send(release).unwrap(); + wait.recv_timeout(Duration::from_secs(30)).unwrap(); + fast(stream, index, request); + }); + let metrics = MetricsServer::new(|_, _| response(b"")); + let child = command( + &temp, + &server, + &metrics.endpoint, + "partial", + workload(1, 0, 1), + ) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let release = seen.recv_timeout(Duration::from_secs(30)).unwrap(); + fs::create_dir(temp.path("partial/wave-000000/metrics.json")).unwrap(); + release.send(()).unwrap(); + assert_eq!(child.wait_with_output().unwrap().status.code(), Some(1)); + assert!(!temp.path("partial/wave-000000/wave.json").exists()); + let output = compare(&temp.path("partial")); + assert_eq!(output.status.code(), Some(2)); + let output: Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!( + output["baseline"][0]["trial_states"][0], + "reserved_unsettled" + ); + let resumed = cli() + .arg("resume") + .arg(temp.path("partial")) + .output() + .unwrap(); + assert_eq!(resumed.status.code(), Some(1)); + assert_eq!(metrics.count.load(Ordering::SeqCst), 2); +} + +#[test] +fn metrics_endpoint_policy_is_admitted_before_any_request() { + let temp = Temp::new(); + let server = Server::new(fast); + let metrics = MetricsServer::new(|_, _| panic!("invalid endpoint dispatched")); + for (index, endpoint) in [ + "http://localhost/metrics", + "http://192.0.2.1/metrics", + "https://user:secret@example.com/metrics", + "https://example.com/metrics?secret=x", + "https://example.com/metrics#fragment", + ] + .into_iter() + .enumerate() + { + let mut cmd = command( + &temp, + &server, + endpoint, + &format!("bad-{index}"), + workload(1, 0, 1), + ); + assert_eq!(cmd.output().unwrap().status.code(), Some(1)); + } + assert_eq!(server.count.load(Ordering::SeqCst), 0); + assert_eq!(metrics.count.load(Ordering::SeqCst), 0); +} + +#[test] +fn metrics_whole_run_deadline_exhausts_and_later_snapshots_skip() { + let temp = Temp::new(); + let server = Server::new(fast); + let (ready, releases) = std::sync::mpsc::sync_channel(32); + let metrics = MetricsServer::new(move |index, _| { + assert!(index < 16, "whole-run deadline budget renewed"); + let (release, wait) = std::sync::mpsc::sync_channel(0); + ready.send(release).unwrap(); + wait.recv_timeout(Duration::from_secs(60)).unwrap(); + response(b"") + }); + let output = command( + &temp, + &server, + &metrics.endpoint, + "deadline", + workload(1, 0, 16), + ) + .output() + .unwrap(); + for release in releases.try_iter() { + release.send(()).unwrap(); + } + successful(&output); + let output = report(&temp.path("deadline")); + let waves = output["baseline_metrics"]["waves"].as_array().unwrap(); + assert_eq!(waves[0]["receipt"]["before"]["status"], "deadline"); + assert_eq!( + waves.last().unwrap()["receipt"]["after"]["status"], + "skipped_budget" + ); + assert!(metrics.count.load(Ordering::SeqCst) <= 15); + assert_eq!(server.count.load(Ordering::SeqCst), 16); + assert_eq!(output["changes"][0]["eligible"], true); +} + +#[test] +fn metrics_declarations_cannot_silently_pair_different_measurement_conditions() { + let temp = Temp::new(); + let server = Server::new(fast); + let first = MetricsServer::new(|_, _| response(b"")); + let second = MetricsServer::new(|_, _| response(b"")); + successful(&run(&temp, &server, "off", &workload(1, 0, 1))); + successful( + &command(&temp, &server, &first.endpoint, "first", workload(1, 0, 1)) + .output() + .unwrap(), + ); + successful( + &command( + &temp, + &server, + &second.endpoint, + "second", + workload(1, 0, 1), + ) + .output() + .unwrap(), + ); + for other in ["off", "second"] { + let output = cli() + .arg("compare") + .arg(temp.path("first")) + .arg(temp.path(other)) + .arg("--json") + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(1)); + assert!(output.stdout.is_empty()); + } + assert_eq!(first.count.load(Ordering::SeqCst), 2); + assert_eq!(second.count.load(Ordering::SeqCst), 2); +} + +#[test] +fn metrics_interrupt_retains_settled_wave_and_refuses_continuation() { + let temp = Temp::new(); + let (ready, releases) = std::sync::mpsc::sync_channel(1); + let server = Server::new(move |mut stream, _, _| { + header(&mut stream, "text/event-stream"); + frame( + &mut stream, + json!({"choices":[{"index":0,"delta":{"content":"partial"}}]}), + ); + let (release, wait) = std::sync::mpsc::sync_channel(0); + ready.send(release).unwrap(); + wait.recv_timeout(Duration::from_secs(30)).unwrap(); + }); + let metrics = MetricsServer::new(|_, _| response(b"")); + let child = command( + &temp, + &server, + &metrics.endpoint, + "interrupted", + workload(1, 0, 2), + ) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let release = releases.recv_timeout(Duration::from_secs(30)).unwrap(); + assert_eq!( + unsafe { libc::kill(child.id() as libc::pid_t, libc::SIGINT) }, + 0 + ); + let output = child.wait_with_output().unwrap(); + release.send(()).unwrap(); + assert_eq!(output.status.code(), Some(2)); + assert_eq!( + wave(&temp, "interrupted", 0)["attempts"][0]["status"], + "interrupted" + ); + assert!(temp.path("interrupted/wave-000000/metrics.json").exists()); + assert!(!temp.path("interrupted/wave-000001").exists()); + assert_eq!(compare(&temp.path("interrupted")).status.code(), Some(2)); + assert_eq!( + cli() + .arg("resume") + .arg(temp.path("interrupted")) + .output() + .unwrap() + .status + .code(), + Some(1) + ); + assert_eq!(metrics.count.load(Ordering::SeqCst), 2); +} + +#[test] +fn metrics_parser_accepts_exact_bounds_and_decodes_full_label_identity() { + let temp = Temp::new(); + let server = Server::new(fast); + let mut labels = (1..16).map(|i| format!("l{i}=\"v\"")).collect::>(); + let escaped = "line\\nquote\\\"slash\\\\"; + let empty = format!("{{l0=\"{escaped}\",{}}}", labels.join(",")); + let padding = "x".repeat(4096 - empty.len()); + labels.insert(0, format!("l0=\"{escaped}{padding}\"")); + let label_set = format!("{{{}}}", labels.join(",")); + assert_eq!(label_set.len(), 4096); + let metrics = MetricsServer::new(move |index, _| { + let mut raw = format!( + "#{}\n{DRAFT}{label_set} {}\n", + "x".repeat(64 * 1024 - 1), + index + 1 + ); + for series in 0..255 { + raw.push_str(&format!("{RUNNING}{{series=\"{series}\"}} 0\n")); + } + response(raw.as_bytes()) + }); + successful( + &command( + &temp, + &server, + &metrics.endpoint, + "bounds", + workload(1, 0, 1), + ) + .output() + .unwrap(), + ); + let output = report(&temp.path("bounds")); + let counter = &output["baseline_metrics"]["waves"][0]["counters"][0]; + assert_eq!(counter["delta"], 1.0); + assert_eq!(counter["labels"].as_object().unwrap().len(), 16); + assert!( + counter["labels"]["l0"] + .as_str() + .unwrap() + .starts_with("line\nquote\"slash\\") + ); + assert_eq!( + output["baseline_metrics"]["waves"][0]["before_series"] + .as_array() + .unwrap() + .len(), + 256 + ); +} + +#[test] +fn metrics_unselected_series_do_not_discard_selected_observations() { + let temp = Temp::new(); + let server = Server::new(fast); + let metrics = MetricsServer::new(|index, _| { + let mut body = String::new(); + for series in 0..1000 { + body.push_str(&format!("unselected{{series=\"{series}\"}} NaN\n")); + } + body.push_str(&format!( + "{DRAFT} {}\n{ACCEPTED} {}\n", + index * 4, + index * 2 + )); + response(body.as_bytes()) + }); + successful( + &command( + &temp, + &server, + &metrics.endpoint, + "selected", + workload(1, 0, 1), + ) + .output() + .unwrap(), + ); + let output = report(&temp.path("selected")); + let wave = &output["baseline_metrics"]["waves"][0]; + assert_eq!(wave["receipt"]["before"]["status"], "complete"); + assert_eq!(wave["before_series"].as_array().unwrap().len(), 2); + assert_eq!(wave["acceptance"][0]["ratio"], 0.5); +} + +#[test] +fn metrics_rehashed_impossible_failure_metadata_is_rejected() { + let temp = Temp::new(); + let server = Server::new(fast); + let metrics = MetricsServer::new(|_, _| response(format!("{DRAFT} 1\n").as_bytes())); + successful( + &command( + &temp, + &server, + &metrics.endpoint, + "metadata", + workload(1, 0, 1), + ) + .output() + .unwrap(), + ); + let root = temp.path("metadata"); + let original = value(root.join("wave-000000/metrics.json")); + for (status, http, scrape) in [ + ( + "unsupported", + Value::Null, + original["before"]["scrape_us"].clone(), + ), + ( + "transport_error", + Value::Null, + original["before"]["scrape_us"].clone(), + ), + ("transport_error", json!(200), json!(0)), + ] { + rewrite_receipt(&root, |receipt| { + *receipt = original.clone(); + receipt["before"]["status"] = json!(status); + receipt["before"]["error"] = json!("fixture failure"); + receipt["before"]["http_status"] = http; + receipt["before"]["scrape_us"] = scrape; + }); + assert_eq!(compare(&root).status.code(), Some(1)); + } + rewrite_receipt(&root, |receipt| *receipt = original); + successful(&compare(&root)); +} + +#[test] +fn metrics_nonstandard_http_failure_does_not_abort_model_collection() { + let temp = Temp::new(); + let server = Server::new(fast); + let metrics = MetricsServer::new(|_, _| { + b"HTTP/1.1 600 Failure\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_vec() + }); + successful( + &command( + &temp, + &server, + &metrics.endpoint, + "http600", + workload(1, 0, 1), + ) + .output() + .unwrap(), + ); + let output = report(&temp.path("http600")); + assert_eq!(output["changes"][0]["eligible"], true); + for side in ["before", "after"] { + assert_eq!( + output["baseline_metrics"]["waves"][0]["receipt"][side]["status"], + "http_error" + ); + assert_eq!( + output["baseline_metrics"]["waves"][0]["receipt"][side]["http_status"], + 600 + ); + } + assert_eq!(server.count.load(Ordering::SeqCst), 1); + assert_eq!(metrics.count.load(Ordering::SeqCst), 2); +} diff --git a/docs/performance/CONTRACT.md b/docs/performance/CONTRACT.md index d952b2b..c8133db 100644 --- a/docs/performance/CONTRACT.md +++ b/docs/performance/CONTRACT.md @@ -195,6 +195,85 @@ collector binary fingerprint and optional deployment declarations. These do not prove which model or hardware an untrusted server used. Raw evidence is not a public-safe export. +## Provider snapshot protocol + +Opt-in plan field `metrics` is a closed version-1 configuration containing +`endpoint`, `allowlist`, `scope`, `cadence`, `deadline_us`, `body_bytes`, +`line_bytes`, `series`, `labels_per_series`, `label_bytes_per_series`, +`run_budget_us`, `retained_raw_bytes` and `max_requests`. The collector freezes +the endpoint and bounds described in the usage guide; the loader rejects +modified policies. The plan-byte hash binds them to every reservation and wave. +Absent `metrics` serializes absent, preserving opted-out plan/request/receipt +shapes. Legacy plans cannot acquire telemetry without execution-session +provenance. Comparison requires equal optional configurations, including endpoint. + +Each opted-in published wave requires `metrics: {file, sha256, overhead_us}`; +`file` is exactly `metrics.json` within that wave directory. The bounded companion +contains `{version, plan_sha256, wave, before, after, measured_origin_unix_ms, +measured_duration_us}`. The measurement duration spans the collector origin +through all lane settlement, not the subsequent scrape or publication. +`overhead_us` covers both telemetry operations including raw and companion +publication; it is excluded from wave preparation/body-publication intervals. +Wall timestamps are observations, not an assumed monotonic clock. + +Each snapshot contains `{started_unix_ms, scrape_us, duration_us, charged_us, +allowance_us, status, error, http_status, raw_bytes, raw_sha256}`. +`scrape_us` covers capture/parse; `duration_us` also includes raw hashing and +publication. Its raw entity or retained +prefix is `metrics-before.bin` or `metrics-after.bin`; skipped snapshots bind +an empty file. Status is `complete`, `parse_error`, `transport_error`, +`http_error`, `unsupported`, `deadline`, `body_limit` or `skipped_budget`. +Only successful, completely captured, identity-encoded HTTP 200 entities supply +samples. Errors remain diagnostics, not zeros or performance-ineligibility +reasons. The loader verifies companion lineage/hash, raw bounds/hashes, policy, +budget arithmetic and complete/parse-error raw semantics. Derived series, +deltas and ratios are recomputed offline, not trusted from serialized values. + +The parser accepts UTF-8 Prometheus text samples, comments, optional integer +timestamps and quoted labels with newline, quote and backslash escapes. It +rejects duplicate selected identities, duplicate selected labels, nonfinite selected +values and negative supported counters. Unselected names are skipped before label/value +parsing; whole-body UTF-8, byte and line bounds still apply. Label bytes include +the encoded braces, names, separators and values of each selected series. +The allowlist is fixed; no histogram expansion or OpenMetrics exemplars are +interpreted. Empty/missing samples remain unavailable, not zero. + +Counter identity is metric name plus canonical complete labels. A delta requires +both finite samples and a nondecreasing value. A decrease reports +`reset_observed`; a missing side reports `missing`, with null delta. +Nondecreasing deltas explicitly leave restart identity unverified. Acceptance +ratios require matching label sets, usable draft and accepted deltas, positive +draft delta and accepted delta no greater than draft delta. Gauges are snapshots, +not deltas. All diagnostics remain server-scoped, not workload-attributed. + +Each scrape deadline is the lesser of 2 seconds and remaining 30-second +whole-run telemetry allowance. Its persisted elapsed duration, including raw +publication and at least a microsecond, is the charge. Remaining telemetry +overhead from the wave reference is charged before the next wave. Scheduler +and filesystem overrun is retained and prevents further +admission once exhausted. No completed scrape gets a fresh full-deadline charge +on resume. Every published companion is reloaded in schedule order to reconstruct +requests, elapsed charges and retained bytes before continuation. Existing +lifecycle rules refuse continuation of any reserved/unsettled scrape or wave; +uncommitted reservations therefore cannot reset the allowance. + +Raw bodies are bounded by 1 MiB each and 16 MiB across the run. Admission reserves +a full body cap before making another request. Companions are bounded by 16 KiB +each and labels are retained only in raw bodies, not JSON receipts. Thus escaped +labels cannot inflate companion publication. Offline JSON can expand raw labels +through escaping and diagnostic projections; its input remains bounded by the +whole-run raw cap and per-scrape series/label limits, not a claim of constant +output size or RSS. Filesystem latency, kernel/TLS buffering and scheduler +starvation are not bounded by the network deadline. Published telemetry overhead +exposes local work rather than claiming zero perturbation. + +Raw before evidence is written outside the measured origin, after reservation. +The after scrape and companion publication follow all lane settlement and precede +wave publication. A failed companion publication leaves the reservation +unsettled; it is inspectable as missing performance evidence, never silently +recovered or resumed. A published wave missing its required companion fails +integrity verification. Telemetry has no separate inspector or recovery workflow. + ## Lifecycle ownership and continuation Each collector/resumer holds a nonblocking exclusive `flock` on the nonsymlink diff --git a/docs/performance/README.md b/docs/performance/README.md index d6cfbfb..a29805d 100644 --- a/docs/performance/README.md +++ b/docs/performance/README.md @@ -264,6 +264,62 @@ reported count. A flag plus a reported zero is **not universal proof of cold cache state**. Model weights, operating-system/JIT/GPU caches and engine state are not inspected or flushed. Missing usage remains unknown, never zero. +## Optional provider snapshots + +Add `--metrics-url https://your-server.example/metrics` to `run` to retain bounded +vLLM Prometheus diagnostics. The endpoint follows the model transport's URL/TLS +policy, but uses a separate client and **never receives model authorization**. +There is no discovery, redirect, retry, idle gate or additional model request. +An endpoint requiring authentication will report a scrape failure; this option +does not add metrics credentials. + +The fixed allowlist contains `vllm:spec_decode_num_draft_tokens_total`, +`vllm:spec_decode_num_accepted_tokens_total`, `vllm:num_requests_running` and +`vllm:num_requests_waiting`. Counters are matched by name and the complete +canonical label set, never summed across engines/models. Before/after samples, +counter deltas, null reset/missing results and compatible acceptance ratios +appear in `compare RUN RUN --json` under `baseline_metrics`; gauges remain +snapshots. A nondecreasing counter does not prove that no restart occurred. +These are **server-wide diagnostics, not this workload's attributed tokens**. +Other clients can contribute to the same series. + +The before scrape follows durable wave reservation and finishes before the +measured origin. The after scrape starts only after all lanes settle, including +failed/interrupted lanes. Scrape timestamps, durations and total telemetry +overhead are separate from measured wave latency. **Outside the timer is not +measurement-neutral**: scraping and evidence publication can change server load, +cache warmth and between-wave cadence. Comparison requires identical telemetry +configuration, including endpoint identity; opted-in and opted-out runs do not +silently qualify as matching conditions. +Signals still cancel model lanes, but telemetry is deadline-bounded rather than +signal-cancelled: an active snapshot and the required after snapshot can delay +exit. Cooperative pause continues to drain and publish the whole admitted wave. + +The frozen protocol allows a deadline of at most 2 seconds per scrape, 1 MiB +body, 64 KiB line, 256 selected sample series, 16 labels and 4096 encoded label-set +bytes per selected series. Comments and unselected samples consume body/line +bounds but are skipped before label/value parsing and selected-series accounting. The whole-run +telemetry allowance is 30 seconds and retained raw data is at most 16 MiB, +with at most twice the planned wave count in requests. Each new scrape gets +the smaller of its deadline and remaining time. Elapsed capture/parse and raw +publication time are charged, not a full deadline for fast scrapes. Companion +publication overhead is charged before the next wave. Raw admission reserves a full +body allowance. Exhaustion retains `skipped_budget`, without another call. +Filesystem synchronization and OS scheduling cannot be given a hard wall-clock +guarantee; actual overhead is retained, and scheduling overrun consumes the +remaining scrape allowance rather than renewing it. + +Resume reconstructs consumption from verified retained snapshots, not a new +budget. Unsettled, interrupted or partially published work is not resumable. +Scrape failures preserve bounded raw bytes and errors without changing otherwise +valid performance eligibility. Missing promised evidence, changed hashes and +local publication failures remain integrity/I/O errors. + +Metrics endpoints, labels and raw bodies can expose private model names, request +identifiers or other server content, including unsupported metrics and comments. +Review all retained bytes before sharing; the allowlist is not a privacy filter. +See the [snapshot schema and bounds](CONTRACT.md#provider-snapshot-protocol). + ## Deployment declarations and privacy `--deployment FILE` optionally records a closed JSON object with nullable