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

relay-general/tests/snapshots/test_fixtures__event_schema.snap

@@ -1228,6 +1228,7 @@ expression: event_json_schema()
         "expectct",
         "expectstaple",
         "transaction",
+        "replayevent",
         "default"
       ]
     },

relay-quotas/src/quota.rs

@@ -103,6 +103,7 @@ impl CategoryUnit {
             DataCategory::Default
             | DataCategory::Error
             | DataCategory::Transaction
+            | DataCategory::ReplayEvent
             | DataCategory::Security => Some(Self::Count),
             DataCategory::Attachment => Some(Self::Bytes),
             DataCategory::Session => Some(Self::Batched),

relay-server/src/actors/envelopes.rs

@@ -1283,6 +1283,7 @@ impl EnvelopeProcessor {
             ItemType::ClientReport => false,
             ItemType::Profile => false,
             ItemType::ReplayRecording => false,
+            ItemType::ReplayEvent => false,
         }
     }
 
@@ -1302,6 +1303,7 @@ impl EnvelopeProcessor {
         // `process_event`.
         let event_item = envelope.take_item_by(|item| item.ty() == ItemType::Event);
         let transaction_item = envelope.take_item_by(|item| item.ty() == ItemType::Transaction);
+        let replay_item = envelope.take_item_by(|item| item.ty() == ItemType::ReplayEvent);
         let security_item = envelope.take_item_by(|item| item.ty() == ItemType::Security);
         let raw_security_item = envelope.take_item_by(|item| item.ty() == ItemType::RawSecurity);
         let form_item = envelope.take_item_by(|item| item.ty() == ItemType::FormData);
@@ -1333,6 +1335,12 @@ impl EnvelopeProcessor {
                 // hint to normalization that we're dealing with a transaction now.
                 self.event_from_json_payload(item, Some(EventType::Transaction))?
             })
+        } else if let Some(mut item) = replay_item {
+            relay_log::trace!("processing json replay event");
+            state.sample_rates = item.take_sample_rates();
+            metric!(timer(RelayTimers::EventProcessingDeserialize), {
+                self.event_from_json_payload(item, Some(EventType::ReplayEvent))?
+            })
         } else if let Some(mut item) = raw_security_item {
             relay_log::trace!("processing security report");
             state.sample_rates = item.take_sample_rates();

relay-server/src/actors/store.rs

@@ -60,6 +60,7 @@ struct Producers {
     metrics: Producer,
     profiles: Producer,
     replay_recordings: Producer,
+    replay_events: Producer,
 }
 
 impl Producers {
@@ -78,6 +79,7 @@ impl Producers {
             KafkaTopic::Metrics => Some(&self.metrics),
             KafkaTopic::Profiles => Some(&self.profiles),
             KafkaTopic::ReplayRecordings => Some(&self.replay_recordings),
+            KafkaTopic::ReplayEvents => Some(&self.replay_events),
         }
     }
 }
@@ -140,6 +142,11 @@ impl StoreForwarder {
                 &mut reused_producers,
                 KafkaTopic::ReplayRecordings,
             )?,
+            replay_events: make_producer(
+                &*config,
+                &mut reused_producers,
+                KafkaTopic::ReplayEvents,
+            )?,
         };
 
         Ok(Self { config, producers })
@@ -193,6 +200,7 @@ impl StoreForwarder {
                 id: id.clone(),
                 chunk_index,
             });
+
             self.produce(topic, attachment_message)?;
             offset += chunk_size;
             chunk_index += 1;
@@ -444,6 +452,27 @@ impl StoreForwarder {
         );
         Ok(())
     }
+    fn produce_replay_event(
+        &self,
+        event_id: EventId,
+        project_id: ProjectId,
+        start_time: Instant,
+        item: &Item,
+    ) -> Result<(), StoreError> {
+        let message = ReplayEventKafkaMessage {
+            event_id,
+            project_id,
+            start_time: UnixTimestamp::from_instant(start_time).as_secs(),
+            payload: item.payload(),
+        };
+        relay_log::trace!("Sending replay event to Kafka");
+        self.produce(KafkaTopic::ReplayEvents, KafkaMessage::ReplayEvent(message))?;
+        metric!(
+            counter(RelayCounters::ProcessingMessageProduced) += 1,
+            event_type = "replay_event"
+        );
+        Ok(())
+    }
 }
 
 /// StoreMessageForwarder is an async actor since the only thing it does is put the messages
@@ -531,6 +560,17 @@ struct EventKafkaMessage {
     /// Attachments that are potentially relevant for processing.
     attachments: Vec<ChunkedAttachment>,
 }
+#[derive(Clone, Debug, Serialize)]
+struct ReplayEventKafkaMessage {
+    /// Raw event payload.
+    payload: Bytes,
+    /// Time at which the event was received by Relay.
+    start_time: u64,
+    /// The event id.
+    event_id: EventId,
+    /// The project id for the current event.
+    project_id: ProjectId,
+}
 
 /// Container payload for chunks of attachments.
 #[derive(Debug, Serialize)]
@@ -631,6 +671,7 @@ enum KafkaMessage {
     Metric(MetricKafkaMessage),
     Profile(ProfileKafkaMessage),
     ReplayRecording(AttachmentKafkaMessage),
+    ReplayEvent(ReplayEventKafkaMessage),
 }
 
 impl KafkaMessage {
@@ -644,6 +685,7 @@ impl KafkaMessage {
             KafkaMessage::Metric(_) => "metric",
             KafkaMessage::Profile(_) => "profile",
             KafkaMessage::ReplayRecording(_) => "replay_recording",
+            KafkaMessage::ReplayEvent(_) => "replay_event",
         }
     }
 
