feat(replays): add replay events ItemType

CHANGELOG.md

@@ -9,7 +9,8 @@
 - Relay is now compatible with CentOS 7 and Red Hat Enterprise Linux 7 onward (kernel version _2.6.32_), depending on _glibc 2.17_ or newer. The `crash-handler` feature, which is currently enabled in the build published to DockerHub, additionally requires _curl 7.29_ or newer. ([#1279](https://github.com/getsentry/relay/pull/1279))
 - Optionally start relay with `--upstream-dsn` to pass a Sentry DSN instead of the URL. This can be convenient when starting Relay in environments close to an SDK, where a DSN is already available. ([#1277](https://github.com/getsentry/relay/pull/1277))
 - Add a new runtime mode `--aws-runtime-api=$AWS_LAMBDA_RUNTIME_API` that integrates Relay with the AWS Extensions API lifecycle. ([#1277](https://github.com/getsentry/relay/pull/1277))
-- Add ReplayRecording ItemType. ([#1236](https://github.com/getsentry/relay/pull/1236))
+- Add Replay ItemTypes. ([#1236](https://github.com/getsentry/relay/pull/1236), ([#1239](https://github.com/getsentry/relay/pull/1239)
+  )
 
 **Bug Fixes**:
 

relay-common/src/constants.rs

@@ -109,6 +109,8 @@ pub enum DataCategory {
     Session = 5,
     /// A profile
     Profile = 6,
+    /// Session Replays
+    Replay = 7,
     /// Any other data category not known by this Relay.
     #[serde(other)]
     Unknown = -1,
@@ -124,6 +126,7 @@ impl DataCategory {
             "security" => Self::Security,
             "attachment" => Self::Attachment,
             "session" => Self::Session,
+            "replay" => Self::Replay,
             "profile" => Self::Profile,
             _ => Self::Unknown,
         }
@@ -138,6 +141,7 @@ impl DataCategory {
             Self::Security => "security",
             Self::Attachment => "attachment",
             Self::Session => "session",
+            Self::Replay => "replay",
             Self::Profile => "profile",
             Self::Unknown => "unknown",
         }

relay-config/src/config.rs

@@ -783,7 +783,9 @@ pub enum KafkaTopic {
     Metrics,
     /// Profiles
     Profiles,
-    /// ReplayRecording, large blobs sent by the replay sdk
+    /// ReplayEvents, breadcrumb + session updates for replays
+    ReplayEvents,
+    /// ReplayRecordings, large blobs sent by the replay sdk
     ReplayRecordings,
 }
 
@@ -807,6 +809,8 @@ pub struct TopicAssignments {
     pub metrics: TopicAssignment,
     /// Stacktrace topic name
     pub profiles: TopicAssignment,
+    /// Replay Events topic name.
+    pub replay_events: TopicAssignment,
     /// Recordings topic name.
     pub replay_recordings: TopicAssignment,
 }
@@ -823,6 +827,7 @@ impl TopicAssignments {
             KafkaTopic::Sessions => &self.sessions,
             KafkaTopic::Metrics => &self.metrics,
             KafkaTopic::Profiles => &self.profiles,
+            KafkaTopic::ReplayEvents => &self.replay_events,
             KafkaTopic::ReplayRecordings => &self.replay_recordings,
         }
     }
@@ -839,6 +844,7 @@ impl Default for TopicAssignments {
             sessions: "ingest-sessions".to_owned().into(),
             metrics: "ingest-metrics".to_owned().into(),
             profiles: "profiles".to_owned().into(),
+            replay_events: "ingest-replay-events".to_owned().into(),
             replay_recordings: "ingest-replay-recordings".to_owned().into(),
         }
     }

relay-quotas/src/quota.rs

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

relay-server/src/actors/envelopes.rs

@@ -1006,10 +1006,11 @@ impl EnvelopeProcessor {
         });
     }
 
-    fn process_replay_recordings(&self, state: &mut ProcessEnvelopeState) {
+    /// Remove replays if the feature flag is not enabled
+    fn process_replays(&self, state: &mut ProcessEnvelopeState) {
         let replays_enabled = state.project_state.has_feature(Feature::Replays);
         state.envelope.retain_items(|item| match item.ty() {
-            ItemType::ReplayRecording => replays_enabled,
+            ItemType::ReplayEvent | ItemType::ReplayRecording => replays_enabled,
             _ => true,
         });
     }
@@ -1329,6 +1330,7 @@ impl EnvelopeProcessor {
             ItemType::MetricBuckets => false,
             ItemType::ClientReport => false,
             ItemType::Profile => false,
+            ItemType::ReplayEvent => false,
             ItemType::ReplayRecording => false,
             // Without knowing more, `Unknown` items are allowed to be repeated
             ItemType::Unknown(_) => false,
@@ -1843,7 +1845,7 @@ impl EnvelopeProcessor {
         self.process_client_reports(state);
         self.process_user_reports(state);
         self.process_profiles(state);
-        self.process_replay_recordings(state);
+        self.process_replays(state);
 
         if state.creates_event() {
             if_processing!({

relay-server/src/actors/project.rs

@@ -54,6 +54,7 @@ pub enum Feature {
     /// Enables ingestion and normalization of profiles.
     #[serde(rename = "organizations:profiling")]
     Profiling,
+    /// Enables ingestion of Session Replays (Replay Recordings and Replay Events)
     #[serde(rename = "organizations:session-replay")]
     Replays,
 

relay-server/src/actors/store.rs

@@ -59,6 +59,7 @@ struct Producers {
     sessions: Producer,
     metrics: Producer,
     profiles: Producer,
+    replay_events: Producer,
     replay_recordings: Producer,
 }
 
@@ -77,6 +78,7 @@ impl Producers {
             KafkaTopic::Sessions => Some(&self.sessions),
             KafkaTopic::Metrics => Some(&self.metrics),
             KafkaTopic::Profiles => Some(&self.profiles),
+            KafkaTopic::ReplayEvents => Some(&self.replay_events),
             KafkaTopic::ReplayRecordings => Some(&self.replay_recordings),
         }
     }
@@ -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 })
@@ -445,6 +452,28 @@ impl StoreForwarder {
         Ok(())
     }
 
+    fn produce_replay_event(
+        &self,
+        replay_id: EventId,
+        project_id: ProjectId,
+        start_time: Instant,
+        item: &Item,
+    ) -> Result<(), StoreError> {
+        let message = ReplayEventKafkaMessage {
+            replay_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(())
+    }
+
     fn produce_replay_recording_chunks(
         &self,
         replay_id: EventId,
@@ -588,6 +617,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.
+    replay_id: EventId,
+    /// The project id for the current event.
+    project_id: ProjectId,
+}
 
 /// Container payload for chunks of attachments.
 #[derive(Debug, Serialize)]
@@ -713,6 +753,7 @@ enum KafkaMessage {
     Session(SessionKafkaMessage),
     Metric(MetricKafkaMessage),
     Profile(ProfileKafkaMessage),
+    ReplayEvent(ReplayEventKafkaMessage),
     ReplayRecording(ReplayRecordingKafkaMessage),
     ReplayRecordingChunk(ReplayRecordingChunkKafkaMessage),
 }
@@ -727,6 +768,7 @@ impl KafkaMessage {
             KafkaMessage::Session(_) => "session",
             KafkaMessage::Metric(_) => "metric",
             KafkaMessage::Profile(_) => "profile",
+            KafkaMessage::ReplayEvent(_) => "replay_event",
             KafkaMessage::ReplayRecording(_) => "replay_recording",
             KafkaMessage::ReplayRecordingChunk(_) => "replay_recording_chunk",
         }
@@ -742,6 +784,7 @@ impl KafkaMessage {
             Self::Session(_message) => Uuid::nil(), // Explicit random partitioning for sessions
             Self::Metric(_message) => Uuid::nil(),  // TODO(ja): Determine a partitioning key
             Self::Profile(_message) => Uuid::nil(),
+            Self::ReplayEvent(message) => message.replay_id.0,
             Self::ReplayRecording(message) => message.replay_id.0,
             Self::ReplayRecordingChunk(message) => message.replay_id.0,
         };
@@ -879,7 +922,12 @@ impl Handler<StoreEnvelope> for StoreForwarder {
                         event_type = "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

@@ -103,6 +103,8 @@ pub enum ItemType {
     ClientReport,
     /// Profile event payload encoded in JSON
     Profile,
+    /// Replay metadata and breadcrumb payload
+    ReplayEvent,
     /// Replay Recording data
     ReplayRecording,
     /// A new item type that is yet unknown by this version of Relay.
@@ -144,6 +146,7 @@ impl fmt::Display for ItemType {
             Self::MetricBuckets => write!(f, "metric_buckets"),
             Self::ClientReport => write!(f, "client_report"),
             Self::Profile => write!(f, "profile"),
+            Self::ReplayEvent => write!(f, "replay_event"),
             Self::ReplayRecording => write!(f, "replay_recording"),
             Self::Unknown(s) => s.fmt(f),
         }
@@ -169,6 +172,7 @@ impl std::str::FromStr for ItemType {
             "metric_buckets" => Self::MetricBuckets,
             "client_report" => Self::ClientReport,
             "profile" => Self::Profile,
+            "replay_event" => Self::ReplayEvent,
             "replay_recording" => Self::ReplayRecording,
             other => Self::Unknown(other.to_owned()),
         })
@@ -625,6 +629,7 @@ impl Item {
             | ItemType::Metrics
             | ItemType::MetricBuckets
             | ItemType::ClientReport
+            | ItemType::ReplayEvent
             | ItemType::ReplayRecording
             | ItemType::Profile => false,
 
@@ -647,6 +652,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/utils/rate_limits.rs

@@ -104,6 +104,7 @@ fn infer_event_category(item: &Item) -> Option<DataCategory> {
         ItemType::FormData => None,
         ItemType::UserReport => None,
         ItemType::Profile => None,
+        ItemType::ReplayEvent => None,
         ItemType::ReplayRecording => None,
         ItemType::ClientReport => None,
         ItemType::Unknown(_) => None,

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

@@ -26,6 +26,7 @@
     replay_recordings_consumer,
     sessions_consumer,
     metrics_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"))

tests/integration/fixtures/processing.py

@@ -51,6 +51,7 @@ def inner(options=None):
                 "outcomes": get_topic_name("outcomes"),
                 "sessions": get_topic_name("sessions"),
                 "metrics": get_topic_name("metrics"),
+                "replay_events": get_topic_name("replay_events"),
                 "replay_recordings": get_topic_name("replay_recordings"),
             }
 
@@ -286,6 +287,11 @@ def replay_recordings_consumer(kafka_consumer):
     return lambda: ReplayRecordingsConsumer(*kafka_consumer("replay_recordings"))
 
 
+@pytest.fixture
+def replay_events_consumer(kafka_consumer):
+    return lambda: ReplayEventsConsumer(*kafka_consumer("replay_events"))
+
+
 class MetricsConsumer(ConsumerBase):
     def get_metric(self, timeout=None):
         message = self.poll(timeout=timeout)
@@ -360,3 +366,14 @@ def get_individual_replay(self):
         v = msgpack.unpackb(message.value(), raw=False, use_list=False)
         assert v["type"] == "replay_recording", v["type"]
         return v
+
+
+class ReplayEventsConsumer(ConsumerBase):
+    def get_replay_event(self):
+        message = self.poll()
+        assert message is not None
+        assert message.error() is None
+
+        event = msgpack.unpackb(message.value(), raw=False, use_list=False)
+        assert event["type"] == "replay_event"
+        return json.loads(event["payload"].decode("utf8")), event