feat(stats): Emit outcomes for applied rate limits

relay-server/src/actors/events.rs

@@ -1533,7 +1533,7 @@ impl Handler<HandleEnvelope> for EventManager {
         let envelope_summary = Rc::new(RefCell::new(EnvelopeSummary::compute(&envelope)));
 
         let future = project
-            .send(CheckEnvelope::fetched(envelope))
+            .send(CheckEnvelope::fetched(envelope, &outcome_producer))
             .map_err(ProcessingError::ScheduleFailed)
             .and_then(|result| result.map_err(ProcessingError::ProjectFailed))
             .and_then(clone!(scoping, envelope_summary, |response| {

relay-server/src/actors/project.rs

@@ -1,5 +1,5 @@
-use std::sync::Arc;
 use std::time::{Duration, Instant};
+use std::{net::IpAddr, sync::Arc};
 
 use actix::prelude::*;
 use chrono::{DateTime, Utc};
@@ -10,19 +10,24 @@ use smallvec::SmallVec;
 use url::Url;
 
 use relay_auth::PublicKey;
-use relay_common::{metric, ProjectId, ProjectKey};
+use relay_common::{metric, DataCategory, ProjectId, ProjectKey};
 use relay_config::Config;
 use relay_filter::{matches_any_origin, FiltersConfig};
-use relay_general::pii::{DataScrubbingConfig, PiiConfig};
-use relay_quotas::{Quota, RateLimits, Scoping};
+use relay_general::{
+    pii::{DataScrubbingConfig, PiiConfig},
+    protocol::EventId,
+};
+use relay_quotas::{Quota, RateLimits, ReasonCode, Scoping};
 use relay_sampling::SamplingConfig;
 
 use crate::actors::outcome::DiscardReason;
 use crate::actors::project_cache::{FetchProjectState, ProjectCache, ProjectError};
 use crate::envelope::Envelope;
 use crate::extractors::RequestMeta;
 use crate::metrics::RelayCounters;
-use crate::utils::{ActorResponse, EnvelopeLimiter, Response};
+use crate::utils::{ActorResponse, EnvelopeLimiter, EnvelopeSummary, Response};
+
+use super::outcome::{Outcome, OutcomeProducer, TrackOutcome};
 
 /// The current status of a project state. Return value of `ProjectState::outdated`.
 #[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
@@ -549,9 +554,12 @@ impl Project {
 
     fn check_envelope(
         &mut self,
-        mut envelope: Envelope,
+        message: CheckEnvelope,
         scoping: &Scoping,
     ) -> Result<CheckedEnvelope, DiscardReason> {
+        let rate_limit_envelope = RateLimitEnvelope::new(&message, scoping);
+        let mut envelope = message.envelope;
+
         if let Some(state) = self.state() {
             state.check_request(envelope.meta(), &self.config)?;
         }
@@ -560,7 +568,9 @@ impl Project {
 
         let quotas = self.state().map(|s| s.get_quotas()).unwrap_or(&[]);
         let envelope_limiter = EnvelopeLimiter::new(|item_scoping, _| {
-            Ok(self.rate_limits.check_with_quotas(quotas, item_scoping))
+            let applied_limits = self.rate_limits.check_with_quotas(quotas, item_scoping);
+            rate_limit_envelope.emit_rate_limit_outcomes(&applied_limits);
+            Ok(applied_limits)
         });
 
         let rate_limits = envelope_limiter.enforce(&mut envelope, scoping)?;
@@ -578,7 +588,7 @@ impl Project {
 
     fn check_envelope_scoped(&mut self, message: CheckEnvelope) -> CheckEnvelopeResponse {
         let scoping = self.get_scoping(message.envelope.meta());
-        let result = self.check_envelope(message.envelope, &scoping);
+        let result = self.check_envelope(message, &scoping);
         CheckEnvelopeResponse { result, scoping }
     }
 }
@@ -595,6 +605,60 @@ impl Actor for Project {
     }
 }
 
+struct RateLimitEnvelope {
+    event_id: Option<EventId>,
+    remote_addr: Option<IpAddr>,
+    envelope_summary: EnvelopeSummary,
+    outcome_producer: Addr<OutcomeProducer>,
+    scoping: Scoping,
+}
+
+impl RateLimitEnvelope {
+    fn new(message: &CheckEnvelope, scoping: &Scoping) -> Self {
+        let envelope = &message.envelope;
+        Self {
+            event_id: envelope.event_id(),
+            remote_addr: envelope.meta().client_addr(),
+            envelope_summary: EnvelopeSummary::compute(&envelope),
+            outcome_producer: message.outcome_producer.clone(),
+            scoping: *scoping,
+        }
+    }
+
+    fn emit_rate_limit_outcomes(&self, applied_limits: &RateLimits) {
+        for applied_limit in applied_limits.iter() {
+            if applied_limit.categories.is_empty() {
+                // Empty categories value indicates that the rate limit applies to all data.
+                if let Some(event_category) = self.envelope_summary.event_category {
+                    self.emit_outcome(&event_category, &applied_limit.reason_code);
+                }
+                if self.envelope_summary.attachment_quantity > 0 {
+                    self.emit_outcome(&DataCategory::Attachment, &applied_limit.reason_code);
+                }
+            } else {
+                for category in applied_limit.categories.iter() {
+                    self.emit_outcome(category, &applied_limit.reason_code);
+                }
+            }
+        }
+    }
+
+    fn emit_outcome(&self, category: &DataCategory, reason_code: &Option<ReasonCode>) {
+        self.outcome_producer.do_send(TrackOutcome {
+            timestamp: Instant::now(),
+            scoping: self.scoping,
+            outcome: Outcome::RateLimited(reason_code.clone()),
+            event_id: self.event_id,
+            remote_addr: self.remote_addr,
+            category: *category,
+            quantity: match category {
+                DataCategory::Attachment => self.envelope_summary.attachment_quantity,
+                _ => 1,
+            },
+        })
+    }
+}
+
 /// Returns the project state if it is already cached.
 ///
 /// This is used for cases when we only want to perform operations that do
@@ -660,26 +724,38 @@ impl Handler<GetProjectState> for Project {
 ///  - Validate origins and public keys
 ///  - Quotas with a limit of `0`
 ///  - Cached rate limits
-#[derive(Debug)]
 pub struct CheckEnvelope {
     envelope: Envelope,
     fetch: bool,
+    outcome_producer: Addr<OutcomeProducer>,
+}
+
+impl std::fmt::Debug for CheckEnvelope {
+    // TODO: Is this good enough? Can we avoid needing it?
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("CheckEnvelope")
+            .field("envelope", &self.envelope)
+            .field("fetch", &self.fetch)
+            .finish()
+    }
 }
 
 impl CheckEnvelope {
     /// Fetches the project state and checks the envelope.
-    pub fn fetched(envelope: Envelope) -> Self {
+    pub fn fetched(envelope: Envelope, outcome_producer: &Addr<OutcomeProducer>) -> Self {
         Self {
             envelope,
             fetch: true,
+            outcome_producer: outcome_producer.clone(),
         }
     }
 
     /// Uses a cached project state and checks the envelope.
-    pub fn cached(envelope: Envelope) -> Self {
+    pub fn cached(envelope: Envelope, outcome_producer: &Addr<OutcomeProducer>) -> Self {
         Self {
             envelope,
             fetch: false,
+            outcome_producer: outcome_producer.clone(),
         }
     }
 }