@@ -658,6 +700,7 @@ impl KafkaMessage {
             Self::Metric(_message) => Uuid::nil(),  // TODO(ja): Determine a partitioning key
             Self::Profile(_message) => Uuid::nil(),
             Self::ReplayRecording(_message) => Uuid::nil(),
+            Self::ReplayEvent(_message) => Uuid::nil(),
         };
 
         if uuid.is_nil() {
@@ -790,6 +833,12 @@ impl Handler<StoreEnvelope> for StoreForwarder {
                     )?;
                     replay_recordings.push(replay_recording);
                 }
+                ItemType::ReplayEvent => self.produce_replay_event(
+                    event_id.ok_or(StoreError::NoEventId)?,
+                    scoping.project_id,
+                    start_time,
+                    item,
+                )?,
                 _ => {}
             }
         }

relay-server/src/envelope.rs

@@ -104,8 +104,10 @@ pub enum ItemType {
     ClientReport,
     /// Profile event payload encoded in JSON
     Profile,
-    /// Replay Recording blob payload
+    /// Replay Recording payload
     ReplayRecording,
+    /// Replay metadata and breadcrumb payload
+    ReplayEvent,
 }
 
 impl ItemType {
@@ -114,6 +116,7 @@ impl ItemType {
         match event_type {
             EventType::Default | EventType::Error => ItemType::Event,
             EventType::Transaction => ItemType::Transaction,
+            EventType::ReplayEvent => ItemType::ReplayEvent,
             EventType::Csp | EventType::Hpkp | EventType::ExpectCt | EventType::ExpectStaple => {
                 ItemType::Security
             }
@@ -139,6 +142,7 @@ impl fmt::Display for ItemType {
             Self::ClientReport => write!(f, "client report"),
             Self::Profile => write!(f, "profile"),
             Self::ReplayRecording => write!(f, "replay_recording"),
+            Self::ReplayEvent => write!(f, "replay_event"),
         }
     }
 }
@@ -557,6 +561,7 @@ impl Item {
             | ItemType::Transaction
             | ItemType::Security
             | ItemType::RawSecurity
+            | ItemType::ReplayEvent
             | ItemType::UnrealReport => true,
 
             // Attachments are only event items if they are crash reports or if they carry partial
@@ -601,6 +606,7 @@ impl Item {
             ItemType::RawSecurity => true,
             ItemType::UnrealReport => true,
             ItemType::UserReport => true,
+            ItemType::ReplayEvent => true,
             ItemType::Session => false,
             ItemType::Sessions => false,
             ItemType::Metrics => false,

relay-server/src/metrics_extraction/transactions.rs

@@ -296,7 +296,9 @@ fn extract_transaction_metrics_inner(
     event: &Event,
     mut push_metric: impl FnMut(Metric),
 ) {
-    if event.ty.value() != Some(&EventType::Transaction) {
+    if event.ty.value() != Some(&EventType::Transaction)
+        && event.ty.value() != Some(&EventType::ReplayEvent)
+    {
         return;
     }
 

relay-server/src/utils/rate_limits.rs

@@ -105,6 +105,7 @@ fn infer_event_category(item: &Item) -> Option<DataCategory> {
         ItemType::UserReport => None,
         ItemType::Profile => None,
         ItemType::ReplayRecording => None,
+        ItemType::ReplayEvent => None,
         // the following items are "internal" item types.  From the perspective of the SDK
         // the use the "internal" data category however this data category is in fact never
         // supposed to be emitted by relay as internal items must not be rate limited.  As

relay-server/src/utils/sizes.rs

@@ -25,6 +25,7 @@ pub fn check_envelope_size_limits(config: &Config, envelope: &Envelope) -> bool
             ItemType::Event
             | ItemType::Transaction
             | ItemType::Security
+            | ItemType::ReplayEvent
             | ItemType::RawSecurity
             | ItemType::FormData => event_size += item.len(),
             ItemType::Attachment | ItemType::UnrealReport | ItemType::ReplayRecording => {

tests/integration/conftest.py

@@ -22,9 +22,10 @@
     outcomes_consumer,
     transactions_consumer,
     attachments_consumer,
-    replay_recordings_consumer,
     sessions_consumer,
     metrics_consumer,
+    replay_recordings_consumer,
+    replay_events_consumer,
 )  # noqa
 
 

tests/integration/fixtures/__init__.py

@@ -166,7 +166,6 @@ def send_envelope(self, project_id, envelope, headers=None, dsn_key_idx=0):
             "X-Sentry-Auth": self.get_auth_header(project_id, dsn_key_idx),
             **(headers or {}),
         }
-
         response = self.post(url, headers=headers, data=envelope.serialize())
         response.raise_for_status()
 
@@ -196,6 +195,14 @@ def send_transaction(self, project_id, payload, item_headers=None):
 
         self.send_envelope(project_id, envelope)
 
+    def send_replay_event(self, project_id, payload, item_headers=None):
+        envelope = Envelope()
+        envelope.add_item(Item(payload=PayloadRef(json=payload), type="replay_event"))
+        if envelope.headers is None:
+            envelope.headers = {}
+
+        self.send_envelope(project_id, envelope)
+
     def send_session_aggregates(self, project_id, payload):
         envelope = Envelope()
         envelope.add_item(Item(payload=PayloadRef(json=payload), type="sessions"))