feat(telemetry): propagate trace context across HTTP and gRPC hops

Modified app/adminhost/src/main.rs +4 −8 View app/adminhost/src/main.rs

@@ -5,8 +5,8 @@

55use axum::Router;

66use config::scoped::{AdminhostConfig, ScopedConfig};

77use tokio::net::TcpListener;

8use tower_http::trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer};

9use tracing::{Level, info, warn};

8use tower_http::trace::TraceLayer;

9use tracing::{info, warn};

1010

1111#[tokio::main]

1212async fn main() -> anyhow::Result<()> {

@@ -58,12 +58,8 @@

5858 let router: Router = health

5959 .layer(

6060 TraceLayer::new_for_http()

61 .make_span_with(

62 DefaultMakeSpan::new()

63 .level(Level::INFO)

64 .include_headers(false),

65 )

66 .on_response(DefaultOnResponse::new().level(Level::INFO)),

61 .make_span_with(export::telemetry::HttpSpan)

62 .on_response(export::telemetry::HttpOnResponse),

6763 )

6864 .layer(export::middleware::AppTraceLayer::new());

6965

Modified app/im/src/main.rs +3 −1 View app/im/src/main.rs

@@ -10,6 +10,7 @@

1010use config::scoped::{ImConfig, ScopedConfig};

1111use hyper_util::rt::TokioIo;

1212use tokio::net::TcpListener;

13use tracing::Instrument as _;

1314use tracing::info;

1415

1516#[tokio::main]

@@ -51,7 +52,8 @@

5152 let io = TokioIo::new(stream);

5253 let service = hyper::service::service_fn(move |req| {

5354 let service = service.clone();

54 async move { gateway::route(service, req).await }

55 let span = export::telemetry::http_span(&req);

56 async move { gateway::route(service, req).await }.instrument(span)

5557 });

5658 if let Err(error) = hyper::server::conn::http1::Builder::new()

5759 .serve_connection(io, service)

Modified app/mcp/src/main.rs +4 −8 View app/mcp/src/main.rs

@@ -6,8 +6,8 @@

66use config::scoped::{McpHostConfig, ScopedConfig};

77use tokio::net::TcpListener;

88use tower_http::limit::RequestBodyLimitLayer;

9use tower_http::trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer};

10use tracing::{Level, info};

9use tower_http::trace::TraceLayer;

10use tracing::info;

1111

1212const MAX_REQUEST_BODY_BYTES: usize = 2 * 1024 * 1024;

1313

@@ -37,12 +37,8 @@

3737 .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_BYTES))

3838 .layer(

3939 TraceLayer::new_for_http()

40 .make_span_with(

41 DefaultMakeSpan::new()

42 .level(Level::INFO)

43 .include_headers(false),

44 )

45 .on_response(DefaultOnResponse::new().level(Level::INFO)),

40 .make_span_with(export::telemetry::HttpSpan)

41 .on_response(export::telemetry::HttpOnResponse),

4642 )

4743 .layer(export::middleware::AppTraceLayer::new());

4844

Modified app/webhook/src/main.rs +2 −1 View app/webhook/src/main.rs

@@ -68,7 +68,8 @@

6868 .merge(export::health::router_with(

6969 Some(service.state.database.clone()),

7070 service.state.redis.clone(),

71 ));

71 ))

72 .layer(export::telemetry::http_trace_layer());

7273 info!(%addr, "webhook scheduler listening");

7374

7475 let server_cancel = cancel.clone();

Modified lib/admin/src/services/system.rs +2 −1 View lib/admin/src/services/system.rs

@@ -190,7 +190,8 @@

190190 };

191191 }

192192 };

