feat(spanv2): Process standalone spans with v2 pipeline (#5290)

There are still a few differences between the pipelines (see differences
in the added integration tests). In follow-ups I will be step by step
take care of them.

Author: Dav1dde

relay-server/src/processing/spans/mod.rs

@@ -120,6 +120,11 @@ impl processing::Processor for SpansProcessor {
             .take_items_by(ItemContainer::<SpanV2>::is_container)
             .into_vec();
 
+        let legacy = envelope
+            .envelope_mut()
+            .take_items_by(|item| matches!(item.ty(), ItemType::Span))
+            .into_vec();
+
         let integrations = envelope
             .envelope_mut()
             .take_items_by(|item| matches!(item.integration(), Some(Integration::Spans(_))))
@@ -128,6 +133,7 @@ impl processing::Processor for SpansProcessor {
         let work = SerializedSpans {
             headers,
             spans,
+            legacy,
             integrations,
         };
         Some(Managed::from_envelope(envelope, work))
@@ -215,11 +221,12 @@ pub struct SerializedSpans {
     /// Original envelope headers.
     headers: EnvelopeHeaders,
 
-    /// A list of spans waiting to be processed.
-    ///
-    /// All items contained here must be spans.
+    /// A list of span 'v2' item containers.
     spans: Vec<Item>,
 
+    /// A list of legacy span 'v1' items.
+    legacy: Vec<Item>,
+
     /// Spans which Relay received from arbitrary integrations.
     integrations: Vec<Item>,
 }
@@ -235,7 +242,10 @@ impl SerializedSpans {
 
 impl Counted for SerializedSpans {
     fn quantities(&self) -> Quantities {
-        let quantity = (outcome_count(&self.spans) + outcome_count(&self.integrations)) as usize;
+        let quantity = (outcome_count(&self.spans)
+            + outcome_count(&self.legacy)
+            + outcome_count(&self.integrations)) as usize;
+
         smallvec::smallvec![
             (DataCategory::Span, quantity),
             (DataCategory::SpanIndexed, quantity),

relay-server/src/processing/spans/process.rs

@@ -2,10 +2,10 @@ use relay_event_normalization::{
     GeoIpLookup, RequiredMode, SchemaProcessor, TimestampProcessor, TrimmingProcessor, eap,
 };
 use relay_event_schema::processor::{ProcessingState, ValueType, process_value};
-use relay_event_schema::protocol::SpanV2;
+use relay_event_schema::protocol::{Span, SpanV2};
 use relay_protocol::Annotated;
 
-use crate::envelope::{ContainerItems, Item, ItemContainer};
+use crate::envelope::{ContainerItems, Item, ItemContainer, WithHeader};
 use crate::extractors::RequestMeta;
 use crate::managed::Managed;
 use crate::processing::Context;
@@ -20,11 +20,18 @@ pub fn expand(spans: Managed<SampledSpans>) -> Managed<ExpandedSpans> {
         let mut all_spans = Vec::new();
 
         for item in &spans.inner.spans {
-            let expanded = expand_span(item);
+            let expanded = expand_span_container(item);
             let expanded = records.or_default(expanded, item);
             all_spans.extend(expanded);
         }
 
+        for item in &spans.inner.legacy {
+            match expand_legacy_span(item) {
+                Ok(span) => all_spans.push(span),
+                Err(err) => drop(records.reject_err(err, item)),
+            }
+        }
+
         spans::integrations::expand_into(&mut all_spans, records, spans.inner.integrations);
 
         ExpandedSpans {
@@ -35,7 +42,7 @@ pub fn expand(spans: Managed<SampledSpans>) -> Managed<ExpandedSpans> {
     })
 }
 
-fn expand_span(item: &Item) -> Result<ContainerItems<SpanV2>> {
+fn expand_span_container(item: &Item) -> Result<ContainerItems<SpanV2>> {
     let spans = ItemContainer::parse(item)
         .map_err(|err| {
             relay_log::debug!("failed to parse span container: {err}");
@@ -46,6 +53,19 @@ fn expand_span(item: &Item) -> Result<ContainerItems<SpanV2>> {
     Ok(spans)
 }
 
+fn expand_legacy_span(item: &Item) -> Result<WithHeader<SpanV2>> {
+    let payload = item.payload();
+
+    let span = Annotated::<Span>::from_json_bytes(&payload)
+        .map_err(|err| {
+            relay_log::debug!("failed to parse span: {err}");
+            Error::Invalid(DiscardReason::InvalidJson)
+        })?
+        .map_value(relay_spans::span_v1_to_span_v2);
+
+    Ok(WithHeader::new(span))
+}
+
 /// Normalizes individual spans.
 pub fn normalize(spans: &mut Managed<ExpandedSpans>, geo_lookup: &GeoIpLookup) {
     spans.retain_with_context(

relay-server/src/services/processor.rs

@@ -319,8 +319,11 @@ impl ProcessingGroup {
                 item.integration(),
                 Some(Integration::Spans(SpansIntegration::OtelV1 { .. }))
             );
+            let is_span = matches!(item.ty(), &ItemType::Span);
 
-            ItemContainer::<SpanV2>::is_container(item) || (exp_feature && is_supported_integration)
+            ItemContainer::<SpanV2>::is_container(item)
+                || (exp_feature && is_span)
+                || (exp_feature && is_supported_integration)
         });
 
         if !span_v2_items.is_empty() {