diff --git a/crates/libsy/src/algorithms/stage.rs b/crates/libsy/src/algorithms/stage.rs index a2612d69..330752e2 100644 --- a/crates/libsy/src/algorithms/stage.rs +++ b/crates/libsy/src/algorithms/stage.rs @@ -232,7 +232,9 @@ mod tests { use super::*; use crate::algorithms::util::stage::DECISION_SOURCE_KEY; - use crate::core::algorithm::{Algorithm, LlmTarget}; + use crate::core::algorithm::{ + Algorithm, AlgorithmMetricValue, LlmTarget, RunObservation, RunObserver, + }; use crate::core::classifier::Score; use crate::core::state::StateValue; use switchyard_protocol::{ @@ -509,9 +511,14 @@ mod tests { let client = Arc::new(RecordingClient::default()); let router = recording_router(client.clone(), config_with_notes())?; let ctx = Context::default(); + let observations = Arc::new(Mutex::new(Vec::new())); + let observed = observations.clone(); + let observer: RunObserver = Arc::new(move |observation| observed.lock().push(observation)); router.clone().run(ctx.clone(), turn_request(false)).await?; - router.run(ctx, turn_request(true)).await?; + router + .run_observed(ctx, turn_request(true), Some(observer)) + .await?; let calls = client.routed(); assert_eq!(calls[0].target, "weak"); @@ -529,6 +536,49 @@ mod tests { "escalating turn should carry the note last: {:?}", calls[1].messages ); + + let metrics = observations + .lock() + .iter() + .filter_map(|observation| match observation { + RunObservation::AlgorithmMetric(metric) => Some(metric.clone()), + _ => None, + }) + .collect::>(); + assert_eq!(metrics.len(), 6); + assert!(metrics.iter().all(|metric| { + metric.algorithm == STAGE_ROUTER + && matches!(metric.value, AlgorithmMetricValue::Histogram(_)) + })); + assert_eq!( + metrics + .iter() + .filter(|metric| metric.name == "switchyard.stage_router.score") + .count(), + 1 + ); + assert_eq!( + metrics + .iter() + .filter(|metric| metric.name == "switchyard.stage_router.confidence") + .count(), + 1 + ); + let mut dimensions = metrics + .iter() + .filter(|metric| metric.name == "switchyard.stage_router.dimension") + .filter_map(|metric| { + metric + .attributes + .first() + .map(|attribute| attribute.value.clone()) + }) + .collect::>(); + dimensions.sort(); + assert_eq!( + dimensions, + ["exploring", "production_intensity", "severity", "spinning"] + ); Ok(()) } diff --git a/crates/libsy/src/algorithms/util/llm_judge.rs b/crates/libsy/src/algorithms/util/llm_judge.rs index 90430e5a..8c813d9c 100644 --- a/crates/libsy/src/algorithms/util/llm_judge.rs +++ b/crates/libsy/src/algorithms/util/llm_judge.rs @@ -18,7 +18,7 @@ use switchyard_protocol::{ }; use super::classifier_contract::ClassifierContract; -use crate::core::algorithm::{Driver, LlmTarget}; +use crate::core::algorithm::{Driver, LlmTarget, MetricAttribute}; use crate::core::classifier::{Classification, Classifier}; use crate::core::state::State; use crate::{LibsyError, Result}; @@ -235,23 +235,32 @@ where }), ) .await - .inspect_err(|error| report_fail_open(judge_model, error, libsy_error_reason(error))) + .inspect_err(|error| { + report_fail_open(driver, judge_model, error, libsy_error_reason(error)) + }) .ok()?; let aggregate = response .llm_response .into_agg() .await - .inspect_err(|error| report_fail_open(judge_model, error, client_error_reason(error))) + .inspect_err(|error| { + report_fail_open(driver, judge_model, error, client_error_reason(error)) + }) .ok()?; self.judge .parse(&aggregate) - .inspect_err(|error| report_fail_open(judge_model, error, "parse_error")) + .inspect_err(|error| report_fail_open(driver, judge_model, error, "parse_error")) .ok() } } /// Logs and counts a judge failure with a bounded label that excludes message content. -fn report_fail_open(judge_model: &str, error: &dyn std::fmt::Display, reason: &'static str) { +fn report_fail_open( + driver: &Driver, + judge_model: &str, + error: &dyn std::fmt::Display, + reason: &'static str, +) { tracing::warn!( target: "libsy", judge_model, @@ -259,7 +268,14 @@ fn report_fail_open(judge_model: &str, error: &dyn std::fmt::Display, reason: &' error = %error, "judge verdict unavailable; routing without one" ); - crate::observability::record_classifier_fail_open(judge_model, reason); + driver.record_counter( + "switchyard.classifier_fail_open", + 1, + [ + MetricAttribute::new("judge_model", judge_model), + MetricAttribute::new("reason", reason), + ], + ); } /// Returns a bounded reason for a judge call that failed at the libsy layer. diff --git a/crates/libsy/src/algorithms/util/stage.rs b/crates/libsy/src/algorithms/util/stage.rs index 24e3dede..89a673c3 100644 --- a/crates/libsy/src/algorithms/util/stage.rs +++ b/crates/libsy/src/algorithms/util/stage.rs @@ -23,7 +23,7 @@ use serde::Deserialize; use super::prompts; use super::tool_signals::ToolSignals; use crate::Result; -use crate::core::algorithm::Driver; +use crate::core::algorithm::{Driver, MetricAttribute}; use crate::core::classifier::{Classification, Classifier, Score}; use crate::core::state::{State, StateValue}; use switchyard_protocol::Request; @@ -43,6 +43,9 @@ const HARD_SEVERITY: f64 = 0.7; const SIGNAL_UNIT: f64 = 0.10; /// Critical severity forces the capable tier regardless of the scorer. const SEVERITY_CRITICAL: f32 = 1.0; +const SCORE_METRIC: &str = "switchyard.stage_router.score"; +const CONFIDENCE_METRIC: &str = "switchyard.stage_router.confidence"; +const DIMENSION_METRIC: &str = "switchyard.stage_router.dimension"; /// The two tiers a turn can route to. #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -340,6 +343,37 @@ pub fn pick_tier(signal: &ToolSignals, mode: PickerMode, confidence_threshold: f } } +/// Records the router's scorer output and four bounded input dimensions. +fn record_metrics(driver: Option<&Driver>, signal: &ToolSignals, outcome: &PickOutcome) { + let Some(driver) = driver else { + return; + }; + let (score, confidence) = match outcome { + PickOutcome::Resolved { + score, confidence, .. + } => (*score, confidence.unwrap_or(score.abs())), + PickOutcome::ConsultClassifier { + score, confidence, .. + } => (*score, *confidence), + }; + driver.record_histogram(SCORE_METRIC, score, []); + driver.record_histogram(CONFIDENCE_METRIC, confidence, []); + + let dimensions = dimensions_from_signal(signal); + for (name, value) in [ + ("severity", dimensions.severity), + ("spinning", dimensions.spinning), + ("exploring", dimensions.exploring), + ("production_intensity", dimensions.production_intensity), + ] { + driver.record_histogram( + DIMENSION_METRIC, + value, + [MetricAttribute::new("dimension", name)], + ); + } +} + /// Build a resolved outcome (a decision made without the classifier). fn resolved( tier: Tier, @@ -484,7 +518,7 @@ impl Classifier for StageClassifier { &self, state: &mut State, request: &mut Request, - _driver: Option<&Driver>, + driver: Option<&Driver>, ) -> Result<(Classification, Option)> { let tool_signals = &state.tool_signals; let Some(signal) = tool_signals else { @@ -494,6 +528,7 @@ impl Classifier for StageClassifier { }; let outcome = pick_tier(signal, self.mode, self.confidence_threshold); + record_metrics(driver, signal, &outcome); match outcome { PickOutcome::Resolved { tier, diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs index bd6212ca..d0ca6776 100644 --- a/crates/libsy/src/core/algorithm.rs +++ b/crates/libsy/src/core/algorithm.rs @@ -53,6 +53,47 @@ pub struct LlmCallObservation { pub usage: Option, } +/// One bounded attribute attached to an algorithm-defined metric. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct MetricAttribute { + /// Stable, low-cardinality attribute name. + pub key: &'static str, + /// Stable, low-cardinality attribute value. + pub value: String, +} + +impl MetricAttribute { + /// Build an attribute from a static key and owned or borrowed value. + pub fn new(key: &'static str, value: impl Into) -> Self { + Self { + key, + value: value.into(), + } + } +} + +/// The value and aggregation kind of an algorithm-defined metric. +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum AlgorithmMetricValue { + /// A non-negative delta added to a cumulative counter. + Counter(u64), + /// One sample recorded in a histogram. + Histogram(f64), +} + +/// One algorithm-defined metric emitted to OpenTelemetry and the run observer. +#[derive(Clone, Debug, PartialEq)] +pub struct AlgorithmMetricObservation { + /// Algorithm that emitted the metric. + pub algorithm: String, + /// Stable OpenTelemetry metric name. One name must always use the same value kind. + pub name: &'static str, + /// Counter delta or histogram sample. + pub value: AlgorithmMetricValue, + /// Bounded dimensions for grouping the metric. + pub attributes: Vec, +} + /// One request-scoped observation emitted by the algorithm runner. #[derive(Clone, Debug)] pub enum RunObservation { @@ -60,6 +101,8 @@ pub enum RunObservation { LlmCall(LlmCallObservation), /// Routing time recorded by the `switchyard.routing_overhead_ms` metric. RoutingOverhead(Duration), + /// An algorithm-defined counter delta or histogram sample. + AlgorithmMetric(AlgorithmMetricObservation), } /// Request-scoped callback for algorithm-run observations. @@ -152,6 +195,7 @@ impl CallLlmRequest { #[derive(Clone)] pub struct Driver { driver: TypeErasedDriver, + algorithm: Arc, // How long the call that served this run took. We need this to calculate routing overhead. routed_call: Arc>>, observer: Option, @@ -161,12 +205,13 @@ impl Driver { /// Build an empty driver with its step channel ready. Created per call by /// [`run_stream`](Algorithm::run_stream). pub(crate) fn new() -> Self { - Self::with_observer(None) + Self::with_observer("", None) } - fn with_observer(observer: Option) -> Self { + fn with_observer(algorithm: &str, observer: Option) -> Self { Self { driver: TypeErasedDriver::new(), + algorithm: Arc::from(algorithm), routed_call: Arc::new(Mutex::new(None)), observer, } @@ -185,6 +230,64 @@ impl Driver { } } + /// Add a delta to an algorithm-defined OpenTelemetry counter and report the + /// same event to this run's observer. Names and attributes must be stable and + /// low-cardinality; never include request or session data. + pub fn record_counter( + &self, + name: &'static str, + delta: u64, + attributes: impl IntoIterator, + ) { + self.record_algorithm_metric( + name, + AlgorithmMetricValue::Counter(delta), + attributes.into_iter().collect(), + ); + } + + /// Record one sample in an algorithm-defined OpenTelemetry histogram and + /// report the same event to this run's observer. Non-finite samples are dropped. + pub fn record_histogram( + &self, + name: &'static str, + sample: f64, + attributes: impl IntoIterator, + ) { + if !sample.is_finite() { + tracing::debug!( + metric = name, + sample, + "dropping non-finite histogram sample" + ); + return; + } + self.record_algorithm_metric( + name, + AlgorithmMetricValue::Histogram(sample), + attributes.into_iter().collect(), + ); + } + + fn record_algorithm_metric( + &self, + name: &'static str, + value: AlgorithmMetricValue, + mut attributes: Vec, + ) { + attributes.retain(|attribute| attribute.key != "algorithm"); + let observation = AlgorithmMetricObservation { + algorithm: self.algorithm.to_string(), + name, + value, + attributes, + }; + observability::record_algorithm_metric(&observation); + if let Some(observer) = &self.observer { + observer(RunObservation::AlgorithmMetric(observation)); + } + } + /// Offload a model call: publish `routed` as a [`Step::CallLlm`] and await the /// consumer's [`Response`]. The call's context travels inside /// [`routed.ctx`](RoutedRequest::ctx). Errors if the stream is closed or the call failed. @@ -626,7 +729,8 @@ pub trait Algorithm: Send + Sync + 'static { /// emitted as an `Err` item. Dropping the stream aborts the spawned algorithm task. /// /// Every invocation owns a separate [`Driver`]. `observer`, when present, receives - /// each completed model call and, after a successful routed run, its routing overhead. + /// completed model calls, algorithm-defined metrics, and, after a successful routed + /// run, its routing overhead. fn run_stream( self: Arc, ctx: Context, @@ -640,7 +744,7 @@ pub trait Algorithm: Send + Sync + 'static { observability::ALGORITHM_KEY.to_string(), self.name().to_string(), ); - let driver = Driver::with_observer(observer); + let driver = Driver::with_observer(self.name(), observer); let task_driver = driver.clone(); let task_ctx = ctx.clone(); let stream = task_driver.stream(); @@ -695,7 +799,7 @@ pub trait Algorithm: Send + Sync + 'static { self.run_observed(ctx, request, None).await } - /// Process a request to completion while reporting each model call to `observer`. + /// Process a request to completion while reporting run observations to `observer`. async fn run_observed( self: Arc, ctx: Context, diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index 37f6cb22..9485f820 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -6,8 +6,9 @@ mod core; pub use core::algorithm::{ - Algorithm, CallLlmRequest, Driver, LlmCallObservation, LlmTarget, LlmTargetSet, RoutedRequest, - RunObservation, RunObserver, Step, StepStream, + Algorithm, AlgorithmMetricObservation, AlgorithmMetricValue, CallLlmRequest, Driver, + LlmCallObservation, LlmTarget, LlmTargetSet, MetricAttribute, RoutedRequest, RunObservation, + RunObserver, Step, StepStream, }; pub use core::classifier::{Classification, Classifier, Score}; pub use core::processor::{Event, Processor}; diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs index bf3f1f1f..21e97e2f 100644 --- a/crates/libsy/src/observability.rs +++ b/crates/libsy/src/observability.rs @@ -2,11 +2,12 @@ // SPDX-License-Identifier: Apache-2.0 //! OpenTelemetry metrics plus `tracing` spans and structured logs for the -//! algorithm layer. +//! algorithm layer, including algorithm-defined metrics emitted through the +//! [`Driver`](crate::Driver). //! -//! The crate's provided run methods call these helpers around the [`Decision`] -//! hook and the offload boundary, so every algorithm is instrumented from the -//! outside and carries no telemetry code of its own. Metrics record through the +//! The crate's provided run methods instrument the [`Decision`] hook and offload +//! boundary from the outside. Algorithms emit domain-specific counters and +//! histograms through the [`Driver`](crate::Driver). Metrics record through the //! OpenTelemetry **global** meter provider under the `switchyard` scope — the host //! installs an SDK provider and exporters; with none installed, recording is a //! no-op. Spans and logs use the `tracing` facade (the async-native surface the @@ -21,9 +22,10 @@ //! Instrument names use the OTel dotted form with the unit baked into the name //! (`switchyard.run_duration_ms`), matching the switchyard metric surface; a //! Prometheus exporter sanitizes them to `switchyard_run_duration_ms`. Attribute -//! cardinality is bounded: `algorithm` and `selected_model` are small -//! configured sets and `outcome` is `ok`/`error`. Nothing per-request becomes a -//! metric attribute — correlation ids ride on the `libsy.run` span instead. +//! cardinality is bounded: `algorithm` and `selected_model` are small configured +//! sets and `outcome` is `ok`/`error`. Algorithm-defined metric attributes follow +//! the same contract. Nothing per-request becomes a metric attribute — correlation +//! ids ride on the `libsy.run` span instead. //! //! Instruments are resolved from the global provider on every record (an //! instrument-cache lookup inside the SDK) so recording follows a meter @@ -45,6 +47,7 @@ use switchyard_protocol::StopReason; use tracing::Span; use tracing_opentelemetry::OpenTelemetrySpanExt; +use crate::core::algorithm::{AlgorithmMetricObservation, AlgorithmMetricValue}; use crate::{Driver, LibsyError, Result}; use switchyard_protocol::{ AggLlmResponse, Context, Decision, LlmClientError, LlmRequest, LlmResponse, LlmResponseChunk, @@ -513,18 +516,28 @@ fn record_routing_overhead( Some(overhead) } -/// Records a judge failure that made the classifier route without a verdict. -pub(crate) fn record_classifier_fail_open(judge_model: &str, reason: &'static str) { - meter() - .u64_counter("switchyard.classifier_fail_open") - .build() - .add( - 1, - &[ - KeyValue::new("judge_model", judge_model.to_string()), - KeyValue::new("reason", reason), - ], - ); +/// Records an algorithm metric after adding its algorithm dimension. +pub(crate) fn record_algorithm_metric(metric: &AlgorithmMetricObservation) { + let mut attributes = Vec::with_capacity(metric.attributes.len() + 1); + attributes.push(KeyValue::new("algorithm", metric.algorithm.clone())); + attributes.extend( + metric + .attributes + .iter() + .map(|attribute| KeyValue::new(attribute.key, attribute.value.clone())), + ); + + let meter = meter(); + match metric.value { + AlgorithmMetricValue::Counter(delta) => meter + .u64_counter(metric.name) + .build() + .add(delta, &attributes), + AlgorithmMetricValue::Histogram(sample) => meter + .f64_histogram(metric.name) + .build() + .record(sample, &attributes), + } } /// Records the resolution of one offloaded model call: the call counter and diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs index 501352f0..e3de5ed6 100644 --- a/crates/libsy/tests/observability.rs +++ b/crates/libsy/tests/observability.rs @@ -33,8 +33,9 @@ use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt}; use tracing_subscriber::registry::LookupSpan; use switchyard_libsy::{ - Algorithm, Driver, LibsyError, LlmClassifierConfig, LlmTarget, LlmTargetSet, LlmTaskClassifier, - Step, TaskClassifierConfig, + Algorithm, AlgorithmMetricValue, Driver, LibsyError, LlmClassifierConfig, LlmTarget, + LlmTargetSet, LlmTaskClassifier, MetricAttribute, RunObservation, RunObserver, Step, + TaskClassifierConfig, }; use switchyard_protocol::{ Context, Decision, LlmResponse, Metadata, Request, Response, RoutedLlmClient, Usage, @@ -290,8 +291,8 @@ fn f64_histogram_count( }) } -/// Latest cumulative sample sum of an `f64` histogram, in whole milliseconds. -fn f64_histogram_sum_ms( +/// Latest cumulative sample sum of an `f64` histogram, truncated to a `u64`. +fn f64_histogram_sum( snapshots: &[ResourceMetrics], name: &str, wanted: &[(&str, &str)], @@ -479,6 +480,47 @@ impl Algorithm for SingleCallAlgo { } } +/// Emits two batches of custom metrics so the observer can aggregate across runs. +struct AlgorithmMetricsAlgo; + +#[async_trait] +impl Algorithm for AlgorithmMetricsAlgo { + fn name(&self) -> &str { + "obs-algorithm-metrics" + } + + async fn create_run_task( + self: Arc, + _ctx: Context, + driver: Driver, + request: Request, + ) -> switchyard_libsy::Result { + let (counter_delta, samples) = match request.llm_request.model.as_deref() { + Some("metrics-first") => (2, [10.0, 20.0]), + _ => (3, [30.0, 40.0]), + }; + let attributes = || { + [ + MetricAttribute::new("source", "test-classifier"), + MetricAttribute::new("algorithm", "spoofed"), + ] + }; + driver.record_counter( + "switchyard.test.algorithm_decisions", + counter_delta, + attributes(), + ); + for sample in samples { + driver.record_histogram("switchyard.test.algorithm_score", sample, attributes()); + } + driver.record_histogram("switchyard.test.algorithm_score", f64::NAN, attributes()); + Ok(Response { + llm_response: LlmResponse::Agg(text_response(None, "metrics recorded")), + metadata: None, + }) + } +} + fn request_with_metadata(session_id: &str, correlation_id: &str) -> Request { Request { llm_request: text_request(Some("auto".to_string()), "hi"), @@ -1169,7 +1211,7 @@ async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_lib ); // The classifier call is the router's own work but the routed call is not, // so overhead lands near the classifier's 60ms, not their 260ms sum. - let overhead = f64_histogram_sum_ms( + let overhead = f64_histogram_sum( &snapshots, "switchyard.routing_overhead_ms", &[("algorithm", "llm_task_classifier")], @@ -1238,3 +1280,100 @@ async fn classifier_fail_open_records_each_failure_stage() -> switchyard_libsy:: } Ok(()) } + +#[tokio::test] +async fn algorithm_metrics_preserve_accumulator_aggregates_for_observer_and_otel() +-> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + let (_store, exporter, provider, _, _) = telemetry(); + let observations = Arc::new(Mutex::new(Vec::new())); + let observed = observations.clone(); + let observer: RunObserver = Arc::new(move |observation| observed.lock().push(observation)); + let algorithm = Arc::new(AlgorithmMetricsAlgo) as Arc; + + for model in ["metrics-first", "metrics-second"] { + let request = Request { + llm_request: text_request(Some(model.to_string()), "record metrics"), + raw_request: None, + metadata: None, + }; + algorithm + .clone() + .run_observed(Context::default(), request, Some(observer.clone())) + .await?; + } + + let observations = observations.lock(); + let metrics = observations + .iter() + .filter_map(|observation| match observation { + RunObservation::AlgorithmMetric(metric) => Some(metric), + _ => None, + }) + .collect::>(); + assert_eq!(metrics.len(), 6); + assert!(metrics.iter().all(|metric| { + metric.algorithm == "obs-algorithm-metrics" + && metric.attributes == [MetricAttribute::new("source", "test-classifier")] + })); + + let counter_total = metrics + .iter() + .filter_map(|metric| match metric.value { + AlgorithmMetricValue::Counter(delta) + if metric.name == "switchyard.test.algorithm_decisions" => + { + Some(delta) + } + _ => None, + }) + .sum::(); + assert_eq!(counter_total, 5); + + let mut samples = metrics + .iter() + .filter_map(|metric| match metric.value { + AlgorithmMetricValue::Histogram(sample) + if metric.name == "switchyard.test.algorithm_score" => + { + Some(sample) + } + _ => None, + }) + .collect::>(); + samples.sort_by(f64::total_cmp); + let count = samples.len() as u64; + let total = samples.iter().sum::(); + let min = samples.first().copied().unwrap_or_default(); + let max = samples.last().copied().unwrap_or_default(); + let average = total / count as f64; + let p50 = samples[samples.len() / 2]; + let p99 = samples[(samples.len() - 1).min((samples.len() as f64 * 0.99) as usize)]; + assert_eq!( + (count, total, min, max, average, p50, p99), + (4, 100.0, 10.0, 40.0, 25.0, 30.0, 40.0) + ); + + let snapshots = flushed_metrics(exporter, provider); + let attributes = [ + ("algorithm", "obs-algorithm-metrics"), + ("source", "test-classifier"), + ]; + assert_eq!( + u64_counter_value( + &snapshots, + "switchyard.test.algorithm_decisions", + &attributes, + ), + Some(5) + ); + assert_eq!( + f64_histogram_count(&snapshots, "switchyard.test.algorithm_score", &attributes,), + Some(4) + ); + assert_eq!( + f64_histogram_sum(&snapshots, "switchyard.test.algorithm_score", &attributes,), + Some(100) + ); + Ok(()) +} diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 13374da5..37180ce0 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -383,6 +383,7 @@ fn stats_observer(stats: StatsAccumulator) -> RunObserver { RunObservation::RoutingOverhead(duration) => { stats.record_routing_overhead(duration.as_secs_f64() * 1_000.0); } + RunObservation::AlgorithmMetric(_) => {} }) }