Modify loggerprovider shutdown, flush to return single Result and handle repeat shutdown calls (#1750)

Author: cijothomas

opentelemetry-appender-log/examples/logs-basic.rs

@@ -32,5 +32,5 @@ async fn main() {
     warn!("warn!");
     info!("test log!");
 
-    logger_provider.force_flush();
+    let _ = logger_provider.shutdown();
 }

opentelemetry-appender-tracing/examples/basic.rs

@@ -24,5 +24,5 @@ fn main() {
     tracing_subscriber::registry().with(layer).init();
 
     error!(name: "my-event-name", target: "my-system", event_id = 20, user_name = "otel", user_email = "[email protected]");
-    drop(provider);
+    let _ = provider.shutdown();
 }

opentelemetry-sdk/CHANGELOG.md

@@ -31,6 +31,8 @@
 - **Breaking** [#1729](https://github.com/open-telemetry/opentelemetry-rust/pull/1729)
   - Update the return type of `TracerProvider.span_processors()` from `&Vec<Box<dyn SpanProcessor>>` to `&[Box<dyn SpanProcessor>]`.
   - Update the return type of `LoggerProvider.log_processors()` from `&Vec<Box<dyn LogProcessor>>` to `&[Box<dyn LogProcessor>]`.
+- **Breaking** [#1750](https://github.com/open-telemetry/opentelemetry-rust/pull/1729)
+  - Update the return type of `LoggerProvider.shutdown()` from `Vec<LogResult<()>>` to `LogResult<()>`.
 
 ## v0.22.1
 

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -4,16 +4,19 @@ use crate::{
     runtime::RuntimeChannel,
 };
 use opentelemetry::{
-    global::{self},
-    logs::LogResult,
+    global,
+    logs::{LogError, LogResult},
     trace::TraceContextExt,
     Context, InstrumentationLibrary,
 };
 
 #[cfg(feature = "logs_level_enabled")]
 use opentelemetry::logs::Severity;
 
-use std::{borrow::Cow, sync::Arc};
+use std::{
+    borrow::Cow,
+    sync::{atomic::Ordering, Arc},
+};
 use std::{sync::atomic::AtomicBool, time::SystemTime};
 
 use once_cell::sync::Lazy;
@@ -105,17 +108,29 @@ impl LoggerProvider {
     }
 
     /// Shuts down this `LoggerProvider`
-    pub fn shutdown(&self) -> Vec<LogResult<()>> {
-        // mark itself as already shutdown
-        self.is_shutdown
-            .store(true, std::sync::atomic::Ordering::Relaxed);
-        // propagate the shutdown signal to processors
-        // it's up to the processor to properly block new logs after shutdown
-        self.inner
-            .processors
-            .iter()
-            .map(|processor| processor.shutdown())
-            .collect()
+    pub fn shutdown(&self) -> LogResult<()> {
+        if self
+            .is_shutdown
+            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
+            .is_ok()
+        {
+            // propagate the shutdown signal to processors
+            // it's up to the processor to properly block new logs after shutdown
+            let mut errs = vec![];
+            for processor in &self.inner.processors {
+                if let Err(err) = processor.shutdown() {
+                    errs.push(err);
+                }
+            }
+
+            if errs.is_empty() {
+                Ok(())
+            } else {
+                Err(LogError::Other(format!("{errs:?}").into()))
+            }
+        } else {
+            Err(LogError::Other("logger provider already shut down".into()))
+        }
     }
 }
 
@@ -485,6 +500,25 @@ mod tests {
         assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 3);
     }
 
+    #[test]
+    fn shutdown_idempotent_test() {
+        let counter = Arc::new(AtomicU64::new(0));
+        let logger_provider = LoggerProvider::builder()
+            .with_log_processor(ShutdownTestLogProcessor::new(counter.clone()))
+            .build();
+
+        let shutdown_res = logger_provider.shutdown();
+        assert!(shutdown_res.is_ok());
+
+        // Subsequent shutdowns should return an error.
+        let shutdown_res = logger_provider.shutdown();
+        assert!(shutdown_res.is_err());
+
+        // Subsequent shutdowns should return an error.
+        let shutdown_res = logger_provider.shutdown();
+        assert!(shutdown_res.is_err());
+    }
+
     #[test]
     fn global_shutdown_test() {
         // cargo test shutdown_test --features=logs
@@ -508,7 +542,7 @@ mod tests {
 
         // explicitly calling shutdown on logger_provider. This will
         // indeed do the shutdown, even if there are loggers still alive.
-        logger_provider.shutdown();
+        let _ = logger_provider.shutdown();
 
         // Assert
 

opentelemetry-sdk/src/logs/log_processor.rs

@@ -729,7 +729,7 @@ mod tests {
             ])))
             .build();
         assert_eq!(exporter.get_resource().unwrap().into_iter().count(), 4);
-        provider.shutdown();
+        let _ = provider.shutdown();
     }
 
     #[tokio::test(flavor = "multi_thread")]

opentelemetry-stdout/examples/basic.rs

@@ -107,7 +107,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     meter_provider.shutdown()?;
 
     #[cfg(feature = "logs")]
-    drop(logger_provider);
+    logger_provider.shutdown()?;
 
     Ok(())
 }