feat(spans): Ingest standalone spans

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 **Features**:
 
+- Ingest OpenTelemetry and standalone Sentry spans via HTTP or an envelope. ([#2620](https://github.com/getsentry/relay/pull/2620))
 - Partition and split metric buckets just before sending. Log outcomes for metrics. ([#2682](https://github.com/getsentry/relay/pull/2682))
 - Return global config ready status to downstream relays. ([#2765](https://github.com/getsentry/relay/pull/2765))
 - Add Mixed JS/Android Profiles events processing. ([#2706](https://github.com/getsentry/relay/pull/2706))

relay-dynamic-config/src/feature.rs

@@ -33,6 +33,9 @@ pub enum Feature {
     /// Enable processing profiles
     #[serde(rename = "organizations:profiling")]
     Profiling,
+    /// Enable standalone span ingestion.
+    #[serde(rename = "organizations:standalone-span-ingestion")]
+    StandaloneSpanIngestion,
     /// Enable metric metadata.
     #[serde(rename = "organizations:metric-meta")]
     MetricMeta,

relay-event-normalization/src/lib.rs

@@ -33,7 +33,11 @@ mod trimming;
 pub mod replay;
 pub use normalize::breakdowns::*;
 pub use normalize::*;
+pub use remove_other::RemoveOtherProcessor;
+pub use schema::SchemaProcessor;
+pub use timestamp::TimestampProcessor;
 pub use transactions::*;
+pub use trimming::TrimmingProcessor;
 pub use user_agent::*;
 
 pub use self::clock_drift::*;

relay-event-normalization/src/normalize/span/tag_extraction.rs

@@ -122,7 +122,7 @@ impl std::fmt::Display for RenderBlockingStatus {
 }
 
 /// Configuration for span tag extraction.
-pub(crate) struct Config {
+pub struct Config {
     /// The maximum allowed size of tag values in bytes. Longer values will be cropped.
     pub max_tag_value_size: usize,
 }
@@ -216,7 +216,7 @@ pub fn extract_shared_tags(event: &Event) -> BTreeMap<SpanTagKey, String> {
 /// [span operations](https://develop.sentry.dev/sdk/performance/span-operations/) and
 /// existing [span data](https://develop.sentry.dev/sdk/performance/span-data-conventions/) fields,
 /// and rely on Sentry conventions and heuristics.
-pub(crate) fn extract_tags(
+pub fn extract_tags(
     span: &Span,
     config: &Config,
     initial_display: Option<Timestamp>,

relay-event-normalization/src/remove_other.rs

@@ -9,6 +9,7 @@ fn create_errors(other: &mut Object<Value>) {
     }
 }
 
+/// Removes unknown, internal and deprecated fields from a payload.
 pub struct RemoveOtherProcessor;
 
 impl Processor for RemoveOtherProcessor {

relay-event-normalization/src/schema.rs

@@ -3,6 +3,7 @@ use relay_event_schema::processor::{
 };
 use relay_protocol::{Array, Empty, Error, ErrorKind, Meta, Object};
 
+/// Validates constraints such as empty strings or arrays and invalid characters.
 pub struct SchemaProcessor;
 
 impl Processor for SchemaProcessor {

relay-event-normalization/src/trimming.rs

@@ -14,12 +14,14 @@ struct BagSizeState {
     size_remaining: usize,
 }
 
+/// Limits data bags to a maximum size and depth.
 #[derive(Default)]
 pub struct TrimmingProcessor {
     bag_size_state: Vec<BagSizeState>,
 }
 
 impl TrimmingProcessor {
+    /// Creates a new trimming processor.
     pub fn new() -> Self {
         Self::default()
     }

relay-server/Cargo.toml

@@ -80,6 +80,7 @@ relay-quotas = { path = "../relay-quotas" }
 relay-redis = { path = "../relay-redis" }
 relay-replays = { path = "../relay-replays" }
 relay-sampling = { path = "../relay-sampling" }
+relay-spans = { path = "../relay-spans" }
 relay-statsd = { path = "../relay-statsd" }
 relay-system = { path = "../relay-system" }
 reqwest = { version = "0.11.1", features = [

relay-server/src/actors/processor.rs

@@ -44,8 +44,7 @@ use tokio::sync::Semaphore;
 use {
     crate::actors::project_cache::UpdateRateLimits,
     crate::utils::{EnvelopeLimiter, ItemAction, MetricsLimiter},
-    relay_event_normalization::{span, StoreConfig, StoreProcessor},
-    relay_event_schema::protocol::Span,
+    relay_event_normalization::{StoreConfig, StoreProcessor},
     relay_metrics::{Aggregator, RedisMetricMetaStore},
     relay_quotas::{RateLimitingError, RedisRateLimiter},
     relay_redis::RedisPool,
@@ -73,6 +72,7 @@ mod profile;
 mod replay;
 mod report;
 mod session;
+mod span;
 
 /// The minimum clock drift for correction to apply.
 const MINIMUM_CLOCK_DRIFT: Duration = Duration::from_secs(55 * 60);
@@ -1123,172 +1123,6 @@ impl EnvelopeProcessorService {
         Ok(())
     }
 
-    #[cfg(feature = "processing")]
-    fn is_span_allowed(&self, span: &Span) -> bool {
-        let Some(op) = span.op.value() else {
-            return false;
-        };
-        let Some(description) = span.description.value() else {
-            return false;
-        };
-        let system: &str = span
-            .data
-            .value()
-            .and_then(|v| v.get("span.system"))
-            .and_then(|system| system.as_str())
-            .unwrap_or_default();
-        op.contains("resource.script")
-            || op.contains("resource.css")
-            || op == "http.client"
-            || op.starts_with("app.")
-            || op.starts_with("ui.load")
-            || op.starts_with("file")
-            || op.starts_with("db")
-                && !(op.contains("clickhouse")
-                    || op.contains("mongodb")
-                    || op.contains("redis")
-                    || op.contains("compiler"))
-                && !(op == "db.sql.query" && (description.contains("\"$") || system == "mongodb"))
-    }
-
-    #[cfg(feature = "processing")]
-    fn extract_spans(&self, state: &mut ProcessEnvelopeState) {
-        // For now, drop any spans submitted by the SDK.
-        state.managed_envelope.retain_items(|item| match item.ty() {
-            ItemType::Span => ItemAction::DropSilently,
-            _ => ItemAction::Keep,
-        });
-
-        // Only extract spans from transactions (not errors).
-        if state.event_type() != Some(EventType::Transaction) {
-            return;
-        };
-
-        // Check feature flag.
-        if !state
-            .project_state
-            .has_feature(Feature::SpanMetricsExtraction)
-        {
-            return;
-        };
-
-        let mut add_span = |span: Annotated<Span>| {
-            let span = match self.validate_span(span) {
-                Ok(span) => span,
-                Err(e) => {
-                    relay_log::error!("Invalid span: {e}");
-                    return;
-                }
-            };
-            let span = match span.to_json() {
-                Ok(span) => span,
-                Err(e) => {
-                    relay_log::error!(error = &e as &dyn Error, "Failed to serialize span");
-                    return;
-                }
-            };
-            let mut item = Item::new(ItemType::Span);
-            item.set_payload(ContentType::Json, span);
-            state.managed_envelope.envelope_mut().add_item(item);
-        };
-
-        let Some(event) = state.event.value() else {
-            return;
-        };
-
-        // Extract transaction as a span.
-        let mut transaction_span: Span = event.into();
-
-        let all_modules_enabled = state
-            .project_state
-            .has_feature(Feature::SpanMetricsExtractionAllModules);
-
-        // Add child spans as envelope items.
-        if let Some(child_spans) = event.spans.value() {
-            for span in child_spans {
-                let Some(inner_span) = span.value() else {
-                    continue;
-                };
-                // HACK: filter spans based on module until we figure out grouping.
-                if !all_modules_enabled && !self.is_span_allowed(inner_span) {
-                    continue;
-                }
-                // HACK: clone the span to set the segment_id. This should happen
-                // as part of normalization once standalone spans reach wider adoption.
-                let mut new_span = inner_span.clone();
-                new_span.is_segment = Annotated::new(false);
-                new_span.received = transaction_span.received.clone();
-                new_span.segment_id = transaction_span.segment_id.clone();
-
-                // If a profile is associated with the transaction, also associate it with its
-                // child spans.
-                new_span.profile_id = transaction_span.profile_id.clone();
-
-                add_span(Annotated::new(new_span));
-            }
-        }
-
-        // Extract tags to add to this span as well
-        let shared_tags = span::tag_extraction::extract_shared_tags(event);
-        transaction_span.sentry_tags = Annotated::new(
-            shared_tags
-                .clone()
-                .into_iter()
-                .map(|(k, v)| (k.sentry_tag_key().to_owned(), Annotated::new(v)))
-                .collect(),
-        );
-        add_span(transaction_span.into());
-    }
-
-    /// Helper for [`Self::extract_spans`].
-    ///
-    /// We do not extract spans with missing fields if those fields are required on the Kafka topic.
-    #[cfg(feature = "processing")]
-    fn validate_span(&self, mut span: Annotated<Span>) -> Result<Annotated<Span>, anyhow::Error> {
-        let inner = span
-            .value_mut()
-            .as_mut()
-            .ok_or(anyhow::anyhow!("empty span"))?;
-        let Span {
-            ref exclusive_time,
-            ref mut tags,
-            ref mut sentry_tags,
-            ..
-        } = inner;
-        // The following required fields are already validated by the `TransactionsProcessor`:
-        // - `timestamp`
-        // - `start_timestamp`
-        // - `trace_id`
-        // - `span_id`
-        //
-        // `is_segment` is set by `extract_span`.
-        exclusive_time
-            .value()
-            .ok_or(anyhow::anyhow!("missing exclusive_time"))?;
-
-        if let Some(sentry_tags) = sentry_tags.value_mut() {
-            sentry_tags.retain(|key, value| match value.value() {
-                Some(s) => {
-                    match key.as_str() {
-                        "group" => {
-                            // Only allow up to 16-char hex strings in group.
-                            s.len() <= 16 && s.chars().all(|c| c.is_ascii_hexdigit())
-                        }
-                        "status_code" => s.parse::<u16>().is_ok(),
-                        _ => true,
-                    }
-                }
-                // Drop empty string values.
-                None => false,
-            });
-        }
-        if let Some(tags) = tags.value_mut() {
-            tags.retain(|_, value| !value.value().is_empty())
-        }
-
-        Ok(span)
-    }
-
     /// Computes the sampling decision on the incoming event
     fn run_dynamic_sampling(&self, state: &mut ProcessEnvelopeState) {
         // Running dynamic sampling involves either:
@@ -1550,6 +1384,7 @@ impl EnvelopeProcessorService {
         );
         replay::process(state, self.inner.config.clone())?;
         profile::filter(state);
+        span::filter(state);
 
         if state.creates_event() {
             // Some envelopes only create events in processing relays; for example, unreal events.
@@ -1591,13 +1426,14 @@ impl EnvelopeProcessorService {
             self.enforce_quotas(state)?;
             profile::process(state, self.inner.config.clone());
             self.process_check_ins(state);
+            span::process(state, self.inner.config.clone());
         });
 
         if state.has_event() {
             self.scrub_event(state)?;
             self.serialize_event(state)?;
             if_processing!({
-                self.extract_spans(state);
+                span::extract_from_event(state);
             });
         }
 

relay-server/src/actors/processor/event.rs

@@ -296,6 +296,7 @@ fn is_duplicate(item: &Item, processing_enabled: bool) -> bool {
         ItemType::ReplayRecording => false,
         ItemType::CheckIn => false,
         ItemType::Span => false,
+        ItemType::OtelSpan => false,
 
         // Without knowing more, `Unknown` items are allowed to be repeated
         ItemType::Unknown(_) => false,