chore(tracing): Use upstream OpenTelemetry for all tracing

This removes all remaining bespoke tracing infrastructure. The main change here is registering a global tracer provider that the trace propagation service can use, replacing the manually wired span data channel.

This leaves in some remnants of the old infrastructure, specifically around the span sink types. These types are fairly viral, so a future PR will clean these up.

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

Author: sfleen

Cargo.lock

@@ -3024,6 +3024,8 @@ dependencies = [
  "percent-encoding",
  "rand 0.9.2",
  "thiserror",
+ "tokio",
+ "tokio-stream",
 ]
 
 [[package]]

linkerd/app/core/src/http_tracing.rs

@@ -11,25 +11,17 @@ use tokio::sync::mpsc;
 pub type SpanSink = mpsc::Sender<ExportSpan>;
 
 pub fn server<S>(
-    sink: Option<SpanSink>,
+    _sink: Option<SpanSink>,
     labels: impl Into<SpanLabels>,
-) -> impl layer::Layer<S, Service = TraceContext<Option<SpanConverter>, S>> + Clone {
-    TraceContext::layer(sink.map(move |sink| SpanConverter {
-        kind: SpanKind::Server,
-        sink,
-        labels: labels.into(),
-    }))
+) -> impl layer::Layer<S, Service = TraceContext<S>> + Clone {
+    TraceContext::layer(SpanKind::Server, labels.into())
 }
 
 pub fn client<S>(
-    sink: Option<SpanSink>,
+    _sink: Option<SpanSink>,
     labels: impl Into<SpanLabels>,
-) -> impl layer::Layer<S, Service = TraceContext<Option<SpanConverter>, S>> + Clone {
-    TraceContext::layer(sink.map(move |sink| SpanConverter {
-        kind: SpanKind::Client,
-        sink,
-        labels: labels.into(),
-    }))
+) -> impl layer::Layer<S, Service = TraceContext<S>> + Clone {
+    TraceContext::layer(SpanKind::Client, labels.into())
 }
 
 #[derive(Clone)]

linkerd/app/src/lib.rs

@@ -399,7 +399,6 @@ impl App {
             admin,
             drain,
             identity,
-            trace_collector: collector,
             start_proxy,
             tap,
             ..
@@ -464,10 +463,6 @@ impl App {
                             tokio::spawn(serve.instrument(info_span!("tap").or_current()));
                         }
 
-                        if let trace_collector::TraceCollector::Enabled(collector) = collector {
-                            tokio::spawn(collector.task.instrument(info_span!("tracing")));
-                        }
-
                         // we don't care if the admin shutdown channel is
                         // dropped or actually triggered.
                         let _ = admin_shutdown_rx.await;

linkerd/app/src/trace_collector.rs

@@ -33,7 +33,6 @@ pub enum TraceCollector {
 pub struct EnabledCollector {
     pub addr: control::ControlAddr,
     pub span_sink: SpanSink,
-    pub task: Task,
 }
 
 impl TraceCollector {

linkerd/app/src/trace_collector/otel_collector.rs

@@ -6,9 +6,7 @@ use std::{
     time::{SystemTime, UNIX_EPOCH},
 };
 use tokio::sync::mpsc;
-use tokio_stream::wrappers::ReceiverStream;
 use tonic::{body::Body as TonicBody, client::GrpcService};
-use tracing::Instrument;
 
 pub(super) struct OtelCollectorAttributes {
     pub hostname: Option<String>,
@@ -28,8 +26,7 @@ where
     S::ResponseBody: Body<Data = tonic::codegen::Bytes> + Send + 'static,
     <S::ResponseBody as Body>::Error: Into<Error> + Send,
 {
-    let (span_sink, spans_rx) = mpsc::channel(crate::trace_collector::SPAN_BUFFER_CAPACITY);
-    let spans_rx = ReceiverStream::new(spans_rx);
+    let (span_sink, _) = mpsc::channel(crate::trace_collector::SPAN_BUFFER_CAPACITY);
 
     let resource = sdk::Resource::builder()
         .with_attribute(KeyValue::new(
@@ -56,14 +53,7 @@ where
         .build();
 
     let addr = addr.clone();
-    let task = Box::pin(
-        opentelemetry::export_spans(svc, spans_rx, resource, legacy_metrics)
-            .instrument(tracing::debug_span!("opentelemetry", peer.addr = %addr).or_current()),
-    );
+    opentelemetry::install_opentelemetry_providers(svc, resource, legacy_metrics);
 
-    EnabledCollector {
-        addr,
-        task,
-        span_sink,
-    }
+    EnabledCollector { addr, span_sink }
 }