relay_profiling/sample/
v2.rs

1//! Sample Format V2
2//!
3//! This version of the sample format expects a collection of samples to be sent with no reference
4//! to the events collected while the profiler was running.
5//!
6//! We collect a profiler ID, meaning to be a random identifier for this specific instance of the
7//! profiler and not a persistent ID. It only needs to be valid from the start of the profiler to
8//! when it stops and will be useful to then group samples on the backend.
9//!
10//! Spans are expected to carry the profiler ID to know which samples are associated with them.
11//!
12use bytes::Bytes;
13use hashbrown::HashMap;
14use serde::{Deserialize, Serialize};
15use std::collections::{BTreeMap, HashSet};
16
17use relay_event_schema::protocol::EventId;
18use relay_protocol::FiniteF64;
19
20use crate::MAX_PROFILE_CHUNK_DURATION;
21use crate::error::ProfileError;
22use crate::measurements::ChunkMeasurement;
23use crate::sample::{DebugMeta, Frame, ThreadMetadata, Version};
24use crate::types::ClientSdk;
25
26const MAX_PROFILE_CHUNK_DURATION_SECS: f64 = MAX_PROFILE_CHUNK_DURATION.as_secs_f64();
27
28#[derive(Debug, Serialize, Deserialize)]
29pub struct ProfileMetadata {
30    /// Random UUID identifying a chunk
31    pub chunk_id: EventId,
32    /// Random UUID for each profiler session
33    pub profiler_id: EventId,
34
35    #[serde(default, skip_serializing_if = "DebugMeta::is_empty")]
36    pub debug_meta: DebugMeta,
37
38    #[serde(skip_serializing_if = "Option::is_none")]
39    pub environment: Option<String>,
40    pub platform: String,
41    #[serde(skip_serializing_if = "Option::is_none")]
42    pub content_type: Option<String>,
43    pub release: Option<String>,
44
45    pub client_sdk: ClientSdk,
46
47    /// Hard-coded string containing "2" to indicate the format version.
48    pub version: Version,
49}
50
51impl relay_protocol::Getter for ProfileMetadata {
52    fn get_value(&self, path: &str) -> Option<relay_protocol::Val<'_>> {
53        match path {
54            "release" => self.release.as_deref().map(|release| release.into()),
55            "platform" => Some(self.platform.as_str().into()),
56            _ => None,
57        }
58    }
59}
60
61/// Index of the stack in the `stacks` field of the profile.
62#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
63#[serde(transparent)]
64pub struct StackId(pub usize);
65
66#[derive(Debug, Serialize, Deserialize)]
67pub struct Sample {
68    /// Unix timestamp in seconds with millisecond precision when the sample
69    /// was captured.
70    pub timestamp: FiniteF64,
71    /// Index of the stack in the `stacks` field of the profile.
72    pub stack_id: StackId,
73    /// Thread or queue identifier
74    pub thread_id: String,
75}
76
77#[derive(Debug, Serialize, Deserialize)]
78pub struct ProfileChunk {
79    // `measurements` contains CPU/memory measurements we do during the capture of the chunk.
80    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
81    pub measurements: BTreeMap<String, ChunkMeasurement>,
82    /// This struct contains all the metadata related to the chunk but all fields are expected to
83    /// be at the top-level of the object.
84    #[serde(flatten)]
85    pub metadata: ProfileMetadata,
86    #[serde(default)]
87    pub profile: ProfileData,
88}
89
90impl ProfileChunk {
91    /// Parses a [`ProfileChunk`] from a JSON `payload`.
92    pub fn parse(payload: &[u8]) -> Result<Self, ProfileError> {
93        let d = &mut serde_json::Deserializer::from_slice(payload);
94        serde_path_to_error::deserialize(d).map_err(ProfileError::InvalidJson)
95    }
96
97    /// Normalizes the [`ProfileChunk`].
98    pub fn normalize(&mut self) -> Result<(), ProfileError> {
99        let platform = self.metadata.platform.as_str();
100        self.profile.normalize(platform)
101    }
102
103    /// Serializes the [`ProfileChunk`] into its JSON form.
104    pub fn serialize(&self) -> Result<Bytes, ProfileError> {
105        serde_json::to_vec(self)
106            .map(Bytes::from)
107            .map_err(|_| ProfileError::CannotSerializePayload)
108    }
109}
110
111impl crate::profile_chunk::ProfileChunk for ProfileChunk {
112    fn platform(&self) -> &str {
113        &self.metadata.platform
114    }
115
116    fn normalize(&mut self) -> Result<(), ProfileError> {
117        ProfileChunk::normalize(self)
118    }
119}
120
121impl relay_filter::Filterable for ProfileChunk {
122    fn release(&self) -> Option<&str> {
123        self.metadata.release.as_deref()
124    }
125}
126
127impl relay_protocol::Getter for ProfileChunk {
128    fn get_value(&self, path: &str) -> Option<relay_protocol::Val<'_>> {
129        self.metadata
130            .get_value(path.strip_prefix(crate::PROFIL_GETTER_PREFIX)?)
131    }
132}
133
134#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
135#[serde(transparent)]
136pub struct FrameId(pub usize);
137
138#[derive(Debug, Default, Serialize, Deserialize)]
139pub struct ProfileData {
140    /// `samples` contains the list of samples referencing a stack and thread identifier.
141    /// If 2 stack of frames captured at 2 different timestamps are identical, you're expected to
142    /// reference the same `stack_id`.
143    pub samples: Vec<Sample>,
144    /// `stacks` contains a list of stacks indicating the index of the frame in the `frames` field.
145    /// We do this to not have to repeat frames in different stacks.
146    pub stacks: Vec<Vec<FrameId>>,
147    /// `frames` contains a list of unique frames found in the profile.
148    pub frames: Vec<Frame>,
149
150    /// `thread_metadata` contains information about the thread or the queue. The identifier is a
151    /// string and can be any unique identifier for the thread or stack (an integer or an address
152    /// for example).
153    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
154    pub thread_metadata: BTreeMap<String, ThreadMetadata>,
155}
156
157impl ProfileData {
158    /// Returns `true` if the [`ProfileData`] does not contain any data.
159    pub fn is_empty(&self) -> bool {
160        let Self {
161            samples,
162            stacks,
163            frames,
164            thread_metadata,
165        } = self;
166
167        samples.is_empty() && stacks.is_empty() && frames.is_empty() && thread_metadata.is_empty()
168    }
169
170    /// Ensures valid profile chunk or returns an error.
171    ///
172    /// Mutates the profile chunk. Removes invalid samples and threads.
173    /// Throws an error if the profile chunk is malformed.
174    /// Removes extra metadata that are not referenced in the samples.
175    pub fn normalize(&mut self, platform: &str) -> Result<(), ProfileError> {
176        self.remove_single_samples_per_thread();
177
178        if self.samples.is_empty() {
179            return Err(ProfileError::NotEnoughSamples);
180        }
181
182        if !self.all_stacks_referenced_by_samples_exist() {
183            return Err(ProfileError::MalformedSamples);
184        }
185
186        if !self.all_frames_referenced_by_stacks_exist() {
187            return Err(ProfileError::MalformedStacks);
188        }
189
190        self.samples.sort_by_key(|s| s.timestamp);
191
192        if self.is_above_max_duration() {
193            return Err(ProfileError::DurationIsTooLong);
194        }
195
196        self.strip_pointer_authentication_code(platform);
197        self.remove_unreferenced_threads();
198
199        Ok(())
200    }
201
202    fn is_above_max_duration(&self) -> bool {
203        if self.samples.is_empty() {
204            return false;
205        }
206        let mut min = self.samples[0].timestamp;
207        let mut max = self.samples[0].timestamp;
208
209        for sample in self.samples.iter().skip(1) {
210            if sample.timestamp < min {
211                min = sample.timestamp
212            } else if sample.timestamp > max {
213                max = sample.timestamp
214            }
215        }
216
217        let duration = max.saturating_sub(min);
218        duration.to_f64() > MAX_PROFILE_CHUNK_DURATION_SECS
219    }
220
221    fn strip_pointer_authentication_code(&mut self, platform: &str) {
222        let addr = match platform {
223            // https://github.com/microsoft/plcrashreporter/blob/748087386cfc517936315c107f722b146b0ad1ab/Source/PLCrashAsyncThread_arm.c#L84
224            "cocoa" => 0x0000000FFFFFFFFF,
225            _ => return,
226        };
227        for frame in &mut self.frames {
228            frame.strip_pointer_authentication_code(addr);
229        }
230    }
231
232    /// Checks that all stacks referenced by the samples exist in the stacks.
233    fn all_stacks_referenced_by_samples_exist(&self) -> bool {
234        self.samples
235            .iter()
236            .all(|sample| self.stacks.get(sample.stack_id.0).is_some())
237    }
238
239    /// Checks that all frames referenced by the stacks exist in the frames.
240    fn all_frames_referenced_by_stacks_exist(&self) -> bool {
241        self.stacks.iter().all(|stack| {
242            stack
243                .iter()
244                .all(|frame_id| self.frames.get(frame_id.0).is_some())
245        })
246    }
247
248    fn remove_unreferenced_threads(&mut self) {
249        let thread_ids = self
250            .samples
251            .iter()
252            .map(|sample| sample.thread_id.clone())
253            .collect::<HashSet<_>>();
254        self.thread_metadata
255            .retain(|thread_id, _| thread_ids.contains(thread_id));
256    }
257
258    /// Removes a sample when it's the only non-idle sample on its thread
259    fn remove_single_samples_per_thread(&mut self) {
260        let mut sample_count_by_thread_id: hashbrown::HashMap<String, u32> = HashMap::new();
261
262        for s in &self.samples {
263            if let Some(stack) = self.stacks.get(s.stack_id.0) {
264                // We only count non-idle samples
265                if stack.is_empty() {
266                    continue;
267                }
268            } else {
269                continue;
270            }
271            *sample_count_by_thread_id
272                .entry(s.thread_id.to_owned())
273                .or_default() += 1;
274        }
275
276        sample_count_by_thread_id.retain(|_, count| *count > 1);
277        self.samples
278            .retain(|sample| sample_count_by_thread_id.contains_key(&sample.thread_id));
279    }
280}
281
282#[cfg(test)]
283mod tests {
284    use relay_protocol::FiniteF64;
285
286    use super::*;
287
288    #[test]
289    fn test_roundtrip() {
290        let first_payload = include_bytes!("../../tests/fixtures/sample/v2/valid.json");
291        let first_parse = ProfileChunk::parse(first_payload);
292        assert!(first_parse.is_ok(), "{first_parse:#?}");
293        let second_payload = serde_json::to_vec(&first_parse.unwrap()).unwrap();
294        let second_parse = ProfileChunk::parse(&second_payload[..]);
295        assert!(second_parse.is_ok(), "{second_parse:#?}");
296    }
297
298    #[test]
299    fn test_samples_are_sorted() {
300        let mut chunk = ProfileData {
301            samples: vec![
302                Sample {
303                    stack_id: StackId(0),
304                    thread_id: "1".into(),
305                    timestamp: FiniteF64::new(60.0).unwrap(),
306                },
307                Sample {
308                    stack_id: StackId(0),
309                    thread_id: "1".to_owned(),
310                    timestamp: FiniteF64::new(30.0).unwrap(),
311                },
312            ],
313            stacks: vec![vec![FrameId(0)]],
314            frames: vec![Default::default()],
315            ..Default::default()
316        };
317
318        assert!(chunk.normalize("python").is_ok());
319
320        let timestamps: Vec<FiniteF64> = chunk.samples.iter().map(|s| s.timestamp).collect();
321
322        assert_eq!(
323            timestamps,
324            vec![FiniteF64::new(30.0).unwrap(), FiniteF64::new(60.0).unwrap(),]
325        );
326    }
327
328    #[test]
329    fn test_is_above_max_duration() {
330        struct TestStruct {
331            name: String,
332            profile: ProfileData,
333            want: bool,
334        }
335
336        let test_cases = [
337            TestStruct {
338                name: "not above max duration".to_owned(),
339                profile: ProfileData {
340                    samples: vec![
341                        Sample {
342                            stack_id: StackId(0),
343                            thread_id: "1".into(),
344                            timestamp: FiniteF64::new(30.0).unwrap(),
345                        },
346                        Sample {
347                            stack_id: StackId(0),
348                            thread_id: "1".to_owned(),
349                            timestamp: FiniteF64::new(60.0).unwrap(),
350                        },
351                    ],
352                    stacks: vec![vec![FrameId(0)]],
353                    frames: vec![Default::default()],
354                    ..Default::default()
355                },
356                want: false,
357            },
358            TestStruct {
359                name: "above max duration".to_owned(),
360                profile: ProfileData {
361                    samples: vec![
362                        Sample {
363                            stack_id: StackId(0),
364                            thread_id: "1".into(),
365                            timestamp: FiniteF64::new(10.0).unwrap(),
366                        },
367                        Sample {
368                            stack_id: StackId(0),
369                            thread_id: "1".to_owned(),
370                            timestamp: FiniteF64::new(80.0).unwrap(),
371                        },
372                    ],
373                    stacks: vec![vec![FrameId(0)]],
374                    frames: vec![Default::default()],
375                    ..Default::default()
376                },
377                want: true,
378            },
379            TestStruct {
380                name: "unsorted samples not above max duration".to_owned(),
381                profile: ProfileData {
382                    samples: vec![
383                        Sample {
384                            stack_id: StackId(0),
385                            thread_id: "1".into(),
386                            timestamp: FiniteF64::new(50.0).unwrap(),
387                        },
388                        Sample {
389                            stack_id: StackId(0),
390                            thread_id: "1".to_owned(),
391                            timestamp: FiniteF64::new(20.0).unwrap(),
392                        },
393                    ],
394                    stacks: vec![vec![FrameId(0)]],
395                    frames: vec![Default::default()],
396                    ..Default::default()
397                },
398                want: false,
399            },
400        ];
401        for test in &test_cases {
402            assert_eq!(
403                test.profile.is_above_max_duration(),
404                test.want,
405                "test <{}> failed",
406                test.name
407            )
408        }
409    }
410
411    #[test]
412    fn test_single_samples_are_removed() {
413        let mut chunk = ProfileData {
414            samples: vec![
415                Sample {
416                    stack_id: StackId(1),
417                    thread_id: "1".into(),
418                    timestamp: FiniteF64::new(60.0).unwrap(),
419                },
420                Sample {
421                    stack_id: StackId(1),
422                    thread_id: "1".into(),
423                    timestamp: FiniteF64::new(60.0).unwrap(),
424                },
425                Sample {
426                    stack_id: StackId(0),
427                    thread_id: "1".into(),
428                    timestamp: FiniteF64::new(60.0).unwrap(),
429                },
430                Sample {
431                    stack_id: StackId(1),
432                    thread_id: "1".into(),
433                    timestamp: FiniteF64::new(60.0).unwrap(),
434                },
435                Sample {
436                    stack_id: StackId(0),
437                    thread_id: "2".to_owned(),
438                    timestamp: FiniteF64::new(30.0).unwrap(),
439                },
440                Sample {
441                    stack_id: StackId(1),
442                    thread_id: "2".into(),
443                    timestamp: FiniteF64::new(60.0).unwrap(),
444                },
445                Sample {
446                    stack_id: StackId(1),
447                    thread_id: "2".into(),
448                    timestamp: FiniteF64::new(60.0).unwrap(),
449                },
450                Sample {
451                    stack_id: StackId(0),
452                    thread_id: "3".to_owned(),
453                    timestamp: FiniteF64::new(30.0).unwrap(),
454                },
455                Sample {
456                    stack_id: StackId(0),
457                    thread_id: "3".to_owned(),
458                    timestamp: FiniteF64::new(30.0).unwrap(),
459                },
460                Sample {
461                    stack_id: StackId(1),
462                    thread_id: "3".into(),
463                    timestamp: FiniteF64::new(60.0).unwrap(),
464                },
465                Sample {
466                    stack_id: StackId(1),
467                    thread_id: "3".into(),
468                    timestamp: FiniteF64::new(60.0).unwrap(),
469                },
470            ],
471            stacks: vec![vec![FrameId(0)], vec![]],
472            frames: vec![Default::default()],
473            ..Default::default()
474        };
475
476        chunk.remove_single_samples_per_thread();
477
478        // Only 4 samples from thread_id 3 are retained.
479        assert_eq!(chunk.samples.len(), 4);
480    }
481}
relay_profiling/sample/v2.rs

relay_profiling/sample/
v2.rs
