From 409e55fb30b36efaa99b898a9f846f0a80ee23a7 Mon Sep 17 00:00:00 2001 From: Chris Klochek Date: Fri, 24 Jul 2026 16:15:24 -0400 Subject: [PATCH] ref(relay): Limit maximum number of logs produced by expansion --- relay-config/src/config.rs | 8 ++++++++ .../src/processing/logs/integrations/mod.rs | 12 ++++++++++-- .../src/processing/logs/integrations/nel.rs | 13 ++++++++----- .../src/processing/logs/integrations/otel.rs | 4 ++-- .../src/processing/logs/integrations/vercel.rs | 6 +++--- relay-server/src/processing/logs/mod.rs | 7 ++++++- relay-server/src/processing/logs/process.rs | 8 ++++++-- 7 files changed, 43 insertions(+), 15 deletions(-) diff --git a/relay-config/src/config.rs b/relay-config/src/config.rs index e57e2e1f9f2..20a58f806ed 100644 --- a/relay-config/src/config.rs +++ b/relay-config/src/config.rs @@ -639,6 +639,8 @@ pub struct Limits { pub max_trace_metric_size: ByteSize, /// The maximum payload size for a log. pub max_log_size: ByteSize, + /// The maximum number of logs that can result from a log expansion. + pub max_expanded_log_count: usize, /// The maximum payload size for a span. pub max_span_size: ByteSize, /// The maximum payload size for an item container. @@ -728,6 +730,7 @@ impl Default for Limits { max_profile_size: ByteSize::mebibytes(50), max_trace_metric_size: ByteSize::mebibytes(1), max_log_size: ByteSize::mebibytes(1), + max_expanded_log_count: 1000, max_span_size: ByteSize::mebibytes(10), max_container_size: ByteSize::mebibytes(12), max_statsd_size: ByteSize::mebibytes(1), @@ -2412,6 +2415,11 @@ impl Config { self.values.limits.max_log_size.as_bytes() } + /// Returns the maximum number of logs that can result from a log expansion. + pub fn max_expanded_log_count(&self) -> usize { + self.values.limits.max_expanded_log_count + } + /// Returns the maximum payload size of a span in bytes. pub fn max_span_size(&self) -> usize { self.values.limits.max_span_size.as_bytes() diff --git a/relay-server/src/processing/logs/integrations/mod.rs b/relay-server/src/processing/logs/integrations/mod.rs index 29c2b070c03..2a8e31ed175 100644 --- a/relay-server/src/processing/logs/integrations/mod.rs +++ b/relay-server/src/processing/logs/integrations/mod.rs @@ -4,7 +4,8 @@ use relay_quotas::DataCategory; use crate::envelope::{ContainerItems, EnvelopeHeaders, Item, WithHeader}; use crate::integrations::{Integration, LogsIntegration}; use crate::managed::RecordKeeper; -use crate::processing::logs::Settings; +use crate::processing::logs::Error::TooManyExpandedLogs; +use crate::processing::logs::{Result, Settings}; mod nel; mod otel; @@ -17,6 +18,7 @@ pub fn expand( item: Item, records: &mut RecordKeeper<'_>, headers: &EnvelopeHeaders, + max_expanded_log_count: usize, ) -> Option<(Settings, ContainerItems)> { let integration = match item.integration() { Some(Integration::Logs(integration)) => integration, @@ -27,7 +29,7 @@ pub fn expand( }; let mut logs = Vec::new(); - let produce = |log: OurLog| { + let produce = |log: OurLog| -> Result<()> { let byte_size = relay_ourlogs::calculate_size(&log); records.modify_by(DataCategory::LogItem, 1); @@ -40,6 +42,12 @@ pub fn expand( }), value: log.into(), }); + + if logs.len() > max_expanded_log_count { + return Err(TooManyExpandedLogs); + } + + Ok(()) }; let payload = item.payload(); diff --git a/relay-server/src/processing/logs/integrations/nel.rs b/relay-server/src/processing/logs/integrations/nel.rs index 6612a48cfcb..3180b8299e8 100644 --- a/relay-server/src/processing/logs/integrations/nel.rs +++ b/relay-server/src/processing/logs/integrations/nel.rs @@ -7,17 +7,20 @@ use crate::processing::logs::{Error, Result, Settings}; use crate::services::outcome::DiscardReason; /// Expands OTeL logs into the [`OurLog`] format. -pub fn expand(payload: &[u8], headers: &EnvelopeHeaders, produce: F) -> Result +pub fn expand(payload: &[u8], headers: &EnvelopeHeaders, mut produce: F) -> Result where - F: FnMut(OurLog), + F: FnMut(OurLog) -> Result<()>, { let received_at = headers.meta().received_at(); - serde_json::from_slice::>(payload) + let filtered = serde_json::from_slice::>(payload) .map_err(|_| Error::Invalid(DiscardReason::InvalidJson))? .into_iter() - .filter_map(|DeserializableAnnotated(nel)| nel::create_log(nel, received_at)) - .for_each(produce); + .filter_map(|DeserializableAnnotated(nel)| nel::create_log(nel, received_at)); + + for log in filtered { + produce(log)?; + } Ok(Settings { infer_user_agent: true, diff --git a/relay-server/src/processing/logs/integrations/otel.rs b/relay-server/src/processing/logs/integrations/otel.rs index 3d51c5b0cb0..1da3cc84a40 100644 --- a/relay-server/src/processing/logs/integrations/otel.rs +++ b/relay-server/src/processing/logs/integrations/otel.rs @@ -9,7 +9,7 @@ use crate::services::outcome::DiscardReason; /// Expands OTeL logs into the [`OurLog`] format. pub fn expand(format: OtelFormat, payload: &[u8], mut produce: F) -> Result where - F: FnMut(OurLog), + F: FnMut(OurLog) -> Result<()>, { let logs = parse_logs_data(format, payload)?; @@ -19,7 +19,7 @@ where let scope = scope_logs.scope.as_ref(); for log_record in scope_logs.log_records { let log = relay_ourlogs::otel_to_sentry_log(log_record, resource, scope); - produce(log); + produce(log)?; } } } diff --git a/relay-server/src/processing/logs/integrations/vercel.rs b/relay-server/src/processing/logs/integrations/vercel.rs index ab332d9e9b9..55ce7e2acc5 100644 --- a/relay-server/src/processing/logs/integrations/vercel.rs +++ b/relay-server/src/processing/logs/integrations/vercel.rs @@ -8,7 +8,7 @@ use crate::services::outcome::DiscardReason; /// Expands Vercel logs into the [`OurLog`] format. pub fn expand(format: VercelLogDrainFormat, payload: &[u8], mut produce: F) -> Result where - F: FnMut(OurLog), + F: FnMut(OurLog) -> Result<()>, { let mut count: i32 = 0; @@ -25,7 +25,7 @@ where for log in logs { count += 1; let ourlog = relay_ourlogs::vercel_log_to_sentry_log(log); - produce(ourlog); + produce(ourlog)?; } } VercelLogDrainFormat::NdJson => { @@ -37,7 +37,7 @@ where if let Ok(log) = serde_json::from_slice::(line) { count += 1; let ourlog = relay_ourlogs::vercel_log_to_sentry_log(log); - produce(ourlog); + produce(ourlog)?; } } } diff --git a/relay-server/src/processing/logs/mod.rs b/relay-server/src/processing/logs/mod.rs index fcc7634b0cd..fc382300fc1 100644 --- a/relay-server/src/processing/logs/mod.rs +++ b/relay-server/src/processing/logs/mod.rs @@ -55,6 +55,9 @@ pub enum Error { /// The log is invalid. #[error("invalid: {0}")] Invalid(DiscardReason), + /// The expanded logs exceed the maximum number allowed. + #[error("expanded logs exeeds limit")] + TooManyExpandedLogs, } impl OutcomeError for Error { @@ -74,6 +77,8 @@ impl OutcomeError for Error { } Self::ProcessingFailed(_) => Some(Outcome::Invalid(DiscardReason::Internal)), Self::Invalid(reason) => Some(Outcome::Invalid(*reason)), + // TODO: Or should this be abuse? Or should this be filtered, or rate-limited? + Self::TooManyExpandedLogs => Some(Outcome::Invalid(DiscardReason::InvalidLog)), }; (outcome, self) @@ -153,7 +158,7 @@ impl processing::Processor for LogsProcessor { // Fast filters, which do not need expanded logs. filter::feature_flag(ctx).reject(&logs)?; - let mut logs = process::expand(logs)?; + let mut logs = process::expand(logs, ctx.config.max_expanded_log_count())?; validate::size(&mut logs, ctx); diff --git a/relay-server/src/processing/logs/process.rs b/relay-server/src/processing/logs/process.rs index 88c891439ef..584c9bac9a3 100644 --- a/relay-server/src/processing/logs/process.rs +++ b/relay-server/src/processing/logs/process.rs @@ -17,7 +17,10 @@ use crate::services::outcome::DiscardReason; /// Parses all serialized logs into their [`ExpandedLogs`] representation. /// /// Individual, invalid logs will be discarded. -pub fn expand(logs: Managed) -> Result, Rejected> { +pub fn expand( + logs: Managed, + max_expanded_log_count: usize, +) -> Result, Rejected> { let trust = logs.headers.meta().request_trust(); logs.try_map(|logs, records| { @@ -44,7 +47,8 @@ pub fn expand(logs: Managed) -> Result, Re let (settings, logs) = match items { LogItems::Container(item) => expand_log_container(&item, trust)?, LogItems::Integration(item) => { - logs::integrations::expand(item, records, &headers).unwrap_or_default() + logs::integrations::expand(item, records, &headers, max_expanded_log_count) + .unwrap_or_default() } };