feat(profiling): Filter out lonely samples

CHANGELOG.md

@@ -18,6 +18,7 @@
 - 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))
 - Batch metrics buckets into logical partitions before sending them as envelopes. ([#1440](https://github.com/getsentry/relay/pull/1440))
+- Filter single samples in cocoa profiles and events with no duration in Android profiles. ([#1445](https://github.com/getsentry/relay/pull/1445))
 - Add looser type requirements for the user.id field. ([#1443](https://github.com/getsentry/relay/pull/1443))
 
 **Features**:

relay-profiling/src/android.rs

@@ -1,4 +1,7 @@
-use android_trace_log::AndroidTraceLog;
+use std::collections::{HashMap, HashSet};
+
+use android_trace_log::chrono::{DateTime, Utc};
+use android_trace_log::{AndroidTraceLog, Clock, Time, Vm};
 use serde::{Deserialize, Serialize};
 
 use relay_general::protocol::EventId;
@@ -41,35 +44,71 @@ struct AndroidProfile {
     #[serde(default, skip_serializing)]
     sampled_profile: String,
 
-    #[serde(default)]
-    profile: Option<AndroidTraceLog>,
+    #[serde(default = "AndroidProfile::default")]
+    profile: AndroidTraceLog,
 }
 
 impl AndroidProfile {
+    fn default() -> AndroidTraceLog {
+        AndroidTraceLog {
+            data_file_overflow: Default::default(),
+            clock: Clock::Global,
+            elapsed_time: Default::default(),
+            total_method_calls: Default::default(),
+            clock_call_overhead: Default::default(),
+            vm: Vm::Dalvik,
+            start_time: Utc::now(),
+            pid: Default::default(),
+            gc_trace: Default::default(),
+            threads: Default::default(),
+            methods: Default::default(),
+            events: Default::default(),
+        }
+    }
+
     fn parse(&mut self) -> Result<(), ProfileError> {
         let profile_bytes = match base64::decode(&self.sampled_profile) {
             Ok(profile) => profile,
             Err(_) => return Err(ProfileError::InvalidBase64Value),
         };
         self.profile = match android_trace_log::parse(&profile_bytes) {
-            Ok(profile) => Some(profile),
+            Ok(profile) => profile,
             Err(_) => return Err(ProfileError::InvalidSampledProfile),
         };
         Ok(())
     }
+
+    /// Removes an event with a duration of 0
+    fn remove_events_with_no_duration(&mut self) {
+        let android_trace = &mut self.profile;
+        let events = &mut android_trace.events;
+        let clock = android_trace.clock;
+        let start_time = android_trace.start_time;
+        let mut timestamps_per_thread_id: HashMap<u16, HashSet<u64>> = HashMap::new();
+
+        for event in events.iter() {
+            let event_time = get_timestamp(clock, start_time, event.time);
+            timestamps_per_thread_id
+                .entry(event.thread_id)
+                .or_default()
+                .insert(event_time);
+        }
+
+        timestamps_per_thread_id.retain(|_, timestamps| timestamps.len() > 1);
+        events.retain(|event| timestamps_per_thread_id.contains_key(&event.thread_id));
+    }
 }
 
 pub fn parse_android_profile(payload: &[u8]) -> Result<Vec<u8>, ProfileError> {
     let mut profile: AndroidProfile =
         serde_json::from_slice(payload).map_err(ProfileError::InvalidJson)?;
 
-    if profile.sampled_profile.is_empty() {
-        return Ok(Vec::new());
+    if !profile.sampled_profile.is_empty() {
+        profile.parse()?;
+        profile.remove_events_with_no_duration();
     }
 
-    profile.parse()?;
-
-    if profile.profile.as_ref().unwrap().events.len() < 2 {
+    if profile.profile.events.is_empty() {
         return Err(ProfileError::NotEnoughSamples);
     }
 
@@ -79,15 +118,53 @@ pub fn parse_android_profile(payload: &[u8]) -> Result<Vec<u8>, ProfileError> {
     }
 }
 
+fn get_timestamp(clock: Clock, start_time: DateTime<Utc>, event_time: Time) -> u64 {
+    match (clock, event_time) {
+        (Clock::Global, Time::Global(time)) => {
+            time.as_nanos() as u64 - start_time.timestamp_nanos() as u64
+        }
+        (
+            Clock::Cpu,
+            Time::Monotonic {
+                cpu: Some(cpu),
+                wall: None,
+            },
+        ) => cpu.as_nanos() as u64,
+        (
+            Clock::Wall,
+            Time::Monotonic {
+                cpu: None,
+                wall: Some(wall),
+            },
+        ) => wall.as_nanos() as u64,
+        (
+            Clock::Dual,
+            Time::Monotonic {
+                cpu: Some(_),
+                wall: Some(wall),
+            },
+        ) => wall.as_nanos() as u64,
+        _ => todo!(),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
     #[test]
     fn test_roundtrip_android() {
-        let payload = include_bytes!("../tests/fixtures/profiles/android.json");
+        let payload = include_bytes!("../tests/fixtures/profiles/android/roundtrip.json");
         let data = parse_android_profile(payload);
         assert!(data.is_ok());
         assert!(parse_android_profile(&(data.unwrap())[..]).is_ok());
     }
+
+    #[test]
+    fn test_remove_invalid_events() {
+        let payload =
+            include_bytes!("../tests/fixtures/profiles/android/remove_invalid_events.json");
+        let data = parse_android_profile(payload);
+        assert!(data.is_err());
+    }
 }

relay-profiling/src/cocoa.rs

@@ -121,11 +121,33 @@ struct CocoaProfile {
     version_name: String,
 }
 
+impl CocoaProfile {
+    /// Removes a sample when it's the only sample on its thread
+    fn remove_single_samples_per_thread(&mut self) {
+        let mut sample_count_by_thread_id: HashMap<u64, u32> = HashMap::new();
+
+        for sample in self.sampled_profile.samples.iter() {
+            *sample_count_by_thread_id
+                .entry(sample.thread_id)
+                .or_default() += 1;
+        }
+
+        // Only keep data from threads with more than 1 sample so we can calculate a duration
+        sample_count_by_thread_id.retain(|_, count| *count > 1);
+
+        self.sampled_profile
+            .samples
+            .retain(|sample| sample_count_by_thread_id.contains_key(&sample.thread_id));
+    }
+}
+
 pub fn parse_cocoa_profile(payload: &[u8]) -> Result<Vec<u8>, ProfileError> {
-    let profile: CocoaProfile =
+    let mut profile: CocoaProfile =
         serde_json::from_slice(payload).map_err(ProfileError::InvalidJson)?;
 
-    if profile.sampled_profile.samples.len() < 2 {
+    profile.remove_single_samples_per_thread();
+
+    if profile.sampled_profile.samples.is_empty() {
         return Err(ProfileError::NotEnoughSamples);
     }
 
@@ -169,4 +191,102 @@ mod tests {
                 other: Map::new(),
             }))), annotated);
     }
+
+    fn generate_profile() -> CocoaProfile {
+        CocoaProfile {
+            debug_meta: DebugMeta { images: Vec::new() },
+            device_is_emulator: true,
+            device_locale: "en_US".to_string(),
+            device_manufacturer: "Apple".to_string(),
+            device_model: "iPhome11,3".to_string(),
+            device_os_build_number: "H3110".to_string(),
+            device_os_name: "iOS".to_string(),
+            device_os_version: "16.0".to_string(),
+            duration_ns: 1337,
+            environment: "testing".to_string(),
+            platform: "cocoa".to_string(),
+            profile_id: EventId::new(),
+            sampled_profile: SampledProfile {
+                thread_metadata: HashMap::new(),
+                samples: Vec::new(),
+                queue_metadata: HashMap::new(),
+            },
+            trace_id: EventId::new(),
+            transaction_id: EventId::new(),
+            transaction_name: "test".to_string(),
+            version_code: "9999".to_string(),
+            version_name: "1.0".to_string(),
+        }
+    }
+
+    #[test]
+    fn test_filter_samples() {
+        let mut profile = generate_profile();
+
+        profile.sampled_profile.samples.extend(vec![
+            Sample {
+                frames: Vec::new(),
+                queue_address: "0xdeadbeef".to_string(),
+                relative_timestamp_ns: 1,
+                thread_id: 1,
+            },
+            Sample {
+                frames: Vec::new(),
+                queue_address: "0xdeadbeef".to_string(),
+                relative_timestamp_ns: 1,
+                thread_id: 1,
+            },
+            Sample {
+                frames: Vec::new(),
+                queue_address: "0xdeadbeef".to_string(),
+                relative_timestamp_ns: 1,
+                thread_id: 2,
+            },
+            Sample {
+                frames: Vec::new(),
+                queue_address: "0xdeadbeef".to_string(),
+                relative_timestamp_ns: 1,
+                thread_id: 3,
+            },
+        ]);
+
+        profile.remove_single_samples_per_thread();
+
+        assert!(profile.sampled_profile.samples.len() == 2);
+    }
+
+    #[test]
+    fn test_parse_profile_with_all_samples_filtered() {
+        let mut profile = generate_profile();
+        profile.sampled_profile.samples.extend(vec![
+            Sample {
+                frames: Vec::new(),
+                queue_address: "0xdeadbeef".to_string(),
+                relative_timestamp_ns: 1,
+                thread_id: 1,
+            },
+            Sample {
+                frames: Vec::new(),
+                queue_address: "0xdeadbeef".to_string(),
+                relative_timestamp_ns: 1,
+                thread_id: 2,
+            },
+            Sample {
+                frames: Vec::new(),
+                queue_address: "0xdeadbeef".to_string(),
+                relative_timestamp_ns: 1,
+                thread_id: 3,
+            },
+            Sample {
+                frames: Vec::new(),
+                queue_address: "0xdeadbeef".to_string(),
+                relative_timestamp_ns: 1,
+                thread_id: 4,
+            },
+        ]);
+
+        let payload = serde_json::to_vec(&profile).unwrap();
+
+        assert!(parse_cocoa_profile(&payload[..]).is_err());
+    }
 }

relay-server/src/actors/processor.rs

@@ -852,6 +852,7 @@ impl EnvelopeProcessor {
     fn process_profiles(&self, state: &mut ProcessEnvelopeState) {
         let profiling_enabled = state.project_state.has_feature(Feature::Profiling);
         let context = &state.envelope_context;
+
         state.envelope.retain_items(|item| {
             match item.ty() {
                 ItemType::Profile => {