chore(tracing): Use upstream tracing propagation

The proxy used its own custom trace propagation for some time. It was built before the upstream OpenTelemetry libraries were available.

This replaces the manual propagation with a combination of the upstream w3c and b3 propagators, modified slightly to only propagate in one format or the other (and not both).

Signed-off-by: Scott Fleener <[email protected]>

Author: sfleen

Cargo.lock

@@ -2084,6 +2084,7 @@ dependencies = [
  "opentelemetry",
  "opentelemetry-proto",
  "opentelemetry-semantic-conventions",
+ "opentelemetry-zipkin",
  "opentelemetry_sdk",
  "tokio",
  "tonic",
@@ -2603,12 +2604,22 @@ dependencies = [
  "futures",
  "hex",
  "http",
+ "linkerd-app-core",
  "linkerd-error",
+ "linkerd-http-box",
+ "linkerd-io",
+ "linkerd-opentelemetry",
  "linkerd-stack",
+ "linkerd-tracing",
+ "opentelemetry",
+ "opentelemetry-http",
  "opentelemetry-semantic-conventions",
+ "opentelemetry_sdk",
  "rand 0.9.2",
  "thiserror",
+ "tokio",
  "tower",
+ "tower-test",
  "tracing",
 ]
 
@@ -2943,6 +2954,18 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "opentelemetry-http"
+version = "0.31.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "http",
+ "opentelemetry",
+]
+
 [[package]]
 name = "opentelemetry-proto"
 version = "0.31.0"
@@ -2966,6 +2989,23 @@ version = "0.31.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e62e29dfe041afb8ed2a6c9737ab57db4907285d999ef8ad3a59092a36bdc846"
 
+[[package]]
+name = "opentelemetry-zipkin"
+version = "0.31.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f27fcd074586dab55936b003c6a499acaabd6debbd539c3f36356bca2ef2fce2"
+dependencies = [
+ "http",
+ "once_cell",
+ "opentelemetry",
+ "opentelemetry-http",
+ "opentelemetry_sdk",
+ "serde",
+ "serde_json",
+ "thiserror",
+ "typed-builder",
+]
+
 [[package]]
 name = "opentelemetry_sdk"
 version = "0.31.0"
@@ -4220,6 +4260,26 @@ version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
 
+[[package]]
+name = "typed-builder"
+version = "0.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd9d30e3a08026c78f246b173243cf07b3696d274debd26680773b6773c2afc7"
+dependencies = [
+ "typed-builder-macro",
+]
+
+[[package]]
+name = "typed-builder-macro"
+version = "0.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3c36781cc0e46a83726d9879608e4cf6c2505237e263a8eb8c24502989cfdb28"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "typenum"
 version = "1.19.0"

linkerd/opentelemetry/Cargo.toml

@@ -15,6 +15,7 @@ linkerd-trace-context = { path = "../trace-context" }
 opentelemetry = { version = "0.31", default-features = false, features = ["trace"] }
 opentelemetry_sdk = { version = "0.31", default-features = false, features = ["trace"] }
 opentelemetry-proto = { version = "0.31" }
+opentelemetry-zipkin = { version = "0.31", default-features = false }
 opentelemetry-semantic-conventions = { version = "0.31", default-features = false, features = ["semconv_experimental"] }
 tonic = { workspace = true, default-features = false, features = [
     "codegen",

linkerd/opentelemetry/src/lib.rs

@@ -2,15 +2,17 @@
 #![forbid(unsafe_code)]
 
 pub mod metrics;
+pub mod propagation;
 
 use self::metrics::Registry;
+use crate::propagation::OrderedPropagator;
 use futures::stream::{Stream, StreamExt};
 use http_body::Body;
 use linkerd_error::Error;
 use linkerd_trace_context::{self as trace_context, export::ExportSpan};
 pub use opentelemetry as otel;
 use opentelemetry::{
-    trace::{SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState},
+    trace::{SpanContext, SpanKind, Status, TraceFlags, TraceState},
     KeyValue,
 };
 pub use opentelemetry_proto as proto;
@@ -56,6 +58,8 @@ where
     )
     .build();
 
+    opentelemetry::global::set_text_map_propagator(OrderedPropagator::new());
+
     SpanExportTask::new(spans, processor).run().await;
 }
 
