feat(processor): Add initial Span Attachment logic (#5363)

Co-authored-by: Joris Bayer <[email protected]>

Author: tobias-wilfert

relay-dynamic-config/src/feature.rs

@@ -129,6 +129,9 @@ pub enum Feature {
     #[doc(hidden)]
     #[serde(rename = "organizations:indexed-spans-extraction")]
     DeprecatedExtractSpansFromEvent,
+    /// Enable the experimental Span Attachment subset of the Span V2 processing pipeline in Relay.
+    #[serde(rename = "projects:span-v2-attachment-processing")]
+    SpanV2AttachmentProcessing,
     /// Forward compatibility.
     #[doc(hidden)]
     #[serde(other)]

relay-event-schema/src/protocol/attachment_v2.rs

@@ -0,0 +1,34 @@
+use relay_protocol::{Annotated, Empty, FromValue, IntoValue, Object, Value};
+
+use crate::processor::ProcessValue;
+use crate::protocol::{Attributes, Timestamp};
+
+use uuid::Uuid;
+
+/// Metadata for a span attachment.
+#[derive(Clone, Debug, Default, PartialEq, Empty, FromValue, IntoValue, ProcessValue)]
+pub struct AttachmentV2Meta {
+    /// Unique identifier for this attachment.
+    #[metastructure(required = true, nonempty = true, trim = false)]
+    pub attachment_id: Annotated<Uuid>,
+
+    /// Timestamp when the attachment was created.
+    #[metastructure(required = true, trim = false)]
+    pub timestamp: Annotated<Timestamp>,
+
+    /// Original filename of the attachment.
+    #[metastructure(pii = "true", max_chars = 256, max_chars_allowance = 40, trim = false)]
+    pub filename: Annotated<String>,
+
+    /// Content type of the attachment body.
+    #[metastructure(required = true, max_chars = 128, trim = false)]
+    pub content_type: Annotated<String>,
+
+    /// Arbitrary attributes on a span attachment.
+    #[metastructure(pii = "maybe")]
+    pub attributes: Annotated<Attributes>,
+
+    /// Additional arbitrary fields for forwards compatibility.
+    #[metastructure(additional_properties, pii = "maybe")]
+    pub other: Object<Value>,
+}

relay-event-schema/src/protocol/mod.rs

@@ -1,5 +1,6 @@
 //! Implements the sentry event protocol.
 
+mod attachment_v2;
 mod attributes;
 mod base;
 mod breadcrumb;
@@ -41,6 +42,7 @@ mod utils;
 #[doc(inline)]
 pub use relay_base_schema::{events::*, spans::*};
 
+pub use self::attachment_v2::*;
 pub use self::attributes::*;
 pub use self::breadcrumb::*;
 pub use self::breakdowns::*;

relay-server/src/envelope/content_type.rs

@@ -35,6 +35,8 @@ pub enum ContentType {
     SpanV2Container,
     /// `application/vnd.sentry.items.trace-metric+json`
     TraceMetricContainer,
+    /// `application/vnd.sentry.attachment.v2`
+    AttachmentV2,
     /// All integration content types.
     Integration(Integration),
     /// Any arbitrary content type not listed explicitly.
@@ -57,6 +59,7 @@ impl ContentType {
             Self::LogContainer => "application/vnd.sentry.items.log+json",
             Self::SpanV2Container => "application/vnd.sentry.items.span.v2+json",
             Self::TraceMetricContainer => "application/vnd.sentry.items.trace-metric+json",
+            Self::AttachmentV2 => "application/vnd.sentry.attachment.v2",
             Self::Integration(integration) => integration.as_content_type(),
             Self::Other(other) => other,
         }
@@ -99,6 +102,8 @@ impl ContentType {
             Some(Self::SpanV2Container)
         } else if ct.eq_ignore_ascii_case(Self::TraceMetricContainer.as_str()) {
             Some(Self::TraceMetricContainer)
+        } else if ct.eq_ignore_ascii_case(Self::AttachmentV2.as_str()) {
+            Some(Self::AttachmentV2)
         } else {
             Integration::from_content_type(ct).map(Self::Integration)
         }

relay-server/src/envelope/item.rs

@@ -6,7 +6,7 @@ use std::ops::AddAssign;
 use uuid::Uuid;
 
 use bytes::Bytes;
-use relay_event_schema::protocol::EventType;
+use relay_event_schema::protocol::{EventType, SpanId};
 use relay_protocol::Value;
 use relay_quotas::DataCategory;
 use serde::{Deserialize, Serialize};
@@ -45,6 +45,8 @@ impl Item {
                 fully_normalized: false,
                 profile_type: None,
                 platform: None,
+                parent_id: None,
+                meta_length: None,
             },
             payload: Bytes::new(),
         }
@@ -129,7 +131,7 @@ impl Item {
             ItemType::Nel => smallvec![],
             ItemType::UnrealReport => smallvec![(DataCategory::Error, item_count)],
             ItemType::Attachment => smallvec![
-                (DataCategory::Attachment, self.len().max(1)),
+                (DataCategory::Attachment, self.attachment_body_size()),
                 (DataCategory::AttachmentItem, item_count),
             ],
             ItemType::Session | ItemType::Sessions => {
@@ -434,6 +436,53 @@ impl Item {
         self.headers.sampled = sampled;
     }
 
+    /// Returns the length of the item.
+    pub fn meta_length(&self) -> Option<u32> {
+        self.headers.meta_length
+    }
+
+    /// Sets the length of the optional meta segment.
+    ///
+    /// Only applicable if the item is an attachment.
+    pub fn set_meta_length(&mut self, meta_length: u32) {
+        self.headers.meta_length = Some(meta_length);
+    }
+
+    /// Returns the parent entity that this item is associated with, if any.
+    ///
+    /// Only applicable if the item is an attachment.
+    pub fn parent_id(&self) -> Option<&ParentId> {
+        self.headers.parent_id.as_ref()
+    }
+
+    /// Sets the parent entity that this item is associated with.
+    pub fn set_parent_id(&mut self, parent_id: ParentId) {
+        self.headers.parent_id = Some(parent_id);
+    }
+
+    /// Returns `true` if this item is an attachment with AttachmentV2 content type.
+    pub fn is_attachment_v2(&self) -> bool {
+        self.ty() == &ItemType::Attachment
+            && self.content_type() == Some(&ContentType::AttachmentV2)
+    }
+
+    /// Returns the attachment payload size.
+    ///
+    /// For AttachmentV2, returns only the size of the actual payload, excluding the attachment meta.
+    /// For Attachment, returns the size of entire payload.
+    ///
+    /// **Note:** This relies on the `meta_length` header which might not be correct as such this
+    /// is best effort.
+    pub fn attachment_body_size(&self) -> usize {
+        if self.is_attachment_v2() {
+            self.len()
+                .saturating_sub(self.meta_length().unwrap_or(0) as usize)
+        } else {
+            self.len()
+        }
+        .max(1)
+    }
+
     /// Returns the specified header value, if present.
     pub fn get_header<K>(&self, name: &K) -> Option<&Value>
     where
@@ -947,6 +996,18 @@ pub struct ItemHeaders {
     #[serde(default, skip)]
     profile_type: Option<ProfileType>,
 
+    /// Content length of an optional meta segment that might be contained in the item.
+    ///
+    /// For the time being such an meta segment is only present for span attachments.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    meta_length: Option<u32>,
+
+    /// Parent entity that this item is associated with, if any.
+    ///
+    /// For the time being only applicable if the item is a span-attachment.
+    #[serde(flatten, skip_serializing_if = "Option::is_none")]
+    parent_id: Option<ParentId>,
+
     /// Other attributes for forward compatibility.
     #[serde(flatten)]
     other: BTreeMap<String, Value>,
@@ -994,6 +1055,18 @@ fn is_true(value: &bool) -> bool {
     *value
 }
 
+/// Parent identifier for an attachment-v2.
+///
+/// Attachments can be associated with different types of parent entities (only spans for now).
+///
+/// SpanId(None) indicates that the item is a span-attachment that is associated with no specific
+/// span.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ParentId {
+    SpanId(Option<SpanId>),
+}
+
 #[cfg(test)]
 mod tests {
     use crate::integrations::OtelFormat;

relay-server/src/managed/counted.rs

@@ -61,6 +61,10 @@ impl Counted for Box<Envelope> {
 
         let data = [
             (DataCategory::Attachment, summary.attachment_quantity),
+            (
+                DataCategory::AttachmentItem,
+                summary.attachment_item_quantity,
+            ),
             (DataCategory::Profile, summary.profile_quantity),
             (DataCategory::ProfileIndexed, summary.profile_quantity),
             (DataCategory::Span, summary.span_quantity),

relay-server/src/managed/envelope.rs

@@ -362,6 +362,14 @@ impl ManagedEnvelope {
             );
         }
 
+        if self.context.summary.attachment_item_quantity > 0 {
+            self.track_outcome(
+                outcome.clone(),
+                DataCategory::AttachmentItem,
+                self.context.summary.attachment_item_quantity,
+            );
+        }
+
         if self.context.summary.monitor_quantity > 0 {
             self.track_outcome(
                 outcome.clone(),

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

@@ -91,6 +91,7 @@ pub fn validate_dsc(spans: &ExpandedSpans) -> Result<()> {
     };
 
     for span in &spans.spans {
+        let span = &span.span;
         let trace_id = get_value!(span.trace_id);
 
         if trace_id != Some(&dsc.trace_id) {
@@ -293,6 +294,7 @@ struct UnsampledSpans {
     spans: Vec<Item>,
     legacy: Vec<Item>,
     integrations: Vec<Item>,
+    attachments: Vec<Item>,
 }
 
 impl From<SerializedSpans> for UnsampledSpans {
@@ -302,12 +304,14 @@ impl From<SerializedSpans> for UnsampledSpans {
             spans,
             legacy,
             integrations,
+            attachments,
         } = value;
 
         Self {
             spans,
             legacy,
             integrations,
+            attachments,
         }
     }
 }
@@ -318,6 +322,22 @@ impl Counted for UnsampledSpans {
             + outcome_count(&self.legacy)
             + outcome_count(&self.integrations)) as usize;
 
-        smallvec::smallvec![(DataCategory::SpanIndexed, quantity)]
+        let mut quantities = smallvec::smallvec![];
+
+        if quantity > 0 {
+            quantities.push((DataCategory::SpanIndexed, quantity));
+        }
+        if !self.attachments.is_empty() {
+            quantities.push((
+                DataCategory::Attachment,
+                self.attachments
+                    .iter()
+                    .map(Item::attachment_body_size)
+                    .sum(),
+            ));
+            quantities.push((DataCategory::AttachmentItem, self.attachments.len()));
+        }
+
+        quantities
     }
 }

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

@@ -5,7 +5,7 @@ use relay_protocol::Annotated;
 use crate::extractors::RequestMeta;
 use crate::managed::Managed;
 use crate::processing::Context;
-use crate::processing::spans::{Error, ExpandedSpans, Result};
+use crate::processing::spans::{Error, ExpandedSpans, Result, SerializedSpans};
 
 /// Filters standalone spans sent for a project which does not allow standalone span ingestion.
 pub fn feature_flag(ctx: Context<'_>) -> Result<()> {
@@ -15,11 +15,25 @@ pub fn feature_flag(ctx: Context<'_>) -> Result<()> {
     }
 }
 
+// Filters span attachments for a project which does not allow for span attachment ingestion.
+pub fn feature_flag_attachment(
+    spans: Managed<SerializedSpans>,
+    ctx: Context<'_>,
+) -> Managed<SerializedSpans> {
+    spans.map(|mut spans, r| {
+        if ctx.should_filter(Feature::SpanV2AttachmentProcessing) {
+            let attachments = std::mem::take(&mut spans.attachments);
+            r.reject_err(Error::FilterFeatureFlag, attachments);
+        }
+        spans
+    })
+}
+
 /// Applies inbound filters to individual spans.
 pub fn filter(spans: &mut Managed<ExpandedSpans>, ctx: Context<'_>) {
     spans.retain_with_context(
         |spans| (&mut spans.spans, spans.headers.meta()),
-        |span, meta, _| filter_span(span, meta, ctx),
+        |span, meta, _| filter_span(&span.span, meta, ctx),
     );
 }