chore(tracing): Use upstream OpenTelemetry for all tracing

Cargo.lock

@@ -3018,6 +3018,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 }
 }