feat(outcomes): add item type to invalid outcome (#4558)

This PR adds the ItemType to the TooLarge outcome.


Fixes getsentry/team-ingest#635

Author: Litarnus

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Add experimental playstation endpoint. ([#4555](https://github.com/getsentry/relay/pull/4555))
 - Add naver.me / Yeti spider on the crawler filter list. ([#4602](https://github.com/getsentry/relay/pull/4602))
+- Add the item type that made the envelope too large to invalid outcomes. ([#4558](https://github.com/getsentry/relay/pull/4558))
 
 **Bug Fixes**:
 

relay-server/src/endpoints/common.rs

@@ -383,7 +383,9 @@ pub async fn handle_envelope(
     if let Err(offender) =
         utils::check_envelope_size_limits(state.config(), managed_envelope.envelope())
     {
-        managed_envelope.reject(Outcome::Invalid(DiscardReason::TooLarge));
+        managed_envelope.reject(Outcome::Invalid(DiscardReason::TooLarge(
+            (&offender).into(),
+        )));
         return Err(BadStoreRequest::Overflow(offender));
     }
 

relay-server/src/services/outcome.rs

@@ -12,6 +12,13 @@ use std::sync::Arc;
 use std::time::Duration;
 use std::{fmt, mem};
 
+use crate::envelope::ItemType;
+#[cfg(feature = "processing")]
+use crate::service::ServiceError;
+use crate::services::processor::{EnvelopeProcessor, SubmitClientReports};
+use crate::services::upstream::{Method, SendQuery, UpstreamQuery, UpstreamRelay};
+use crate::statsd::RelayCounters;
+use crate::utils::SleepHandle;
 use chrono::{DateTime, SecondsFormat, Utc};
 use relay_base_schema::organization::OrganizationId;
 use relay_base_schema::project::ProjectId;
@@ -29,13 +36,6 @@ use relay_statsd::metric;
 use relay_system::{Addr, FromMessage, Interface, NoResponse, Service};
 use serde::{Deserialize, Serialize};
 
-#[cfg(feature = "processing")]
-use crate::service::ServiceError;
-use crate::services::processor::{EnvelopeProcessor, SubmitClientReports};
-use crate::services::upstream::{Method, SendQuery, UpstreamQuery, UpstreamRelay};
-use crate::statsd::RelayCounters;
-use crate::utils::SleepHandle;
-
 /// Defines the structure of the HTTP outcomes requests
 #[derive(Debug, Default, Deserialize, Serialize)]
 pub struct SendOutcomes {
@@ -205,6 +205,9 @@ impl Outcome {
     /// Returns the `reason` code field of this outcome.
     pub fn to_reason(&self) -> Option<Cow<'_, str>> {
         match self {
+            Outcome::Invalid(DiscardReason::TooLarge(too_large_reason)) => Some(Cow::Owned(
+                format!("too_large:{}", too_large_reason.as_str()),
+            )),
             Outcome::Invalid(discard_reason) => Some(Cow::Borrowed(discard_reason.name())),
             Outcome::Filtered(filter_key) => Some(filter_key.clone().name()),
             Outcome::FilteredSampling(rule_ids) => Some(Cow::Owned(format!("Sampled:{rule_ids}"))),
@@ -356,7 +359,7 @@ pub enum DiscardReason {
     EmptyEnvelope,
 
     /// (Relay) The event payload exceeds the maximum size limit for the respective endpoint.
-    TooLarge,
+    TooLarge(DiscardItemType),
 
     /// (Legacy) A store request was received with an invalid method.
     ///
@@ -476,7 +479,7 @@ impl DiscardReason {
             DiscardReason::AuthVersion => "auth_version",
             DiscardReason::AuthClient => "auth_client",
             DiscardReason::NoData => "no_data",
-            DiscardReason::TooLarge => "too_large",
+            DiscardReason::TooLarge(discard) => discard.as_str(),
             DiscardReason::DisallowedMethod => "disallowed_method",
             DiscardReason::ContentType => "content_type",
             DiscardReason::MultiProjectId => "multi_project_id",
@@ -524,6 +527,149 @@ impl fmt::Display for DiscardReason {
     }
 }
 
+/// Similar to [`ItemType`] but it does not have any additional information in the
+/// Unknown variant so that it can derive [`Copy`] and be used from [`DiscardReason`].
+/// The variants should be the same as [`ItemType`].
+#[derive(Copy, Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub enum DiscardItemType {
+    /// Event payload encoded in JSON.
+    Event,
+    /// Transaction event payload encoded in JSON.
+    Transaction,
+    /// Security report event payload encoded in JSON.
+    Security,
+    /// Raw payload of an arbitrary attachment.
+    Attachment,
+    /// Multipart form data collected into a stream of JSON tuples.
+    FormData,
+    /// Security report as sent by the browser in JSON.
+    RawSecurity,
+    /// NEL report as sent by the browser.
+    Nel,
+    /// Raw compressed UE4 crash report.
+    UnrealReport,
+    /// User feedback encoded as JSON.
+    UserReport,
+    /// Session update data.
+    Session,
+    /// Aggregated session data.
+    Sessions,
+    /// Individual metrics in text encoding.
+    Statsd,
+    /// Buckets of preaggregated metrics encoded as JSON.
+    MetricBuckets,
+    /// Client internal report (eg: outcomes).
+    ClientReport,
+    /// Profile event payload encoded as JSON.
+    Profile,
+    /// Replay metadata and breadcrumb payload.
+    ReplayEvent,
+    /// Replay Recording data.
+    ReplayRecording,
+    /// Replay Video data.
+    ReplayVideo,
+    /// Monitor check-in encoded as JSON.
+    CheckIn,
+    /// A log from the [OTEL Log format](https://opentelemetry.io/docs/specs/otel/logs/data-model/#log-and-event-record-definition)
+    OtelLog,
+    /// A log for the log product, not internal logs.
+    Log,
+    /// A standalone span.
+    Span,
+    /// A standalone OpenTelemetry span serialized as JSON.
+    OtelSpan,
+    /// An OTLP TracesData container.
+    OtelTracesData,
+    /// UserReport as an Event
+    UserReportV2,
+    /// ProfileChunk is a chunk of a profiling session.
+    ProfileChunk,
+    /// A new item type that is yet unknown by this version of Relay.
+    ///
+    /// By default, items of this type are forwarded without modification. Processing Relays and
+    /// Relays explicitly configured to do so will instead drop those items. This allows
+    /// forward-compatibility with new item types where we expect outdated Relays.
+    Unknown,
+    // Keep `Unknown` last in the list. Add new items above `Unknown`.
+}
+
+impl DiscardItemType {
+    /// Returns the enum variant as string representation in snake case.
+    ///
+    /// For example, `DiscardItemType::UserReportV2` gets converted into `user_report_v2`.
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Self::Unknown => "unknown",
+            Self::Event => "event",
+            Self::Transaction => "transaction",
+            Self::Security => "security",
+            Self::Attachment => "attachment",
+            Self::FormData => "form_data",
+            Self::RawSecurity => "raw_security",
+            Self::Nel => "nel",
+            Self::UnrealReport => "unreal_report",
+            Self::UserReport => "user_report",
+            Self::Session => "session",
+            Self::Sessions => "sessions",
+            Self::Statsd => "statsd",
+            Self::MetricBuckets => "metric_buckets",
+            Self::ClientReport => "client_report",
+            Self::Profile => "profile",
+            Self::ReplayEvent => "replay_event",
+            Self::ReplayRecording => "replay_recording",
+            Self::ReplayVideo => "replay_video",
+            Self::CheckIn => "check_in",
+            Self::OtelLog => "otel_log",
+            Self::Log => "log",
+            Self::Span => "span",
+            Self::OtelSpan => "otel_span",
+            Self::OtelTracesData => "otel_traces_data",
+            Self::UserReportV2 => "user_report_v2",
+            Self::ProfileChunk => "profile_chunk",
+        }
+    }
+}
+
+impl From<&ItemType> for DiscardItemType {
+    fn from(value: &ItemType) -> Self {
+        match value {
+            ItemType::Event => DiscardItemType::Event,
+            ItemType::Transaction => DiscardItemType::Transaction,
+            ItemType::Security => DiscardItemType::Security,
+            ItemType::Attachment => DiscardItemType::Attachment,
+            ItemType::FormData => DiscardItemType::FormData,
+            ItemType::RawSecurity => DiscardItemType::RawSecurity,
+            ItemType::Nel => DiscardItemType::Nel,
+            ItemType::UnrealReport => DiscardItemType::UnrealReport,
+            ItemType::UserReport => DiscardItemType::UserReport,
+            ItemType::Session => DiscardItemType::Session,
+            ItemType::Sessions => DiscardItemType::Sessions,
+            ItemType::Statsd => DiscardItemType::Statsd,
+            ItemType::MetricBuckets => DiscardItemType::MetricBuckets,
+            ItemType::ClientReport => DiscardItemType::ClientReport,
+            ItemType::Profile => DiscardItemType::Profile,
+            ItemType::ReplayEvent => DiscardItemType::ReplayEvent,
+            ItemType::ReplayRecording => DiscardItemType::ReplayRecording,
+            ItemType::ReplayVideo => DiscardItemType::ReplayVideo,
+            ItemType::CheckIn => DiscardItemType::CheckIn,
+            ItemType::OtelLog => DiscardItemType::OtelLog,
+            ItemType::Log => DiscardItemType::Log,
+            ItemType::Span => DiscardItemType::Span,
+            ItemType::OtelSpan => DiscardItemType::OtelSpan,
+            ItemType::OtelTracesData => DiscardItemType::OtelTracesData,
+            ItemType::UserReportV2 => DiscardItemType::UserReportV2,
+            ItemType::ProfileChunk => DiscardItemType::ProfileChunk,
+            ItemType::Unknown(_) => DiscardItemType::Unknown,
+        }
+    }
+}
+
+impl From<ItemType> for DiscardItemType {
+    fn from(value: ItemType) -> Self {
+        From::from(&value)
+    }
+}
+
 /// Raw representation of an outcome for serialization.
 ///
 /// The JSON serialization of this structure is placed on the Kafka topic and used in the HTTP
@@ -754,7 +900,10 @@ impl ClientReportOutcomeProducer {
             Outcome::FilteredSampling(_) => &mut client_report.filtered_sampling_events,
             Outcome::RateLimited(_) => &mut client_report.rate_limited_events,
             _ => {
-                // Cannot convert this outcome to a client report.
+                relay_log::debug!(
+                    "Outcome '{}' cannot be converted to client report",
+                    msg.outcome
+                );
                 return;
             }
         };
@@ -1076,4 +1225,20 @@ mod tests {
             MatchedRuleIds([1000, 1004, 1400, 1500, 0].map(RuleId).into())
         );
     }
+
+    #[test]
+    fn test_outcome_discard_reason() {
+        assert_eq!(
+            Some(Cow::from("too_large:attachment")),
+            Outcome::Invalid(DiscardReason::TooLarge(DiscardItemType::Attachment)).to_reason()
+        );
+        assert_eq!(
+            Some(Cow::from("too_large:unknown")),
+            Outcome::Invalid(DiscardReason::TooLarge(DiscardItemType::Unknown)).to_reason()
+        );
+        assert_eq!(
+            Some(Cow::from("too_large:unreal_report")),
+            Outcome::Invalid(DiscardReason::TooLarge(DiscardItemType::UnrealReport)).to_reason()
+        );
+    }
 }

relay-server/src/services/processor.rs

@@ -477,7 +477,7 @@ pub enum ProcessingError {
     InvalidUnrealReport(#[source] Unreal4Error),
 
     #[error("event payload too large")]
-    PayloadTooLarge,
+    PayloadTooLarge(ItemType),
 
     #[error("invalid transaction event")]
     InvalidTransaction,
@@ -536,7 +536,9 @@ impl ProcessingError {
     fn to_outcome(&self) -> Option<Outcome> {
         match self {
             // General outcomes for invalid events
-            Self::PayloadTooLarge => Some(Outcome::Invalid(DiscardReason::TooLarge)),
+            Self::PayloadTooLarge(item_type) => {
+                Some(Outcome::Invalid(DiscardReason::TooLarge(item_type.into())))
+            }
             Self::InvalidJson(_) => Some(Outcome::Invalid(DiscardReason::InvalidJson)),
             Self::InvalidMsgpack(_) => Some(Outcome::Invalid(DiscardReason::InvalidMsgpack)),
             Self::InvalidSecurityType(_) => {
@@ -588,7 +590,7 @@ impl ProcessingError {
 impl From<Unreal4Error> for ProcessingError {
     fn from(err: Unreal4Error) -> Self {
         match err.kind() {
-            Unreal4ErrorKind::TooLarge => Self::PayloadTooLarge,
+            Unreal4ErrorKind::TooLarge => Self::PayloadTooLarge(ItemType::UnrealReport),
             _ => ProcessingError::InvalidUnrealReport(err),
         }
     }

relay-server/src/services/processor/event.rs

@@ -602,7 +602,7 @@ fn extract_attached_event(
     // Protect against blowing up during deserialization. Attachments can have a significantly
     // larger size than regular events and may cause significant processing delays.
     if item.len() > config.max_event_size() {
-        return Err(ProcessingError::PayloadTooLarge);
+        return Err(ProcessingError::PayloadTooLarge(item.ty().to_owned()));
     }
 
     let payload = item.payload();
@@ -625,7 +625,7 @@ fn parse_msgpack_breadcrumbs(
     // blowing up during deserialization. As approximation, we use the maximum event payload
     // size as bound, which is roughly in the right ballpark.
     if item.len() > config.max_event_size() {
-        return Err(ProcessingError::PayloadTooLarge);
+        return Err(ProcessingError::PayloadTooLarge(item.ty().to_owned()));
     }
 
     let payload = item.payload();

relay-server/src/services/store.rs

@@ -36,7 +36,7 @@ use uuid::Uuid;
 use crate::metrics::{ArrayEncoding, BucketEncoder, MetricOutcomes};
 use crate::service::ServiceError;
 use crate::services::global_config::GlobalConfigHandle;
-use crate::services::outcome::{DiscardReason, Outcome, TrackOutcome};
+use crate::services::outcome::{DiscardItemType, DiscardReason, Outcome, TrackOutcome};
 use crate::services::processor::Processed;
 use crate::statsd::{RelayCounters, RelayGauges, RelayTimers};
 use crate::utils::{FormDataIter, TypedEnvelope};
@@ -776,7 +776,9 @@ impl StoreService {
             self.outcome_aggregator.send(TrackOutcome {
                 category: DataCategory::Replay,
                 event_id,
-                outcome: Outcome::Invalid(DiscardReason::TooLarge),
+                outcome: Outcome::Invalid(DiscardReason::TooLarge(
+                    DiscardItemType::ReplayRecording,
+                )),
                 quantity: 1,
                 remote_addr: None,
                 scoping,

relay-server/src/utils/unreal.rs

@@ -8,13 +8,14 @@ use relay_event_schema::protocol::{
 };
 use relay_protocol::{Annotated, Array, Empty, Object, Value};
 use symbolic_unreal::{
-    Unreal4Context, Unreal4Crash, Unreal4Error, Unreal4ErrorKind, Unreal4FileType, Unreal4LogEntry,
+    Unreal4Context, Unreal4Crash, Unreal4Error, Unreal4FileType, Unreal4LogEntry,
 };
 
 use crate::constants::{
     ITEM_NAME_BREADCRUMBS1, ITEM_NAME_BREADCRUMBS2, ITEM_NAME_EVENT, UNREAL_USER_HEADER,
 };
 use crate::envelope::{AttachmentType, ContentType, Envelope, Item, ItemType};
+use crate::services::processor::ProcessingError;
 
 /// Maximum number of unreal logs to parse for breadcrumbs.
 const MAX_NUM_UNREAL_LOGS: usize = 40;
@@ -50,7 +51,7 @@ pub fn expand_unreal_envelope(
     unreal_item: Item,
     envelope: &mut Envelope,
     config: &Config,
-) -> Result<(), Unreal4Error> {
+) -> Result<(), ProcessingError> {
     let payload = unreal_item.payload();
     let crash = Unreal4Crash::parse_with_limit(&payload, config.max_envelope_size())?;
 
@@ -90,8 +91,8 @@ pub fn expand_unreal_envelope(
         envelope.add_item(item);
     }
 
-    if super::check_envelope_size_limits(config, envelope).is_err() {
-        return Err(Unreal4ErrorKind::TooLarge.into());
+    if let Err(offender) = super::check_envelope_size_limits(config, envelope) {
+        return Err(ProcessingError::PayloadTooLarge(offender));
     }
 
     Ok(())