fix(profiling): Classify profile chunks for rate limits and outcomes

Author: Dav1dde

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 - Add experimental playstation endpoint. ([#4555](https://github.com/getsentry/relay/pull/4555))
 
+**Bug Fixes**:
+
+- Separates profiles into backend and ui profiles. ([#4595](https://github.com/getsentry/relay/pull/4595))
+
 **Internal**:
 
 - Add ui chunk profiling data category. ([#4593](https://github.com/getsentry/relay/pull/4593))

Cargo.lock

@@ -14,6 +14,7 @@ workspace = true
 
 [dependencies]
 android_trace_log = { workspace = true, features = ["serde"] }
+bytes = { workspace = true }
 chrono = { workspace = true }
 data-encoding = { workspace = true }
 itertools = { workspace = true }

relay-profiling/Cargo.toml

@@ -42,6 +42,8 @@
 use std::error::Error;
 use std::net::IpAddr;
 use std::time::Duration;
+
+use bytes::Bytes;
 use url::Url;
 
 use relay_base_schema::project::ProjectId;
@@ -80,6 +82,12 @@ const MAX_PROFILE_CHUNK_DURATION: Duration = Duration::from_secs(66);
 /// Same format as event IDs.
 pub type ProfileId = EventId;
 
+#[derive(Debug, Clone, Copy)]
+pub enum ProfileType {
+    Backend,
+    Ui,
+}
+
 #[derive(Debug, Deserialize)]
 struct MinimalProfile {
     #[serde(alias = "profile_id", alias = "chunk_id")]
@@ -275,41 +283,74 @@ pub fn expand_profile(
     }
 }
 
-pub fn expand_profile_chunk(
-    payload: &[u8],
-    client_ip: Option<IpAddr>,
-    filter_settings: &ProjectFiltersConfig,
-    global_config: &GlobalConfig,
-) -> Result<Vec<u8>, ProfileError> {
-    let profile = match minimal_profile_from_json(payload) {
-        Ok(profile) => profile,
-        Err(err) => {
-            relay_log::warn!(
-                error = &err as &dyn Error,
-                from = "minimal",
-                "invalid profile chunk",
-            );
-            return Err(ProfileError::InvalidJson(err));
+/// Intermediate type for all processing on a profile chunk.
+pub struct ProfileChunk {
+    profile: MinimalProfile,
+    payload: Bytes,
+}
+
+impl ProfileChunk {
+    /// Parses a new [`Self`] from raw bytes.
+    pub fn new(payload: Bytes) -> Result<Self, ProfileError> {
+        match minimal_profile_from_json(&payload) {
+            Ok(profile) => Ok(Self { profile, payload }),
+            Err(err) => {
+                relay_log::debug!(
+                    error = &err as &dyn Error,
+                    from = "minimal",
+                    "invalid profile chunk",
+                );
+                Err(ProfileError::InvalidJson(err))
+            }
         }
-    };
+    }
 
-    if let Err(filter_stat_key) = relay_filter::should_filter(
-        &profile,
-        client_ip,
-        filter_settings,
-        global_config.filters(),
-    ) {
-        return Err(ProfileError::Filtered(filter_stat_key));
+    /// Returns the [`ProfileType`] this chunk belongs to.
+    ///
+    /// The profile type is currently determined based on the contained profile
+    /// platform. It determines the data category this profile chunk belongs to.
+    ///
+    /// This needs to be synchronized with the implementation in Sentry:
+    /// https://github.com/getsentry/sentry/blob/ed2e1c8bcd0d633e6f828fcfbeefbbdd98ef3dba/src/sentry/profiles/task.py#L995
+    pub fn profile_type(&self) -> ProfileType {
+        // Synchronized with:
+        //
+        match self.profile.platform.as_str() {
+            "cocoa" | "android" | "javascript" => ProfileType::Ui,
+            _ => ProfileType::Backend,
+        }
     }
 
-    match (profile.platform.as_str(), profile.version) {
-        ("android", _) => android::chunk::parse(payload),
-        (_, sample::Version::V2) => {
-            let mut profile = sample::v2::parse(payload)?;
-            profile.normalize()?;
-            serde_json::to_vec(&profile).map_err(|_| ProfileError::CannotSerializePayload)
+    /// Applies inbound filters to the profile chunk.
+    ///
+    /// The profile needs to be filtered (rejected) when this returns an error.
+    pub fn filter(
+        &self,
+        client_ip: Option<IpAddr>,
+        filter_settings: &ProjectFiltersConfig,
+        global_config: &GlobalConfig,
+    ) -> Result<(), ProfileError> {
+        relay_filter::should_filter(
+            &self.profile,
+            client_ip,
+            filter_settings,
+            global_config.filters(),
+        )
+        .map_err(ProfileError::Filtered)
+    }
+
+    /// Normalizes and 'expands' the profile chunk into its normalized form Sentry expects.
+    pub fn expand(&self) -> Result<Vec<u8>, ProfileError> {
+        match (self.profile.platform.as_str(), self.profile.version) {
+            ("android", _) => android::chunk::parse(&self.payload),
+            (_, sample::Version::V2) => {
+                let mut profile = sample::v2::parse(&self.payload)?;
+                profile.normalize()?;
+                Ok(serde_json::to_vec(&profile)
+                    .map_err(|_| ProfileError::CannotSerializePayload)?)
+            }
+            (_, _) => Err(ProfileError::PlatformNotSupported),
         }
-        (_, _) => Err(ProfileError::PlatformNotSupported),
     }
 }
 

relay-profiling/src/lib.rs

@@ -7,7 +7,7 @@ pub mod v1;
 pub mod v2;
 
 /// Possible values for the version field of the Sample Format.
-#[derive(Debug, Serialize, Deserialize, Clone, Default, PartialEq, Eq)]
+#[derive(Debug, Serialize, Deserialize, Copy, Clone, Default, PartialEq, Eq)]
 pub enum Version {
     #[default]
     Unknown,

relay-profiling/src/sample/mod.rs

@@ -31,6 +31,7 @@
 //! ```
 
 use relay_base_schema::project::ProjectKey;
+use relay_profiling::ProfileType;
 use std::borrow::Borrow;
 use std::collections::BTreeMap;
 use std::fmt;
@@ -596,6 +597,12 @@ pub struct ItemHeaders {
     #[serde(default, skip)]
     ingest_span_in_eap: bool,
 
+    /// Tracks whether the item is a backend or ui profile chunk.
+    ///
+    /// NOTE: This is internal-only and not exposed into the Envelope.
+    #[serde(default, skip)]
+    profile_type: Option<ProfileType>,
+
     /// Other attributes for forward compatibility.
     #[serde(flatten)]
     other: BTreeMap<String, Value>,
@@ -673,6 +680,7 @@ impl Item {
                 sampled: true,
                 fully_normalized: false,
                 ingest_span_in_eap: false,
+                profile_type: None,
             },
             payload: Bytes::new(),
         }
@@ -725,7 +733,11 @@ impl Item {
             ItemType::Span | ItemType::OtelSpan => smallvec![(DataCategory::Span, 1)],
             // NOTE: semantically wrong, but too expensive to parse.
             ItemType::OtelTracesData => smallvec![(DataCategory::Span, 1)],
-            ItemType::ProfileChunk => smallvec![(DataCategory::ProfileChunk, 1)], // TODO: should be seconds?
+            ItemType::ProfileChunk => match self.headers.profile_type {
+                Some(ProfileType::Backend) => smallvec![(DataCategory::ProfileChunk, 1)],
+                Some(ProfileType::Ui) => smallvec![(DataCategory::ProfileChunkUi, 1)],
+                None => smallvec![],
+            },
             ItemType::Unknown(_) => smallvec![],
         }
     }
@@ -890,6 +902,16 @@ impl Item {
         self.headers.ingest_span_in_eap = ingest_span_in_eap;
     }
 
+    /// Returns the associated profile type of a profile chunk.
+    pub fn profile_type(&self) -> Option<ProfileType> {
+        self.headers.profile_type
+    }
+
+    /// Set the profile type of the profile chunk.
+    pub fn set_profile_type(&mut self, profile_type: ProfileType) {
+        self.headers.profile_type = Some(profile_type);
+    }
+
     /// Gets the `sampled` flag.
     pub fn sampled(&self) -> bool {
         self.headers.sampled

relay-server/src/envelope.rs

@@ -1922,15 +1922,25 @@ impl EnvelopeProcessorService {
         &self,
         managed_envelope: &mut TypedEnvelope<ProfileChunkGroup>,
         project_info: Arc<ProjectInfo>,
+        _rate_limits: Arc<RateLimits>,
     ) -> Result<Option<ProcessingExtractedMetrics>, ProcessingError> {
         profile_chunk::filter(managed_envelope, project_info.clone());
         if_processing!(self.inner.config, {
             profile_chunk::process(
                 managed_envelope,
-                project_info,
+                project_info.clone(),
                 &self.inner.global_config.current(),
                 &self.inner.config,
             );
+
+            self.enforce_quotas(
+                managed_envelope,
+                Annotated::empty(),
+                &mut ProcessingExtractedMetrics::new(),
+                project_info,
+                _rate_limits,
+            )
+            .await?;
         });
 
         Ok(None)
@@ -1943,7 +1953,7 @@ impl EnvelopeProcessorService {
         config: Arc<Config>,
         project_id: ProjectId,
         project_info: Arc<ProjectInfo>,
-        #[allow(unused_variables)] rate_limits: Arc<RateLimits>,
+        _rate_limits: Arc<RateLimits>,
     ) -> Result<Option<ProcessingExtractedMetrics>, ProcessingError> {
         #[allow(unused_mut)]
         let mut extracted_metrics = ProcessingExtractedMetrics::new();
@@ -1964,7 +1974,7 @@ impl EnvelopeProcessorService {
                 Annotated::empty(),
                 &mut extracted_metrics,
                 project_info.clone(),
-                rate_limits,
+                _rate_limits,
             )
             .await?;
         });
@@ -2297,7 +2307,9 @@ impl EnvelopeProcessorService {
                 rate_limits,
                 reservoir_counters
             ),
-            ProcessingGroup::ProfileChunk => run!(process_profile_chunks, project_info),
+            ProcessingGroup::ProfileChunk => {
+                run!(process_profile_chunks, project_info, rate_limits)
+            }
             // Currently is not used.
             ProcessingGroup::Metrics => {
                 // In proxy mode we simply forward the metrics.
@@ -3650,7 +3662,7 @@ impl UpstreamRequest for SendMetricsRequest {
 
 /// Container for global and project level [`Quota`].
 #[cfg(feature = "processing")]
-#[derive(Copy, Clone)]
+#[derive(Copy, Clone, Debug)]
 struct CombinedQuotas<'a> {
     global_quotas: &'a [Quota],
     project_quotas: &'a [Quota],

relay-server/src/services/processor.rs

@@ -1,6 +1,7 @@
 //! Profile chunks processor code.
 
 use relay_dynamic_config::Feature;
+use relay_profiling::ProfileError;
 use std::sync::Arc;
 
 use crate::envelope::ItemType;
@@ -40,44 +41,55 @@ pub fn process(
 ) {
     let client_ip = managed_envelope.envelope().meta().client_addr();
     let filter_settings = &project_info.config.filter_settings;
+
     let continuous_profiling_enabled =
         if project_info.has_feature(Feature::ContinuousProfilingBetaIngest) {
             project_info.has_feature(Feature::ContinuousProfilingBeta)
         } else {
             project_info.has_feature(Feature::ContinuousProfiling)
         };
+
     managed_envelope.retain_items(|item| match item.ty() {
         ItemType::ProfileChunk => {
             if !continuous_profiling_enabled {
                 return ItemAction::DropSilently;
             }
 
-            match relay_profiling::expand_profile_chunk(
-                &item.payload(),
-                client_ip,
-                filter_settings,
-                global_config,
-            ) {
-                Ok(payload) => {
-                    if payload.len() <= config.max_profile_size() {
-                        item.set_payload(ContentType::Json, payload);
-                        ItemAction::Keep
-                    } else {
-                        ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
-                            relay_profiling::discard_reason(
-                                relay_profiling::ProfileError::ExceedSizeLimit,
-                            ),
-                        )))
-                    }
-                }
-                Err(relay_profiling::ProfileError::Filtered(filter_stat_key)) => {
-                    ItemAction::Drop(Outcome::Filtered(filter_stat_key))
-                }
-                Err(err) => ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
-                    relay_profiling::discard_reason(err),
-                ))),
+            let chunk = match relay_profiling::ProfileChunk::new(item.payload()) {
+                Ok(chunk) => chunk,
+                Err(err) => return error_to_action(err),
+            };
+            // Important: set the profile type to get outcomes in the correct category.
+            item.set_profile_type(chunk.profile_type());
+
+            if let Err(err) = chunk.filter(client_ip, filter_settings, global_config) {
+                return error_to_action(err);
+            }
+
+            let payload = match chunk.expand() {
+                Ok(expanded) => expanded,
+                Err(err) => return error_to_action(err),
+            };
+
+            if payload.len() > config.max_profile_size() {
+                return error_to_action(relay_profiling::ProfileError::ExceedSizeLimit);
             }
+
+            item.set_payload(ContentType::Json, payload);
+            ItemAction::Keep
         }
         _ => ItemAction::Keep,
     });
 }
+
+#[cfg(feature = "processing")]
+fn error_to_action(err: ProfileError) -> ItemAction {
+    match err {
+        ProfileError::Filtered(filter_stat_key) => {
+            ItemAction::Drop(Outcome::Filtered(filter_stat_key))
+        }
+        err => ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
+            relay_profiling::discard_reason(err),
+        ))),
+    }
+}

relay-server/src/services/processor/profile_chunk.rs

@@ -470,6 +470,14 @@ impl ManagedEnvelope {
             );
         }
 
+        if self.context.summary.profile_chunk_ui_quantity > 0 {
+            self.track_outcome(
+                outcome.clone(),
+                DataCategory::ProfileChunkUi,
+                self.context.summary.profile_chunk_ui_quantity,
+            );
+        }
+
         self.finish(RelayCounters::EnvelopeRejected, handling);
     }