feat(spans): Extract transaction from segment span

relay-sampling/src/evaluation.rs

@@ -65,9 +65,9 @@ impl<'a> ReservoirEvaluator<'a> {
         }
     }
 
-    /// Returns a shared reference to the reservoir counters.
+    /// Gets shared ownership of the reference counters.
     pub fn counters(&self) -> ReservoirCounters {
-        self.counters.clone()
+        Arc::clone(&self.counters)
     }
 
     /// Sets the Redis pool and organiation ID for the [`ReservoirEvaluator`].

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

@@ -15,6 +15,7 @@ use relay_event_normalization::{
 };
 use relay_event_schema::processor::{process_value, ProcessingState};
 use relay_event_schema::protocol::{BrowserContext, Contexts, Event, EventId, Span, SpanData};
+use relay_log::protocol::{Attachment, AttachmentType};
 use relay_metrics::{aggregator::AggregatorConfig, MetricNamespace, UnixTimestamp};
 use relay_pii::PiiProcessor;
 use relay_protocol::{Annotated, Empty};
@@ -28,7 +29,13 @@ use crate::services::processor::{
     Addrs, ProcessEnvelope, ProcessEnvelopeState, ProcessingError, ProcessingGroup, SpanGroup,
     TransactionGroup,
 };
+use crate::statsd::{RelayCounters, RelayHistograms};
 use crate::utils::{sample, ItemAction, ManagedEnvelope};
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+#[error(transparent)]
+struct ValidationError(#[from] anyhow::Error);
 
 pub fn process(
     state: &mut ProcessEnvelopeState<SpanGroup>,
@@ -88,6 +95,15 @@ pub fn process(
             _ => return ItemAction::Keep,
         };
 
+        set_segment_attributes(&mut annotated_span);
+
+        if should_extract_transactions && !item.transaction_extracted() {
+            if let Some(transaction) = convert_to_transaction(&annotated_span) {
+                extracted_transactions.push(transaction);
+                item.set_transaction_extracted(true);
+            }
+        }
+
         if let Err(e) = normalize(
             &mut annotated_span,
             normalize_span_config.clone(),
@@ -98,13 +114,6 @@ pub fn process(
             return ItemAction::Drop(Outcome::Invalid(DiscardReason::Internal));
         };
 
-        if should_extract_transactions && !item.transaction_extracted() {
-            if let Some(transaction) = convert_to_transaction(&annotated_span) {
-                extracted_transactions.push(transaction);
-                item.set_transaction_extracted(true);
-            }
-        }
-
         if let Some(config) = span_metrics_extraction_config {
             let Some(span) = annotated_span.value_mut() else {
                 return ItemAction::Drop(Outcome::Invalid(DiscardReason::Internal));
@@ -173,6 +182,7 @@ pub fn process(
         ItemAction::Keep
     });
 
+    let mut transaction_count = 0;
     for mut transaction in extracted_transactions {
         // Give each transaction event a new random ID:
         transaction.id = EventId::new().into();
@@ -186,6 +196,8 @@ pub fn process(
                     item.set_spans_extracted(true);
                 }
 
+                transaction_count += 1;
+
                 addrs.envelope_processor.send(ProcessEnvelope {
                     envelope: ManagedEnvelope::standalone(
                         envelope,
@@ -195,14 +207,21 @@ pub fn process(
                     ),
                     project_state: state.project_state.clone(),
                     sampling_project_state: state.sampling_project_state.clone(),
-                    reservoir_counters: state.reservoir.counters().clone(),
+                    reservoir_counters: state.reservoir.counters(),
                 });
             }
             Err(e) => {
                 relay_log::error!("Failed to create event envelope: {e}");
             }
         }
     }
+
+    if transaction_count > 0 {
+        relay_statsd::metric!(counter(RelayCounters::TransactionsFromSpans) += transaction_count);
+        relay_statsd::metric!(
+            histogram(RelayHistograms::TransactionsFromSpansPerEnvelope) = transaction_count as u64
+        );
+    }
 }
 
 pub fn extract_from_event(
@@ -371,6 +390,19 @@ fn get_normalize_span_config<'a>(
     }
 }
 
+fn set_segment_attributes(span: &mut Annotated<Span>) {
+    let Some(span) = span.value_mut() else { return };
+
+    // TODO: A span might be a segment span even if the parent_id is not empty
+    // (parent within a trace). I.e. do not overwrite here.
+    let is_segment = span.parent_span_id.is_empty();
+
+    span.is_segment = Annotated::new(is_segment);
+    if is_segment {
+        span.segment_id = span.span_id.clone();
+    }
+}
+
 /// Normalizes a standalone span.
 fn normalize(
     annotated_span: &mut Annotated<Span>,
@@ -438,14 +470,8 @@ fn normalize(
         );
     }
 
-    let is_segment = span.parent_span_id.is_empty();
-    span.is_segment = Annotated::new(is_segment);
     span.received = Annotated::new(received_at.into());
 
-    if is_segment {
-        span.segment_id = span.span_id.clone();
-    }
-
     if let Some(transaction) = span
         .data
         .value_mut()

relay-server/src/statsd.rs

@@ -382,6 +382,9 @@ pub enum RelayTimers {
     /// This metric is tagged with:
     ///  - `type`: The type of the health check, `liveness` or `readiness`.
     HealthCheckDuration,
+
+    /// Measurees how many transactions were created from segment spans in a single envelope.
+    TransactionsFromSpansPerEnvelope,
 }
 
 impl TimerMetric for RelayTimers {
@@ -420,6 +423,7 @@ impl TimerMetric for RelayTimers {
             RelayTimers::BufferMessageProcessDuration => "buffer.message.duration",
             RelayTimers::ProjectCacheTaskDuration => "project_cache.task.duration",
             RelayTimers::HealthCheckDuration => "health.message.duration",
+            RelayTimers::TransactionsFromSpansPerEnvelope => "transactions_from_spans_per_envelope",
         }
     }
 }