193 match client.get(url).send().await {

193 let span = export::telemetry::http_client_span("GET", &url);

194 match export::telemetry::send(span, client.get(url)).await {

194195 Ok(response) => DependencyDto {

195196 ok: response.status().is_success(),

196197 latency_ms: Some(latency_ms(start)),

Modified lib/admin-rpc/src/serve.rs +1 −0 View lib/admin-rpc/src/serve.rs

@@ -16,6 +16,7 @@

1616) -> Result<(), tonic::transport::Error> {

1717 let interceptor = TrustedBackend::new(trusted_keys);

1818 Server::builder()

19 .layer(export::telemetry::TraceContextLayer)

1920 .add_service(AdminServer::with_interceptor(

2021 AdminRpc::new(service),

2122 interceptor,

Modified lib/agent/src/client.rs +2 −2 View lib/agent/src/client.rs

@@ -278,8 +278,8 @@

278278 if !self.api_key.is_empty() {

279279 request = request.bearer_auth(&self.api_key);

280280 }

281 let response = request

282 .send()

281 let span = export::telemetry::http_client_span("POST", &self.endpoint);

282 let response = export::telemetry::send(span, request)

283283 .await

284284 .map_err(|e| AgentError::Transport(transport_message(&e)))?;

285285 let status = response.status();

Modified lib/chat/src/services/outbound.rs +13 −10 View lib/chat/src/services/outbound.rs

@@ -540,16 +540,19 @@

540540 }

541541 let client = client_builder.build().map_err(|e| e.to_string())?;

542542

543 let response = client

544 .post(&webhook.url)

545 .header("content-type", "application/json")

546 .header("x-gitdata-event", &delivery.event_type)

547 .header("x-gitdata-delivery", delivery.id.to_string())

548 .header("x-gitdata-signature", &signature)

549 .body(body)

550 .send()

551 .await

552 .map_err(|e| format!("request failed: {e}"))?;

543 let span = export::telemetry::http_client_span("POST", &webhook.url);

544 let response = export::telemetry::send(

545 span,

546 client

547 .post(&webhook.url)

548 .header("content-type", "application/json")

549 .header("x-gitdata-event", &delivery.event_type)

550 .header("x-gitdata-delivery", delivery.id.to_string())

551 .header("x-gitdata-signature", &signature)

552 .body(body),

553 )

554 .await

555 .map_err(|e| format!("request failed: {e}"))?;

553556 let status = response.status().as_u16();

554557 if (200..300).contains(&status) {

555558 Ok(format!("delivered {status}"))

Modified lib/ci/src/outbound.rs +13 −10 View lib/ci/src/outbound.rs

@@ -227,16 +227,19 @@

227227 builder = builder.resolve(&host, *addr);

228228 }

229229 let client = builder.build().map_err(|e| e.to_string())?;

230 let response = client

231 .post(url)

232 .header("content-type", "application/json")

233 .header("x-gitdata-event", &delivery.event_type)

234 .header("x-gitdata-delivery", delivery.id.to_string())

235 .header("x-gitdata-signature", signature)

236 .body(body)

237 .send()

238 .await

239 .map_err(|e| format!("request failed: {e}"))?;

230 let span = export::telemetry::http_client_span("POST", url);

231 let response = export::telemetry::send(

232 span,

233 client

234 .post(url)

235 .header("content-type", "application/json")

236 .header("x-gitdata-event", &delivery.event_type)

237 .header("x-gitdata-delivery", delivery.id.to_string())

238 .header("x-gitdata-signature", signature)

239 .body(body),

240 )

241 .await

242 .map_err(|e| format!("request failed: {e}"))?;

240243 let status = response.status().as_u16() as i32;

241244 let text = response

242245 .text()

Modified lib/export/src/telemetry.rs +97 −10 View lib/export/src/telemetry.rs

@@ -3,10 +3,10 @@

33

44use axum::extract::MatchedPath;

55use axum::http::{HeaderMap, Request, Response};

6use opentelemetry::Context;

67use opentelemetry::global;

78use opentelemetry::propagation::TextMapCompositePropagator;

89use opentelemetry::trace::TracerProvider as _;

9use opentelemetry::Context;

1010use opentelemetry_http::{HeaderExtractor, HeaderInjector};

1111use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality};

1212use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator};

@@ -103,13 +103,11 @@

103103 OtlpSignal::Traces => "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",

104104 OtlpSignal::Metrics => "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT",

105105 };

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 })

106 ["OTEL_EXPORTER_OTLP_ENDPOINT", specific].iter().any(|key| {

107 std::env::var(key)

108 .map(|value| !value.trim().is_empty())

109 .unwrap_or(false)

110 })

113111}

114112

115113fn tracer_provider() -> Option<SdkTracerProvider> {

@@ -117,7 +115,10 @@

117115 return None;

118116 }

119117

