From 3a078ca7232d27baec78faa641683d6c6bad1fc5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 8 May 2026 04:39:19 +0000 Subject: [PATCH] soft-agent: latency histograms on /metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two Prometheus-format histograms to the existing /metrics endpoint: - soft_step_duration_seconds{outcome=Idle|Processed|Error} measures one full step's wall-clock from mailbox poll to return. Idle covers the cheap mailbox-empty path (~16us observed locally); Processed covers handler dispatch + gate + execute; Error captures dispatch failures. - soft_action_execute_duration_seconds{kind=...} measures only executor.execute() wall-clock — captures the real I/O cost of A2A sends, MCP calls, LLM calls — without contaminating it with gate evaluation or trace bookkeeping. Standard Prometheus latency buckets (1 ms → 10 s exponential): 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0. Each series carries cumulative bucket_counts + sum_micros + count, all `AtomicU64` for cross-thread reads from the soft-a2a server's /metrics handler. soft-agent::metrics: new private `LabeledHistogram` + public `Metrics::observe_step_duration(outcome, duration)` and `Metrics::observe_action_execute_duration(kind, duration)`. Render path appends histogram series after the existing counter series; Prometheus content-type and ordering preserved. soft-agent::runner: `step()` records `Instant::now()` at entry, calls `observe_step_duration(outcome, elapsed)` on each return path (Idle, Processed, Error). `executor.execute(action)` is now wrapped with its own `Instant` so the action-kind histogram captures only execution time. 5 new metrics unit tests + 1 new soft-a2a route test cover: - bucket boundaries (7 ms lands in 0.01+, not 0.005) - cumulative bucket semantics across multiple observations - per-label independence (Idle vs Processed) - empty histograms still emit type lines - HTTP route serves histogram format with right content-type cargo fmt clean, cargo clippy --workspace --all-targets -- -D warnings clean, full workspace test suite: 93 passed / 0 failed. Live verification on a hand-driven depot: 3 RequestSession messages across ~600 ms produced 31 Idle observations totaling 0.495 ms (~16 us/poll) and 3 Processed observations all in the ≤1 ms bucket. --- crates/soft-a2a/tests/metrics_route.rs | 20 +++ crates/soft-agent/src/metrics.rs | 220 ++++++++++++++++++++++++- crates/soft-agent/src/runner.rs | 12 +- 3 files changed, 245 insertions(+), 7 deletions(-) diff --git a/crates/soft-a2a/tests/metrics_route.rs b/crates/soft-a2a/tests/metrics_route.rs index fe0cef5..297c695 100644 --- a/crates/soft-a2a/tests/metrics_route.rs +++ b/crates/soft-a2a/tests/metrics_route.rs @@ -75,6 +75,26 @@ fn metrics_returns_prometheus_text_when_wired() { assert!(body.contains("soft_actions_allowed_total{kind=\"send_a2a\"} 1")); } +#[test] +fn metrics_route_renders_histogram_format() { + use std::time::Duration; + + let m = Arc::new(Metrics::new("hist-route")); + m.observe_step_duration("Processed", Duration::from_millis(7)); + m.observe_action_execute_duration("send_a2a", Duration::from_millis(3)); + + let addr = boot(Some(Arc::clone(&m))); + let (status, body, _) = get(&addr, "/metrics"); + assert_eq!(status, 200); + + // Histogram preamble + a representative bucket + sum + count. + assert!(body.contains("# TYPE soft_step_duration_seconds histogram")); + assert!(body.contains("soft_step_duration_seconds_bucket{outcome=\"Processed\",le=\"0.01\"} 1")); + assert!(body.contains("soft_step_duration_seconds_count{outcome=\"Processed\"} 1")); + assert!(body + .contains("soft_action_execute_duration_seconds_bucket{kind=\"send_a2a\",le=\"0.005\"} 1")); +} + #[test] fn metrics_reflects_concurrent_increments() { let m = Arc::new(Metrics::new("concur")); diff --git a/crates/soft-agent/src/metrics.rs b/crates/soft-agent/src/metrics.rs index de4e601..1ac292c 100644 --- a/crates/soft-agent/src/metrics.rs +++ b/crates/soft-agent/src/metrics.rs @@ -1,17 +1,20 @@ -//! Lightweight Prometheus-format counters for the runner. +//! Lightweight Prometheus-format counters and histograms for the +//! runner. //! -//! All counters are atomic, `Arc`-shared, and safe to read from a +//! All metrics are atomic, `Arc`-shared, and safe to read from a //! different thread than the one running the runner — soft-a2a's HTTP //! server reads them to render `/metrics` while the runner increments //! them from its step loop. //! -//! Only counters here, no histograms or gauges. Keeping the surface -//! minimal: every metric is `monotonic_total{label="..."}`. Render via +//! Two metric kinds: counters (`monotonic_total{label="..."}`) and +//! histograms (cumulative bucket counts + sum + count, in +//! `*_seconds`-suffixed Prometheus convention). Render via //! [`Metrics::render_prometheus`]. use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Mutex; +use std::time::Duration; /// Per-label counter map. Lock contention is negligible: the runner /// holds the lock for at most one `entry().or_insert(0).fetch_add(1)` @@ -107,8 +110,119 @@ impl LabeledCounter2 { } } -/// Set of counters the runner increments. Construct one per agent and -/// share via `Arc::clone` between the runner and the HTTP server. +/// Standard Prometheus latency buckets (seconds). Covers 1 ms through +/// 10 s exponentially — appropriate for agent step latencies (mostly +/// sub-100 ms) and outliers (LLM/MCP calls in the hundreds-of-ms to +/// multi-second range). +const DEFAULT_LATENCY_BUCKETS_SECONDS: &[f64] = &[ + 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, +]; + +/// Per-label histogram. Each labeled series carries: one `AtomicU64` +/// per bucket (cumulative count of observations ≤ that boundary), an +/// `AtomicU64` running sum (in microseconds — fits 5800 years), and +/// an `AtomicU64` total observation count. +struct HistogramSeries { + bucket_counts: Vec, + sum_micros: AtomicU64, + count: AtomicU64, +} + +impl HistogramSeries { + fn new(num_buckets: usize) -> Self { + let mut bucket_counts = Vec::with_capacity(num_buckets + 1); + for _ in 0..=num_buckets { + bucket_counts.push(AtomicU64::new(0)); + } + Self { + bucket_counts, + sum_micros: AtomicU64::new(0), + count: AtomicU64::new(0), + } + } + + fn observe(&self, seconds: f64, buckets: &[f64]) { + // Increment every bucket whose upper bound is ≥ the observed + // value. Prometheus histograms are cumulative — a single + // observation is counted in its own bucket *and* every wider + // bucket above it. + for (i, &boundary) in buckets.iter().enumerate() { + if seconds <= boundary { + self.bucket_counts[i].fetch_add(1, Ordering::Relaxed); + } + } + // +Inf bucket (last entry) always increments. + self.bucket_counts[buckets.len()].fetch_add(1, Ordering::Relaxed); + + let micros = (seconds * 1_000_000.0).round() as u64; + self.sum_micros.fetch_add(micros, Ordering::Relaxed); + self.count.fetch_add(1, Ordering::Relaxed); + } +} + +struct LabeledHistogram { + name: &'static str, + label_key: &'static str, + buckets: &'static [f64], + inner: Mutex>, +} + +impl LabeledHistogram { + fn new(name: &'static str, label_key: &'static str, buckets: &'static [f64]) -> Self { + Self { + name, + label_key, + buckets, + inner: Mutex::new(HashMap::new()), + } + } + + fn observe(&self, label: &str, duration: Duration) { + let seconds = duration.as_secs_f64(); + let mut map = self.inner.lock().expect("metrics map poisoned"); + let series = map + .entry(label.to_string()) + .or_insert_with(|| HistogramSeries::new(self.buckets.len())); + series.observe(seconds, self.buckets); + } + + fn render(&self, out: &mut String) { + let map = self.inner.lock().expect("metrics map poisoned"); + out.push_str(&format!("# TYPE {} histogram\n", self.name)); + let mut keys: Vec<&String> = map.keys().collect(); + keys.sort(); + for label in keys { + let series = &map[label]; + let escaped = escape_label(label); + for (i, &boundary) in self.buckets.iter().enumerate() { + let count = series.bucket_counts[i].load(Ordering::Relaxed); + out.push_str(&format!( + "{}_bucket{{{}=\"{}\",le=\"{}\"}} {}\n", + self.name, self.label_key, escaped, boundary, count, + )); + } + let inf_count = series.bucket_counts[self.buckets.len()].load(Ordering::Relaxed); + out.push_str(&format!( + "{}_bucket{{{}=\"{}\",le=\"+Inf\"}} {}\n", + self.name, self.label_key, escaped, inf_count, + )); + let sum_seconds = series.sum_micros.load(Ordering::Relaxed) as f64 / 1_000_000.0; + out.push_str(&format!( + "{}_sum{{{}=\"{}\"}} {}\n", + self.name, self.label_key, escaped, sum_seconds, + )); + let total = series.count.load(Ordering::Relaxed); + out.push_str(&format!( + "{}_count{{{}=\"{}\"}} {}\n", + self.name, self.label_key, escaped, total, + )); + } + } +} + +/// Set of counters and histograms the runner records. Construct one +/// per agent and share via `Arc::clone` between the runner and the +/// HTTP server. pub struct Metrics { agent: String, messages_received: LabeledCounter, @@ -118,6 +232,8 @@ pub struct Metrics { gate_verdict: LabeledCounter, tick_fires: LabeledCounter, steps: LabeledCounter, + step_duration: LabeledHistogram, + action_execute_duration: LabeledHistogram, } impl Metrics { @@ -131,6 +247,16 @@ impl Metrics { gate_verdict: LabeledCounter::new("soft_gate_verdict_total", "verdict"), tick_fires: LabeledCounter::new("soft_tick_fires_total", "topic"), steps: LabeledCounter::new("soft_steps_total", "outcome"), + step_duration: LabeledHistogram::new( + "soft_step_duration_seconds", + "outcome", + DEFAULT_LATENCY_BUCKETS_SECONDS, + ), + action_execute_duration: LabeledHistogram::new( + "soft_action_execute_duration_seconds", + "kind", + DEFAULT_LATENCY_BUCKETS_SECONDS, + ), } } @@ -162,6 +288,19 @@ impl Metrics { self.steps.inc(outcome); } + /// Record one full step's wall-clock duration, labeled by + /// outcome (`Idle`, `Processed`, `Error`). + pub fn observe_step_duration(&self, outcome: &str, duration: Duration) { + self.step_duration.observe(outcome, duration); + } + + /// Record one action-execution wall-clock duration, labeled by + /// action kind. Captures only execution itself — not gate + /// evaluation or trace bookkeeping. + pub fn observe_action_execute_duration(&self, kind: &str, duration: Duration) { + self.action_execute_duration.observe(kind, duration); + } + /// Render all metrics as Prometheus text exposition format. /// The agent name is emitted as a constant `# HELP` comment so /// scrapers can identify the source even without re-labeling. @@ -178,6 +317,8 @@ impl Metrics { self.gate_verdict.render(&mut out); self.tick_fires.render(&mut out); self.steps.render(&mut out); + self.step_duration.render(&mut out); + self.action_execute_duration.render(&mut out); out } @@ -233,4 +374,71 @@ mod tests { let txt = m.render_prometheus(); assert!(txt.contains("topic=\"weird\\\"topic\"")); } + + #[test] + fn histogram_observation_lands_in_correct_buckets() { + let m = Metrics::new("hist-test"); + // 7 ms — should land in 0.01, 0.025, 0.05, ... +Inf buckets. + m.observe_step_duration("Processed", Duration::from_millis(7)); + let txt = m.render_prometheus(); + + // 0.005 bucket excludes 7 ms; 0.01 and above include it. + assert!( + txt.contains("soft_step_duration_seconds_bucket{outcome=\"Processed\",le=\"0.005\"} 0") + ); + assert!( + txt.contains("soft_step_duration_seconds_bucket{outcome=\"Processed\",le=\"0.01\"} 1") + ); + assert!( + txt.contains("soft_step_duration_seconds_bucket{outcome=\"Processed\",le=\"+Inf\"} 1") + ); + assert!(txt.contains("soft_step_duration_seconds_count{outcome=\"Processed\"} 1")); + // Sum lands at 0.007 seconds. + assert!( + txt.contains("soft_step_duration_seconds_sum{outcome=\"Processed\"} 0.007"), + "missing sum line in:\n{txt}" + ); + } + + #[test] + fn histogram_handles_multiple_observations_under_same_label() { + let m = Metrics::new("hist-multi"); + m.observe_action_execute_duration("send_a2a", Duration::from_millis(2)); + m.observe_action_execute_duration("send_a2a", Duration::from_millis(20)); + m.observe_action_execute_duration("send_a2a", Duration::from_millis(2000)); + + let txt = m.render_prometheus(); + // 2 ms ≤ 0.005; 20 ms exceeds 0.005 but ≤ 0.025; 2 s exceeds + // every bucket up to 2.5 s. + assert!(txt.contains( + "soft_action_execute_duration_seconds_bucket{kind=\"send_a2a\",le=\"0.005\"} 1" + )); + assert!(txt.contains( + "soft_action_execute_duration_seconds_bucket{kind=\"send_a2a\",le=\"0.025\"} 2" + )); + assert!(txt.contains( + "soft_action_execute_duration_seconds_bucket{kind=\"send_a2a\",le=\"2.5\"} 3" + )); + assert!(txt.contains("soft_action_execute_duration_seconds_count{kind=\"send_a2a\"} 3")); + } + + #[test] + fn histogram_renders_type_line_when_empty() { + let m = Metrics::new("hist-empty"); + let txt = m.render_prometheus(); + assert!(txt.contains("# TYPE soft_step_duration_seconds histogram")); + assert!(txt.contains("# TYPE soft_action_execute_duration_seconds histogram")); + // No bucket lines yet. + assert!(!txt.contains("soft_step_duration_seconds_bucket")); + } + + #[test] + fn histogram_observations_are_independent_per_label() { + let m = Metrics::new("hist-labels"); + m.observe_step_duration("Idle", Duration::from_millis(1)); + m.observe_step_duration("Processed", Duration::from_millis(50)); + let txt = m.render_prometheus(); + assert!(txt.contains("soft_step_duration_seconds_count{outcome=\"Idle\"} 1")); + assert!(txt.contains("soft_step_duration_seconds_count{outcome=\"Processed\"} 1")); + } } diff --git a/crates/soft-agent/src/runner.rs b/crates/soft-agent/src/runner.rs index 7fffef9..620b3e8 100644 --- a/crates/soft-agent/src/runner.rs +++ b/crates/soft-agent/src/runner.rs @@ -17,7 +17,7 @@ use std::collections::HashMap; use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use indexmap::IndexMap; use lex_bytecode::Value as LexValue; @@ -143,10 +143,13 @@ impl Runner { /// Process at most one inbound message. Returns [`StepReport::Idle`] /// if the mailbox is empty. pub fn step(&mut self) -> Result { + let step_start = Instant::now(); let msg = match self.mailbox.try_recv() { Some(m) => m, None => { self.metrics.inc_step("Idle"); + self.metrics + .observe_step_duration("Idle", step_start.elapsed()); return Ok(StepReport::Idle); } }; @@ -169,6 +172,8 @@ impl Runner { Ok(p) => p, Err(e) => { self.metrics.inc_step("Error"); + self.metrics + .observe_step_duration("Error", step_start.elapsed()); return Err(e); } }; @@ -244,7 +249,10 @@ impl Runner { // Execute. Executor errors don't count as denials — the // action passed all gates; the failure is downstream and is // recorded in the trace's `action.executed` outcome. + let exec_start = Instant::now(); let outcome = self.executor.execute(action).map_err(|e| e.to_string()); + self.metrics + .observe_action_execute_duration(kind, exec_start.elapsed()); self.trace .record_effect("action.executed", summary, outcome); self.metrics.inc_action_allowed(kind); @@ -252,6 +260,8 @@ impl Runner { } self.metrics.inc_step("Processed"); + self.metrics + .observe_step_duration("Processed", step_start.elapsed()); Ok(StepReport::Processed { allowed, denied }) }