feat: add shutdown with timeout for log exporter

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

@@ -3,6 +3,7 @@
 use opentelemetry::otel_debug;
 use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
+use std::time;
 
 impl LogExporter for OtlpHttpClient {
     async fn export(&self, batch: LogBatch<'_>) -> OTelSdkResult {
@@ -46,7 +47,7 @@
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
         let mut client_guard = self.client.lock().map_err(|e| {
             OTelSdkError::InternalFailure(format!("Failed to acquire client lock: {}", e))
         })?;
@@ -58,6 +59,10 @@
         Ok(())
     }
 
+    fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(time::Duration::from_secs(5))
+    }
+
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
         self.resource = resource.into();
     }

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

@@ -5,6 +5,7 @@
 };
 use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
+use std::time;
 use tokio::sync::Mutex;
 use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
 
@@ -84,7 +85,7 @@
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
         // TODO: Implement actual shutdown
         // Due to the use of tokio::sync::Mutex to guard
         // the inner client, we need to await the call to lock the mutex

opentelemetry-otlp/src/logs.rs

@@ -4,9 +4,9 @@
 
 #[cfg(feature = "grpc-tonic")]
 use opentelemetry::otel_debug;
-use std::fmt::Debug;
-
 use opentelemetry_sdk::{error::OTelSdkResult, logs::LogBatch};
+use std::fmt::Debug;
+use std::time;
 
 use crate::{ExporterBuildError, HasExportConfig, NoExporterBuilderSet};
 
