Extract W3C traceparent header and set it as Span parent

Author: ncoiffier-celonis

quickwit/Cargo.lock

@@ -285,6 +285,7 @@ tower-http = { version = "0.6", features = [
   "compression-gzip",
   "compression-zstd",
   "cors",
+  "trace",
 ] }
 tracing = "0.1"
 tracing-opentelemetry = "0.32"

quickwit/Cargo.toml

@@ -207,6 +207,24 @@ pub fn set_parent_span_from_request_metadata(request_metadata: &tonic::metadata:
     let _ = Span::current().set_parent(parent_cx);
 }
 
+/// `HeaderMap` extracts OpenTelemetry tracing keys from HTTP headers.
+struct HeaderMap<'a>(&'a http::HeaderMap);
+
+impl Extractor for HeaderMap<'_> {
+    fn get(&self, key: &str) -> Option<&str> {
+        self.0.get(key).and_then(|metadata| metadata.to_str().ok())
+    }
+
+    fn keys(&self) -> Vec<&str> {
+        self.0.keys().map(|key| key.as_str()).collect()
+    }
+}
+
+/// Extracts an OpenTelemetry context from HTTP [`http::HeaderMap`].
+pub fn extract_context_from_request_headers(headers: &http::HeaderMap) -> ::opentelemetry::Context {
+    global::get_text_map_propagator(|prop| prop.extract(&HeaderMap(headers)))
+}
+
 impl search::SortOrder {
     #[inline(always)]
     pub fn compare_opt<T: Ord>(&self, this: &Option<T>, other: &Option<T>) -> Ordering {

quickwit/quickwit-proto/src/lib.rs

@@ -52,6 +52,7 @@ tonic-reflection = { workspace = true }
 tower = { workspace = true, features = ["limit"] }
 tower-http = { workspace = true }
 tracing = { workspace = true }
+tracing-opentelemetry = { workspace = true }
 utoipa = { workspace = true }
 warp = { workspace = true, features = ["server"] }
 zstd = { workspace = true }
@@ -84,10 +85,14 @@ assert-json-diff = { workspace = true }
 http = { workspace = true }
 itertools = { workspace = true }
 mockall = { workspace = true }
+opentelemetry = { workspace = true }
+opentelemetry_sdk = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }
 tonic = { workspace = true }
+tracing-opentelemetry = { workspace = true }
+tracing-subscriber = { workspace = true }
 
 quickwit-actors = { workspace = true, features = ["testsuite"] }
 quickwit-cluster = { workspace = true, features = ["testsuite"] }

quickwit/quickwit-serve/Cargo.toml

@@ -21,6 +21,7 @@ use hyper_util::server::conn::auto::Builder;
 use hyper_util::service::TowerToHyperService;
 use quickwit_common::tower::BoxFutureInfaillible;
 use quickwit_config::{disable_ingest_v1, enable_ingest_v2};
+use quickwit_proto::extract_context_from_request_headers;
 use quickwit_search::SearchService;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::{TcpListener, TcpStream};
@@ -30,7 +31,9 @@ use tower::ServiceBuilder;
 use tower_http::compression::CompressionLayer;
 use tower_http::compression::predicate::{NotForContentType, Predicate, SizeAbove};
 use tower_http::cors::{AllowOrigin, CorsLayer};
-use tracing::{error, info};
+use tower_http::trace::TraceLayer;
+use tracing::{Level, error, info};
+use tracing_opentelemetry::OpenTelemetrySpanExt;
 use warp::filters::log::Info;
 use warp::hyper::http::HeaderValue;
 use warp::hyper::{Method, StatusCode, http};
@@ -208,7 +211,21 @@ pub(crate) async fn start_rest_server(
     let compression_predicate = CompressionPredicate::from_env().and(NotForContentType::IMAGES);
     let cors = build_cors(&quickwit_services.node_config.rest_config.cors_allow_origins);
 
+    let trace_layer = TraceLayer::new_for_http().make_span_with(|request: &http::Request<_>| {
+        let span = tracing::span!(
+            Level::INFO,
+            "http_request",
+            otel.kind = "Server",
+            http.method = %request.method(),
+            http.target = %request.uri(),
+        );
+        let ctx = extract_context_from_request_headers(request.headers());
+        let _ = span.set_parent(ctx);
+        span
+    });
+
     let service = ServiceBuilder::new()
+        .layer(trace_layer)
         .layer(
             CompressionLayer::new()
                 .zstd(true)