feat(server): Add rule id to outcomes coming from transaction sampling

CHANGELOG.md

@@ -7,6 +7,7 @@
 - Emit the `quantity` field for outcomes of events. This field describes the total size in bytes for attachments or the event count for all other categories. A separate outcome is emitted for attachments in a rejected envelope, if any, in addition to the event outcome. ([#942](https://github.com/getsentry/relay/pull/942))
 - Add experimental metrics ingestion without bucketing or pre-aggregation. ([#948](https://github.com/getsentry/relay/pull/948))
 - Disallow empty metric names, require alphabetic start. ([#952](https://github.com/getsentry/relay/pull/952))
+- Add rule id to outcomes coming from transaction sampling. ([#953](https://github.com/getsentry/relay/pull/953))
 
 ## 21.3.0
 

relay-sampling/src/lib.rs

@@ -30,6 +30,17 @@ pub enum RuleType {
     Error,
 }
 
+/// The result of a sampling operation returned by [`TraceContext::should_keep`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SamplingResult {
+    /// Keep the event.
+    Keep,
+    /// Drop the event, due to the rule with provided identifier.
+    Drop(RuleId),
+    /// No decision can be made.
+    NoDecision,
+}
+
 /// A condition that checks the values using the equality operator.
 ///
 /// For string values it supports case-insensitive comparison.
@@ -492,14 +503,26 @@ pub struct TraceContext {
 }
 
 impl TraceContext {
-    /// Returns the decision of whether to sample or not a trace based on the configuration rules.
+    /// Returns whether a trace should be retained based on sampling rules.
     ///
-    /// If None then a decision can't be made either because of an invalid of missing trace context or
-    /// because no applicable sampling rule could be found.
-    pub fn should_sample(&self, ip_addr: Option<IpAddr>, config: &SamplingConfig) -> Option<bool> {
-        let rule = get_matching_trace_rule(config, self, ip_addr, RuleType::Trace)?;
-        let rate = pseudo_random_from_uuid(self.trace_id)?;
-        Some(rate < rule.sample_rate)
+    /// If [`SamplingResult::NoDecision`] is returned, then no rule matched this trace. In this
+    /// case, the caller may decide whether to keep the trace or not. The same is returned if the
+    /// configuration is invalid.
+    pub fn should_keep(&self, ip_addr: Option<IpAddr>, config: &SamplingConfig) -> SamplingResult {
+        if let Some(rule) = get_matching_trace_rule(config, self, ip_addr, RuleType::Trace) {
+            let rate = match pseudo_random_from_uuid(self.trace_id) {
+                None => return SamplingResult::NoDecision,
+                Some(rate) => rate,
+            };
+
+            if rate < rule.sample_rate {
+                SamplingResult::Keep
+            } else {
+                SamplingResult::Drop(rule.id)
+            }
+        } else {
+            SamplingResult::NoDecision
+        }
     }
 }
 
@@ -552,15 +575,18 @@ pub fn pseudo_random_from_uuid(id: Uuid) -> Option<f64> {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use std::net::{IpAddr as NetIpAddr, Ipv4Addr};
+    use std::str::FromStr;
+
     use insta::assert_ron_snapshot;
+
     use relay_general::protocol::{
         Csp, Exception, Headers, IpAddr, JsonLenientString, LenientString, LogEntry, PairList,
         Request, User, Values,
     };
     use relay_general::types::Annotated;
-    use std::net::{IpAddr as NetIpAddr, Ipv4Addr};
-    use std::str::FromStr;
+
+    use super::*;
 
     fn eq(name: &str, value: &[&str], ignore_case: bool) -> RuleCondition {
         RuleCondition::Eq(EqCondition {

relay-server/src/actors/events.rs

@@ -9,7 +9,6 @@ use actix::prelude::*;
 use chrono::{DateTime, Duration as SignedDuration, Utc};
 use failure::Fail;
 use futures::prelude::*;
-use relay_metrics::Metric;
 use serde_json::Value as SerdeValue;
 
 use relay_common::{clone, metric, ProjectId, UnixTimestamp};
@@ -23,9 +22,10 @@ use relay_general::protocol::{
 use relay_general::store::ClockDriftProcessor;
 use relay_general::types::{Annotated, Array, FromValue, Object, ProcessingAction, Value};
 use relay_log::LogError;
+use relay_metrics::Metric;
 use relay_quotas::{DataCategory, RateLimits};
 use relay_redis::RedisPool;
-use relay_sampling::RuleId;
+use relay_sampling::{RuleId, SamplingResult};
 
 use crate::actors::outcome::{DiscardReason, Outcome, OutcomeProducer, TrackOutcome};
 use crate::actors::project::{
@@ -37,9 +37,7 @@ use crate::envelope::{self, AttachmentType, ContentType, Envelope, Item, ItemTyp
 use crate::http::{HttpError, RequestBuilder};
 use crate::metrics::{RelayCounters, RelayHistograms, RelaySets, RelayTimers};
 use crate::service::ServerError;
-use crate::utils::{
-    self, ChunkedFormDataAggregator, EnvelopeSummary, FormDataIter, FutureExt, SamplingResult,
-};
+use crate::utils::{self, ChunkedFormDataAggregator, EnvelopeSummary, FormDataIter, FutureExt};
 
 #[cfg(feature = "processing")]
 use {
@@ -130,10 +128,10 @@ enum ProcessingError {
     #[fail(display = "event exceeded its configured lifetime")]
     Timeout,
 
-    #[fail(display = "envelope empty, transaction removed by sampling")]
-    TransactionSampled,
+    #[fail(display = "trace dropped by sampling rule {}", _0)]
+    TraceSampled(RuleId),
 
-    #[fail(display = "envelope empty, event removed by sampling rule:{}", _0)]
+    #[fail(display = "event dropped by sampling rule {}", _0)]
     EventSampled(RuleId),
 }
 
@@ -173,7 +171,7 @@ impl ProcessingError {
             }
 
             // Dynamic sampling (not an error, just discarding messages that were removed by sampling)
-            Self::TransactionSampled => Some(Outcome::Invalid(DiscardReason::TransactionSampled)),
+            Self::TraceSampled(rule_id) => Some(Outcome::FilteredSampling(rule_id)),
             Self::EventSampled(rule_id) => Some(Outcome::FilteredSampling(rule_id)),
 
             // If we send to an upstream, we don't emit outcomes.
@@ -1553,15 +1551,8 @@ impl Handler<HandleEnvelope> for EventManager {
                 }
             }))
             .and_then(move |envelope| {
-                utils::sample_transaction(envelope, sampling_project, false, processing_enabled)
-                    .map_err(|()| (ProcessingError::TransactionSampled))
-            })
-            .and_then(|envelope| {
-                if envelope.is_empty() {
-                    Err(ProcessingError::TransactionSampled)
-                } else {
-                    Ok(envelope)
-                }
+                utils::sample_trace(envelope, sampling_project, false, processing_enabled)
+                    .map_err(ProcessingError::TraceSampled)
             })
             .and_then(clone!(project, |envelope| {
                 // get the state for the current project. we can always fetch the cached version
@@ -1798,11 +1789,11 @@ impl Handler<GetCapturedEnvelope> for EventManager {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use chrono::{DateTime, TimeZone, Utc};
 
     use crate::extractors::RequestMeta;
 
-    use chrono::{DateTime, TimeZone, Utc};
+    use super::*;
 
     fn create_breadcrumbs_item(breadcrumbs: &[(Option<DateTime<Utc>>, &str)]) -> Item {
         let mut data = Vec::new();

relay-server/src/endpoints/common.rs

@@ -16,6 +16,7 @@ use relay_config::Config;
 use relay_general::protocol::{EventId, EventType};
 use relay_log::LogError;
 use relay_quotas::RateLimits;
+use relay_sampling::RuleId;
 
 use crate::actors::events::{QueueEnvelope, QueueEnvelopeError};
 use crate::actors::outcome::{DiscardReason, Outcome, TrackOutcome};
@@ -82,7 +83,7 @@ pub enum BadStoreRequest {
     EventRejected(DiscardReason),
 
     #[fail(display = "envelope empty due to sampling")]
-    TraceSampled(Option<EventId>),
+    TraceSampled(RuleId),
 }
 
 impl BadStoreRequest {
@@ -131,9 +132,7 @@ impl BadStoreRequest {
                     .longest_error()
                     .map(|r| Outcome::RateLimited(r.reason_code.clone()));
             }
-            //TODO fix this when we decide how to map empty Envelopes due to the trace being
-            //removed by sampling
-            BadStoreRequest::TraceSampled(_) => Outcome::Invalid(DiscardReason::Internal),
+            BadStoreRequest::TraceSampled(rule_id) => Outcome::FilteredSampling(*rule_id),
 
             // should actually never create an outcome
             BadStoreRequest::InvalidEventId => Outcome::Invalid(DiscardReason::Internal),
@@ -492,21 +491,14 @@ where
                     Box::new(response) as RetVal
                 }))
                 .and_then(move |(envelope, rate_limits, sampling_project)| {
-                    // do the fast path transaction sampling (if we can't do it here
-                    // we'll try again after the envelope is queued)
-                    let event_id = envelope.event_id();
-
-                    utils::sample_transaction(
+                    utils::sample_trace(
                         envelope,
                         sampling_project.clone(),
                         true,
                         processing_enabled,
                     )
                     .then(move |result| match result {
-                        Err(()) => Err(BadStoreRequest::TraceSampled(event_id)),
-                        Ok(envelope) if envelope.is_empty() => {
-                            Err(BadStoreRequest::TraceSampled(event_id))
-                        }
+                        Err(rule_id) => Err(BadStoreRequest::TraceSampled(rule_id)),
                         Ok(envelope) => Ok((envelope, rate_limits, sampling_project)),
                     })
                 })
@@ -564,9 +556,8 @@ where
                 return Ok(create_response(*event_id.borrow()));
             }
 
-            if let BadStoreRequest::TraceSampled(event_id) = error {
-                relay_log::debug!("creating response for trace sampled event");
-                return Ok(create_response(event_id));
+            if matches!(error, BadStoreRequest::TraceSampled(_)) {
+                return Ok(create_response(*event_id.borrow()));
             }
 
             let response = error.error_response();