Streamify log expansion (for discussion, not review)

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()

relay-otel/src/lib.rs

@@ -100,8 +100,8 @@ pub fn otel_value_to_attribute(otel_value: OtelValue) -> Option<Attribute> {
 /// Applies Otel scopes into Sentry [`Attributes`].
 pub fn otel_scope_into_attributes(
     attributes: &mut Attributes,
-    resource: Option<&Resource>,
-    scope: Option<&InstrumentationScope>,
+    resource: &Option<Resource>,
+    scope: &Option<InstrumentationScope>,
 ) {
     for attribute in resource.into_iter().flat_map(|s| &s.attributes) {
         if let Some(attr) = attribute

relay-ourlogs/src/otel_to_sentry.rs

@@ -69,8 +69,8 @@ fn otel_body_to_sentry_body(body: Option<AnyValue>) -> Option<String> {
 /// Transforms an OpenTelemetry log record to a Sentry log.
 pub fn otel_to_sentry_log(
     otel_log: OtelLogRecord,
-    resource: Option<&Resource>,
-    scope: Option<&InstrumentationScope>,
+    resource: &Option<Resource>,
+    scope: &Option<InstrumentationScope>,
 ) -> OurLog {
     let OtelLogRecord {
         time_unix_nano,
@@ -101,7 +101,7 @@ pub fn otel_to_sentry_log(
         attribute_data.insert(EVENT__NAME, event_name.to_owned());
     }
     if let Some(resource) = resource
-        && let Some(platform) = otel_resource_to_platform(resource)
+        && let Some(platform) = otel_resource_to_platform(&resource)
     {
         attribute_data.insert(SENTRY__PLATFORM, platform.to_owned());
     }

relay-server/src/processing/logs/integrations/mod.rs

@@ -4,7 +4,7 @@
 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::{Result, Settings};
 
 mod nel;
 mod otel;
@@ -17,6 +17,7 @@
     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,
@@ -26,37 +27,49 @@
         }
     };
 
-    let mut logs = Vec::new();
-    let produce = |log: OurLog| {
-        let byte_size = relay_ourlogs::calculate_size(&log);
-
-        records.modify_by(DataCategory::LogItem, 1);
-        records.modify_by(DataCategory::LogByte, byte_size as isize);
+    let payload = item.payload();
 
-        logs.push(WithHeader {
-            header: Some(OurLogHeader {
-                byte_size: Some(byte_size),
-                other: Default::default(),
-            }),
-            value: log.into(),
-        });
+    let log_stream: Result<Box<dyn Iterator<Item = OurLog>>> = match integration {
+        LogsIntegration::Nel => nel::expand2(&payload, headers),
+        LogsIntegration::OtelV1 { format } => otel::expand2(format, &payload),
+        LogsIntegration::VercelDrainLog { format } => vercel::expand2(format, &payload),
     };
 
-    let payload = item.payload();
-
     let settings = match integration {
-        LogsIntegration::Nel => nel::expand(&payload, headers, produce),
-        LogsIntegration::OtelV1 { format } => otel::expand(format, &payload, produce),
-        LogsIntegration::VercelDrainLog { format } => vercel::expand(format, &payload, produce),
+        LogsIntegration::Nel => Settings {
+            infer_user_agent: true,
+            infer_ip: false,
+        },
+        LogsIntegration::OtelV1 { format: _ } => Settings::default(),
+        LogsIntegration::VercelDrainLog { format: _ } => Settings::default(),
     };
-    let settings = match settings {
-        Ok(settings) => settings,
+
+    let (log_stream, settings) = match log_stream {
+        Ok(log_stream) => (log_stream, settings),
         Err(err) => {
             let _ = records.reject_err(err, &item);
             return None;
         }
     };
 
+    let logs = log_stream
+        .take(max_expanded_log_count)
+        .map(|log| {
+            let byte_size = relay_ourlogs::calculate_size(&log);
+
+            records.modify_by(DataCategory::LogItem, 1);
+            records.modify_by(DataCategory::LogByte, byte_size as isize);
+
+            WithHeader {
+                header: Some(OurLogHeader {
+                    byte_size: Some(byte_size),
+                    other: Default::default(),
+                }),
+                value: log.into(),
+            }
+        })
+        .collect();
+
     // Undo all the base item quantities, as they will be completely taken over by the parsed
     // contents, which contains an arbitrary amount of items (even 0).
     for (category, quantity) in item.quantities() {

relay-server/src/processing/logs/integrations/nel.rs

@@ -3,24 +3,20 @@ use relay_event_schema::protocol::OurLog;
 use relay_protocol::DeserializableAnnotated;
 
 use crate::envelope::EnvelopeHeaders;
-use crate::processing::logs::{Error, Result, Settings};
+use crate::processing::logs::{Error, Result};
 use crate::services::outcome::DiscardReason;
 
 /// Expands OTeL logs into the [`OurLog`] format.
-pub fn expand<F>(payload: &[u8], headers: &EnvelopeHeaders, produce: F) -> Result<Settings>
-where
-    F: FnMut(OurLog),
-{
+pub fn expand2(
+    payload: &[u8],
+    headers: &EnvelopeHeaders,
+) -> Result<Box<dyn Iterator<Item = OurLog>>> {
     let received_at = headers.meta().received_at();
 
-    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);
-
-    Ok(Settings {
-        infer_user_agent: true,
-        infer_ip: false,
-    })
+    Ok(Box::new(
+        serde_json::from_slice::<Vec<_>>(payload)
+            .map_err(|_| Error::Invalid(DiscardReason::InvalidJson))?
+            .into_iter()
+            .filter_map(move |DeserializableAnnotated(nel)| nel::create_log(nel, received_at)),
+    ))
 }

relay-server/src/processing/logs/integrations/otel.rs

@@ -1,30 +1,33 @@
+use std::ops::Deref;
+
 use opentelemetry_proto::tonic::logs::v1::LogsData;
 use prost::Message as _;
 use relay_event_schema::protocol::OurLog;
 
 use crate::integrations::OtelFormat;
-use crate::processing::logs::{Error, Result, Settings};
+use crate::processing::logs::{Error, Result};
 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),
-{
-    let logs = parse_logs_data(format, payload)?;
-
-    for resource_logs in logs.resource_logs {
-        let resource = resource_logs.resource.as_ref();
-        for scope_logs in resource_logs.scope_logs {
-            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);
-            }
-        }
-    }
+pub fn expand2(format: OtelFormat, payload: &[u8]) -> Result<Box<dyn Iterator<Item = OurLog>>> {
+    let logs: LogsData = parse_logs_data(format, payload).unwrap();
 
-    Ok(Settings::default())
+    Ok(Box::new(logs.resource_logs.into_iter().flat_map(
+        |resource_logs| {
+            let resource = std::cell::RefCell::new(resource_logs.resource);
+            resource_logs
+                .scope_logs
+                .into_iter()
+                .flat_map(move |scope_logs| {
+                    let scope = scope_logs.scope;
+                    let r = resource.clone();
+                    scope_logs.log_records.into_iter().map(move |log_record| {
+                        let b = r.borrow();
+                        relay_ourlogs::otel_to_sentry_log(log_record, b.deref(), &scope)
+                    })
+                })
+        },
+    )))
 }
 
 fn parse_logs_data(format: OtelFormat, payload: &[u8]) -> Result<LogsData, Error> {

relay-server/src/processing/logs/integrations/vercel.rs

@@ -2,51 +2,33 @@ use relay_event_schema::protocol::OurLog;
 use relay_ourlogs::VercelLog;
 
 use crate::integrations::VercelLogDrainFormat;
-use crate::processing::logs::{Error, Result, Settings};
+use crate::processing::logs::{Error, Result};
 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),
-{
-    let mut count: i32 = 0;
-
+pub fn expand2<'a>(
+    format: VercelLogDrainFormat,
+    payload: &'a [u8],
+) -> Result<Box<dyn Iterator<Item = OurLog> + 'a>> {
     match format {
-        VercelLogDrainFormat::Json => {
-            let logs = serde_json::from_slice::<Vec<VercelLog>>(payload).map_err(|e| {
-                relay_log::debug!(
-                    error = &e as &dyn std::error::Error,
-                    "Failed to parse logs data as JSON"
-                );
-                Error::Invalid(DiscardReason::InvalidJson)
-            })?;
-
-            for log in logs {
-                count += 1;
-                let ourlog = relay_ourlogs::vercel_log_to_sentry_log(log);
-                produce(ourlog);
-            }
-        }
-        VercelLogDrainFormat::NdJson => {
-            for line in payload.split(|&b| b == b'\n') {
-                if line.is_empty() {
-                    continue;
-                }
-
-                if let Ok(log) = serde_json::from_slice::<VercelLog>(line) {
-                    count += 1;
-                    let ourlog = relay_ourlogs::vercel_log_to_sentry_log(log);
-                    produce(ourlog);
-                }
-            }
-        }
+        VercelLogDrainFormat::Json => Ok(Box::new(
+            serde_json::from_slice::<Vec<VercelLog>>(payload)
+                .map_err(|e| {
+                    relay_log::debug!(
+                        error = &e as &dyn std::error::Error,
+                        "Failed to parse logs data as JSON"
+                    );
+                    Error::Invalid(DiscardReason::InvalidJson)
+                })?
+                .into_iter()
+                .map(relay_ourlogs::vercel_log_to_sentry_log),
+        )),
+        VercelLogDrainFormat::NdJson => Ok(Box::new(
+            payload
+                .split(|&b| b == b'\n')
+                .filter(|l| !l.is_empty())
+                .flat_map(serde_json::from_slice::<VercelLog>)
+                .map(relay_ourlogs::vercel_log_to_sentry_log),
+        )),
     }
-
-    if count == 0 {
-        relay_log::debug!("Failed to parse any logs from vercel log drain payload");
-        return Err(Error::Invalid(DiscardReason::InvalidJson));
-    }
-
-    Ok(Settings::default())
 }

relay-server/src/processing/logs/mod.rs

@@ -55,6 +55,9 @@
     /// 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 @@
             }
             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 @@
         // 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);
 

relay-server/src/processing/logs/process.rs

@@ -17,7 +17,10 @@
 /// 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| {
@@ -44,7 +47,8 @@
         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()
             }
         };
 

relay-server/src/processing/spans/integrations/otel.rs

@@ -14,11 +14,11 @@ where
     let traces = parse_traces_data(format, payload)?;
 
     for resource_spans in traces.resource_spans {
-        let resource = resource_spans.resource.as_ref();
+        let resource = resource_spans.resource;
         for scope_spans in resource_spans.scope_spans {
-            let scope = scope_spans.scope.as_ref();
+            let scope = scope_spans.scope;
             for span in scope_spans.spans {
-                let span = relay_spans::otel_to_sentry_span_v2(span, resource, scope);
+                let span = relay_spans::otel_to_sentry_span_v2(span, &resource, &scope);
                 produce(span);
             }
         }