Merge branch 'master' into feat/metrics-ingestion

* master:
  feat(stats): Add quantity to TrackOutcome and new attachment outcomes (#942)

Author: jan-auer

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 **Internal**:
 
+- 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))
 
 ## 21.3.0

relay-server/src/actors/events.rs

@@ -37,7 +37,9 @@ 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, SamplingResult};
+use crate::utils::{
+    self, ChunkedFormDataAggregator, EnvelopeSummary, FormDataIter, FutureExt, SamplingResult,
+};
 
 #[cfg(feature = "processing")]
 use {
@@ -1522,19 +1524,19 @@ impl Handler<HandleEnvelope> for EventManager {
             start_time,
             sampling_project,
         } = message;
-        let event_category = envelope.get_event_category();
 
         let event_id = envelope.event_id();
         let remote_addr = envelope.meta().client_addr();
 
         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::compute(&envelope)));
 
         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);
@@ -1543,7 +1545,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)),
                 }
             }))
@@ -1577,6 +1582,7 @@ impl Handler<HandleEnvelope> for EventManager {
                     })
                     .map_err(ProcessingError::ScheduleFailed)
                     .flatten()
+                // TODO: Update envelope_summary once the rate-limiting code emits its own outcomes.
             })
             .and_then(clone!(project, |processed| {
                 let rate_limits = processed.rate_limits;
@@ -1714,13 +1720,6 @@ impl Handler<HandleEnvelope> for EventManager {
                     }
                 }
 
-                // Envelopes not containing events (such as standalone attachment uploads or user
-                // reports) should never create outcomes.
-                let category = match event_category {
-                    Some(event_category) => event_category,
-                    None => return,
-                };
-
                 let outcome = error.to_outcome();
                 if let Some(Outcome::Invalid(DiscardReason::Internal)) = outcome {
                     // Errors are only logged for what we consider an internal discard reason. These
@@ -1738,15 +1737,31 @@ impl Handler<HandleEnvelope> for EventManager {
                     return;
                 }
 
+                let envelope_summary = envelope_summary.borrow();
                 if let Some(outcome) = outcome {
-                    outcome_producer.do_send(TrackOutcome {
-                        timestamp: Instant::now(),
-                        scoping: *scoping.borrow(),
-                        outcome,
-                        event_id,
-                        remote_addr,
-                        category,
-                    })
+                    if let Some(category) = envelope_summary.event_category {
+                        outcome_producer.do_send(TrackOutcome {
+                            timestamp: Instant::now(),
+                            scoping: *scoping.borrow(),
+                            outcome: outcome.clone(),
+                            event_id,
+                            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,
+                            remote_addr,
+                            category: DataCategory::Attachment,
+                            quantity: envelope_summary.attachment_quantity,
+                        });
+                    }
                 }
             })
             .then(move |x, slf, _| {

relay-server/src/actors/outcome.rs

@@ -84,6 +84,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 {
@@ -316,6 +318,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 {
@@ -351,6 +356,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 {
@@ -411,19 +411,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)
@@ -447,6 +447,8 @@ where
                         Some(envelope) => envelope,
                         None => return Err(BadStoreRequest::RateLimited(checked.rate_limits)),
                     };
+                    // TODO: Update envelope_summary from checked.envelope, once the rate-limiting
+                    // code in CheckEnvelope emits its own outcomes.
 
                     if check_envelope_size_limits(&config, &envelope) {
                         Ok((envelope, checked.rate_limits))
@@ -531,15 +533,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,
                     });
                 }
             }

relay-server/src/envelope.rs

@@ -41,14 +41,13 @@ use failure::Fail;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use smallvec::SmallVec;
 
-use relay_common::DataCategory;
 use relay_general::protocol::{EventId, EventType};
 use relay_general::types::Value;
 use relay_sampling::TraceContext;
 
 use crate::constants::DEFAULT_EVENT_RETENTION;
 use crate::extractors::{PartialMeta, RequestMeta};
-use crate::utils::{infer_event_category, ErrorBoundary};
+use crate::utils::ErrorBoundary;
 
 pub const CONTENT_TYPE: &str = "application/x-sentry-envelope";
 
@@ -961,13 +960,6 @@ impl Envelope {
             .write_all(buf)
             .map_err(EnvelopeError::PayloadIoFailed)
     }
-
-    /// Return the data category type of the event item, if any, in this envelope.
-    pub fn get_event_category(&self) -> Option<DataCategory> {
-        self.items().find_map(infer_event_category)
-        // There are some cases where multiple items may have different categories, but returning
-        // the first is good enough for now.
-    }
 }
 
 #[cfg(test)]

