feat(stats): Add quantity to TrackOutcome and new attachment outcomes

relay-server/src/actors/events.rs

@@ -35,7 +35,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, FormDataIter, FutureExt};
+use crate::utils::{self, ChunkedFormDataAggregator, EnvelopeSummary, FormDataIter, FutureExt};
 
 #[cfg(feature = "processing")]
 use {
@@ -1483,12 +1483,13 @@ impl Handler<HandleEnvelope> for EventManager {
 
         let scoping = Rc::new(RefCell::new(envelope.meta().get_partial_scoping()));
         let is_received = Rc::new(AtomicBool::from(false));
+        let envelope_summary = Rc::new(RefCell::new(EnvelopeSummary::empty()));
 
         let future = project
             .send(CheckEnvelope::fetched(envelope))
             .map_err(ProcessingError::ScheduleFailed)
             .and_then(|result| result.map_err(ProcessingError::ProjectFailed))
-            .and_then(clone!(scoping, |response| {
+            .and_then(clone!(scoping, envelope_summary, |response| {
                 // Use the project id from the loaded project state to account for redirects.
                 let project_id = response.scoping.project_id.value();
                 metric!(set(RelaySets::UniqueProjects) = project_id as i64);
@@ -1497,7 +1498,10 @@ impl Handler<HandleEnvelope> for EventManager {
 
                 let checked = response.result.map_err(ProcessingError::EventRejected)?;
                 match checked.envelope {
-                    Some(envelope) => Ok(envelope),
+                    Some(envelope) => {
+                        envelope_summary.replace(EnvelopeSummary::compute(&envelope));
+                        Ok(envelope)
+                    }
                     None => Err(ProcessingError::RateLimited(checked.rate_limits)),
                 }
             }))
@@ -1696,11 +1700,25 @@ impl Handler<HandleEnvelope> for EventManager {
                     outcome_producer.do_send(TrackOutcome {
                         timestamp: Instant::now(),
                         scoping: *scoping.borrow(),
-                        outcome,
+                        outcome: outcome.clone(),
                         event_id,
                         remote_addr,
                         category,
-                    })
+                        quantity: 1,
+                    });
+
+                    let envelope_summary = envelope_summary.borrow();
+                    if envelope_summary.attachment_quantity > 0 {
+                        outcome_producer.do_send(TrackOutcome {
+                            timestamp: start_time,
+                            scoping: *scoping.borrow(),
+                            outcome,
+                            event_id,
+                            remote_addr,
+                            category: DataCategory::Attachment,
+                            quantity: envelope_summary.attachment_quantity,
+                        });
+                    }
                 }
             })
             .then(move |x, slf, _| {

relay-server/src/actors/outcome.rs

@@ -83,6 +83,8 @@ pub struct TrackOutcome {
     pub remote_addr: Option<IpAddr>,
     /// The event's data category.
     pub category: DataCategory,
+    /// The number of events or total attachment size in bytes.
+    pub quantity: usize,
 }
 
 impl Message for TrackOutcome {
@@ -313,6 +315,9 @@ pub struct TrackRawOutcome {
     /// The event's data category.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub category: Option<u8>,
+    /// The number of events or total attachment size in bytes.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub quantity: Option<usize>,
 }
 
 impl TrackRawOutcome {
@@ -348,6 +353,7 @@ impl TrackRawOutcome {
             remote_addr: msg.remote_addr.map(|addr| addr.to_string()),
             source,
             category: msg.category.value(),
+            quantity: Some(msg.quantity),
         }
     }
 }

relay-server/src/endpoints/common.rs

@@ -11,7 +11,7 @@ use failure::Fail;
 use futures::prelude::*;
 use serde::Deserialize;
 
-use relay_common::{clone, metric, tryf};
+use relay_common::{clone, metric, tryf, DataCategory};
 use relay_config::Config;
 use relay_general::protocol::{EventId, EventType};
 use relay_log::LogError;
@@ -26,7 +26,7 @@ use crate::envelope::{AttachmentType, Envelope, EnvelopeError, ItemType, Items};
 use crate::extractors::RequestMeta;
 use crate::metrics::RelayCounters;
 use crate::service::{ServiceApp, ServiceState};
-use crate::utils::{self, ApiErrorResponse, FormDataIter, MultipartError};
+use crate::utils::{self, ApiErrorResponse, EnvelopeSummary, FormDataIter, MultipartError};
 
 #[derive(Fail, Debug)]
 pub enum BadStoreRequest {
@@ -410,19 +410,19 @@ where
 
     let scoping = Rc::new(RefCell::new(meta.get_partial_scoping()));
     let event_id = Rc::new(RefCell::new(None));
-    let event_category = Rc::new(RefCell::new(None));
+    let envelope_summary = Rc::new(RefCell::new(EnvelopeSummary::empty()));
     let config = request.state().config();
     let processing_enabled = config.processing_enabled();
 
     let future = project_manager
         .send(GetProject { public_key })
         .map_err(BadStoreRequest::ScheduleFailed)
-        .and_then(clone!(event_id, event_category, scoping, |project| {
+        .and_then(clone!(event_id, envelope_summary, scoping, |project| {
             extract_envelope(&request, meta)
                 .into_future()
-                .and_then(clone!(event_id, event_category, |envelope| {
+                .and_then(clone!(event_id, envelope_summary, |envelope| {
                     event_id.replace(envelope.event_id());
-                    event_category.replace(envelope.get_event_category());
+                    envelope_summary.replace(EnvelopeSummary::compute(&envelope));
 
                     if envelope.is_empty() {
                         Err(BadStoreRequest::EmptyEnvelope)
@@ -446,6 +446,8 @@ where
                         Some(envelope) => envelope,
                         None => return Err(BadStoreRequest::RateLimited(checked.rate_limits)),
                     };
+                    // TODO: Ensure that outcomes are emitted correctly for rate-limited attachments
+                    // so that it is safe to update envelope_summary here.
 
                     if check_envelope_size_limits(&config, &envelope) {
                         Ok((envelope, checked.rate_limits))
@@ -530,15 +532,29 @@ where
         .or_else(move |error: BadStoreRequest| {
             metric!(counter(RelayCounters::EnvelopeRejected) += 1);
 
-            if let Some(category) = *event_category.borrow() {
-                if let Some(outcome) = error.to_outcome() {
+            let envelope_summary = envelope_summary.borrow();
+            if let Some(outcome) = error.to_outcome() {
+                if let Some(category) = envelope_summary.event_category {
                     outcome_producer.do_send(TrackOutcome {
                         timestamp: start_time,
                         scoping: *scoping.borrow(),
-                        outcome,
+                        outcome: outcome.clone(),
                         event_id: *event_id.borrow(),
                         remote_addr,
                         category,
+                        quantity: 1,
+                    });
+                }
+
+                if envelope_summary.attachment_quantity > 0 {
+                    outcome_producer.do_send(TrackOutcome {
+                        timestamp: start_time,
+                        scoping: *scoping.borrow(),
+                        outcome,
+                        event_id: *event_id.borrow(),
+                        remote_addr,
+                        category: DataCategory::Attachment,
+                        quantity: envelope_summary.attachment_quantity,
                     });
                 }
             }

tests/integration/test_outcome.py

@@ -104,6 +104,7 @@ def test_outcomes_non_processing(relay, mini_sentry, event_type):
         "event_id": event_id,
         "remote_addr": "127.0.0.1",
         "category": 2 if event_type == "transaction" else 1,
+        "quantity": 1,
     }
     assert outcome == expected_outcome
 
@@ -303,6 +304,7 @@ def test_outcome_forwarding(
         "event_id": event_id,
         "remote_addr": "127.0.0.1",
         "category": 2 if event_type == "transaction" else 1,
+        "quantity": 1,
     }
     outcome.pop("timestamp")
 
@@ -377,6 +379,7 @@ def test_outcomes_forwarding_rate_limited(
         "event_id": event_id,
         "source": "processing-layer",
         "category": 1,
+        "quantity": 1,
     }
     assert outcome == expected_outcome