120 let exporter = match opentelemetry_otlp::SpanExporter::builder().with_http().build() {

118 let exporter = match opentelemetry_otlp::SpanExporter::builder()

119 .with_http()

120 .build()

121 {

121122 Ok(exporter) => exporter,

122123 Err(err) => {

123124 eprintln!("failed to build otlp span exporter, traces disabled: {err}");

@@ -156,7 +157,10 @@

156157pub fn traceparent() -> Option<String> {

157158 let mut headers = HeaderMap::new();

158159 global::get_text_map_propagator(|propagator| {

159 propagator.inject_context(&Span::current().context(), &mut HeaderInjector(&mut headers));

160 propagator.inject_context(

161 &Span::current().context(),

162 &mut HeaderInjector(&mut headers),

163 );

160164 });

161165 headers

162166 .get("traceparent")

@@ -199,6 +203,89 @@

199203 }

200204}

201205

206/// `tower_http::trace::MakeSpan` adapter shared by every axum/tonic host.

207#[derive(Clone, Copy, Default)]

208pub struct HttpSpan;

209

210impl<B> tower_http::trace::MakeSpan<B> for HttpSpan {

211 fn make_span(&mut self, request: &Request<B>) -> Span {

212 http_span(request)

213 }

214}

215

216/// `tower_http::trace::OnResponse` adapter keeping the request log line and

217/// recording the HTTP semantic-convention attributes on the span.

218#[derive(Clone, Copy, Default)]

219pub struct HttpOnResponse;

220

221impl<B> tower_http::trace::OnResponse<B> for HttpOnResponse {

222 fn on_response(self, response: &Response<B>, latency: Duration, span: &Span) {

223 on_response(response, latency, span);

224 tracing::info!(

225 parent: span,

226 status = response.status().as_u16(),

227 latency_ms = latency.as_millis() as u64,

228 "finished processing request",

229 );

230 }

231}

232

233pub type HttpTraceLayer = tower_http::trace::TraceLayer<

234 tower_http::classify::SharedClassifier<tower_http::classify::ServerErrorsAsFailures>,

235 HttpSpan,

236 tower_http::trace::DefaultOnRequest,

237 HttpOnResponse,

238>;

239

240/// Ready-made server trace layer for hosts that build plain axum routers.

241pub fn http_trace_layer() -> HttpTraceLayer {

242 tower_http::trace::TraceLayer::new_for_http()

243 .make_span_with(HttpSpan)

244 .on_response(HttpOnResponse)

245}

246

247/// Body-agnostic tower layer for servers whose body type must stay untouched

248/// (tonic gRPC): it adopts the remote parent, creates the server span and runs

249/// the inner future inside it.

250#[derive(Clone, Copy, Default)]

251pub struct TraceContextLayer;

252

253impl<S> tower::Layer<S> for TraceContextLayer {

254 type Service = TraceContextService<S>;

255

256 fn layer(&self, inner: S) -> Self::Service {

257 TraceContextService { inner }

258 }

259}

260

261#[derive(Clone)]

262pub struct TraceContextService<S> {

263 inner: S,

264}

265

266impl<S, B, ResBody> tower::Service<Request<B>> for TraceContextService<S>

267where

268 S: tower::Service<Request<B>, Response = Response<ResBody>>,

269 S::Future: Send + 'static,

270 B: 'static,

271{

272 type Response = S::Response;

273 type Error = S::Error;

274 type Future = tracing::instrument::Instrumented<S::Future>;

275

276 fn poll_ready(

277 &mut self,

278 cx: &mut std::task::Context<'_>,

279 ) -> std::task::Poll<Result<(), Self::Error>> {

280 self.inner.poll_ready(cx)

281 }

282

283 fn call(&mut self, request: Request<B>) -> Self::Future {

284 let span = http_span(&request);

285 self.inner.call(request).instrument(span)

286 }

287}

288

202289/// Client span for an outbound HTTP call; the trace context of this span is

203290/// what gets injected into the request, so the remote hop parents to it.

204291pub fn http_client_span(method: &str, url: &str) -> Span {

Modified lib/export/Cargo.toml +1 −0 View lib/export/Cargo.toml

@@ -26,6 +26,7 @@

2626utoipa = { workspace = true }

2727rand.workspace = true

2828tower = { workspace = true }

29tower-http = { workspace = true }

2930serde_json = { workspace = true }

3031uuid = { workspace = true }

3132bytes = { workspace = true }

Modified lib/git/src/smart/handler.rs +1 −0 View lib/git/src/smart/handler.rs

@@ -24,6 +24,7 @@

2424 .head(dispatch),

2525 )

2626 .with_state(service)

27 .layer(export::telemetry::http_trace_layer())

2728}

2829

2930async fn dispatch(

Modified lib/push/src/client.rs +2 −3 View lib/push/src/client.rs

@@ -140,9 +140,8 @@

140140 for (name, value) in request.headers() {

141141 outbound = outbound.header(name.as_str(), value.as_bytes());

142142 }

143 let response = outbound

144 .body(request.into_body())

145 .send()

143 let span = export::telemetry::http_client_span("POST", &request.uri().to_string());

144 let response = export::telemetry::send(span, outbound.body(request.into_body()))

146145 .await

147146 .map_err(|error| PushError::Transport(error.to_string()))?;

148147

Modified lib/sandbox-rpc/src/lib.rs +23 −0 View lib/sandbox-rpc/src/lib.rs

@@ -5,13 +5,18 @@

55use std::time::Duration;

66

77use config::sandbox_rpc::SandboxRpcConfig;

8use opentelemetry::global;

9use opentelemetry::propagation::Injector;

810use tonic::metadata::{Ascii, MetadataValue};

911use tonic::service::Interceptor;

1012use tonic::service::interceptor::InterceptedService;

1113use tonic::transport::{Channel, Endpoint};

1214use tonic::{Request, Status};

15use tracing::Span;

16use tracing_opentelemetry::OpenTelemetrySpanExt;

1317

1418pub const TRUSTED_KEY_HEADER: &str = "x-trusted-key";

19pub const TRACEPARENT_HEADER: &str = "traceparent";

1520pub const SCOPE_STREAM: &str = "stream";

1621

1722#[derive(Debug, thiserror::Error)]

@@ -49,11 +54,29 @@

4954 value: MetadataValue<Ascii>,

5055}

5156

57#[derive(Default)]

58struct TraceCarrier(std::collections::HashMap<String, String>);

59

60impl Injector for TraceCarrier {

61 fn set(&mut self, key: &str, value: String) {

62 self.0.insert(key.to_string(), value);

63 }

64}

65

5266impl Interceptor for TrustedKey {

5367 fn call(&mut self, mut request: Request<()>) -> Result<Request<()>, Status> {

5468 request

5569 .metadata_mut()

5670 .insert(TRUSTED_KEY_HEADER, self.value.clone());

71 let mut carrier = TraceCarrier::default();

72 global::get_text_map_propagator(|propagator| {

73 propagator.inject_context(&Span::current().context(), &mut carrier);

74 });

75 if let Some(value) = carrier.0.get(TRACEPARENT_HEADER)

76 && let Ok(value) = MetadataValue::try_from(value.as_str())

77 {

78 request.metadata_mut().insert(TRACEPARENT_HEADER, value);

79 }

5780 Ok(request)

5881 }

5982}

Modified lib/sandbox-rpc/Cargo.toml +2 −0 View lib/sandbox-rpc/Cargo.toml

@@ -5,11 +5,13 @@

55

66[dependencies]

77config = { workspace = true }

8opentelemetry = { workspace = true }

89prost = { workspace = true }

910thiserror = { workspace = true }

1011tonic = { workspace = true }

1112tonic-prost = { workspace = true }

1213tracing = { workspace = true }

14tracing-opentelemetry = { workspace = true }

1315

1416[build-dependencies]

1517tonic-prost-build = { workspace = true }

No newline at end of file

Modified lib/webapi/src/lib.rs +3 −8 View lib/webapi/src/lib.rs

@@ -8,8 +8,7 @@

88use export::middleware::AppTraceLayer;

99use session_layer::{SessionLayer, store};

1010use tower_http::limit::RequestBodyLimitLayer;

11use tower_http::trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer};