relay-server/src/utils/rate_limits.rs

@@ -85,7 +85,7 @@ pub fn parse_rate_limits(scoping: &Scoping, string: &str) -> RateLimits {
 ///   to be set on the event item.
 /// - `Attachment`: If the attachment creates an event (e.g. for minidumps), the category is assumed
 ///   to be `Error`.
-pub fn infer_event_category(item: &Item) -> Option<DataCategory> {
+fn infer_event_category(item: &Item) -> Option<DataCategory> {
     match item.ty() {
         ItemType::Event => Some(DataCategory::Error),
         ItemType::Transaction => Some(DataCategory::Transaction),

tests/integration/fixtures/processing.py

@@ -207,23 +207,42 @@ def category_value(category):
 
 
 class OutcomesConsumer(ConsumerBase):
-    def get_outcome(self):
-        outcome = self.poll()
-        assert outcome is not None
-        assert outcome.error() is None
-        return json.loads(outcome.value())
+    def _poll_all(self):
+        while True:
+            outcome = self.poll()
+            if outcome is None:
+                return
+            else:
+                yield outcome
+
+    def get_outcomes(self):
+        outcomes = list(self._poll_all())
+        for outcome in outcomes:
+            assert outcome.error() is None
+        return [json.loads(outcome.value()) for outcome in outcomes]
 
-    def assert_rate_limited(self, reason, key_id=None, category=None):
-        outcome = self.get_outcome()
-        assert outcome["outcome"] == 2, outcome
-        assert outcome["reason"] == reason
-        if key_id is not None:
-            assert outcome["key_id"] == key_id
-        if category is not None:
-            value = category_value(category)
-            assert outcome["category"] == value, outcome["category"]
-        else:
+    def get_outcome(self):
+        outcomes = self.get_outcomes()
+        assert len(outcomes) > 0, "No outcomes were consumed"
+        assert len(outcomes) == 1, "More than one outcome was consumed"
+        return outcomes[0]
+
+    def assert_rate_limited(self, reason, key_id=None, categories=None):
+        if categories is None:
+            outcome = self.get_outcome()
             assert isinstance(outcome["category"], int)
+            outcomes = [outcome]
+        else:
+            outcomes = self.get_outcomes()
+            expected = set(category_value(category) for category in categories)
+            actual = set(outcome["category"] for outcome in outcomes)
+            assert actual == expected, (actual, expected)
+
+        for outcome in outcomes:
+            assert outcome["outcome"] == 2, outcome
+            assert outcome["reason"] == reason
+            if key_id is not None:
+                assert outcome["key_id"] == key_id
 
     def assert_dropped_internal(self):
         outcome = self.get_outcome()

tests/integration/test_minidump.py

@@ -464,11 +464,15 @@ def test_minidump_ratelimit(
 
     # First minidump returns 200 but is rate limited in processing
     relay.send_minidump(project_id=project_id, files=attachments)
-    outcomes_consumer.assert_rate_limited("static_disabled_quota", category="error")
+    outcomes_consumer.assert_rate_limited(
+        "static_disabled_quota", categories=["error", "attachment"]
+    )
 
     # Minidumps never return rate limits
     relay.send_minidump(project_id=project_id, files=attachments)
-    outcomes_consumer.assert_rate_limited("static_disabled_quota", category="error")
+    outcomes_consumer.assert_rate_limited(
+        "static_disabled_quota", categories=["error", "attachment"]
+    )
 
 
 def test_crashpad_annotations(mini_sentry, relay_with_processing, attachments_consumer):

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
 

tests/integration/test_store.py

@@ -538,7 +538,7 @@ def test_processing_quotas(
 
     if outcomes_consumer is not None:
         outcomes_consumer.assert_rate_limited(
-            "get_lost", key_id=key_id, category=category
+            "get_lost", key_id=key_id, categories=[category]
         )
     else:
         # since we don't wait for the outcome, wait a little for the event to go through
@@ -558,7 +558,7 @@ def test_processing_quotas(
         assert rest == "%s:key:get_lost" % category
         if outcomes_consumer is not None:
             outcomes_consumer.assert_rate_limited(
-                "get_lost", key_id=key_id, category=category
+                "get_lost", key_id=key_id, categories=[category]
             )
 
     for i in range(10):