@@ -642,6 +646,8 @@ pub enum RelayCounters {
     /// This metric is tagged with:
     /// - `decision`: "drop" if dynamic sampling drops the envelope, else "keep".
     DynamicSamplingDecision,
+    /// Counts how many transactions were created from segment spans.
+    TransactionsFromSpans,
 }
 
 impl CounterMetric for RelayCounters {
@@ -680,6 +686,7 @@ impl CounterMetric for RelayCounters {
             RelayCounters::OpenTelemetryEvent => "event.opentelemetry",
             RelayCounters::GlobalConfigFetched => "global_config.fetch",
             RelayCounters::DynamicSamplingDecision => "dynamic_sampling_decision",
+            RelayCounters::TransactionsFromSpans => "transactions_from_spans",
         }
     }
 }

tests/integration/test_spans.py

@@ -410,7 +410,7 @@ def test_span_ingestion(
     relay.send_envelope(
         project_id,
         envelope,
-        headers={
+        headers={  # Set browser header to verify that `d:transactions/measurements.score.total@ratio` is extracted only once.
             "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36"
         },
     )
@@ -475,9 +475,7 @@ def test_span_ingestion(
         headers={"Content-Type": "application/x-protobuf"},
     )
 
-    print("Waiting for spans..")
     spans = list(spans_consumer.get_spans(timeout=10.0, max_attempts=6))
-    print("Done waiting for spans.")
 
     for span in spans:
         span.pop("received", None)
@@ -592,7 +590,6 @@ def test_span_ingestion(
         },
     ]
 
-    print("Asserting emptiness")
     spans_consumer.assert_empty()
 
     # If transaction extraction is enabled, expect transactions:
@@ -614,12 +611,9 @@ def test_span_ingestion(
             # No errors during normalization:
             assert not transaction.get("errors")
 
-    print("Asserting emptiness")
     transactions_consumer.assert_empty()
 
-    print("Waiting for metrics")
     metrics = [metric for (metric, _headers) in metrics_consumer.get_metrics()]
-    print("/Waiting for metrics")
     metrics.sort(key=lambda m: (m["name"], sorted(m["tags"].items()), m["timestamp"]))
     for metric in metrics:
         try: