fix(spans): Extract metrics from transaction spans

CHANGELOG.md

@@ -7,6 +7,7 @@
 - Enable `db.redis` span metrics extraction. ([#3283](https://github.com/getsentry/relay/pull/3283))
 - Add a data category for continuous profiling. ([#3284](https://github.com/getsentry/relay/pull/3284))
 - Apply rate limits to span metrics. ([#3255](https://github.com/getsentry/relay/pull/3255))
+- Extract metrics from transaction spans. ([#3273](https://github.com/getsentry/relay/pull/3273))
 - Implement volume metric stats. ([#3281](https://github.com/getsentry/relay/pull/3281))
 
 ## 24.3.0

relay-server/src/metrics_extraction/event.rs

@@ -5,6 +5,7 @@ use relay_metrics::Bucket;
 use relay_quotas::DataCategory;
 
 use crate::metrics_extraction::generic::{self, Extractable};
+use crate::services::processor::extract_transaction_span;
 use crate::statsd::RelayTimers;
 
 impl Extractable for Event {
@@ -47,6 +48,9 @@ pub fn extract_metrics(event: &Event, config: &MetricExtractionConfig) -> Vec<Bu
     let mut metrics = generic::extract_metrics(event, config);
 
     relay_statsd::metric!(timer(RelayTimers::EventProcessingSpanMetricsExtraction), {
+        let transaction_span = extract_transaction_span(event);
+        metrics.extend(generic::extract_metrics(&transaction_span, config));
+
         if let Some(spans) = event.spans.value() {
             for annotated_span in spans {
                 if let Some(span) = annotated_span.value() {
@@ -1245,7 +1249,7 @@ mod tests {
             .filter(|b| &*b.name == "c:spans/usage@none")
             .collect::<Vec<_>>();
 
-        let expected_usage = 8; // We count all spans received by Relay
+        let expected_usage = 9; // We count all spans received by Relay, plus one for the transaction
         assert_eq!(usage_metrics.len(), expected_usage);
         for m in usage_metrics {
             assert!(m.tags.is_empty());
@@ -1440,4 +1444,38 @@ mod tests {
             assert!(metrics.iter().any(|b| &*b.name == mri));
         }
     }
+
+    #[test]
+    fn extracts_span_metrics_from_transaction() {
+        let event = r#"
+            {
+                "type": "transaction",
+                "timestamp": "2021-04-26T08:00:05+0100",
+                "start_timestamp": "2021-04-26T08:00:00+0100",
+                "contexts": {
+                    "trace": {
+                        "op": "db.query"
+                    }
+                }
+            }
+            "#;
+        let event = Annotated::<Event>::from_json(event).unwrap();
+
+        // Create a project config with the relevant feature flag. Sanitize to fill defaults.
+        let mut project = ProjectConfig {
+            features: [Feature::ExtractSpansAndSpanMetricsFromEvent]
+                .into_iter()
+                .collect(),
+            ..ProjectConfig::default()
+        };
+        project.sanitize();
+
+        let config = project.metric_extraction.ok().unwrap();
+        let metrics = extract_metrics(event.value().unwrap(), &config);
+
+        assert_eq!(metrics.len(), 2);
+        assert_eq!(&*metrics[0].name, "c:spans/usage@none");
+        assert_eq!(&*metrics[1].name, "c:spans/count_per_op@none");
+        assert_eq!(&*metrics[1].tags["span.op"], "db.query");
+    }
 }

relay-server/src/metrics_extraction/snapshots/relay_server__metrics_extraction__event__tests__extract_span_metrics.snap

@@ -3,6 +3,51 @@ source: relay-server/src/metrics_extraction/event.rs
 expression: metrics
 ---
 [
+    Bucket {
+        timestamp: UnixTimestamp(1619420400),
+        width: 0,
+        name: "c:spans/usage@none",
+        value: Counter(
+            1.0,
+        ),
+        tags: {},
+        metadata: BucketMetadata {
+            merges: 1,
+        },
+    },
+    Bucket {
+        timestamp: UnixTimestamp(1619420400),
+        width: 0,
+        name: "d:spans/exclusive_time@millisecond",
+        value: Distribution(
+            [
+                59000.0,
+            ],
+        ),
+        tags: {
+            "environment": "fake_environment",
+            "span.op": "mYOp",
+            "transaction": "gEt /api/:version/users/",
+            "transaction.op": "myop",
+        },
+        metadata: BucketMetadata {
+            merges: 1,
+        },
+    },
+    Bucket {
+        timestamp: UnixTimestamp(1619420400),
+        width: 0,
+        name: "c:spans/count_per_op@none",
+        value: Counter(
+            1.0,
+        ),
+        tags: {
+            "span.op": "mYOp",
+        },
+        metadata: BucketMetadata {
+            merges: 1,
+        },
+    },
     Bucket {
         timestamp: UnixTimestamp(1597976302),
         width: 0,

relay-server/src/metrics_extraction/snapshots/relay_server__metrics_extraction__event__tests__extract_span_metrics_mobile.snap

@@ -422,6 +422,64 @@ expression: "(&event.value().unwrap().spans, metrics)"
         },
     ],
     [
+        Bucket {
+            timestamp: UnixTimestamp(1619420400),
+            width: 0,
+            name: "c:spans/usage@none",
+            value: Counter(
+                1.0,
+            ),
+            tags: {},
+            metadata: BucketMetadata {
+                merges: 1,
+            },
+        },
+        Bucket {
+            timestamp: UnixTimestamp(1619420400),
+            width: 0,
+            name: "d:spans/exclusive_time@millisecond",
+            value: Distribution(
+                [
+                    59000.0,
+                ],
+            ),
+            tags: {
+                "span.op": "default",
+                "transaction": "gEt /api/:version/users/",
+            },
+            metadata: BucketMetadata {
+                merges: 1,
+            },
+        },
+        Bucket {
+            timestamp: UnixTimestamp(1619420400),
+            width: 0,
+            name: "c:spans/count_per_op@none",
+            value: Counter(
+                1.0,
+            ),
+            tags: {
+                "span.op": "default",
+            },
+            metadata: BucketMetadata {
+                merges: 1,
+            },
+        },
+        Bucket {
+            timestamp: UnixTimestamp(1619420400),
+            width: 0,
+            name: "c:spans/count_per_segment@none",
+            value: Counter(
+                1.0,
+            ),
+            tags: {
+                "release": "1.2.3",
+                "transaction": "gEt /api/:version/users/",
+            },
+            metadata: BucketMetadata {
+                merges: 1,
+            },
+        },
         Bucket {
             timestamp: UnixTimestamp(1597976302),
             width: 0,

relay-server/src/services/processor.rs

@@ -94,6 +94,8 @@ mod replay;
 mod report;
 mod session;
 mod span;
+pub use span::extract_transaction_span;
+
 #[cfg(feature = "processing")]
 mod unreal;
 

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

@@ -1,6 +1,9 @@
 //! Processor code related to standalone spans.
 
 use relay_dynamic_config::Feature;
+use relay_event_normalization::span::tag_extraction;
+use relay_event_schema::protocol::{Event, Span};
+use relay_protocol::Annotated;
 
 use crate::services::processor::SpanGroup;
 use crate::{envelope::ItemType, services::processor::ProcessEnvelopeState, utils::ItemAction};
@@ -26,3 +29,23 @@ pub fn filter(state: &mut ProcessEnvelopeState<SpanGroup>) {
         _ => ItemAction::Keep,
     });
 }
+
+/// Creates a span from the transaction and applies tag extraction on it.
+pub fn extract_transaction_span(event: &Event) -> Span {
+    let mut transaction_span: Span = event.into();
+
+    let mut shared_tags = tag_extraction::extract_shared_tags(event);
+    if let Some(span_op) = transaction_span.op.value() {
+        shared_tags.insert(tag_extraction::SpanTagKey::SpanOp, span_op.to_owned());
+    }
+
+    transaction_span.sentry_tags = Annotated::new(
+        shared_tags
+            .clone()
+            .into_iter()
+            .map(|(k, v)| (k.sentry_tag_key().to_owned(), Annotated::new(v)))
+            .collect(),
+    );
+
+    transaction_span
+}

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

@@ -23,6 +23,7 @@ use relay_spans::{otel_to_sentry_span, otel_trace::Span as OtelSpan};
 use crate::envelope::{ContentType, Item, ItemType};
 use crate::metrics_extraction::generic::extract_metrics;
 use crate::services::outcome::{DiscardReason, Outcome};
+use crate::services::processor::span::extract_transaction_span;
 use crate::services::processor::{
     ProcessEnvelopeState, ProcessingError, SpanGroup, TransactionGroup,
 };
@@ -191,8 +192,7 @@ pub fn extract_from_event(state: &mut ProcessEnvelopeState<TransactionGroup>) {
         return;
     };
 
-    // Extract transaction as a span.
-    let mut transaction_span: Span = event.into();
+    let transaction_span = extract_transaction_span(event);
 
     // Add child spans as envelope items.
     if let Some(child_spans) = event.spans.value() {
@@ -216,20 +216,6 @@ pub fn extract_from_event(state: &mut ProcessEnvelopeState<TransactionGroup>) {
         }
     }
 
-    // Extract tags to add to this span as well
-    let mut shared_tags = tag_extraction::extract_shared_tags(event);
-
-    if let Some(span_op) = transaction_span.op.value() {
-        shared_tags.insert(tag_extraction::SpanTagKey::SpanOp, span_op.to_owned());
-    }
-
-    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());
 }
 
@@ -246,24 +232,24 @@ pub fn maybe_discard_transaction(state: &mut ProcessEnvelopeState<TransactionGro
 #[derive(Clone, Debug)]
 struct NormalizeSpanConfig<'a> {
     /// The time at which the event was received in this Relay.
-    pub received_at: DateTime<Utc>,
+    received_at: DateTime<Utc>,
     /// Allowed time range for spans.
-    pub timestamp_range: std::ops::Range<UnixTimestamp>,
+    timestamp_range: std::ops::Range<UnixTimestamp>,
     /// The maximum allowed size of tag values in bytes. Longer values will be cropped.
-    pub max_tag_value_size: usize,
+    max_tag_value_size: usize,
     /// Configuration for generating performance score measurements for web vitals
-    pub performance_score: Option<&'a PerformanceScoreConfig>,
+    performance_score: Option<&'a PerformanceScoreConfig>,
     /// Configuration for measurement normalization in transaction events.
     ///
     /// Has an optional [`relay_event_normalization::MeasurementsConfig`] from both the project and the global level.
     /// If at least one is provided, then normalization will truncate custom measurements
     /// and add units of known built-in measurements.
-    pub measurements: Option<DynamicMeasurementsConfig<'a>>,
+    measurements: Option<DynamicMeasurementsConfig<'a>>,
     /// The maximum length for names of custom measurements.
     ///
     /// Measurements with longer names are removed from the transaction event and replaced with a
     /// metadata entry.
-    pub max_name_and_unit_len: Option<usize>,
+    max_name_and_unit_len: Option<usize>,
 }
 
 fn get_normalize_span_config<'a>(

relay-server/src/testutils.rs

@@ -7,6 +7,7 @@ use relay_dynamic_config::ErrorBoundary;
 use relay_event_schema::protocol::EventId;
 use relay_protocol::RuleCondition;
 use relay_sampling::config::{DecayingFunction, RuleId, RuleType, SamplingRule, SamplingValue};
+
 use relay_sampling::{DynamicSamplingContext, SamplingConfig};
 use relay_system::Addr;
 use relay_test::mock_service;

tests/integration/fixtures/processing.py

@@ -174,7 +174,8 @@ def assert_empty(self, timeout=None):
         test message ends up in the same partition as the message we are checking).
         """
         # First, give Relay a bit of time to process
-        assert self.poll(timeout=0.2) is None
+        rv = self.poll(timeout=0.2)
+        assert rv is None, f"not empty: {rv.value()}"
 
         # Then, send a custom message to ensure we're not just timing out
         message = json.dumps({"__test__": uuid.uuid4().hex}).encode("utf8")