Span links stored as Vector instead of EvictedQueue (#1313)

Author: cijothomas

opentelemetry-datadog/src/exporter/model/mod.rs

@@ -193,7 +193,11 @@ pub(crate) mod tests {
         trace::{SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState},
         KeyValue,
     };
-    use opentelemetry_sdk::{self, trace::EvictedQueue, InstrumentationLibrary, Resource};
+    use opentelemetry_sdk::{
+        self,
+        trace::{EvictedQueue, SpanLinks},
+        InstrumentationLibrary, Resource,
+    };
     use std::borrow::Cow;
     use std::time::{Duration, SystemTime};
 
@@ -216,7 +220,7 @@ pub(crate) mod tests {
         let capacity = 3;
         let attributes = vec![KeyValue::new("span.type", "web")];
         let events = EvictedQueue::new(capacity);
-        let links = EvictedQueue::new(capacity);
+        let links = SpanLinks::default();
         let resource = Resource::new(vec![KeyValue::new("host.name", "test")]);
 
         trace::SpanData {

opentelemetry-jaeger/src/exporter/mod.rs

@@ -102,7 +102,7 @@ impl SpanExporter for Exporter {
     }
 }
 
-fn links_to_references(links: EvictedQueue<Link>) -> Option<Vec<jaeger::SpanRef>> {
+fn links_to_references(links: &[Link]) -> Option<Vec<jaeger::SpanRef>> {
     if !links.is_empty() {
         let refs = links
             .iter()
@@ -139,7 +139,7 @@ fn convert_otel_span_into_jaeger_span(span: SpanData, export_instrument_lib: boo
         span_id: i64::from_be_bytes(span.span_context.span_id().to_bytes()),
         parent_span_id: i64::from_be_bytes(span.parent_span_id.to_bytes()),
         operation_name: span.name.into_owned(),
-        references: links_to_references(span.links),
+        references: links_to_references(span.links.as_ref()),
         flags: span.span_context.trace_flags().to_u8() as i32,
         start_time: span
             .start_time

opentelemetry-proto/src/transform/trace.rs

@@ -93,7 +93,7 @@ pub mod tonic {
                                 dropped_attributes_count: event.dropped_attributes_count,
                             })
                             .collect(),
-                        dropped_links_count: source_span.links.dropped_count(),
+                        dropped_links_count: source_span.links.dropped_count,
                         links: source_span.links.into_iter().map(Into::into).collect(),
                         status: Some(Status {
                             code: status::StatusCode::from(&source_span.status).into(),
@@ -204,7 +204,7 @@ pub mod grpcio {
                                 dropped_attributes_count: event.dropped_attributes_count,
                             })
                             .collect(),
-                        dropped_links_count: source_span.links.dropped_count(),
+                        dropped_links_count: source_span.links.dropped_count,
                         links: source_span.links.into_iter().map(Into::into).collect(),
                         status: Some(Status {
                             code: status::StatusCode::from(&source_span.status).into(),

opentelemetry-sdk/CHANGELOG.md

@@ -2,6 +2,22 @@
 
 ## vNext
 
+### Changed
+
+- **Breaking**
+[#1313](https://github.com/open-telemetry/opentelemetry-rust/issues/1313)
+  Changes how Span links are stored to achieve performance gains. See below for
+  details:
+
+  *Behavior Change*: When enforcing `max_links_per_span` from `SpanLimits`,
+  links are kept in the first-come order. The previous "eviction" based approach
+  is no longer performed.
+  
+  *Breaking Change Affecting Exporter authors*:
+  
+  `SpanData` now stores `links` as `SpanLinks` instead of `EvictedQueue` where
+  `SpanLinks` is a struct with a `Vec` of links and `dropped_count`.
+
 ## v0.21.0
 
 ### Added

opentelemetry-sdk/benches/batch_span_processor.rs

@@ -5,7 +5,7 @@ use opentelemetry::trace::{
 use opentelemetry_sdk::export::trace::SpanData;
 use opentelemetry_sdk::runtime::Tokio;
 use opentelemetry_sdk::testing::trace::NoopSpanExporter;
-use opentelemetry_sdk::trace::{BatchSpanProcessor, EvictedQueue, SpanProcessor};
+use opentelemetry_sdk::trace::{BatchSpanProcessor, EvictedQueue, SpanLinks, SpanProcessor};
 use opentelemetry_sdk::Resource;
 use std::borrow::Cow;
 use std::sync::Arc;
@@ -30,7 +30,7 @@ fn get_span_data() -> Vec<SpanData> {
             attributes: Vec::new(),
             dropped_attributes_count: 0,
             events: EvictedQueue::new(12),
-            links: EvictedQueue::new(12),
+            links: SpanLinks::default(),
             status: Status::Unset,
             resource: Cow::Owned(Resource::empty()),
             instrumentation_lib: Default::default(),

opentelemetry-sdk/src/export/trace.rs

@@ -1,7 +1,7 @@
 //! Trace exporters
 use crate::Resource;
 use futures_util::future::BoxFuture;
-use opentelemetry::trace::{Event, Link, SpanContext, SpanId, SpanKind, Status, TraceError};
+use opentelemetry::trace::{Event, SpanContext, SpanId, SpanKind, Status, TraceError};
 use opentelemetry::KeyValue;
 use std::borrow::Cow;
 use std::fmt::Debug;
@@ -89,7 +89,7 @@ pub struct SpanData {
     /// Span events
     pub events: crate::trace::EvictedQueue<Event>,
     /// Span Links
-    pub links: crate::trace::EvictedQueue<Link>,
+    pub links: crate::trace::SpanLinks,
     /// Span status
     pub status: Status,
     /// Resource contains attributes representing an entity that produced this span.

opentelemetry-sdk/src/testing/trace/mod.rs

@@ -7,7 +7,7 @@ use crate::{
         trace::{ExportResult, SpanData, SpanExporter},
         ExportError,
     },
-    trace::{Config, EvictedQueue},
+    trace::{Config, EvictedQueue, SpanLinks},
     InstrumentationLibrary,
 };
 use async_trait::async_trait;
@@ -37,7 +37,7 @@ pub fn new_test_export_span_data() -> SpanData {
         attributes: Vec::new(),
         dropped_attributes_count: 0,
         events: EvictedQueue::new(config.span_limits.max_events_per_span),
-        links: EvictedQueue::new(config.span_limits.max_links_per_span),
+        links: SpanLinks::default(),
         status: Status::Unset,
         resource: config.resource,
         instrumentation_lib: InstrumentationLibrary::default(),

opentelemetry-sdk/src/trace/links.rs

@@ -0,0 +1,31 @@
+//! # Span Links
+
+use std::ops::Deref;
+
+use opentelemetry::trace::Link;
+/// Stores span links along with dropped count.
+#[derive(Clone, Debug, Default, PartialEq)]
+#[non_exhaustive]
+pub struct SpanLinks {
+    /// The links stored as a vector. Could be empty if there are no links.
+    pub links: Vec<Link>,
+    /// The number of links dropped from the span.
+    pub dropped_count: u32,
+}
+
+impl Deref for SpanLinks {
+    type Target = [Link];
+
+    fn deref(&self) -> &Self::Target {
+        &self.links
+    }
+}
+
+impl IntoIterator for SpanLinks {
+    type Item = Link;
+    type IntoIter = std::vec::IntoIter<Self::Item>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.links.into_iter()
+    }
+}

opentelemetry-sdk/src/trace/mod.rs

@@ -10,6 +10,7 @@ mod config;
 mod evicted_hash_map;
 mod evicted_queue;
 mod id_generator;
+mod links;
 mod provider;
 mod sampler;
 mod span;
@@ -21,6 +22,7 @@ pub use config::{config, Config};
 pub use evicted_hash_map::EvictedHashMap;
 pub use evicted_queue::EvictedQueue;
 pub use id_generator::{aws::XrayIdGenerator, IdGenerator, RandomIdGenerator};
+pub use links::SpanLinks;
 pub use provider::{Builder, TracerProvider};
 pub use sampler::{Sampler, ShouldSample};
 pub use span::Span;
@@ -39,14 +41,19 @@ mod runtime_tests;
 #[cfg(all(test, feature = "testing"))]
 mod tests {
     use super::*;
-    use crate::testing::trace::InMemorySpanExporterBuilder;
+    use crate::{
+        testing::trace::InMemorySpanExporterBuilder, trace::span_limit::DEFAULT_MAX_LINKS_PER_SPAN,
+    };
     use opentelemetry::{
-        trace::{Span, Tracer, TracerProvider as _},
+        trace::{
+            Link, Span, SpanBuilder, SpanContext, SpanId, TraceFlags, TraceId, Tracer,
+            TracerProvider as _,
+        },
         KeyValue,
     };
 
     #[test]
-    fn tracing_in_span() {
+    fn in_span() {
         // Arrange
         let exporter = InMemorySpanExporterBuilder::new().build();
         let provider = TracerProvider::builder()
@@ -70,7 +77,7 @@ mod tests {
     }
 
     #[test]
-    fn tracing_tracer_start() {
+    fn tracer_start() {
         // Arrange
         let exporter = InMemorySpanExporterBuilder::new().build();
         let provider = TracerProvider::builder()
@@ -93,4 +100,44 @@ mod tests {
         assert_eq!(span.name, "span_name");
         assert_eq!(span.instrumentation_lib.name, "test_tracer");
     }
+
+    #[test]
+    fn exceed_span_links_limit() {
+        // Arrange
+        let exporter = InMemorySpanExporterBuilder::new().build();
+        let provider = TracerProvider::builder()
+            .with_span_processor(SimpleSpanProcessor::new(Box::new(exporter.clone())))
+            .build();
+
+        // Act
+        let tracer = provider.tracer("test_tracer");
+
+        let mut links = Vec::new();
+        for _i in 0..(DEFAULT_MAX_LINKS_PER_SPAN * 2) {
+            links.push(Link::new(
+                SpanContext::new(
+                    TraceId::from_u128(12),
+                    SpanId::from_u64(12),
+                    TraceFlags::default(),
+                    false,
+                    Default::default(),
+                ),
+                Vec::new(),
+            ))
+        }
+
+        let span_builder = SpanBuilder::from_name("span_name").with_links(links);
+        let mut span = tracer.build(span_builder);
+        span.end();
+        provider.force_flush();
+
+        // Assert
+        let exported_spans = exporter
+            .get_finished_spans()
+            .expect("Spans are expected to be exported.");
+        assert_eq!(exported_spans.len(), 1);
+        let span = &exported_spans[0];
+        assert_eq!(span.name, "span_name");
+        assert_eq!(span.links.len(), DEFAULT_MAX_LINKS_PER_SPAN as usize);
+    }
 }

opentelemetry-sdk/src/trace/span.rs

@@ -44,7 +44,7 @@ pub(crate) struct SpanData {
     /// Span events
     pub(crate) events: crate::trace::EvictedQueue<trace::Event>,
     /// Span Links
-    pub(crate) links: crate::trace::EvictedQueue<trace::Link>,
+    pub(crate) links: crate::trace::SpanLinks,
     /// Span status
     pub(crate) status: Status,
 }
@@ -252,8 +252,9 @@ mod tests {
     use crate::testing::trace::NoopSpanExporter;
     use crate::trace::span_limit::{
         DEFAULT_MAX_ATTRIBUTES_PER_EVENT, DEFAULT_MAX_ATTRIBUTES_PER_LINK,
-        DEFAULT_MAX_ATTRIBUTES_PER_SPAN,
+        DEFAULT_MAX_ATTRIBUTES_PER_SPAN, DEFAULT_MAX_LINKS_PER_SPAN,
     };
+    use crate::trace::SpanLinks;
     use opentelemetry::trace::{Link, SpanBuilder, TraceFlags, TraceId, Tracer};
     use opentelemetry::{trace::Span as _, trace::TracerProvider, KeyValue};
     use std::time::Duration;
@@ -272,7 +273,7 @@ mod tests {
             attributes: Vec::new(),
             dropped_attributes_count: 0,
             events: crate::trace::EvictedQueue::new(config.span_limits.max_events_per_span),
-            links: crate::trace::EvictedQueue::new(config.span_limits.max_links_per_span),
+            links: SpanLinks::default(),
             status: Status::Unset,
         };
         (tracer, data)
@@ -610,11 +611,44 @@ mod tests {
             .clone()
             .expect("span data should not be empty as we already set it before")
             .links;
-        let link_vec: Vec<_> = link_queue.iter().collect();
+        let link_vec: Vec<_> = link_queue.links;
         let processed_link = link_vec.get(0).expect("should have at least one link");
         assert_eq!(processed_link.attributes.len(), 128);
     }
 
+    #[test]
+    fn exceed_span_links_limit() {
+        let exporter = NoopSpanExporter::new();
+        let provider_builder =
+            crate::trace::TracerProvider::builder().with_simple_exporter(exporter);
+        let provider = provider_builder.build();
+        let tracer = provider.tracer("opentelemetry-test");
+
+        let mut links = Vec::new();
+        for _i in 0..(DEFAULT_MAX_LINKS_PER_SPAN * 2) {
+            links.push(Link::new(
+                SpanContext::new(
+                    TraceId::from_u128(12),
+                    SpanId::from_u64(12),
+                    TraceFlags::default(),
+                    false,
+                    Default::default(),
+                ),
+                Vec::new(),
+            ))
+        }
+
+        let span_builder = tracer.span_builder("test").with_links(links);
+        let span = tracer.build(span_builder);
+        let link_queue = span
+            .data
+            .clone()
+            .expect("span data should not be empty as we already set it before")
+            .links;
+        let link_vec: Vec<_> = link_queue.links;
+        assert_eq!(link_vec.len(), DEFAULT_MAX_LINKS_PER_SPAN as usize);
+    }
+
     #[test]
     fn test_span_exported_data() {
         let provider = crate::trace::TracerProvider::builder()

opentelemetry-sdk/src/trace/span_processor.rs

@@ -721,7 +721,7 @@ mod tests {
     use crate::testing::trace::{
         new_test_export_span_data, new_test_exporter, new_tokio_test_exporter,
     };
-    use crate::trace::{BatchConfig, EvictedQueue};
+    use crate::trace::{BatchConfig, EvictedQueue, SpanLinks};
     use async_trait::async_trait;
     use opentelemetry::trace::{SpanContext, SpanId, SpanKind, Status};
     use std::fmt::Debug;
@@ -751,7 +751,7 @@ mod tests {
             attributes: Vec::new(),
             dropped_attributes_count: 0,
             events: EvictedQueue::new(0),
-            links: EvictedQueue::new(0),
+            links: SpanLinks::default(),
             status: Status::Unset,
             resource: Default::default(),
             instrumentation_lib: Default::default(),