switchyard_server/
observability.rs

1// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Process-wide tracing and OpenTelemetry setup for server hosts.
5
6use std::env;
7use std::sync::OnceLock;
8
9use axum::http::HeaderMap;
10use opentelemetry::propagation::{Extractor, TextMapPropagator};
11use opentelemetry::trace::TracerProvider as _;
12use opentelemetry_sdk::Resource;
13use opentelemetry_sdk::propagation::TraceContextPropagator;
14use opentelemetry_sdk::trace::SdkTracerProvider;
15use tracing_opentelemetry::OpenTelemetrySpanExt;
16use tracing_subscriber::layer::SubscriberExt as _;
17use tracing_subscriber::util::SubscriberInitExt as _;
18use tracing_subscriber::{EnvFilter, Layer as _};
19
20use crate::{ServerError, ServerResult, metrics};
21
22const DEFAULT_LOG_FILTER: &str = "switchyard_server=info,libsy=info,opentelemetry=warn";
23const DEFAULT_SERVICE_NAME: &str = "switchyard-server";
24
25struct Observability {
26    tracer_provider: Option<SdkTracerProvider>,
27}
28
29static OBSERVABILITY: OnceLock<Result<Observability, String>> = OnceLock::new();
30
31/// Installs metrics and tracing once for either the binary or an embedded host.
32pub fn initialize_observability() -> ServerResult<()> {
33    match OBSERVABILITY.get_or_init(initialize) {
34        Ok(_) => Ok(()),
35        Err(error) => Err(ServerError::new(error.clone())),
36    }
37}
38
39/// Flushes pending OTLP telemetry without shutting down process-wide providers.
40pub fn flush_observability() {
41    if let Some(Ok(observability)) = OBSERVABILITY.get()
42        && let Some(provider) = &observability.tracer_provider
43        && let Err(error) = provider.force_flush()
44    {
45        tracing::warn!(error = %error, "failed to flush OpenTelemetry traces");
46    }
47    metrics::flush();
48}
49
50/// Creates the server request span with any incoming W3C trace context as its parent.
51pub(crate) fn request_span(headers: &HeaderMap) -> tracing::Span {
52    let parent = TraceContextPropagator::new().extract(&HeaderExtractor(headers));
53    let span = tracing::info_span!(
54        target: "switchyard_server",
55        "switchyard.request",
56        otel.kind = "server",
57        openinference.span.kind = "CHAIN",
58    );
59    let _ = span.set_parent(parent);
60    span
61}
62
63struct HeaderExtractor<'a>(&'a HeaderMap);
64
65impl Extractor for HeaderExtractor<'_> {
66    fn get(&self, key: &str) -> Option<&str> {
67        self.0.get(key).and_then(|value| value.to_str().ok())
68    }
69
70    fn keys(&self) -> Vec<&str> {
71        self.0.keys().map(|name| name.as_str()).collect()
72    }
73}
74
75pub(crate) fn otlp_enabled(signal: &str) -> bool {
76    if env_var_is_true("OTEL_SDK_DISABLED") {
77        return false;
78    }
79    if env::var(format!("OTEL_{signal}_EXPORTER"))
80        .ok()
81        .filter(|value| !value.trim().is_empty())
82        .is_some_and(|value| {
83            !value
84                .split(',')
85                .any(|exporter| exporter.trim().eq_ignore_ascii_case("otlp"))
86        })
87    {
88        return false;
89    }
90    [
91        "OTEL_EXPORTER_OTLP_ENDPOINT",
92        &format!("OTEL_EXPORTER_OTLP_{signal}_ENDPOINT"),
93    ]
94    .into_iter()
95    .any(|name| env::var(name).is_ok_and(|value| !value.trim().is_empty()))
96}
97
98pub(crate) fn resource() -> Resource {
99    let service_name = env::var("OTEL_SERVICE_NAME")
100        .ok()
101        .filter(|value| !value.trim().is_empty())
102        .unwrap_or_else(|| DEFAULT_SERVICE_NAME.to_string());
103    Resource::builder().with_service_name(service_name).build()
104}
105
106fn initialize() -> Result<Observability, String> {
107    metrics::registry()?;
108
109    let tracer_provider = otlp_enabled("TRACES")
110        .then(build_tracer_provider)
111        .transpose()?;
112    let filter = log_filter()?;
113    let format = tracing_subscriber::fmt::layer()
114        .with_ansi(false)
115        .with_writer(std::io::stderr)
116        .with_filter(filter);
117
118    if let Some(provider) = &tracer_provider {
119        let tracer = provider.tracer("switchyard");
120        tracing_subscriber::registry()
121            .with(format)
122            .with(
123                tracing_opentelemetry::layer()
124                    .with_tracer(tracer)
125                    .with_filter(log_filter()?),
126            )
127            .try_init()
128            .map_err(|error| format!("failed to initialize tracing: {error}"))?;
129    } else {
130        tracing_subscriber::registry()
131            .with(format)
132            .try_init()
133            .map_err(|error| format!("failed to initialize tracing: {error}"))?;
134    }
135
136    Ok(Observability { tracer_provider })
137}
138
139fn log_filter() -> Result<EnvFilter, String> {
140    EnvFilter::try_from_default_env()
141        .or_else(|_| EnvFilter::try_new(DEFAULT_LOG_FILTER))
142        .map_err(|error| format!("invalid tracing filter: {error}"))
143}
144
145fn build_tracer_provider() -> Result<SdkTracerProvider, String> {
146    let exporter = opentelemetry_otlp::SpanExporter::builder()
147        .with_http()
148        .build()
149        .map_err(|error| format!("failed to initialize OTLP trace exporter: {error}"))?;
150    let provider = SdkTracerProvider::builder()
151        .with_resource(resource())
152        .with_batch_exporter(exporter)
153        .build();
154    opentelemetry::global::set_tracer_provider(provider.clone());
155    Ok(provider)
156}
157
158fn env_var_is_true(name: &str) -> bool {
159    env::var(name).is_ok_and(|value| matches!(value.to_ascii_lowercase().as_str(), "true" | "1"))
160}
161
162#[cfg(test)]
163mod tests {
164    use axum::http::{HeaderMap, HeaderValue};
165    use opentelemetry::trace::{TraceContextExt, TracerProvider as _};
166    use opentelemetry_sdk::trace::SdkTracerProvider;
167    use tracing_opentelemetry::OpenTelemetrySpanExt as _;
168    use tracing_subscriber::layer::SubscriberExt as _;
169
170    use super::request_span;
171
172    #[test]
173    fn request_span_continues_incoming_w3c_trace_context() {
174        let provider = SdkTracerProvider::builder().build();
175        let tracer = provider.tracer("request-span-test");
176        let subscriber =
177            tracing_subscriber::registry().with(tracing_opentelemetry::layer().with_tracer(tracer));
178        let mut headers = HeaderMap::new();
179        headers.insert(
180            "traceparent",
181            HeaderValue::from_static("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"),
182        );
183        headers.insert(
184            "tracestate",
185            HeaderValue::from_static("vendor=opaque-value"),
186        );
187
188        tracing::subscriber::with_default(subscriber, || {
189            let span = request_span(&headers);
190            let context = span.context();
191            let current = context.span();
192            let span_context = current.span_context();
193            assert_eq!(
194                span_context.trace_id().to_string(),
195                "4bf92f3577b34da6a3ce929d0e0e4736"
196            );
197            assert_eq!(span_context.trace_state().header(), "vendor=opaque-value");
198        });
199    }
200}
switchyard_server/observability.rs

switchyard_server/
observability.rs
