Streamify log expansion (for discussion, not review)

[klochek]

No description provided.

[sentry-warden]

OTel log expansion panics on malformed input instead of returning an error

otel::expand2 unwraps the parsing result of attacker-controlled payload bytes, causing a panic on invalid input instead of returning a rejection error.

Evidence

• The hunk adds otel::expand2(format, &payload) at mod.rs:32, feeding attacker-controlled payload bytes into the new function.
• otel::expand2 in relay-server/src/processing/logs/integrations/otel.rs:15 calls parse_logs_data(format, payload).unwrap().
• parse_logs_data correctly returns Result<LogsData, Error>, propagating JSON and protobuf parse failures.
• The .unwrap() converts every parse failure into an uncatchable panic, crashing the worker thread instead of returning a DiscardReason rejection outcome to the caller.
• Sibling expand2 functions in nel.rs and vercel.rs properly propagate errors via ? and map_err, confirming the omission is unintended.

Also found at 2 additional locations

• relay-server/src/processing/logs/integrations/otel.rs:13
• relay-server/src/processing/logs/process.rs:50-51

_{Identified by Warden · wrdn-dos-review · RMP-KHP}

[sentry-warden]

max_expanded_log_count cap is post-materialization for integrations and skipped for containers

The max_expanded_log_count limit passed to process::expand is not enforced before integration payloads are fully parsed into memory, and the container expansion path ignores it entirely.

Evidence

• mod.rs:161 passes ctx.config.max_expanded_log_count() into process::expand.
• In process.rs, the LogItems::Container branch calls expand_log_container(&item, trust) without the cap, so all container logs are parsed into memory and returned.
• In process.rs, the LogItems::Integration branch passes the cap to integrations::expand(), which applies .take(max_expanded_log_count) at integrations/mod.rs:68.
• nel::expand2 eagerly parses the full payload with serde_json::from_slice::<Vec<_>>(payload) before returning an iterator.
• vercel::expand2 (JSON path) eagerly parses with serde_json::from_slice::<Vec<VercelLog>>(payload) before returning an iterator.
• otel::expand2 eagerly parses the full payload via parse_logs_data (serde_json::from_slice or LogsData::decode) before returning an iterator.
• Because the cap via .take() operates on an iterator over already-materialized data, the memory and parsing cost for the full integration payload is paid before the limit takes effect.

_{Identified by Warden · wrdn-dos-review · HQH-YVE}

[Dav1dde]

I wonder if we can derive this from the total size, so we don't have to derive a count when we already have a size.

[Dav1dde]

This could also be streamed, then the question arises if the stream shouldn't be consumed here?

[Dav1dde]

I don't think this RefCell is necessary, there shouldn't be a problem with passing in a & to the nested closures.

This entire stream may be easier expressed if you build up a stream of (resource, scope, log) items via something like .flat_map(repeat(resource).zip(scopes)) then convert them.

[klochek]

It's because of all the nesting--that resource_logs from the very-first flat_map is only alive long enough to return the inner iterator, and so it really does need to be owned by the inner closures, but the only way to do that (and avoid the cloning) is to stick it in a ref-cell