feat: Add timeout duration to shutdown

opentelemetry-appender-tracing/benches/log-attributes.rs

@@ -21,7 +21,6 @@
 | otel_11_attributes   | 625 ns       | +106 ns   | // vec! initial capacity is 5. 11th attribute causes vec! to be reallocated
 | otel_12_attributes   | 676 ns       | +51 ns    |
 */
-
 use criterion::{criterion_group, criterion_main, Criterion};
 use opentelemetry::InstrumentationScope;
 use opentelemetry_appender_tracing::layer as tracing_layer;
@@ -30,6 +29,7 @@ use opentelemetry_sdk::logs::{LogProcessor, SdkLogRecord, SdkLoggerProvider};
 use opentelemetry_sdk::Resource;
 #[cfg(not(target_os = "windows"))]
 use pprof::criterion::{Output, PProfProfiler};
+use std::time::Duration;
 use tracing::error;
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::Registry;
@@ -44,7 +44,7 @@ impl LogProcessor for NoopProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 }

opentelemetry-appender-tracing/benches/logs.rs

@@ -22,7 +22,6 @@
     | ot_layer_disabled           | 12 ns       |
     | ot_layer_enabled            | 186 ns      |
 */
-
 use criterion::{criterion_group, criterion_main, Criterion};
 use opentelemetry::InstrumentationScope;
 use opentelemetry_appender_tracing::layer as tracing_layer;
@@ -31,6 +30,7 @@ use opentelemetry_sdk::logs::{LogProcessor, SdkLogRecord, SdkLoggerProvider};
 use opentelemetry_sdk::Resource;
 #[cfg(not(target_os = "windows"))]
 use pprof::criterion::{Output, PProfProfiler};
+use std::time::Duration;
 use tracing::error;
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::Layer;
@@ -54,7 +54,7 @@ impl LogProcessor for NoopProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 

opentelemetry-appender-tracing/src/layer.rs

@@ -232,6 +232,7 @@ mod tests {
     use opentelemetry_sdk::logs::{LogBatch, LogExporter};
     use opentelemetry_sdk::logs::{SdkLogRecord, SdkLoggerProvider};
     use opentelemetry_sdk::trace::{Sampler, SdkTracerProvider};
+    use std::time::Duration;
     use tracing::{error, warn};
     use tracing_subscriber::prelude::__tracing_subscriber_SubscriberExt;
     use tracing_subscriber::util::SubscriberInitExt;
@@ -821,7 +822,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
             Ok(())
         }
     }

opentelemetry-proto/src/transform/logs.rs

@@ -229,6 +229,7 @@ mod tests {
     use opentelemetry_sdk::logs::LogProcessor;
     use opentelemetry_sdk::logs::SdkLoggerProvider;
     use opentelemetry_sdk::{logs::LogBatch, logs::SdkLogRecord, Resource};
+    use std::time;
 
     #[derive(Debug)]
     struct MockProcessor;
@@ -240,7 +241,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown(&self, _timeout: time::Duration) -> OTelSdkResult {
             Ok(())
         }
     }

opentelemetry-sdk/benches/log.rs

@@ -17,6 +17,7 @@ RAM: 64.0 GB
 
 use opentelemetry::time::now;
 use std::collections::HashMap;
+use std::time::Duration;
 
 use criterion::{criterion_group, criterion_main, Criterion};
 
@@ -38,7 +39,7 @@ impl LogProcessor for NoopProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 }

opentelemetry-sdk/benches/log_exporter.rs

@@ -13,6 +13,7 @@
 use opentelemetry::time::now;
 use opentelemetry_sdk::error::OTelSdkResult;
 use std::sync::Mutex;
+use std::time::Duration;
 
 use criterion::{criterion_group, criterion_main, Criterion};
 
@@ -73,7 +74,7 @@ impl LogProcessor for ExportingProcessorWithFuture {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 }
@@ -104,7 +105,7 @@ impl LogProcessor for ExportingProcessorWithoutFuture {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 }

opentelemetry-sdk/benches/log_processor.rs

@@ -15,6 +15,7 @@ use opentelemetry::time::now;
 use std::{
     sync::{Arc, Mutex},
     thread::sleep,
+    time::Duration,
 };
 
 use criterion::{criterion_group, criterion_main, Criterion};
@@ -54,7 +55,7 @@ impl LogProcessor for NoopProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 }
@@ -71,7 +72,7 @@ impl LogProcessor for CloningProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 }
@@ -117,7 +118,7 @@ impl LogProcessor for SendToChannelProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 }

opentelemetry-sdk/benches/metric.rs

@@ -1,6 +1,3 @@
-use rand::Rng;
-use std::sync::{Arc, Weak};
-
 use criterion::{criterion_group, criterion_main, Bencher, Criterion};
 use opentelemetry::{
     metrics::{Counter, Histogram, MeterProvider as _},
@@ -15,6 +12,9 @@ use opentelemetry_sdk::{
     },
     Resource,
 };
+use rand::Rng;
+use std::sync::{Arc, Weak};
+use std::time::Duration;
 
 #[derive(Clone, Debug)]
 struct SharedReader(Arc<dyn MetricReader>);
