diff --git a/opentelemetry-sdk/CHANGELOG.md b/opentelemetry-sdk/CHANGELOG.md index 4f91ae7f75..7ccd490475 100644 --- a/opentelemetry-sdk/CHANGELOG.md +++ b/opentelemetry-sdk/CHANGELOG.md @@ -2,6 +2,24 @@ ## vNext +- **Breaking behavioral change:** `SdkLogRecord::add_attribute` now deduplicates + attribute keys by default (last-write-wins), so exported log records conform to + the OpenTelemetry specification requirement that attributes form a map of unique + keys. This changes observable output for any code that previously called + `add_attribute` with the same key more than once. + + **Migration:** If you relied on the previous push-only behavior (e.g. for + performance reasons or because downstream consumers tolerated duplicate keys), + opt out via the provider builder: + + ```rust + let provider = SdkLoggerProvider::builder() + .with_log_record_attribute_deduplication(false) + .build(); + ``` + + Fixes [#3497]. + ## 0.32.1 Released 2026-May-23 diff --git a/opentelemetry-sdk/Cargo.toml b/opentelemetry-sdk/Cargo.toml index 1a10758afa..b0579cb73f 100644 --- a/opentelemetry-sdk/Cargo.toml +++ b/opentelemetry-sdk/Cargo.toml @@ -124,6 +124,11 @@ name = "log" harness = false required-features = ["logs"] +[[bench]] +name = "log_record_attribute_dedup" +harness = false +required-features = ["logs"] + [[bench]] name = "bound_instruments" harness = false diff --git a/opentelemetry-sdk/benches/log_record_attribute_dedup.rs b/opentelemetry-sdk/benches/log_record_attribute_dedup.rs new file mode 100644 index 0000000000..156423f9b7 --- /dev/null +++ b/opentelemetry-sdk/benches/log_record_attribute_dedup.rs @@ -0,0 +1,118 @@ +//! Benchmarks for `SdkLogRecord::add_attribute` deduplication overhead. +//! +//! Run with: +//! `cargo bench --bench log_record_attribute_dedup --features logs` +//! +//! Apple M4 Max, macOS 25.3.0 (2026-06-05) — median of 3 independent runs +//! +//! Results (unique keys — typical case, no duplicates in the batch): +//! | Scenario | Dedup ON | Dedup OFF (before) | Overhead | +//! |-------------------------|-----------|--------------------|----------| +//! | add_1_unique_attribute | 25.3 ns | 25.5 ns | ~1.0x | +//! | add_5_unique_attributes | 130.7 ns | 122.9 ns | ~1.06x | +//! | add_9_unique_attributes | 296.9 ns | 220.3 ns | ~1.35x | +//! +//! Results (repeated key — duplicate writes to the same key): +//! | Scenario | Dedup ON | Dedup OFF (before) | +//! |----------------|----------|--------------------| +//! | add_5_same_key | 32.0 ns | 32.1 ns | +//! +//! Note: criterion does not fail CI on regression by itself. These numbers are +//! reference values for human review in PR #3537 / issue #3497. + +use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion}; +use opentelemetry::logs::{AnyValue, LogRecord as _, Logger, LoggerProvider}; +use opentelemetry::Key; +use opentelemetry_sdk::logs::SdkLoggerProvider; + +struct BenchLoggers { + dedup_on: ::Logger, + dedup_off: ::Logger, +} + +impl BenchLoggers { + fn new() -> Self { + let dedup_on = SdkLoggerProvider::builder().build().logger("bench"); + let dedup_off = SdkLoggerProvider::builder() + .with_log_record_attribute_deduplication(false) + .build() + .logger("bench"); + Self { + dedup_on, + dedup_off, + } + } +} + +fn bench_add_unique_attributes(c: &mut Criterion) { + let loggers = BenchLoggers::new(); + let mut group = c.benchmark_group("LogRecord_AddUniqueAttributes"); + + for count in [1usize, 5, 9] { + let keys: Vec = (0..count).map(|i| Key::new(format!("key{i}"))).collect(); + + group.bench_with_input(BenchmarkId::new("dedup_on", count), &keys, |b, keys| { + b.iter(|| { + let mut record = loggers.dedup_on.create_log_record(); + for (i, key) in keys.iter().enumerate() { + record.add_attribute(key.clone(), AnyValue::Int(i as i64)); + } + black_box(record.attributes_iter().count()); + }); + }); + + group.bench_with_input(BenchmarkId::new("dedup_off", count), &keys, |b, keys| { + b.iter(|| { + let mut record = loggers.dedup_off.create_log_record(); + for (i, key) in keys.iter().enumerate() { + record.add_attribute(key.clone(), AnyValue::Int(i as i64)); + } + black_box(record.attributes_iter().count()); + }); + }); + } + + group.finish(); +} + +fn bench_add_repeated_key(c: &mut Criterion) { + let loggers = BenchLoggers::new(); + let mut group = c.benchmark_group("LogRecord_AddRepeatedKey"); + let key = Key::new("key"); + + group.bench_function("dedup_on_5_writes", |b| { + b.iter(|| { + let mut record = loggers.dedup_on.create_log_record(); + for i in 0..5 { + record.add_attribute(key.clone(), AnyValue::Int(i)); + } + black_box(record.attributes_iter().count()); + }); + }); + + group.bench_function("dedup_off_5_writes", |b| { + b.iter(|| { + let mut record = loggers.dedup_off.create_log_record(); + for i in 0..5 { + record.add_attribute(key.clone(), AnyValue::Int(i)); + } + black_box(record.attributes_iter().count()); + }); + }); + + group.finish(); +} + +fn criterion_benchmark(c: &mut Criterion) { + bench_add_unique_attributes(c); + bench_add_repeated_key(c); +} + +criterion_group! { + name = benches; + config = Criterion::default() + .warm_up_time(std::time::Duration::from_secs(1)) + .measurement_time(std::time::Duration::from_secs(2)); + targets = criterion_benchmark +} +criterion_main!(benches); diff --git a/opentelemetry-sdk/src/growable_array.rs b/opentelemetry-sdk/src/growable_array.rs index 525f44e949..f5611422cd 100644 --- a/opentelemetry-sdk/src/growable_array.rs +++ b/opentelemetry-sdk/src/growable_array.rs @@ -80,6 +80,21 @@ impl< } } + /// Gets a mutable reference to the value at the specified index. + /// + /// Returns `None` if the index is out of bounds. + #[allow(dead_code)] + #[inline] + pub(crate) fn get_mut(&mut self, index: usize) -> Option<&mut T> { + if index < self.count { + Some(&mut self.inline[index]) + } else if let Some(ref mut overflow) = self.overflow { + overflow.get_mut(index - MAX_INLINE_CAPACITY) + } else { + None + } + } + /// Returns the number of elements in the `GrowableArray`. #[allow(dead_code)] #[inline] @@ -179,6 +194,17 @@ mod tests { type KeyValuePair = Option<(Key, AnyValue)>; + #[test] + fn test_get_mut() { + let mut collection = GrowableArray::::new(); + collection.push(1); + collection.push(2); + *collection.get_mut(0).expect("index 0") = 10; + *collection.get_mut(1).expect("index 1") = 20; + assert_eq!(collection.get(0), Some(&10)); + assert_eq!(collection.get(1), Some(&20)); + } + #[test] fn test_push_and_get() { let mut collection = GrowableArray::::new(); diff --git a/opentelemetry-sdk/src/logs/logger.rs b/opentelemetry-sdk/src/logs/logger.rs index b2c789c175..2359d31648 100644 --- a/opentelemetry-sdk/src/logs/logger.rs +++ b/opentelemetry-sdk/src/logs/logger.rs @@ -27,7 +27,7 @@ impl opentelemetry::logs::Logger for SdkLogger { type LogRecord = SdkLogRecord; fn create_log_record(&self) -> Self::LogRecord { - SdkLogRecord::new() + SdkLogRecord::with_deduplicate_attributes(self.provider.deduplicate_log_record_attributes()) } /// Emit a `LogRecord`. diff --git a/opentelemetry-sdk/src/logs/logger_provider.rs b/opentelemetry-sdk/src/logs/logger_provider.rs index e8765b456b..215be80c74 100644 --- a/opentelemetry-sdk/src/logs/logger_provider.rs +++ b/opentelemetry-sdk/src/logs/logger_provider.rs @@ -22,6 +22,7 @@ fn noop_logger_provider() -> &'static SdkLoggerProvider { inner: Arc::new(LoggerProviderInner { processors: Vec::new(), is_shutdown: AtomicBool::new(true), + deduplicate_log_record_attributes: true, }), }) } @@ -82,6 +83,10 @@ impl SdkLoggerProvider { &self.inner.processors } + pub(crate) fn deduplicate_log_record_attributes(&self) -> bool { + self.inner.deduplicate_log_record_attributes + } + /// Force flush all remaining logs in log processors and return results. pub fn force_flush(&self) -> OTelSdkResult { let result: Vec<_> = self @@ -135,6 +140,7 @@ impl SdkLoggerProvider { struct LoggerProviderInner { processors: Vec>, is_shutdown: AtomicBool, + deduplicate_log_record_attributes: bool, } impl LoggerProviderInner { @@ -179,11 +185,22 @@ impl Drop for LoggerProviderInner { } } -#[derive(Debug, Default)] +#[derive(Debug)] /// Builder for provider attributes. pub struct LoggerProviderBuilder { processors: Vec>, resource: Option, + deduplicate_log_record_attributes: bool, +} + +impl Default for LoggerProviderBuilder { + fn default() -> Self { + Self { + processors: Vec::new(), + resource: None, + deduplicate_log_record_attributes: true, + } + } } impl LoggerProviderBuilder { @@ -287,6 +304,16 @@ impl LoggerProviderBuilder { LoggerProviderBuilder { resource, ..self } } + /// Controls whether log record attribute keys are deduplicated (last-write-wins). + /// + /// Enabled by default per the OpenTelemetry specification. Set to `false` to + /// retain the previous push-only behavior and avoid the lookup cost on + /// `add_attribute`. + pub fn with_log_record_attribute_deduplication(mut self, enabled: bool) -> Self { + self.deduplicate_log_record_attributes = enabled; + self + } + /// Create a new provider from this configuration. pub fn build(self) -> SdkLoggerProvider { let resource = self.resource.unwrap_or(Resource::builder().build()); @@ -299,6 +326,7 @@ impl LoggerProviderBuilder { inner: Arc::new(LoggerProviderInner { processors, is_shutdown: AtomicBool::new(false), + deduplicate_log_record_attributes: self.deduplicate_log_record_attributes, }), }; @@ -793,6 +821,7 @@ mod tests { flush_called.clone(), ))], is_shutdown: AtomicBool::new(false), + deduplicate_log_record_attributes: true, }); { @@ -833,6 +862,7 @@ mod tests { flush_called.clone(), ))], is_shutdown: AtomicBool::new(false), + deduplicate_log_record_attributes: true, }); // Create a scope to test behavior when providers are dropped diff --git a/opentelemetry-sdk/src/logs/mod.rs b/opentelemetry-sdk/src/logs/mod.rs index eef1405791..bdcd8b63fa 100644 --- a/opentelemetry-sdk/src/logs/mod.rs +++ b/opentelemetry-sdk/src/logs/mod.rs @@ -115,6 +115,29 @@ mod tests { assert_eq!(&resource, log.resource.borrow()); } + #[test] + fn log_record_attribute_deduplication_disabled_via_builder() { + let exporter: InMemoryLogExporter = InMemoryLogExporter::default(); + let provider = SdkLoggerProvider::builder() + .with_log_processor(SimpleLogProcessor::new(exporter.clone())) + .with_log_record_attribute_deduplication(false) + .build(); + + let logger = provider.logger("test-logger"); + let mut log_record = logger.create_log_record(); + log_record.add_attribute("key", "first"); + log_record.add_attribute("key", "second"); + logger.emit(log_record); + + let exported_logs = exporter + .get_emitted_logs() + .expect("Logs are expected to be exported."); + let log = exported_logs + .first() + .expect("Atleast one log is expected to be present."); + assert_eq!(log.record.attributes_len(), 2); + } + #[test] fn logger_attributes() { let exporter: InMemoryLogExporter = InMemoryLogExporter::default(); diff --git a/opentelemetry-sdk/src/logs/record.rs b/opentelemetry-sdk/src/logs/record.rs index 746e0f5b9b..3f5cc54fec 100644 --- a/opentelemetry-sdk/src/logs/record.rs +++ b/opentelemetry-sdk/src/logs/record.rs @@ -50,6 +50,10 @@ pub struct SdkLogRecord { /// Additional attributes associated with this record pub(crate) attributes: LogRecordAttributes, + + /// When true, `add_attribute` replaces an existing key (last-write-wins). + /// When false, attributes are appended without deduplication. + pub(crate) deduplicate_attributes: bool, } impl opentelemetry::logs::LogRecord for SdkLogRecord { @@ -101,7 +105,21 @@ impl opentelemetry::logs::LogRecord for SdkLogRecord { K: Into, V: Into, { - self.attributes.push(Some((key.into(), value.into()))); + let key = key.into(); + let value = value.into(); + if self.deduplicate_attributes { + for i in 0..self.attributes.len() { + if let Some((existing_key, existing_value)) = + self.attributes.get_mut(i).and_then(|opt| opt.as_mut()) + { + if *existing_key == key { + *existing_value = value; + return; + } + } + } + } + self.attributes.push(Some((key, value))); } fn set_trace_context( @@ -119,8 +137,15 @@ impl opentelemetry::logs::LogRecord for SdkLogRecord { } impl SdkLogRecord { - /// Crate only default constructor + /// Crate-only default constructor with attribute deduplication enabled. + /// + /// Used from `#[cfg(test)]` modules; `clippy --lib` cannot see those call sites. + #[allow(dead_code)] pub(crate) fn new() -> Self { + Self::with_deduplicate_attributes(true) + } + + pub(crate) fn with_deduplicate_attributes(deduplicate_attributes: bool) -> Self { SdkLogRecord { event_name: None, target: None, @@ -131,6 +156,7 @@ impl SdkLogRecord { severity_number: None, body: None, attributes: LogRecordAttributes::default(), + deduplicate_attributes, } } @@ -307,6 +333,54 @@ mod tests { assert!(log_record.attributes_contains(&key, &value)); } + #[test] + fn test_add_attribute_deduplicates_last_write_wins() { + let mut log_record = SdkLogRecord::new(); + log_record.add_attribute("key", "first"); + log_record.add_attribute("key", "second"); + assert_eq!(log_record.attributes_len(), 1); + assert!( + log_record.attributes_contains(&Key::new("key"), &AnyValue::String("second".into())) + ); + } + + #[test] + fn test_add_attributes_deduplicates_within_batch() { + let mut log_record = SdkLogRecord::new(); + log_record.add_attributes([("key", "first"), ("key", "second")]); + log_record.add_attribute(Key::new("other"), AnyValue::Int(1)); + assert_eq!(log_record.attributes_len(), 2); + assert!( + log_record.attributes_contains(&Key::new("key"), &AnyValue::String("second".into())) + ); + assert!(log_record.attributes_contains(&Key::new("other"), &AnyValue::Int(1))); + } + + #[test] + fn test_add_attribute_skips_deduplication_when_disabled() { + let mut log_record = SdkLogRecord::with_deduplicate_attributes(false); + log_record.add_attribute("key", "first"); + log_record.add_attribute("key", "second"); + assert_eq!(log_record.attributes_len(), 2); + assert!(log_record.attributes_contains(&Key::new("key"), &AnyValue::String("first".into()))); + assert!( + log_record.attributes_contains(&Key::new("key"), &AnyValue::String("second".into())) + ); + } + + #[test] + fn test_add_attribute_deduplicates_beyond_inline_capacity() { + let mut log_record = SdkLogRecord::new(); + for i in 0..PREALLOCATED_ATTRIBUTE_CAPACITY { + log_record.add_attribute(format!("key{i}"), i as i64); + } + log_record.add_attribute("key0", "updated"); + assert_eq!(log_record.attributes_len(), PREALLOCATED_ATTRIBUTE_CAPACITY); + assert!( + log_record.attributes_contains(&Key::new("key0"), &AnyValue::String("updated".into())) + ); + } + #[test] fn compare_trace_context() { let trace_context = TraceContext { @@ -339,6 +413,7 @@ mod tests { severity_number: Some(Severity::Error), body: Some(AnyValue::String("Test body".into())), attributes: LogRecordAttributes::new(), + deduplicate_attributes: true, trace_context: Some(TraceContext { trace_id: TraceId::from(1), span_id: SpanId::from(1),