From 369317c11a9370f8e9418e984ee16e4706e0f519 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 10 Aug 2026 16:09:02 -0700 Subject: [PATCH 1/9] feat(server): expose stage router stats Signed-off-by: nachiketb --- crates/switchyard-server/README.md | 6 +- crates/switchyard-server/src/lib.rs | 17 +- crates/switchyard-server/src/metrics.rs | 27 ++- crates/switchyard-server/src/stats.rs | 2 + .../src/stats/accumulator.rs | 3 + .../switchyard-server/src/stats/algorithm.rs | 168 +++++++++++++++ .../src/stats/algorithm/stage_router.rs | 193 ++++++++++++++++++ crates/switchyard-server/tests/server.rs | 12 +- docs/known_issues.md | 2 +- .../stage_router_routing.md | 4 +- 10 files changed, 420 insertions(+), 14 deletions(-) create mode 100644 crates/switchyard-server/src/stats/algorithm.rs create mode 100644 crates/switchyard-server/src/stats/algorithm/stage_router.rs diff --git a/crates/switchyard-server/README.md b/crates/switchyard-server/README.md index b5960eb71..dd57c57d7 100644 --- a/crates/switchyard-server/README.md +++ b/crates/switchyard-server/README.md @@ -122,7 +122,7 @@ documented in [Stage-Router Routing](../../docs/routing_algorithms/stage_router_ | `POST` | `/v1/responses` | OpenAI Responses | | `POST` | `/v1/messages/count_tokens` | Token count from a route's Anthropic target | | `GET` | `/v1/models` | Routes served by this deployment | -| `GET` | `/v1/stats` | Per-model request, token, and cost totals | +| `GET` | `/v1/stats` | Per-model usage plus curated algorithm stats | | `POST` | `/v1/stats/reset` | Clear accumulated stats | | `GET` | `/metrics` | Prometheus text, see [Metrics](#metrics) | | `GET` | `/health` | Liveness | @@ -131,6 +131,10 @@ Requests name a route by its `id`, so `POST /v1/chat/completions` with `"model": routes through the `[routes.general]` entry above. Any of the three request formats can address any route, and the server translates between them. +For `stage_router`, `algorithm_stats.stage_router` groups routing decisions by source and semantic +target and summarizes its score, confidence, and input-dimension histograms. These values reset +with `/v1/stats/reset`; the process-lifetime counters on `/metrics` remain cumulative. + Token counting selects an Anthropic-format completion target, preferring target names or model IDs containing `opus`, `sonnet`, then `haiku`. Other ties preserve the route's target order. diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 1f69e0b02..6e2d35a67 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -45,7 +45,9 @@ use tracing::{Instrument, Level}; use switchyard_translation::{WireFormat, decode_request}; use crate::response::into_http_response; -use crate::stats::{StatsAccumulator, StatsSnapshot, prefix_probe, tracking_enabled_from_env}; +use crate::stats::{ + AlgorithmStats, StatsAccumulator, StatsSnapshot, prefix_probe, tracking_enabled_from_env, +}; pub use observability::{flush_observability, initialize_observability}; @@ -133,6 +135,7 @@ pub struct ServerState { routes: Arc>, metrics: prometheus::Registry, stats: StatsAccumulator, + algorithm_stats: AlgorithmStats, routing_log: Option, track_cache_eligibility: bool, } @@ -219,10 +222,17 @@ impl ServerState { return Err(ServerError::new("at least one algorithm route is required")); } let metrics = metrics::registry().map_err(ServerError::new)?; + let algorithm_stats = AlgorithmStats::new( + metrics.clone(), + entries + .values() + .map(|entry| entry.algorithm.name().to_string()), + ); Ok(Self { routes: Arc::new(entries), metrics, stats: StatsAccumulator::default(), + algorithm_stats, routing_log: None, track_cache_eligibility: tracking_enabled_from_env(), }) @@ -1016,11 +1026,14 @@ async fn models(State(state): State) -> Json { } async fn get_stats(State(state): State) -> Json { - Json(state.stats.snapshot()) + let mut snapshot = state.stats.snapshot(); + snapshot.algorithm_stats = state.algorithm_stats.snapshot(); + Json(snapshot) } async fn reset_stats(State(state): State) -> Json { state.stats.reset(); + state.algorithm_stats.reset(); Json(json!({"status": "reset"})) } diff --git a/crates/switchyard-server/src/metrics.rs b/crates/switchyard-server/src/metrics.rs index 2469e0251..27a5ee12e 100644 --- a/crates/switchyard-server/src/metrics.rs +++ b/crates/switchyard-server/src/metrics.rs @@ -18,6 +18,17 @@ pub(crate) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8" const ROUTING_OVERHEAD_BUCKETS_MS: &[f64] = &[ 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, 2500.0, 5000.0, ]; +const STAGE_ROUTER_SCORE_BUCKETS: &[f64] = &[-1.0, -0.75, -0.5, -0.25, 0.0, 0.25, 0.5, 0.75, 1.0]; +const STAGE_ROUTER_UNIT_BUCKETS: &[f64] = &[0.0, 0.1, 0.25, 0.5, 0.75, 0.9, 1.0]; + +const STAGE_ROUTER_SCORE_METRIC: &str = "switchyard.stage_router.score"; +const STAGE_ROUTER_UNIT_METRICS: &[&str] = &[ + "switchyard.stage_router.confidence", + "switchyard.stage_router.severity", + "switchyard.stage_router.spinning", + "switchyard.stage_router.exploring", + "switchyard.stage_router.production_intensity", +]; struct Metrics { registry: Registry, @@ -42,7 +53,7 @@ fn initialize() -> Result { .map_err(|error| format!("failed to initialize Prometheus metrics: {error}"))?; let mut builder = SdkMeterProvider::builder() .with_reader(exporter) - .with_view(routing_overhead_buckets) + .with_view(histogram_buckets) .with_resource(crate::observability::resource()); if crate::observability::otlp_enabled("METRICS") { let exporter = opentelemetry_otlp::MetricExporter::builder() @@ -70,13 +81,19 @@ pub(crate) fn flush() { } } -fn routing_overhead_buckets(instrument: &Instrument) -> Option { - if instrument.name() != "switchyard.routing_overhead_ms" { +fn histogram_buckets(instrument: &Instrument) -> Option { + let boundaries = if instrument.name() == "switchyard.routing_overhead_ms" { + ROUTING_OVERHEAD_BUCKETS_MS + } else if instrument.name() == STAGE_ROUTER_SCORE_METRIC { + STAGE_ROUTER_SCORE_BUCKETS + } else if STAGE_ROUTER_UNIT_METRICS.contains(&instrument.name()) { + STAGE_ROUTER_UNIT_BUCKETS + } else { return None; - } + }; Stream::builder() .with_aggregation(Aggregation::ExplicitBucketHistogram { - boundaries: ROUTING_OVERHEAD_BUCKETS_MS.to_vec(), + boundaries: boundaries.to_vec(), // Cumulative min/max cover the whole process, so they aren't useful. record_min_max: false, }) diff --git a/crates/switchyard-server/src/stats.rs b/crates/switchyard-server/src/stats.rs index ebb644578..74f9dda51 100644 --- a/crates/switchyard-server/src/stats.rs +++ b/crates/switchyard-server/src/stats.rs @@ -4,7 +4,9 @@ //! Process-local JSON stats accounting for the Rust server. mod accumulator; +mod algorithm; mod cache_eligibility; pub(crate) use accumulator::{StatsAccumulator, StatsSnapshot, TokenUsage}; +pub(crate) use algorithm::AlgorithmStats; pub(crate) use cache_eligibility::{prefix_probe, tracking_enabled_from_env}; diff --git a/crates/switchyard-server/src/stats/accumulator.rs b/crates/switchyard-server/src/stats/accumulator.rs index ee5561808..9b7d91032 100644 --- a/crates/switchyard-server/src/stats/accumulator.rs +++ b/crates/switchyard-server/src/stats/accumulator.rs @@ -9,6 +9,7 @@ use std::sync::Arc; use parking_lot::{Mutex, MutexGuard}; use serde::Serialize; +use super::algorithm::AlgorithmStatsSnapshot; use super::cache_eligibility::PrefixProbe; const MAX_LATENCY_SAMPLES: usize = 10_000; @@ -165,6 +166,7 @@ impl StatsAccumulatorInner { routing_overhead: self.routing_overhead.snapshot(), routing_fallbacks: self.routing_fallbacks, classifier, + algorithm_stats: AlgorithmStatsSnapshot::default(), } } } @@ -275,6 +277,7 @@ pub(crate) struct StatsSnapshot { pub routing_overhead: LatencyHistogramSnapshot, pub routing_fallbacks: RoutingFallbackStats, pub classifier: ClassifierStatsSnapshot, + pub algorithm_stats: AlgorithmStatsSnapshot, } /// Legacy fallback counters retained in the stats response shape. diff --git a/crates/switchyard-server/src/stats/algorithm.rs b/crates/switchyard-server/src/stats/algorithm.rs new file mode 100644 index 000000000..b9b461059 --- /dev/null +++ b/crates/switchyard-server/src/stats/algorithm.rs @@ -0,0 +1,168 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Server-owned projections of algorithm OpenTelemetry metrics. + +mod stage_router; + +use std::collections::BTreeSet; +use std::sync::Arc; + +use parking_lot::Mutex; +use prometheus::Registry; +use serde::Serialize; + +use stage_router::{StageRouterCumulative, StageRouterStatsSnapshot}; + +const STAGE_ROUTER: &str = "stage_router"; + +/// Cumulative algorithm metrics and the baseline used by `/v1/stats/reset`. +#[derive(Clone)] +pub(crate) struct AlgorithmStats { + inner: Arc, +} + +struct AlgorithmStatsInner { + registry: Registry, + configured: BTreeSet, + baseline: Mutex, +} + +#[derive(Clone, Debug, Default)] +struct AlgorithmMetrics { + stage_router: StageRouterCumulative, +} + +/// Curated algorithm-specific data included in the JSON stats response. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct AlgorithmStatsSnapshot { + #[serde(skip_serializing_if = "Option::is_none")] + pub stage_router: Option, +} + +impl AlgorithmStats { + /// Starts algorithm stats at the registry's current cumulative values. + pub(crate) fn new(registry: Registry, configured: impl IntoIterator) -> Self { + let baseline = collect(®istry); + Self { + inner: Arc::new(AlgorithmStatsInner { + registry, + configured: configured.into_iter().collect(), + baseline: Mutex::new(baseline), + }), + } + } + + /// Projects cumulative OpenTelemetry metrics since the last baseline. + pub(crate) fn snapshot(&self) -> AlgorithmStatsSnapshot { + let current = collect(&self.inner.registry); + let baseline = self.inner.baseline.lock(); + AlgorithmStatsSnapshot { + stage_router: self + .inner + .configured + .contains(STAGE_ROUTER) + .then(|| current.stage_router.delta(&baseline.stage_router)), + } + } + + /// Moves the JSON baseline without resetting process-lifetime OpenTelemetry metrics. + pub(crate) fn reset(&self) { + *self.inner.baseline.lock() = collect(&self.inner.registry); + } +} + +fn collect(registry: &Registry) -> AlgorithmMetrics { + let families = registry.gather(); + AlgorithmMetrics { + stage_router: StageRouterCumulative::collect(&families), + } +} + +#[cfg(test)] +mod tests { + use opentelemetry::KeyValue; + use opentelemetry::metrics::MeterProvider as _; + use opentelemetry_sdk::metrics::SdkMeterProvider; + + use super::*; + + #[test] + fn stage_router_projection_preserves_decisions_scores_and_reset_baseline() { + let registry = Registry::new(); + let exporter = opentelemetry_prometheus::exporter() + .with_registry(registry.clone()) + .build() + .unwrap_or_else(|error| panic!("failed to build metrics exporter: {error}")); + let provider = SdkMeterProvider::builder().with_reader(exporter).build(); + let meter = provider.meter("switchyard"); + let stats = AlgorithmStats::new(registry, [STAGE_ROUTER.to_string()]); + + meter + .u64_counter("switchyard.stage_router.routing_decisions") + .build() + .add( + 2, + &[ + KeyValue::new("decision_source", "dimensions"), + KeyValue::new("target_name", "model/efficient"), + ], + ); + meter + .u64_counter("switchyard.stage_router.routing_decisions") + .build() + .add( + 1, + &[ + KeyValue::new("decision_source", "dimensions"), + KeyValue::new("target_name", "model/capable"), + ], + ); + for value in [0.5, -0.25] { + meter + .f64_histogram("switchyard.stage_router.score") + .build() + .record(value, &[]); + } + meter + .f64_histogram("switchyard.stage_router.confidence") + .build() + .record(0.75, &[]); + + let snapshot = stats.snapshot(); + let stage = snapshot + .stage_router + .unwrap_or_else(|| panic!("stage-router stats missing")); + let dimensions = &stage.routing_decisions["dimensions"]; + assert_eq!(dimensions.total, 3); + assert_eq!(dimensions.targets["model/efficient"], 2); + assert_eq!(dimensions.targets["model/capable"], 1); + assert_eq!(stage.scoring.score.count, 2); + assert_eq!(stage.scoring.score.sum, 0.25); + assert_eq!(stage.scoring.score.avg, 0.125); + assert_eq!(stage.scoring.confidence.avg, 0.75); + + stats.reset(); + assert_eq!( + stats.snapshot().stage_router, + Some(StageRouterStatsSnapshot::default()) + ); + + meter + .u64_counter("switchyard.stage_router.routing_decisions") + .build() + .add( + 1, + &[ + KeyValue::new("decision_source", "override"), + KeyValue::new("target_name", "model/capable"), + ], + ); + let after_reset = stats + .snapshot() + .stage_router + .unwrap_or_else(|| panic!("stage-router stats missing after reset")); + assert_eq!(after_reset.routing_decisions["override"].total, 1); + assert!(!after_reset.routing_decisions.contains_key("dimensions")); + } +} diff --git a/crates/switchyard-server/src/stats/algorithm/stage_router.rs b/crates/switchyard-server/src/stats/algorithm/stage_router.rs new file mode 100644 index 000000000..7f7d1fe9c --- /dev/null +++ b/crates/switchyard-server/src/stats/algorithm/stage_router.rs @@ -0,0 +1,193 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Stage-router projection from cumulative Prometheus metric families. + +use std::collections::BTreeMap; + +use prometheus::proto::{Metric, MetricFamily}; +use serde::Serialize; + +const ROUTING_DECISIONS_METRIC: &str = "switchyard_stage_router_routing_decisions_total"; +const SCORE_METRIC: &str = "switchyard_stage_router_score"; +const CONFIDENCE_METRIC: &str = "switchyard_stage_router_confidence"; +const SEVERITY_METRIC: &str = "switchyard_stage_router_severity"; +const SPINNING_METRIC: &str = "switchyard_stage_router_spinning"; +const EXPLORING_METRIC: &str = "switchyard_stage_router_exploring"; +const PRODUCTION_INTENSITY_METRIC: &str = "switchyard_stage_router_production_intensity"; + +#[derive(Clone, Debug, Default)] +pub(super) struct StageRouterCumulative { + decisions: BTreeMap, + score: HistogramTotal, + confidence: HistogramTotal, + severity: HistogramTotal, + spinning: HistogramTotal, + exploring: HistogramTotal, + production_intensity: HistogramTotal, +} + +#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)] +struct DecisionKey { + source: String, + target: String, +} + +#[derive(Clone, Copy, Debug, Default)] +struct HistogramTotal { + count: u64, + sum: f64, +} + +/// Human-readable stage-router stats derived from its native metrics. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct StageRouterStatsSnapshot { + pub routing_decisions: BTreeMap, + pub scoring: ScoringStatsSnapshot, +} + +/// One decision source, retaining the semantic targets it selected. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct DecisionStatsSnapshot { + pub total: u64, + pub targets: BTreeMap, +} + +/// Stage scorer output and input-dimension summaries. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct ScoringStatsSnapshot { + pub score: MetricSummary, + pub confidence: MetricSummary, + pub dimensions: DimensionStatsSnapshot, +} + +/// Input dimensions evaluated by the stage scorer. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct DimensionStatsSnapshot { + pub severity: MetricSummary, + pub spinning: MetricSummary, + pub exploring: MetricSummary, + pub production_intensity: MetricSummary, +} + +/// Exact count, sum, and average since the last JSON stats reset. +#[derive(Clone, Copy, Debug, Default, PartialEq, Serialize)] +pub(crate) struct MetricSummary { + pub count: u64, + pub sum: f64, + pub avg: f64, +} + +impl StageRouterCumulative { + pub(super) fn collect(families: &[MetricFamily]) -> Self { + Self { + decisions: collect_decisions(families), + score: collect_histogram(families, SCORE_METRIC), + confidence: collect_histogram(families, CONFIDENCE_METRIC), + severity: collect_histogram(families, SEVERITY_METRIC), + spinning: collect_histogram(families, SPINNING_METRIC), + exploring: collect_histogram(families, EXPLORING_METRIC), + production_intensity: collect_histogram(families, PRODUCTION_INTENSITY_METRIC), + } + } + + pub(super) fn delta(&self, baseline: &Self) -> StageRouterStatsSnapshot { + let mut routing_decisions: BTreeMap = BTreeMap::new(); + for (key, current) in &self.decisions { + let count = current.saturating_sub(*baseline.decisions.get(key).unwrap_or(&0)); + if count == 0 { + continue; + } + let source = routing_decisions.entry(key.source.clone()).or_default(); + source.total = source.total.saturating_add(count); + source.targets.insert(key.target.clone(), count); + } + StageRouterStatsSnapshot { + routing_decisions, + scoring: ScoringStatsSnapshot { + score: self.score.delta(baseline.score), + confidence: self.confidence.delta(baseline.confidence), + dimensions: DimensionStatsSnapshot { + severity: self.severity.delta(baseline.severity), + spinning: self.spinning.delta(baseline.spinning), + exploring: self.exploring.delta(baseline.exploring), + production_intensity: self + .production_intensity + .delta(baseline.production_intensity), + }, + }, + } + } +} + +impl HistogramTotal { + fn delta(self, baseline: Self) -> MetricSummary { + let count = self.count.saturating_sub(baseline.count); + if count == 0 { + return MetricSummary::default(); + } + let sum = round4(self.sum - baseline.sum); + MetricSummary { + count, + sum, + avg: round4(sum / count as f64), + } + } +} + +fn collect_decisions(families: &[MetricFamily]) -> BTreeMap { + let mut decisions = BTreeMap::new(); + for metric in metrics(families, ROUTING_DECISIONS_METRIC) { + let Some(source) = label(metric, "decision_source") else { + continue; + }; + let Some(target) = label(metric, "target_name") else { + continue; + }; + let Some(counter) = metric.get_counter().as_ref() else { + continue; + }; + let value = counter.value(); + if value.is_finite() && value > 0.0 { + let count = decisions + .entry(DecisionKey { + source: source.to_string(), + target: target.to_string(), + }) + .or_insert(0u64); + *count = count.saturating_add(value as u64); + } + } + decisions +} + +fn collect_histogram(families: &[MetricFamily], name: &str) -> HistogramTotal { + metrics(families, name).fold(HistogramTotal::default(), |mut total, metric| { + let Some(histogram) = metric.get_histogram().as_ref() else { + return total; + }; + total.count = total.count.saturating_add(histogram.sample_count()); + total.sum += histogram.sample_sum(); + total + }) +} + +fn metrics<'a>(families: &'a [MetricFamily], name: &'a str) -> impl Iterator { + families + .iter() + .filter(move |family| family.name() == name) + .flat_map(|family| family.get_metric()) +} + +fn label<'a>(metric: &'a Metric, name: &str) -> Option<&'a str> { + metric + .get_label() + .iter() + .find(|label| label.name() == name) + .map(|label| label.value()) +} + +fn round4(value: f64) -> f64 { + let rounded = (value * 10_000.0).round() / 10_000.0; + if rounded == 0.0 { 0.0 } else { rounded } +} diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index 0e3272cb0..65bd5ab79 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -294,6 +294,7 @@ async fn stats_exposes_the_exact_empty_schema_and_no_legacy_alias() -> TestResul "total_tokens": empty_token_totals(), "models": {}, }, + "algorithm_stats": {}, }) ); assert_eq!( @@ -754,11 +755,11 @@ format = "openai_chat" base_url = "{base_url}" [targets.strong] -id = "model/strong" +id = "model/stats-strong" llm_client = "upstream" [targets.weak] -id = "model/weak" +id = "model/stats-weak" llm_client = "upstream" [routes.stage] @@ -798,9 +799,14 @@ confidence_threshold = 0.5 .headers .get("x-model-router-selected-model") .and_then(|value| value.to_str().ok()), - Some("model/strong"), + Some("model/stats-strong"), "a critical error should escalate on the signals alone" ); + let stats = send(&app, "GET", "/v1/stats", None).await?.json()?; + assert_eq!( + stats["algorithm_stats"]["stage_router"]["routing_decisions"]["override"]["targets"]["model/stats-strong"], + 1 + ); Ok(()) } diff --git a/docs/known_issues.md b/docs/known_issues.md index d1ed129f9..a1b9a345c 100644 --- a/docs/known_issues.md +++ b/docs/known_issues.md @@ -2,7 +2,7 @@ ## 0.2.0 1. Buffered upstream work continues after the client disconnects, so a cancelled request can still incur provider cost. -2. Routing-tier attribution is missing from `GET /v1/stats` and `/metrics` for LLM-classifier judge failures that route to the default target, escalation decisions, and `stage_router` fallback decisions. +2. Routing-tier attribution is missing from `GET /v1/stats` and `/metrics` for LLM-classifier judge failures that route to the default target and escalation decisions. 3. The retry recovery counter stays at zero after a successful upstream retry. 4. `x-switchyard-session-id` is not recorded in native session stats. 5. The native server does not send the documented `X-Switchyard-Version` header upstream. diff --git a/docs/routing_algorithms/stage_router_routing.md b/docs/routing_algorithms/stage_router_routing.md index 207586134..bba42fbd2 100644 --- a/docs/routing_algorithms/stage_router_routing.md +++ b/docs/routing_algorithms/stage_router_routing.md @@ -109,8 +109,8 @@ staying on the default tier.) | `1.0` | yes (required) | Classifier-driven. Tool signals only apply hard overrides; other turns reach the classifier. | The signal-vs-classifier split is dataset-dependent. Measure it in -production: `/v1/stats` reports traffic by tier and model, while response headers -and structured decision logs explain individual selections. +production: `/v1/stats` reports stage-router decisions by source and semantic +target, while response headers and structured decision logs explain individual selections. ### Calibrating the threshold from run data From 6cb86c134a8fb6deb99a31464ae0e10f7dfa00a2 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 10 Aug 2026 16:17:12 -0700 Subject: [PATCH 2/9] refactor(server): consolidate stats ownership Signed-off-by: nachiketb --- crates/switchyard-server/src/lib.rs | 15 ++++--------- crates/switchyard-server/src/stats.rs | 1 - .../src/stats/accumulator.rs | 21 ++++++++++++++----- .../switchyard-server/src/stats/algorithm.rs | 6 ++++++ 4 files changed, 26 insertions(+), 17 deletions(-) diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 6e2d35a67..87cc6bcdd 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -45,9 +45,7 @@ use tracing::{Instrument, Level}; use switchyard_translation::{WireFormat, decode_request}; use crate::response::into_http_response; -use crate::stats::{ - AlgorithmStats, StatsAccumulator, StatsSnapshot, prefix_probe, tracking_enabled_from_env, -}; +use crate::stats::{StatsAccumulator, StatsSnapshot, prefix_probe, tracking_enabled_from_env}; pub use observability::{flush_observability, initialize_observability}; @@ -135,7 +133,6 @@ pub struct ServerState { routes: Arc>, metrics: prometheus::Registry, stats: StatsAccumulator, - algorithm_stats: AlgorithmStats, routing_log: Option, track_cache_eligibility: bool, } @@ -222,7 +219,7 @@ impl ServerState { return Err(ServerError::new("at least one algorithm route is required")); } let metrics = metrics::registry().map_err(ServerError::new)?; - let algorithm_stats = AlgorithmStats::new( + let stats = StatsAccumulator::new( metrics.clone(), entries .values() @@ -231,8 +228,7 @@ impl ServerState { Ok(Self { routes: Arc::new(entries), metrics, - stats: StatsAccumulator::default(), - algorithm_stats, + stats, routing_log: None, track_cache_eligibility: tracking_enabled_from_env(), }) @@ -1026,14 +1022,11 @@ async fn models(State(state): State) -> Json { } async fn get_stats(State(state): State) -> Json { - let mut snapshot = state.stats.snapshot(); - snapshot.algorithm_stats = state.algorithm_stats.snapshot(); - Json(snapshot) + Json(state.stats.snapshot()) } async fn reset_stats(State(state): State) -> Json { state.stats.reset(); - state.algorithm_stats.reset(); Json(json!({"status": "reset"})) } diff --git a/crates/switchyard-server/src/stats.rs b/crates/switchyard-server/src/stats.rs index 74f9dda51..fb070073b 100644 --- a/crates/switchyard-server/src/stats.rs +++ b/crates/switchyard-server/src/stats.rs @@ -8,5 +8,4 @@ mod algorithm; mod cache_eligibility; pub(crate) use accumulator::{StatsAccumulator, StatsSnapshot, TokenUsage}; -pub(crate) use algorithm::AlgorithmStats; pub(crate) use cache_eligibility::{prefix_probe, tracking_enabled_from_env}; diff --git a/crates/switchyard-server/src/stats/accumulator.rs b/crates/switchyard-server/src/stats/accumulator.rs index 9b7d91032..b7de170f8 100644 --- a/crates/switchyard-server/src/stats/accumulator.rs +++ b/crates/switchyard-server/src/stats/accumulator.rs @@ -7,9 +7,10 @@ use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; use parking_lot::{Mutex, MutexGuard}; +use prometheus::Registry; use serde::Serialize; -use super::algorithm::AlgorithmStatsSnapshot; +use super::algorithm::{AlgorithmStats, AlgorithmStatsSnapshot}; use super::cache_eligibility::PrefixProbe; const MAX_LATENCY_SAMPLES: usize = 10_000; @@ -26,12 +27,21 @@ pub(crate) struct TokenUsage { } /// Thread-safe process-local stats store. -#[derive(Clone, Debug, Default)] +#[derive(Clone, Default)] pub(crate) struct StatsAccumulator { inner: Arc>, + algorithm_stats: AlgorithmStats, } impl StatsAccumulator { + /// Creates a stats store that projects metrics for the configured algorithms. + pub(crate) fn new(registry: Registry, configured: impl IntoIterator) -> Self { + Self { + inner: Arc::default(), + algorithm_stats: AlgorithmStats::new(registry, configured), + } + } + /// Records one successful routed backend call. pub(crate) fn record_success(&self, model: impl Into, backend_latency_ms: f64) { let mut inner = self.lock(); @@ -117,12 +127,13 @@ impl StatsAccumulator { /// Returns a serializable point-in-time snapshot. pub(crate) fn snapshot(&self) -> StatsSnapshot { let inner = self.lock().clone(); - inner.snapshot() + inner.snapshot(self.algorithm_stats.snapshot()) } /// Clears all accumulated stats. pub(crate) fn reset(&self) { *self.lock() = StatsAccumulatorInner::default(); + self.algorithm_stats.reset(); } fn lock(&self) -> MutexGuard<'_, StatsAccumulatorInner> { @@ -151,7 +162,7 @@ impl StatsAccumulatorInner { self.by_classifier.entry(model).or_default() } - fn snapshot(&self) -> StatsSnapshot { + fn snapshot(&self, algorithm_stats: AlgorithmStatsSnapshot) -> StatsSnapshot { let (models, total_tokens) = build_model_snapshots(&self.by_model, self.total_requests); let classifier = build_classifier_snapshot( &self.by_classifier, @@ -166,7 +177,7 @@ impl StatsAccumulatorInner { routing_overhead: self.routing_overhead.snapshot(), routing_fallbacks: self.routing_fallbacks, classifier, - algorithm_stats: AlgorithmStatsSnapshot::default(), + algorithm_stats, } } } diff --git a/crates/switchyard-server/src/stats/algorithm.rs b/crates/switchyard-server/src/stats/algorithm.rs index b9b461059..e051ccd16 100644 --- a/crates/switchyard-server/src/stats/algorithm.rs +++ b/crates/switchyard-server/src/stats/algorithm.rs @@ -22,6 +22,12 @@ pub(crate) struct AlgorithmStats { inner: Arc, } +impl Default for AlgorithmStats { + fn default() -> Self { + Self::new(Registry::new(), std::iter::empty()) + } +} + struct AlgorithmStatsInner { registry: Registry, configured: BTreeSet, From 270cd47f0075ddb9ef782da019d0d29052662340 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 10 Aug 2026 16:21:33 -0700 Subject: [PATCH 3/9] docs: preserve released known issues Signed-off-by: nachiketb --- docs/known_issues.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/known_issues.md b/docs/known_issues.md index a1b9a345c..d1ed129f9 100644 --- a/docs/known_issues.md +++ b/docs/known_issues.md @@ -2,7 +2,7 @@ ## 0.2.0 1. Buffered upstream work continues after the client disconnects, so a cancelled request can still incur provider cost. -2. Routing-tier attribution is missing from `GET /v1/stats` and `/metrics` for LLM-classifier judge failures that route to the default target and escalation decisions. +2. Routing-tier attribution is missing from `GET /v1/stats` and `/metrics` for LLM-classifier judge failures that route to the default target, escalation decisions, and `stage_router` fallback decisions. 3. The retry recovery counter stays at zero after a successful upstream retry. 4. `x-switchyard-session-id` is not recorded in native session stats. 5. The native server does not send the documented `X-Switchyard-Version` header upstream. From 5c0f3d61a07f5bec59555537644279857f7af760 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 10 Aug 2026 16:24:13 -0700 Subject: [PATCH 4/9] test(server): colocate stage router stats coverage Signed-off-by: nachiketb --- .../switchyard-server/src/stats/algorithm.rs | 88 ------------------ .../src/stats/algorithm/stage_router.rs | 90 +++++++++++++++++++ 2 files changed, 90 insertions(+), 88 deletions(-) diff --git a/crates/switchyard-server/src/stats/algorithm.rs b/crates/switchyard-server/src/stats/algorithm.rs index e051ccd16..48e97877f 100644 --- a/crates/switchyard-server/src/stats/algorithm.rs +++ b/crates/switchyard-server/src/stats/algorithm.rs @@ -84,91 +84,3 @@ fn collect(registry: &Registry) -> AlgorithmMetrics { stage_router: StageRouterCumulative::collect(&families), } } - -#[cfg(test)] -mod tests { - use opentelemetry::KeyValue; - use opentelemetry::metrics::MeterProvider as _; - use opentelemetry_sdk::metrics::SdkMeterProvider; - - use super::*; - - #[test] - fn stage_router_projection_preserves_decisions_scores_and_reset_baseline() { - let registry = Registry::new(); - let exporter = opentelemetry_prometheus::exporter() - .with_registry(registry.clone()) - .build() - .unwrap_or_else(|error| panic!("failed to build metrics exporter: {error}")); - let provider = SdkMeterProvider::builder().with_reader(exporter).build(); - let meter = provider.meter("switchyard"); - let stats = AlgorithmStats::new(registry, [STAGE_ROUTER.to_string()]); - - meter - .u64_counter("switchyard.stage_router.routing_decisions") - .build() - .add( - 2, - &[ - KeyValue::new("decision_source", "dimensions"), - KeyValue::new("target_name", "model/efficient"), - ], - ); - meter - .u64_counter("switchyard.stage_router.routing_decisions") - .build() - .add( - 1, - &[ - KeyValue::new("decision_source", "dimensions"), - KeyValue::new("target_name", "model/capable"), - ], - ); - for value in [0.5, -0.25] { - meter - .f64_histogram("switchyard.stage_router.score") - .build() - .record(value, &[]); - } - meter - .f64_histogram("switchyard.stage_router.confidence") - .build() - .record(0.75, &[]); - - let snapshot = stats.snapshot(); - let stage = snapshot - .stage_router - .unwrap_or_else(|| panic!("stage-router stats missing")); - let dimensions = &stage.routing_decisions["dimensions"]; - assert_eq!(dimensions.total, 3); - assert_eq!(dimensions.targets["model/efficient"], 2); - assert_eq!(dimensions.targets["model/capable"], 1); - assert_eq!(stage.scoring.score.count, 2); - assert_eq!(stage.scoring.score.sum, 0.25); - assert_eq!(stage.scoring.score.avg, 0.125); - assert_eq!(stage.scoring.confidence.avg, 0.75); - - stats.reset(); - assert_eq!( - stats.snapshot().stage_router, - Some(StageRouterStatsSnapshot::default()) - ); - - meter - .u64_counter("switchyard.stage_router.routing_decisions") - .build() - .add( - 1, - &[ - KeyValue::new("decision_source", "override"), - KeyValue::new("target_name", "model/capable"), - ], - ); - let after_reset = stats - .snapshot() - .stage_router - .unwrap_or_else(|| panic!("stage-router stats missing after reset")); - assert_eq!(after_reset.routing_decisions["override"].total, 1); - assert!(!after_reset.routing_decisions.contains_key("dimensions")); - } -} diff --git a/crates/switchyard-server/src/stats/algorithm/stage_router.rs b/crates/switchyard-server/src/stats/algorithm/stage_router.rs index 7f7d1fe9c..4b3d0e59d 100644 --- a/crates/switchyard-server/src/stats/algorithm/stage_router.rs +++ b/crates/switchyard-server/src/stats/algorithm/stage_router.rs @@ -191,3 +191,93 @@ fn round4(value: f64) -> f64 { let rounded = (value * 10_000.0).round() / 10_000.0; if rounded == 0.0 { 0.0 } else { rounded } } + +#[cfg(test)] +mod tests { + use opentelemetry::KeyValue; + use opentelemetry::metrics::MeterProvider as _; + use opentelemetry_sdk::metrics::SdkMeterProvider; + use prometheus::Registry; + + use super::*; + use crate::stats::algorithm::{AlgorithmStats, STAGE_ROUTER}; + + #[test] + fn stage_router_projection_preserves_decisions_scores_and_reset_baseline() { + let registry = Registry::new(); + let exporter = opentelemetry_prometheus::exporter() + .with_registry(registry.clone()) + .build() + .unwrap_or_else(|error| panic!("failed to build metrics exporter: {error}")); + let provider = SdkMeterProvider::builder().with_reader(exporter).build(); + let meter = provider.meter("switchyard"); + let stats = AlgorithmStats::new(registry, [STAGE_ROUTER.to_string()]); + + meter + .u64_counter("switchyard.stage_router.routing_decisions") + .build() + .add( + 2, + &[ + KeyValue::new("decision_source", "dimensions"), + KeyValue::new("target_name", "model/efficient"), + ], + ); + meter + .u64_counter("switchyard.stage_router.routing_decisions") + .build() + .add( + 1, + &[ + KeyValue::new("decision_source", "dimensions"), + KeyValue::new("target_name", "model/capable"), + ], + ); + for value in [0.5, -0.25] { + meter + .f64_histogram("switchyard.stage_router.score") + .build() + .record(value, &[]); + } + meter + .f64_histogram("switchyard.stage_router.confidence") + .build() + .record(0.75, &[]); + + let snapshot = stats.snapshot(); + let stage = snapshot + .stage_router + .unwrap_or_else(|| panic!("stage-router stats missing")); + let dimensions = &stage.routing_decisions["dimensions"]; + assert_eq!(dimensions.total, 3); + assert_eq!(dimensions.targets["model/efficient"], 2); + assert_eq!(dimensions.targets["model/capable"], 1); + assert_eq!(stage.scoring.score.count, 2); + assert_eq!(stage.scoring.score.sum, 0.25); + assert_eq!(stage.scoring.score.avg, 0.125); + assert_eq!(stage.scoring.confidence.avg, 0.75); + + stats.reset(); + assert_eq!( + stats.snapshot().stage_router, + Some(StageRouterStatsSnapshot::default()) + ); + + meter + .u64_counter("switchyard.stage_router.routing_decisions") + .build() + .add( + 1, + &[ + KeyValue::new("decision_source", "override"), + KeyValue::new("target_name", "model/capable"), + ], + ); + let after_reset = stats + .snapshot() + .stage_router + .unwrap_or_else(|| panic!("stage-router stats missing after reset")); + assert_eq!(after_reset.routing_decisions["override"].total, 1); + assert!(!after_reset.routing_decisions.contains_key("dimensions")); + } +} From bf07592550305e353bd0609e86915d3f3ef3485e Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 10 Aug 2026 16:33:47 -0700 Subject: [PATCH 5/9] refactor(server): simplify algorithm stats state Signed-off-by: nachiketb --- crates/switchyard-server/src/stats.rs | 2 +- .../src/stats/accumulator.rs | 53 +++++++++--- .../switchyard-server/src/stats/algorithm.rs | 86 ------------------- .../switchyard-server/src/stats/algorithms.rs | 48 +++++++++++ .../{algorithm => algorithms}/stage_router.rs | 8 +- 5 files changed, 95 insertions(+), 102 deletions(-) delete mode 100644 crates/switchyard-server/src/stats/algorithm.rs create mode 100644 crates/switchyard-server/src/stats/algorithms.rs rename crates/switchyard-server/src/stats/{algorithm => algorithms}/stage_router.rs (97%) diff --git a/crates/switchyard-server/src/stats.rs b/crates/switchyard-server/src/stats.rs index fb070073b..1a419ffd2 100644 --- a/crates/switchyard-server/src/stats.rs +++ b/crates/switchyard-server/src/stats.rs @@ -4,7 +4,7 @@ //! Process-local JSON stats accounting for the Rust server. mod accumulator; -mod algorithm; +mod algorithms; mod cache_eligibility; pub(crate) use accumulator::{StatsAccumulator, StatsSnapshot, TokenUsage}; diff --git a/crates/switchyard-server/src/stats/accumulator.rs b/crates/switchyard-server/src/stats/accumulator.rs index b7de170f8..fec28ec8e 100644 --- a/crates/switchyard-server/src/stats/accumulator.rs +++ b/crates/switchyard-server/src/stats/accumulator.rs @@ -3,14 +3,14 @@ //! Thread-safe stats accumulator and serializable snapshot schema. -use std::collections::{BTreeMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashSet}; use std::sync::Arc; use parking_lot::{Mutex, MutexGuard}; use prometheus::Registry; use serde::Serialize; -use super::algorithm::{AlgorithmStats, AlgorithmStatsSnapshot}; +use super::algorithms::{AlgorithmMetrics, AlgorithmStatsSnapshot}; use super::cache_eligibility::PrefixProbe; const MAX_LATENCY_SAMPLES: usize = 10_000; @@ -27,18 +27,22 @@ pub(crate) struct TokenUsage { } /// Thread-safe process-local stats store. -#[derive(Clone, Default)] +#[derive(Clone)] pub(crate) struct StatsAccumulator { inner: Arc>, - algorithm_stats: AlgorithmStats, +} + +impl Default for StatsAccumulator { + fn default() -> Self { + Self::new(Registry::new(), std::iter::empty()) + } } impl StatsAccumulator { /// Creates a stats store that projects metrics for the configured algorithms. pub(crate) fn new(registry: Registry, configured: impl IntoIterator) -> Self { Self { - inner: Arc::default(), - algorithm_stats: AlgorithmStats::new(registry, configured), + inner: Arc::new(Mutex::new(StatsAccumulatorInner::new(registry, configured))), } } @@ -126,14 +130,12 @@ impl StatsAccumulator { /// Returns a serializable point-in-time snapshot. pub(crate) fn snapshot(&self) -> StatsSnapshot { - let inner = self.lock().clone(); - inner.snapshot(self.algorithm_stats.snapshot()) + self.lock().snapshot() } /// Clears all accumulated stats. pub(crate) fn reset(&self) { - *self.lock() = StatsAccumulatorInner::default(); - self.algorithm_stats.reset(); + self.lock().reset(); } fn lock(&self) -> MutexGuard<'_, StatsAccumulatorInner> { @@ -141,7 +143,6 @@ impl StatsAccumulator { } } -#[derive(Clone, Debug, Default)] struct StatsAccumulatorInner { by_model: BTreeMap, total_requests: u64, @@ -151,9 +152,29 @@ struct StatsAccumulatorInner { by_classifier: BTreeMap, classifier_requests: u64, classifier_errors: u64, + algorithm_registry: Registry, + configured_algorithms: BTreeSet, + algorithm_baseline: AlgorithmMetrics, } impl StatsAccumulatorInner { + fn new(registry: Registry, configured: impl IntoIterator) -> Self { + let algorithm_baseline = AlgorithmMetrics::collect(®istry); + Self { + by_model: BTreeMap::new(), + total_requests: 0, + total_errors: 0, + routing_overhead: LatencyHistogram::default(), + routing_fallbacks: RoutingFallbackStats::default(), + by_classifier: BTreeMap::new(), + classifier_requests: 0, + classifier_errors: 0, + algorithm_registry: registry, + configured_algorithms: configured.into_iter().collect(), + algorithm_baseline, + } + } + fn model_stats_mut(&mut self, model: String) -> &mut ModelStats { self.by_model.entry(model).or_default() } @@ -162,13 +183,15 @@ impl StatsAccumulatorInner { self.by_classifier.entry(model).or_default() } - fn snapshot(&self, algorithm_stats: AlgorithmStatsSnapshot) -> StatsSnapshot { + fn snapshot(&self) -> StatsSnapshot { let (models, total_tokens) = build_model_snapshots(&self.by_model, self.total_requests); let classifier = build_classifier_snapshot( &self.by_classifier, self.classifier_requests, self.classifier_errors, ); + let algorithm_stats = AlgorithmMetrics::collect(&self.algorithm_registry) + .snapshot(&self.algorithm_baseline, &self.configured_algorithms); StatsSnapshot { total_requests: self.total_requests, total_errors: self.total_errors, @@ -180,6 +203,12 @@ impl StatsAccumulatorInner { algorithm_stats, } } + + fn reset(&mut self) { + let registry = self.algorithm_registry.clone(); + let configured = std::mem::take(&mut self.configured_algorithms); + *self = Self::new(registry, configured); + } } #[derive(Clone, Debug, Default)] diff --git a/crates/switchyard-server/src/stats/algorithm.rs b/crates/switchyard-server/src/stats/algorithm.rs deleted file mode 100644 index 48e97877f..000000000 --- a/crates/switchyard-server/src/stats/algorithm.rs +++ /dev/null @@ -1,86 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -//! Server-owned projections of algorithm OpenTelemetry metrics. - -mod stage_router; - -use std::collections::BTreeSet; -use std::sync::Arc; - -use parking_lot::Mutex; -use prometheus::Registry; -use serde::Serialize; - -use stage_router::{StageRouterCumulative, StageRouterStatsSnapshot}; - -const STAGE_ROUTER: &str = "stage_router"; - -/// Cumulative algorithm metrics and the baseline used by `/v1/stats/reset`. -#[derive(Clone)] -pub(crate) struct AlgorithmStats { - inner: Arc, -} - -impl Default for AlgorithmStats { - fn default() -> Self { - Self::new(Registry::new(), std::iter::empty()) - } -} - -struct AlgorithmStatsInner { - registry: Registry, - configured: BTreeSet, - baseline: Mutex, -} - -#[derive(Clone, Debug, Default)] -struct AlgorithmMetrics { - stage_router: StageRouterCumulative, -} - -/// Curated algorithm-specific data included in the JSON stats response. -#[derive(Clone, Debug, Default, PartialEq, Serialize)] -pub(crate) struct AlgorithmStatsSnapshot { - #[serde(skip_serializing_if = "Option::is_none")] - pub stage_router: Option, -} - -impl AlgorithmStats { - /// Starts algorithm stats at the registry's current cumulative values. - pub(crate) fn new(registry: Registry, configured: impl IntoIterator) -> Self { - let baseline = collect(®istry); - Self { - inner: Arc::new(AlgorithmStatsInner { - registry, - configured: configured.into_iter().collect(), - baseline: Mutex::new(baseline), - }), - } - } - - /// Projects cumulative OpenTelemetry metrics since the last baseline. - pub(crate) fn snapshot(&self) -> AlgorithmStatsSnapshot { - let current = collect(&self.inner.registry); - let baseline = self.inner.baseline.lock(); - AlgorithmStatsSnapshot { - stage_router: self - .inner - .configured - .contains(STAGE_ROUTER) - .then(|| current.stage_router.delta(&baseline.stage_router)), - } - } - - /// Moves the JSON baseline without resetting process-lifetime OpenTelemetry metrics. - pub(crate) fn reset(&self) { - *self.inner.baseline.lock() = collect(&self.inner.registry); - } -} - -fn collect(registry: &Registry) -> AlgorithmMetrics { - let families = registry.gather(); - AlgorithmMetrics { - stage_router: StageRouterCumulative::collect(&families), - } -} diff --git a/crates/switchyard-server/src/stats/algorithms.rs b/crates/switchyard-server/src/stats/algorithms.rs new file mode 100644 index 000000000..eb351b3eb --- /dev/null +++ b/crates/switchyard-server/src/stats/algorithms.rs @@ -0,0 +1,48 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Server-owned projections of algorithm OpenTelemetry metrics. + +mod stage_router; + +use std::collections::BTreeSet; + +use prometheus::Registry; +use serde::Serialize; + +use stage_router::{StageRouterCumulative, StageRouterStatsSnapshot}; + +pub(super) const STAGE_ROUTER: &str = "stage_router"; + +#[derive(Clone, Debug, Default)] +pub(super) struct AlgorithmMetrics { + stage_router: StageRouterCumulative, +} + +/// Curated algorithm-specific data included in the JSON stats response. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct AlgorithmStatsSnapshot { + #[serde(skip_serializing_if = "Option::is_none")] + pub stage_router: Option, +} + +impl AlgorithmMetrics { + pub(super) fn collect(registry: &Registry) -> Self { + let families = registry.gather(); + Self { + stage_router: StageRouterCumulative::collect(&families), + } + } + + pub(super) fn snapshot( + &self, + baseline: &Self, + configured: &BTreeSet, + ) -> AlgorithmStatsSnapshot { + AlgorithmStatsSnapshot { + stage_router: configured + .contains(STAGE_ROUTER) + .then(|| self.stage_router.delta(&baseline.stage_router)), + } + } +} diff --git a/crates/switchyard-server/src/stats/algorithm/stage_router.rs b/crates/switchyard-server/src/stats/algorithms/stage_router.rs similarity index 97% rename from crates/switchyard-server/src/stats/algorithm/stage_router.rs rename to crates/switchyard-server/src/stats/algorithms/stage_router.rs index 4b3d0e59d..65a71a9f9 100644 --- a/crates/switchyard-server/src/stats/algorithm/stage_router.rs +++ b/crates/switchyard-server/src/stats/algorithms/stage_router.rs @@ -200,7 +200,7 @@ mod tests { use prometheus::Registry; use super::*; - use crate::stats::algorithm::{AlgorithmStats, STAGE_ROUTER}; + use crate::stats::{StatsAccumulator, algorithms::STAGE_ROUTER}; #[test] fn stage_router_projection_preserves_decisions_scores_and_reset_baseline() { @@ -211,7 +211,7 @@ mod tests { .unwrap_or_else(|error| panic!("failed to build metrics exporter: {error}")); let provider = SdkMeterProvider::builder().with_reader(exporter).build(); let meter = provider.meter("switchyard"); - let stats = AlgorithmStats::new(registry, [STAGE_ROUTER.to_string()]); + let stats = StatsAccumulator::new(registry, [STAGE_ROUTER.to_string()]); meter .u64_counter("switchyard.stage_router.routing_decisions") @@ -246,6 +246,7 @@ mod tests { let snapshot = stats.snapshot(); let stage = snapshot + .algorithm_stats .stage_router .unwrap_or_else(|| panic!("stage-router stats missing")); let dimensions = &stage.routing_decisions["dimensions"]; @@ -259,7 +260,7 @@ mod tests { stats.reset(); assert_eq!( - stats.snapshot().stage_router, + stats.snapshot().algorithm_stats.stage_router, Some(StageRouterStatsSnapshot::default()) ); @@ -275,6 +276,7 @@ mod tests { ); let after_reset = stats .snapshot() + .algorithm_stats .stage_router .unwrap_or_else(|| panic!("stage-router stats missing after reset")); assert_eq!(after_reset.routing_decisions["override"].total, 1); From 92f42d91d5faf2d0ba2bc3b3a08f2ad7a48e235d Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 10 Aug 2026 16:40:42 -0700 Subject: [PATCH 6/9] refactor(server): simplify stage router stats state Signed-off-by: nachiketb --- crates/switchyard-server/src/lib.rs | 10 ++--- .../src/stats/accumulator.rs | 40 +++++++++++-------- .../switchyard-server/src/stats/algorithms.rs | 34 +--------------- .../src/stats/algorithms/stage_router.rs | 10 ++--- 4 files changed, 34 insertions(+), 60 deletions(-) diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 87cc6bcdd..a39417a2d 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -219,12 +219,10 @@ impl ServerState { return Err(ServerError::new("at least one algorithm route is required")); } let metrics = metrics::registry().map_err(ServerError::new)?; - let stats = StatsAccumulator::new( - metrics.clone(), - entries - .values() - .map(|entry| entry.algorithm.name().to_string()), - ); + let has_stage_router = entries + .values() + .any(|entry| entry.algorithm.name() == "stage_router"); + let stats = StatsAccumulator::new(metrics.clone(), has_stage_router); Ok(Self { routes: Arc::new(entries), metrics, diff --git a/crates/switchyard-server/src/stats/accumulator.rs b/crates/switchyard-server/src/stats/accumulator.rs index fec28ec8e..d2f11d762 100644 --- a/crates/switchyard-server/src/stats/accumulator.rs +++ b/crates/switchyard-server/src/stats/accumulator.rs @@ -3,14 +3,14 @@ //! Thread-safe stats accumulator and serializable snapshot schema. -use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; use parking_lot::{Mutex, MutexGuard}; use prometheus::Registry; use serde::Serialize; -use super::algorithms::{AlgorithmMetrics, AlgorithmStatsSnapshot}; +use super::algorithms::{AlgorithmStatsSnapshot, StageRouterCumulative}; use super::cache_eligibility::PrefixProbe; const MAX_LATENCY_SAMPLES: usize = 10_000; @@ -34,15 +34,18 @@ pub(crate) struct StatsAccumulator { impl Default for StatsAccumulator { fn default() -> Self { - Self::new(Registry::new(), std::iter::empty()) + Self::new(Registry::new(), false) } } impl StatsAccumulator { - /// Creates a stats store that projects metrics for the configured algorithms. - pub(crate) fn new(registry: Registry, configured: impl IntoIterator) -> Self { + /// Creates a stats store with optional stage-router metric projection. + pub(crate) fn new(registry: Registry, stage_router: bool) -> Self { Self { - inner: Arc::new(Mutex::new(StatsAccumulatorInner::new(registry, configured))), + inner: Arc::new(Mutex::new(StatsAccumulatorInner::new( + registry, + stage_router, + ))), } } @@ -153,13 +156,15 @@ struct StatsAccumulatorInner { classifier_requests: u64, classifier_errors: u64, algorithm_registry: Registry, - configured_algorithms: BTreeSet, - algorithm_baseline: AlgorithmMetrics, + stage_router_baseline: Option, } impl StatsAccumulatorInner { - fn new(registry: Registry, configured: impl IntoIterator) -> Self { - let algorithm_baseline = AlgorithmMetrics::collect(®istry); + fn new(registry: Registry, stage_router: bool) -> Self { + let stage_router_baseline = stage_router.then(|| { + let families = registry.gather(); + StageRouterCumulative::collect(&families) + }); Self { by_model: BTreeMap::new(), total_requests: 0, @@ -170,8 +175,7 @@ impl StatsAccumulatorInner { classifier_requests: 0, classifier_errors: 0, algorithm_registry: registry, - configured_algorithms: configured.into_iter().collect(), - algorithm_baseline, + stage_router_baseline, } } @@ -190,8 +194,10 @@ impl StatsAccumulatorInner { self.classifier_requests, self.classifier_errors, ); - let algorithm_stats = AlgorithmMetrics::collect(&self.algorithm_registry) - .snapshot(&self.algorithm_baseline, &self.configured_algorithms); + let stage_router = self.stage_router_baseline.as_ref().map(|baseline| { + let families = self.algorithm_registry.gather(); + StageRouterCumulative::collect(&families).delta(baseline) + }); StatsSnapshot { total_requests: self.total_requests, total_errors: self.total_errors, @@ -200,14 +206,14 @@ impl StatsAccumulatorInner { routing_overhead: self.routing_overhead.snapshot(), routing_fallbacks: self.routing_fallbacks, classifier, - algorithm_stats, + algorithm_stats: AlgorithmStatsSnapshot { stage_router }, } } fn reset(&mut self) { let registry = self.algorithm_registry.clone(); - let configured = std::mem::take(&mut self.configured_algorithms); - *self = Self::new(registry, configured); + let stage_router = self.stage_router_baseline.is_some(); + *self = Self::new(registry, stage_router); } } diff --git a/crates/switchyard-server/src/stats/algorithms.rs b/crates/switchyard-server/src/stats/algorithms.rs index eb351b3eb..e03dabff5 100644 --- a/crates/switchyard-server/src/stats/algorithms.rs +++ b/crates/switchyard-server/src/stats/algorithms.rs @@ -5,19 +5,10 @@ mod stage_router; -use std::collections::BTreeSet; - -use prometheus::Registry; use serde::Serialize; -use stage_router::{StageRouterCumulative, StageRouterStatsSnapshot}; - -pub(super) const STAGE_ROUTER: &str = "stage_router"; - -#[derive(Clone, Debug, Default)] -pub(super) struct AlgorithmMetrics { - stage_router: StageRouterCumulative, -} +pub(super) use stage_router::StageRouterCumulative; +use stage_router::StageRouterStatsSnapshot; /// Curated algorithm-specific data included in the JSON stats response. #[derive(Clone, Debug, Default, PartialEq, Serialize)] @@ -25,24 +16,3 @@ pub(crate) struct AlgorithmStatsSnapshot { #[serde(skip_serializing_if = "Option::is_none")] pub stage_router: Option, } - -impl AlgorithmMetrics { - pub(super) fn collect(registry: &Registry) -> Self { - let families = registry.gather(); - Self { - stage_router: StageRouterCumulative::collect(&families), - } - } - - pub(super) fn snapshot( - &self, - baseline: &Self, - configured: &BTreeSet, - ) -> AlgorithmStatsSnapshot { - AlgorithmStatsSnapshot { - stage_router: configured - .contains(STAGE_ROUTER) - .then(|| self.stage_router.delta(&baseline.stage_router)), - } - } -} diff --git a/crates/switchyard-server/src/stats/algorithms/stage_router.rs b/crates/switchyard-server/src/stats/algorithms/stage_router.rs index 65a71a9f9..34baf4d60 100644 --- a/crates/switchyard-server/src/stats/algorithms/stage_router.rs +++ b/crates/switchyard-server/src/stats/algorithms/stage_router.rs @@ -17,7 +17,7 @@ const EXPLORING_METRIC: &str = "switchyard_stage_router_exploring"; const PRODUCTION_INTENSITY_METRIC: &str = "switchyard_stage_router_production_intensity"; #[derive(Clone, Debug, Default)] -pub(super) struct StageRouterCumulative { +pub(in crate::stats) struct StageRouterCumulative { decisions: BTreeMap, score: HistogramTotal, confidence: HistogramTotal, @@ -79,7 +79,7 @@ pub(crate) struct MetricSummary { } impl StageRouterCumulative { - pub(super) fn collect(families: &[MetricFamily]) -> Self { + pub(in crate::stats) fn collect(families: &[MetricFamily]) -> Self { Self { decisions: collect_decisions(families), score: collect_histogram(families, SCORE_METRIC), @@ -91,7 +91,7 @@ impl StageRouterCumulative { } } - pub(super) fn delta(&self, baseline: &Self) -> StageRouterStatsSnapshot { + pub(in crate::stats) fn delta(&self, baseline: &Self) -> StageRouterStatsSnapshot { let mut routing_decisions: BTreeMap = BTreeMap::new(); for (key, current) in &self.decisions { let count = current.saturating_sub(*baseline.decisions.get(key).unwrap_or(&0)); @@ -200,7 +200,7 @@ mod tests { use prometheus::Registry; use super::*; - use crate::stats::{StatsAccumulator, algorithms::STAGE_ROUTER}; + use crate::stats::StatsAccumulator; #[test] fn stage_router_projection_preserves_decisions_scores_and_reset_baseline() { @@ -211,7 +211,7 @@ mod tests { .unwrap_or_else(|error| panic!("failed to build metrics exporter: {error}")); let provider = SdkMeterProvider::builder().with_reader(exporter).build(); let meter = provider.meter("switchyard"); - let stats = StatsAccumulator::new(registry, [STAGE_ROUTER.to_string()]); + let stats = StatsAccumulator::new(registry, true); meter .u64_counter("switchyard.stage_router.routing_decisions") From d2505779e2bc2d48a79400f50caeb1588b7ad9db Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 10 Aug 2026 16:46:14 -0700 Subject: [PATCH 7/9] refactor(server): isolate algorithm stats extensions Signed-off-by: nachiketb --- crates/switchyard-server/src/lib.rs | 8 +-- crates/switchyard-server/src/metrics.rs | 17 +------ crates/switchyard-server/src/stats.rs | 1 + .../src/stats/accumulator.rs | 46 ++++++++--------- .../switchyard-server/src/stats/algorithms.rs | 50 ++++++++++++++++++- .../src/stats/algorithms/stage_router.rs | 30 +++++++++-- 6 files changed, 101 insertions(+), 51 deletions(-) diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index a39417a2d..7ad32d81b 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -219,10 +219,10 @@ impl ServerState { return Err(ServerError::new("at least one algorithm route is required")); } let metrics = metrics::registry().map_err(ServerError::new)?; - let has_stage_router = entries - .values() - .any(|entry| entry.algorithm.name() == "stage_router"); - let stats = StatsAccumulator::new(metrics.clone(), has_stage_router); + let stats = StatsAccumulator::new( + metrics.clone(), + entries.values().map(|entry| entry.algorithm.name()), + ); Ok(Self { routes: Arc::new(entries), metrics, diff --git a/crates/switchyard-server/src/metrics.rs b/crates/switchyard-server/src/metrics.rs index 27a5ee12e..69ccc0a0b 100644 --- a/crates/switchyard-server/src/metrics.rs +++ b/crates/switchyard-server/src/metrics.rs @@ -18,17 +18,6 @@ pub(crate) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8" const ROUTING_OVERHEAD_BUCKETS_MS: &[f64] = &[ 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, 2500.0, 5000.0, ]; -const STAGE_ROUTER_SCORE_BUCKETS: &[f64] = &[-1.0, -0.75, -0.5, -0.25, 0.0, 0.25, 0.5, 0.75, 1.0]; -const STAGE_ROUTER_UNIT_BUCKETS: &[f64] = &[0.0, 0.1, 0.25, 0.5, 0.75, 0.9, 1.0]; - -const STAGE_ROUTER_SCORE_METRIC: &str = "switchyard.stage_router.score"; -const STAGE_ROUTER_UNIT_METRICS: &[&str] = &[ - "switchyard.stage_router.confidence", - "switchyard.stage_router.severity", - "switchyard.stage_router.spinning", - "switchyard.stage_router.exploring", - "switchyard.stage_router.production_intensity", -]; struct Metrics { registry: Registry, @@ -84,12 +73,8 @@ pub(crate) fn flush() { fn histogram_buckets(instrument: &Instrument) -> Option { let boundaries = if instrument.name() == "switchyard.routing_overhead_ms" { ROUTING_OVERHEAD_BUCKETS_MS - } else if instrument.name() == STAGE_ROUTER_SCORE_METRIC { - STAGE_ROUTER_SCORE_BUCKETS - } else if STAGE_ROUTER_UNIT_METRICS.contains(&instrument.name()) { - STAGE_ROUTER_UNIT_BUCKETS } else { - return None; + crate::stats::algorithm_histogram_buckets(instrument.name())? }; Stream::builder() .with_aggregation(Aggregation::ExplicitBucketHistogram { diff --git a/crates/switchyard-server/src/stats.rs b/crates/switchyard-server/src/stats.rs index 1a419ffd2..822145759 100644 --- a/crates/switchyard-server/src/stats.rs +++ b/crates/switchyard-server/src/stats.rs @@ -8,4 +8,5 @@ mod algorithms; mod cache_eligibility; pub(crate) use accumulator::{StatsAccumulator, StatsSnapshot, TokenUsage}; +pub(crate) use algorithms::histogram_buckets as algorithm_histogram_buckets; pub(crate) use cache_eligibility::{prefix_probe, tracking_enabled_from_env}; diff --git a/crates/switchyard-server/src/stats/accumulator.rs b/crates/switchyard-server/src/stats/accumulator.rs index d2f11d762..56d2df552 100644 --- a/crates/switchyard-server/src/stats/accumulator.rs +++ b/crates/switchyard-server/src/stats/accumulator.rs @@ -10,7 +10,7 @@ use parking_lot::{Mutex, MutexGuard}; use prometheus::Registry; use serde::Serialize; -use super::algorithms::{AlgorithmStatsSnapshot, StageRouterCumulative}; +use super::algorithms::{AlgorithmStats, AlgorithmStatsSnapshot}; use super::cache_eligibility::PrefixProbe; const MAX_LATENCY_SAMPLES: usize = 10_000; @@ -34,18 +34,18 @@ pub(crate) struct StatsAccumulator { impl Default for StatsAccumulator { fn default() -> Self { - Self::new(Registry::new(), false) + Self::new(Registry::new(), std::iter::empty()) } } impl StatsAccumulator { - /// Creates a stats store with optional stage-router metric projection. - pub(crate) fn new(registry: Registry, stage_router: bool) -> Self { + /// Creates a stats store for the supplied algorithm names. + pub(crate) fn new<'a>( + registry: Registry, + algorithms: impl IntoIterator, + ) -> Self { Self { - inner: Arc::new(Mutex::new(StatsAccumulatorInner::new( - registry, - stage_router, - ))), + inner: Arc::new(Mutex::new(StatsAccumulatorInner::new(registry, algorithms))), } } @@ -155,16 +155,11 @@ struct StatsAccumulatorInner { by_classifier: BTreeMap, classifier_requests: u64, classifier_errors: u64, - algorithm_registry: Registry, - stage_router_baseline: Option, + algorithm_stats: AlgorithmStats, } impl StatsAccumulatorInner { - fn new(registry: Registry, stage_router: bool) -> Self { - let stage_router_baseline = stage_router.then(|| { - let families = registry.gather(); - StageRouterCumulative::collect(&families) - }); + fn new<'a>(registry: Registry, algorithms: impl IntoIterator) -> Self { Self { by_model: BTreeMap::new(), total_requests: 0, @@ -174,8 +169,7 @@ impl StatsAccumulatorInner { by_classifier: BTreeMap::new(), classifier_requests: 0, classifier_errors: 0, - algorithm_registry: registry, - stage_router_baseline, + algorithm_stats: AlgorithmStats::new(registry, algorithms), } } @@ -194,10 +188,6 @@ impl StatsAccumulatorInner { self.classifier_requests, self.classifier_errors, ); - let stage_router = self.stage_router_baseline.as_ref().map(|baseline| { - let families = self.algorithm_registry.gather(); - StageRouterCumulative::collect(&families).delta(baseline) - }); StatsSnapshot { total_requests: self.total_requests, total_errors: self.total_errors, @@ -206,14 +196,20 @@ impl StatsAccumulatorInner { routing_overhead: self.routing_overhead.snapshot(), routing_fallbacks: self.routing_fallbacks, classifier, - algorithm_stats: AlgorithmStatsSnapshot { stage_router }, + algorithm_stats: self.algorithm_stats.snapshot(), } } fn reset(&mut self) { - let registry = self.algorithm_registry.clone(); - let stage_router = self.stage_router_baseline.is_some(); - *self = Self::new(registry, stage_router); + self.by_model.clear(); + self.total_requests = 0; + self.total_errors = 0; + self.routing_overhead = LatencyHistogram::default(); + self.routing_fallbacks = RoutingFallbackStats::default(); + self.by_classifier.clear(); + self.classifier_requests = 0; + self.classifier_errors = 0; + self.algorithm_stats.reset(); } } diff --git a/crates/switchyard-server/src/stats/algorithms.rs b/crates/switchyard-server/src/stats/algorithms.rs index e03dabff5..c50e0721a 100644 --- a/crates/switchyard-server/src/stats/algorithms.rs +++ b/crates/switchyard-server/src/stats/algorithms.rs @@ -5,10 +5,20 @@ mod stage_router; +use std::collections::HashSet; + +use prometheus::Registry; use serde::Serialize; -pub(super) use stage_router::StageRouterCumulative; -use stage_router::StageRouterStatsSnapshot; +use stage_router::{StageRouterCumulative, StageRouterStatsSnapshot}; + +const STAGE_ROUTER: &str = "stage_router"; + +/// Owns algorithm metric baselines behind the generic server stats interface. +pub(super) struct AlgorithmStats { + registry: Registry, + stage_router_baseline: Option, +} /// Curated algorithm-specific data included in the JSON stats response. #[derive(Clone, Debug, Default, PartialEq, Serialize)] @@ -16,3 +26,39 @@ pub(crate) struct AlgorithmStatsSnapshot { #[serde(skip_serializing_if = "Option::is_none")] pub stage_router: Option, } + +impl AlgorithmStats { + pub(super) fn new<'a>( + registry: Registry, + algorithms: impl IntoIterator, + ) -> Self { + let algorithms: HashSet<_> = algorithms.into_iter().collect(); + let families = registry.gather(); + Self { + stage_router_baseline: algorithms + .contains(STAGE_ROUTER) + .then(|| StageRouterCumulative::collect(&families)), + registry, + } + } + + pub(super) fn snapshot(&self) -> AlgorithmStatsSnapshot { + let families = self.registry.gather(); + AlgorithmStatsSnapshot { + stage_router: self + .stage_router_baseline + .as_ref() + .map(|baseline| StageRouterCumulative::collect(&families).delta(baseline)), + } + } + + pub(super) fn reset(&mut self) { + if let Some(baseline) = &mut self.stage_router_baseline { + *baseline = StageRouterCumulative::collect(&self.registry.gather()); + } + } +} + +pub(crate) fn histogram_buckets(metric: &str) -> Option<&'static [f64]> { + stage_router::histogram_buckets(metric) +} diff --git a/crates/switchyard-server/src/stats/algorithms/stage_router.rs b/crates/switchyard-server/src/stats/algorithms/stage_router.rs index 34baf4d60..795f7c859 100644 --- a/crates/switchyard-server/src/stats/algorithms/stage_router.rs +++ b/crates/switchyard-server/src/stats/algorithms/stage_router.rs @@ -16,8 +16,20 @@ const SPINNING_METRIC: &str = "switchyard_stage_router_spinning"; const EXPLORING_METRIC: &str = "switchyard_stage_router_exploring"; const PRODUCTION_INTENSITY_METRIC: &str = "switchyard_stage_router_production_intensity"; +const SCORE_INSTRUMENT: &str = "switchyard.stage_router.score"; +const UNIT_INSTRUMENTS: &[&str] = &[ + "switchyard.stage_router.confidence", + "switchyard.stage_router.severity", + "switchyard.stage_router.spinning", + "switchyard.stage_router.exploring", + "switchyard.stage_router.production_intensity", +]; + +const SCORE_BUCKETS: &[f64] = &[-1.0, -0.75, -0.5, -0.25, 0.0, 0.25, 0.5, 0.75, 1.0]; +const UNIT_BUCKETS: &[f64] = &[0.0, 0.1, 0.25, 0.5, 0.75, 0.9, 1.0]; + #[derive(Clone, Debug, Default)] -pub(in crate::stats) struct StageRouterCumulative { +pub(super) struct StageRouterCumulative { decisions: BTreeMap, score: HistogramTotal, confidence: HistogramTotal, @@ -79,7 +91,7 @@ pub(crate) struct MetricSummary { } impl StageRouterCumulative { - pub(in crate::stats) fn collect(families: &[MetricFamily]) -> Self { + pub(super) fn collect(families: &[MetricFamily]) -> Self { Self { decisions: collect_decisions(families), score: collect_histogram(families, SCORE_METRIC), @@ -91,7 +103,7 @@ impl StageRouterCumulative { } } - pub(in crate::stats) fn delta(&self, baseline: &Self) -> StageRouterStatsSnapshot { + pub(super) fn delta(&self, baseline: &Self) -> StageRouterStatsSnapshot { let mut routing_decisions: BTreeMap = BTreeMap::new(); for (key, current) in &self.decisions { let count = current.saturating_sub(*baseline.decisions.get(key).unwrap_or(&0)); @@ -192,6 +204,16 @@ fn round4(value: f64) -> f64 { if rounded == 0.0 { 0.0 } else { rounded } } +pub(super) fn histogram_buckets(metric: &str) -> Option<&'static [f64]> { + if metric == SCORE_INSTRUMENT { + Some(SCORE_BUCKETS) + } else if UNIT_INSTRUMENTS.contains(&metric) { + Some(UNIT_BUCKETS) + } else { + None + } +} + #[cfg(test)] mod tests { use opentelemetry::KeyValue; @@ -211,7 +233,7 @@ mod tests { .unwrap_or_else(|error| panic!("failed to build metrics exporter: {error}")); let provider = SdkMeterProvider::builder().with_reader(exporter).build(); let meter = provider.meter("switchyard"); - let stats = StatsAccumulator::new(registry, true); + let stats = StatsAccumulator::new(registry, ["stage_router"]); meter .u64_counter("switchyard.stage_router.routing_decisions") From f6359ae3e5e38ccef0ffbd77a79d1f16e9edd272 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Tue, 11 Aug 2026 09:52:13 -0700 Subject: [PATCH 8/9] refactor(metrics): colocate stage histogram boundaries Signed-off-by: nachiketb --- crates/libsy/src/algorithms/util/stage.rs | 29 ++++++++++++++----- crates/switchyard-server/src/metrics.rs | 14 ++++----- crates/switchyard-server/src/stats.rs | 1 - .../switchyard-server/src/stats/algorithms.rs | 4 --- .../src/stats/algorithms/stage_router.rs | 22 -------------- 5 files changed, 27 insertions(+), 43 deletions(-) diff --git a/crates/libsy/src/algorithms/util/stage.rs b/crates/libsy/src/algorithms/util/stage.rs index 32fcf8f52..1976f6b94 100644 --- a/crates/libsy/src/algorithms/util/stage.rs +++ b/crates/libsy/src/algorithms/util/stage.rs @@ -62,6 +62,11 @@ const EXPLORING_METRIC: &str = "switchyard.stage_router.exploring"; /// Distribution of production-oriented tool activity. const PRODUCTION_INTENSITY_METRIC: &str = "switchyard.stage_router.production_intensity"; +// Histogram boundaries live with the instruments so every host exports the +// same stage-router distributions without duplicating algorithm knowledge. +const SCORE_BUCKETS: &[f64] = &[-1.0, -0.75, -0.5, -0.25, 0.0, 0.25, 0.5, 0.75, 1.0]; +const UNIT_BUCKETS: &[f64] = &[0.0, 0.1, 0.25, 0.5, 0.75, 0.9, 1.0]; + /// The two tiers a turn can route to. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum Tier { @@ -288,15 +293,23 @@ fn record_score_metrics(signal: &ToolSignals, outcome: &PickOutcome) { }; let dimensions = dimensions_from_signal(signal); let meter = meter(); - for (name, value) in [ - (SCORE_METRIC, score), - (CONFIDENCE_METRIC, confidence), - (SEVERITY_METRIC, dimensions.severity), - (SPINNING_METRIC, dimensions.spinning), - (EXPLORING_METRIC, dimensions.exploring), - (PRODUCTION_INTENSITY_METRIC, dimensions.production_intensity), + for (name, value, boundaries) in [ + (SCORE_METRIC, score, SCORE_BUCKETS), + (CONFIDENCE_METRIC, confidence, UNIT_BUCKETS), + (SEVERITY_METRIC, dimensions.severity, UNIT_BUCKETS), + (SPINNING_METRIC, dimensions.spinning, UNIT_BUCKETS), + (EXPLORING_METRIC, dimensions.exploring, UNIT_BUCKETS), + ( + PRODUCTION_INTENSITY_METRIC, + dimensions.production_intensity, + UNIT_BUCKETS, + ), ] { - meter.f64_histogram(name).build().record(value, &[]); + meter + .f64_histogram(name) + .with_boundaries(boundaries.to_vec()) + .build() + .record(value, &[]); } } diff --git a/crates/switchyard-server/src/metrics.rs b/crates/switchyard-server/src/metrics.rs index 69ccc0a0b..2469e0251 100644 --- a/crates/switchyard-server/src/metrics.rs +++ b/crates/switchyard-server/src/metrics.rs @@ -42,7 +42,7 @@ fn initialize() -> Result { .map_err(|error| format!("failed to initialize Prometheus metrics: {error}"))?; let mut builder = SdkMeterProvider::builder() .with_reader(exporter) - .with_view(histogram_buckets) + .with_view(routing_overhead_buckets) .with_resource(crate::observability::resource()); if crate::observability::otlp_enabled("METRICS") { let exporter = opentelemetry_otlp::MetricExporter::builder() @@ -70,15 +70,13 @@ pub(crate) fn flush() { } } -fn histogram_buckets(instrument: &Instrument) -> Option { - let boundaries = if instrument.name() == "switchyard.routing_overhead_ms" { - ROUTING_OVERHEAD_BUCKETS_MS - } else { - crate::stats::algorithm_histogram_buckets(instrument.name())? - }; +fn routing_overhead_buckets(instrument: &Instrument) -> Option { + if instrument.name() != "switchyard.routing_overhead_ms" { + return None; + } Stream::builder() .with_aggregation(Aggregation::ExplicitBucketHistogram { - boundaries: boundaries.to_vec(), + boundaries: ROUTING_OVERHEAD_BUCKETS_MS.to_vec(), // Cumulative min/max cover the whole process, so they aren't useful. record_min_max: false, }) diff --git a/crates/switchyard-server/src/stats.rs b/crates/switchyard-server/src/stats.rs index 822145759..1a419ffd2 100644 --- a/crates/switchyard-server/src/stats.rs +++ b/crates/switchyard-server/src/stats.rs @@ -8,5 +8,4 @@ mod algorithms; mod cache_eligibility; pub(crate) use accumulator::{StatsAccumulator, StatsSnapshot, TokenUsage}; -pub(crate) use algorithms::histogram_buckets as algorithm_histogram_buckets; pub(crate) use cache_eligibility::{prefix_probe, tracking_enabled_from_env}; diff --git a/crates/switchyard-server/src/stats/algorithms.rs b/crates/switchyard-server/src/stats/algorithms.rs index c50e0721a..27bfc0ff6 100644 --- a/crates/switchyard-server/src/stats/algorithms.rs +++ b/crates/switchyard-server/src/stats/algorithms.rs @@ -58,7 +58,3 @@ impl AlgorithmStats { } } } - -pub(crate) fn histogram_buckets(metric: &str) -> Option<&'static [f64]> { - stage_router::histogram_buckets(metric) -} diff --git a/crates/switchyard-server/src/stats/algorithms/stage_router.rs b/crates/switchyard-server/src/stats/algorithms/stage_router.rs index 795f7c859..9383d1576 100644 --- a/crates/switchyard-server/src/stats/algorithms/stage_router.rs +++ b/crates/switchyard-server/src/stats/algorithms/stage_router.rs @@ -16,18 +16,6 @@ const SPINNING_METRIC: &str = "switchyard_stage_router_spinning"; const EXPLORING_METRIC: &str = "switchyard_stage_router_exploring"; const PRODUCTION_INTENSITY_METRIC: &str = "switchyard_stage_router_production_intensity"; -const SCORE_INSTRUMENT: &str = "switchyard.stage_router.score"; -const UNIT_INSTRUMENTS: &[&str] = &[ - "switchyard.stage_router.confidence", - "switchyard.stage_router.severity", - "switchyard.stage_router.spinning", - "switchyard.stage_router.exploring", - "switchyard.stage_router.production_intensity", -]; - -const SCORE_BUCKETS: &[f64] = &[-1.0, -0.75, -0.5, -0.25, 0.0, 0.25, 0.5, 0.75, 1.0]; -const UNIT_BUCKETS: &[f64] = &[0.0, 0.1, 0.25, 0.5, 0.75, 0.9, 1.0]; - #[derive(Clone, Debug, Default)] pub(super) struct StageRouterCumulative { decisions: BTreeMap, @@ -204,16 +192,6 @@ fn round4(value: f64) -> f64 { if rounded == 0.0 { 0.0 } else { rounded } } -pub(super) fn histogram_buckets(metric: &str) -> Option<&'static [f64]> { - if metric == SCORE_INSTRUMENT { - Some(SCORE_BUCKETS) - } else if UNIT_INSTRUMENTS.contains(&metric) { - Some(UNIT_BUCKETS) - } else { - None - } -} - #[cfg(test)] mod tests { use opentelemetry::KeyValue; From a66ee4f5d97cbc93ec26c8df7a1d18dae02eae3d Mon Sep 17 00:00:00 2001 From: nachiketb Date: Tue, 11 Aug 2026 10:16:38 -0700 Subject: [PATCH 9/9] refactor(stats): simplify stage metric summaries Signed-off-by: nachiketb --- .../src/stats/algorithms/stage_router.rs | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/crates/switchyard-server/src/stats/algorithms/stage_router.rs b/crates/switchyard-server/src/stats/algorithms/stage_router.rs index 9383d1576..9b3fc113f 100644 --- a/crates/switchyard-server/src/stats/algorithms/stage_router.rs +++ b/crates/switchyard-server/src/stats/algorithms/stage_router.rs @@ -70,12 +70,11 @@ pub(crate) struct DimensionStatsSnapshot { pub production_intensity: MetricSummary, } -/// Exact count, sum, and average since the last JSON stats reset. +/// Exact count and mean since the last JSON stats reset. #[derive(Clone, Copy, Debug, Default, PartialEq, Serialize)] pub(crate) struct MetricSummary { pub count: u64, - pub sum: f64, - pub avg: f64, + pub mean: f64, } impl StageRouterCumulative { @@ -129,8 +128,7 @@ impl HistogramTotal { let sum = round4(self.sum - baseline.sum); MetricSummary { count, - sum, - avg: round4(sum / count as f64), + mean: round4(sum / count as f64), } } } @@ -254,9 +252,8 @@ mod tests { assert_eq!(dimensions.targets["model/efficient"], 2); assert_eq!(dimensions.targets["model/capable"], 1); assert_eq!(stage.scoring.score.count, 2); - assert_eq!(stage.scoring.score.sum, 0.25); - assert_eq!(stage.scoring.score.avg, 0.125); - assert_eq!(stage.scoring.confidence.avg, 0.75); + assert_eq!(stage.scoring.score.mean, 0.125); + assert_eq!(stage.scoring.confidence.mean, 0.75); stats.reset(); assert_eq!(