Skip to content

ref(relay): Limit maximum number of logs produced by expansion #6263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
klochek wants to merge 1 commit into
base: master
Choose a base branch
from
Open

ref(relay): Limit maximum number of logs produced by expansion #6263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions relay-config/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -638,7 +638,9 @@
/// The maximum payload size for a trace metric.
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.

Check warning on line 642 in relay-config/src/config.rs

View check run for this annotation

@sentry/warden / warden: wrdn-dos-review

`max_expanded_log_count` is only enforced for integrations, not log containers 

The new `max_expanded_log_count` limit only applies to log integrations (OTel, Vercel, NEL) and not to `LogItems::Container`. A container with many small logs bypasses the count cap entirely, causing unbounded per-request CPU and memory usage during normalization, filtering, and scrubbing.
Comment on lines 641 to +642

@sentry-warden sentry-warden Bot Jul 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

max_expanded_log_count is only enforced for integrations, not log containers

The new max_expanded_log_count limit only applies to log integrations (OTel, Vercel, NEL) and not to LogItems::Container. A container with many small logs bypasses the count cap entirely, causing unbounded per-request CPU and memory usage during normalization, filtering, and scrubbing.

Evidence
  • relay-server/src/processing/logs/process.rs:expand() receives max_expanded_log_count but only passes it to integrations::expand() for LogItems::Integration.
  • For LogItems::Container, it calls expand_log_container() with no count parameter, so the cap is never checked.
  • A container is bounded by max_container_size (default 12 MB), but with small logs that still allows tens or hundreds of thousands of items—far above the max_expanded_log_count default of 1000.
  • Every log from the container then goes through validate::size, normalize, filter, scrub, and normalize_derived with no early count-based rejection, unlike the integration path where produce() returns Err(TooManyExpandedLogs) once the cap is exceeded.

Identified by Warden · wrdn-dos-review · MG4-3M7

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.
Expand Down Expand Up @@ -728,6 +730,7 @@
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),
Expand Down Expand Up @@ -2412,6 +2415,11 @@
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()
Expand Down
12 changes: 10 additions & 2 deletions relay-server/src/processing/logs/integrations/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -17,6 +18,7 @@ pub fn expand(
item: Item,
records: &mut RecordKeeper<'_>,
headers: &EnvelopeHeaders,
max_expanded_log_count: usize,
) -> Option<(Settings, ContainerItems<OurLog>)> {
let integration = match item.integration() {
Some(Integration::Logs(integration)) => integration,
Expand All @@ -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);
Expand All @@ -40,6 +42,12 @@ pub fn expand(
}),
value: log.into(),
});

if logs.len() > max_expanded_log_count {
return Err(TooManyExpandedLogs);
}
Comment on lines +46 to +48

@Dav1dde Dav1dde Jul 28, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this check isn't enough, as we're already deserializing (either via serde_json or prost) the entire array.

@klochek klochek Jul 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So, it's the entire array, but we've only expanded (with accompanying attributes) logs.len() full log entries. It's the amplification by the attributes that can make this size a lot larger than we'd like.

@Dav1dde Dav1dde Jul 28, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Parsing is already too much. The amplification of attributes is just a bigger multiplier. The OurLog struct is 208 bytes, the minimum size for a valid log item is 2 bytes in JSON (3 if we count the ,, 2 in protobuf). That is a 100x memory multiplier without considering the attribute amplification. With a size limit of 12 MiB per container, that means in a parsed state this is ~1 GiB. Then we have additional overheads like WithHeader<OurLog> and over-allocation of the Vec (capacity). To fully address the issue we already need to limit the count during parsing.


Ok(())

@cursor cursor Bot Jul 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Debug quantity mismatch on limit

Medium Severity

When expansion hits TooManyExpandedLogs, produce has already called records.modify_by for each log emitted so far, but expand returns None and process::expand still finishes try_map with empty logs via unwrap_or_default. Debug RecordKeeper::success then fails quantity balancing and panics.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 409e55f. Configure here.

};

let payload = item.payload();
Expand Down
13 changes: 8 additions & 5 deletions relay-server/src/processing/logs/integrations/nel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<F>(payload: &[u8], headers: &EnvelopeHeaders, produce: F) -> Result<Settings>
pub fn expand<F>(payload: &[u8], headers: &EnvelopeHeaders, mut produce: F) -> Result<Settings>
where
F: FnMut(OurLog),
F: FnMut(OurLog) -> Result<()>,
{
let received_at = headers.meta().received_at();

serde_json::from_slice::<Vec<_>>(payload)
let filtered = serde_json::from_slice::<Vec<_>>(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,
Expand Down
4 changes: 2 additions & 2 deletions relay-server/src/processing/logs/integrations/otel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ use crate::services::outcome::DiscardReason;
/// Expands OTeL logs into the [`OurLog`] format.
pub fn expand<F>(format: OtelFormat, payload: &[u8], mut produce: F) -> Result<Settings>
where
F: FnMut(OurLog),
F: FnMut(OurLog) -> Result<()>,
{
let logs = parse_logs_data(format, payload)?;

Expand All @@ -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)?;
}
}
}
Expand Down
6 changes: 3 additions & 3 deletions relay-server/src/processing/logs/integrations/vercel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ use crate::services::outcome::DiscardReason;
/// Expands Vercel logs into the [`OurLog`] format.
pub fn expand<F>(format: VercelLogDrainFormat, payload: &[u8], mut produce: F) -> Result<Settings>
where
F: FnMut(OurLog),
F: FnMut(OurLog) -> Result<()>,
{
let mut count: i32 = 0;

Expand All @@ -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 => {
Expand All @@ -37,7 +37,7 @@ where
if let Ok(log) = serde_json::from_slice::<VercelLog>(line) {
count += 1;
let ourlog = relay_ourlogs::vercel_log_to_sentry_log(log);
produce(ourlog);
produce(ourlog)?;
}
}
}
Expand Down
7 changes: 6 additions & 1 deletion relay-server/src/processing/logs/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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)
Expand Down Expand Up @@ -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);

Expand Down
8 changes: 6 additions & 2 deletions relay-server/src/processing/logs/process.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<SerializedLogs>) -> Result<Managed<ExpandedLogs>, Rejected<Error>> {
pub fn expand(
logs: Managed<SerializedLogs>,
max_expanded_log_count: usize,
) -> Result<Managed<ExpandedLogs>, Rejected<Error>> {
let trust = logs.headers.meta().request_trust();

logs.try_map(|logs, records| {
Expand All @@ -44,7 +47,8 @@ pub fn expand(logs: Managed<SerializedLogs>) -> Result<Managed<ExpandedLogs>, 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()
}
};

Expand Down
Loading