feat: add-shutdown-with-timeout-for-log-provider-and-processor (#2941)

Co-authored-by: Lalit Kumar Bhasin <[email protected]>
Co-authored-by: Cijo Thomas <[email protected]>

Author: 3 people

examples/tracing-http-propagator/src/server.rs

@@ -120,10 +120,6 @@ impl LogProcessor for EnrichWithBaggageLogProcessor {
     fn force_flush(&self) -> OTelSdkResult {
         Ok(())
     }
-
-    fn shutdown(&self) -> OTelSdkResult {
-        Ok(())
-    }
 }
 
 /// A custom span processor that enriches spans with baggage attributes. Baggage

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

@@ -43,10 +43,6 @@ impl LogProcessor for NoopProcessor {
     fn force_flush(&self) -> OTelSdkResult {
         Ok(())
     }
-
-    fn shutdown(&self) -> OTelSdkResult {
-        Ok(())
-    }
 }
 
 /// Creates a single benchmark for a specific number of attributes

opentelemetry-appender-tracing/benches/logs.rs

@@ -54,10 +54,6 @@ impl LogProcessor for NoopProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
-        Ok(())
-    }
-
     fn event_enabled(
         &self,
         _level: opentelemetry::logs::Severity,

opentelemetry-appender-tracing/src/layer.rs

@@ -877,10 +877,6 @@ mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
-
-        fn shutdown(&self) -> OTelSdkResult {
-            Ok(())
-        }
     }
 
     #[cfg(feature = "spec_unstable_logs_enabled")]

opentelemetry-proto/src/transform/logs.rs

@@ -239,10 +239,6 @@ mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
-
-        fn shutdown(&self) -> OTelSdkResult {
-            Ok(())
-        }
     }
 
     fn create_test_log_data(

opentelemetry-sdk/CHANGELOG.md

@@ -26,6 +26,7 @@ also modified to suppress telemetry before invoking exporters.
     instead of the string `"true"`.
     [#2878](https://github.com/open-telemetry/opentelemetry-rust/issues/2878)
 - The `shutdown_with_timeout` method is added to LogExporter trait.
+- The `shutdown_with_timeout` method is added to LogProvider and LogProcessor trait.
 - *Breaking* `MetricError`, `MetricResult` no longer public (except when
   `spec_unstable_metrics_views` feature flag is enabled). `OTelSdkResult` should
   be used instead, wherever applicable. [#2906](https://github.com/open-telemetry/opentelemetry-rust/pull/2906)

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -132,7 +132,6 @@ pub struct BatchLogProcessor {
     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 @@ impl LogProcessor for BatchLogProcessor {
         }
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&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 @@ impl LogProcessor for BatchLogProcessor {
         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 @@ impl LogProcessor for BatchLogProcessor {
                                 name: "BatchLogProcessor.Shutdown.Timeout",
                                 message = "BatchLogProcessor shutdown timing out."
                             );
-                            OTelSdkError::Timeout(self.shutdown_timeout)
+                            OTelSdkError::Timeout(timeout)
                         }
                         _ => {
                             otel_error!(
@@ -489,7 +488,6 @@ impl BatchLogProcessor {
             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)),

opentelemetry-sdk/src/logs/concurrent_log_processor.rs

@@ -1,4 +1,5 @@
 use opentelemetry::{otel_info, InstrumentationScope};
+use std::time::Duration;
 
 use crate::{error::OTelSdkResult, Resource};
 
@@ -43,8 +44,8 @@ impl<T: LogExporter> LogProcessor for SimpleConcurrentLogProcessor<T> {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
-        self.exporter.shutdown()
+    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
+        self.exporter.shutdown_with_timeout(timeout)
     }
 
     #[cfg(feature = "spec_unstable_logs_enabled")]

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,13 @@ 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_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+        Ok(())
+    }
+    /// Shuts down the processor with default timeout.
+    fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(Duration::from_secs(5))
+    }
     #[cfg(feature = "spec_unstable_logs_enabled")]
     /// Check if logging is enabled
     fn event_enabled(&self, _level: Severity, _target: &str, _name: Option<&str>) -> bool {
@@ -133,10 +140,6 @@ pub(crate) mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
-
-        fn shutdown(&self) -> OTelSdkResult {
-            Ok(())
-        }
     }
 
     #[derive(Debug)]
@@ -163,10 +166,6 @@ pub(crate) mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
-
-        fn shutdown(&self) -> OTelSdkResult {
-            Ok(())
-        }
     }
 
     #[test]

opentelemetry-sdk/src/logs/log_processor_with_async_runtime.rs

@@ -628,10 +628,6 @@ mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
-
-        fn shutdown(&self) -> OTelSdkResult {
-            Ok(())
-        }
     }
 
     #[derive(Debug)]
@@ -658,10 +654,6 @@ mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
-
-        fn shutdown(&self) -> OTelSdkResult {
-            Ok(())
-        }
     }
     #[test]
     fn test_log_data_modification_by_multiple_processors() {