feat(telemetry): install W3C propagators and centralize telemetry helpers
- export: install trace context + baggage propagators at init - export: add http server/client span helpers, remote parent adoption, traceparent capture/injection and queue-boundary helpers - export: own the OTLP meter provider and axum metrics layer; skip health and static asset paths - apihost: use the shared meter provider and one shutdown entry point
ZhenYi committed
Modified app/apihost/src/main.rs app/apihost/src/main.rs +4 −13 View app/apihost/src/main.rs
@@ -25,13 +25,9 @@
2525 .context("building AppState")?;
2626 info!("dependencies ready");
2727
28 let meter_provider = match webapi::metrics::init_meter_provider() {
29 Ok(provider) => Some(provider),
30 Err(err) => {
31 warn!(error = %err, "failed to init meter provider; metrics disabled");
32 None
33 }
34 };
28 if let Err(err) = export::telemetry::init_meter_provider() {
29 warn!(error = %err, "failed to init meter provider; metrics disabled");
30 }
3531
3632 tokio::spawn(chat::worker::run_background_tasks(state.clone()));
3733 tokio::spawn(auth::worker::run_deletion_executor(state.clone()));
@@ -74,12 +70,7 @@
7470 .await
7571 .context("axum server error")?;
7672
77 if let Some(meter_provider) = meter_provider
78 && let Err(err) = meter_provider.shutdown()
79 {
80 warn!(error = %err, "failed to shut down meter provider");
81 }
82 export::telemetry::flush_traces();
73 export::telemetry::shutdown_telemetry();
8374
8475 info!("apihost shut down");
8576 Ok(())
Modified lib/export/src/telemetry.rs lib/export/src/telemetry.rs +206 −16 View lib/export/src/telemetry.rs
@@ -1,16 +1,37 @@
11use std::sync::OnceLock;
2use std::time::Duration;
23
4use axum::extract::MatchedPath;
5use axum::http::{HeaderMap, Request, Response};
6use opentelemetry::global;
7use opentelemetry::propagation::TextMapCompositePropagator;
38use opentelemetry::trace::TracerProvider as _;
9use opentelemetry::Context;
10use opentelemetry_http::{HeaderExtractor, HeaderInjector};
11use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};
12use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator};
413use opentelemetry_sdk::trace::SdkTracerProvider;
5use tracing::{info, warn};
14use tracing::{Instrument as _, Span, info, warn};
15use tracing_opentelemetry::OpenTelemetrySpanExt;
616use tracing_subscriber::EnvFilter;
717use tracing_subscriber::layer::SubscriberExt;
818use tracing_subscriber::util::SubscriberInitExt;
919
1020/// Installed tracer provider, kept so shutdown can flush the batch processor.
1121static TRACER_PROVIDER: OnceLock<SdkTracerProvider> = OnceLock::new();
22/// Installed meter provider, kept so shutdown can flush the periodic reader.
23static METER_PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();
24
25const METRICS_EXPORT_INTERVAL: Duration = Duration::from_secs(30);
1226
1327pub fn init_tracing() {
28 // W3C trace context is what lets `traceparent` survive HTTP, gRPC and
29 // queue payloads across hosts; installing it is a no-op for log-only runs.
30 global::set_text_map_propagator(TextMapCompositePropagator::new(vec![
31 Box::new(TraceContextPropagator::new()),
32 Box::new(BaggagePropagator::new()),
33 ]));
34
1435 // The filter is installed on the registry rather than on the fmt layer so it
1536 // governs span *creation*: without it every debug-level span in the process
1637 // (hyper/tower internals) becomes an exported span.
@@ -31,22 +52,72 @@
3152 }
3253}
3354
34/// Traces are opt-in through the standard OTLP environment variables: setting
35/// `OTEL_EXPORTER_OTLP_ENDPOINT` turns export on, anything else (including a bad
36/// endpoint) only downgrades to log-only. Telemetry must never stop a host from
37/// booting, so every failure path here logs and returns `None`.
38fn tracer_provider() -> Option<SdkTracerProvider> {
39 if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
40 .unwrap_or_default()
41 .is_empty()
42 {
43 return None;
55pub fn init_meter_provider() -> Result<SdkMeterProvider, String> {
56 if !otlp_configured(OtlpSignal::Metrics) {
57 return Err("OTLP endpoint is not configured".to_string());
4458 }
4559
46 let exporter = match opentelemetry_otlp::SpanExporter::builder()
60 let exporter = opentelemetry_otlp::MetricExporter::builder()
4761 .with_http()
62 .with_temporality(Temporality::default())
4863 .build()
49 {
64 .map_err(|error| error.to_string())?;
65
66 let reader = PeriodicReader::builder(exporter)
67 .with_interval(METRICS_EXPORT_INTERVAL)
68 .build();
69
70 let provider = SdkMeterProvider::builder().with_reader(reader).build();
71 global::set_meter_provider(provider.clone());
72 let _ = METER_PROVIDER.set(provider.clone());
73 Ok(provider)
74}
75
76/// HTTP server metrics for axum-style routers. Paths without a matched route
77/// (static assets, the SPA fallback) are skipped so raw URLs never become
78/// metric labels.
79pub fn metrics_layer() -> axum_otel_metrics::HttpMetricsLayer {
80 axum_otel_metrics::HttpMetricsLayerBuilder::new()
81 .with_skipper(axum_otel_metrics::PathSkipper::new(skip_metrics_path))
82 .build()
83}
84
85fn skip_metrics_path(path: &str) -> bool {
86 matches!(path, "/healthz" | "/readyz" | "/favicon.ico")
87 || path.starts_with("/assets/")
88 || path.starts_with("/.well-known/")
89}
90
91#[derive(Clone, Copy)]
92enum OtlpSignal {
93 Traces,
94 Metrics,
95}
96
97/// Telemetry is opt-in through the standard OTLP environment variables:
98/// setting `OTEL_EXPORTER_OTLP_ENDPOINT` turns export on, anything else
99/// (including a bad endpoint) only downgrades to log-only. Telemetry must never
100/// stop a host from booting, so every failure path here logs and returns `None`.
101fn otlp_configured(signal: OtlpSignal) -> bool {
102 let specific = match signal {
103 OtlpSignal::Traces => "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
104 OtlpSignal::Metrics => "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
105 };
106 ["OTEL_EXPORTER_OTLP_ENDPOINT", specific]
107 .iter()
108 .any(|key| {
109 std::env::var(key)
110 .map(|value| !value.trim().is_empty())
111 .unwrap_or(false)
112 })
113}
114
115fn tracer_provider() -> Option<SdkTracerProvider> {
116 if !otlp_configured(OtlpSignal::Traces) {
117 return None;
118 }
119
120 let exporter = match opentelemetry_otlp::SpanExporter::builder().with_http().build() {
50121 Ok(exporter) => exporter,
51122 Err(err) => {
52123 eprintln!("failed to build otlp span exporter, traces disabled: {err}");
@@ -61,6 +132,116 @@
61132 )
62133}
63134
135/// Adopts the W3C trace context carried by inbound headers, when present.
136pub fn set_remote_parent(span: &Span, headers: &HeaderMap) {
137 let parent = global::get_text_map_propagator(|propagator| {
138 propagator.extract_with_context(&Context::new(), &HeaderExtractor(headers))
139 });
140 let _ = span.set_parent(parent);
141}
142
143/// Continues a trace across a queue boundary; `None` keeps the ambient parent.
144pub fn set_parent_from_traceparent(span: &Span, traceparent: Option<&str>) {
145 let Some(traceparent) = traceparent else {
146 return;
147 };
148 let mut headers = HeaderMap::new();
149 if let Ok(value) = traceparent.parse() {
150 headers.insert("traceparent", value);
151 }
152 set_remote_parent(span, &headers);
153}
154
155/// Serialises the current trace context for a queue payload.
156pub fn traceparent() -> Option<String> {
157 let mut headers = HeaderMap::new();
158 global::get_text_map_propagator(|propagator| {
159 propagator.inject_context(&Span::current().context(), &mut HeaderInjector(&mut headers));
160 });
161 headers
162 .get("traceparent")
163 .and_then(|value| value.to_str().ok())
164 .map(str::to_owned)
165}
166
167pub fn http_span<B>(request: &Request<B>) -> Span {
168 let method = request.method().as_str();
169 let route = request
170 .extensions()
171 .get::<MatchedPath>()
172 .map_or_else(|| request.uri().path(), |path| path.as_str());
173 let name = format!("{method} {route}");
174 let span = tracing::info_span!(
175 "http.server",
176 "otel.name" = %name,
177 "otel.kind" = "server",
178 "http.request.method" = %method,
179 "url.path" = %request.uri().path(),
180 "http.route" = %route,
181 "http.response.status_code" = tracing::field::Empty,
182 "latency_ms" = tracing::field::Empty,
183 "request_id" = tracing::field::Empty,
184 );
185 if let Some(request_id) = request.extensions().get::<crate::middleware::RequestId>() {
186 span.record("request_id", request_id.0.as_str());
187 }
188 set_remote_parent(&span, request.headers());
189 span
190}
191
192pub fn on_response<B>(response: &Response<B>, latency: Duration, span: &Span) {
193 let status = response.status().as_u16();
194 span.record("http.response.status_code", i64::from(status));
195 span.record("latency_ms", latency.as_millis() as i64);
196 if status >= 500 {
197 span.record("otel.status_code", "error");
198 span.record("error.type", i64::from(status));
199 }
200}
201
202/// Client span for an outbound HTTP call; the trace context of this span is
203/// what gets injected into the request, so the remote hop parents to it.
204pub fn http_client_span(method: &str, url: &str) -> Span {
205 let name = format!("{method} {url}");
206 tracing::info_span!(
207 "http.client",
208 "otel.name" = %name,
209 "otel.kind" = "client",
210 "http.request.method" = %method,
211 "url.full" = %url,
212 "http.response.status_code" = tracing::field::Empty,
213 )
214}
215
216/// Sends a request inside `span` with `traceparent` injected into its headers.
217pub async fn send(
218 span: Span,
219 request: reqwest::RequestBuilder,
220) -> Result<reqwest::Response, reqwest::Error> {
221 let mut headers = HeaderMap::new();
222 global::get_text_map_propagator(|propagator| {
223 propagator.inject_context(&span.context(), &mut HeaderInjector(&mut headers));
224 });
225
226 let result = request
227 .headers(headers)
228 .send()
229 .instrument(span.clone())
230 .await;
231 match &result {
232 Ok(response) => {
233 span.record(
234 "http.response.status_code",
235 i64::from(response.status().as_u16()),
236 );
237 }
238 Err(_) => {
239 span.record("otel.status_code", "error");
240 }
241 }
242 result
243}
244
64245pub async fn shutdown_signal() {
65246 let ctrl_c = async {
66247 if let Err(err) = tokio::signal::ctrl_c().await {
@@ -89,12 +270,21 @@
89270 _ = terminate => info!("received SIGTERM"),
90271 }
91272
273 shutdown_telemetry();
274}
275
276/// Drains spans queued in the batch processor and metrics queued in the
277/// periodic reader. Without this the tail of every deployment's telemetry is
278/// lost on rollout, which is exactly when you look at it.
279pub fn shutdown_telemetry() {
92280 flush_traces();
281 if let Some(provider) = METER_PROVIDER.get()
282 && let Err(err) = provider.shutdown()
283 {
284 warn!(error = %err, "failed to shut down meter provider");
285 }
93286}
94287
95/// Drains spans queued in the batch processor. Without this the tail of every
96/// deployment's trace output is lost on rollout, which is exactly when you look
97/// at traces.
98288pub fn flush_traces() {
99289 if let Some(provider) = TRACER_PROVIDER.get()
100290 && let Err(err) = provider.shutdown()
Modified lib/export/Cargo.toml lib/export/Cargo.toml +3 −0 View lib/export/Cargo.toml
@@ -16,9 +16,12 @@
1616sha2 = { workspace = true }
1717tracing = { workspace = true }
1818tracing-subscriber = { workspace = true }
19axum-otel-metrics = { workspace = true }
1920opentelemetry = { workspace = true }
2021opentelemetry_sdk = { workspace = true }
22opentelemetry-http = { workspace = true }
2123opentelemetry-otlp = { workspace = true }
24reqwest = { workspace = true }
2225tracing-opentelemetry = { workspace = true }
2326utoipa = { workspace = true }
2427rand.workspace = true
Modified lib/webapi/src/lib.rs lib/webapi/src/lib.rs +1 −2 View lib/webapi/src/lib.rs
@@ -1,5 +1,4 @@
11pub mod cleanup;
2pub mod metrics;
32pub mod openapi;
43pub mod search;
54pub mod spa;
@@ -109,7 +108,7 @@
109108 )
110109 .on_response(DefaultOnResponse::new().level(Level::INFO)),
111110 )
112 .layer(metrics::metrics_layer())
111 .layer(export::telemetry::metrics_layer())
113112 .layer(AppTraceLayer::new());
114113
115114 Ok(router.merge(ci::actions_runtime::router().with_state(ci)))
Deleted lib/webapi/src/metrics.rs lib/webapi/src/metrics.rs +0 −28
@@ -1,28 +0,0 @@
1use std::time::Duration;
2
3use axum_otel_metrics::{HttpMetricsLayer, HttpMetricsLayerBuilder};
4use opentelemetry::global;
5use opentelemetry_otlp::MetricExporter;
6use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};
7
8const EXPORT_INTERVAL: Duration = Duration::from_secs(30);
9
10pub fn init_meter_provider() -> Result<SdkMeterProvider, String> {
11 let exporter = MetricExporter::builder()
12 .with_http()
13 .with_temporality(Temporality::default())
14 .build()
15 .map_err(|e| e.to_string())?;
16
17 let reader = PeriodicReader::builder(exporter)
18 .with_interval(EXPORT_INTERVAL)
19 .build();
20
21 let provider = SdkMeterProvider::builder().with_reader(reader).build();
22 global::set_meter_provider(provider.clone());
23 Ok(provider)
24}
25
26pub fn metrics_layer() -> HttpMetricsLayer {
27 HttpMetricsLayerBuilder::new().build()
28}
Modified lib/webapi/Cargo.toml lib/webapi/Cargo.toml +0 −4 View lib/webapi/Cargo.toml
@@ -25,10 +25,6 @@
2525tower-http = { workspace = true }
2626tracing = { workspace = true }
2727utoipa = { workspace = true }
28axum-otel-metrics = { workspace = true }
29opentelemetry = { workspace = true }
30opentelemetry_sdk = { workspace = true }
31opentelemetry-otlp = { workspace = true }
3228web = { workspace = true }
3329serde = { workspace = true, features = ["derive"] }
3430serde_json = { workspace = true }
Modified Cargo.lock Cargo.lock +3 −4 View Cargo.lock
@@ -2448,6 +2448,7 @@
24482448version = "1.0.0"
24492449dependencies = [
24502450 "axum",
2451 "axum-otel-metrics",
24512452 "bytes",
24522453 "config",
24532454 "email",
@@ -2455,10 +2456,12 @@
24552456 "imaging",
24562457 "jsonwebtoken 9.3.1",
24572458 "opentelemetry",
2459 "opentelemetry-http",
24582460 "opentelemetry-otlp",
24592461 "opentelemetry_sdk",
24602462 "rand 0.8.7",
24612463 "redis",
2464 "reqwest 0.12.28",
24622465 "sea-orm",
24632466 "serde",
24642467 "serde_json",
@@ -10716,7 +10719,6 @@
1071610719 "agents",
1071710720 "auth",
1071810721 "axum",
10719 "axum-otel-metrics",
1072010722 "chat",
1072110723 "ci",
1072210724 "config",
@@ -10725,9 +10727,6 @@
1072510727 "import",
1072610728 "issues",
1072710729 "models",
10728 "opentelemetry",
10729 "opentelemetry-otlp",
10730 "opentelemetry_sdk",
1073110730 "pulls",
1073210731 "sea-orm",
1073310732 "search",
Modified Cargo.toml Cargo.toml +1 −0 View Cargo.toml
@@ -59,6 +59,7 @@
5959axum-otel-metrics = "0.14.1"
6060opentelemetry = "0.32"
6161opentelemetry_sdk = { version = "0.32.1", features = ["rt-tokio", "trace"] }
62opentelemetry-http = "0.32"
6263# reqwest-rustls: the default client feature set compiles reqwest with no TLS backend,
6364# so an https OTEL_EXPORTER_OTLP_ENDPOINT would fail every export at request time.
6465opentelemetry-otlp = { version = "0.32", features = ["reqwest-rustls"] }