feat(spans): Accept V2 spans and convert them to V1 spans

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Add logic to extract event json from userdata in prosperodumps. ([#4755](https://github.com/getsentry/relay/pull/4755)
 - Add browser name/version to logs. ([#4757](https://github.com/getsentry/relay/pull/4757))
+- Accept standalone spans in the V2 format. This feature is still highly experimental! ([#4771](https://github.com/getsentry/relay/pull/4771))
 
 **Internal**:
 

relay-server/src/envelope/container.rs

@@ -1,3 +1,11 @@
+#![cfg_attr(
+    not(any(test, feature = "processing")),
+    expect(
+        dead_code,
+        reason = "A number of functions and data in this module are only used in processing or tests."
+    )
+)]
+
 use std::marker::PhantomData;
 
 use bytes::BufMut;
@@ -163,6 +171,10 @@ impl ContainerItem for relay_event_schema::protocol::OurLog {
     const CONTENT_TYPE: ContentType = ContentType::LogContainer;
 }
 
+impl ContainerItem for relay_event_schema::protocol::SpanV2 {
+    const CONTENT_TYPE: ContentType = ContentType::SpanV2Container;
+}
+
 /// (De-)Serializes a list of Annotated items with metadata.
 #[derive(Debug)]
 struct AnnotatedItems<T>(ContainerItems<T>);

relay-server/src/envelope/content_type.rs

@@ -9,24 +9,26 @@ pub const CONTENT_TYPE: &str = "application/x-sentry-envelope";
 /// This is an optimized enum intended to reduce allocations for common content types.
 #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub enum ContentType {
-    /// text/plain
+    /// `text/plain`
     Text,
-    /// application/json
+    /// `application/json`
     Json,
-    /// application/x-msgpack
+    /// `application/x-msgpack`
     MsgPack,
-    /// application/octet-stream
+    /// `application/octet-stream`
     OctetStream,
-    /// application/x-dmp
+    /// `application/x-dmp`
     Minidump,
-    /// text/xml and application/xml
+    /// `text/xml` and `application/xml`
     Xml,
-    /// application/x-sentry-envelope
+    /// `application/x-sentry-envelope`
     Envelope,
-    /// "application/x-protobuf"
+    /// `application/x-protobuf`
     Protobuf,
     /// `application/vnd.sentry.items.log+json`
     LogContainer,
+    /// `application/vnd.sentry.items.span.v2+json`
+    SpanV2Container,
     /// Any arbitrary content type not listed explicitly.
     Other(String),
 }
@@ -44,6 +46,7 @@ impl ContentType {
             Self::Envelope => CONTENT_TYPE,
             Self::Protobuf => "application/x-protobuf",
             Self::LogContainer => "application/vnd.sentry.items.log+json",
+            Self::SpanV2Container => "application/vnd.sentry.items.span.v2+json",
             Self::Other(ref other) => other,
         }
     }
@@ -69,6 +72,8 @@ impl ContentType {
             Some(Self::Protobuf)
         } else if ct.eq_ignore_ascii_case(Self::LogContainer.as_str()) {
             Some(Self::LogContainer)
+        } else if ct.eq_ignore_ascii_case(Self::SpanV2Container.as_str()) {
+            Some(Self::SpanV2Container)
         } else {
             None
         }

relay-server/src/envelope/mod.rs

@@ -51,13 +51,11 @@ use crate::constants::DEFAULT_EVENT_RETENTION;
 use crate::extractors::{PartialMeta, RequestMeta};
 
 mod attachment;
-#[cfg(feature = "processing")]
 mod container;
 mod content_type;
 mod item;
 
 pub use self::attachment::*;
-#[cfg(feature = "processing")]
 pub use self::container::*;
 pub use self::content_type::*;
 pub use self::item::*;

relay-server/src/services/processor.rs

@@ -2166,6 +2166,7 @@ impl EnvelopeProcessorService {
     ) -> Result<Option<ProcessingExtractedMetrics>, ProcessingError> {
         let mut extracted_metrics = ProcessingExtractedMetrics::new();
 
+        span::expand_v2_spans(managed_envelope)?;
         span::filter(managed_envelope, config.clone(), project_info.clone());
         span::convert_otel_traces_data(managed_envelope);
 

relay-server/src/services/processor/span.rs

@@ -12,7 +12,7 @@ use relay_protocol::Annotated;
 use relay_quotas::DataCategory;
 use relay_spans::otel_trace::TracesData;
 
-use crate::envelope::{ContentType, Item, ItemType};
+use crate::envelope::{ContentType, Item, ItemContainer, ItemType};
 use crate::services::outcome::{DiscardReason, Outcome};
 use crate::services::processor::{SpanGroup, should_filter};
 use crate::utils::ItemAction;
@@ -25,6 +25,8 @@ use crate::services::projects::project::ProjectInfo;
 pub use processing::*;
 use relay_config::Config;
 
+use super::ProcessingError;
+
 pub fn filter(
     managed_envelope: &mut TypedEnvelope<SpanGroup>,
     config: Arc<Config>,
@@ -46,6 +48,65 @@ pub fn filter(
     });
 }
 
+/// Expands V2 spans to V1 spans.
+///
+/// This expands one item (contanining multiple V2 spans) into several
+/// (containing one V1 span each).
+pub fn expand_v2_spans(
+    managed_envelope: &mut TypedEnvelope<SpanGroup>,
+) -> Result<(), ProcessingError> {
+    let span_v2_items = managed_envelope
+        .envelope_mut()
+        .take_items_by(is_span_v2_item);
+
+    // V2 spans must always be sent as an `ItemContainer`, currently it is not allowed to
+    // send multiple containers for V2 spans.
+    //
+    // This restriction may be lifted in the future, this is why this validation only happens
+    // when processing is enabled, allowing it to be changed easily in the future.
+    //
+    // This limit mostly exists to incentivise SDKs to batch multiple spans into a single container,
+    // technically it can be removed without issues.
+    if span_v2_items.len() > 1 {
+        return Err(ProcessingError::DuplicateItem(ItemType::Span));
+    }
+    for span_v2_item in span_v2_items {
+        let spans_v2 = match ItemContainer::parse(&span_v2_item) {
+            Ok(spans_v2) => spans_v2,
+            Err(err) => {
+                relay_log::debug!("failed to parse V2 spans: {err}");
+                track_invalid(
+                    managed_envelope,
+                    DiscardReason::InvalidSpan,
+                    span_v2_item.item_count().unwrap_or(1) as usize,
+                );
+                continue;
+            }
+        };
+
+        for span_v2 in spans_v2.into_items() {
+            let span_v1 = span_v2.map_value(relay_spans::span_v2_to_span_v1);
+            match span_v1.to_json() {
+                Ok(payload) => {
+                    let mut new_item = Item::new(ItemType::Span);
+                    new_item.set_payload(ContentType::Json, payload);
+                    managed_envelope.envelope_mut().add_item(new_item);
+                }
+                Err(err) => {
+                    relay_log::debug!("failed to serialize span: {}", err);
+                    track_invalid(managed_envelope, DiscardReason::Internal, 1);
+                }
+            }
+        }
+    }
+
+    Ok(())
+}
+
+fn is_span_v2_item(item: &Item) -> bool {
+    item.ty() == &ItemType::Span && item.content_type() == Some(&ContentType::SpanV2Container)
+}
+
 pub fn convert_otel_traces_data(managed_envelope: &mut TypedEnvelope<SpanGroup>) {
     let envelope = managed_envelope.envelope_mut();
 
@@ -60,7 +121,7 @@ fn convert_traces_data(item: Item, managed_envelope: &mut TypedEnvelope<SpanGrou
         Err(reason) => {
             // NOTE: logging quantity=1 is semantically wrong, but we cannot know the real quantity
             // without parsing.
-            track_invalid(managed_envelope, reason);
+            track_invalid(managed_envelope, reason, 1);
             return;
         }
     };
@@ -102,7 +163,7 @@ fn convert_traces_data(item: Item, managed_envelope: &mut TypedEnvelope<SpanGrou
                 }
 
                 let Ok(payload) = serde_json::to_vec(&span) else {
-                    track_invalid(managed_envelope, DiscardReason::Internal);
+                    track_invalid(managed_envelope, DiscardReason::Internal, 1);
                     continue;
                 };
                 let mut item = Item::new(ItemType::OtelSpan);
@@ -114,9 +175,17 @@ fn convert_traces_data(item: Item, managed_envelope: &mut TypedEnvelope<SpanGrou
     managed_envelope.update(); // update envelope summary
 }
 
-fn track_invalid(managed_envelope: &mut TypedEnvelope<SpanGroup>, reason: DiscardReason) {
-    managed_envelope.track_outcome(Outcome::Invalid(reason), DataCategory::Span, 1);
-    managed_envelope.track_outcome(Outcome::Invalid(reason), DataCategory::SpanIndexed, 1);
+fn track_invalid(
+    managed_envelope: &mut TypedEnvelope<SpanGroup>,
+    reason: DiscardReason,
+    quantity: usize,
+) {
+    managed_envelope.track_outcome(Outcome::Invalid(reason), DataCategory::Span, quantity);
+    managed_envelope.track_outcome(
+        Outcome::Invalid(reason),
+        DataCategory::SpanIndexed,
+        quantity,
+    );
 }
 
 fn parse_traces_data(item: Item) -> Result<TracesData, DiscardReason> {

relay-spans/src/lib.rs

@@ -7,6 +7,7 @@
 )]
 
 pub use crate::otel_to_sentry::otel_to_sentry_span;
+pub use crate::v2_to_v1::span_v2_to_span_v1;
 
 pub use opentelemetry_proto::tonic::trace::v1 as otel_trace;