Forward profiles of non-sampled transactions

CHANGELOG.md

@@ -14,6 +14,7 @@
 - Copy event measurements to span & normalize span measurements. ([#2953](https://github.com/getsentry/relay/pull/2953))
 - Add `allow_negative` to `BuiltinMeasurementKey`. Filter out negative BuiltinMeasurements if `allow_negative` is false. ([#2982](https://github.com/getsentry/relay/pull/2982))
 - Add possiblity to block metrics or their tags with glob-patterns. ([#2954](https://github.com/getsentry/relay/pull/2954), [#2973](https://github.com/getsentry/relay/pull/2973))
+- Forward profiles of non-sampled transactions ([#2940](https://github.com/getsentry/relay/pull/2940))
 
 **Bug Fixes**:
 

relay-dynamic-config/src/feature.rs

@@ -40,6 +40,11 @@ pub enum Feature {
     /// Enable the Relay cardinality limiter.
     #[serde(rename = "organizations:relay-cardinality-limiter")]
     CardinalityLimiter,
+    /// Enable processing and extracting data from profiles that would normally be dropped by dynamic sampling.
+    ///
+    /// This is required for [slowest function aggregation](https://github.com/getsentry/snuba/blob/b5311b404a6bd73a9e1997a46d38e7df88e5f391/snuba/snuba_migrations/functions/0001_functions.py#L209-L256). The profile payload will be dropped on the sentry side.
+    #[serde(rename = "projects:profiling-ingest-unsampled-profiles")]
+    IngestUnsampledProfiles,
 
     /// Deprecated, still forwarded for older downstream Relays.
     #[serde(rename = "organizations:transaction-name-mark-scrubbed-as-sanitized")]

relay-dynamic-config/src/global.rs

@@ -45,11 +45,19 @@ impl GlobalConfig {
 #[serde(default, rename_all = "camelCase")]
 #[cfg_attr(feature = "jsonschema", derive(JsonSchema))]
 pub struct Options {
-    // Example:
-    // ```rs
-    // #[serde(default, rename = "relay.some-option.name")]
-    // pub some_option: Vec<u32>,
-    // ```
+    /// List of platform names for which we allow using unsampled profiles for the purpose
+    /// of improving profile (function) metrics
+    #[serde(rename = "profiling.profile_metrics.unsampled_profiles.platforms")]
+    pub profile_metrics_allowed_platforms: Vec<String>,
+
+    /// Sample rate for tuning the amount of unsampled profiles that we "let through"
+    #[serde(rename = "profiling.profile_metrics.unsampled_profiles.sample_rate")]
+    pub profile_metrics_sample_rate: f32,
+
+    /// Kill switch for shutting down profile metrics
+    #[serde(rename = "profiling.profile_metrics.unsampled_profiles.enabled")]
+    pub unsampled_profiles_enabled: bool,
+
     /// All other unknown options.
     #[serde(flatten)]
     other: HashMap<String, Value>,
@@ -75,6 +83,7 @@ mod tests {
             }),
             options: Some(Options {
                 other: HashMap::from([("relay.unknown".to_owned(), Value::Bool(true))]),
+                ..Default::default()
             }),
         };
 

relay-server/Cargo.toml

@@ -62,6 +62,7 @@ mime_guess = { version = "2.0.4", optional = true }
 minidump = { version = "0.15.2", optional = true }
 multer = "2.0.4"
 once_cell = { workspace = true }
+rand = { workspace = true }
 regex = { workspace = true }
 relay-auth = { path = "../relay-auth" }
 relay-aws-extension = { path = "../relay-aws-extension" }

relay-server/src/envelope.rs

@@ -517,6 +517,14 @@ pub struct ItemHeaders {
     #[serde(default, skip_serializing_if = "is_false")]
     metrics_extracted: bool,
 
+    /// `false` if the sampling decision is "drop".
+    ///
+    /// In the most common use case, the item is dropped when the sampling decision is "drop".
+    /// For profiles with the feature enabled, however, we keep all profile items and mark the ones
+    /// for which the transaction was dropped as `sampled: false`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    sampled: Option<bool>,
+
     /// Other attributes for forward compatibility.
     #[serde(flatten)]
     other: BTreeMap<String, Value>,
@@ -564,6 +572,7 @@ impl Item {
                 sample_rates: None,
                 other: BTreeMap::new(),
                 metrics_extracted: false,
+                sampled: None,
             },
             payload: Bytes::new(),
         }
@@ -739,6 +748,16 @@ impl Item {
         self.headers.metrics_extracted = metrics_extracted;
     }
 
+    /// Gets the `sampled` flag.
+    pub fn sampled(&self) -> Option<bool> {
+        self.headers.sampled
+    }
+
+    /// Sets the `sampled` flag.
+    pub fn set_sampled(&mut self, sampled: bool) {
+        self.headers.sampled = Some(sampled);
+    }
+
     /// Returns the specified header value, if present.
     pub fn get_header<K>(&self, name: &K) -> Option<&Value>
     where

relay-server/src/services/processor.rs

@@ -1193,7 +1193,11 @@ impl EnvelopeProcessorService {
             self.extract_metrics(state)?;
         }
 
-        dynamic_sampling::sample_envelope_items(state);
+        dynamic_sampling::sample_envelope_items(
+            state,
+            &self.inner.config,
+            &self.inner.global_config.current(),
+        );
 
         if_processing!(self.inner.config, {
             event::store(state, &self.inner.config, self.inner.geoip_lookup.as_ref())?;
@@ -1213,11 +1217,8 @@ impl EnvelopeProcessorService {
         Ok(())
     }
 
-    /// Processes standalone attachments.
-    fn process_standalone_attachments(
-        &self,
-        state: &mut ProcessEnvelopeState,
-    ) -> Result<(), ProcessingError> {
+    /// Processes standalone items that require an event ID, but do not have an event on the same envelope.
+    fn process_standalone(&self, state: &mut ProcessEnvelopeState) -> Result<(), ProcessingError> {
         profile::filter(state);
 
         if_processing!(self.inner.config, {
@@ -1337,7 +1338,11 @@ impl EnvelopeProcessorService {
                 self.extract_metrics(state)?;
             }
 
-            dynamic_sampling::sample_envelope_items(state);
+            dynamic_sampling::sample_envelope_items(
+                state,
+                &self.inner.config,
+                &self.inner.global_config.current(),
+            );
 
             if_processing!(self.inner.config, {
                 event::store(state, &self.inner.config, self.inner.geoip_lookup.as_ref())?;
@@ -1379,7 +1384,7 @@ impl EnvelopeProcessorService {
             ProcessingGroup::Error => self.process_errors(state)?,
             ProcessingGroup::Transaction => self.process_transactions(state)?,
             ProcessingGroup::Session => self.process_sessions(state)?,
-            ProcessingGroup::Standalone => self.process_standalone_attachments(state)?,
+            ProcessingGroup::Standalone => self.process_standalone(state)?,
             ProcessingGroup::ClientReport => self.process_client_reports(state)?,
             ProcessingGroup::Replay => self.process_replays(state)?,
             ProcessingGroup::CheckIn => self.process_checkins(state)?,

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

@@ -5,16 +5,17 @@ use std::ops::ControlFlow;
 use chrono::Utc;
 use relay_base_schema::events::EventType;
 use relay_config::Config;
-use relay_dynamic_config::ErrorBoundary;
+use relay_dynamic_config::{ErrorBoundary, Feature, GlobalConfig};
 use relay_event_schema::protocol::{Contexts, Event, TraceContext};
 use relay_protocol::{Annotated, Empty};
 use relay_sampling::config::{RuleType, SamplingMode};
 use relay_sampling::evaluation::{ReservoirEvaluator, SamplingEvaluator};
 use relay_sampling::{DynamicSamplingContext, SamplingConfig};
 
+use crate::envelope::ItemType;
 use crate::services::outcome::Outcome;
-use crate::services::processor::ProcessEnvelopeState;
-use crate::utils::{self, SamplingResult};
+use crate::services::processor::{profile, ProcessEnvelopeState};
+use crate::utils::{self, ItemAction, SamplingResult};
 
 /// Ensures there is a valid dynamic sampling context and corresponding project state.
 ///
@@ -96,17 +97,31 @@ pub fn run(state: &mut ProcessEnvelopeState, config: &Config) {
 }
 
 /// Apply the dynamic sampling decision from `compute_sampling_decision`.
-pub fn sample_envelope_items(state: &mut ProcessEnvelopeState) {
+pub fn sample_envelope_items(
+    state: &mut ProcessEnvelopeState,
+    config: &Config,
+    global_config: &GlobalConfig,
+) {
     if let SamplingResult::Match(sampling_match) = std::mem::take(&mut state.sampling_result) {
         // We assume that sampling is only supposed to work on transactions.
-        if state.event_type() == Some(EventType::Transaction) && sampling_match.should_drop() {
-            let matched_rules = sampling_match.into_matched_rules();
+        let Some(event) = state.event.value() else {
+            return;
+        };
+        if event.ty.value() == Some(&EventType::Transaction) && sampling_match.should_drop() {
+            let unsampled_profiles_enabled = forward_unsampled_profiles(state, global_config);
 
+            let matched_rules = sampling_match.into_matched_rules();
             let outcome = Outcome::FilteredSampling(matched_rules.clone());
-            state
-                .managed_envelope
-                .retain_items(|_| utils::ItemAction::Drop(outcome.clone()));
-
+            state.managed_envelope.retain_items(|item| {
+                if unsampled_profiles_enabled && item.ty() == &ItemType::Profile {
+                    item.set_sampled(false);
+                    // Transfer metadata to the profile before the transaction gets dropped:
+                    let (_, item_action) = profile::expand_profile(item, event, config);
+                    item_action
+                } else {
+                    ItemAction::Drop(outcome.clone())
+                }
+            });
             // The event is no longer in the envelope, so we need to handle it separately:
             state.reject_event(outcome);
         }
@@ -226,6 +241,32 @@ pub fn tag_error_with_sampling_decision(state: &mut ProcessEnvelopeState, config
     }
 }
 
+/// Determines whether profiles that would otherwise be dropped by dynamic sampling should be kept.
+fn forward_unsampled_profiles(state: &ProcessEnvelopeState, global_config: &GlobalConfig) -> bool {
+    let Some(global_options) = &global_config.options else {
+        return false;
+    };
+
+    if !global_options.unsampled_profiles_enabled {
+        return false;
+    }
+
+    let event_platform = state
+        .event
+        .value()
+        .and_then(|e| e.platform.as_str())
+        .unwrap_or("");
+
+    state
+        .project_state
+        .has_feature(Feature::IngestUnsampledProfiles)
+        && global_options
+            .profile_metrics_allowed_platforms
+            .iter()
+            .any(|s| s == event_platform)
+        && rand::random::<f32>() < global_options.profile_metrics_sample_rate
+}
+
 #[cfg(test)]
 mod tests {
 

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

@@ -1,14 +1,15 @@
 //! Profiles related processor code.
 
 #[cfg(feature = "processing")]
-use {crate::envelope::ContentType, relay_config::Config, relay_dynamic_config::Feature};
+use relay_dynamic_config::Feature;
 
 use relay_base_schema::events::EventType;
-use relay_event_schema::protocol::{Contexts, ProfileContext};
-use relay_profiling::ProfileError;
+use relay_config::Config;
+use relay_event_schema::protocol::{Contexts, Event, ProfileContext};
+use relay_profiling::{ProfileError, ProfileId};
 use relay_protocol::Annotated;
 
-use crate::envelope::ItemType;
+use crate::envelope::{ContentType, Item, ItemType};
 use crate::services::outcome::{DiscardReason, Outcome};
 use crate::services::processor::ProcessEnvelopeState;
 use crate::utils::ItemAction;
@@ -23,10 +24,14 @@ pub fn filter(state: &mut ProcessEnvelopeState) {
         .count();
     let mut profile_id = None;
     state.managed_envelope.retain_items(|item| match item.ty() {
-        // Drop profile without a transaction in the same envelope.
-        ItemType::Profile if transaction_count == 0 => ItemAction::DropSilently,
         // First profile found in the envelope, we'll keep it if metadata are valid.
         ItemType::Profile if profile_id.is_none() => {
+            // Drop profile without a transaction in the same envelope.
+            let profile_allowed = transaction_count > 0 || matches!(item.sampled(), Some(false));
+            if !profile_allowed {
+                return ItemAction::DropSilently;
+            }
+
             match relay_profiling::parse_metadata(&item.payload(), state.project_id) {
                 Ok(id) => {
                     profile_id = Some(id);
@@ -77,24 +82,9 @@ pub fn process(state: &mut ProcessEnvelopeState, config: &Config) {
             let Some(event) = state.event.value() else {
                 return ItemAction::DropSilently;
             };
-            match relay_profiling::expand_profile(&item.payload(), event) {
-                Ok((profile_id, payload)) => {
-                    if payload.len() <= config.max_profile_size() {
-                        found_profile_id = Some(profile_id);
-                        item.set_payload(ContentType::Json, payload);
-                        ItemAction::Keep
-                    } else {
-                        ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
-                            relay_profiling::discard_reason(
-                                relay_profiling::ProfileError::ExceedSizeLimit,
-                            ),
-                        )))
-                    }
-                }
-                Err(err) => ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
-                    relay_profiling::discard_reason(err),
-                ))),
-            }
+            let (profile_id, action) = expand_profile(item, event, config);
+            found_profile_id = profile_id;
+            action
         }
         _ => ItemAction::Keep,
     });
@@ -110,6 +100,32 @@ pub fn process(state: &mut ProcessEnvelopeState, config: &Config) {
     }
 }
 
+/// Transfers transaction metadata to profile and check its size.
+pub fn expand_profile(
+    item: &mut Item,
+    event: &Event,
+    config: &Config,
+) -> (Option<ProfileId>, ItemAction) {
+    let mut profile_id = None;
+    let item_action = match relay_profiling::expand_profile(&item.payload(), event) {
+        Ok((id, payload)) => {
+            profile_id = Some(id);
+            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(err) => ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
+            relay_profiling::discard_reason(err),
+        ))),
+    };
+    (profile_id, item_action)
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;