diff --git a/crates/aisix-obs/src/metrics.rs b/crates/aisix-obs/src/metrics.rs index 287f43d4..1e586c00 100644 --- a/crates/aisix-obs/src/metrics.rs +++ b/crates/aisix-obs/src/metrics.rs @@ -24,14 +24,12 @@ use metrics_exporter_prometheus::{ Matcher, PrometheusBuilder, PrometheusHandle, PrometheusRecorder, }; use std::collections::HashMap; -use std::hash::{Hash, Hasher}; +use std::hash::Hasher; use std::sync::Arc; use std::sync::Mutex; use std::sync::OnceLock; use std::time::Duration; -use parking_lot::RwLock; - /// Metric names (public so the admin `/metrics` handler and tests can /// refer to them without typo risk). pub const M_REQUESTS_TOTAL: &str = "aisix_requests_total"; @@ -381,9 +379,19 @@ pub struct Metrics { struct MetricsInner { recorder: PrometheusRecorder, handle: PrometheusHandle, - proxy_in_flight: Mutex>, - request_series: RwLock, - usage_series: RwLock, + /// Per-(endpoint, protocol) in-flight counts. A linear scan over a + /// bounded slot list (route templates × protocols): the steady-state + /// hit is two pointer-length string compares with no allocation and + /// no hashing, where the map this replaced allocated two `String` + /// keys and ran SipHash on every request edge. Slots for drained + /// pairs stay at zero rather than being removed — the set is bounded, + /// and reuse beats churn. + proxy_in_flight: Mutex>, + /// Process-unique id prefixed onto every worker-cache key, so + /// thread-local entries minted for one instance's recorder can never + /// serve another instance (parallel tests build many `Metrics`). + /// Pre-formatted once — per-emit `fmt::write` was a visible cost. + worker_key_prefix: String, /// Constant `env_id` label for the SLO latency histograms — one DP /// process serves exactly one environment. `"unknown"` when the DP /// runs standalone (no control plane). @@ -403,37 +411,29 @@ struct ConfigLabelState { last_stale_kinds: std::collections::HashSet, } -const REQUEST_SERIES_CACHE_CAPACITY: usize = 1024; -const USAGE_SERIES_CACHE_CAPACITY: usize = 1024; - +/// Per-thread cap on each worker-cache map. Label sets are bounded by +/// construction (operator-configured names, fixed vocabularies), so +/// production never approaches this; it is a safety valve against an +/// unforeseen unbounded dimension pinning memory in every worker. +const WORKER_CACHE_CAPACITY: usize = 1024; + +/// Separator joining label values into a worker-cache key — a control +/// byte that no bounded label vocabulary contains. A value that DOES +/// contain it (nothing today) falls back to the uncached emit path +/// rather than risk two label sets aliasing one key. +const WORKER_KEY_SEP: char = '\u{1f}'; + +/// The per-request series handles every request-shaped emit resolves +/// through. Each field registers lazily on first use so the proxy-only +/// paths never mint `aisix_llm_*` series (and vice versa) — the same +/// series-sparsity the plain macro path had. #[derive(Default)] -struct RequestSeriesCache { - entries: HashMap>, - len: usize, -} - -struct CachedRequestSeries { - labels: RequestLabelsOwned, - handles: RequestSeriesHandles, -} - struct RequestSeriesHandles { - proxy_requests: metrics::Counter, - proxy_failed_requests: Option, - proxy_duration: metrics::Histogram, - llm_requests: metrics::Counter, - llm_duration: metrics::Histogram, -} - -#[derive(Default)] -struct UsageSeriesCache { - entries: HashMap>, - len: usize, -} - -struct CachedUsageSeries { - labels: UsageLabelsOwned, - handles: UsageSeriesHandles, + proxy_requests: OnceLock, + proxy_failed_requests: OnceLock, + proxy_duration: OnceLock, + llm_requests: OnceLock, + llm_duration: OnceLock, } /// Usage dimensions are registered independently so a zero-valued token or @@ -446,157 +446,162 @@ struct UsageSeriesHandles { spend_micro_usd: OnceLock, } -struct RequestLabelsOwned { - endpoint: String, - inbound_protocol: String, - provider: String, - model: String, - upstream_model: String, - provider_key_id: String, - provider_key_name: String, - api_key_id: String, - team_id: String, - user_id: String, - user_name: String, - stream: bool, - is_fallback: bool, - status: u16, - outcome: RequestOutcome, +// ── Per-worker handle cache ──────────────────────────────────────────────── +// +// `metrics::counter!`-family macros rebuild a `Key` (one owned `String` per +// label), hash it, and probe the recorder's sharded registry on EVERY emit. +// `metrics::Counter`/`Gauge`/`Histogram` are `Arc`-backed handles wired +// straight to the series' storage, so registering once per label set and +// reusing the handle removes all of that from the steady-state path. +// +// The cache is `thread_local!`, not shared: this process runs one +// current-thread runtime per pinned core (thread-per-core), and a shared +// map — like the two `RwLock` caches this replaced — puts one contended +// cache line (the lock word) in front of every emit on every worker. The +// spike for AISIX-Cloud#1259 item 3b measured that shared-lock variant +// recovering almost nothing (+0.5% throughput) while the thread-local +// variant recovered +4.4%; per-worker duplication of a bounded handle set +// is the whole trick. +// +// Correctness properties: +// - Keys are `instance id \x1f site \x1f label values…`, so two `Metrics` +// instances on one thread (parallel tests) can never serve each other's +// recorder, and one site's entries can never answer another site. +// - Values never alias: label values are joined with a control byte no +// bounded label vocabulary contains, and a value that does contain it +// falls back to the uncached emit instead of being cached. +// - Eviction only drops OUR reference. The series and its value live in +// the recorder's registry; re-registering the same labels returns a +// handle to the SAME storage, so counts continue exactly where they +// left off. + +/// FNV-1a, hand-rolled: three instructions per byte, no dependency, and +/// no DoS surface — every byte hashed here is operator-bounded config +/// vocabulary, never attacker-chosen cardinality (#451 keeps raw client +/// strings out of label values by contract). +struct FnvHasher(u64); + +impl Default for FnvHasher { + fn default() -> Self { + Self(0xcbf2_9ce4_8422_2325) + } } -impl RequestLabelsOwned { - fn new(labels: RequestLabels<'_>) -> Self { - Self { - endpoint: labels.endpoint.to_string(), - inbound_protocol: labels.inbound_protocol.to_string(), - provider: labels.provider.to_string(), - model: labels.model.to_string(), - upstream_model: labels.upstream_model.to_string(), - provider_key_id: labels.provider_key_id.to_string(), - provider_key_name: labels.provider_key_name.to_string(), - api_key_id: labels.api_key_id.to_string(), - team_id: labels.team_id.to_string(), - user_id: labels.user_id.to_string(), - user_name: labels.user_name.to_string(), - stream: labels.stream, - is_fallback: labels.is_fallback, - status: labels.status, - outcome: labels.outcome, - } +impl Hasher for FnvHasher { + fn finish(&self) -> u64 { + self.0 } - fn matches(&self, labels: RequestLabels<'_>) -> bool { - self.endpoint == labels.endpoint - && self.inbound_protocol == labels.inbound_protocol - && self.provider == labels.provider - && self.model == labels.model - && self.upstream_model == labels.upstream_model - && self.provider_key_id == labels.provider_key_id - && self.provider_key_name == labels.provider_key_name - && self.api_key_id == labels.api_key_id - && self.team_id == labels.team_id - && self.user_id == labels.user_id - && self.user_name == labels.user_name - && self.stream == labels.stream - && self.is_fallback == labels.is_fallback - && self.status == labels.status - && self.outcome == labels.outcome + fn write(&mut self, bytes: &[u8]) { + for &b in bytes { + self.0 ^= u64::from(b); + self.0 = self.0.wrapping_mul(0x0000_0100_0000_01b3); + } } } -struct UsageLabelsOwned { - endpoint: String, - inbound_protocol: String, - provider: String, - model: String, - upstream_model: String, - provider_key_id: String, - provider_key_name: String, - api_key_id: String, - team_id: String, - user_id: String, - user_name: String, +type FnvMap = HashMap, T, std::hash::BuildHasherDefault>; + +/// One worker's handle maps, one per handle shape. Field-per-shape rather +/// than a value enum so every site gets its concrete handle type back +/// without a runtime discriminant. +#[derive(Default)] +struct WorkerCache { + counters: FnvMap, + gauges: FnvMap, + histograms: FnvMap, + request_series: FnvMap, + usage_series: FnvMap, + /// Reused key buffer: the steady-state emit builds its lookup key in + /// place and allocates nothing; only a first-sight miss allocates the + /// stored `Box`. + key_buf: String, } -impl UsageLabelsOwned { - fn new(labels: UsageLabels<'_>) -> Self { - Self { - endpoint: labels.endpoint.to_string(), - inbound_protocol: labels.inbound_protocol.to_string(), - provider: labels.provider.to_string(), - model: labels.model.to_string(), - upstream_model: labels.upstream_model.to_string(), - provider_key_id: labels.provider_key_id.to_string(), - provider_key_name: labels.provider_key_name.to_string(), - api_key_id: labels.api_key_id.to_string(), - team_id: labels.team_id.to_string(), - user_id: labels.user_id.to_string(), - user_name: labels.user_name.to_string(), - } +thread_local! { + static WORKER_CACHE: std::cell::RefCell = + std::cell::RefCell::new(WorkerCache::default()); +} + +/// Selects which [`WorkerCache`] map a handle shape lives in. +/// Hands back a handle-shape's map TOGETHER with the built key. The pair +/// comes from disjoint `WorkerCache` fields, which the per-impl field +/// split proves to the borrow checker — a `&mut WorkerCache -> &mut map` +/// signature would pin the whole cache and forbid reading the key. +trait WorkerCached: Sized { + fn slot_with_key(cache: &mut WorkerCache) -> (&mut FnvMap, &str); +} + +impl WorkerCached for metrics::Counter { + fn slot_with_key(cache: &mut WorkerCache) -> (&mut FnvMap, &str) { + (&mut cache.counters, cache.key_buf.as_str()) } +} - fn matches(&self, labels: UsageLabels<'_>) -> bool { - self.endpoint == labels.endpoint - && self.inbound_protocol == labels.inbound_protocol - && self.provider == labels.provider - && self.model == labels.model - && self.upstream_model == labels.upstream_model - && self.provider_key_id == labels.provider_key_id - && self.provider_key_name == labels.provider_key_name - && self.api_key_id == labels.api_key_id - && self.team_id == labels.team_id - && self.user_id == labels.user_id - && self.user_name == labels.user_name +impl WorkerCached for metrics::Gauge { + fn slot_with_key(cache: &mut WorkerCache) -> (&mut FnvMap, &str) { + (&mut cache.gauges, cache.key_buf.as_str()) } } -impl RequestSeriesCache { - fn get(&self, hash: u64, labels: RequestLabels<'_>) -> Option<&RequestSeriesHandles> { - self.entries - .get(&hash)? - .iter() - .find(|entry| entry.labels.matches(labels)) - .map(|entry| &entry.handles) +impl WorkerCached for metrics::Histogram { + fn slot_with_key(cache: &mut WorkerCache) -> (&mut FnvMap, &str) { + (&mut cache.histograms, cache.key_buf.as_str()) } +} - fn insert(&mut self, hash: u64, labels: RequestLabelsOwned, handles: RequestSeriesHandles) { - if self.len >= REQUEST_SERIES_CACHE_CAPACITY { - if let Some(evicted_hash) = self.entries.keys().next().copied() { - if let Some(evicted) = self.entries.remove(&evicted_hash) { - self.len -= evicted.len(); - } - } - } - self.entries - .entry(hash) - .or_default() - .push(CachedRequestSeries { labels, handles }); - self.len += 1; +impl WorkerCached for RequestSeriesHandles { + fn slot_with_key(cache: &mut WorkerCache) -> (&mut FnvMap, &str) { + (&mut cache.request_series, cache.key_buf.as_str()) } } -impl UsageSeriesCache { - fn get(&self, hash: u64, labels: UsageLabels<'_>) -> Option<&UsageSeriesHandles> { - self.entries - .get(&hash)? - .iter() - .find(|entry| entry.labels.matches(labels)) - .map(|entry| &entry.handles) +impl WorkerCached for UsageSeriesHandles { + fn slot_with_key(cache: &mut WorkerCache) -> (&mut FnvMap, &str) { + (&mut cache.usage_series, cache.key_buf.as_str()) + } +} + +/// Writes one emit's label values into the worker-cache key buffer. +/// Numeric/bool writers exist so `u16` statuses and flags key without a +/// heap allocation; `dirty` flips when a value contains the separator, +/// which sends that emit down the uncached path. +struct WorkerKey<'a> { + buf: &'a mut String, + dirty: bool, +} + +impl WorkerKey<'_> { + fn label(&mut self, value: &str) { + if value.contains(WORKER_KEY_SEP) { + self.dirty = true; + } + self.buf.push(WORKER_KEY_SEP); + self.buf.push_str(value); } - fn insert(&mut self, hash: u64, labels: UsageLabelsOwned, handles: UsageSeriesHandles) { - if self.len >= USAGE_SERIES_CACHE_CAPACITY { - if let Some(evicted_hash) = self.entries.keys().next().copied() { - if let Some(evicted) = self.entries.remove(&evicted_hash) { - self.len -= evicted.len(); - } + fn label_u16(&mut self, value: u16) { + self.buf.push(WORKER_KEY_SEP); + // Hand-rolled digits: `fmt::write` machinery was a visible + // per-emit cost for what is a five-byte-max ASCII render. + let mut digits = [0u8; 5]; + let mut i = digits.len(); + let mut v = value; + loop { + i -= 1; + digits[i] = b'0' + (v % 10) as u8; + v /= 10; + if v == 0 { + break; } } - self.entries - .entry(hash) - .or_default() - .push(CachedUsageSeries { labels, handles }); - self.len += 1; + self.buf + .push_str(std::str::from_utf8(&digits[i..]).expect("ascii digits")); + } + + fn label_bool(&mut self, value: bool) { + self.buf.push(WORKER_KEY_SEP); + self.buf.push(if value { 't' } else { 'f' }); } } @@ -646,13 +651,16 @@ impl Metrics { .expect("bucket lists are validated non-empty") .build_recorder(); let handle = recorder.handle(); + static NEXT_INSTANCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); Self { inner: Arc::new(MetricsInner { recorder, handle, - proxy_in_flight: Mutex::new(HashMap::new()), - request_series: RwLock::new(RequestSeriesCache::default()), - usage_series: RwLock::new(UsageSeriesCache::default()), + proxy_in_flight: Mutex::new(Vec::new()), + worker_key_prefix: format!( + "{:x}", + NEXT_INSTANCE.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + ), env_id: if env_id.is_empty() { "unknown".to_string() } else { @@ -686,6 +694,12 @@ impl Metrics { /// `source_connected`) are emitted only in etcd mode. Label churn on the /// info/rejected gauges (`hash_info`, `rejected_resources`) zeroes the /// prior label set so the exposition never carries two live samples. + /// + /// Deliberately NOT routed through the per-worker handle cache: this + /// runs once per scrape (not per request), and the zeroing discipline + /// above works on churning label sets — exactly the shape a + /// first-seen handle cache handles worst. The macro path's per-call + /// registration cost is irrelevant at scrape frequency. pub fn sync_config_status(&self, view: &aisix_core::ConfigMetricsView) { use aisix_core::SourceKind; let etcd = matches!(view.source_kind, SourceKind::Etcd); @@ -791,23 +805,42 @@ impl Metrics { outcome: RequestOutcome, duration: Duration, ) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_REQUESTS_TOTAL, - "provider" => provider.to_string(), - "model" => model.to_string(), - "status" => status.to_string(), - "outcome" => outcome.as_str().to_string(), - ) - .increment(1); - metrics::histogram!( - M_REQUEST_DURATION, - "provider" => provider.to_string(), - "model" => model.to_string(), - "status" => status.to_string(), - ) - .record(duration.as_secs_f64()); - }); + self.cached_counter( + M_REQUESTS_TOTAL, + 1, + |k| { + k.label(provider); + k.label(model); + k.label_u16(status); + k.label(outcome.as_str()); + }, + || { + metrics::counter!( + M_REQUESTS_TOTAL, + "provider" => provider.to_string(), + "model" => model.to_string(), + "status" => status.to_string(), + "outcome" => outcome.as_str().to_string(), + ) + }, + ); + self.cached_histogram( + M_REQUEST_DURATION, + duration.as_secs_f64(), + |k| { + k.label(provider); + k.label(model); + k.label_u16(status); + }, + || { + metrics::histogram!( + M_REQUEST_DURATION, + "provider" => provider.to_string(), + "model" => model.to_string(), + "status" => status.to_string(), + ) + }, + ); } /// Record one inbound authentication decision on @@ -815,15 +848,25 @@ impl Metrics { /// proxy auth choke point for every credential judgment — allowed /// or denied, API-key and JWT paths alike. pub fn record_auth_decision(&self, method: &str, allowed: bool, reason: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_AUTH_DECISIONS_TOTAL, - "method" => method.to_string(), - "result" => if allowed { "allowed" } else { "denied" }.to_string(), - "reason" => if reason.is_empty() { "none" } else { reason }.to_string(), - ) - .increment(1); - }); + let result = if allowed { "allowed" } else { "denied" }; + let reason = if reason.is_empty() { "none" } else { reason }; + self.cached_counter( + M_AUTH_DECISIONS_TOTAL, + 1, + |k| { + k.label(method); + k.label(result); + k.label(reason); + }, + || { + metrics::counter!( + M_AUTH_DECISIONS_TOTAL, + "method" => method.to_string(), + "result" => result.to_string(), + "reason" => reason.to_string(), + ) + }, + ); } /// Record one request's guardrail outcome. Called once per request from @@ -831,18 +874,27 @@ impl Metrics { /// `guardrail_blocked` / `guardrail_bypassed_reason` fields. An empty /// `bypass_reason` means no bypass occurred. pub fn record_guardrail_outcome(&self, blocked: bool, bypass_reason: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - if blocked { - metrics::counter!(M_GUARDRAIL_BLOCKS_TOTAL).increment(1); - } - if !bypass_reason.is_empty() { - metrics::counter!( - M_GUARDRAIL_BYPASSES_TOTAL, - "reason" => bypass_reason.to_string(), - ) - .increment(1); - } - }); + if blocked { + self.cached_counter( + M_GUARDRAIL_BLOCKS_TOTAL, + 1, + |_| {}, + || metrics::counter!(M_GUARDRAIL_BLOCKS_TOTAL), + ); + } + if !bypass_reason.is_empty() { + self.cached_counter( + M_GUARDRAIL_BYPASSES_TOTAL, + 1, + |k| k.label(bypass_reason), + || { + metrics::counter!( + M_GUARDRAIL_BYPASSES_TOTAL, + "reason" => bypass_reason.to_string(), + ) + }, + ); + } } /// Record one guardrail member execution on @@ -850,18 +902,30 @@ impl Metrics { /// chain fold through the `GuardrailMetricsSink` impl below — once per /// member per hook pass, on every handler. pub fn record_guardrail_execution(&self, exec: &aisix_core::GuardrailExecution<'_>) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::histogram!( - M_GUARDRAIL_LATENCY_SECONDS, - "env_id" => self.inner.env_id.clone(), - "guardrail" => exec.guardrail_name.to_string(), - "kind" => exec.kind.to_string(), - "phase" => exec.phase.to_string(), - "result" => exec.result.to_string(), - "error_type" => exec.error_type.unwrap_or("none").to_string(), - ) - .record(exec.elapsed.as_secs_f64()); - }); + // `env_id` is constant per instance and the cache key is already + // instance-scoped, so it stays out of the key. + self.cached_histogram( + M_GUARDRAIL_LATENCY_SECONDS, + exec.elapsed.as_secs_f64(), + |k| { + k.label(exec.guardrail_name); + k.label(exec.kind); + k.label(exec.phase); + k.label(exec.result); + k.label(exec.error_type.unwrap_or("none")); + }, + || { + metrics::histogram!( + M_GUARDRAIL_LATENCY_SECONDS, + "env_id" => self.inner.env_id.clone(), + "guardrail" => exec.guardrail_name.to_string(), + "kind" => exec.kind.to_string(), + "phase" => exec.phase.to_string(), + "result" => exec.result.to_string(), + "error_type" => exec.error_type.unwrap_or("none").to_string(), + ) + }, + ); } /// Count one rate-limit rejection. `scope` is the exceeded @@ -872,80 +936,118 @@ impl Metrics { /// Recorded at the quota gate, the one point every endpoint funnels /// through (AISIX-Cloud#892). pub fn record_ratelimit_rejection(&self, scope: &str, layer: &str, policy_id: Option<&str>) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_RATELIMIT_REJECTIONS, - "scope" => scope.to_string(), - "layer" => layer.to_string(), - "policy_id" => policy_id.unwrap_or_default().to_string(), - ) - .increment(1); - }); + let policy_id = policy_id.unwrap_or_default(); + self.cached_counter( + M_RATELIMIT_REJECTIONS, + 1, + |k| { + k.label(scope); + k.label(layer); + k.label(policy_id); + }, + || { + metrics::counter!( + M_RATELIMIT_REJECTIONS, + "scope" => scope.to_string(), + "layer" => layer.to_string(), + "policy_id" => policy_id.to_string(), + ) + }, + ); } /// Count one cache-gate outcome. `policy` is the matched policy's /// name, `outcome` one of the fixed [`M_CACHE_REQUESTS_TOTAL`] /// values. pub fn record_cache_event(&self, policy: &str, outcome: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_CACHE_REQUESTS_TOTAL, - "policy" => policy.to_string(), - "outcome" => outcome.to_string(), - ) - .increment(1); - }); + self.cached_counter( + M_CACHE_REQUESTS_TOTAL, + 1, + |k| { + k.label(policy); + k.label(outcome); + }, + || { + metrics::counter!( + M_CACHE_REQUESTS_TOTAL, + "policy" => policy.to_string(), + "outcome" => outcome.to_string(), + ) + }, + ); } /// Record one successful cache semantic-layer embedding call. pub fn record_cache_semantic_embed(&self, policy: &str, elapsed: Duration) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::histogram!( - M_CACHE_SEMANTIC_EMBED_SECONDS, - "policy" => policy.to_string(), - ) - .record(elapsed.as_secs_f64()); - }); + self.cached_histogram( + M_CACHE_SEMANTIC_EMBED_SECONDS, + elapsed.as_secs_f64(), + |k| k.label(policy), + || { + metrics::histogram!( + M_CACHE_SEMANTIC_EMBED_SECONDS, + "policy" => policy.to_string(), + ) + }, + ); } /// Count one failed cache semantic-layer embedding attempt. `cause` /// is one of the fixed [`M_CACHE_SEMANTIC_EMBED_FAILURES_TOTAL`] /// values. pub fn record_cache_semantic_embed_failure(&self, policy: &str, cause: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_CACHE_SEMANTIC_EMBED_FAILURES_TOTAL, - "policy" => policy.to_string(), - "cause" => cause.to_string(), - ) - .increment(1); - }); + self.cached_counter( + M_CACHE_SEMANTIC_EMBED_FAILURES_TOTAL, + 1, + |k| { + k.label(policy); + k.label(cause); + }, + || { + metrics::counter!( + M_CACHE_SEMANTIC_EMBED_FAILURES_TOTAL, + "policy" => policy.to_string(), + "cause" => cause.to_string(), + ) + }, + ); } /// Count one failed semantic-store operation. `op` is one of the /// fixed [`M_CACHE_SEMANTIC_STORE_FAILURES_TOTAL`] values. pub fn record_cache_semantic_store_failure(&self, policy: &str, op: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_CACHE_SEMANTIC_STORE_FAILURES_TOTAL, - "policy" => policy.to_string(), - "op" => op.to_string(), - ) - .increment(1); - }); + self.cached_counter( + M_CACHE_SEMANTIC_STORE_FAILURES_TOTAL, + 1, + |k| { + k.label(policy); + k.label(op); + }, + || { + metrics::counter!( + M_CACHE_SEMANTIC_STORE_FAILURES_TOTAL, + "policy" => policy.to_string(), + "op" => op.to_string(), + ) + }, + ); } /// Count a request the client abandoned before it produced a response /// head. `endpoint` must already be a bounded route template — see /// [`M_PROXY_CLIENT_CANCELLED_TOTAL`]. pub fn record_client_cancelled(&self, endpoint: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_PROXY_CLIENT_CANCELLED_TOTAL, - "endpoint" => endpoint.to_string(), - ) - .increment(1); - }); + self.cached_counter( + M_PROXY_CLIENT_CANCELLED_TOTAL, + 1, + |k| k.label(endpoint), + || { + metrics::counter!( + M_PROXY_CLIENT_CANCELLED_TOTAL, + "endpoint" => endpoint.to_string(), + ) + }, + ); } /// Count a request refused by the request-body cap. `endpoint` must @@ -957,117 +1059,170 @@ impl Metrics { inbound_protocol: &str, outcome: &str, ) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_PROXY_BODY_LIMIT_REJECTIONS_TOTAL, - "endpoint" => endpoint.to_string(), - "inbound_protocol" => inbound_protocol.to_string(), - "outcome" => outcome.to_string(), - ) - .increment(1); - }); + self.cached_counter( + M_PROXY_BODY_LIMIT_REJECTIONS_TOTAL, + 1, + |k| { + k.label(endpoint); + k.label(inbound_protocol); + k.label(outcome); + }, + || { + metrics::counter!( + M_PROXY_BODY_LIMIT_REJECTIONS_TOTAL, + "endpoint" => endpoint.to_string(), + "inbound_protocol" => inbound_protocol.to_string(), + "outcome" => outcome.to_string(), + ) + }, + ); } pub fn record_tokens(&self, provider: &str, model: &str, total_tokens: u64) { if total_tokens == 0 { return; } - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_TOKENS_CONSUMED, - "provider" => provider.to_string(), - "model" => model.to_string(), - ) - .increment(total_tokens); - }); + self.cached_counter( + M_TOKENS_CONSUMED, + total_tokens, + |k| { + k.label(provider); + k.label(model); + }, + || { + metrics::counter!( + M_TOKENS_CONSUMED, + "provider" => provider.to_string(), + "model" => model.to_string(), + ) + }, + ); } pub fn increment_proxy_in_flight(&self, endpoint: &str, inbound_protocol: &str) { - let value = { - let mut counters = self.inner.proxy_in_flight.lock().expect("lock in-flight"); - let value = counters - .entry((endpoint.to_string(), inbound_protocol.to_string())) - .or_insert(0); - *value += 1; - *value - }; - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::gauge!( - M_PROXY_IN_FLIGHT, - "endpoint" => endpoint.to_string(), - "inbound_protocol" => inbound_protocol.to_string(), - ) - .set(value as f64); - }); + self.apply_in_flight_delta(endpoint, inbound_protocol, 1); } pub fn decrement_proxy_in_flight(&self, endpoint: &str, inbound_protocol: &str) { - let value = { - let mut counters = self.inner.proxy_in_flight.lock().expect("lock in-flight"); - let key = (endpoint.to_string(), inbound_protocol.to_string()); - let value = counters.entry(key.clone()).or_insert(0); - *value = (*value - 1).max(0); - let current = *value; - if current == 0 { - counters.remove(&key); - } - current + self.apply_in_flight_delta(endpoint, inbound_protocol, -1); + } + + /// Apply one in-flight edge and publish the resulting count on the + /// gauge WHILE the slot lock is held, so concurrent edges on one pair + /// cannot publish out of order and strand a stale value on an + /// endpoint that then goes idle. The count clamps at zero (a + /// decrement without its increment must not wedge the gauge + /// negative). No path takes this lock from inside the worker cache, + /// so emitting under it cannot deadlock. + /// + /// Precondition: `endpoint` must already be a bounded route template + /// (`normalize_endpoint_label` output, #451) and `inbound_protocol` a + /// fixed vocabulary. The slot vector's boundedness — and the linear + /// scan's cost — depend on it; a raw request path here would grow the + /// vector without bound. + fn apply_in_flight_delta(&self, endpoint: &str, inbound_protocol: &str, delta: i64) { + let mut slots = self.inner.proxy_in_flight.lock().expect("lock in-flight"); + let value = if let Some(slot) = slots + .iter_mut() + .find(|(e, p, _)| e == endpoint && p == inbound_protocol) + { + slot.2 = (slot.2 + delta).max(0); + slot.2 + } else { + let value = delta.max(0); + slots.push((endpoint.to_string(), inbound_protocol.to_string(), value)); + value }; - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::gauge!( - M_PROXY_IN_FLIGHT, - "endpoint" => endpoint.to_string(), - "inbound_protocol" => inbound_protocol.to_string(), - ) - .set(value as f64); + self.set_proxy_in_flight_gauge(endpoint, inbound_protocol, value); + } + + /// Shared gauge emit for the increment/decrement pair — one cache + /// entry, since both sides address the same series. + fn set_proxy_in_flight_gauge(&self, endpoint: &str, inbound_protocol: &str, value: i64) { + self.cached_gauge( + M_PROXY_IN_FLIGHT, + value as f64, + |k| { + k.label(endpoint); + k.label(inbound_protocol); + }, + || { + metrics::gauge!( + M_PROXY_IN_FLIGHT, + "endpoint" => endpoint.to_string(), + "inbound_protocol" => inbound_protocol.to_string(), + ) + }, + ); + } + + /// Emit through this worker's cached handle for `(site, label values)`, + /// registering via `register` on the first sight of that label set on + /// this thread. + /// + /// `build_key` writes the label VALUES in a fixed per-site order; + /// `use_handle` performs the actual increment/set/record. On the + /// steady-state path this allocates nothing and touches no shared + /// state beyond the series' own value atomics. + /// + /// Invariant: `register` and `use_handle` must not re-enter any + /// `Metrics` emit (they would hit the `RefCell` re-borrow) — they only + /// register or write handles, and registration cannot emit. + fn with_worker_handle( + &self, + site: &'static str, + build_key: impl FnOnce(&mut WorkerKey<'_>), + register: impl FnOnce() -> H, + use_handle: impl FnOnce(&H), + ) { + WORKER_CACHE.with(|cell| { + let cache = &mut *cell.borrow_mut(); + cache.key_buf.clear(); + let dirty = { + let mut key = WorkerKey { + buf: &mut cache.key_buf, + dirty: false, + }; + key.buf.push_str(&self.inner.worker_key_prefix); + key.buf.push(WORKER_KEY_SEP); + key.buf.push_str(site); + build_key(&mut key); + key.dirty + }; + if dirty { + // A label value contained the key separator: emit through + // a freshly registered handle instead of risking two label + // sets aliasing one cache key. Same series, slow path. + use_handle(®ister()); + return; + } + let (map, key) = H::slot_with_key(cache); + if let Some(handle) = map.get(key) { + use_handle(handle); + return; + } + let handle = register(); + use_handle(&handle); + if map.len() >= WORKER_CACHE_CAPACITY { + if let Some(evicted) = map.keys().next().cloned() { + map.remove(&evicted); + } + } + map.insert(Box::from(key), handle); }); } - fn request_series_hash(labels: RequestLabels<'_>) -> u64 { - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - labels.endpoint.hash(&mut hasher); - labels.inbound_protocol.hash(&mut hasher); - labels.provider.hash(&mut hasher); - labels.model.hash(&mut hasher); - labels.upstream_model.hash(&mut hasher); - labels.provider_key_id.hash(&mut hasher); - labels.provider_key_name.hash(&mut hasher); - labels.api_key_id.hash(&mut hasher); - labels.team_id.hash(&mut hasher); - labels.user_id.hash(&mut hasher); - labels.user_name.hash(&mut hasher); - labels.stream.hash(&mut hasher); - labels.is_fallback.hash(&mut hasher); - labels.status.hash(&mut hasher); - labels.outcome.as_str().hash(&mut hasher); - hasher.finish() - } - - fn usage_series_hash(labels: UsageLabels<'_>) -> u64 { - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - labels.endpoint.hash(&mut hasher); - labels.inbound_protocol.hash(&mut hasher); - labels.provider.hash(&mut hasher); - labels.model.hash(&mut hasher); - labels.upstream_model.hash(&mut hasher); - labels.provider_key_id.hash(&mut hasher); - labels.provider_key_name.hash(&mut hasher); - labels.api_key_id.hash(&mut hasher); - labels.team_id.hash(&mut hasher); - labels.user_id.hash(&mut hasher); - labels.user_name.hash(&mut hasher); - hasher.finish() - } - - fn register_request_series(&self, labels: RequestLabels<'_>) -> RequestSeriesHandles { - metrics::with_local_recorder(&self.inner.recorder, || RequestSeriesHandles { - proxy_requests: labels.request_counter(M_PROXY_REQUESTS_TOTAL), - proxy_failed_requests: (labels.outcome != RequestOutcome::Success) - .then(|| labels.request_counter(M_PROXY_FAILED_REQUESTS_TOTAL)), - proxy_duration: labels.request_duration_histogram(M_PROXY_REQUEST_DURATION), - llm_requests: labels.request_counter(M_LLM_REQUESTS_TOTAL), - llm_duration: labels.request_duration_histogram(M_LLM_REQUEST_DURATION), - }) + /// Test-only view of this thread's cached request-series entry count. + /// Each `#[test]` runs on its own thread, so the count starts at zero + /// and covers exactly that test's emits. + #[cfg(test)] + fn worker_request_series_len() -> usize { + WORKER_CACHE.with(|cell| cell.borrow().request_series.len()) + } + + #[cfg(test)] + fn worker_usage_series_len() -> usize { + WORKER_CACHE.with(|cell| cell.borrow().usage_series.len()) } fn with_request_series( @@ -1075,74 +1230,137 @@ impl Metrics { labels: RequestLabels<'_>, record: impl FnOnce(&RequestSeriesHandles), ) { - let hash = Self::request_series_hash(labels); - { - let cache = self.inner.request_series.read(); - if let Some(handles) = cache.get(hash, labels) { - record(handles); - return; - } - } - - let handles = self.register_request_series(labels); - let mut cache = self.inner.request_series.write(); - if let Some(existing) = cache.get(hash, labels) { - record(existing); - return; - } - cache.insert(hash, RequestLabelsOwned::new(labels), handles); - record( - cache - .get(hash, labels) - .expect("request series was just inserted"), + self.with_worker_handle( + "request_series", + |k| { + k.label(labels.endpoint); + k.label(labels.inbound_protocol); + k.label(labels.provider); + k.label(labels.model); + k.label(labels.upstream_model); + k.label(labels.provider_key_id); + k.label(labels.provider_key_name); + k.label(labels.api_key_id); + k.label(labels.team_id); + k.label(labels.user_id); + k.label(labels.user_name); + k.label_bool(labels.stream); + k.label_bool(labels.is_fallback); + k.label_u16(labels.status); + k.label(labels.outcome.as_str()); + }, + RequestSeriesHandles::default, + record, ); } fn with_usage_series(&self, labels: UsageLabels<'_>, record: impl FnOnce(&UsageSeriesHandles)) { - let hash = Self::usage_series_hash(labels); - { - let cache = self.inner.usage_series.read(); - if let Some(handles) = cache.get(hash, labels) { - record(handles); - return; - } - } + self.with_worker_handle( + "usage_series", + |k| { + k.label(labels.endpoint); + k.label(labels.inbound_protocol); + k.label(labels.provider); + k.label(labels.model); + k.label(labels.upstream_model); + k.label(labels.provider_key_id); + k.label(labels.provider_key_name); + k.label(labels.api_key_id); + k.label(labels.team_id); + k.label(labels.user_id); + k.label(labels.user_name); + }, + UsageSeriesHandles::default, + record, + ); + } - let mut cache = self.inner.usage_series.write(); - if cache.get(hash, labels).is_none() { - cache.insert( - hash, - UsageLabelsOwned::new(labels), - UsageSeriesHandles::default(), - ); - } - record( - cache - .get(hash, labels) - .expect("usage series was just inserted"), + /// Resolve one lazily-registered handle in a series bundle. The + /// recorder guard is paid only on the one-time init path; a cache-hit + /// emit never touches it. + fn init_handle<'h, T>(&self, slot: &'h OnceLock, init: impl FnOnce() -> T) -> &'h T { + slot.get_or_init(|| metrics::with_local_recorder(&self.inner.recorder, init)) + } + + /// Increment a worker-cached counter. `register` runs under the + /// recorder guard on the one-time miss path only. + fn cached_counter( + &self, + site: &'static str, + by: u64, + build_key: impl FnOnce(&mut WorkerKey<'_>), + register: impl FnOnce() -> metrics::Counter, + ) { + self.with_worker_handle( + site, + build_key, + || metrics::with_local_recorder(&self.inner.recorder, register), + |counter| counter.increment(by), + ); + } + + /// Set a worker-cached gauge, same contract as [`Self::cached_counter`]. + fn cached_gauge( + &self, + site: &'static str, + value: f64, + build_key: impl FnOnce(&mut WorkerKey<'_>), + register: impl FnOnce() -> metrics::Gauge, + ) { + self.with_worker_handle( + site, + build_key, + || metrics::with_local_recorder(&self.inner.recorder, register), + |gauge| gauge.set(value), + ); + } + + /// Record into a worker-cached histogram, same contract as + /// [`Self::cached_counter`]. + fn cached_histogram( + &self, + site: &'static str, + value: f64, + build_key: impl FnOnce(&mut WorkerKey<'_>), + register: impl FnOnce() -> metrics::Histogram, + ) { + self.with_worker_handle( + site, + build_key, + || metrics::with_local_recorder(&self.inner.recorder, register), + |histogram| histogram.record(value), ); } pub fn record_proxy_request(&self, labels: RequestLabels<'_>, duration: Duration) { - metrics::with_local_recorder(&self.inner.recorder, || { - labels.request_counter(M_PROXY_REQUESTS_TOTAL).increment(1); - labels - .request_duration_histogram(M_PROXY_REQUEST_DURATION) - .record(duration.as_secs_f64()); + self.with_request_series(labels, |h| { + self.init_handle(&h.proxy_requests, || { + labels.request_counter(M_PROXY_REQUESTS_TOTAL) + }) + .increment(1); + self.init_handle(&h.proxy_duration, || { + labels.request_duration_histogram(M_PROXY_REQUEST_DURATION) + }) + .record(duration.as_secs_f64()); if labels.outcome != RequestOutcome::Success { - labels - .request_counter(M_PROXY_FAILED_REQUESTS_TOTAL) - .increment(1); + self.init_handle(&h.proxy_failed_requests, || { + labels.request_counter(M_PROXY_FAILED_REQUESTS_TOTAL) + }) + .increment(1); } }); } pub fn record_llm_request(&self, labels: RequestLabels<'_>, duration: Duration) { - metrics::with_local_recorder(&self.inner.recorder, || { - labels.request_counter(M_LLM_REQUESTS_TOTAL).increment(1); - labels - .request_duration_histogram(M_LLM_REQUEST_DURATION) - .record(duration.as_secs_f64()); + self.with_request_series(labels, |h| { + self.init_handle(&h.llm_requests, || { + labels.request_counter(M_LLM_REQUESTS_TOTAL) + }) + .increment(1); + self.init_handle(&h.llm_duration, || { + labels.request_duration_histogram(M_LLM_REQUEST_DURATION) + }) + .record(duration.as_secs_f64()); }); } @@ -1150,13 +1368,29 @@ impl Metrics { /// All request handlers emit these together with the same end-to-end /// duration. pub fn record_proxy_and_llm_request(&self, labels: RequestLabels<'_>, duration: Duration) { - self.with_request_series(labels, |handles| { - handles.proxy_requests.increment(1); - handles.proxy_duration.record(duration.as_secs_f64()); - handles.llm_requests.increment(1); - handles.llm_duration.record(duration.as_secs_f64()); - if let Some(failed) = &handles.proxy_failed_requests { - failed.increment(1); + let secs = duration.as_secs_f64(); + self.with_request_series(labels, |h| { + self.init_handle(&h.proxy_requests, || { + labels.request_counter(M_PROXY_REQUESTS_TOTAL) + }) + .increment(1); + self.init_handle(&h.proxy_duration, || { + labels.request_duration_histogram(M_PROXY_REQUEST_DURATION) + }) + .record(secs); + self.init_handle(&h.llm_requests, || { + labels.request_counter(M_LLM_REQUESTS_TOTAL) + }) + .increment(1); + self.init_handle(&h.llm_duration, || { + labels.request_duration_histogram(M_LLM_REQUEST_DURATION) + }) + .record(secs); + if labels.outcome != RequestOutcome::Success { + self.init_handle(&h.proxy_failed_requests, || { + labels.request_counter(M_PROXY_FAILED_REQUESTS_TOTAL) + }) + .increment(1); } }); } @@ -1171,56 +1405,70 @@ impl Metrics { return; } self.with_usage_series(labels, |handles| { - metrics::with_local_recorder(&self.inner.recorder, || { - if usage.input_tokens > 0 { - handles - .input_tokens - .get_or_init(|| labels.counter(M_LLM_INPUT_TOKENS_TOTAL)) - .increment(u64::from(usage.input_tokens)); - } - if usage.output_tokens > 0 { - handles - .output_tokens - .get_or_init(|| labels.counter(M_LLM_OUTPUT_TOKENS_TOTAL)) - .increment(u64::from(usage.output_tokens)); - } - if usage.total_tokens > 0 { - handles - .total_tokens - .get_or_init(|| labels.counter(M_LLM_TOTAL_TOKENS_TOTAL)) - .increment(u64::from(usage.total_tokens)); - } - if let Some(value) = spend_micro_usd { - handles - .spend_micro_usd - .get_or_init(|| labels.counter(M_LLM_SPEND_MICRO_USD_TOTAL)) - .increment(value); - } - }); - }); - } + if usage.input_tokens > 0 { + self.init_handle(&handles.input_tokens, || { + labels.counter(M_LLM_INPUT_TOKENS_TOTAL) + }) + .increment(u64::from(usage.input_tokens)); + } + if usage.output_tokens > 0 { + self.init_handle(&handles.output_tokens, || { + labels.counter(M_LLM_OUTPUT_TOKENS_TOTAL) + }) + .increment(u64::from(usage.output_tokens)); + } + if usage.total_tokens > 0 { + self.init_handle(&handles.total_tokens, || { + labels.counter(M_LLM_TOTAL_TOKENS_TOTAL) + }) + .increment(u64::from(usage.total_tokens)); + } + if let Some(value) = spend_micro_usd { + self.init_handle(&handles.spend_micro_usd, || { + labels.counter(M_LLM_SPEND_MICRO_USD_TOTAL) + }) + .increment(value); + } + }); + } pub fn record_time_to_first_token(&self, labels: UsageLabels<'_>, ttft: Duration) { if ttft.is_zero() { return; } - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::histogram!( - M_LLM_TTFT, - "endpoint" => labels.endpoint.to_string(), - "inbound_protocol" => labels.inbound_protocol.to_string(), - "provider" => labels.provider.to_string(), - "model" => labels.model.to_string(), - "upstream_model" => labels.upstream_model.to_string(), - "provider_key_id" => labels.provider_key_id.to_string(), - "provider_key_name" => labels.provider_key_name.to_string(), - "api_key_id" => labels.api_key_id.to_string(), - "team_id" => labels.team_id.to_string(), - "user_id" => labels.user_id.to_string(), - "user_name" => labels.user_name.to_string(), - ) - .record(ttft.as_secs_f64()); - }); + self.cached_histogram( + M_LLM_TTFT, + ttft.as_secs_f64(), + |k| { + k.label(labels.endpoint); + k.label(labels.inbound_protocol); + k.label(labels.provider); + k.label(labels.model); + k.label(labels.upstream_model); + k.label(labels.provider_key_id); + k.label(labels.provider_key_name); + k.label(labels.api_key_id); + k.label(labels.team_id); + k.label(labels.user_id); + k.label(labels.user_name); + }, + || { + metrics::histogram!( + M_LLM_TTFT, + "endpoint" => labels.endpoint.to_string(), + "inbound_protocol" => labels.inbound_protocol.to_string(), + "provider" => labels.provider.to_string(), + "model" => labels.model.to_string(), + "upstream_model" => labels.upstream_model.to_string(), + "provider_key_id" => labels.provider_key_id.to_string(), + "provider_key_name" => labels.provider_key_name.to_string(), + "api_key_id" => labels.api_key_id.to_string(), + "team_id" => labels.team_id.to_string(), + "user_id" => labels.user_id.to_string(), + "user_name" => labels.user_name.to_string(), + ) + }, + ); } /// #890 req-4: record token volume for the inbound `client_type` on the @@ -1255,64 +1503,91 @@ impl Metrics { if input_tokens == 0 && output_tokens == 0 && total_tokens == 0 { return; } - metrics::with_local_recorder(&self.inner.recorder, || { - if input_tokens > 0 { - metrics::counter!( - M_LLM_TOKENS_BY_CLIENT_TOTAL, - "client_type" => client_type.to_string(), - "model" => model.to_string(), - "token_type" => "input", - ) - .increment(input_tokens); - } - if output_tokens > 0 { - metrics::counter!( - M_LLM_TOKENS_BY_CLIENT_TOTAL, - "client_type" => client_type.to_string(), - "model" => model.to_string(), - "token_type" => "output", - ) - .increment(output_tokens); + for (token_type, count) in [ + ("input", input_tokens), + ("output", output_tokens), + ("total", total_tokens), + ] { + if count == 0 { + continue; } - if total_tokens > 0 { + self.cached_counter( + M_LLM_TOKENS_BY_CLIENT_TOTAL, + count, + |k| { + k.label(client_type); + k.label(model); + k.label(token_type); + }, + || { + metrics::counter!( + M_LLM_TOKENS_BY_CLIENT_TOTAL, + "client_type" => client_type.to_string(), + "model" => model.to_string(), + "token_type" => token_type, + ) + }, + ); + } + } + + /// Shared cached emit for the deployment counter family. + fn cached_deployment_counter(&self, metric: &'static str, labels: DeploymentLabels<'_>) { + self.cached_counter( + metric, + 1, + |k| { + k.label(labels.provider); + k.label(labels.model); + k.label(labels.upstream_model); + k.label(labels.provider_key_id); + }, + || { metrics::counter!( - M_LLM_TOKENS_BY_CLIENT_TOTAL, - "client_type" => client_type.to_string(), - "model" => model.to_string(), - "token_type" => "total", + metric, + "provider" => labels.provider.to_string(), + "model" => labels.model.to_string(), + "upstream_model" => labels.upstream_model.to_string(), + "provider_key_id" => labels.provider_key_id.to_string(), ) - .increment(total_tokens); - } - }); + }, + ); } pub fn record_deployment_request(&self, labels: DeploymentLabels<'_>, outcome: RequestOutcome) { - metrics::with_local_recorder(&self.inner.recorder, || { - labels.record_counter(M_DEPLOYMENT_REQUESTS_TOTAL); - match outcome { - RequestOutcome::Success => labels.record_counter(M_DEPLOYMENT_SUCCESS_TOTAL), - _ => labels.record_counter(M_DEPLOYMENT_FAILURE_TOTAL), + self.cached_deployment_counter(M_DEPLOYMENT_REQUESTS_TOTAL, labels); + match outcome { + RequestOutcome::Success => { + self.cached_deployment_counter(M_DEPLOYMENT_SUCCESS_TOTAL, labels) } - }); + _ => self.cached_deployment_counter(M_DEPLOYMENT_FAILURE_TOTAL, labels), + } } pub fn set_deployment_state(&self, labels: DeploymentLabels<'_>, state: DeploymentState) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::gauge!( - M_DEPLOYMENT_STATE, - "provider" => labels.provider.to_string(), - "model" => labels.model.to_string(), - "upstream_model" => labels.upstream_model.to_string(), - "provider_key_id" => labels.provider_key_id.to_string(), - ) - .set(state.as_f64()); - }); + self.cached_gauge( + M_DEPLOYMENT_STATE, + state.as_f64(), + |k| { + k.label(labels.provider); + k.label(labels.model); + k.label(labels.upstream_model); + k.label(labels.provider_key_id); + }, + || { + metrics::gauge!( + M_DEPLOYMENT_STATE, + "provider" => labels.provider.to_string(), + "model" => labels.model.to_string(), + "upstream_model" => labels.upstream_model.to_string(), + "provider_key_id" => labels.provider_key_id.to_string(), + ) + }, + ); } pub fn record_deployment_cooldown(&self, labels: DeploymentLabels<'_>) { - metrics::with_local_recorder(&self.inner.recorder, || { - labels.record_counter(M_DEPLOYMENT_COOLED_DOWN_TOTAL); - }); + self.cached_deployment_counter(M_DEPLOYMENT_COOLED_DOWN_TOTAL, labels); } pub fn record_routing_fallback(&self, success: bool, model: &str) { @@ -1321,9 +1596,12 @@ impl Metrics { } else { M_ROUTING_FAILED_FALLBACKS_TOTAL }; - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!(metric, "model" => model.to_string()).increment(1); - }); + self.cached_counter( + metric, + 1, + |k| k.label(model), + || metrics::counter!(metric, "model" => model.to_string()), + ); } pub fn set_rate_limit_remaining( @@ -1333,62 +1611,99 @@ impl Metrics { requests: Option, tokens: Option, ) { - metrics::with_local_recorder(&self.inner.recorder, || { - if let Some(value) = requests { - metrics::gauge!( - M_RATELIMIT_REMAINING_REQUESTS, - "api_key_id" => api_key_id.to_string(), - "model" => model.to_string(), - ) - .set(value as f64); - } - if let Some(value) = tokens { + if let Some(value) = requests { + self.cached_gauge( + M_RATELIMIT_REMAINING_REQUESTS, + value as f64, + |k| { + k.label(api_key_id); + k.label(model); + }, + || { + metrics::gauge!( + M_RATELIMIT_REMAINING_REQUESTS, + "api_key_id" => api_key_id.to_string(), + "model" => model.to_string(), + ) + }, + ); + } + if let Some(value) = tokens { + self.cached_gauge( + M_RATELIMIT_REMAINING_TOKENS, + value as f64, + |k| { + k.label(api_key_id); + k.label(model); + }, + || { + metrics::gauge!( + M_RATELIMIT_REMAINING_TOKENS, + "api_key_id" => api_key_id.to_string(), + "model" => model.to_string(), + ) + }, + ); + } + } + + /// Shared cached emit for the budget gauge family. + fn cached_budget_gauge(&self, metric: &'static str, labels: BudgetLabels<'_>, value: f64) { + self.cached_gauge( + metric, + value, + |k| { + k.label(labels.api_key_id); + k.label(labels.team_id); + k.label(labels.user_id); + }, + || { metrics::gauge!( - M_RATELIMIT_REMAINING_TOKENS, - "api_key_id" => api_key_id.to_string(), - "model" => model.to_string(), + metric, + "api_key_id" => labels.api_key_id.to_string(), + "team_id" => labels.team_id.to_string(), + "user_id" => labels.user_id.to_string(), ) - .set(value as f64); - } - }); + }, + ); } pub fn set_budget_gauges(&self, labels: BudgetLabels<'_>, budget: BudgetGauges) { - metrics::with_local_recorder(&self.inner.recorder, || { - labels.record_gauge(M_BUDGET_DETAILS_PRESENT, 1.0); - if let Some(value) = budget.limit_usd { - labels.record_gauge(M_BUDGET_LIMIT_USD, value); - } - if let Some(value) = budget.spent_usd { - labels.record_gauge(M_BUDGET_SPENT_USD, value); - } - if let Some(value) = budget.remaining_usd { - labels.record_gauge(M_BUDGET_REMAINING_USD, value); - } - if let Some(value) = budget.reset_seconds { - labels.record_gauge(M_BUDGET_RESET_SECONDS, value as f64); - } - }); + self.cached_budget_gauge(M_BUDGET_DETAILS_PRESENT, labels, 1.0); + if let Some(value) = budget.limit_usd { + self.cached_budget_gauge(M_BUDGET_LIMIT_USD, labels, value); + } + if let Some(value) = budget.spent_usd { + self.cached_budget_gauge(M_BUDGET_SPENT_USD, labels, value); + } + if let Some(value) = budget.remaining_usd { + self.cached_budget_gauge(M_BUDGET_REMAINING_USD, labels, value); + } + if let Some(value) = budget.reset_seconds { + self.cached_budget_gauge(M_BUDGET_RESET_SECONDS, labels, value as f64); + } } pub fn clear_budget_gauges(&self, labels: BudgetLabels<'_>) { - metrics::with_local_recorder(&self.inner.recorder, || { - labels.record_gauge(M_BUDGET_DETAILS_PRESENT, 0.0); - }); + self.cached_budget_gauge(M_BUDGET_DETAILS_PRESENT, labels, 0.0); } pub fn record_redis_failure(&self, operation: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!(M_REDIS_FAILURES_TOTAL, "operation" => operation.to_string()) - .increment(1); - }); + self.cached_counter( + M_REDIS_FAILURES_TOTAL, + 1, + |k| k.label(operation), + || metrics::counter!(M_REDIS_FAILURES_TOTAL, "operation" => operation.to_string()), + ); } pub fn record_usage_event_drop(&self, reason: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!(M_USAGE_EVENT_DROPS_TOTAL, "reason" => reason.to_string()) - .increment(1); - }); + self.cached_counter( + M_USAGE_EVENT_DROPS_TOTAL, + 1, + |k| k.label(reason), + || metrics::counter!(M_USAGE_EVENT_DROPS_TOTAL, "reason" => reason.to_string()), + ); } /// Issue #408: bump on every `UsageSink::try_emit` call (the @@ -1410,33 +1725,56 @@ impl Metrics { status_code: u16, inbound_protocol: &'static str, ) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_USAGE_EVENT_EMITS_TOTAL, - "handler" => handler, - "status_code" => status_bucket(status_code), - "inbound_protocol" => inbound_protocol, - ) - .increment(1); - }); + let status_class = status_bucket(status_code); + self.cached_counter( + M_USAGE_EVENT_EMITS_TOTAL, + 1, + |k| { + k.label(handler); + k.label(status_class); + k.label(inbound_protocol); + }, + || { + metrics::counter!( + M_USAGE_EVENT_EMITS_TOTAL, + "handler" => handler, + "status_code" => status_class, + "inbound_protocol" => inbound_protocol, + ) + }, + ); } pub fn record_otlp_fanout_drop(&self, exporter: &str, reason: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!( - M_OTLP_FANOUT_DROPS_TOTAL, - "exporter" => exporter.to_string(), - "reason" => reason.to_string(), - ) - .increment(1); - }); + self.cached_counter( + M_OTLP_FANOUT_DROPS_TOTAL, + 1, + |k| { + k.label(exporter); + k.label(reason); + }, + || { + metrics::counter!( + M_OTLP_FANOUT_DROPS_TOTAL, + "exporter" => exporter.to_string(), + "reason" => reason.to_string(), + ) + }, + ); } pub fn record_otlp_fanout_failure(&self, exporter: &str) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::counter!(M_OTLP_FANOUT_FAILURES_TOTAL, "exporter" => exporter.to_string()) - .increment(1); - }); + self.cached_counter( + M_OTLP_FANOUT_FAILURES_TOTAL, + 1, + |k| k.label(exporter), + || { + metrics::counter!( + M_OTLP_FANOUT_FAILURES_TOTAL, + "exporter" => exporter.to_string(), + ) + }, + ); } /// Observe one request's client-perceived end-to-end latency on @@ -1444,18 +1782,50 @@ impl Metrics { /// at handler return for non-streaming requests and failures, at /// stream completion for committed streams. pub fn record_request_e2e_latency(&self, labels: LatencyLabels<'_>, elapsed: Duration) { - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::histogram!( - M_REQUEST_E2E_LATENCY_SECONDS, - "env_id" => self.inner.env_id.clone(), - "endpoint" => labels.endpoint.to_string(), - "model" => non_empty_or_unknown(labels.model), - "provider" => non_empty_or_unknown(labels.provider), - "status_class" => status_bucket(labels.status), - "streaming" => bool_str(labels.streaming), - ) - .record(elapsed.as_secs_f64()); - }); + self.cached_latency_histogram(M_REQUEST_E2E_LATENCY_SECONDS, labels, elapsed); + } + + /// Shared cached emit for the two SLO latency histograms — identical + /// label shape, `env_id` constant per instance (the cache key is + /// already instance-scoped, so it stays out of the key). + fn cached_latency_histogram( + &self, + metric: &'static str, + labels: LatencyLabels<'_>, + elapsed: Duration, + ) { + let model = if labels.model.is_empty() { + "unknown" + } else { + labels.model + }; + let provider = if labels.provider.is_empty() { + "unknown" + } else { + labels.provider + }; + self.cached_histogram( + metric, + elapsed.as_secs_f64(), + |k| { + k.label(labels.endpoint); + k.label(model); + k.label(provider); + k.label(status_bucket(labels.status)); + k.label_bool(labels.streaming); + }, + || { + metrics::histogram!( + metric, + "env_id" => self.inner.env_id.clone(), + "endpoint" => labels.endpoint.to_string(), + "model" => model.to_string(), + "provider" => provider.to_string(), + "status_class" => status_bucket(labels.status), + "streaming" => bool_str(labels.streaming), + ) + }, + ); } /// Observe a streaming request's time-to-first-token on @@ -1465,18 +1835,7 @@ impl Metrics { if ttft.is_zero() { return; } - metrics::with_local_recorder(&self.inner.recorder, || { - metrics::histogram!( - M_REQUEST_TTFT_SECONDS, - "env_id" => self.inner.env_id.clone(), - "endpoint" => labels.endpoint.to_string(), - "model" => non_empty_or_unknown(labels.model), - "provider" => non_empty_or_unknown(labels.provider), - "status_class" => status_bucket(labels.status), - "streaming" => bool_str(labels.streaming), - ) - .record(ttft.as_secs_f64()); - }); + self.cached_latency_histogram(M_REQUEST_TTFT_SECONDS, labels, ttft); } } @@ -1507,14 +1866,6 @@ pub struct LatencyLabels<'a> { } /// Missing dimensions default to `"unknown"`, never an empty label value. -fn non_empty_or_unknown(v: &str) -> String { - if v.is_empty() { - "unknown".to_string() - } else { - v.to_string() - } -} - /// Bucket an HTTP status code into one of `2xx` / `3xx` / `4xx` / /// `5xx` / `other` (the last covers 1xx and out-of-range). Used by /// the UsageEvent emission counter (#408) to keep prometheus label @@ -1909,19 +2260,6 @@ impl Default for DeploymentLabels<'_> { } } -impl DeploymentLabels<'_> { - fn record_counter(&self, metric: &'static str) { - metrics::counter!( - metric, - "provider" => self.provider.to_string(), - "model" => self.model.to_string(), - "upstream_model" => self.upstream_model.to_string(), - "provider_key_id" => self.provider_key_id.to_string(), - ) - .increment(1); - } -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DeploymentState { Healthy, @@ -1956,18 +2294,6 @@ impl Default for BudgetLabels<'_> { } } -impl BudgetLabels<'_> { - fn record_gauge(&self, metric: &'static str, value: f64) { - metrics::gauge!( - metric, - "api_key_id" => self.api_key_id.to_string(), - "team_id" => self.team_id.to_string(), - "user_id" => self.user_id.to_string(), - ) - .set(value); - } -} - #[derive(Debug, Clone, Copy, Default)] pub struct BudgetGauges { pub limit_usd: Option, @@ -2181,11 +2507,11 @@ mod tests { user_id: "user-1", user_name: "alice", }; - assert_eq!(m.inner.request_series.read().len, 0); + assert_eq!(Metrics::worker_request_series_len(), 0); m.record_proxy_and_llm_request(labels, Duration::from_millis(25)); m.record_proxy_and_llm_request(labels, Duration::from_millis(20)); assert_eq!( - m.inner.request_series.read().len, + Metrics::worker_request_series_len(), 1, "repeated labels must reuse one registered handle set" ); @@ -2243,61 +2569,43 @@ mod tests { #[test] fn request_series_handle_cache_is_bounded() { let metrics = Metrics::new(false); - for index in 0..=REQUEST_SERIES_CACHE_CAPACITY { - let model = format!("model-{index}"); - metrics.record_proxy_and_llm_request( - RequestLabels { - model: &model, - ..RequestLabels::default() - }, - Duration::from_millis(1), - ); + // Two passes over capacity+1 distinct label sets: pass one fills + // the worker cache and forces at least one eviction; pass two + // re-emits everything, so every evicted entry re-registers. + for _ in 0..2 { + for index in 0..=WORKER_CACHE_CAPACITY { + let model = format!("model-{index}"); + metrics.record_proxy_and_llm_request( + RequestLabels { + model: &model, + ..RequestLabels::default() + }, + Duration::from_millis(1), + ); + } } assert_eq!( - metrics.inner.request_series.read().len, - REQUEST_SERIES_CACHE_CAPACITY, + Metrics::worker_request_series_len(), + WORKER_CACHE_CAPACITY, "request series handle cache must stay at its fixed capacity" ); - let evicted_model = (0..=REQUEST_SERIES_CACHE_CAPACITY) - .map(|index| format!("model-{index}")) - .find(|model| { - let labels = RequestLabels { - model, - ..RequestLabels::default() - }; - let hash = Metrics::request_series_hash(labels); - metrics - .inner - .request_series - .read() - .get(hash, labels) - .is_none() - }) - .expect("one series must have been evicted"); - metrics.record_proxy_and_llm_request( - RequestLabels { - model: &evicted_model, - ..RequestLabels::default() - }, - Duration::from_millis(1), - ); - - assert_eq!( - metrics.inner.request_series.read().len, - REQUEST_SERIES_CACHE_CAPACITY, - "re-registering an evicted series must not grow the cache" - ); + // Every label set — evicted or cached — must have continued its + // own Prometheus series: eviction drops our handle, never the + // series or its value. let rendered = metrics.render(); - assert!( - rendered.lines().any(|line| { - line.starts_with(M_PROXY_REQUESTS_TOTAL) - && line.contains(&format!("model=\"{evicted_model}\"")) - && line.ends_with(" 2") - }), - "re-registering an evicted handle must continue the existing Prometheus series" - ); + for index in 0..=WORKER_CACHE_CAPACITY { + let model_label = format!("model=\"model-{index}\""); + assert!( + rendered.lines().any(|line| { + line.starts_with(M_PROXY_REQUESTS_TOTAL) + && line.contains(&model_label) + && line.ends_with(" 2") + }), + "series for model-{index} must show both emits (eviction must not reset it)" + ); + } } #[test] @@ -2323,79 +2631,814 @@ mod tests { } #[test] - fn request_series_cache_distinguishes_hash_collisions() { + fn separator_bearing_label_values_never_alias_cached_series() { let metrics = Metrics::new(false); + // These two label sets join to the SAME byte sequence under the + // cache-key separator; the dirty-key fallback must keep them + // distinct series (uncached, correct) rather than folding them + // into one cache entry. let labels_a = RequestLabels { - model: "collision-a", - outcome: RequestOutcome::Success, + model: "x\u{1f}y", + upstream_model: "z", ..RequestLabels::default() }; let labels_b = RequestLabels { - model: "collision-b", - outcome: RequestOutcome::Success, + model: "x", + upstream_model: "y\u{1f}z", ..RequestLabels::default() }; - let handles_a = metrics.register_request_series(labels_a); - let handles_b = metrics.register_request_series(labels_b); - let forced_hash = 7; - { - let mut cache = metrics.inner.request_series.write(); - cache.insert(forced_hash, RequestLabelsOwned::new(labels_a), handles_a); - cache.insert(forced_hash, RequestLabelsOwned::new(labels_b), handles_b); - cache - .get(forced_hash, labels_a) - .expect("first colliding label set must remain addressable") - .proxy_requests - .increment(1); - cache - .get(forced_hash, labels_b) - .expect("second colliding label set must remain addressable") - .proxy_requests - .increment(2); - } + metrics.record_proxy_and_llm_request(labels_a, Duration::from_millis(1)); + metrics.record_proxy_and_llm_request(labels_b, Duration::from_millis(1)); + metrics.record_proxy_and_llm_request(labels_b, Duration::from_millis(1)); + assert_eq!( + Metrics::worker_request_series_len(), + 0, + "a separator-bearing label value must never be cached" + ); let rendered = metrics.render(); - assert!(rendered.lines().any(|line| { - line.starts_with(M_PROXY_REQUESTS_TOTAL) - && line.contains("model=\"collision-a\"") - && line.ends_with(" 1") - })); - assert!(rendered.lines().any(|line| { - line.starts_with(M_PROXY_REQUESTS_TOTAL) - && line.contains("model=\"collision-b\"") - && line.ends_with(" 2") - })); + let series: Vec<&str> = rendered + .lines() + .filter(|line| line.starts_with(M_PROXY_REQUESTS_TOTAL)) + .collect(); + assert_eq!( + series.len(), + 2, + "aliasing would fold the two label sets into one series: {series:?}" + ); + assert!(series.iter().any(|line| line.ends_with(" 1"))); + assert!(series.iter().any(|line| line.ends_with(" 2"))); } #[test] - fn concurrent_request_series_misses_register_once_and_record_every_call() { - const THREADS: usize = 16; + fn two_instances_on_one_thread_never_share_cached_handles() { + let a = Metrics::new(false); + let b = Metrics::new(false); + a.record_tokens("openai", "shared-model", 1); + b.record_tokens("openai", "shared-model", 2); + + // If the worker cache ignored the instance id, `b`'s emit would + // land on `a`'s handle: `a` would render 3 and `b` nothing. + assert!(a + .render() + .lines() + .any(|line| line.starts_with(M_TOKENS_CONSUMED) && line.ends_with(" 1"))); + assert!(b + .render() + .lines() + .any(|line| line.starts_with(M_TOKENS_CONSUMED) && line.ends_with(" 2"))); + } + /// Pins the deployment counter family against a cache-key/label + /// drift: the key builder and the register closure in + /// `cached_deployment_counter` list the labels independently, and + /// dropping one from the KEY would silently alias two deployments + /// onto one series while every single-label-set test stayed green. + #[test] + fn deployment_counters_stay_distinct_per_label_set() { let metrics = Metrics::new(false); - let barrier = Arc::new(std::sync::Barrier::new(THREADS)); - let threads = (0..THREADS) - .map(|_| { - let metrics = metrics.clone(); - let barrier = Arc::clone(&barrier); - std::thread::spawn(move || { - barrier.wait(); - metrics.record_proxy_and_llm_request( - RequestLabels::default(), - Duration::from_millis(1), - ); + let dep_a = DeploymentLabels { + provider: "openai", + model: "gpt", + upstream_model: "gpt-4o", + provider_key_id: "pk-a", + }; + // Differs ONLY in provider_key_id — the label a key-builder + // regression is most likely to drop. + let dep_b = DeploymentLabels { + provider_key_id: "pk-b", + ..dep_a + }; + metrics.record_deployment_request(dep_a, RequestOutcome::Success); + metrics.record_deployment_request(dep_a, RequestOutcome::Success); + metrics.record_deployment_request(dep_b, RequestOutcome::UpstreamError); + + let rendered = metrics.render(); + let series = |metric: &str, key: &str| { + rendered + .lines() + .find(|l| { + l.starts_with(metric) && l.contains(&format!("provider_key_id=\"{key}\"")) }) - }) - .collect::>(); - for thread in threads { - thread.join().expect("metric recording thread panicked"); + .map(str::to_owned) + }; + let requests_a = series(M_DEPLOYMENT_REQUESTS_TOTAL, "pk-a") + .expect("deployment A must have its own requests series"); + assert!(requests_a.ends_with(" 2"), "got: {requests_a}"); + for label in [ + "provider=\"openai\"", + "model=\"gpt\"", + "upstream_model=\"gpt-4o\"", + ] { + assert!(requests_a.contains(label), "missing {label}: {requests_a}"); } + let requests_b = series(M_DEPLOYMENT_REQUESTS_TOTAL, "pk-b") + .expect("deployment B must have its own requests series"); + assert!(requests_b.ends_with(" 1"), "got: {requests_b}"); + assert!(series(M_DEPLOYMENT_SUCCESS_TOTAL, "pk-a").is_some_and(|l| l.ends_with(" 2"))); + assert!(series(M_DEPLOYMENT_FAILURE_TOTAL, "pk-b").is_some_and(|l| l.ends_with(" 1"))); + // The outcome split must not cross-pollinate. + assert!(series(M_DEPLOYMENT_SUCCESS_TOTAL, "pk-b").is_none()); + assert!(series(M_DEPLOYMENT_FAILURE_TOTAL, "pk-a").is_none()); + } - assert_eq!( - metrics.inner.request_series.read().len, - 1, - "concurrent first misses must converge on one cached handle set" + /// Pins the budget gauge family's values, field gating and clear + /// semantics. (Key-drift protection is NOT this test's job — its two + /// label sets differ in every dimension, so dropping one key label + /// still yields distinct keys; `budget_gauge_key_covers_every_label` + /// below is the drift guard.) + #[test] + fn budget_gauges_stay_distinct_per_label_set_and_clear() { + let metrics = Metrics::new(false); + let key_a = BudgetLabels { + api_key_id: "ak-a", + team_id: "team-a", + user_id: "user-a", + }; + let key_b = BudgetLabels { + api_key_id: "ak-b", + team_id: "team-b", + user_id: "user-b", + }; + metrics.set_budget_gauges( + key_a, + BudgetGauges { + limit_usd: Some(100.0), + spent_usd: Some(40.0), + remaining_usd: Some(60.0), + reset_seconds: Some(3600), + }, ); - let rendered = metrics.render(); + metrics.set_budget_gauges( + key_b, + BudgetGauges { + spent_usd: Some(7.0), + ..BudgetGauges::default() + }, + ); + + let rendered = metrics.render(); + let series = |metric: &str, team: &str| { + rendered + .lines() + .find(|l| l.starts_with(metric) && l.contains(&format!("team_id=\"{team}\""))) + .map(str::to_owned) + }; + for (metric, value) in [ + (M_BUDGET_DETAILS_PRESENT, "1"), + (M_BUDGET_LIMIT_USD, "100"), + (M_BUDGET_SPENT_USD, "40"), + (M_BUDGET_REMAINING_USD, "60"), + (M_BUDGET_RESET_SECONDS, "3600"), + ] { + let line = series(metric, "team-a") + .unwrap_or_else(|| panic!("{metric} missing for team-a:\n{rendered}")); + assert!(line.ends_with(&format!(" {value}")), "got: {line}"); + assert!(line.contains("api_key_id=\"ak-a\"") && line.contains("user_id=\"user-a\"")); + } + // Key B must have its OWN spent series, not overwrite A's. + assert!(series(M_BUDGET_SPENT_USD, "team-b").is_some_and(|l| l.ends_with(" 7"))); + // Unset dimensions register nothing for B. + assert!(series(M_BUDGET_LIMIT_USD, "team-b").is_none()); + + metrics.clear_budget_gauges(key_a); + let rendered = metrics.render(); + let details = |team: &str| { + rendered + .lines() + .find(|l| { + l.starts_with(M_BUDGET_DETAILS_PRESENT) + && l.contains(&format!("team_id=\"{team}\"")) + }) + .map(str::to_owned) + }; + assert!(details("team-a").is_some_and(|l| l.ends_with(" 0"))); + assert!(details("team-b").is_some_and(|l| l.ends_with(" 1"))); + } + + /// Pins the legacy spec-7 pair's full label sets, accumulation, and + /// the outcome-stays-off-durations invariant. (Its two label sets + /// differ in several dimensions at once, so this test cannot catch a + /// single dropped key label; `legacy_request_key_covers_every_label` + /// below is the drift guard.) + #[test] + fn legacy_request_series_stay_fully_labelled_and_distinct() { + let metrics = Metrics::new(false); + metrics.record_request( + "openai", + "my-gpt4", + 200, + RequestOutcome::Success, + Duration::from_millis(120), + ); + metrics.record_request( + "openai", + "my-gpt4", + 200, + RequestOutcome::Success, + Duration::from_millis(80), + ); + metrics.record_request( + "openai", + "other-model", + 404, + RequestOutcome::ClientError, + Duration::from_millis(5), + ); + + let rendered = metrics.render(); + let counter_a = rendered + .lines() + .find(|l| l.starts_with(M_REQUESTS_TOTAL) && l.contains("status=\"200\"")) + .expect("200 series must render"); + assert!( + counter_a.ends_with(" 2"), + "cached handle must accumulate: {counter_a}" + ); + for label in [ + "provider=\"openai\"", + "model=\"my-gpt4\"", + "outcome=\"success\"", + ] { + assert!(counter_a.contains(label), "missing {label}: {counter_a}"); + } + let counter_b = rendered + .lines() + .find(|l| l.starts_with(M_REQUESTS_TOTAL) && l.contains("status=\"404\"")) + .expect("404 series must render distinctly"); + assert!(counter_b.ends_with(" 1"), "got: {counter_b}"); + assert!(counter_b.contains("model=\"other-model\"")); + + // Duration summary: carries provider/model/status, never outcome, + // and counts per (model, status) series independently. + let dur_count_a = rendered + .lines() + .find(|l| { + l.starts_with(&format!("{M_REQUEST_DURATION}_count")) + && l.contains("model=\"my-gpt4\"") + }) + .expect("duration count for my-gpt4 must render"); + assert!(dur_count_a.ends_with(" 2") && dur_count_a.contains("status=\"200\"")); + let dur_count_b = rendered + .lines() + .find(|l| { + l.starts_with(&format!("{M_REQUEST_DURATION}_count")) + && l.contains("model=\"other-model\"") + }) + .expect("duration count for other-model must render"); + assert!(dur_count_b.ends_with(" 1") && dur_count_b.contains("status=\"404\"")); + for line in rendered.lines() { + if line.starts_with(M_REQUEST_DURATION) { + assert!( + !line.contains("outcome="), + "outcome must stay off durations: {line}" + ); + } + } + } + + // ── Vary-one-label key-drift guards ──────────────────────────────── + // + // The worker cache duplicates each family's label list in two places: + // the key builder and the register closure. A label present in the + // register closure but DROPPED from the key builder aliases every + // pair of label sets that differ only in that label — silently, with + // correct-looking output for single-label-set traffic. These tests + // emit a base label set twice plus one variant per label differing + // ONLY in that label, then assert one rendered series per label set: + // any single dropped key label folds its variant into the base series + // and fails both the count and the base-total assertion. + // (An independent audit demonstrated by mutation that multi-dimension + // pairs do NOT catch single-label drops; each variant here differs in + // exactly one.) + + /// Rendered value lines for one exact metric name (brace-delimited, + /// so `aisix_x` never matches `aisix_x_count` and vice versa). + fn series_lines<'r>(rendered: &'r str, metric: &str) -> Vec<&'r str> { + rendered + .lines() + .filter(|l| l.starts_with(metric) && l.as_bytes().get(metric.len()) == Some(&b'{')) + .collect() + } + + #[track_caller] + fn assert_one_series_per_label_set(rendered: &str, metric: &str, expected: usize) { + let series = series_lines(rendered, metric); + assert_eq!( + series.len(), + expected, + "{metric}: a dropped key label folds a variant into the base series\n{rendered}" + ); + assert_eq!( + series.iter().filter(|l| l.ends_with(" 2")).count(), + 1, + "{metric}: exactly the base series must show both base emits\n{rendered}" + ); + } + + #[test] + fn request_series_key_covers_every_label() { + let base = RequestLabels::default(); + let variants = [ + RequestLabels { + endpoint: "/v1/messages", + ..base + }, + RequestLabels { + inbound_protocol: "anthropic", + ..base + }, + RequestLabels { + provider: "p2", + ..base + }, + RequestLabels { + model: "m2", + ..base + }, + RequestLabels { + upstream_model: "um2", + ..base + }, + RequestLabels { + provider_key_id: "pk2", + ..base + }, + RequestLabels { + provider_key_name: "pkn2", + ..base + }, + RequestLabels { + api_key_id: "ak2", + ..base + }, + RequestLabels { + team_id: "t2", + ..base + }, + RequestLabels { + user_id: "u2", + ..base + }, + RequestLabels { + user_name: "n2", + ..base + }, + RequestLabels { + stream: true, + ..base + }, + RequestLabels { + is_fallback: true, + ..base + }, + RequestLabels { + status: 201, + ..base + }, + RequestLabels { + outcome: RequestOutcome::ClientError, + ..base + }, + ]; + let m = Metrics::new(false); + m.record_proxy_and_llm_request(base, Duration::from_millis(1)); + m.record_proxy_and_llm_request(base, Duration::from_millis(1)); + for v in &variants { + m.record_proxy_and_llm_request(*v, Duration::from_millis(1)); + } + assert_one_series_per_label_set(&m.render(), M_PROXY_REQUESTS_TOTAL, 1 + variants.len()); + } + + #[test] + fn usage_series_key_covers_every_label() { + let base = UsageLabels::default(); + let variants = [ + UsageLabels { + endpoint: "/v1/messages", + ..base + }, + UsageLabels { + inbound_protocol: "anthropic", + ..base + }, + UsageLabels { + provider: "p2", + ..base + }, + UsageLabels { + model: "m2", + ..base + }, + UsageLabels { + upstream_model: "um2", + ..base + }, + UsageLabels { + provider_key_id: "pk2", + ..base + }, + UsageLabels { + provider_key_name: "pkn2", + ..base + }, + UsageLabels { + api_key_id: "ak2", + ..base + }, + UsageLabels { + team_id: "t2", + ..base + }, + UsageLabels { + user_id: "u2", + ..base + }, + UsageLabels { + user_name: "n2", + ..base + }, + ]; + let m = Metrics::new(false); + let one_token = LlmUsage { + input_tokens: 1, + ..LlmUsage::default() + }; + m.record_llm_usage(base, one_token); + m.record_llm_usage(base, one_token); + for v in &variants { + m.record_llm_usage(*v, one_token); + } + assert_one_series_per_label_set(&m.render(), M_LLM_INPUT_TOKENS_TOTAL, 1 + variants.len()); + } + + #[test] + fn legacy_request_key_covers_every_label() { + let m = Metrics::new(false); + let emit = |provider, model, status, outcome| { + m.record_request(provider, model, status, outcome, Duration::from_millis(1)); + }; + emit("openai", "m", 200, RequestOutcome::Success); + emit("openai", "m", 200, RequestOutcome::Success); + emit("p2", "m", 200, RequestOutcome::Success); + emit("openai", "m2", 200, RequestOutcome::Success); + emit("openai", "m", 201, RequestOutcome::Success); + emit("openai", "m", 200, RequestOutcome::ClientError); + + let rendered = m.render(); + assert_one_series_per_label_set(&rendered, M_REQUESTS_TOTAL, 5); + // The duration histogram keys on (provider, model, status) only: + // the outcome variant lands on the base duration series, so the + // base count is 3 across 4 series. A status/model/provider drop + // from the HISTOGRAM key collapses its series count below 4. + let dur = series_lines(&rendered, &format!("{M_REQUEST_DURATION}_count")); + assert_eq!(dur.len(), 4, "{rendered}"); + assert_eq!( + dur.iter().filter(|l| l.ends_with(" 3")).count(), + 1, + "{rendered}" + ); + } + + #[test] + fn budget_gauge_key_covers_every_label() { + let m = Metrics::new(false); + let base = BudgetLabels { + api_key_id: "ak", + team_id: "t", + user_id: "u", + }; + let variants = [ + BudgetLabels { + api_key_id: "ak2", + ..base + }, + BudgetLabels { + team_id: "t2", + ..base + }, + BudgetLabels { + user_id: "u2", + ..base + }, + ]; + let spend = |v| BudgetGauges { + spent_usd: Some(v), + ..BudgetGauges::default() + }; + m.set_budget_gauges(base, spend(1.0)); + for (i, v) in variants.iter().enumerate() { + m.set_budget_gauges(*v, spend(2.0 + i as f64)); + } + let rendered = m.render(); + let series = series_lines(&rendered, M_BUDGET_SPENT_USD); + assert_eq!(series.len(), 1 + variants.len(), "{rendered}"); + // Gauges overwrite rather than sum: a dropped key label makes the + // LAST variant's value land on the base series instead. + assert!( + series.iter().any(|l| l.contains("api_key_id=\"ak\"") + && l.contains("team_id=\"t\"") + && l.ends_with(" 1")), + "{rendered}" + ); + } + + #[test] + fn llm_ttft_key_covers_every_label() { + let base = UsageLabels::default(); + let variants = [ + UsageLabels { + endpoint: "/v1/messages", + ..base + }, + UsageLabels { + inbound_protocol: "anthropic", + ..base + }, + UsageLabels { + provider: "p2", + ..base + }, + UsageLabels { + model: "m2", + ..base + }, + UsageLabels { + upstream_model: "um2", + ..base + }, + UsageLabels { + provider_key_id: "pk2", + ..base + }, + UsageLabels { + provider_key_name: "pkn2", + ..base + }, + UsageLabels { + api_key_id: "ak2", + ..base + }, + UsageLabels { + team_id: "t2", + ..base + }, + UsageLabels { + user_id: "u2", + ..base + }, + UsageLabels { + user_name: "n2", + ..base + }, + ]; + let m = Metrics::new(false); + m.record_time_to_first_token(base, Duration::from_millis(1)); + m.record_time_to_first_token(base, Duration::from_millis(1)); + for v in &variants { + m.record_time_to_first_token(*v, Duration::from_millis(1)); + } + assert_one_series_per_label_set( + &m.render(), + &format!("{M_LLM_TTFT}_count"), + 1 + variants.len(), + ); + } + + #[test] + fn latency_histogram_key_covers_every_label() { + let base = LatencyLabels { + endpoint: "/v1/chat/completions", + model: "m", + provider: "p", + status: 200, + streaming: false, + }; + let variants = [ + LatencyLabels { + endpoint: "/v1/messages", + ..base + }, + LatencyLabels { + model: "m2", + ..base + }, + LatencyLabels { + provider: "p2", + ..base + }, + // The key uses the status CLASS, so the variant must change + // the bucket, not just the code. + LatencyLabels { + status: 404, + ..base + }, + LatencyLabels { + streaming: true, + ..base + }, + ]; + let m = Metrics::new(false); + m.record_request_e2e_latency(base, Duration::from_millis(1)); + m.record_request_e2e_latency(base, Duration::from_millis(1)); + for v in &variants { + m.record_request_e2e_latency(*v, Duration::from_millis(1)); + } + assert_one_series_per_label_set( + &m.render(), + &format!("{M_REQUEST_E2E_LATENCY_SECONDS}_count"), + 1 + variants.len(), + ); + } + + #[test] + fn auth_decision_key_covers_every_label() { + let m = Metrics::new(false); + m.record_auth_decision("api_key", true, ""); + m.record_auth_decision("api_key", true, ""); + m.record_auth_decision("jwt", true, ""); + m.record_auth_decision("api_key", false, ""); + m.record_auth_decision("api_key", true, "key_expired"); + assert_one_series_per_label_set(&m.render(), M_AUTH_DECISIONS_TOTAL, 4); + } + + #[test] + fn guardrail_latency_key_covers_every_label() { + let base = aisix_core::GuardrailExecution { + guardrail_name: "g", + kind: "keyword", + phase: "input", + result: "allowed", + error_type: None, + elapsed: Duration::from_millis(1), + }; + let variants = [ + aisix_core::GuardrailExecution { + guardrail_name: "g2", + ..base + }, + aisix_core::GuardrailExecution { + kind: "pii", + ..base + }, + aisix_core::GuardrailExecution { + phase: "output", + ..base + }, + aisix_core::GuardrailExecution { + result: "blocked", + ..base + }, + aisix_core::GuardrailExecution { + error_type: Some("timeout"), + ..base + }, + ]; + let m = Metrics::new(false); + m.record_guardrail_execution(&base); + m.record_guardrail_execution(&base); + for v in &variants { + m.record_guardrail_execution(v); + } + assert_one_series_per_label_set( + &m.render(), + &format!("{M_GUARDRAIL_LATENCY_SECONDS}_count"), + 1 + variants.len(), + ); + } + + #[test] + fn ratelimit_rejection_key_covers_every_label() { + let m = Metrics::new(false); + m.record_ratelimit_rejection("requests", "api_key", None); + m.record_ratelimit_rejection("requests", "api_key", None); + m.record_ratelimit_rejection("tokens", "api_key", None); + m.record_ratelimit_rejection("requests", "model", None); + m.record_ratelimit_rejection("requests", "api_key", Some("p1")); + assert_one_series_per_label_set(&m.render(), M_RATELIMIT_REJECTIONS, 4); + } + + #[test] + fn tokens_by_client_key_covers_every_label() { + let m = Metrics::new(false); + m.record_llm_tokens_by_client("cli", "m", 1, 0, 0); + m.record_llm_tokens_by_client("cli", "m", 1, 0, 0); + m.record_llm_tokens_by_client("cli2", "m", 1, 0, 0); + m.record_llm_tokens_by_client("cli", "m2", 1, 0, 0); + assert_one_series_per_label_set(&m.render(), M_LLM_TOKENS_BY_CLIENT_TOTAL, 3); + } + + #[test] + fn consumed_tokens_key_covers_every_label() { + let m = Metrics::new(false); + m.record_tokens("p", "m", 1); + m.record_tokens("p", "m", 1); + m.record_tokens("p2", "m", 1); + m.record_tokens("p", "m2", 1); + assert_one_series_per_label_set(&m.render(), M_TOKENS_CONSUMED, 3); + } + + #[test] + fn usage_event_emit_key_covers_every_label() { + let m = Metrics::new(false); + m.record_usage_event_emit("chat", 200, "openai"); + m.record_usage_event_emit("chat", 200, "openai"); + m.record_usage_event_emit("embeddings", 200, "openai"); + m.record_usage_event_emit("chat", 404, "openai"); + m.record_usage_event_emit("chat", 200, "anthropic"); + assert_one_series_per_label_set(&m.render(), M_USAGE_EVENT_EMITS_TOTAL, 4); + } + + #[test] + fn in_flight_gauge_key_covers_every_label() { + let m = Metrics::new(false); + m.increment_proxy_in_flight("/v1/chat/completions", "openai"); + m.increment_proxy_in_flight("/v1/messages", "openai"); + m.increment_proxy_in_flight("/v1/chat/completions", "anthropic"); + let rendered = m.render(); + assert_eq!( + series_lines(&rendered, M_PROXY_IN_FLIGHT).len(), + 3, + "a dropped key label folds distinct endpoints/protocols into one gauge\n{rendered}" + ); + } + + #[test] + fn in_flight_gauge_clamps_an_unmatched_decrement_at_zero() { + let m = Metrics::new(false); + m.decrement_proxy_in_flight("/v1/chat/completions", "openai"); + let rendered = m.render(); + let line = rendered + .lines() + .find(|l| l.starts_with(M_PROXY_IN_FLIGHT)) + .expect("in-flight gauge must render"); + assert!( + line.ends_with(" 0"), + "unmatched decrement must clamp: {line}" + ); + } + + /// Pins the slot predicate on BOTH fields: a regression comparing only + /// `endpoint` would route the anthropic edges through the openai slot, + /// corrupting VALUES while the series count (guarded above) stays 3. + #[test] + fn in_flight_slots_stay_distinct_per_endpoint_and_protocol() { + let m = Metrics::new(false); + m.increment_proxy_in_flight("/v1/chat/completions", "openai"); + m.increment_proxy_in_flight("/v1/chat/completions", "anthropic"); + m.increment_proxy_in_flight("/v1/messages", "anthropic"); + m.decrement_proxy_in_flight("/v1/chat/completions", "anthropic"); + + let rendered = m.render(); + let value = |endpoint: &str, protocol: &str| { + rendered + .lines() + .find(|l| { + l.starts_with(M_PROXY_IN_FLIGHT) + && l.contains(&format!("endpoint=\"{endpoint}\"")) + && l.contains(&format!("inbound_protocol=\"{protocol}\"")) + }) + .and_then(|l| l.rsplit(' ').next()) + .map(str::to_owned) + }; + assert_eq!( + value("/v1/chat/completions", "openai").as_deref(), + Some("1") + ); + assert_eq!( + value("/v1/chat/completions", "anthropic").as_deref(), + Some("0") + ); + assert_eq!(value("/v1/messages", "anthropic").as_deref(), Some("1")); + } + + #[test] + fn concurrent_request_series_misses_register_once_and_record_every_call() { + const THREADS: usize = 16; + + let metrics = Metrics::new(false); + let barrier = Arc::new(std::sync::Barrier::new(THREADS)); + let threads = (0..THREADS) + .map(|_| { + let metrics = metrics.clone(); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + barrier.wait(); + metrics.record_proxy_and_llm_request( + RequestLabels::default(), + Duration::from_millis(1), + ); + }) + }) + .collect::>(); + for thread in threads { + thread.join().expect("metric recording thread panicked"); + } + + // Each thread registers into its own worker cache, but every + // registration for the same labels resolves to the SAME registry + // series — the sums below are the contract. + let rendered = metrics.render(); for metric in [ M_PROXY_REQUESTS_TOTAL, M_LLM_REQUESTS_TOTAL, @@ -2435,7 +3478,7 @@ mod tests { }, ); assert_eq!( - metrics.inner.usage_series.read().len, + Metrics::worker_usage_series_len(), 1, "the first non-zero usage sample must create one cached label set" ); @@ -2466,7 +3509,7 @@ mod tests { }, ); assert_eq!( - metrics.inner.usage_series.read().len, + Metrics::worker_usage_series_len(), 1, "repeated labels must reuse the cached usage handles" ); @@ -2516,11 +3559,7 @@ mod tests { thread.join().expect("metric recording thread panicked"); } - assert_eq!( - metrics.inner.usage_series.read().len, - 1, - "concurrent first misses must converge on one cached usage handle set" - ); + // Per-thread caches; the shared-series sums below are the contract. let rendered = metrics.render(); for (metric, value) in [ (M_LLM_INPUT_TOKENS_TOTAL, THREADS), @@ -2540,7 +3579,7 @@ mod tests { #[test] fn usage_series_handle_cache_is_bounded() { let metrics = Metrics::new(false); - for index in 0..=USAGE_SERIES_CACHE_CAPACITY { + for index in 0..=WORKER_CACHE_CAPACITY { let model = format!("usage-model-{index}"); metrics.record_llm_usage( UsageLabels { @@ -2555,44 +3594,12 @@ mod tests { } assert_eq!( - metrics.inner.usage_series.read().len, - USAGE_SERIES_CACHE_CAPACITY, + Metrics::worker_usage_series_len(), + WORKER_CACHE_CAPACITY, "usage series handle cache must stay at its fixed capacity" ); } - #[test] - fn usage_series_cache_distinguishes_hash_collisions() { - let labels_a = UsageLabels { - model: "usage-collision-a", - ..UsageLabels::default() - }; - let labels_b = UsageLabels { - model: "usage-collision-b", - ..UsageLabels::default() - }; - let forced_hash = 7; - let mut cache = UsageSeriesCache::default(); - cache.insert( - forced_hash, - UsageLabelsOwned::new(labels_a), - UsageSeriesHandles::default(), - ); - cache.insert( - forced_hash, - UsageLabelsOwned::new(labels_b), - UsageSeriesHandles::default(), - ); - - assert!(cache.get(forced_hash, labels_a).is_some()); - assert!(cache.get(forced_hash, labels_b).is_some()); - assert_eq!( - cache.entries[&forced_hash].len(), - 2, - "colliding hashes must retain both exact label sets" - ); - } - #[test] fn individual_request_recorders_do_not_create_unrelated_metric_families() { let proxy_only = Metrics::new(false);