diff --git a/opentelemetry-sdk/CHANGELOG.md b/opentelemetry-sdk/CHANGELOG.md index 1c9f434c22..8578f74450 100644 --- a/opentelemetry-sdk/CHANGELOG.md +++ b/opentelemetry-sdk/CHANGELOG.md @@ -2,6 +2,14 @@ ## vNext +- `SdkLoggerProvider` now enforces a maximum attribute count per log record + (default `128`). Attributes beyond the limit are rejected eagerly in + `add_attribute`, so a misbehaving caller cannot grow a record's memory past + the limit, and the number of rejected attributes is tracked and exposed via + `SdkLogRecord::dropped_attributes_count()`. The limit is configurable via + `LoggerProviderBuilder::with_max_attributes_per_log` or the + `OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT` environment variable, which falls back + to `OTEL_ATTRIBUTE_COUNT_LIMIT`. - Bound instruments are now available for `Gauge` and `UpDownCounter` via the new `BoundGauge` and `BoundUpDownCounter` types exposed by the `opentelemetry` crate. Requires the `experimental_metrics_bound_instruments` diff --git a/opentelemetry-sdk/src/lib.rs b/opentelemetry-sdk/src/lib.rs index 42348d9e3b..6368953c3c 100644 --- a/opentelemetry-sdk/src/lib.rs +++ b/opentelemetry-sdk/src/lib.rs @@ -111,6 +111,12 @@ //! | `OTEL_BSP_MAX_EXPORT_BATCH_SIZE` | Maximum batch size. Must be less than or equal to `OTEL_BSP_MAX_QUEUE_SIZE`. | `512` | //! | `OTEL_BSP_MAX_CONCURRENT_EXPORTS` | Maximum number of concurrent exports. Honored by `span_processor_with_async_runtime::BatchSpanProcessor`; thread-based `BatchSpanProcessor` exports serially. For concurrent exports, enable `experimental_trace_batch_span_processor_with_async_runtime` and use the async-runtime processor. | `1` | //! +//! ### Logs: Log Record Limits +//! +//! | Variable | Description | Default | +//! |---|---|---| +//! | `OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT` | Maximum number of attributes allowed on a log record. Falls back to `OTEL_ATTRIBUTE_COUNT_LIMIT` when unset. | `128` | +//! //! ### Logs: Batch Log Record Processor (BLRP) //! //! | Variable | Description | Default | diff --git a/opentelemetry-sdk/src/logs/logger.rs b/opentelemetry-sdk/src/logs/logger.rs index b2c789c175..9bbd72955b 100644 --- a/opentelemetry-sdk/src/logs/logger.rs +++ b/opentelemetry-sdk/src/logs/logger.rs @@ -27,7 +27,9 @@ impl opentelemetry::logs::Logger for SdkLogger { type LogRecord = SdkLogRecord; fn create_log_record(&self) -> Self::LogRecord { - SdkLogRecord::new() + // Stamp the record with the configured attribute limit so that + // `add_attribute` can enforce it eagerly as attributes are added. + SdkLogRecord::new_with_limit(self.provider.max_attributes_per_log()) } /// Emit a `LogRecord`. diff --git a/opentelemetry-sdk/src/logs/logger_provider.rs b/opentelemetry-sdk/src/logs/logger_provider.rs index e8765b456b..7653e87a4b 100644 --- a/opentelemetry-sdk/src/logs/logger_provider.rs +++ b/opentelemetry-sdk/src/logs/logger_provider.rs @@ -1,3 +1,4 @@ +use super::record::DEFAULT_MAX_ATTRIBUTES_PER_LOG; use super::{BatchLogProcessor, LogProcessor, SdkLogger, SimpleLogProcessor}; use crate::error::{OTelSdkError, OTelSdkResult}; use crate::logs::LogExporter; @@ -22,10 +23,18 @@ fn noop_logger_provider() -> &'static SdkLoggerProvider { inner: Arc::new(LoggerProviderInner { processors: Vec::new(), is_shutdown: AtomicBool::new(true), + max_attributes_per_log: DEFAULT_MAX_ATTRIBUTES_PER_LOG, }), }) } +#[inline] +fn read_u32_env(key: &str) -> Option { + std::env::var(key) + .ok() + .and_then(|value| value.parse::().ok()) +} + #[derive(Debug, Clone)] /// Handles the creation and coordination of [`Logger`]s. /// @@ -82,6 +91,12 @@ impl SdkLoggerProvider { &self.inner.processors } + /// Maximum number of attributes retained per log record. Attributes beyond + /// this limit are dropped when the record is emitted. + pub(crate) fn max_attributes_per_log(&self) -> u32 { + self.inner.max_attributes_per_log + } + /// Force flush all remaining logs in log processors and return results. pub fn force_flush(&self) -> OTelSdkResult { let result: Vec<_> = self @@ -135,6 +150,7 @@ impl SdkLoggerProvider { struct LoggerProviderInner { processors: Vec>, is_shutdown: AtomicBool, + max_attributes_per_log: u32, } impl LoggerProviderInner { @@ -184,6 +200,7 @@ impl Drop for LoggerProviderInner { pub struct LoggerProviderBuilder { processors: Vec>, resource: Option, + max_attributes_per_log: Option, } impl LoggerProviderBuilder { @@ -287,6 +304,20 @@ impl LoggerProviderBuilder { LoggerProviderBuilder { resource, ..self } } + /// Specify the maximum number of attributes retained per log record. + /// + /// Attributes added beyond this limit are dropped when the record is + /// emitted. When not set, this is read from the + /// `OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT` environment variable, which falls + /// back to `OTEL_ATTRIBUTE_COUNT_LIMIT`, and otherwise defaults to `128`. + /// A value set here takes precedence over the environment variables. + pub fn with_max_attributes_per_log(self, max_attributes: u32) -> Self { + LoggerProviderBuilder { + max_attributes_per_log: Some(max_attributes), + ..self + } + } + /// Create a new provider from this configuration. pub fn build(self) -> SdkLoggerProvider { let resource = self.resource.unwrap_or(Resource::builder().build()); @@ -295,10 +326,20 @@ impl LoggerProviderBuilder { processor.set_resource(&resource); } + // Precedence: explicit builder value, then + // `OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT`, then the general + // `OTEL_ATTRIBUTE_COUNT_LIMIT`, then the default. + let max_attributes_per_log = self + .max_attributes_per_log + .or_else(|| read_u32_env("OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT")) + .or_else(|| read_u32_env("OTEL_ATTRIBUTE_COUNT_LIMIT")) + .unwrap_or(DEFAULT_MAX_ATTRIBUTES_PER_LOG); + let logger_provider = SdkLoggerProvider { inner: Arc::new(LoggerProviderInner { processors, is_shutdown: AtomicBool::new(false), + max_attributes_per_log, }), }; @@ -793,6 +834,7 @@ mod tests { flush_called.clone(), ))], is_shutdown: AtomicBool::new(false), + max_attributes_per_log: DEFAULT_MAX_ATTRIBUTES_PER_LOG, }); { @@ -833,6 +875,7 @@ mod tests { flush_called.clone(), ))], is_shutdown: AtomicBool::new(false), + max_attributes_per_log: DEFAULT_MAX_ATTRIBUTES_PER_LOG, }); // Create a scope to test behavior when providers are dropped @@ -1007,4 +1050,138 @@ mod tests { Ok(()) } } + + fn emit_record_with_attributes(provider: &SdkLoggerProvider, count: usize) { + let logger = provider.logger("test"); + let mut record = logger.create_log_record(); + for i in 0..count { + record.add_attribute(Key::new(format!("key{i}")), AnyValue::Int(i as i64)); + } + logger.emit(record); + provider.force_flush().unwrap(); + } + + #[test] + fn log_record_attributes_capped_at_default_limit() { + let exporter = InMemoryLogExporter::default(); + let provider = SdkLoggerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + assert_eq!( + provider.max_attributes_per_log(), + DEFAULT_MAX_ATTRIBUTES_PER_LOG + ); + + emit_record_with_attributes(&provider, 200); + + let emitted = exporter.get_emitted_logs().unwrap(); + assert_eq!( + emitted[0].record.attributes_iter().count(), + DEFAULT_MAX_ATTRIBUTES_PER_LOG as usize + ); + assert_eq!( + emitted[0].record.dropped_attributes_count(), + 200 - DEFAULT_MAX_ATTRIBUTES_PER_LOG + ); + } + + #[test] + fn log_record_attributes_within_limit_are_kept() { + let exporter = InMemoryLogExporter::default(); + let provider = SdkLoggerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + + emit_record_with_attributes(&provider, 10); + + let emitted = exporter.get_emitted_logs().unwrap(); + assert_eq!(emitted[0].record.attributes_iter().count(), 10); + assert_eq!(emitted[0].record.dropped_attributes_count(), 0); + } + + #[test] + fn builder_max_attributes_per_log_takes_precedence() { + temp_env::with_vars( + [ + ("OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT", Some("16")), + ("OTEL_ATTRIBUTE_COUNT_LIMIT", Some("8")), + ], + || { + let exporter = InMemoryLogExporter::default(); + let provider = SdkLoggerProvider::builder() + .with_simple_exporter(exporter.clone()) + .with_max_attributes_per_log(4) + .build(); + assert_eq!(provider.max_attributes_per_log(), 4); + + emit_record_with_attributes(&provider, 10); + let emitted = exporter.get_emitted_logs().unwrap(); + assert_eq!(emitted[0].record.attributes_iter().count(), 4); + assert_eq!(emitted[0].record.dropped_attributes_count(), 6); + }, + ); + } + + #[test] + fn logrecord_env_var_takes_precedence_over_general() { + temp_env::with_vars( + [ + ("OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT", Some("16")), + ("OTEL_ATTRIBUTE_COUNT_LIMIT", Some("8")), + ], + || { + let provider = SdkLoggerProvider::builder().build(); + assert_eq!(provider.max_attributes_per_log(), 16); + }, + ); + } + + #[test] + fn general_env_var_is_fallback() { + temp_env::with_vars( + [ + ("OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT", None::<&str>), + ("OTEL_ATTRIBUTE_COUNT_LIMIT", Some("8")), + ], + || { + let provider = SdkLoggerProvider::builder().build(); + assert_eq!(provider.max_attributes_per_log(), 8); + }, + ); + } + + #[test] + fn invalid_env_var_falls_back_to_default() { + temp_env::with_vars( + [ + ("OTEL_LOGRECORD_ATTRIBUTE_COUNT_LIMIT", Some("not-a-number")), + ("OTEL_ATTRIBUTE_COUNT_LIMIT", None::<&str>), + ], + || { + let provider = SdkLoggerProvider::builder().build(); + assert_eq!( + provider.max_attributes_per_log(), + DEFAULT_MAX_ATTRIBUTES_PER_LOG + ); + }, + ); + } + + #[test] + fn noop_logger_after_shutdown_uses_default_limit() { + let provider = SdkLoggerProvider::builder().build(); + provider.shutdown().unwrap(); + + // After shutdown, new loggers refer to the no-op provider, which is + // configured with the default attribute limit. + let logger = provider.logger("after-shutdown"); + let mut record = logger.create_log_record(); + for i in 0..(DEFAULT_MAX_ATTRIBUTES_PER_LOG as usize + 10) { + record.add_attribute(Key::new(format!("k{i}")), AnyValue::Int(i as i64)); + } + assert_eq!( + record.attributes_iter().count(), + DEFAULT_MAX_ATTRIBUTES_PER_LOG as usize + ); + } } diff --git a/opentelemetry-sdk/src/logs/record.rs b/opentelemetry-sdk/src/logs/record.rs index 746e0f5b9b..9e7c44fbf4 100644 --- a/opentelemetry-sdk/src/logs/record.rs +++ b/opentelemetry-sdk/src/logs/record.rs @@ -11,6 +11,9 @@ use std::{borrow::Cow, time::SystemTime}; // up to 5 attributes is the most common case. const PREALLOCATED_ATTRIBUTE_CAPACITY: usize = 5; +/// Default maximum number of attributes retained per log record. +pub(crate) const DEFAULT_MAX_ATTRIBUTES_PER_LOG: u32 = 128; + /// Represents a collection of log record attributes with a predefined capacity. /// /// This type uses `GrowableArray` to store key-value pairs of log attributes, where each attribute is an `Option<(Key, AnyValue)>`. @@ -19,7 +22,7 @@ const PREALLOCATED_ATTRIBUTE_CAPACITY: usize = 5; pub(crate) type LogRecordAttributes = GrowableArray, PREALLOCATED_ATTRIBUTE_CAPACITY>; -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone)] #[non_exhaustive] /// LogRecord represents all data carried by a log record, and /// is provided to `LogExporter`s as input. @@ -50,6 +53,32 @@ pub struct SdkLogRecord { /// Additional attributes associated with this record pub(crate) attributes: LogRecordAttributes, + + /// Number of attributes dropped because the attribute count limit was + /// reached. + pub(crate) dropped_attributes_count: u32, + + /// Maximum number of attributes retained on this record. Attributes added + /// beyond this limit are dropped and counted in `dropped_attributes_count`. + /// This is configuration, not exported data, and is excluded from equality. + pub(crate) max_attributes: u32, +} + +impl PartialEq for SdkLogRecord { + fn eq(&self, other: &Self) -> bool { + // `max_attributes` is build-time configuration rather than data carried + // by the record, so it is intentionally excluded from equality. + self.event_name == other.event_name + && self.target == other.target + && self.timestamp == other.timestamp + && self.observed_timestamp == other.observed_timestamp + && self.trace_context == other.trace_context + && self.severity_text == other.severity_text + && self.severity_number == other.severity_number + && self.body == other.body + && self.attributes == other.attributes + && self.dropped_attributes_count == other.dropped_attributes_count + } } impl opentelemetry::logs::LogRecord for SdkLogRecord { @@ -101,7 +130,15 @@ impl opentelemetry::logs::LogRecord for SdkLogRecord { K: Into, V: Into, { - self.attributes.push(Some((key.into(), value.into()))); + // Enforce the attribute count limit eagerly so that a misbehaving + // caller adding an unbounded number of attributes cannot grow the + // record's memory beyond the configured limit. Attributes past the + // limit are dropped and counted. + if self.attributes.len() < self.max_attributes as usize { + self.attributes.push(Some((key.into(), value.into()))); + } else { + self.dropped_attributes_count += 1; + } } fn set_trace_context( @@ -119,8 +156,15 @@ impl opentelemetry::logs::LogRecord for SdkLogRecord { } impl SdkLogRecord { - /// Crate only default constructor + /// Crate only default constructor, using the default attribute limit. + #[cfg(test)] pub(crate) fn new() -> Self { + Self::new_with_limit(DEFAULT_MAX_ATTRIBUTES_PER_LOG) + } + + /// Crate only constructor that sets the maximum number of attributes + /// retained on the record. + pub(crate) fn new_with_limit(max_attributes: u32) -> Self { SdkLogRecord { event_name: None, target: None, @@ -131,9 +175,18 @@ impl SdkLogRecord { severity_number: None, body: None, attributes: LogRecordAttributes::default(), + dropped_attributes_count: 0, + max_attributes, } } + /// Returns the number of attributes dropped because the attribute count + /// limit was reached. + #[inline] + pub fn dropped_attributes_count(&self) -> u32 { + self.dropped_attributes_count + } + /// Returns the event name #[inline] pub fn event_name(&self) -> Option<&'static str> { @@ -344,6 +397,8 @@ mod tests { span_id: SpanId::from(1), trace_flags: Some(TraceFlags::default()), }), + dropped_attributes_count: 0, + max_attributes: DEFAULT_MAX_ATTRIBUTES_PER_LOG, }; log_record.add_attribute(Key::new("key"), AnyValue::String("value".into())); @@ -357,6 +412,32 @@ mod tests { assert_ne!(log_record, log_record_different); } + #[test] + fn add_attribute_enforces_limit_and_counts_drops() { + let mut log_record = SdkLogRecord::new_with_limit(3); + for i in 0..10 { + log_record.add_attribute(Key::new(format!("key{i}")), AnyValue::Int(i)); + } + // Only the first 3 attributes are retained; the storage never grows + // past the limit. + assert_eq!(log_record.attributes_iter().count(), 3); + assert_eq!(log_record.dropped_attributes_count(), 7); + // The retained attributes are the first ones added. + assert!(log_record.attributes_contains(&Key::new("key0"), &AnyValue::Int(0))); + assert!(!log_record.attributes_contains(&Key::new("key3"), &AnyValue::Int(3))); + } + + #[test] + fn add_attribute_excludes_max_attributes_from_equality() { + let mut a = SdkLogRecord::new_with_limit(5); + let mut b = SdkLogRecord::new_with_limit(500); + a.add_attribute(Key::new("k"), AnyValue::Int(1)); + b.add_attribute(Key::new("k"), AnyValue::Int(1)); + // Differing limits alone must not make otherwise-identical records + // unequal. + assert_eq!(a, b); + } + #[test] fn compare_log_record_target_borrowed_eq_owned() { let log_record_borrowed = SdkLogRecord {