feat(replays): Add user-agent parsing to replays processor

CHANGELOG.md

@@ -16,6 +16,11 @@
 - Defer dropping of projects to a background thread to speed up project cache eviction. ([#1410](https://github.com/getsentry/relay/pull/1410))
 - Update store service to use generic Addr and minor changes to generic Addr. ([#1415](https://github.com/getsentry/relay/pull/1415))
 - Added new Register for the Services that is initialized later than the current. ([#1421](https://github.com/getsentry/relay/pull/1421))
+
+**Features**:
+
+- Make Redis connection pool configurable. ([#1418](https://github.com/getsentry/relay/pull/1418))
+- Add user-agent parsing to replays processor. ([#1420](https://github.com/getsentry/relay/pull/1420))
 - Improve the release name used when reporting data to Sentry to include both the version and exact build. ([#1428](https://github.com/getsentry/relay/pull/1428))
 
 ## 22.8.0

py/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
 - Add `transaction_info` to event payloads, including the transaction's source and internal original transaction name. ([#1330](https://github.com/getsentry/relay/pull/1330))
+- Add user-agent parsing to replays processor. ([#1420](https://github.com/getsentry/relay/pull/1420))
 
 ## 0.8.13
 

relay-general/src/user_agent.rs

@@ -1,4 +1,10 @@
 //! Utility functions for working with user agents.
+//!
+//! NOTICE:
+//!
+//! Adding user_agent parsing to your module will incur a latency penalty in the test suite.
+//! Because of this some integration tests may fail. To fix this, you will need to add a timeout
+//! to your consumer.
 
 use once_cell::sync::Lazy;
 use uaparser::{Parser, UserAgentParser};

relay-replays/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "relay-replays"
+authors = ["Sentry <oss@sentry.io>"]
+description = "Replays functionality for Relay"
+homepage = "https://getsentry.github.io/relay/"
+repository = "https://github.com/getsentry/relay"
+version = "22.7.0"
+edition = "2018"
+license-file = "../LICENSE"
+publish = false
+
+[dependencies]
+relay-common = { path = "../relay-common" }
+relay-general = { path = "../relay-general" }
+relay-log = { path = "../relay-log" }
+serde = { version = "1.0.114", features = ["derive"] }
+serde_json = "1.0.55"
+relay-filter = { path = "../relay-filter" }
+rand = "0.6.5"
+rand_pcg = "0.1.2"
+unicase = "2.6.0"
+
+[dev-dependencies]
+insta = { version = "1.1.0", features = ["ron"] }

relay-replays/src/lib.rs

@@ -0,0 +1,280 @@
+//! Replay processing and normalization module.
+//!
+//! Replays are multi-part values sent from Sentry integrations spanning arbitrary time-periods.
+//! They are ingested incrementally and are available for viewing after the first two segments
+//! (segment 0 and 1) have completed ingestion.
+//!
+//! # Protocol
+//!
+//! Relay is expecting a JSON object with some mandatory metadata.  However, environment and user
+//! metadata is usually sent in addition to the minimal payload.
+//!
+//! ```json
+//! {
+//!     "type": "replay_event",
+//!     "replay_id": "d2132d31b39445f1938d7e21b6bf0ec4",
+//!     "event_id": "63c5b0f895441a94340183c5f1e74cd4",
+//!     "segment_id": 0,
+//!     "timestamp": 1597976392.6542819,
+//!     "replay_start_timestamp": 1597976392.6542819,
+//!     "urls": ["https://sentry.io"],
+//!     "error_ids": ["d2132d31b39445f1938d7e21b6bf0ec4"],
+//!     "trace_ids": ["63c5b0f895441a94340183c5f1e74cd4"],
+//!     "requests": {
+//!         "headers": {"User-Agent": "Mozilla/5.0..."}
+//!     },
+//! }
+//! ```
+use std::collections::HashMap;
+use std::fmt::Write;
+use std::net::IpAddr;
+
+use serde::{Deserialize, Serialize};
+use serde_json::Error;
+
+use relay_general::user_agent::{parse_device, parse_os, parse_user_agent, Device};
+
+pub fn normalize_replay_event(
+    replay_bytes: &[u8],
+    detected_ip_address: Option<IpAddr>,
+) -> Result<Vec<u8>, Error> {
+    //! Validates and accepts a replay-event item as input and returns modified output.
+    //!
+    //! The output of this function will return a valid replay-event payload with the addition of
+    //! browser, device, and operating-system metadata.  The output will also contain a default
+    //! ip-address value for the user if none was provided.
+    let mut replay_input: ReplayInput = serde_json::from_slice(replay_bytes)?;
+
+    // Set user-agent metadata from requests object.
+    replay_input.set_user_agent_meta();
+
+    // Set user ip-address if needed.
+    match detected_ip_address {
+        Some(ip_address) => replay_input.set_user_ip_address(ip_address),
+        None => (),
+    }
+
+    serde_json::to_vec(&replay_input)
+}
+
+#[derive(Debug, Default, Deserialize, Serialize)]
+struct ReplayInput {
+    #[serde(rename = "type")]
+    ty: String,
+    replay_id: String,
+    event_id: String,
+    segment_id: u8,
+    timestamp: f64,
+    replay_start_timestamp: Option<f64>,
+    contexts: Option<Contexts>,
+    dist: Option<String>,
+    platform: Option<String>,
+    environment: Option<String>,
+    release: Option<String>,
+    tags: Option<HashMap<String, String>>,
+    sdk: Option<VersionedMeta>,
+    #[serde(default)]
+    urls: Vec<String>,
+    #[serde(default)]
+    error_ids: Vec<String>,
+    #[serde(default)]
+    trace_ids: Vec<String>,
+    #[serde(default)]
+    user: User,
+    #[serde(default)]
+    requests: Requests,
+}
+
+impl ReplayInput {
+    fn set_user_agent_meta(&mut self) {
+        let user_agent = &self.requests.headers.user_agent;
+
+        let device = parse_device(user_agent);
+
+        let ua = parse_user_agent(user_agent);
+        let browser_struct = VersionedMeta {
+            name: ua.family,
+            version: get_version(&ua.major, &ua.minor, &ua.patch),
+        };
+
+        let os = parse_os(user_agent);
+        let os_struct = VersionedMeta {
+            name: os.family,
+            version: get_version(&os.major, &os.minor, &os.patch),
+        };
+
+        self.contexts = Some(Contexts {
+            device: Some(device),
+            browser: Some(browser_struct),
+            os: Some(os_struct),
+        })
+    }
+
+    fn set_user_ip_address(&mut self, ip_address: IpAddr) {
+        self.user
+            .ip_address
+            .get_or_insert_with(|| ip_address.to_string());
+    }
+}
+
+fn get_version(
+    major: &Option<String>,
+    minor: &Option<String>,
+    patch: &Option<String>,
+) -> Option<String> {
+    let mut version = major.clone()?;
+
+    if let Some(minor) = minor {
+        write!(version, ".{}", minor).ok();
+        if let Some(patch) = patch {
+            write!(version, ".{}", patch).ok();
+        }
+    }
+
+    Some(version)
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+struct Contexts {
+    browser: Option<VersionedMeta>,
+    device: Option<Device>,
+    os: Option<VersionedMeta>,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+struct VersionedMeta {
+    name: String,
+    version: Option<String>,
+}
+
+#[derive(Debug, Default, Deserialize, Serialize)]
+struct User {
+    id: Option<String>,
+    username: Option<String>,
+    email: Option<String>,
+    ip_address: Option<String>,
+}
+
+#[derive(Debug, Default, Deserialize, Serialize)]
+#[serde(default)]
+struct Requests {
+    url: Option<String>,
+    headers: Headers,
+}
+
+#[derive(Debug, Default, Deserialize, Serialize)]
+#[serde(default)]
+struct Headers {
+    #[serde(rename = "User-Agent")]
+    user_agent: String,
+}
+
+#[cfg(test)]
+mod tests {
+    use std::net::{IpAddr, Ipv4Addr};
+
+    use crate::ReplayInput;
+
+    #[test]
+    fn test_set_ip_address() {
+        let ip_address = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
+
+        // IP-Address was not set.
+        let payload = include_bytes!("../tests/fixtures/replay.json");
+        let mut replay_input: ReplayInput = serde_json::from_slice(payload).unwrap();
+        replay_input.set_user_ip_address(ip_address);
+        assert!(*"192.168.11.12" == replay_input.user.ip_address.unwrap());
+    }
+
+    #[test]
+    fn test_set_ip_address_missing_user() {
+        let ip_address = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
+
+        // IP-Address set.
+        let payload = include_bytes!("../tests/fixtures/replay_missing_user.json");
+        let mut replay_input: ReplayInput = serde_json::from_slice(payload).unwrap();
+        replay_input.set_user_ip_address(ip_address);
+        assert!(*"127.0.0.1" == replay_input.user.ip_address.unwrap());
+    }
+
+    #[test]
+    fn test_set_ip_address_missing_user_ip_address() {
+        let ip_address = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
+
+        // IP-Address set.
+        let payload = include_bytes!("../tests/fixtures/replay_missing_user_ip_address.json");
+        let mut replay_input: ReplayInput = serde_json::from_slice(payload).unwrap();
+        replay_input.set_user_ip_address(ip_address);
+        assert!(*"127.0.0.1" == replay_input.user.ip_address.unwrap());
+    }
+
+    #[test]
+    fn test_set_user_agent_meta() {
+        let payload = include_bytes!("../tests/fixtures/replay.json");
+        let mut replay_input: ReplayInput = serde_json::from_slice(payload).unwrap();
+        replay_input.set_user_agent_meta();
+
+        match replay_input.contexts {
+            Some(contexts) => {
+                match contexts.browser {
+                    Some(browser) => {
+                        assert!(browser.name == *"Safari");
+                        assert!(browser.version.unwrap() == *"15.5");
+                    }
+                    None => unreachable!(),
+                }
+                match contexts.os {
+                    Some(os) => {
+                        assert!(os.name == *"Mac OS X");
+                        assert!(os.version.unwrap() == *"10.15.7");
+                    }
+                    None => unreachable!(),
+                }
+                match contexts.device {
+                    Some(device) => {
+                        assert!(device.family == *"Mac");
+                        assert!(device.brand.unwrap() == *"Apple");
+                        assert!(device.model.unwrap() == *"Mac");
+                    }
+                    None => unreachable!(),
+                }
+            }
+            None => unreachable!(),
+        }
+    }
+
+    #[test]
+    fn test_set_user_agent_meta_no_requests() {
+        let payload = include_bytes!("../tests/fixtures/replay_no_requests.json");
+        let mut replay_input: ReplayInput = serde_json::from_slice(payload).unwrap();
+        replay_input.set_user_agent_meta();
+
+        match replay_input.contexts {
+            Some(contexts) => {
+                match contexts.browser {
+                    Some(browser) => {
+                        assert!(browser.name == *"Other");
+                        assert!(browser.version.is_none());
+                    }
+                    None => unreachable!(),
+                }
+                match contexts.os {
+                    Some(os) => {
+                        assert!(os.name == *"Other");
+                        assert!(os.version.is_none());
+                    }
+                    None => unreachable!(),
+                }
+                match contexts.device {
+                    Some(device) => {
+                        assert!(device.family == *"Other");
+                        assert!(device.brand.is_none());
+                        assert!(device.model.is_none());
+                    }
+                    None => unreachable!(),
+                }
+            }
+            None => unreachable!(),
+        }
+    }
+}