feat(profiles): Emit "accepted" outcomes for profiles filtered by sampling

CHANGELOG.md

@@ -12,6 +12,7 @@
 - Copy transaction tags to the profile. ([#1982](https://github.com/getsentry/relay/pull/1982))
 - Lower default max compressed replay recording segment size to 10 MiB. ([#2031](https://github.com/getsentry/relay/pull/2031))
 - Add a data category for indexed profiles. ([#2051](https://github.com/getsentry/relay/pull/2051))
+- Differentiate between `Profile` and `ProfileIndexed` outcomes. ([#2054](https://github.com/getsentry/relay/pull/2054))
 
 ## 23.4.0
 

relay-server/src/actors/outcome.rs

@@ -148,12 +148,16 @@ impl FromMessage<Self> for TrackOutcome {
 /// Defines the possible outcomes from processing an event.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub enum Outcome {
-    // /// The event has been accepted and handled completely.
-    // ///
-    // /// This is never emitted by Relay as the event may be discarded by the processing pipeline
-    // /// after Relay. Only the `save_event` task in Sentry finally accepts an event.
-    // #[allow(dead_code)]
-    // Accepted,
+    /// The event has been accepted and handled completely.
+    ///
+    /// For events and most other types, this is never emitted by Relay as the event
+    /// may be discarded by the processing pipeline after Relay.
+    /// Only the `save_event` task in Sentry finally accepts an event.
+    ///
+    /// The only data type for which this outcome is emitted by Relay is [`DataCategory::Profile`].
+    /// (See [`crate::actors::processor::EnvelopeProcessor`])
+    #[cfg(feature = "processing")]
+    Accepted,
     /// The event has been filtered due to a configured filter.
     Filtered(FilterStatKey),
 
@@ -178,6 +182,8 @@ impl Outcome {
     /// Returns the raw numeric value of this outcome for the JSON and Kafka schema.
     fn to_outcome_id(&self) -> OutcomeId {
         match self {
+            #[cfg(feature = "processing")]
+            Outcome::Accepted => OutcomeId::ACCEPTED,
             Outcome::Filtered(_) | Outcome::FilteredSampling(_) => OutcomeId::FILTERED,
             Outcome::RateLimited(_) => OutcomeId::RATE_LIMITED,
             Outcome::Invalid(_) => OutcomeId::INVALID,
@@ -189,6 +195,8 @@ impl Outcome {
     /// Returns the `reason` code field of this outcome.
     fn to_reason(&self) -> Option<Cow<str>> {
         match self {
+            #[cfg(feature = "processing")]
+            Outcome::Accepted => None,
             Outcome::Invalid(discard_reason) => Some(Cow::Borrowed(discard_reason.name())),
             Outcome::Filtered(filter_key) => Some(Cow::Borrowed(filter_key.name())),
             Outcome::FilteredSampling(rule_ids) => Some(Cow::Owned(format!("Sampled:{rule_ids}"))),
@@ -221,6 +229,8 @@ impl Outcome {
 impl fmt::Display for Outcome {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
+            #[cfg(feature = "processing")]
+            Outcome::Accepted => write!(f, "accepted"),
             Outcome::Filtered(key) => write!(f, "filtered by {key}"),
             Outcome::FilteredSampling(rule_ids) => write!(f, "sampling rule {rule_ids}"),
             Outcome::RateLimited(None) => write!(f, "rate limited"),
@@ -787,14 +797,16 @@ impl OutcomeBroker {
     }
 
     #[cfg(feature = "processing")]
-    fn send_kafka_message(
+    fn send_kafka_message_inner(
         &self,
         producer: &KafkaOutcomesProducer,
         organization_id: u64,
         message: TrackRawOutcome,
     ) -> Result<(), OutcomeError> {
         relay_log::trace!("Tracking kafka outcome: {:?}", message);
 
+        send_outcome_metric(&message, "kafka");
+
         let payload = serde_json::to_string(&message).map_err(OutcomeError::SerializationError)?;
 
         // At the moment, we support outcomes with optional EventId.
@@ -824,11 +836,23 @@ impl OutcomeBroker {
         }
     }
 
+    #[cfg(feature = "processing")]
+    fn send_kafka_message(
+        &self,
+        producer: &KafkaOutcomesProducer,
+        organization_id: u64,
+        message: TrackRawOutcome,
+    ) -> Result<(), OutcomeError> {
+        for message in transform_outcome(message) {
+            self.send_kafka_message_inner(producer, organization_id, message)?;
+        }
+        Ok(())
+    }
+
     fn handle_track_outcome(&self, message: TrackOutcome, config: &Config) {
         match self {
             #[cfg(feature = "processing")]
             Self::Kafka(kafka_producer) => {
-                send_outcome_metric(&message, "kafka");
                 let organization_id = message.scoping.organization_id;
                 let raw_message = TrackRawOutcome::from_outcome(message, config);
                 if let Err(error) =
@@ -853,7 +877,6 @@ impl OutcomeBroker {
         match self {
             #[cfg(feature = "processing")]
             Self::Kafka(kafka_producer) => {
-                send_outcome_metric(&message, "kafka");
                 let sharding_id = message.org_id.unwrap_or_else(|| message.project_id.value());
                 if let Err(error) = self.send_kafka_message(kafka_producer, sharding_id, message) {
                     relay_log::error!("failed to produce outcome: {}", LogError(&error));
@@ -869,6 +892,56 @@ impl OutcomeBroker {
     }
 }
 
+/// Returns true if the outcome represents profiles dropped by dynamic sampling.
+#[cfg(feature = "processing")]
+fn is_sampled_profile(outcome: &TrackRawOutcome) -> bool {
+    // Older external Relays will still emit a `Profile` outcome.
+    // Newer Relays will emit a `ProfileIndexed` outcome.
+    (outcome.category == Some(DataCategory::Profile as u8)
+        || outcome.category == Some(DataCategory::ProfileIndexed as u8))
+        && outcome.outcome == OutcomeId::FILTERED
+        && outcome
+            .reason
+            .as_deref()
+            .map_or(false, |reason| reason.starts_with("Sampled:"))
+}
+
+/// Transform outcome into one or more derived outcome messages before sending it to kafka.
+#[cfg(feature = "processing")]
+fn transform_outcome(mut outcome: TrackRawOutcome) -> impl Iterator<Item = TrackRawOutcome> {
+    let mut extra = None;
+    if is_sampled_profile(&outcome) {
+        // Profiles that were dropped by dynamic sampling still count as "processed",
+        // so we emit the FILTERED outcome only for the "indexed" category instead.
+        outcome.category = Some(DataCategory::ProfileIndexed as u8);
+
+        // "processed" profiles are an abstract data category that does not represent actual data
+        // going through our pipeline. Instead, the number of accepted "processed" profiles is counted as
+        //
+        //     processed_profiles = indexed_profiles + sampled_profiles
+        //
+        // The "processed" outcome for indexed_profiles is generated in processing
+        // (see `EnvelopeProcessor::count_processed_profiles()`),
+        // but for profiles dropped by dynamic sampling, all we have is the FILTERED outcome,
+        // which we transform into an ACCEPTED outcome here.
+        //
+        // The reason for doing this transformation in the kafka producer is that it should apply
+        // to both `TrackOutcome` and `TrackRawOutcome`, and it should only happen _once_.
+        //
+        // In the future, we might actually extract metrics from profiles before dynamic sampling,
+        // like we do for transactions. At that point, this code should be removed, and we should
+        // enforce rate limits and emit outcomes based on the collect profile metric, as we do for
+        // transactions.
+        extra = Some(TrackRawOutcome {
+            outcome: OutcomeId::ACCEPTED,
+            reason: None,
+            category: Some(DataCategory::Profile as u8),
+            ..outcome.clone()
+        });
+    }
+    Some(outcome).into_iter().chain(extra)
+}
+
 #[derive(Debug)]
 enum ProducerInner {
     #[cfg(feature = "processing")]

relay-server/src/actors/processor.rs

@@ -1070,6 +1070,54 @@ impl EnvelopeProcessorService {
         })
     }
 
+    /// Count the number of profiles that are in the envelope and emit accepted outcome.
+    ///
+    /// "processed" profiles are an abstract data category that does not represent actual data
+    /// going through our pipeline. Instead, the number of accepted "processed" profiles is counted as
+    ///
+    /// ```text
+    /// processed_profiles = indexed_profiles + sampled_profiles
+    /// ```
+    ///
+    /// The "processed" outcome for sampled profiles is generated by the Kafka producer
+    /// (see `transform_outcome` in [`crate::actors::store`]), but for "indexed" profiles, we count
+    /// the corresponding number of processed profiles here.
+    ///
+    /// NOTE: Instead of emitting a [processed](`DataCategory::Profile`) outcome here,
+    /// we could also do it in sentry, in the same place where the [indexed](`DataCategory::ProfileIndexed`)
+    /// outcome is logged. We do it here to be consistent with profiles that are dropped by dynamic sampling,
+    /// which also count as "processed" even though they did not pass through the `process_profiles` step yet.
+    ///
+    ///
+    /// In the future, we might actually extract metrics from profiles before dynamic sampling,
+    /// like we do for transactions. At that point, this code should be removed, and we should
+    /// enforce rate limits and emit outcomes based on the collect profile metric, as we do for
+    /// transactions.
+    #[cfg(feature = "processing")]
+    fn count_processed_profiles(&self, state: &mut ProcessEnvelopeState) {
+        let profile_count: usize = state
+            .managed_envelope
+            .envelope()
+            .items()
+            .filter(|item| item.ty() == &ItemType::Profile)
+            .map(|item| item.quantity())
+            .sum();
+
+        if profile_count == 0 {
+            return;
+        }
+
+        self.outcome_aggregator.send(TrackOutcome {
+            timestamp: state.managed_envelope.received_at(),
+            scoping: state.managed_envelope.scoping(),
+            outcome: Outcome::Accepted,
+            event_id: None,
+            remote_addr: None,
+            category: DataCategory::Profile,
+            quantity: profile_count as u32, // truncates to `u32::MAX`
+        })
+    }
+
     /// Process profiles and set the profile ID in the profile context on the transaction if successful
     #[cfg(feature = "processing")]
     fn process_profiles(&self, state: &mut ProcessEnvelopeState) {
@@ -2245,6 +2293,9 @@ impl EnvelopeProcessorService {
         self.process_replays(state)?;
         self.filter_profiles(state);
 
+        // After filtering, we need to update the envelope summary:
+        state.managed_envelope.update();
+
         if state.creates_event() {
             // Some envelopes only create events in processing relays; for example, unreal events.
             // This makes it possible to get in this code block while not really having an event in
@@ -2276,6 +2327,11 @@ impl EnvelopeProcessorService {
 
         if_processing!({
             self.enforce_quotas(state)?;
+            // Any profile that reaches this point counts as "processed", regardless of whether
+            // they survive the actual `process_profiles` step. This is to be consistent with
+            // profiles that are dropped by dynamic sampling, which also count as "processed"
+            // even though they did not pass through the `process_profiles` step yet.
+            self.count_processed_profiles(state);
             // We need the event parsed in order to set the profile context on it
             self.process_profiles(state);
             self.process_check_ins(state);

relay-server/src/envelope.rs

@@ -550,7 +550,11 @@ impl Item {
             ItemType::Metrics | ItemType::MetricBuckets => None,
             ItemType::FormData => None,
             ItemType::UserReport => None,
-            ItemType::Profile => Some(DataCategory::Profile),
+            ItemType::Profile => Some(if indexed {
+                DataCategory::ProfileIndexed
+            } else {
+                DataCategory::Profile
+            }),
             ItemType::ReplayEvent | ItemType::ReplayRecording => Some(DataCategory::Replay),
             ItemType::ClientReport => None,
             ItemType::CheckIn => Some(DataCategory::Monitor),

relay-server/src/utils/managed_envelope.rs

@@ -312,7 +312,11 @@ impl ManagedEnvelope {
         if self.context.summary.profile_quantity > 0 {
             self.track_outcome(
                 outcome,
-                DataCategory::Profile,
+                if self.context.summary.event_metrics_extracted {
+                    DataCategory::ProfileIndexed
+                } else {
+                    DataCategory::Profile
+                },
                 self.context.summary.profile_quantity,
             );
         }

relay-server/src/utils/rate_limits.rs

@@ -510,8 +510,13 @@ where
 
             // It makes no sense to store profiles without transactions, so if the event
             // is rate limited, rate limit profiles as well.
+            let profile_category = if summary.event_metrics_extracted {
+                DataCategory::ProfileIndexed
+            } else {
+                DataCategory::Profile
+            };
             enforcement.profiles =
-                CategoryLimit::new(DataCategory::Profile, summary.profile_quantity, longest);
+                CategoryLimit::new(profile_category, summary.profile_quantity, longest);
 
             rate_limits.merge(event_limits);
         }
@@ -548,7 +553,11 @@ where
             let item_scoping = scoping.item(DataCategory::Profile);
             let profile_limits = (self.check)(item_scoping, summary.profile_quantity)?;
             enforcement.profiles = CategoryLimit::new(
-                DataCategory::Profile,
+                if summary.event_metrics_extracted {
+                    DataCategory::ProfileIndexed
+                } else {
+                    DataCategory::Profile
+                },
                 summary.profile_quantity,
                 profile_limits.longest(),
             );