Integrate OpenTelemetry into the proxy

OpenCensus is a deprecated protocol and is slated to be removed from upstream collectors soon.

This wires up the proxy to optionally use OpenTelmetry as the format for exported traces. Currently, this defaults to the existing OpenCensus exporter, and we can switch the default later.

[#10111](linkerd/linkerd2#10111)

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

Cargo.lock

@@ -1126,6 +1126,7 @@ dependencies = [
  "linkerd-app-outbound",
  "linkerd-error",
  "linkerd-opencensus",
+ "linkerd-opentelemetry",
  "linkerd-tonic-stream",
  "rangemap",
  "regex",
@@ -1184,6 +1185,7 @@ dependencies = [
  "linkerd-meshtls",
  "linkerd-metrics",
  "linkerd-opencensus",
+ "linkerd-opentelemetry",
  "linkerd-proxy-api-resolve",
  "linkerd-proxy-balance",
  "linkerd-proxy-client-policy",
@@ -1743,6 +1745,7 @@ dependencies = [
  "http-body",
  "linkerd-error",
  "linkerd-metrics",
+ "linkerd-trace-context",
  "opencensus-proto",
  "tokio",
  "tokio-stream",
@@ -1759,6 +1762,7 @@ dependencies = [
  "http-body",
  "linkerd-error",
  "linkerd-metrics",
+ "linkerd-trace-context",
  "opentelemetry",
  "opentelemetry-proto",
  "opentelemetry_sdk",

linkerd/app/Cargo.toml

@@ -25,6 +25,7 @@ linkerd-app-inbound = { path = "./inbound" }
 linkerd-app-outbound = { path = "./outbound" }
 linkerd-error = { path = "../error" }
 linkerd-opencensus = { path = "../opencensus" }
+linkerd-opentelemetry = { path = "../opentelemetry" }
 linkerd-tonic-stream = { path = "../tonic-stream" }
 rangemap = "1"
 regex = "1"

linkerd/app/core/Cargo.toml

@@ -47,6 +47,7 @@ linkerd-io = { path = "../../io" }
 linkerd-meshtls = { path = "../../meshtls", default-features = false }
 linkerd-metrics = { path = "../../metrics", features = ["process", "stack"] }
 linkerd-opencensus = { path = "../../opencensus" }
+linkerd-opentelemetry = { path = "../../opentelemetry" }
 linkerd-proxy-api-resolve = { path = "../../proxy/api-resolve" }
 linkerd-proxy-balance = { path = "../../proxy/balance" }
 linkerd-proxy-core = { path = "../../proxy/core" }

linkerd/app/core/src/http_tracing.rs

@@ -1,139 +1,76 @@
 use linkerd_error::Error;
-use linkerd_opencensus::proto::trace::v1 as oc;
 use linkerd_stack::layer;
-use linkerd_trace_context::{self as trace_context, TraceContext};
-use std::{collections::HashMap, sync::Arc};
-use thiserror::Error;
+use linkerd_trace_context::{
+    self as trace_context,
+    export::{ExportSpan, SpanKind, SpanLabels},
+    Span, TraceContext,
+};
+use std::{str::FromStr, sync::Arc};
 use tokio::sync::mpsc;
 
-pub type OpenCensusSink = Option<mpsc::Sender<oc::Span>>;
-pub type Labels = Arc<HashMap<String, String>>;
-
-/// SpanConverter converts trace_context::Span objects into OpenCensus agent
-/// protobuf span objects. SpanConverter receives trace_context::Span objects by
-/// implmenting the SpanSink trait. For each span that it receives, it converts
-/// it to an OpenCensus span and then sends it on the provided mpsc::Sender.
-#[derive(Clone)]
-pub struct SpanConverter {
-    kind: Kind,
-    sink: mpsc::Sender<oc::Span>,
-    labels: Labels,
+#[derive(Debug, Copy, Clone, Default)]
+pub enum CollectorProtocol {
+    #[default]
+    OpenCensus,
+    OpenTelemetry,
 }
 
-#[derive(Debug, Error)]
-#[error("ID '{:?} should have {} bytes, but it has {}", self.id, self.expected_size, self.actual_size)]
-pub struct IdLengthError {
-    id: Vec<u8>,
-    expected_size: usize,
-    actual_size: usize,
+impl FromStr for CollectorProtocol {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        if s.eq_ignore_ascii_case("opencensus") {
+            Ok(Self::OpenCensus)
+        } else if s.eq_ignore_ascii_case("opentelemetry") {
+            Ok(Self::OpenTelemetry)
+        } else {
+            Err(())
+        }
+    }
 }
 
+pub type SpanSink = mpsc::Sender<ExportSpan>;
+
 pub fn server<S>(
-    sink: OpenCensusSink,
-    labels: impl Into<Labels>,
+    sink: Option<SpanSink>,
+    labels: impl Into<SpanLabels>,
 ) -> impl layer::Layer<S, Service = TraceContext<Option<SpanConverter>, S>> + Clone {
-    SpanConverter::layer(Kind::Server, sink, labels)
+    TraceContext::layer(sink.map(move |sink| SpanConverter {
+        kind: SpanKind::Server,
+        sink,
+        labels: labels.into(),
+    }))
 }
 
 pub fn client<S>(
-    sink: OpenCensusSink,
-    labels: impl Into<Labels>,
+    sink: Option<SpanSink>,
+    labels: impl Into<SpanLabels>,
 ) -> impl layer::Layer<S, Service = TraceContext<Option<SpanConverter>, S>> + Clone {
-    SpanConverter::layer(Kind::Client, sink, labels)
-}
-
-#[derive(Copy, Clone, Debug, PartialEq)]
-enum Kind {
-    Server = 1,
-    Client = 2,
+    TraceContext::layer(sink.map(move |sink| SpanConverter {
+        kind: SpanKind::Client,
+        sink,
+        labels: labels.into(),
+    }))
 }
 
-impl SpanConverter {
-    fn layer<S>(
-        kind: Kind,
-        sink: OpenCensusSink,
-        labels: impl Into<Labels>,
-    ) -> impl layer::Layer<S, Service = TraceContext<Option<Self>, S>> + Clone {
-        TraceContext::layer(sink.map(move |sink| Self {
-            kind,
-            sink,
-            labels: labels.into(),
-        }))
-    }
-
-    fn mk_span(&self, mut span: trace_context::Span) -> Result<oc::Span, IdLengthError> {
-        let mut attributes = HashMap::<String, oc::AttributeValue>::new();
-        for (k, v) in self.labels.iter() {
-            attributes.insert(
-                k.clone(),
-                oc::AttributeValue {
-                    value: Some(oc::attribute_value::Value::StringValue(truncatable(
-                        v.clone(),
-                    ))),
-                },
-            );
-        }
-        for (k, v) in span.labels.drain() {
-            attributes.insert(
-                k.to_string(),
-                oc::AttributeValue {
-                    value: Some(oc::attribute_value::Value::StringValue(truncatable(v))),
-                },
-            );
-        }
-        Ok(oc::Span {
-            trace_id: into_bytes(span.trace_id, 16)?,
-            span_id: into_bytes(span.span_id, 8)?,
-            tracestate: None,
-            parent_span_id: into_bytes(span.parent_id, 8)?,
-            name: Some(truncatable(span.span_name)),
-            kind: self.kind as i32,
-            start_time: Some(span.start.into()),
-            end_time: Some(span.end.into()),
-            attributes: Some(oc::span::Attributes {
-                attribute_map: attributes,
-                dropped_attributes_count: 0,
-            }),
-            stack_trace: None,
-            time_events: None,
-            links: None,
-            status: None, // TODO: this is gRPC status; we must read response trailers to populate this
-            resource: None,
-            same_process_as_parent_span: Some(self.kind == Kind::Client),
-            child_span_count: None,
-        })
-    }
+#[derive(Clone)]
+pub struct SpanConverter {
+    kind: SpanKind,
+    sink: SpanSink,
+    labels: SpanLabels,
 }
 
 impl trace_context::SpanSink for SpanConverter {
-    #[inline]
     fn is_enabled(&self) -> bool {
         true
     }
 
-    fn try_send(&mut self, span: trace_context::Span) -> Result<(), Error> {
-        let span = self.mk_span(span)?;
-        self.sink.try_send(span).map_err(Into::into)
-    }
-}
-
-fn into_bytes(id: trace_context::Id, size: usize) -> Result<Vec<u8>, IdLengthError> {
-    let bytes: Vec<u8> = id.into();
-    if bytes.len() == size {
-        Ok(bytes)
-    } else {
-        let actual_size = bytes.len();
-        Err(IdLengthError {
-            id: bytes,
-            expected_size: size,
-            actual_size,
-        })
-    }
-}
-
-fn truncatable(value: String) -> oc::TruncatableString {
-    oc::TruncatableString {
-        value,
-        truncated_byte_count: 0,
+    fn try_send(&mut self, span: Span) -> Result<(), Error> {
+        self.sink.try_send(ExportSpan {
+            span,
+            kind: self.kind,
+            labels: Arc::clone(&self.labels),
+        })?;
+        Ok(())
     }
 }

linkerd/app/core/src/lib.rs

@@ -40,6 +40,7 @@ pub use linkerd_http_metrics as http_metrics;
 pub use linkerd_idle_cache as idle_cache;
 pub use linkerd_io as io;
 pub use linkerd_opencensus as opencensus;
+pub use linkerd_opentelemetry as opentelemetry;
 pub use linkerd_service_profiles as profiles;
 pub use linkerd_stack_metrics as stack_metrics;
 pub use linkerd_stack_tracing as stack_tracing;
@@ -65,7 +66,7 @@ pub struct ProxyRuntime {
     pub identity: identity::creds::Receiver,
     pub metrics: metrics::Proxy,
     pub tap: proxy::tap::Registry,
-    pub span_sink: http_tracing::OpenCensusSink,
+    pub span_sink: Option<http_tracing::SpanSink>,
     pub drain: drain::Watch,
 }
 

linkerd/app/core/src/metrics.rs

@@ -9,7 +9,7 @@
 pub use crate::transport::labels::{TargetAddr, TlsAccept};
 use crate::{
     classify::Class,
-    control, http_metrics, opencensus, profiles, stack_metrics,
+    control, http_metrics, opencensus, opentelemetry, profiles, stack_metrics,
     svc::Param,
     tls,
     transport::{self, labels::TlsConnect},
@@ -39,6 +39,7 @@ pub struct Metrics {
     pub proxy: Proxy,
     pub control: ControlHttp,
     pub opencensus: opencensus::metrics::Registry,
+    pub opentelemetry: opentelemetry::metrics::Registry,
 }
 
 #[derive(Clone, Debug)]
@@ -191,11 +192,13 @@ impl Metrics {
         };
 
         let (opencensus, opencensus_report) = opencensus::metrics::new();
+        let (opentelemetry, opentelemetry_report) = opentelemetry::metrics::new();
 
         let metrics = Metrics {
             proxy,
             control,
             opencensus,
+            opentelemetry,
         };
 
         let report = endpoint_report
@@ -205,6 +208,7 @@ impl Metrics {
             .and_report(control_report)
             .and_report(transport_report)
             .and_report(opencensus_report)
+            .and_report(opentelemetry_report)
             .and_report(stack);
 
         (metrics, report)

linkerd/app/inbound/src/lib.rs

@@ -18,11 +18,13 @@ mod server;
 #[cfg(any(test, feature = "test-util", fuzzing))]
 pub mod test_util;
 
+#[cfg(fuzzing)]
+pub use self::http::fuzz as http_fuzz;
 pub use self::{metrics::InboundMetrics, policy::DefaultPolicy};
 use linkerd_app_core::{
     config::{ConnectConfig, ProxyConfig, QueueConfig},
     drain,
-    http_tracing::OpenCensusSink,
+    http_tracing::SpanSink,
     identity, io,
     proxy::{tap, tcp},
     svc,
@@ -33,9 +35,6 @@ use std::{fmt::Debug, time::Duration};
 use thiserror::Error;
 use tracing::debug_span;
 
-#[cfg(fuzzing)]
-pub use self::http::fuzz as http_fuzz;
-
 #[derive(Clone, Debug)]
 pub struct Config {
     pub allow_discovery: NameMatch,
@@ -67,7 +66,7 @@ struct Runtime {
     metrics: InboundMetrics,
     identity: identity::creds::Receiver,
     tap: tap::Registry,
-    span_sink: OpenCensusSink,
+    span_sink: Option<SpanSink>,
     drain: drain::Watch,
 }
 

linkerd/app/outbound/src/http/server.rs

@@ -47,7 +47,7 @@ impl<T> Outbound<svc::ArcNewCloneHttp<T>> {
                 .check_new_service::<T, http::Request<_>>()
                 .push(ServerRescue::layer(config.emit_headers))
                 .check_new_service::<T, http::Request<_>>()
-                // Initiates OpenCensus tracing.
+                // Initiates OpenTelemetry tracing.
                 .push_on_service(http_tracing::server(rt.span_sink.clone(), trace_labels()))
                 .push_on_service(http::BoxResponse::layer())
                 // Convert origin form HTTP/1 URIs to absolute form for Hyper's

linkerd/app/outbound/src/lib.rs

@@ -6,11 +6,11 @@
 #![allow(opaque_hidden_inferred_bound)]
 #![forbid(unsafe_code)]
 
+use linkerd_app_core::http_tracing::SpanSink;
 use linkerd_app_core::{
     config::{ProxyConfig, QueueConfig},
     drain,
     exp_backoff::ExponentialBackoff,
-    http_tracing::OpenCensusSink,
     identity, io,
     metrics::prom,
     profiles,
@@ -95,7 +95,7 @@ struct Runtime {
     metrics: OutboundMetrics,
     identity: identity::NewClient,
     tap: tap::Registry,
-    span_sink: OpenCensusSink,
+    span_sink: Option<SpanSink>,
     drain: drain::Watch,
 }