switchyard_llm_client/
metrics.rs

1// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Metric labelling inherited from Python
5
6use std::time::{Duration, Instant};
7
8use futures::{StreamExt, stream};
9use std::{
10    sync::OnceLock,
11    sync::atomic::{AtomicU64, Ordering},
12};
13
14use opentelemetry::metrics::ObservableGauge;
15use opentelemetry::{KeyValue, global};
16use switchyard_libsy::Result;
17use switchyard_protocol::{LlmResponse, LlmResponseChunk, ModelId, Response, WireFormat};
18use switchyard_translation::{DiagnosticSeverity, TranslationDiagnostic};
19
20static TOTAL_REQUESTS: AtomicU64 = AtomicU64::new(0);
21static TOTAL_ERRORS: AtomicU64 = AtomicU64::new(0);
22static TOTAL_GAUGES: OnceLock<(ObservableGauge<u64>, ObservableGauge<u64>)> = OnceLock::new();
23
24/// Registers process-wide compatibility gauges with the installed global meter provider.
25pub fn initialize() {
26    TOTAL_GAUGES.get_or_init(|| {
27        let meter = global::meter("switchyard");
28        let requests = meter
29            .u64_observable_gauge("switchyard.total_requests")
30            .with_callback(|observer| {
31                observer.observe(TOTAL_REQUESTS.load(Ordering::Relaxed), &[]);
32            })
33            .build();
34        let errors = meter
35            .u64_observable_gauge("switchyard.total_errors")
36            .with_callback(|observer| {
37                observer.observe(TOTAL_ERRORS.load(Ordering::Relaxed), &[]);
38            })
39            .build();
40        (requests, errors)
41    });
42}
43
44pub(crate) const fn is_retryable_http_status(status: u16) -> bool {
45    status == 408 || status == 429 || (status >= 500 && status <= 599)
46}
47
48/// Maps an HTTP status code (or lack thereof) to a normalized outcome metric label:
49/// - `"ok"` for HTTP 2xx success responses.
50/// - `"retryable_error"` for retryable HTTP statuses (408, 429, 5xx) or non-HTTP failures (`None`).
51/// - `"other_error"` for all other client or non-retryable statuses.
52pub const fn http_outcome_label(status: Option<u16>) -> &'static str {
53    match status {
54        Some(200..=299) => "ok",
55        Some(status) if is_retryable_http_status(status) => "retryable_error",
56        None => "retryable_error",
57        Some(_) => "other_error",
58    }
59}
60
61/// Limit the cardinality of the HTTP status code.
62/// This matches Python, but likely we should log the status code directly. Most of these never
63/// appear.
64pub const fn http_status_code_label(status: Option<u16>) -> &'static str {
65    match status {
66        None => "none",
67        Some(200) => "200",
68        Some(400) => "400",
69        Some(401) => "401",
70        Some(403) => "403",
71        Some(404) => "404",
72        Some(408) => "408",
73        Some(409) => "409",
74        Some(422) => "422",
75        Some(429) => "429",
76        Some(500) => "500",
77        Some(502) => "502",
78        Some(503) => "503",
79        Some(504) => "504",
80        Some(100..=199) => "1xx",
81        Some(200..=299) => "2xx",
82        Some(300..=399) => "3xx",
83        Some(400..=499) => "4xx",
84        Some(500..=599) => "5xx",
85        Some(_) => "other",
86    }
87}
88
89pub(crate) fn record_upstream_attempt(status: Option<u16>) {
90    global::meter("switchyard")
91        .u64_counter("switchyard.upstream_attempts")
92        .build()
93        .add(
94            1,
95            &[
96                KeyValue::new("outcome", http_outcome_label(status)),
97                KeyValue::new("code", http_status_code_label(status)),
98            ],
99        );
100}
101
102/// Records an upstream operation that succeeded after at least one retry.
103pub(crate) fn record_retry_recovered() {
104    global::meter("switchyard")
105        .u64_counter("switchyard.router_retry_recovered")
106        .build()
107        .add(1, &[]);
108}
109
110/// Records translation diagnostics as metrics and structured warnings.
111/// Diagnostic messages and JSON paths are logged, not used as metric labels.
112pub fn record_translation_diagnostics(
113    diagnostics: &[TranslationDiagnostic],
114    operation: &'static str,
115    format: WireFormat,
116) {
117    for diagnostic in diagnostics {
118        let severity = diagnostic_severity_label(&diagnostic.severity);
119        global::meter("switchyard")
120            .u64_counter("switchyard.translation_diagnostics")
121            .build()
122            .add(
123                1,
124                &[
125                    KeyValue::new("code", diagnostic.code.clone()),
126                    KeyValue::new("format", format.as_str()),
127                    KeyValue::new("operation", operation),
128                    KeyValue::new("severity", severity),
129                ],
130            );
131        tracing::warn!(
132            target: "libsy",
133            code = %diagnostic.code,
134            format = format.as_str(),
135            operation,
136            severity,
137            diagnostic = %diagnostic.message,
138            path = diagnostic.path.as_deref().unwrap_or(""),
139            "LLM protocol translation emitted a diagnostic"
140        );
141    }
142}
143
144const fn diagnostic_severity_label(severity: &DiagnosticSeverity) -> &'static str {
145    match severity {
146        DiagnosticSeverity::Info => "info",
147        DiagnosticSeverity::Warning => "warning",
148        DiagnosticSeverity::Error => "error",
149    }
150}
151
152/// Records the time needed to produce the routing outcome, including classifier calls,
153/// target resolution, request rewrites, and decision publishing.
154pub(crate) fn record_routing_overhead(algorithm: &str, overhead: Duration) {
155    global::meter("switchyard")
156        .f64_histogram("switchyard.routing_overhead_ms")
157        .build()
158        .record(
159            overhead.as_secs_f64() * 1000.0,
160            &[KeyValue::new("algorithm", algorithm.to_string())],
161        );
162}
163
164/// Records the final outcome, retaining ownership until a streamed answer ends or is dropped.
165pub(crate) fn observe_routed_request(
166    algorithm: &str,
167    selected_model: &ModelId,
168    answer_duration: Option<Duration>,
169    result: Result<Response>,
170) -> Result<Response> {
171    let mut call = RoutedCallMetrics {
172        algorithm: algorithm.to_owned(),
173        model: selected_model.clone(),
174        duration: answer_duration,
175        started: Instant::now(),
176        outcome: Some(result.is_ok()),
177    };
178    result.map(|mut response| {
179        if let LlmResponse::Stream(stream) = response.llm_response {
180            call.outcome = None;
181            response.llm_response = LlmResponse::Stream(Box::pin(stream::unfold(
182                (stream, call),
183                |(mut stream, mut call)| async move {
184                    let Some(item) = stream.next().await else {
185                        call.outcome.get_or_insert(true);
186                        return None;
187                    };
188                    match &item {
189                        Err(_) => call.outcome = Some(false),
190                        Ok(event) => {
191                            for chunk in event.normalized() {
192                                match chunk {
193                                    LlmResponseChunk::StreamError { .. }
194                                    | LlmResponseChunk::DecodeError { .. } => {
195                                        call.outcome = Some(false)
196                                    }
197                                    LlmResponseChunk::MessageStop { .. } => {
198                                        call.outcome.get_or_insert(true);
199                                    }
200                                    _ => {}
201                                }
202                            }
203                        }
204                    }
205                    Some((item, (stream, call)))
206                },
207            )));
208        }
209        response
210    })
211}
212
213struct RoutedCallMetrics {
214    algorithm: String,
215    model: ModelId,
216    duration: Option<Duration>,
217    started: Instant,
218    // Dropping an open stream is a failure; a terminal message permits a successful drop.
219    outcome: Option<bool>,
220}
221
222impl Drop for RoutedCallMetrics {
223    fn drop(&mut self) {
224        let is_success = self.outcome.unwrap_or(false);
225        let duration = self
226            .duration
227            .map(|duration| duration + self.started.elapsed());
228        if let Some(duration) = duration {
229            record_answer_call(&self.algorithm, &self.model, duration, is_success);
230        }
231        record_routed_request(&self.model, duration, is_success);
232    }
233}
234
235/// Records one terminal model call made after routing, preserving the libsy call metric surface.
236pub(crate) fn record_answer_call(
237    algorithm: &str,
238    selected_model: &ModelId,
239    duration: Duration,
240    is_success: bool,
241) {
242    let attributes = [
243        KeyValue::new("algorithm", algorithm.to_string()),
244        KeyValue::new("selected_model", selected_model.to_string()),
245        KeyValue::new("outcome", if is_success { "ok" } else { "error" }),
246    ];
247    let meter = global::meter("switchyard");
248    meter
249        .u64_counter("switchyard.llm_calls")
250        .build()
251        .add(1, &attributes);
252    meter
253        .f64_histogram("switchyard.llm_call_duration_ms")
254        .build()
255        .record(duration.as_secs_f64() * 1000.0, &attributes);
256}
257
258/// Records one terminal routed request after the algorithm has produced an outcome.
259pub(crate) fn record_routed_request(
260    selected_model: &ModelId,
261    answer_duration: Option<Duration>,
262    is_success: bool,
263) {
264    TOTAL_REQUESTS.fetch_add(1, Ordering::Relaxed);
265    let attributes = [KeyValue::new("model", selected_model.to_string())];
266    let meter = global::meter("switchyard");
267    if is_success {
268        meter
269            .u64_counter("switchyard.requests")
270            .build()
271            .add(1, &attributes);
272        if let Some(duration) = answer_duration {
273            meter
274                .f64_histogram("switchyard.model_call_latency_ms")
275                .build()
276                .record(duration.as_secs_f64() * 1000.0, &attributes);
277        }
278    } else {
279        TOTAL_ERRORS.fetch_add(1, Ordering::Relaxed);
280        meter
281            .u64_counter("switchyard.errors")
282            .build()
283            .add(1, &attributes);
284    }
285}
286
287#[cfg(test)]
288mod tests {
289    use std::io::{self, Write};
290    use std::sync::{Arc, Mutex};
291
292    use super::*;
293
294    #[derive(Default)]
295    struct LogBuffer(Mutex<Vec<u8>>);
296
297    impl Write for &LogBuffer {
298        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
299            self.0.lock().expect("log buffer lock").extend(bytes);
300            Ok(bytes.len())
301        }
302
303        fn flush(&mut self) -> io::Result<()> {
304            Ok(())
305        }
306    }
307
308    #[test]
309    fn outcome_labels_match_the_retry_policy() {
310        assert_eq!(http_outcome_label(Some(200)), "ok");
311        for status in [408, 429, 500, 502, 599] {
312            assert!(is_retryable_http_status(status));
313            assert_eq!(http_outcome_label(Some(status)), "retryable_error");
314        }
315        for status in [400, 409, 499, 600] {
316            assert!(!is_retryable_http_status(status));
317            assert_eq!(http_outcome_label(Some(status)), "other_error");
318        }
319        assert_eq!(http_outcome_label(None), "retryable_error");
320    }
321
322    // One diagnostic produces one warning containing its structured troubleshooting fields.
323    #[test]
324    fn translation_diagnostic_emits_one_structured_warning() {
325        let output = Arc::new(LogBuffer::default());
326        let subscriber = tracing_subscriber::fmt()
327            .with_ansi(false)
328            .without_time()
329            .with_writer(output.clone())
330            .finish();
331        tracing::subscriber::with_default(subscriber, || {
332            record_translation_diagnostics(
333                &[TranslationDiagnostic::warning(
334                    "lossy_conversion",
335                    "dropped unsupported JSON Schema constraints",
336                )
337                .at_path("$.response_format")],
338                "request_encode",
339                WireFormat::AnthropicMessages,
340            );
341        });
342
343        let bytes = output.0.lock().expect("log buffer lock").clone();
344        let logs = String::from_utf8(bytes).expect("captured log must be UTF-8");
345        let warnings = logs
346            .lines()
347            .filter(|line| line.contains("LLM protocol translation emitted a diagnostic"))
348            .collect::<Vec<_>>();
349        assert_eq!(warnings.len(), 1, "diagnostic warnings: {logs}");
350        for field in [
351            "WARN",
352            "libsy",
353            "lossy_conversion",
354            "anthropic_messages",
355            "request_encode",
356            "warning",
357            "dropped unsupported JSON Schema constraints",
358            "$.response_format",
359        ] {
360            assert!(warnings[0].contains(field), "missing {field:?} in {logs}");
361        }
362    }
363}
switchyard_llm_client/metrics.rs

switchyard_llm_client/
metrics.rs
