Merge remote-tracking branch 'refs/remotes/origin/chore/semantic-commits' into chore/semantic-commits

Author: scottgerring

README.md

@@ -186,6 +186,7 @@ you're more than welcome to participate!
 ### Approvers
 
 * [Shaun Cox](https://github.com/shaun-cox)
+* [Scott Gerring](https://github.com/scottgerring)
 
 ### Emeritus
 

opentelemetry-otlp/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## vNext
 
 - The `OTEL_EXPORTER_OTLP_TIMEOUT`, `OTEL_EXPORTER_OTLP_TRACES_TIMEOUT`, `OTEL_EXPORTER_OTLP_METRICS_TIMEOUT` and `OTEL_EXPORTER_OTLP_LOGS_TIMEOUT` are changed from seconds to miliseconds.
+- Fixed `.with_headers()` in `HttpExporterBuilder` to correctly support multiple key/value pairs. [#2699](https://github.com/open-telemetry/opentelemetry-rust/pull/2699)
 
 ## 0.28.0
 
@@ -41,7 +42,7 @@ Released 2024-Nov-11
 - Update `opentelemetry-http` dependency version to 0.27
 - Update `opentelemetry-proto` dependency version to 0.27
 
-- **BREAKING**: 
+- **BREAKING**:
   - ([#2217](https://github.com/open-telemetry/opentelemetry-rust/pull/2217)) **Replaced**: The `MetricsExporterBuilder` interface is modified from `with_temporality_selector` to `with_temporality` example can be seen below:
     Previous Signature:
     ```rust
@@ -88,9 +89,9 @@ Released 2024-Nov-11
       - `MetricsExporterBuilder` -> `MetricExporterBuilder`
 
   - [#2263](https://github.com/open-telemetry/opentelemetry-rust/pull/2263)
-  Support `hyper` client for opentelemetry-otlp. This can be enabled using flag `hyper-client`. 
+  Support `hyper` client for opentelemetry-otlp. This can be enabled using flag `hyper-client`.
   Refer example: https://github.com/open-telemetry/opentelemetry-rust/tree/main/opentelemetry-otlp/examples/basic-otlp-http
-  
+
 ## v0.26.0
 Released 2024-Sep-30
 

opentelemetry-otlp/src/exporter/http/mod.rs

@@ -449,13 +449,13 @@ impl<B: HasHttpConfig> WithHttpConfig for B {
 
     fn with_headers(mut self, headers: HashMap<String, String>) -> Self {
         // headers will be wrapped, so we must do some logic to unwrap first.
-        self.http_client_config()
+        let http_client_headers = self
+            .http_client_config()
             .headers
-            .iter_mut()
-            .zip(headers)
-            .for_each(|(http_client_headers, (key, value))| {
-                http_client_headers.insert(key, super::url_decode(&value).unwrap_or(value));
-            });
+            .get_or_insert(HashMap::new());
+        headers.into_iter().for_each(|(key, value)| {
+            http_client_headers.insert(key, super::url_decode(&value).unwrap_or(value));
+        });
         self
     }
 }
@@ -671,11 +671,14 @@ mod tests {
     }
 
     #[test]
-    fn test_http_exporter_builder_with_header() {
+    fn test_http_exporter_builder_with_headers() {
         use std::collections::HashMap;
         // Arrange
         let initial_headers = HashMap::from([("k1".to_string(), "v1".to_string())]);
-        let extra_headers = HashMap::from([("k2".to_string(), "v2".to_string())]);
+        let extra_headers = HashMap::from([
+            ("k2".to_string(), "v2".to_string()),
+            ("k3".to_string(), "v3".to_string()),
+        ]);
         let expected_headers = initial_headers.iter().chain(extra_headers.iter()).fold(
             HashMap::new(),
             |mut acc, (k, v)| {

opentelemetry-sdk/CHANGELOG.md

@@ -33,17 +33,24 @@
     }
 }
 ```
+
 - **Breaking** The SpanExporter::export() method no longer requires a mutable reference to self.
-  Before: 
+  Before:
   ```rust
-    async fn export(&mut self, batch: Vec<SpanData>) -> OTelSdkResult 
+    async fn export(&mut self, batch: Vec<SpanData>) -> OTelSdkResult
   ```
-  After: 
-  ```rust 
-    async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult  
+  After:
+  ```rust
+    async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult
   ```
+
   Custom exporters will need to internally synchronize any mutable state, if applicable.
-  
+
+- Bug Fix: `BatchLogProcessor` now correctly calls `shutdown` on the exporter
+  when its `shutdown` is invoked.
+
+- Reduced some info level logs to debug
+
 ## 0.28.0
 
 Released 2025-Feb-10
@@ -85,7 +92,7 @@ field as metadata, a feature introduced in version 0.1.40.
 - `Resource::{new,empty,from_detectors,new_with_defaults,from_schema_url,merge,default}`.
    To create Resources you should only use `Resource::builder()` or `Resource::builder_empty()`. See
    [#2322](https://github.com/open-telemetry/opentelemetry-rust/pull/2322) for a migration guide.
-  
+
   Example Usage:
 
   ```rust
@@ -173,7 +180,7 @@ field as metadata, a feature introduced in version 0.1.40.
   - Does not support `hyper`, `reqwest` HTTP Clients
   - Does not support multiple concurrent exports (`with_max_concurrent_exports`
      is not supported). This existed only for traces.
-  
+
   If this applies to you, you can get the old behavior back by following steps
   below:
   - Enable one or more of the feature flag from below
@@ -241,7 +248,7 @@ limit.
 - *Breaking*: Rename namespaces for InMemoryExporters. (The module is still
   under "testing" feature flag)
   before:
-  
+
   ```rust
   opentelemetry_sdk::testing::logs::{InMemoryLogExporter,
   InMemoryLogExporterBuilder};
@@ -252,7 +259,7 @@ limit.
   ```
 
   now:
-  
+
   ```rust
   opentelemetry_sdk::logs::{InMemoryLogExporter, InMemoryLogExporterBuilder};
   opentelemetry_sdk::trace::{InMemorySpanExporter,

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -23,7 +23,7 @@ use crate::{
 };
 use std::sync::mpsc::{self, RecvTimeoutError, SyncSender};
 
-use opentelemetry::{otel_debug, otel_error, otel_info, otel_warn, InstrumentationScope};
+use opentelemetry::{otel_debug, otel_error, otel_warn, InstrumentationScope};
 
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::{cmp::min, env, sync::Mutex};
@@ -342,7 +342,7 @@ impl BatchLogProcessor {
         let handle = thread::Builder::new()
             .name("OpenTelemetry.Logs.BatchProcessor".to_string())
             .spawn(move || {
-                otel_info!(
+                otel_debug!(
                     name: "BatchLogProcessor.ThreadStarted",
                     interval_in_millisecs = config.scheduled_delay.as_millis(),
                     max_export_batch_size = config.max_export_batch_size,
@@ -436,6 +436,7 @@ impl BatchLogProcessor {
                                 &current_batch_size,
                                 &config,
                             );
+                            let _ = exporter.shutdown();
                             let _ = sender.send(result);
 
                             otel_debug!(
@@ -475,7 +476,7 @@ impl BatchLogProcessor {
                         }
                     }
                 }
-                otel_info!(
+                otel_debug!(
                     name: "BatchLogProcessor.ThreadStopped"
                 );
             })
@@ -925,7 +926,8 @@ mod tests {
         processor.shutdown().unwrap();
         // todo: expect to see errors here. How should we assert this?
         processor.emit(&mut record, &instrumentation);
-        assert_eq!(1, exporter.get_emitted_logs().unwrap().len())
+        assert_eq!(1, exporter.get_emitted_logs().unwrap().len());
+        assert!(exporter.is_shutdown_called());
     }
 
     #[tokio::test(flavor = "current_thread")]

opentelemetry-sdk/src/logs/in_memory_exporter.rs

@@ -4,6 +4,7 @@ use crate::logs::{LogBatch, LogExporter};
 use crate::Resource;
 use opentelemetry::InstrumentationScope;
 use std::borrow::Cow;
+use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, Mutex};
 
 type LogResult<T> = Result<T, OTelSdkError>;
@@ -42,6 +43,7 @@ pub struct InMemoryLogExporter {
     logs: Arc<Mutex<Vec<OwnedLogData>>>,
     resource: Arc<Mutex<Resource>>,
     should_reset_on_shutdown: bool,
+    shutdown_called: Arc<AtomicBool>,
 }
 
 impl Default for InMemoryLogExporter {
@@ -124,6 +126,7 @@ impl InMemoryLogExporterBuilder {
             logs: Arc::new(Mutex::new(Vec::new())),
             resource: Arc::new(Mutex::new(Resource::builder().build())),
             should_reset_on_shutdown: self.reset_on_shutdown,
+            shutdown_called: Arc::new(AtomicBool::new(false)),
         }
     }
 
@@ -137,6 +140,12 @@ impl InMemoryLogExporterBuilder {
 }
 
 impl InMemoryLogExporter {
+    /// Returns true if shutdown was called.
+    pub fn is_shutdown_called(&self) -> bool {
+        self.shutdown_called
+            .load(std::sync::atomic::Ordering::Relaxed)
+    }
+
     /// Returns the logs emitted via Logger as a vector of `LogDataWithResource`.
     ///
     /// # Example
@@ -203,6 +212,8 @@ impl LogExporter for InMemoryLogExporter {
     }
 
     fn shutdown(&mut self) -> OTelSdkResult {
+        self.shutdown_called
+            .store(true, std::sync::atomic::Ordering::Relaxed);
         if self.should_reset_on_shutdown {
             self.reset();
         }

opentelemetry-sdk/src/logs/simple_log_processor.rs

@@ -223,7 +223,8 @@ mod tests {
 
         processor.emit(&mut record, &instrumentation);
 
-        assert_eq!(1, exporter.get_emitted_logs().unwrap().len())
+        assert_eq!(1, exporter.get_emitted_logs().unwrap().len());
+        assert!(exporter.is_shutdown_called());
     }
 
     #[test]

opentelemetry-sdk/src/metrics/meter_provider.rs

@@ -110,7 +110,7 @@ impl SdkMeterProvider {
     /// There is no guaranteed that all telemetry be flushed or all resources have
     /// been released on error.
     pub fn shutdown(&self) -> OTelSdkResult {
-        otel_info!(
+        otel_debug!(
             name: "MeterProvider.Shutdown",
             message = "User initiated shutdown of MeterProvider."
         );
@@ -164,7 +164,7 @@ impl Drop for SdkMeterProviderInner {
                     reason = format!("{}", err)
                 );
             } else {
-                otel_info!(
+                otel_debug!(
                     name: "MeterProvider.Drop.ShutdownCompleted",
                 );
             }
@@ -310,7 +310,7 @@ impl MeterProviderBuilder {
             }),
         };
 
-        otel_info!(
+        otel_debug!(
             name: "MeterProvider.Built",
         );
         meter_provider
@@ -562,6 +562,25 @@ mod tests {
         assert_eq!(provider.inner.meters.lock().unwrap().len(), 5);
     }
 
+    #[test]
+    fn same_meter_reused_same_scope_attributes() {
+        let meter_provider = super::SdkMeterProvider::builder().build();
+        let make_scope = |attributes| {
+            InstrumentationScope::builder("test.meter")
+                .with_version("v0.1.0")
+                .with_schema_url("http://example.com")
+                .with_attributes(attributes)
+                .build()
+        };
+
+        let _meter1 =
+            meter_provider.meter_with_scope(make_scope(vec![KeyValue::new("key", "value1")]));
+        let _meter2 =
+            meter_provider.meter_with_scope(make_scope(vec![KeyValue::new("key", "value1")]));
+
+        assert_eq!(meter_provider.inner.meters.lock().unwrap().len(), 1);
+    }
+
     #[test]
     fn with_resource_multiple_calls_ensure_additive() {
         let builder = SdkMeterProvider::builder()

opentelemetry-sdk/src/metrics/mod.rs

@@ -782,7 +782,7 @@ mod tests {
             .build();
 
         // Act
-        // Meters are identical except for scope attributes, but scope attributes are not an identifying property.
+        // Meters are identical.
         // Hence there should be a single metric stream output for this test.
         let make_scope = |attributes| {
             InstrumentationScope::builder("test.meter")
@@ -795,7 +795,7 @@ mod tests {
         let meter1 =
             meter_provider.meter_with_scope(make_scope(vec![KeyValue::new("key", "value1")]));
         let meter2 =
-            meter_provider.meter_with_scope(make_scope(vec![KeyValue::new("key", "value2")]));
+            meter_provider.meter_with_scope(make_scope(vec![KeyValue::new("key", "value1")]));
 
         let counter1 = meter1
             .u64_counter("my_counter")

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -160,7 +160,7 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
             .spawn(move || {
                 let mut interval_start = Instant::now();
                 let mut remaining_interval = interval;
-                otel_info!(
+                otel_debug!(
                     name: "PeriodReaderThreadStarted",
                     interval_in_millisecs = interval.as_millis(),
                 );
@@ -190,16 +190,16 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
 
                             if export_result.is_err() {
                                 if response_sender.send(false).is_err() {
-                                    otel_info!(
+                                    otel_debug!(
                                         name: "PeriodReader.Flush.ResponseSendError",
-                                        message = "PeriodicReader's flush has failed, but unable to send this info back to caller. 
+                                        message = "PeriodicReader's flush has failed, but unable to send this info back to caller.
                                         This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the flush timeout."
                                     );
                                 }
                             } else if response_sender.send(true).is_err() {
-                                otel_info!(
+                                otel_debug!(
                                     name: "PeriodReader.Flush.ResponseSendError",
-                                    message = "PeriodicReader's flush has completed successfully, but unable to send this info back to caller. 
+                                    message = "PeriodicReader's flush has completed successfully, but unable to send this info back to caller.
                                     This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the flush timeout."
                                 );
                             }
@@ -253,14 +253,14 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
                                 if response_sender.send(false).is_err() {
                                     otel_info!(
                                         name: "PeriodReaderThreadShutdown.ResponseSendError",
-                                        message = "PeriodicReader's shutdown has failed, but unable to send this info back to caller. 
+                                        message = "PeriodicReader's shutdown has failed, but unable to send this info back to caller.
                                         This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the shutdown timeout."
                                     );
                                 }
                             } else if response_sender.send(true).is_err() {
-                                otel_info!(
+                                otel_debug!(
                                     name: "PeriodReaderThreadShutdown.ResponseSendError",
-                                    message = "PeriodicReader completed its shutdown, but unable to send this info back to caller. 
+                                    message = "PeriodicReader completed its shutdown, but unable to send this info back to caller.
                                     This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the shutdown timeout."
                                 );
                             }
@@ -312,7 +312,7 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
                         }
                     }
                 }
-                otel_info!(
+                otel_debug!(
                     name: "PeriodReaderThreadStopped"
                 );
             });
@@ -499,7 +499,7 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
     /// This function SHOULD be obtained from the exporter.
     ///
     /// If not configured, the Cumulative temporality SHOULD be used.
-    ///  
+    ///
     /// [metric-reader]: https://github.com/open-telemetry/opentelemetry-specification/blob/0a78571045ca1dca48621c9648ec3c832c3c541c/specification/metrics/sdk.md#metricreader
     fn temporality(&self, kind: InstrumentKind) -> Temporality {
         kind.temporality_preference(self.inner.temporality(kind))