chore(tracing): Use upstream tracing batch exporter

The proxy used its own custom trace exporter for some time. It was built before the upstream OpenTelemetry libraries were available, and they have outlived their usefulness.

This replaces the custom exporter with a batch exporter configured to export to the same endpoint with the same service configuration.

In the future, this exporter can be installed as a global default trace processor, which would decouple it from the service layer where the proxy generates spans for requests.

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

Author: sfleen

linkerd/app/src/trace_collector/otel_collector.rs

@@ -1,13 +1,6 @@
 use super::EnabledCollector;
 use linkerd_app_core::{control::ControlAddr, proxy::http::Body, Error};
-use linkerd_opentelemetry::{
-    self as opentelemetry, metrics,
-    proto::{
-        tonic::common::v1::{any_value, AnyValue, KeyValue},
-        transform::common::tonic::ResourceAttributesWithSchema,
-    },
-    semconv,
-};
+use linkerd_opentelemetry::{self as opentelemetry, metrics, otel::KeyValue, sdk, semconv};
 use std::{
     collections::HashMap,
     time::{SystemTime, UNIX_EPOCH},
@@ -29,7 +22,7 @@ pub(super) fn create_collector<S>(
     legacy_metrics: metrics::Registry,
 ) -> EnabledCollector
 where
-    S: GrpcService<TonicBody> + Clone + Send + 'static,
+    S: GrpcService<TonicBody> + Clone + Send + Sync + 'static,
     S::Error: Into<Error>,
     S::Future: Send,
     S::ResponseBody: Body<Data = tonic::codegen::Bytes> + Send + 'static,
@@ -38,37 +31,33 @@ where
     let (span_sink, spans_rx) = mpsc::channel(crate::trace_collector::SPAN_BUFFER_CAPACITY);
     let spans_rx = ReceiverStream::new(spans_rx);
 
-    let mut resources = ResourceAttributesWithSchema::default();
-
-    resources
-        .attributes
-        .0
-        .push((std::process::id() as i64).with_key(semconv::attribute::PROCESS_PID));
-
-    resources.attributes.0.push(
-        SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .map(|d| d.as_secs() as i64)
-            .unwrap_or_else(|e| -(e.duration().as_secs() as i64))
-            .with_key("process.start_timestamp"),
-    );
-    resources.attributes.0.push(
-        attributes
-            .hostname
-            .unwrap_or_default()
-            .with_key(semconv::attribute::HOST_NAME),
-    );
-
-    resources.attributes.0.extend(
-        attributes
-            .extra
-            .into_iter()
-            .map(|(key, value)| value.with_key(&key)),
-    );
+    let resource = sdk::Resource::builder()
+        .with_attribute(KeyValue::new(
+            semconv::attribute::PROCESS_PID,
+            std::process::id() as i64,
+        ))
+        .with_attribute(KeyValue::new(
+            "process.start_timestamp",
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .map(|d| d.as_secs() as i64)
+                .unwrap_or_else(|e| -(e.duration().as_secs() as i64)),
+        ))
+        .with_attribute(KeyValue::new(
+            semconv::attribute::HOST_NAME,
+            attributes.hostname.unwrap_or_default(),
+        ))
+        .with_attributes(
+            attributes
+                .extra
+                .into_iter()
+                .map(|(k, v)| KeyValue::new(k, v)),
+        )
+        .build();
 
     let addr = addr.clone();
     let task = Box::pin(
-        opentelemetry::export_spans(svc, spans_rx, resources, legacy_metrics)
+        opentelemetry::export_spans(svc, spans_rx, resource, legacy_metrics)
             .instrument(tracing::debug_span!("opentelemetry", peer.addr = %addr).or_current()),
     );
 
@@ -78,41 +67,3 @@ where
         span_sink,
     }
 }
-
-trait IntoAnyValue
-where
-    Self: Sized,
-{
-    fn into_any_value(self) -> AnyValue;
-
-    fn with_key(self, key: &str) -> KeyValue {
-        KeyValue {
-            key: key.to_string(),
-            value: Some(self.into_any_value()),
-        }
-    }
-}
-
-impl IntoAnyValue for String {
-    fn into_any_value(self) -> AnyValue {
-        AnyValue {
-            value: Some(any_value::Value::StringValue(self)),
-        }
-    }
-}
-
-impl IntoAnyValue for &str {
-    fn into_any_value(self) -> AnyValue {
-        AnyValue {
-            value: Some(any_value::Value::StringValue(self.to_string())),
-        }
-    }
-}
-
-impl IntoAnyValue for i64 {
-    fn into_any_value(self) -> AnyValue {
-        AnyValue {
-            value: Some(any_value::Value::IntValue(self)),
-        }
-    }
-}