acbf7ea
use std::sync::OnceLock;
use std::time::Duration;
use axum::extract::MatchedPath;
use axum::http::{HeaderMap, Request, Response};
use opentelemetry::Context;
use opentelemetry::global;
use opentelemetry::propagation::TextMapCompositePropagator;
use opentelemetry::trace::TracerProvider as _;
use opentelemetry_http::{HeaderExtractor, HeaderInjector};
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};
use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator};
use opentelemetry_sdk::trace::SdkTracerProvider;
use tracing::{Instrument as _, Span, info, warn};
use tracing_opentelemetry::OpenTelemetrySpanExt;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
/// Installed tracer provider, kept so shutdown can flush the batch processor.
static TRACER_PROVIDER: OnceLock<SdkTracerProvider> = OnceLock::new();
/// Installed meter provider, kept so shutdown can flush the periodic reader.
static METER_PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();
const METRICS_EXPORT_INTERVAL: Duration = Duration::from_secs(30);
pub fn init_tracing() {
// W3C trace context is what lets `traceparent` survive HTTP, gRPC and
// queue payloads across hosts; installing it is a no-op for log-only runs.
global::set_text_map_propagator(TextMapCompositePropagator::new(vec![
Box::new(TraceContextPropagator::new()),
Box::new(BaggagePropagator::new()),
]));
// The filter is installed on the registry rather than on the fmt layer so it
// governs span *creation*: without it every debug-level span in the process
// (hyper/tower internals) becomes an exported span.
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
let subscriber = tracing_subscriber::registry()
.with(tracing_subscriber::fmt::layer())
.with(filter);
match tracer_provider() {
Some(provider) => {
let _ = TRACER_PROVIDER.set(provider.clone());
subscriber
.with(tracing_opentelemetry::layer().with_tracer(provider.tracer("gitdata")))
.init();
info!("opentelemetry trace export enabled");
}
None => subscriber.init(),
}
}
pub fn init_meter_provider() -> Result<SdkMeterProvider, String> {
if !otlp_configured(OtlpSignal::Metrics) {
return Err("OTLP endpoint is not configured".to_string());
}
let exporter = opentelemetry_otlp::MetricExporter::builder()
.with_http()
.with_temporality(Temporality::default())
.build()
.map_err(|error| error.to_string())?;
let reader = PeriodicReader::builder(exporter)
.with_interval(METRICS_EXPORT_INTERVAL)
.build();
let provider = SdkMeterProvider::builder().with_reader(reader).build();
global::set_meter_provider(provider.clone());
let _ = METER_PROVIDER.set(provider.clone());
Ok(provider)
}
/// HTTP server metrics for axum-style routers. Paths without a matched route
/// (static assets, the SPA fallback) are skipped so raw URLs never become
/// metric labels.
pub fn metrics_layer() -> axum_otel_metrics::HttpMetricsLayer {
axum_otel_metrics::HttpMetricsLayerBuilder::new()
.with_skipper(axum_otel_metrics::PathSkipper::new(skip_metrics_path))
.build()
}
fn skip_metrics_path(path: &str) -> bool {
matches!(path, "/healthz" | "/readyz" | "/favicon.ico")
|| path.starts_with("/assets/")
|| path.starts_with("/.well-known/")
}
#[derive(Clone, Copy)]
enum OtlpSignal {
Traces,
Metrics,
}
/// Telemetry is opt-in through the standard OTLP environment variables:
/// setting `OTEL_EXPORTER_OTLP_ENDPOINT` turns export on, anything else
/// (including a bad endpoint) only downgrades to log-only. Telemetry must never
/// stop a host from booting, so every failure path here logs and returns `None`.
fn otlp_configured(signal: OtlpSignal) -> bool {
let specific = match signal {
OtlpSignal::Traces => "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
OtlpSignal::Metrics => "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",
};
["OTEL_EXPORTER_OTLP_ENDPOINT", specific].iter().any(|key| {
std::env::var(key)
.map(|value| !value.trim().is_empty())
.unwrap_or(false)
})
}
fn tracer_provider() -> Option<SdkTracerProvider> {
if !otlp_configured(OtlpSignal::Traces) {
return None;
}
let exporter = match opentelemetry_otlp::SpanExporter::builder()
.with_http()
.build()
{
Ok(exporter) => exporter,
Err(err) => {
eprintln!("failed to build otlp span exporter, traces disabled: {err}");
return None;
}
};
Some(
SdkTracerProvider::builder()
.with_batch_exporter(exporter)
.build(),
)
}
/// Adopts the W3C trace context carried by inbound headers, when present.
pub fn set_remote_parent(span: &Span, headers: &HeaderMap) {
let parent = global::get_text_map_propagator(|propagator| {
propagator.extract_with_context(&Context::new(), &HeaderExtractor(headers))
});
let _ = span.set_parent(parent);
}
/// Continues a trace across a queue boundary; `None` keeps the ambient parent.
pub fn set_parent_from_traceparent(span: &Span, traceparent: Option<&str>) {
let Some(traceparent) = traceparent else {
return;
};
let mut headers = HeaderMap::new();
if let Ok(value) = traceparent.parse() {
headers.insert("traceparent", value);
}
set_remote_parent(span, &headers);
}
/// Serialises the current trace context for a queue payload.
pub fn traceparent() -> Option<String> {
let mut headers = HeaderMap::new();
global::get_text_map_propagator(|propagator| {
propagator.inject_context(
&Span::current().context(),
&mut HeaderInjector(&mut headers),
);
});
headers
.get("traceparent")
.and_then(|value| value.to_str().ok())
.map(str::to_owned)
}
/// Serialises the request id for a queue payload.
pub fn request_id() -> Option<String> {
crate::middleware::request_id()
}
pub fn http_span<B>(request: &Request<B>) -> Span {
let method = request.method().as_str();
let route = request
.extensions()
.get::<MatchedPath>()
.map_or_else(|| request.uri().path(), |path| path.as_str());
let name = format!("{method} {route}");
let span = tracing::info_span!(
"http.server",
"otel.name" = %name,
"otel.kind" = "server",
"http.request.method" = %method,
"url.path" = %request.uri().path(),
"http.route" = %route,
"http.response.status_code" = tracing::field::Empty,
"latency_ms" = tracing::field::Empty,
"request_id" = tracing::field::Empty,
);
if let Some(request_id) = request.extensions().get::<crate::middleware::RequestId>() {
span.record("request_id", request_id.0.as_str());
}
set_remote_parent(&span, request.headers());
span
}
pub fn on_response<B>(response: &Response<B>, latency: Duration, span: &Span) {
let status = response.status().as_u16();
span.record("http.response.status_code", i64::from(status));
span.record("latency_ms", latency.as_millis() as i64);
if status >= 500 {
span.record("otel.status_code", "error");
span.record("error.type", i64::from(status));
}
}
/// `tower_http::trace::MakeSpan` adapter shared by every axum/tonic host.
#[derive(Clone, Copy, Default)]
pub struct HttpSpan;
impl<B> tower_http::trace::MakeSpan<B> for HttpSpan {
fn make_span(&mut self, request: &Request<B>) -> Span {
http_span(request)
}
}
/// `tower_http::trace::OnResponse` adapter keeping the request log line and
/// recording the HTTP semantic-convention attributes on the span.
#[derive(Clone, Copy, Default)]
pub struct HttpOnResponse;
impl<B> tower_http::trace::OnResponse<B> for HttpOnResponse {
fn on_response(self, response: &Response<B>, latency: Duration, span: &Span) {
on_response(response, latency, span);
tracing::info!(
parent: span,
status = response.status().as_u16(),
latency_ms = latency.as_millis() as u64,
"finished processing request",
);
}
}
pub type HttpTraceLayer = tower_http::trace::TraceLayer<
tower_http::classify::SharedClassifier<tower_http::classify::ServerErrorsAsFailures>,
HttpSpan,
tower_http::trace::DefaultOnRequest,
HttpOnResponse,
>;
/// Ready-made server trace layer for hosts that build plain axum routers.
pub fn http_trace_layer() -> HttpTraceLayer {
tower_http::trace::TraceLayer::new_for_http()
.make_span_with(HttpSpan)
.on_response(HttpOnResponse)
}
/// Body-agnostic tower layer for servers whose body type must stay untouched
/// (tonic gRPC): it adopts the remote parent, creates the server span and runs
/// the inner future inside it.
#[derive(Clone, Copy, Default)]
pub struct TraceContextLayer;
impl<S> tower::Layer<S> for TraceContextLayer {
type Service = TraceContextService<S>;
fn layer(&self, inner: S) -> Self::Service {
TraceContextService { inner }
}
}
#[derive(Clone)]
pub struct TraceContextService<S> {
inner: S,
}
impl<S, B, ResBody> tower::Service<Request<B>> for TraceContextService<S>
where
S: tower::Service<Request<B>, Response = Response<ResBody>>,
S::Future: Send + 'static,
B: 'static,
{
type Response = S::Response;
type Error = S::Error;
type Future = tracing::instrument::Instrumented<S::Future>;
fn poll_ready(
&mut self,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, request: Request<B>) -> Self::Future {
let span = http_span(&request);
self.inner.call(request).instrument(span)
}
}
/// Client span for an outbound HTTP call; the trace context of this span is
/// what gets injected into the request, so the remote hop parents to it.
pub fn http_client_span(method: &str, url: &str) -> Span {
let name = format!("{method} {url}");
tracing::info_span!(
"http.client",
"otel.name" = %name,
"otel.kind" = "client",
"http.request.method" = %method,
"url.full" = %url,
"http.response.status_code" = tracing::field::Empty,
)
}
/// Sends a request inside `span` with `traceparent` injected into its headers.
pub async fn send(
span: Span,
request: reqwest::RequestBuilder,
) -> Result<reqwest::Response, reqwest::Error> {
let mut headers = HeaderMap::new();
global::get_text_map_propagator(|propagator| {
propagator.inject_context(&span.context(), &mut HeaderInjector(&mut headers));
});
if let Some(request_id) = crate::middleware::request_id()
&& let Ok(value) = request_id.parse()
{
headers.insert(crate::middleware::REQUEST_ID_HEADER, value);
}
let result = request
.headers(headers)
.send()
.instrument(span.clone())
.await;
match &result {
Ok(response) => {
span.record(
"http.response.status_code",
i64::from(response.status().as_u16()),
);
}
Err(_) => {
span.record("otel.status_code", "error");
}
}
result
}
pub async fn shutdown_signal() {
let ctrl_c = async {
if let Err(err) = tokio::signal::ctrl_c().await {
warn!(error = %err, "failed to install ctrl_c handler");
}
};
#[cfg(unix)]
let terminate = async {
use tokio::signal::unix::{SignalKind, signal};
match signal(SignalKind::terminate()) {
Ok(mut s) => {
s.recv().await;
}
Err(err) => {
warn!(error = %err, "failed to install SIGTERM handler");
}
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => info!("received SIGINT"),
_ = terminate => info!("received SIGTERM"),
}
shutdown_telemetry();
}
/// Drains spans queued in the batch processor and metrics queued in the
/// periodic reader. Without this the tail of every deployment's telemetry is
/// lost on rollout, which is exactly when you look at it.
pub fn shutdown_telemetry() {
flush_traces();
if let Some(provider) = METER_PROVIDER.get()
&& let Err(err) = provider.shutdown()
{
warn!(error = %err, "failed to shut down meter provider");
}
}
pub fn flush_traces() {
if let Some(provider) = TRACER_PROVIDER.get()
&& let Err(err) = provider.shutdown()
{
warn!(error = %err, "failed to flush tracer provider");
}
}