@@ -157,7 +157,7 @@
         }
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
         match &self.client {
             #[cfg(feature = "grpc-tonic")]
             SupportedTransportClient::Tonic(client) => client.shutdown(),

opentelemetry-sdk/CHANGELOG.md

@@ -25,6 +25,7 @@ also modified to suppress telemetry before invoking exporters.
   - Fixed the overflow attribute to correctly use the boolean value `true`
     instead of the string `"true"`.
     [#2878](https://github.com/open-telemetry/opentelemetry-rust/issues/2878)
+- *Breaking* The `shutdown_with_timeout` method is added to LogExporter trait. This is breaking change for custom `LogExporter` authors.
 - *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/benches/log_enabled.rs

@@ -21,6 +21,7 @@ use opentelemetry_sdk::logs::{
 use opentelemetry_sdk::Resource;
 #[cfg(not(target_os = "windows"))]
 use pprof::criterion::{Output, PProfProfiler};
+use std::time;
 
 #[derive(Debug)]
 struct NoopExporter;
@@ -29,7 +30,7 @@ impl LogExporter for NoopExporter {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
         Ok(())
     }
 

opentelemetry-sdk/src/logs/export.rs

@@ -6,6 +6,7 @@ use crate::Resource;
 use opentelemetry::logs::Severity;
 use opentelemetry::InstrumentationScope;
 use std::fmt::Debug;
+use std::time;
 
 /// A batch of log records to be exported by a `LogExporter`.
 ///
@@ -134,10 +135,11 @@ pub trait LogExporter: Send + Sync + Debug {
         &self,
         batch: LogBatch<'_>,
     ) -> impl std::future::Future<Output = OTelSdkResult> + Send;
-
     /// Shuts down the exporter.
+    fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult;
+    /// Shuts down the exporter with a default timeout.
     fn shutdown(&self) -> OTelSdkResult {
-        Ok(())
+        self.shutdown_with_timeout(time::Duration::from_secs(5))
     }
     #[cfg(feature = "spec_unstable_logs_enabled")]
     /// Check if logs are enabled.

opentelemetry-sdk/src/logs/in_memory_exporter.rs

@@ -7,6 +7,7 @@ use opentelemetry::InstrumentationScope;
 use std::borrow::Cow;
 use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, Mutex};
+use std::time;
 
 /// An in-memory logs exporter that stores logs data in memory..
 ///
@@ -205,7 +206,7 @@ impl LogExporter for InMemoryLogExporter {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
         self.shutdown_called
             .store(true, std::sync::atomic::Ordering::Relaxed);
         if self.should_reset_on_shutdown {
@@ -214,6 +215,10 @@ impl LogExporter for InMemoryLogExporter {
         Ok(())
     }
 
+    fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(time::Duration::from_secs(5))
+    }
+
     fn set_resource(&mut self, resource: &Resource) {
         let mut res_guard = self.resource.lock().expect("Resource lock poisoned");
         *res_guard = resource.clone();

opentelemetry-sdk/src/logs/log_processor.rs

@@ -81,6 +81,7 @@ pub(crate) mod tests {
     use opentelemetry::logs::{Logger, LoggerProvider};
     use opentelemetry::{InstrumentationScope, Key};
     use std::sync::{Arc, Mutex};
+    use std::time;
 
     #[derive(Debug, Clone)]
     pub(crate) struct MockLogExporter {
@@ -92,7 +93,7 @@ pub(crate) mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
             Ok(())
         }
 

opentelemetry-sdk/src/logs/log_processor_with_async_runtime.rs

@@ -309,6 +309,7 @@ mod tests {
     use opentelemetry::KeyValue;
     use opentelemetry::{InstrumentationScope, Key};
     use std::sync::{Arc, Mutex};
+    use std::time;
     use std::time::Duration;
 
     #[derive(Debug, Clone)]
@@ -321,10 +322,14 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
             Ok(())
         }
 
+        fn shutdown(&self) -> OTelSdkResult {
+            self.shutdown_with_timeout(time::Duration::from_secs(5))
+        }
+
         fn set_resource(&mut self, resource: &Resource) {
             self.resource
                 .lock()

opentelemetry-sdk/src/logs/logger_provider.rs

@@ -291,7 +291,7 @@
     use std::fmt::{Debug, Formatter};
     use std::sync::atomic::AtomicU64;
     use std::sync::Mutex;
-    use std::thread;
+    use std::{thread, time};
 
     struct ShutdownTestLogProcessor {
         is_shutdown: Arc<Mutex<bool>>,
@@ -364,7 +364,7 @@
             *res = resource.clone();
         }
 
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
             Ok(())
         }
     }

opentelemetry-sdk/src/logs/simple_log_processor.rs

@@ -164,6 +164,7 @@
     use opentelemetry::KeyValue;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::sync::{Arc, Mutex};
+    use std::time;
     use std::time::Duration;
 
     #[derive(Debug, Clone)]
@@ -195,6 +196,9 @@
             }
             Ok(())
         }
+        fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
+            Ok(())
+        }
     }
 
     #[test]
@@ -460,7 +464,7 @@
     }
 
     impl LogExporter for ReentrantLogExporter {
-        fn shutdown(&self) -> OTelSdkResult {
+        fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
             Ok(())
         }
 

opentelemetry-stdout/src/logs/exporter.rs

@@ -5,6 +5,7 @@
 use opentelemetry_sdk::Resource;
 use std::sync::atomic;
 use std::sync::atomic::Ordering;
+use std::time;
 
 /// An OpenTelemetry exporter that writes Logs to stdout on export.
 pub struct LogExporter {
@@ -57,7 +58,7 @@
         }
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
         self.is_shutdown.store(true, atomic::Ordering::SeqCst);
         Ok(())
     }

stress/src/logs.rs

@@ -24,6 +24,7 @@ use opentelemetry_sdk::logs::concurrent_log_processor::SimpleConcurrentLogProces
 use opentelemetry_sdk::logs::SdkLoggerProvider;
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
 use opentelemetry_sdk::Resource;
+use std::time;
 use tracing::error;
 use tracing_subscriber::prelude::*;
 
@@ -52,7 +53,7 @@ impl LogExporter for NoopExporter {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
         Ok(())
     }