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-otel/src/lib.rs b/relay-otel/src/lib.rs index 758774140d7..b813ca7e70c 100644 --- a/relay-otel/src/lib.rs +++ b/relay-otel/src/lib.rs @@ -100,8 +100,8 @@ pub fn otel_value_to_attribute(otel_value: OtelValue) -> Option { /// Applies Otel scopes into Sentry [`Attributes`]. pub fn otel_scope_into_attributes( attributes: &mut Attributes, - resource: Option<&Resource>, - scope: Option<&InstrumentationScope>, + resource: &Option, + scope: &Option, ) { for attribute in resource.into_iter().flat_map(|s| &s.attributes) { if let Some(attr) = attribute diff --git a/relay-ourlogs/src/otel_to_sentry.rs b/relay-ourlogs/src/otel_to_sentry.rs index cf306d86810..b45223e66ef 100644 --- a/relay-ourlogs/src/otel_to_sentry.rs +++ b/relay-ourlogs/src/otel_to_sentry.rs @@ -69,8 +69,8 @@ fn otel_body_to_sentry_body(body: Option) -> Option { /// 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, + scope: &Option, ) -> 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()); } diff --git a/relay-server/src/processing/logs/integrations/mod.rs b/relay-server/src/processing/logs/integrations/mod.rs index 29c2b070c03..29260816d4e 100644 --- a/relay-server/src/processing/logs/integrations/mod.rs +++ b/relay-server/src/processing/logs/integrations/mod.rs @@ -4,7 +4,7 @@ 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::{Result, Settings}; mod nel; mod otel; @@ -17,6 +17,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, @@ -26,37 +27,49 @@ pub fn expand( } }; - 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>> = 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() { diff --git a/relay-server/src/processing/logs/integrations/nel.rs b/relay-server/src/processing/logs/integrations/nel.rs index 6612a48cfcb..81b17046000 100644 --- a/relay-server/src/processing/logs/integrations/nel.rs +++ b/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(payload: &[u8], headers: &EnvelopeHeaders, produce: F) -> Result -where - F: FnMut(OurLog), -{ +pub fn expand2( + payload: &[u8], + headers: &EnvelopeHeaders, +) -> Result>> { let received_at = headers.meta().received_at(); - 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); - - Ok(Settings { - infer_user_agent: true, - infer_ip: false, - }) + Ok(Box::new( + serde_json::from_slice::>(payload) + .map_err(|_| Error::Invalid(DiscardReason::InvalidJson))? + .into_iter() + .filter_map(move |DeserializableAnnotated(nel)| nel::create_log(nel, received_at)), + )) } diff --git a/relay-server/src/processing/logs/integrations/otel.rs b/relay-server/src/processing/logs/integrations/otel.rs index 3d51c5b0cb0..03a178d1e35 100644 --- a/relay-server/src/processing/logs/integrations/otel.rs +++ b/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(format: OtelFormat, payload: &[u8], mut produce: F) -> Result -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>> { + 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 { diff --git a/relay-server/src/processing/logs/integrations/vercel.rs b/relay-server/src/processing/logs/integrations/vercel.rs index ab332d9e9b9..4bdda3b90f2 100644 --- a/relay-server/src/processing/logs/integrations/vercel.rs +++ b/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(format: VercelLogDrainFormat, payload: &[u8], mut produce: F) -> Result -where - F: FnMut(OurLog), -{ - let mut count: i32 = 0; - +pub fn expand2<'a>( + format: VercelLogDrainFormat, + payload: &'a [u8], +) -> Result + 'a>> { match format { - VercelLogDrainFormat::Json => { - let logs = serde_json::from_slice::>(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::(line) { - count += 1; - let ourlog = relay_ourlogs::vercel_log_to_sentry_log(log); - produce(ourlog); - } - } - } + VercelLogDrainFormat::Json => Ok(Box::new( + serde_json::from_slice::>(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::) + .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()) } diff --git a/relay-server/src/processing/logs/mod.rs b/relay-server/src/processing/logs/mod.rs index fcc7634b0cd..953a6e22790 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() } }; diff --git a/relay-server/src/processing/spans/integrations/otel.rs b/relay-server/src/processing/spans/integrations/otel.rs index 58ace372622..3195abedb53 100644 --- a/relay-server/src/processing/spans/integrations/otel.rs +++ b/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); } } diff --git a/relay-spans/src/otel_to_sentry_v2.rs b/relay-spans/src/otel_to_sentry_v2.rs index c9555144397..ffbd6e24643 100644 --- a/relay-spans/src/otel_to_sentry_v2.rs +++ b/relay-spans/src/otel_to_sentry_v2.rs @@ -31,8 +31,8 @@ use relay_protocol::{Annotated, Error, Value}; /// All other attributes are carried over from the OTEL span to the Sentry span. pub fn otel_to_sentry_span( otel_span: OtelSpan, - resource: Option<&Resource>, - scope: Option<&InstrumentationScope>, + resource: &Option, + scope: &Option, ) -> SentrySpanV2 { let OtelSpan { trace_id, @@ -66,11 +66,11 @@ pub fn otel_to_sentry_span( let mut sentry_attributes = Attributes::new(); - relay_otel::otel_scope_into_attributes(&mut sentry_attributes, resource, scope); + relay_otel::otel_scope_into_attributes(&mut sentry_attributes, &resource, &scope); sentry_attributes.insert(SENTRY__ORIGIN, "auto.otlp.spans".to_owned()); - if let Some(resource) = resource - && let Some(platform) = otel_resource_to_platform(resource) + if let Some(resource) = &resource + && let Some(platform) = otel_resource_to_platform(&resource) { sentry_attributes.insert(SENTRY__PLATFORM, platform.to_owned()); }