ref(profiles): Counting profiles, V2 (#2163)

Instead of counting processed profiles in two different places (see
#2054), add a `has_profile` tag
to the transaction counter metric, and define

```
processed_profiles := count(processed_transactions) WHERE has_profile = true
```

## Changes

This PR contains the following changes:

* Revert PRs around counting profiles.
* Add `has_profile(s)` tag on `transaction.duration` metric.
#2165
* Update metrics billing consumer to emit `accepted` outcome for
profiles. getsentry/sentry#50047


## Caveats

With this PR, rate limits will be applied consistently if they are
issued for `DataCategory.Transaction`. However, if a rate limit (e.g.
spike protection) is issued _only_ for `DataCategory.Profile`, it will
not be enforced.

## Rollout Order

1. Merge and deploy getsentry/sentry#50047. The
billing metrics consumer will now listen for profiles, but not receive
any.
3. Deploy processing Relays. Profile counts will go down, because
PoP-Relays are not sending the `has_profile` tag yet.
4. Deploy PoP Relays. Profile counts should go back to normal.

## In case of rollback

1. First revert the sentry change
(getsentry/sentry#50047) and deploy it to stop
counting profiles.
2. Then revert the Relay change (this PR) and deploy to processing
Relays and PoPs.

ref: #2158

---------

Co-authored-by: Iker Barriocanal <[email protected]>

Author: jjbayer

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+**Bug Fixes**:
+
+- Make counting of total profiles consistent with total transactions. ([#2163](https://github.com/getsentry/relay/pull/2163))
+
 **Features**:
 
 - Add support for X-Vercel-Forwarded-For header. ([#2124](https://github.com/getsentry/relay/pull/2124))

relay-profiling/src/error.rs

@@ -26,6 +26,8 @@ pub enum ProfileError {
     MalformedSamples,
     #[error("exceed size limit")]
     ExceedSizeLimit,
+    #[error("too many profiles")]
+    TooManyProfiles,
     #[error("duration is too long")]
     DurationIsTooLong,
 }

relay-profiling/src/outcomes.rs

@@ -14,6 +14,7 @@ pub fn discard_reason(err: ProfileError) -> &'static str {
         ProfileError::NoTransactionAssociated => "profiling_no_transaction_associated",
         ProfileError::NotEnoughSamples => "profiling_not_enough_samples",
         ProfileError::PlatformNotSupported => "profiling_platform_not_supported",
+        ProfileError::TooManyProfiles => "profiling_too_many_profiles",
         ProfileError::DurationIsTooLong => "profiling_duration_is_too_long",
     }
 }

relay-server/src/actors/outcome.rs

@@ -147,16 +147,12 @@ 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.
-    ///
-    /// 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 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 filtered due to a configured filter.
     Filtered(FilterStatKey),
 
@@ -181,8 +177,6 @@ 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,
@@ -194,8 +188,6 @@ 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}"))),
@@ -228,8 +220,6 @@ 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"),
@@ -796,16 +786,14 @@ impl OutcomeBroker {
     }
 
     #[cfg(feature = "processing")]
-    fn send_kafka_message_inner(
+    fn send_kafka_message(
         &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.
@@ -835,23 +823,11 @@ 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) =
@@ -876,6 +852,7 @@ 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!(error = &error as &dyn Error, "failed to produce outcome");
@@ -891,54 +868,6 @@ impl OutcomeBroker {
     }
 }
 
-/// Returns true if the outcome represents profiles dropped by dynamic sampling.
-#[cfg(feature = "processing")]
-fn is_sampled_profile(outcome: &TrackRawOutcome) -> bool {
-    (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

@@ -13,6 +13,7 @@ use chrono::{DateTime, Duration as SignedDuration, Utc};
 use flate2::write::{GzEncoder, ZlibEncoder};
 use flate2::Compression;
 use once_cell::sync::OnceCell;
+use relay_profiling::ProfileError;
 use serde_json::Value as SerdeValue;
 use tokio::sync::Semaphore;
 
@@ -307,6 +308,9 @@ struct ProcessEnvelopeState {
 
     /// The managed envelope before processing.
     managed_envelope: ManagedEnvelope,
+
+    /// Whether there is a profiling item in the envelope.
+    has_profile: bool,
 }
 
 impl ProcessEnvelopeState {
@@ -1061,15 +1065,29 @@ impl EnvelopeProcessorService {
 
     /// Remove profiles from the envelope if they can not be parsed
     fn filter_profiles(&self, state: &mut ProcessEnvelopeState) {
+        let mut found_profile = false;
         state.managed_envelope.retain_items(|item| match item.ty() {
-            ItemType::Profile => match relay_profiling::parse_metadata(&item.payload()) {
-                Ok(_) => ItemAction::Keep,
-                Err(err) => ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
-                    relay_profiling::discard_reason(err),
-                ))),
-            },
+            ItemType::Profile => {
+                if !found_profile {
+                    match relay_profiling::parse_metadata(&item.payload()) {
+                        Ok(_) => {
+                            found_profile = true;
+                            ItemAction::Keep
+                        }
+                        Err(err) => ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
+                            relay_profiling::discard_reason(err),
+                        ))),
+                    }
+                } else {
+                    // We found a second profile, drop it.
+                    return ItemAction::Drop(Outcome::Invalid(DiscardReason::Profiling(
+                        relay_profiling::discard_reason(ProfileError::TooManyProfiles),
+                    )));
+                }
+            }
             _ => ItemAction::Keep,
         });
+        state.has_profile = found_profile;
     }
 
     /// Normalize monitor check-ins and remove invalid ones.
@@ -1097,61 +1115,6 @@ 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_mut()
-            .items_mut()
-            .filter(|item| item.ty() == &ItemType::Profile)
-            .map(|item| {
-                item.set_profile_counted_as_processed();
-                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`
-        });
-
-        // TODO: At this point, we should also ensure that the envelope summary gets recomputed.
-        // But recomputing the summary after extracting the event is currently problematic, because it
-        // sets the envelope type to `None`. This needs to be solved in a follow-up.
-    }
-
     /// 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) {
@@ -1396,6 +1359,7 @@ impl EnvelopeProcessorService {
             sampling_project_state,
             project_id,
             managed_envelope,
+            has_profile: false,
         })
     }
 
@@ -2129,6 +2093,7 @@ impl EnvelopeProcessorService {
                         event,
                         transaction_from_dsc,
                         &state.sampling_result,
+                        state.has_profile,
                         &mut state.extracted_metrics.project_metrics,
                         &mut state.extracted_metrics.sampling_metrics,
                     );
@@ -2429,9 +2394,6 @@ 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
@@ -2463,11 +2425,6 @@ 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);
@@ -2770,6 +2727,7 @@ impl Service for EnvelopeProcessorService {
 
 #[cfg(test)]
 mod tests {
+    use std::env;
     use std::str::FromStr;
 
     use chrono::{DateTime, TimeZone, Utc};
@@ -2992,6 +2950,7 @@ mod tests {
                     outcome_aggregator.clone(),
                     test_store.clone(),
                 ),
+                has_profile: false,
             };
 
             // TODO: This does not test if the sampling decision is actually applied. This should be