12use tracing::Level;

11use tower_http::trace::TraceLayer;

1312

1413const MAX_REQUEST_BODY_BYTES: usize = 2 * 1024 * 1024;

1514

@@ -101,12 +100,8 @@

101100 .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_BYTES))

102101 .layer(

103102 TraceLayer::new_for_http()

104 .make_span_with(

105 DefaultMakeSpan::new()

106 .level(Level::INFO)

107 .include_headers(false),

108 )

109 .on_response(DefaultOnResponse::new().level(Level::INFO)),

103 .make_span_with(export::telemetry::HttpSpan)

104 .on_response(export::telemetry::HttpOnResponse),

110105 )

111106 .layer(export::telemetry::metrics_layer())

112107 .layer(AppTraceLayer::new());

Modified Cargo.lock +3 −0 View Cargo.lock

@@ -2471,6 +2471,7 @@

24712471 "time 0.3.55",

24722472 "tokio",

24732473 "tower",

2474 "tower-http 0.7.0",

24742475 "tracing",

24752476 "tracing-opentelemetry",

24762477 "tracing-subscriber",

@@ -8314,12 +8315,14 @@

83148315version = "1.0.0"

83158316dependencies = [

83168317 "config",

8318 "opentelemetry",

83178319 "prost 0.14.4",

83188320 "thiserror 2.0.20",

83198321 "tonic",

83208322 "tonic-prost",

83218323 "tonic-prost-build",

83228324 "tracing",

8325 "tracing-opentelemetry",

83238326]

83248327

83258328[[package]]