@@ -32,8 +32,8 @@ impl MetricReader for SharedReader {
         self.0.force_flush()
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
-        self.0.shutdown()
+    fn shutdown(&self, timeout: Duration) -> OTelSdkResult {
+        self.0.shutdown(timeout)
     }
 
     fn temporality(&self, kind: InstrumentKind) -> Temporality {

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -132,7 +132,6 @@
     message_sender: SyncSender<BatchMessage>, // Control channel to store control messages for the worker thread
     handle: Mutex<Option<thread::JoinHandle<()>>>,
     forceflush_timeout: Duration,
-    shutdown_timeout: Duration,
     export_log_message_sent: Arc<AtomicBool>,
     current_batch_size: Arc<AtomicUsize>,
     max_export_batch_size: usize,
@@ -256,7 +255,7 @@
         }
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, timeout: Duration) -> OTelSdkResult {
         let dropped_logs = self.dropped_logs_count.load(Ordering::Relaxed);
         let max_queue_size = self.max_queue_size;
         if dropped_logs > 0 {
@@ -272,7 +271,7 @@
         match self.message_sender.try_send(BatchMessage::Shutdown(sender)) {
             Ok(_) => {
                 receiver
-                    .recv_timeout(self.shutdown_timeout)
+                    .recv_timeout(timeout)
                     .map(|_| {
                         // join the background thread after receiving back the
                         // shutdown signal
@@ -287,7 +286,7 @@
                                 name: "BatchLogProcessor.Shutdown.Timeout",
                                 message = "BatchLogProcessor shutdown timing out."
                             );
-                            OTelSdkError::Timeout(self.shutdown_timeout)
+                            OTelSdkError::Timeout(timeout)
                         }
                         _ => {
                             otel_error!(
@@ -488,7 +487,6 @@
             message_sender,
             handle: Mutex::new(Some(handle)),
             forceflush_timeout: Duration::from_secs(5), // TODO: make this configurable
-            shutdown_timeout: Duration::from_secs(5),   // TODO: make this configurable
             dropped_logs_count: AtomicUsize::new(0),
             max_queue_size,
             export_log_message_sent: Arc::new(AtomicBool::new(false)),
@@ -961,7 +959,7 @@
 
         processor.emit(&mut record, &instrumentation);
         processor.force_flush().unwrap();
-        processor.shutdown().unwrap();
+        processor.shutdown(Duration::from_secs(5)).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());
@@ -973,27 +971,27 @@
         let exporter = InMemoryLogExporterBuilder::default().build();
         let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
 
-        processor.shutdown().unwrap();
+        processor.shutdown(Duration::from_secs(5)).unwrap();
     }
 
     #[tokio::test(flavor = "current_thread")]
     async fn test_batch_log_processor_shutdown_with_async_runtime_current_flavor_current_thread() {
         let exporter = InMemoryLogExporterBuilder::default().build();
         let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
-        processor.shutdown().unwrap();
+        processor.shutdown(Duration::from_secs(5)).unwrap();
     }
 
     #[tokio::test(flavor = "multi_thread")]
     async fn test_batch_log_processor_shutdown_with_async_runtime_multi_flavor_multi_thread() {
         let exporter = InMemoryLogExporterBuilder::default().build();
         let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
-        processor.shutdown().unwrap();
+        processor.shutdown(Duration::from_secs(5)).unwrap();
     }
 
     #[tokio::test(flavor = "multi_thread")]
     async fn test_batch_log_processor_shutdown_with_async_runtime_multi_flavor_current_thread() {
         let exporter = InMemoryLogExporterBuilder::default().build();
         let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
-        processor.shutdown().unwrap();
+        processor.shutdown(Duration::from_secs(5)).unwrap();
     }
 }

opentelemetry-sdk/src/logs/log_processor.rs

@@ -34,6 +34,7 @@ use opentelemetry::logs::Severity;
 use opentelemetry::InstrumentationScope;
 
 use std::fmt::Debug;
+use std::time::Duration;
 
 /// The interface for plugging into a [`SdkLogger`].
 ///
@@ -56,7 +57,7 @@ pub trait LogProcessor: Send + Sync + Debug {
     /// Shuts down the processor.
     /// After shutdown returns the log processor should stop processing any logs.
     /// It's up to the implementation on when to drop the LogProcessor.
-    fn shutdown(&self) -> OTelSdkResult;
+    fn shutdown(&self, timeout: Duration) -> OTelSdkResult;
     #[cfg(feature = "spec_unstable_logs_enabled")]
     /// Check if logging is enabled
     fn event_enabled(&self, _level: Severity, _target: &str, _name: Option<&str>) -> bool {
@@ -81,6 +82,7 @@ pub(crate) mod tests {
     use opentelemetry::logs::{Logger, LoggerProvider};
     use opentelemetry::{InstrumentationScope, Key};
     use std::sync::{Arc, Mutex};
+    use std::time::Duration;
 
     #[derive(Debug, Clone)]
     pub(crate) struct MockLogExporter {
@@ -138,7 +140,7 @@ pub(crate) mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
             Ok(())
         }
     }
@@ -168,7 +170,7 @@ pub(crate) mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
             Ok(())
         }
     }