Propagate OpenTelemetry trace headers into request headers

Author: LarryOsterman

Cargo.lock

@@ -188,6 +188,9 @@ impl Policy for RequestInstrumentationPolicy {
                 span.set_attribute(HTTP_REQUEST_RESEND_COUNT_ATTRIBUTE, retry_count.0.into());
             }
 
+            // Propagate the headers for distributed tracing into the request.
+            span.propagate_headers(request);
+
             let result = next[0].send(ctx, request, &next[1..]).await;
 
             if let Some(err) = result.as_ref().err() {
@@ -242,6 +245,7 @@ mod tests {
     use azure_core_test::http::MockHttpClient;
     use futures::future::BoxFuture;
     use std::sync::{Arc, Mutex};
+    use typespec_client_core::http::headers::HeaderName;
 
     #[derive(Debug)]
     struct MockTracingProvider {
@@ -386,7 +390,15 @@ mod tests {
             todo!()
         }
 
-        fn propagate_headers(&self, _request: &mut Request) {}
+        /// Insert two dummy headers for distributed tracing.
+        // cspell: ignore traceparent tracestate
+        fn propagate_headers(&self, request: &mut Request) {
+            request.insert_header(
+                HeaderName::from_static("traceparent"),
+                "00-<trace_id>-<span_id>-01",
+            );
+            request.insert_header(HeaderName::from_static("tracestate"), "<key>=<value>");
+        }
     }
 
     impl AsAny for MockSpan {
@@ -546,6 +558,11 @@ mod tests {
                 Box::pin(async move {
                     assert_eq!(req.url().host_str(), Some("example.com"));
                     assert_eq!(req.method(), &Method::Get);
+                    assert_eq!(
+                        req.headers()
+                            .get_optional_str(HeaderName::from_static("traceparent")),
+                        Some("00-<trace_id>-<span_id>-01")
+                    );
                     Ok(RawResponse::from_bytes(
                         StatusCode::Ok,
                         Headers::new(),

sdk/core/azure_core/src/http/policies/request_instrumentation.rs

@@ -18,9 +18,12 @@ edition.workspace = true
 azure_core.workspace = true
 opentelemetry = { version = "0.30", features = ["trace"] }
 opentelemetry-http = "0.30.0"
+opentelemetry_sdk = "0.30"
+reqwest.workspace = true
 tracing.workspace = true
 typespec_client_core.workspace = true
 
+
 [dev-dependencies]
 azure_core_test = { workspace = true, features = ["tracing"] }
 azure_core_test_macros.workspace = true

sdk/core/azure_core_opentelemetry/Cargo.toml

@@ -8,7 +8,10 @@ use azure_core::{
     tracing::{AsAny, AttributeValue, Span, SpanGuard, SpanStatus},
     Result,
 };
-use opentelemetry::trace::TraceContextExt;
+use opentelemetry::{propagation::TextMapPropagator, trace::TraceContextExt};
+use opentelemetry_http::HeaderInjector;
+use opentelemetry_sdk::propagation::TraceContextPropagator;
+use reqwest::header::HeaderMap;
 use std::{error::Error as StdError, sync::Arc};
 
 /// newtype for Azure Core SpanKind to enable conversion to OpenTelemetry SpanKind
@@ -76,7 +79,34 @@ impl Span for OpenTelemetrySpan {
         self.context.span().set_status(otel_status);
     }
 
-    fn propagate_headers(&self, _request: &mut azure_core::http::Request) {}
+    fn propagate_headers(&self, request: &mut azure_core::http::Request) {
+        // A TraceContextPropagator is used to inject trace context information into HTTP headers.
+        let trace_propagator = TraceContextPropagator::new();
+        // We need to map between a reqwest header map (which is what the OpenTelemetry SDK requires)
+        // and the Azure Core request headers.
+        //
+        // We start with an empty header map and inject the OpenTelemetry headers into it.
+        let mut header_map = HeaderMap::new();
+        trace_propagator.inject_context(&self.context, &mut HeaderInjector(&mut header_map));
+
+        // We then insert each of the headers from the OpenTelemetry header map into the
+        // Request's header map.
+        for (key, value) in header_map.into_iter() {
+            // Note: The OpenTelemetry HeaderInjector will always produce unique header names, so we don't need to
+            // handle the multiple headers case here.
+            if let Some(key) = key {
+                // Convert HeaderName to &str for insertion.
+                let value_str = value.to_str().unwrap().to_string();
+                request.insert_header(
+                    azure_core::http::headers::HeaderName::from(key.to_string()),
+                    azure_core::http::headers::HeaderValue::from(value_str),
+                );
+            } else {
+                // If the key is invalid, we skip it
+                tracing::warn!("Invalid header key: {:?}", key);
+            }
+        }
+    }
 
     fn set_current(
         &self,
@@ -117,7 +147,7 @@ impl Drop for OpenTelemetrySpanGuard {
 #[cfg(test)]
 mod tests {
     use crate::telemetry::OpenTelemetryTracerProvider;
-    use azure_core::http::Context as AzureContext;
+    use azure_core::http::{Context as AzureContext, Url};
     use azure_core::tracing::{Attribute, AttributeValue, SpanKind, SpanStatus, TracerProvider};
     use opentelemetry::trace::TraceContextExt;
     use opentelemetry::{Context, Key, KeyValue, Value};
@@ -158,6 +188,34 @@ mod tests {
         }
     }
 
+    // cspell: ignore traceparent tracestate
+    #[test]
+    fn test_open_telemetry_span_propagate() {
+        let (otel_tracer_provider, otel_exporter) = create_exportable_tracer_provider();
+
+        let tracer_provider = OpenTelemetryTracerProvider::new(otel_tracer_provider);
+        assert!(tracer_provider.is_ok());
+        let tracer =
+            tracer_provider
+                .unwrap()
+                .get_tracer(Some("Microsoft.SpecialCase"), "test", "0.1.0");
+        let span = tracer.start_span("test_span", SpanKind::Client, vec![]);
+        let mut request = azure_core::http::Request::new(
+            Url::parse("http://example.com").unwrap(),
+            azure_core::http::Method::Get,
+        );
+        span.propagate_headers(&mut request);
+        trace!("Request headers after propagation: {:?}", request.headers());
+        let traceparent = azure_core::http::headers::HeaderName::from("traceparent");
+        let tracestate = azure_core::http::headers::HeaderName::from("tracestate");
+        request.headers().get_as::<String, _>(&traceparent).unwrap();
+        request.headers().get_as::<String, _>(&tracestate).unwrap();
+        span.end();
+
+        let finished_spans = otel_exporter.get_finished_spans().unwrap();
+        assert_eq!(finished_spans.len(), 1);
+    }
+
     #[test]
     fn test_open_telemetry_span_hierarchy() {
         let (otel_tracer_provider, otel_exporter) = create_exportable_tracer_provider();