feat(replays): Add ReplayVideo envelope item type (#3105)

Add support for replay-video event types. Changes: - Refactors event validation in the processor. This is to DRY the processing logic between the three replay event types. - Adds new replay_video field to replay-recording Kafka message. - Add outcomes. - Configure rate-limits. Relates to: * getsentry/sentry#63255 --------- Co-authored-by: Joshua Ferge <[email protected]> Co-authored-by: Josh Ferge <[email protected]> Co-authored-by: Joris Bayer <[email protected]>
getsentry · Feb 27, 2024 · 6ec6039 · 6ec6039
1 parent dc6ece7
commit 6ec6039
Show file tree

Hide file tree

Showing 15 changed files with 437 additions and 202 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -6,6 +6,7 @@
 
 - Extend GPU context with data for Unreal Engine crash reports. ([#3144](https://github.com/getsentry/relay/pull/3144))
 - Parametrize transaction in dynamic sampling context. ([#3141](https://github.com/getsentry/relay/pull/3141))
+- Adds ReplayVideo envelope-item type. ([#3105](https://github.com/getsentry/relay/pull/3105))
 - Parse & scrub span description for supabase. ([#3153](https://github.com/getsentry/relay/pull/3153), [#3156](https://github.com/getsentry/relay/pull/3156))
 
 **Bug Fixes**:

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/relay-replays/Cargo.toml b/relay-replays/Cargo.toml
@@ -10,6 +10,7 @@ license-file = "../LICENSE.md"
 publish = false
 
 [dependencies]
+bytes = { version = "1.4.0" }
 flate2 = "1.0.19"
 once_cell = { workspace = true }
 relay-common = { path = "../relay-common" }

diff --git a/relay-server/Cargo.toml b/relay-server/Cargo.toml
@@ -23,7 +23,6 @@ processing = [
     "dep:symbolic-common",
     "dep:symbolic-unreal",
     "dep:zstd",
-    "bytes/serde",
     "relay-cardinality/redis",
     "relay-config/processing",
     "relay-kafka/producer",
@@ -47,7 +46,7 @@ axum-server = "0.4.7"
 backoff = "0.4.0"
 brotli = "3.3.4"
 bytecount = "0.6.0"
-bytes = { version = "1.4.0" }
+bytes = { version = "1.4.0", features = ["serde"] }
 chrono = { workspace = true, features = ["clock"] }
 data-encoding = "2.3.3"
 flate2 = "1.0.19"
@@ -97,6 +96,7 @@ reqwest = { version = "0.11.1", features = [
 rmp-serde = "1.1.1"
 rust-embed = { version = "8.0.0", optional = true }
 serde = { workspace = true }
+serde_bytes = "0.11"
 serde_json = { workspace = true }
 smallvec = { workspace = true, features = ["drain_filter"] }
 sqlx = { version = "0.7.3", features = [

diff --git a/relay-server/src/envelope.rs b/relay-server/src/envelope.rs
@@ -114,6 +114,8 @@ pub enum ItemType {
     ReplayEvent,
     /// Replay Recording data.
     ReplayRecording,
+    /// Replay Video data.
+    ReplayVideo,
     /// Monitor check-in encoded as JSON.
     CheckIn,
     /// A standalone span.
@@ -168,6 +170,7 @@ impl ItemType {
             Self::Profile => "profile",
             Self::ReplayEvent => "replay_event",
             Self::ReplayRecording => "replay_recording",
+            Self::ReplayVideo => "replay_video",
             Self::CheckIn => "check_in",
             Self::Span => "span",
             Self::OtelSpan => "otel_span",
@@ -222,6 +225,7 @@ impl std::str::FromStr for ItemType {
             "profile" => Self::Profile,
             "replay_event" => Self::ReplayEvent,
             "replay_recording" => Self::ReplayRecording,
+            "replay_video" => Self::ReplayVideo,
             "check_in" => Self::CheckIn,
             "span" => Self::Span,
             "otel_span" => Self::OtelSpan,
@@ -661,7 +665,9 @@ impl Item {
             } else {
                 DataCategory::Profile
             }),
-            ItemType::ReplayEvent | ItemType::ReplayRecording => Some(DataCategory::Replay),
+            ItemType::ReplayEvent | ItemType::ReplayRecording | ItemType::ReplayVideo => {
+                Some(DataCategory::Replay)
+            }
             ItemType::ClientReport => None,
             ItemType::CheckIn => Some(DataCategory::Monitor),
             ItemType::Span | ItemType::OtelSpan => Some(if indexed {
@@ -873,6 +879,7 @@ impl Item {
             | ItemType::ClientReport
             | ItemType::ReplayEvent
             | ItemType::ReplayRecording
+            | ItemType::ReplayVideo
             | ItemType::Profile
             | ItemType::CheckIn
             | ItemType::Span
@@ -907,6 +914,7 @@ impl Item {
             ItemType::MetricMeta => false,
             ItemType::ClientReport => false,
             ItemType::ReplayRecording => false,
+            ItemType::ReplayVideo => false,
             ItemType::Profile => true,
             ItemType::CheckIn => false,
             ItemType::Span => false,
@@ -1765,6 +1773,22 @@ mod tests {
         assert_eq!(items[0].ty(), &ItemType::ReplayRecording);
     }
 
+    #[test]
+    fn test_deserialize_envelope_replay_video() {
+        let bytes = Bytes::from(
+            "\
+             {\"event_id\":\"9ec79c33ec9942ab8353589fcb2e04dc\",\"dsn\":\"https://e12d836b15bb49d7bbf99e64295d995b:@sentry.io/42\"}\n\
+             {\"type\":\"replay_video\"}\n\
+             helloworld\n\
+             ",
+        );
+
+        let envelope = Envelope::parse_bytes(bytes).unwrap();
+        assert_eq!(envelope.len(), 1);
+        let items: Vec<_> = envelope.items().collect();
+        assert_eq!(items[0].ty(), &ItemType::ReplayVideo);
+    }
+
     #[test]
     fn test_deserialize_envelope_view_hierarchy() {
         let bytes = Bytes::from(

diff --git a/relay-server/src/services/outcome.rs b/relay-server/src/services/outcome.rs
@@ -366,6 +366,7 @@ pub enum DiscardReason {
     InvalidReplayEventNoPayload,
     InvalidReplayEventPii,
     InvalidReplayRecordingEvent,
+    InvalidReplayVideoEvent,
 
     /// (Relay) Profiling related discard reasons
     Profiling(&'static str),
@@ -413,6 +414,7 @@ impl DiscardReason {
             DiscardReason::InvalidReplayEventNoPayload => "invalid_replay_no_payload",
             DiscardReason::InvalidReplayEventPii => "invalid_replay_pii_scrubber_failed",
             DiscardReason::InvalidReplayRecordingEvent => "invalid_replay_recording",
+            DiscardReason::InvalidReplayVideoEvent => "invalid_replay_video",
             DiscardReason::Profiling(reason) => reason,
             DiscardReason::InvalidSpan => "invalid_span",
         }

diff --git a/relay-server/src/services/processor.rs b/relay-server/src/services/processor.rs
@@ -226,7 +226,7 @@ impl ProcessingGroup {
         let replay_items = envelope.take_items_by(|item| {
             matches!(
                 item.ty(),
-                &ItemType::ReplayEvent | &ItemType::ReplayRecording
+                &ItemType::ReplayEvent | &ItemType::ReplayRecording | &ItemType::ReplayVideo
             )
         });
         if !replay_items.is_empty() {

diff --git a/relay-server/src/services/processor/event.rs b/relay-server/src/services/processor/event.rs
@@ -464,6 +464,7 @@ fn is_duplicate(item: &Item, processing_enabled: bool) -> bool {
         ItemType::Profile => false,
         ItemType::ReplayEvent => false,
         ItemType::ReplayRecording => false,
+        ItemType::ReplayVideo => false,
         ItemType::CheckIn => false,
         ItemType::Span => false,
         ItemType::OtelSpan => false,