feat(metrics-summaries): Emit summaries to their own topic

CHANGELOG.md

@@ -9,6 +9,7 @@
 **Internal**:
 
 - Set the span op on segments. ([#3082](https://github.com/getsentry/relay/pull/3082))
+- Push metrics summaries to their own topic. ([#3045](https://github.com/getsentry/relay/pull/3045))
 
 ## 24.1.2
 

relay-kafka/src/config.rs

@@ -51,14 +51,16 @@ pub enum KafkaTopic {
     Monitors,
     /// Standalone spans without a transaction.
     Spans,
+    /// Summary for metrics collected during a span.
+    MetricsSummaries,
 }
 
 impl KafkaTopic {
     /// Returns iterator over the variants of [`KafkaTopic`].
     /// It will have to be adjusted if the new variants are added.
     pub fn iter() -> std::slice::Iter<'static, Self> {
         use KafkaTopic::*;
-        static TOPICS: [KafkaTopic; 13] = [
+        static TOPICS: [KafkaTopic; 14] = [
             Events,
             Attachments,
             Transactions,
@@ -72,6 +74,7 @@ impl KafkaTopic {
             ReplayRecordings,
             Monitors,
             Spans,
+            MetricsSummaries,
         ];
         TOPICS.iter()
     }
@@ -111,6 +114,8 @@ pub struct TopicAssignments {
     pub monitors: TopicAssignment,
     /// Standalone spans without a transaction.
     pub spans: TopicAssignment,
+    /// Summary for metrics collected during a span.
+    pub metrics_summaries: TopicAssignment,
 }
 
 impl TopicAssignments {
@@ -131,6 +136,7 @@ impl TopicAssignments {
             KafkaTopic::ReplayRecordings => &self.replay_recordings,
             KafkaTopic::Monitors => &self.monitors,
             KafkaTopic::Spans => &self.spans,
+            KafkaTopic::MetricsSummaries => &self.metrics_summaries,
         }
     }
 }
@@ -152,6 +158,7 @@ impl Default for TopicAssignments {
             replay_recordings: "ingest-replay-recordings".to_owned().into(),
             monitors: "ingest-monitors".to_owned().into(),
             spans: "snuba-spans".to_owned().into(),
+            metrics_summaries: "snuba-metrics-summaries".to_owned().into(),
         }
     }
 }

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

@@ -299,7 +299,7 @@ fn normalize(
 
     let NormalizeSpanConfig {
         received_at,
-        timestamp_range: timestmap_range,
+        timestamp_range,
         max_tag_value_size,
         performance_score,
         measurements,
@@ -323,7 +323,7 @@ fn normalize(
     )?;
 
     if let Some(span) = annotated_span.value() {
-        validate_span(span, Some(&timestmap_range))?;
+        validate_span(span, Some(&timestamp_range))?;
     }
     process_value(
         annotated_span,

relay-server/src/services/store.rs

@@ -871,6 +871,12 @@ impl StoreService {
             }
         };
 
+        span.duration_ms = ((span.end_timestamp - span.start_timestamp) * 1e3) as u32;
+        span.event_id = event_id;
+        span.project_id = scoping.project_id.value();
+        span.retention_days = retention_days;
+        span.start_timestamp_ms = (span.start_timestamp * 1e3) as u64;
+
         if let Some(measurements) = &mut span.measurements {
             measurements.retain(|_, v| {
                 v.as_ref()
@@ -879,30 +885,7 @@ impl StoreService {
             });
         }
 
-        if let Some(metrics_summary) = &mut span.metrics_summary {
-            metrics_summary.retain(|_, mut v| {
-                if let Some(v) = &mut v {
-                    v.retain(|v| {
-                        if let Some(v) = v {
-                            return v.min.is_some()
-                                || v.max.is_some()
-                                || v.sum.is_some()
-                                || v.count.is_some();
-                        }
-                        false
-                    });
-                    !v.is_empty()
-                } else {
-                    false
-                }
-            });
-        }
-
-        span.duration_ms = ((span.end_timestamp - span.start_timestamp) * 1e3) as u32;
-        span.event_id = event_id;
-        span.project_id = scoping.project_id.value();
-        span.retention_days = retention_days;
-        span.start_timestamp_ms = (span.start_timestamp * 1e3) as u64;
+        self.produce_metrics_summary(scoping, start_time, item, &span);
 
         self.produce(
             KafkaTopic::Spans,
@@ -927,6 +910,121 @@ impl StoreService {
 
         Ok(())
     }
+
+    fn produce_metrics_summary(
+        &self,
+        scoping: Scoping,
+        start_time: Instant,
+        item: &Item,
+        span: &SpanKafkaMessage,
+    ) {
+        let payload = item.payload();
+        let d = &mut Deserializer::from_slice(&payload);
+        let mut metrics_summary: SpanWithMetricsSummary = match serde_path_to_error::deserialize(d)
+        {
+            Ok(span) => span,
+            Err(error) => {
+                relay_log::error!(
+                    error = &error as &dyn std::error::Error,
+                    "failed to parse span"
+                );
+                self.outcome_aggregator.send(TrackOutcome {
+                    category: DataCategory::SpanIndexed,
+                    event_id: None,
+                    outcome: Outcome::Invalid(DiscardReason::InvalidSpan),
+                    quantity: 1,
+                    remote_addr: None,
+                    scoping,
+                    timestamp: instant_to_date_time(start_time),
+                });
+                return;
+            }
+        };
+        let Some(metrics_summary) = &mut metrics_summary.metrics_summary else {
+            return;
+        };
+        let &SpanKafkaMessage {
+            duration_ms,
+            end_timestamp,
+            is_segment,
+            project_id,
+            received,
+            retention_days,
+            segment_id,
+            span_id,
+            trace_id,
+            ..
+        } = span;
+
+        metrics_summary.retain(|_, mut v| {
+            if let Some(v) = &mut v {
+                v.retain(|v| {
+                    if let Some(v) = v {
+                        return v.min.is_some()
+                            || v.max.is_some()
+                            || v.sum.is_some()
+                            || v.count.is_some();
+                    }
+                    false
+                });
+                !v.is_empty()
+            } else {
+                false
+            }
+        });
+
+        let group = span
+            .sentry_tags
+            .as_ref()
+            .and_then(|sentry_tags| sentry_tags.get("group"))
+            .map_or("", String::as_str);
+
+        for (mri, summaries) in metrics_summary {
+            let Some(summaries) = summaries else {
+                continue;
+            };
+            for summary in summaries {
+                let Some(SpanMetricsSummary {
+                    count,
+                    max,
+                    min,
+                    sum,
+                    tags,
+                }) = summary
+                else {
+                    continue;
+                };
+                // Ignore immediate errors on produce.
+                if let Err(error) = self.produce(
+                    KafkaTopic::MetricsSummaries,
+                    scoping.organization_id,
+                    KafkaMessage::MetricsSummary(MetricsSummaryKafkaMessage {
+                        count,
+                        duration_ms,
+                        end_timestamp,
+                        group,
+                        is_segment,
+                        max,
+                        min,
+                        mri,
+                        project_id,
+                        received,
+                        retention_days,
+                        segment_id: segment_id.unwrap_or_default(),
+                        span_id,
+                        sum,
+                        tags,
+                        trace_id,
+                    }),
+                ) {
+                    relay_log::error!(
+                        error = &error as &dyn std::error::Error,
+                        "failed to push metrics summary to kafka",
+                    );
+                }
+            }
+        }
+    }
 }
 
 impl Service for StoreService {
@@ -1211,22 +1309,6 @@ struct SpanMeasurement {
     value: Option<f64>,
 }
 
-#[derive(Debug, Deserialize, Serialize)]
-struct SpanMetricsSummary {
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    count: Option<u64>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    max: Option<f64>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    min: Option<f64>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    sum: Option<f64>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    tags: Option<BTreeMap<String, String>>,
-}
-
-type SpanMetricsSummaries = Vec<Option<SpanMetricsSummary>>;
-
 #[derive(Debug, Deserialize, Serialize)]
 struct SpanKafkaMessage<'a> {
     #[serde(skip_serializing)]
@@ -1247,13 +1329,6 @@ struct SpanKafkaMessage<'a> {
 
     #[serde(borrow, default, skip_serializing_if = "Option::is_none")]
     measurements: Option<BTreeMap<Cow<'a, str>, Option<SpanMeasurement>>>,
-    #[serde(
-        borrow,
-        default,
-        rename = "_metrics_summary",
-        skip_serializing_if = "Option::is_none"
-    )]
-    metrics_summary: Option<BTreeMap<Cow<'a, str>, Option<SpanMetricsSummaries>>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     parent_span_id: Option<&'a str>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -1269,7 +1344,7 @@ struct SpanKafkaMessage<'a> {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     segment_id: Option<&'a str>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    sentry_tags: Option<&'a RawValue>,
+    sentry_tags: Option<BTreeMap<&'a str, String>>,
     span_id: &'a str,
     #[serde(default)]
     start_timestamp_ms: u64,
@@ -1278,6 +1353,54 @@ struct SpanKafkaMessage<'a> {
     trace_id: &'a str,
 }
 
+#[derive(Debug, Deserialize)]
+struct SpanMetricsSummary {
+    #[serde(default)]
+    count: Option<u64>,
+    #[serde(default)]
+    max: Option<f64>,
+    #[serde(default)]
+    min: Option<f64>,
+    #[serde(default)]
+    sum: Option<f64>,
+    #[serde(default)]
+    tags: BTreeMap<String, String>,
+}
+
+type SpanMetricsSummaries = Vec<Option<SpanMetricsSummary>>;
+
+#[derive(Debug, Deserialize)]
+struct SpanWithMetricsSummary {
+    #[serde(default, rename(deserialize = "_metrics_summary"))]
+    metrics_summary: Option<BTreeMap<String, Option<SpanMetricsSummaries>>>,
+}
+
+#[derive(Debug, Serialize)]
+struct MetricsSummaryKafkaMessage<'a> {
+    duration_ms: u32,
+    end_timestamp: f64,
+    group: &'a str,
+    is_segment: bool,
+    mri: &'a str,
+    project_id: u64,
+    received: f64,
+    retention_days: u16,
+    segment_id: &'a str,
+    span_id: &'a str,
+    trace_id: &'a str,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    count: &'a Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    max: &'a Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    min: &'a Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    sum: &'a Option<f64>,
+    #[serde(skip_serializing_if = "BTreeMap::is_empty")]
+    tags: &'a BTreeMap<String, String>,
+}
+
 /// An enum over all possible ingest messages.
 #[derive(Debug, Serialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
@@ -1299,6 +1422,7 @@ enum KafkaMessage<'a> {
     ReplayRecordingNotChunked(ReplayRecordingNotChunkedKafkaMessage),
     CheckIn(CheckInKafkaMessage),
     Span(SpanKafkaMessage<'a>),
+    MetricsSummary(MetricsSummaryKafkaMessage<'a>),
 }
 
 impl Message for KafkaMessage<'_> {
@@ -1315,6 +1439,7 @@ impl Message for KafkaMessage<'_> {
             KafkaMessage::ReplayRecordingNotChunked(_) => "replay_recording_not_chunked",
             KafkaMessage::CheckIn(_) => "check_in",
             KafkaMessage::Span(_) => "span",
+            KafkaMessage::MetricsSummary(_) => "metrics_summary",
         }
     }
 
@@ -1337,7 +1462,8 @@ impl Message for KafkaMessage<'_> {
             Self::Session(_)
             | Self::Profile(_)
             | Self::ReplayRecordingNotChunked(_)
-            | Self::Span(_) => Uuid::nil(),
+            | Self::Span(_)
+            | Self::MetricsSummary(_) => Uuid::nil(),
 
             // TODO(ja): Determine a partitioning key
             Self::Metric { .. } => Uuid::nil(),
@@ -1383,6 +1509,10 @@ impl Message for KafkaMessage<'_> {
             KafkaMessage::Span(message) => {
                 serde_json::to_vec(message).map_err(ClientError::InvalidJson)
             }
+            KafkaMessage::MetricsSummary(message) => {
+                serde_json::to_vec(message).map_err(ClientError::InvalidJson)
+            }
+
             _ => rmp_serde::to_vec_named(&self).map_err(ClientError::InvalidMsgPack),
         }
     }

tests/integration/conftest.py

@@ -30,6 +30,7 @@
     monitors_consumer,
     spans_consumer,
     profiles_consumer,
+    metrics_summaries_consumer,
 )