feat(replays): add replay events ItemType

CHANGELOG.md

@@ -43,6 +43,8 @@
 - Refactor aggregation error, recover from errors more gracefully. ([#1240](https://github.com/getsentry/relay/pull/1240))
 - Remove/reject nul-bytes from metric strings. ([#1235](https://github.com/getsentry/relay/pull/1235))
 - Remove the unused "internal" data category. ([#1245](https://github.com/getsentry/relay/pull/1245))
+- Add ReplayEvent ItemType. ([#1239](https://github.com/getsentry/relay/pull/1239))
+- Add ReplayRecording ItemType. ([#1236](https://github.com/getsentry/relay/pull/1236))
 - Add the client and version as `sdk` tag to extracted session metrics in the format `name/version`. ([#1248](https://github.com/getsentry/relay/pull/1248))
 - Expose `shutdown_timeout` in `OverridableConfig` ([#1247](https://github.com/getsentry/relay/pull/1247))
 - Normalize all profiles and reject invalid ones. ([#1250](https://github.com/getsentry/relay/pull/1250))

relay-common/src/constants.rs

@@ -109,6 +109,8 @@ pub enum DataCategory {
     Session = 5,
     /// A profile
     Profile = 6,
+    /// Replay Events
+    ReplayEvent = 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_event" => Self::ReplayEvent,
             "profile" => Self::Profile,
             _ => Self::Unknown,
         }
@@ -138,6 +141,7 @@ impl DataCategory {
             Self::Security => "security",
             Self::Attachment => "attachment",
             Self::Session => "session",
+            Self::ReplayEvent => "replay_event",
             Self::Profile => "profile",
             Self::Unknown => "unknown",
         }

relay-config/src/config.rs

@@ -779,6 +779,10 @@ pub enum KafkaTopic {
     Metrics,
     /// Profiles
     Profiles,
+    /// ReplayEvents, breadcrumb + session updates for replays
+    ReplayEvents,
+    /// ReplayRecording, large blobs sent by the replay sdk
+    ReplayRecordings,
 }
 
 /// Configuration for topics.
@@ -801,6 +805,10 @@ 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,
 }
 
 impl TopicAssignments {
@@ -815,6 +823,8 @@ impl TopicAssignments {
             KafkaTopic::Sessions => &self.sessions,
             KafkaTopic::Metrics => &self.metrics,
             KafkaTopic::Profiles => &self.profiles,
+            KafkaTopic::ReplayEvents => &self.replay_events,
+            KafkaTopic::ReplayRecordings => &self.replay_recordings,
         }
     }
 }
@@ -830,6 +840,8 @@ 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::ReplayEvent
             | DataCategory::Security
             | DataCategory::Profile => Some(Self::Count),
             DataCategory::Attachment => Some(Self::Bytes),

relay-server/src/actors/envelopes.rs

@@ -1006,6 +1006,22 @@ impl EnvelopeProcessor {
         });
     }
 
+    /// 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::ReplayEvent | ItemType::ReplayRecording => {
+                    if !replays_enabled {
+                        return false;
+                    }
+                    true
+                }
+                _ => true, // Keep all other item types
+            }
+        });
+    }
+
     /// Creates and initializes the processing state.
     ///
     /// This applies defaults to the envelope and initializes empty rate limits.
@@ -1320,7 +1336,8 @@ 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,
         }
@@ -1834,6 +1851,7 @@ impl EnvelopeProcessor {
         self.process_client_reports(state);
         self.process_user_reports(state);
         self.process_profiles(state);
+        self.process_replays(state);
 
         if state.creates_event() {
             if_processing!({

relay-server/src/actors/project.rs

@@ -55,6 +55,9 @@ pub enum Feature {
     #[serde(rename = "organizations:profiling")]
     Profiling,
 
+    #[serde(rename = "organizations:session-replay")]
+    Replays,
+
     /// Unused.
     ///
     /// This used to control the initial experimental metrics extraction for sessions and has been

relay-server/src/actors/store.rs

@@ -38,6 +38,8 @@ const MAX_EXPLODED_SESSIONS: usize = 100;
 /// Fallback name used for attachment items without a `filename` header.
 const UNNAMED_ATTACHMENT: &str = "Unnamed Attachment";
 
+const REPLAY_RECORDINGS_ATTACHMENT_NAME: &str = "rr";
+
 #[derive(Fail, Debug)]
 pub enum StoreError {
     #[fail(display = "failed to send kafka message")]
@@ -59,6 +61,8 @@ struct Producers {
     sessions: Producer,
     metrics: Producer,
     profiles: Producer,
+    replay_events: Producer,
+    replay_recordings: Producer,
 }
 
 impl Producers {
@@ -76,6 +80,8 @@ 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),
         }
     }
 }
@@ -133,6 +139,16 @@ impl StoreForwarder {
             sessions: make_producer(&*config, &mut reused_producers, KafkaTopic::Sessions)?,
             metrics: make_producer(&*config, &mut reused_producers, KafkaTopic::Metrics)?,
             profiles: make_producer(&*config, &mut reused_producers, KafkaTopic::Profiles)?,
+            replay_recordings: make_producer(
+                &*config,
+                &mut reused_producers,
+                KafkaTopic::ReplayRecordings,
+            )?,
+            replay_events: make_producer(
+                &*config,
+                &mut reused_producers,
+                KafkaTopic::ReplayEvents,
+            )?,
         };
 
         Ok(Self { config, producers })
@@ -177,7 +193,6 @@ impl StoreForwarder {
         while offset < size {
             let max_chunk_size = self.config.attachment_chunk_size();
             let chunk_size = std::cmp::min(max_chunk_size, size - offset);
-
             let attachment_message = KafkaMessage::AttachmentChunk(AttachmentChunkKafkaMessage {
                 payload: payload.slice(offset, offset + chunk_size),
                 event_id,
@@ -439,6 +454,75 @@ 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,
+        project_id: ProjectId,
+        item: &Item,
+    ) -> Result<ChunkedAttachment, StoreError> {
+        let id = Uuid::new_v4().to_string();
+
+        let mut chunk_index = 0;
+        let mut offset = 0;
+        let payload = item.payload();
+        let size = item.len();
+
+        // This skips chunks for empty replay recordings. The consumer does not require chunks for
+        // empty replay recordings. `chunks` will be `0` in this case.
+        while offset < size {
+            let max_chunk_size = self.config.attachment_chunk_size();
+            let chunk_size = std::cmp::min(max_chunk_size, size - offset);
+
+            let replay_recording_chunk_message =
+                KafkaMessage::ReplayRecordingChunk(ReplayRecordingChunkKafkaMessage {
+                    payload: payload.slice(offset, offset + chunk_size),
+                    replay_id,
+                    project_id,
+                    id: id.clone(),
+                    chunk_index,
+                });
+            self.produce(KafkaTopic::ReplayRecordings, replay_recording_chunk_message)?;
+            offset += chunk_size;
+            chunk_index += 1;
+        }
+
+        // The chunk_index is incremented after every loop iteration. After we exit the loop, it
+        // is one larger than the last chunk, so it is equal to the number of chunks.
+
+        Ok(ChunkedAttachment {
+            id,
+            name: REPLAY_RECORDINGS_ATTACHMENT_NAME.to_owned(),
+            content_type: item
+                .content_type()
+                .map(|content_type| content_type.as_str().to_owned()),
+            attachment_type: item.attachment_type().unwrap_or_default(),
+            chunks: chunk_index,
+            size: Some(size),
+            rate_limited: Some(item.rate_limited()),
+        })
+    }
 }
 
 /// StoreMessageForwarder is an async actor since the only thing it does is put the messages
@@ -526,6 +610,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)]
@@ -558,6 +653,32 @@ struct AttachmentKafkaMessage {
     attachment: ChunkedAttachment,
 }
 
+/// Container payload for chunks of attachments.
+#[derive(Debug, Serialize)]
+struct ReplayRecordingChunkKafkaMessage {
+    /// Chunk payload of the replay recording.
+    payload: Bytes,
+    /// The replay id.
+    replay_id: EventId,
+    /// The project id for the current replay.
+    project_id: ProjectId,
+    /// The recording ID within the replay.
+    ///
+    /// The triple `(project_id, replay_id, id)` identifies a replay recording chunk uniquely.
+    id: String,
+    /// Sequence number of chunk. Starts at 0 and ends at `ReplayRecordingKafkaMessage.num_chunks - 1`.
+    chunk_index: usize,
+}
+#[derive(Debug, Serialize)]
+struct ReplayRecordingKafkaMessage {
+    /// The replay id.
+    replay_id: EventId,
+    /// The project id for the current event.
+    project_id: ProjectId,
+    /// The recording attachment.
+    replay_recording: ChunkedAttachment,
+}
+
 /// User report for an event wrapped up in a message ready for consumption in Kafka.
 ///
 /// Is always independent of an event and can be sent as part of any envelope.
@@ -626,6 +747,9 @@ enum KafkaMessage {
     Session(SessionKafkaMessage),
     Metric(MetricKafkaMessage),
     Profile(ProfileKafkaMessage),
+    ReplayEvent(ReplayEventKafkaMessage),
+    ReplayRecording(ReplayRecordingKafkaMessage),
+    ReplayRecordingChunk(ReplayRecordingChunkKafkaMessage),
 }
 
 impl KafkaMessage {
@@ -638,6 +762,9 @@ impl KafkaMessage {
             KafkaMessage::Session(_) => "session",
             KafkaMessage::Metric(_) => "metric",
             KafkaMessage::Profile(_) => "profile",
+            KafkaMessage::ReplayEvent(_) => "replay_event",
+            KafkaMessage::ReplayRecording(_) => "replay_recording",
+            KafkaMessage::ReplayRecordingChunk(_) => "replay_recording_chunk",
         }
     }
 
@@ -651,6 +778,9 @@ 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) => Uuid::nil(),
+            Self::ReplayRecording(message) => message.replay_id.0,
+            Self::ReplayRecordingChunk(message) => message.replay_id.0,
         };
 
         if uuid.is_nil() {
@@ -766,6 +896,32 @@ impl Handler<StoreEnvelope> for StoreForwarder {
                     start_time,
                     item,
                 )?,
+                ItemType::ReplayRecording => {
+                    let replay_recording = self.produce_replay_recording_chunks(
+                        event_id.ok_or(StoreError::NoEventId)?,
+                        scoping.project_id,
+                        item,
+                    )?;
+                    relay_log::trace!("Sending individual replay_recordings of envelope to kafka");
+                    let replay_recording_message =
+                        KafkaMessage::ReplayRecording(ReplayRecordingKafkaMessage {
+                            replay_id: event_id.ok_or(StoreError::NoEventId)?,
+                            project_id: scoping.project_id,
+                            replay_recording,
+                        });
+
+                    self.produce(KafkaTopic::ReplayRecordings, replay_recording_message)?;
+                    metric!(
+                        counter(RelayCounters::ProcessingMessageProduced) += 1,
+                        event_type = "replay_recording"
+                    );
+                }
+                ItemType::ReplayEvent => self.produce_replay_event(
+                    event_id.ok_or(StoreError::NoEventId)?,
+                    scoping.project_id,
+                    start_time,
+                    item,
+                )?,
                 _ => {}
             }
         }