feat(replays): add replay events itemtype

Author: JoshFerge

CHANGELOG.md

@@ -11,6 +11,7 @@
 
 - Add sampling + tagging by event platform and transaction op. Some (unused) tagging rules from 22.4.0 have been renamed. ([#1231](https://github.com/getsentry/relay/pull/1231))
 - Add ReplayRecording ItemType. ([#1236](https://github.com/getsentry/relay/pull/1236))
+- Add ReplayEvent ItemType. ([#1239](https://github.com/getsentry/relay/pull/1239))
 
 ## 22.4.0
 

relay-common/src/constants.rs

@@ -36,6 +36,8 @@ pub enum EventType {
     ExpectStaple,
     /// Performance monitoring transactions carrying spans.
     Transaction,
+    /// Replay Events for session replays
+    ReplayEvent,
     /// All events that do not qualify as any other type.
     #[serde(other)]
     Default,
@@ -86,6 +88,7 @@ impl fmt::Display for EventType {
             EventType::ExpectCt => write!(f, "expectct"),
             EventType::ExpectStaple => write!(f, "expectstaple"),
             EventType::Transaction => write!(f, "transaction"),
+            EventType::ReplayEvent => write!(f, "replay_event"),
         }
     }
 }
@@ -109,6 +112,8 @@ pub enum DataCategory {
     Session = 5,
     /// Reserved data category that shall not appear in the outcomes.
     Internal = -2,
+    /// Replay Events
+    ReplayEvent = 6,
     /// Any other data category not known by this Relay.
     #[serde(other)]
     Unknown = -1,
@@ -125,6 +130,7 @@ impl DataCategory {
             "attachment" => Self::Attachment,
             "session" => Self::Session,
             "internal" => Self::Internal,
+            "replay_event" => Self::ReplayEvent,
             _ => Self::Unknown,
         }
     }
@@ -139,6 +145,7 @@ impl DataCategory {
             Self::Attachment => "attachment",
             Self::Session => "session",
             Self::Internal => "internal",
+            Self::ReplayEvent => "replay_event",
             Self::Unknown => "unknown",
         }
     }
@@ -175,6 +182,7 @@ impl From<EventType> for DataCategory {
         match ty {
             EventType::Default | EventType::Error => Self::Error,
             EventType::Transaction => Self::Transaction,
+            EventType::ReplayEvent => Self::ReplayEvent,
             EventType::Csp | EventType::Hpkp | EventType::ExpectCt | EventType::ExpectStaple => {
                 Self::Security
             }

relay-config/src/config.rs

@@ -767,8 +767,10 @@ pub enum KafkaTopic {
     Metrics,
     /// Profiles
     Profiles,
-    /// ReplayRecordings, large blobs sent by the replay sdk
+    /// ReplayRecordings, large blobs
     ReplayRecordings,
+    /// ReplayEvents, breadcrumb + session updates for replays
+    ReplayEvents,
 }
 
 /// Configuration for topics.
@@ -793,6 +795,8 @@ pub struct TopicAssignments {
     pub profiles: TopicAssignment,
     /// Recordings topic name.
     pub replay_recordings: TopicAssignment,
+    /// Replay Events topic name.
+    pub replay_events: TopicAssignment,
 }
 
 impl TopicAssignments {
@@ -808,6 +812,7 @@ impl TopicAssignments {
             KafkaTopic::Metrics => &self.metrics,
             KafkaTopic::Profiles => &self.profiles,
             KafkaTopic::ReplayRecordings => &self.replay_recordings,
+            KafkaTopic::ReplayEvents => &self.replay_events,
         }
     }
 }
@@ -824,6 +829,7 @@ impl Default for TopicAssignments {
             metrics: "ingest-metrics".to_owned().into(),
             profiles: "profiles".to_owned().into(),
             replay_recordings: "ingest-replay-recordings".to_owned().into(),
+            replay_events: "ingest-replay-events".to_owned().into(),
         }
     }
 }

relay-general/src/store/normalize.rs

@@ -114,7 +114,9 @@ impl<'a> NormalizeProcessor<'a> {
 
     /// Ensure measurements interface is only present for transaction events
     fn normalize_measurements(&self, event: &mut Event) {
-        if event.ty.value() != Some(&EventType::Transaction) {
+        if event.ty.value() != Some(&EventType::Transaction)
+            && event.ty.value() != Some(&EventType::ReplayEvent)
+        {
             // Only transaction events may have a measurements interface
             event.measurements = Annotated::empty();
         }
@@ -129,7 +131,9 @@ impl<'a> NormalizeProcessor<'a> {
     }
 
     fn normalize_spans(&self, event: &mut Event) {
-        if event.ty.value() == Some(&EventType::Transaction) {
+        if event.ty.value() == Some(&EventType::Transaction)
+            || event.ty.value() == Some(&EventType::ReplayEvent)
+        {
             spans::normalize_spans(event, &self.config.span_attributes);
         }
     }
@@ -260,6 +264,9 @@ impl<'a> NormalizeProcessor<'a> {
         if event.ty.value() == Some(&EventType::Transaction) {
             return EventType::Transaction;
         }
+        if event.ty.value() == Some(&EventType::ReplayEvent) {
+            return EventType::ReplayEvent;
+        }
 
         // The SDKs do not describe event types, and we must infer them from available attributes.
         let has_exceptions = event