chore(tracing): Add tracing export tests

This adds tests for the tracing exporter. This will help a future PR that modifies this to use the upstream tracing exporter.

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

Author: sfleen

Cargo.lock

@@ -600,7 +600,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.1",
 ]
 
 [[package]]
@@ -635,9 +635,9 @@ checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
 
 [[package]]
 name = "flate2"
-version = "1.1.4"
+version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc5a4e564e38c699f2880d3fda590bedc2e69f3f84cd48b457bd892ce61d0aa9"
+checksum = "bfe33edd8e85a12a67454e37f8c75e730830d83e313556ab9ebf9ee7fbeb3bfb"
 dependencies = [
  "crc32fast",
  "miniz_oxide",
@@ -1224,7 +1224,7 @@ dependencies = [
  "portable-atomic",
  "portable-atomic-util",
  "serde",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -2056,15 +2056,21 @@ name = "linkerd-opentelemetry"
 version = "0.1.0"
 dependencies = [
  "futures",
+ "hex",
+ "http",
  "http-body",
+ "http-body-util",
  "linkerd-error",
  "linkerd-metrics",
  "linkerd-trace-context",
+ "linkerd-tracing",
  "opentelemetry",
  "opentelemetry-proto",
  "opentelemetry_sdk",
  "tokio",
  "tonic",
+ "tonic-prost",
+ "tower-test",
  "tracing",
 ]
 
@@ -2834,7 +2840,7 @@ version = "0.50.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.59.0",
+ "windows-sys 0.61.1",
 ]
 
 [[package]]
@@ -3476,7 +3482,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys 0.4.15",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -3489,7 +3495,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys 0.9.2",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -3790,7 +3796,7 @@ dependencies = [
  "getrandom 0.3.1",
  "once_cell",
  "rustix 1.0.2",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.1",
 ]
 
 [[package]]

linkerd/opentelemetry/Cargo.toml

@@ -20,3 +20,12 @@ tonic = { workspace = true, default-features = false, features = [
 ] }
 tokio = { version = "1", features = ["macros", "sync", "time"] }
 tracing = { workspace = true }
+hex = "0.4.3"
+http-body-util = "0.1.3"
+
+[dev-dependencies]
+http = "1"
+linkerd-tracing = { path = "../tracing" }
+tonic-prost = { workspace = true, default-features = false }
+tokio = { version = "1", features = ["test-util"] }
+tower-test = { workspace = true }

linkerd/opentelemetry/src/lib.rs

@@ -276,3 +276,135 @@ fn convert_span(span: ExportSpan) -> Result<SpanData, Error> {
         instrumentation_scope: Default::default(),
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use linkerd_trace_context::{export::SpanKind, Id, Span};
+    use opentelemetry_proto::tonic::{common::v1::InstrumentationScope, resource::v1::Resource};
+    use std::{collections::HashMap, sync::Arc, time::SystemTime};
+    use tokio::sync::mpsc;
+    use tonic::codegen::{tokio_stream::wrappers::ReceiverStream, tokio_stream::StreamExt, Bytes};
+    use tonic_prost::ProstDecoder;
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn send_span() {
+        let trace_id = Id::from(Bytes::from(
+            hex::decode("0123456789abcedffedcba9876543210").expect("decode"),
+        ));
+        let span_id = Id::from(Bytes::from(
+            hex::decode("fedcba9876543210").expect("decode"),
+        ));
+        let parent_id = Id::from(Bytes::from(
+            hex::decode("0123456789abcedf").expect("decode"),
+        ));
+        let span_name = "test".to_string();
+
+        let start = SystemTime::now();
+        let end = SystemTime::now();
+
+        let span = ExportSpan {
+            span: Span {
+                trace_id: trace_id.clone(),
+                span_id: span_id.clone(),
+                parent_id: parent_id.clone(),
+                span_name: span_name.clone(),
+                start,
+                end,
+                labels: HashMap::new(),
+            },
+            kind: SpanKind::Server,
+            labels: Arc::new(Default::default()),
+        };
+
+        let mut req = send_mock_request(span).await;
+
+        assert_eq!(req.resource_spans.len(), 1);
+        let mut resource_span = req.resource_spans.remove(0);
+        assert_eq!(
+            resource_span.resource,
+            Some(Resource {
+                attributes: vec![],
+                dropped_attributes_count: 0,
+                entity_refs: vec![],
+            })
+        );
+        assert_eq!(resource_span.schema_url, "");
+        assert_eq!(resource_span.scope_spans.len(), 1);
+
+        let mut scope_span = resource_span.scope_spans.remove(0);
+        assert_eq!(scope_span.schema_url, "");
+        assert_eq!(
+            scope_span.scope,
+            Some(InstrumentationScope {
+                name: "".to_string(),
+                version: "".to_string(),
+                attributes: vec![],
+                dropped_attributes_count: 0,
+            })
+        );
+        assert_eq!(scope_span.spans.len(), 1);
+
+        let span = scope_span.spans.remove(0);
+        assert_eq!(
+            span.span_id,
+            span_id.into_bytes::<8>().expect("into_bytes").to_vec()
+        );
+        assert_eq!(
+            span.parent_span_id,
+            parent_id.into_bytes::<8>().expect("into_bytes").to_vec()
+        );
+        assert_eq!(
+            span.trace_id,
+            trace_id.into_bytes::<16>().expect("into_bytes").to_vec()
+        );
+        assert_eq!(span.name, span_name);
+        assert_eq!(
+            span.start_time_unix_nano,
+            start
+                .duration_since(SystemTime::UNIX_EPOCH)
+                .expect("duration")
+                .as_nanos() as u64
+        );
+        assert_eq!(
+            span.end_time_unix_nano,
+            end.duration_since(SystemTime::UNIX_EPOCH)
+                .expect("duration")
+                .as_nanos() as u64
+        );
+        assert_eq!(span.flags, 0b11_0000_0000);
+    }
+
+    async fn send_mock_request(span: ExportSpan) -> ExportTraceServiceRequest {
+        let _trace = linkerd_tracing::test::trace_init();
+        let (span_tx, span_rx) = mpsc::channel(1);
+
+        let (inner, mut handle) = tower_test::mock::pair::<
+            http::Request<tonic::body::Body>,
+            http::Response<tonic::body::Body>,
+        >();
+        handle.allow(1);
+
+        span_tx.try_send(span).expect("Must have space");
+
+        let (metrics, _) = metrics::new();
+        tokio::spawn(export_spans(
+            inner,
+            ReceiverStream::new(span_rx),
+            ResourceAttributesWithSchema::default(),
+            metrics,
+        ));
+
+        let (req, _tx) = handle.next_request().await.expect("next request");
+        let req = tonic::Request::from_http(req);
+        let mut req = tonic::codec::Streaming::<ExportTraceServiceRequest>::new_request(
+            ProstDecoder::default(),
+            req.into_inner(),
+            None,
+            None,
+        );
+        let req = req.next().await.expect("next").expect("must succeed");
+
+        req
+    }
+}

linkerd/trace-context/src/lib.rs

@@ -16,7 +16,7 @@ use thiserror::Error;
 
 const SPAN_ID_LEN: usize = 8;
 
-#[derive(Debug, Default, Eq, PartialEq)]
+#[derive(Debug, Default, Clone, Eq, PartialEq)]
 pub struct Id(Vec<u8>);
 
 #[derive(Debug, Error)]