@@ -151,12 +155,12 @@ fn convert_span(span: ExportSpan) -> Result<SpanData, Error> {
     for (k, v) in labels.iter() {
         attributes.push(KeyValue::new(k.clone(), v.clone()));
     }
-    for (k, v) in span.labels.into_iter() {
-        attributes.push(KeyValue::new(k, v));
+    for kv in span.labels.into_iter() {
+        attributes.push(kv);
     }
     let is_remote = kind != trace_context::export::SpanKind::Client;
     Ok(SpanData {
-        parent_span_id: SpanId::from_bytes(span.parent_id.into_bytes()?),
+        parent_span_id: span.parent_id,
         parent_span_is_remote: true, // we do not originate any spans locally
         span_kind: match kind {
             trace_context::export::SpanKind::Server => SpanKind::Server,
@@ -170,8 +174,8 @@ fn convert_span(span: ExportSpan) -> Result<SpanData, Error> {
         links: SpanLinks::default(),
         status: Status::Unset, // TODO: this is gRPC status; we must read response trailers to populate this
         span_context: SpanContext::new(
-            TraceId::from_bytes(span.trace_id.into_bytes()?),
-            SpanId::from_bytes(span.span_id.into_bytes()?),
+            span.trace_id,
+            span.span_id,
             TraceFlags::default(),
             is_remote,
             TraceState::NONE,

linkerd/opentelemetry/src/propagation.rs

@@ -0,0 +1,79 @@
+use opentelemetry::{
+    propagation::{text_map_propagator::FieldIter, Extractor, Injector, TextMapPropagator},
+    Context,
+};
+use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator};
+
+#[derive(Copy, Clone, Eq, PartialEq)]
+enum PropagationFormat {
+    W3C,
+    B3,
+}
+
+#[derive(Debug)]
+pub struct OrderedPropagator {
+    w3c: TraceContextPropagator,
+    b3: opentelemetry_zipkin::Propagator,
+    baggage: BaggagePropagator,
+    fields: Vec<String>,
+}
+
+impl OrderedPropagator {
+    pub fn new() -> Self {
+        let w3c = TraceContextPropagator::new();
+        let b3 = opentelemetry_zipkin::Propagator::new();
+        let baggage = BaggagePropagator::new();
+
+        Self {
+            fields: w3c
+                .fields()
+                .chain(b3.fields())
+                .chain(baggage.fields())
+                .map(|s| s.to_string())
+                .collect(),
+            w3c,
+            b3,
+            baggage,
+        }
+    }
+}
+
+impl Default for OrderedPropagator {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl TextMapPropagator for OrderedPropagator {
+    fn inject_context(&self, cx: &Context, injector: &mut dyn Injector) {
+        match cx.get::<PropagationFormat>() {
+            None => {}
+            Some(PropagationFormat::W3C) => {
+                self.w3c.inject_context(cx, injector);
+            }
+            Some(PropagationFormat::B3) => {
+                self.b3.inject_context(cx, injector);
+            }
+        }
+        self.baggage.inject_context(cx, injector);
+    }
+
+    fn extract_with_context(&self, cx: &Context, extractor: &dyn Extractor) -> Context {
+        let cx = if self.w3c.fields().any(|f| extractor.get(f).is_some()) {
+            self.w3c
+                .extract_with_context(cx, extractor)
+                .with_value(PropagationFormat::W3C)
+        } else if self.b3.fields().any(|f| extractor.get(f).is_some()) {
+            self.b3
+                .extract_with_context(cx, extractor)
+                .with_value(PropagationFormat::B3)
+        } else {
+            cx.clone()
+        };
+        self.baggage.extract_with_context(&cx, extractor)
+    }
+
+    fn fields(&self) -> FieldIter<'_> {
+        FieldIter::new(self.fields.as_slice())
+    }
+}

linkerd/trace-context/Cargo.toml

@@ -15,8 +15,20 @@ hex = "0.4"
 http = { workspace = true }
 linkerd-error = { path = "../error" }
 linkerd-stack = { path = "../stack" }
+opentelemetry = { version = "0.31", default-features = false, features = ["trace"] }
+opentelemetry_sdk = { version = "0.31", default-features = false, features = ["trace"] }
+opentelemetry-http = { version = "0.31", default-features = false }
 opentelemetry-semantic-conventions = { version = "0.31", default-features = false, features = ["semconv_experimental"] }
 rand = "0.9"
 thiserror = "2"
 tower = { workspace = true, default-features = false, features = ["util"] }
 tracing = { workspace = true }
+
+[dev-dependencies]
+linkerd-app-core = { path = "../app/core" }
+linkerd-http-box = { path = "../http/box" }
+linkerd-io = { path = "../io" }
+linkerd-tracing = { path = "../tracing" }
+linkerd-opentelemetry = { path = "../opentelemetry" }
+tokio = { version = "1", features = ["test-util"] }
+tower-test = { workspace = true }

linkerd/trace-context/src/export.rs

@@ -8,6 +8,15 @@ pub enum SpanKind {
     Client = 2,
 }
 
+impl From<SpanKind> for opentelemetry::trace::SpanKind {
+    fn from(value: SpanKind) -> Self {
+        match value {
+            SpanKind::Server => opentelemetry::trace::SpanKind::Server,
+            SpanKind::Client => opentelemetry::trace::SpanKind::Client,
+        }
+    }
+}
+
 pub type SpanLabels = Arc<HashMap<String, String>>;
 
 #[derive(Debug)]

linkerd/trace-context/src/lib.rs

@@ -2,20 +2,16 @@
 #![forbid(unsafe_code)]
 
 pub mod export;
-mod propagation;
 mod service;
 
 pub use self::service::TraceContext;
 use bytes::Bytes;
 use linkerd_error::Error;
-use rand::Rng;
-use std::collections::HashMap;
+use opentelemetry::{KeyValue, SpanId, TraceId};
 use std::fmt;
 use std::time::SystemTime;
 use thiserror::Error;
 
-const SPAN_ID_LEN: usize = 8;
-
 #[derive(Debug, Default)]
 pub struct Id(Vec<u8>);
 
@@ -49,13 +45,13 @@ pub struct InsufficientBytes;
 
 #[derive(Debug)]
 pub struct Span {
-    pub trace_id: Id,
-    pub span_id: Id,
-    pub parent_id: Id,
+    pub trace_id: TraceId,
+    pub span_id: SpanId,
+    pub parent_id: SpanId,
     pub span_name: String,
     pub start: SystemTime,
     pub end: SystemTime,
-    pub labels: HashMap<&'static str, String>,
+    pub labels: Vec<KeyValue>,
 }
 
 pub trait SpanSink {
@@ -78,14 +74,6 @@ impl<K: SpanSink> SpanSink for Option<K> {
 
 // === impl Id ===
 
-impl Id {
-    fn new_span_id<R: Rng>(rng: &mut R) -> Self {
-        let mut bytes = vec![0; SPAN_ID_LEN];
-        rng.fill(bytes.as_mut_slice());
-        Self(bytes)
-    }
-}
-
 impl From<Id> for Vec<u8> {
     fn from(Id(bytes): Id) -> Vec<u8> {
         bytes