Merge branch 'main' into otel-prometheus-0.30

Author: TommyCpp

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

@@ -361,7 +361,7 @@ impl HasTonicConfig for TonicExporterBuilder {
 /// ```
 /// # #[cfg(all(feature = "trace", feature = "grpc-tonic"))]
 /// # {
-/// use crate::opentelemetry_otlp::{WithExportConfig, WithTonicConfig};
+/// use opentelemetry_otlp::{WithExportConfig, WithTonicConfig};
 /// let exporter_builder = opentelemetry_otlp::SpanExporter::builder()
 ///     .with_tonic()
 ///     .with_compression(opentelemetry_otlp::Compression::Gzip);
@@ -373,6 +373,32 @@ pub trait WithTonicConfig {
     fn with_tls_config(self, tls_config: ClientTlsConfig) -> Self;
 
     /// Set custom metadata entries to send to the collector.
+    ///
+    /// **Note**: This method is additive - calling it multiple times will merge
+    /// the metadata entries. If the same key is provided in multiple calls,
+    /// the last value will override previous ones.
+    ///
+    /// # Example
+    /// ```no_run
+    /// # #[cfg(feature = "grpc-tonic")]
+    /// # {
+    /// use tonic::metadata::MetadataMap;
+    /// use opentelemetry_otlp::WithTonicConfig;
+    ///
+    /// let mut metadata1 = MetadataMap::new();
+    /// metadata1.insert("key1", "value1".parse().unwrap());
+    ///
+    /// let mut metadata2 = MetadataMap::new();
+    /// metadata2.insert("key2", "value2".parse().unwrap());
+    ///
+    /// let exporter = opentelemetry_otlp::SpanExporter::builder()
+    ///     .with_tonic()
+    ///     .with_metadata(metadata1)  // Adds key1=value1
+    ///     .with_metadata(metadata2)  // Adds key2=value2 (both are present)
+    ///     .build()?;
+    /// # }
+    /// # Ok::<(), Box<dyn std::error::Error>>(())
+    /// ```
     fn with_metadata(self, metadata: MetadataMap) -> Self;
 
     /// Set the compression algorithm to use when communicating with the collector.
@@ -387,8 +413,65 @@ pub trait WithTonicConfig {
     fn with_channel(self, channel: tonic::transport::Channel) -> Self;
 
     /// Use a custom `interceptor` to modify each outbound request.
-    /// this can be used to modify the grpc metadata, for example
+    /// This can be used to modify the gRPC metadata, for example
     /// to inject auth tokens.
+    ///
+    /// **Note**: Calling this method multiple times will replace the previous
+    /// interceptor. If you need multiple interceptors, chain them together
+    /// before passing to this method.
+    ///
+    /// # Examples
+    ///
+    /// ## Single interceptor
+    /// ```no_run
+    /// # #[cfg(feature = "grpc-tonic")]
+    /// # {
+    /// use tonic::{Request, Status};
+    /// use opentelemetry_otlp::WithTonicConfig;
+    ///
+    /// fn auth_interceptor(mut req: Request<()>) -> Result<Request<()>, Status> {
+    ///     req.metadata_mut().insert("authorization", "Bearer token".parse().unwrap());
+    ///     Ok(req)
+    /// }
+    ///
+    /// let exporter = opentelemetry_otlp::SpanExporter::builder()
+    ///     .with_tonic()
+    ///     .with_interceptor(auth_interceptor)
+    ///     .build()?;
+    /// # }
+    /// # Ok::<(), Box<dyn std::error::Error>>(())
+    /// ```
+    ///
+    /// ## Multiple interceptors (chaining)
+    /// ```no_run
+    /// # #[cfg(feature = "grpc-tonic")]
+    /// # {
+    /// use tonic::{Request, Status};
+    /// use opentelemetry_otlp::WithTonicConfig;
+    ///
+    /// fn auth_interceptor(mut req: Request<()>) -> Result<Request<()>, Status> {
+    ///     req.metadata_mut().insert("authorization", "Bearer token".parse().unwrap());
+    ///     Ok(req)
+    /// }
+    ///
+    /// fn logging_interceptor(req: Request<()>) -> Result<Request<()>, Status> {
+    ///     println!("Sending gRPC request with metadata: {:?}", req.metadata());
+    ///     Ok(req)
+    /// }
+    ///
+    /// // Chain interceptors by wrapping them
+    /// fn combined_interceptor(req: Request<()>) -> Result<Request<()>, Status> {
+    ///     let req = logging_interceptor(req)?;
+    ///     auth_interceptor(req)
+    /// }
+    ///
+    /// let exporter = opentelemetry_otlp::SpanExporter::builder()
+    ///     .with_tonic()
+    ///     .with_interceptor(combined_interceptor)
+    ///     .build()?;
+    /// # }
+    /// # Ok::<(), Box<dyn std::error::Error>>(())
+    /// ```
     fn with_interceptor<I>(self, interceptor: I) -> Self
     where
         I: tonic::service::Interceptor + Clone + Send + Sync + 'static;

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -360,7 +360,7 @@ impl BatchLogProcessor {
                     logs: &mut Vec<LogsData>,
                     last_export_time: &mut Instant,
                     current_batch_size: &AtomicUsize,
-                    config: &BatchConfig,
+                    max_export_size: usize,
                 ) -> OTelSdkResult
                 where
                     E: LogExporter + Send + Sync + 'static,
@@ -373,7 +373,7 @@ impl BatchLogProcessor {
                         // Get upto `max_export_batch_size` amount of logs log records from the channel and push them to the logs vec
                         while let Ok(log) = logs_receiver.try_recv() {
                             logs.push(log);
-                            if logs.len() == config.max_export_batch_size {
+                            if logs.len() == max_export_size {
                                 break;
                             }
                         }
@@ -409,7 +409,7 @@ impl BatchLogProcessor {
                                 &mut logs,
                                 &mut last_export_time,
                                 &current_batch_size,
-                                &config,
+                                max_export_batch_size,
                             );
                         }
                         Ok(BatchMessage::ForceFlush(sender)) => {
@@ -420,7 +420,7 @@ impl BatchLogProcessor {
                                 &mut logs,
                                 &mut last_export_time,
                                 &current_batch_size,
-                                &config,
+                                max_export_batch_size,
                             );
                             let _ = sender.send(result);
                         }
@@ -432,7 +432,7 @@ impl BatchLogProcessor {
                                 &mut logs,
                                 &mut last_export_time,
                                 &current_batch_size,
-                                &config,
+                                max_export_batch_size,
                             );
                             let _ = exporter.shutdown();
                             let _ = sender.send(result);
@@ -460,7 +460,7 @@ impl BatchLogProcessor {
                                 &mut logs,
                                 &mut last_export_time,
                                 &current_batch_size,
-                                &config,
+                                max_export_batch_size,
                             );
                         }
                         Err(RecvTimeoutError::Disconnected) => {

opentelemetry-sdk/src/logs/in_memory_exporter.rs

@@ -132,6 +132,7 @@ impl InMemoryLogExporterBuilder {
 
     /// If set, the records will not be [`InMemoryLogExporter::reset`] on shutdown.
     #[cfg(test)]
+    #[allow(dead_code)]
     pub(crate) fn keep_records_on_shutdown(self) -> Self {
         Self {
             reset_on_shutdown: false,

opentelemetry-sdk/src/trace/in_memory_exporter.rs

@@ -2,7 +2,7 @@ use crate::error::{OTelSdkError, OTelSdkResult};
 use crate::resource::Resource;
 use crate::trace::{SpanData, SpanExporter};
 use crate::InMemoryExporterError;
-use std::sync::{Arc, Mutex};
+use std::sync::{atomic::AtomicBool, Arc, Mutex};
 use std::time::Duration;
 
 /// An in-memory span exporter that stores span data in memory.
@@ -51,6 +51,8 @@ use std::time::Duration;
 pub struct InMemorySpanExporter {
     spans: Arc<Mutex<Vec<SpanData>>>,
     resource: Arc<Mutex<Resource>>,
+    should_reset_on_shutdown: bool,
+    shutdown_called: Arc<AtomicBool>,
 }
 
 impl Default for InMemorySpanExporter {
@@ -67,7 +69,9 @@ impl Default for InMemorySpanExporter {
 /// let exporter = InMemorySpanExporterBuilder::new().build();
 /// ```
 #[derive(Clone, Debug)]
-pub struct InMemorySpanExporterBuilder {}
+pub struct InMemorySpanExporterBuilder {
+    reset_on_shutdown: bool,
+}
 
 impl Default for InMemorySpanExporterBuilder {
     fn default() -> Self {
@@ -78,19 +82,38 @@ impl Default for InMemorySpanExporterBuilder {
 impl InMemorySpanExporterBuilder {
     /// Creates a new instance of the `InMemorySpanExporterBuilder`.
     pub fn new() -> Self {
-        Self {}
+        Self {
+            reset_on_shutdown: true,
+        }
     }
 
     /// Creates a new instance of the `InMemorySpanExporter`.
     pub fn build(&self) -> InMemorySpanExporter {
         InMemorySpanExporter {
             spans: 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)),
+        }
+    }
+
+    /// If set, the records will not be [`InMemorySpanExporter::reset`] on shutdown.
+    #[cfg(test)]
+    #[allow(dead_code)]
+    pub(crate) fn keep_records_on_shutdown(self) -> Self {
+        Self {
+            reset_on_shutdown: false,
         }
     }
 }
 
 impl InMemorySpanExporter {
+    /// Returns true if shutdown was called.
+    pub fn is_shutdown_called(&self) -> bool {
+        self.shutdown_called
+            .load(std::sync::atomic::Ordering::Relaxed)
+    }
+
     /// Returns the finished span as a vector of `SpanData`.
     ///
     /// # Errors
@@ -140,7 +163,11 @@ impl SpanExporter for InMemorySpanExporter {
     }
 
     fn shutdown_with_timeout(&mut self, _timeout: Duration) -> OTelSdkResult {
-        self.reset();
+        self.shutdown_called
+            .store(true, std::sync::atomic::Ordering::Relaxed);
+        if self.should_reset_on_shutdown {
+            self.reset();
+        }
         Ok(())
     }