fix: Remove mut ref requirement for shutdown LogExporter

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

@@ -46,7 +46,7 @@
         Ok(())
     }
 
-    fn shutdown(&mut self) -> OTelSdkResult {
+    fn shutdown(&self) -> OTelSdkResult {
         let mut client_guard = self.client.lock().map_err(|e| {
             OTelSdkError::InternalFailure(format!("Failed to acquire client lock: {}", e))
         })?;

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

@@ -7,21 +7,20 @@
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
 use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
 
-use opentelemetry_proto::transform::logs::tonic::group_logs_by_resource_and_scope;
-
 use super::BoxInterceptor;
-use tokio::sync::Mutex;
+use opentelemetry_proto::transform::logs::tonic::group_logs_by_resource_and_scope;
+use std::sync::Mutex;
 
 pub(crate) struct TonicLogsClient {
-    inner: Option<ClientInner>,
+    inner: Mutex<Option<ClientInner>>,
     #[allow(dead_code)]
     // <allow dead> would be removed once we support set_resource for metrics.
     resource: opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema,
 }
 
 struct ClientInner {
     client: LogsServiceClient<Channel>,
-    interceptor: Mutex<BoxInterceptor>,
+    interceptor: BoxInterceptor,
 }
 
 impl fmt::Debug for TonicLogsClient {
@@ -46,30 +45,36 @@
         otel_debug!(name: "TonicsLogsClientBuilt");
 
         TonicLogsClient {
-            inner: Some(ClientInner {
+            inner: Mutex::new(Some(ClientInner {
                 client,
-                interceptor: Mutex::new(interceptor),
-            }),
+                interceptor,
+            })),
             resource: Default::default(),
         }
     }
 }
 
 impl LogExporter for TonicLogsClient {
     async fn export(&self, batch: LogBatch<'_>) -> OTelSdkResult {
-        let (mut client, metadata, extensions) = match &self.inner {
-            Some(inner) => {
-                let (m, e, _) = inner
-                    .interceptor
-                    .lock()
-                    .await // tokio::sync::Mutex doesn't return a poisoned error, so we can safely use the interceptor here
-                    .call(Request::new(()))
-                    .map_err(|e| OTelSdkError::InternalFailure(format!("error: {:?}", e)))?
-                    .into_parts();
-                (inner.client.clone(), m, e)
-            }
-            None => return Err(OTelSdkError::AlreadyShutdown),
-        };
+        let (mut client, metadata, extensions) = self
+            .inner
+            .lock()
+            .map_err(|e| OTelSdkError::InternalFailure(format!("Failed to acquire lock: {e:?}")))
+            .and_then(|mut inner| match &mut *inner {
+                Some(inner) => {
+                    let (m, e, _) = inner
+                        .interceptor
+                        .call(Request::new(()))
+                        .map_err(|e| {
+                            OTelSdkError::InternalFailure(format!(
+                                "unexpected status while exporting {e:?}"
+                            ))
+                        })?
+                        .into_parts();
+                    Ok((inner.client.clone(), m, e))
+                }
+                None => Err(OTelSdkError::AlreadyShutdown),
+            })?;
 
         let resource_logs = group_logs_by_resource_and_scope(batch, &self.resource);
 
@@ -86,8 +91,13 @@
         Ok(())
     }
 
-    fn shutdown(&mut self) -> OTelSdkResult {
-        match self.inner.take() {
+    fn shutdown(&self) -> OTelSdkResult {
+        match self
+            .inner
+            .lock()
+            .map_err(|e| OTelSdkError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
+            .take()
+        {
             Some(_) => Ok(()), // Successfully took `inner`, indicating a successful shutdown.
             None => Err(OTelSdkError::AlreadyShutdown), // `inner` was already `None`, meaning it's already shut down.
         }

opentelemetry-sdk/src/logs/export.rs

@@ -136,7 +136,7 @@
     ) -> impl std::future::Future<Output = OTelSdkResult> + Send;
 
     /// Shuts down the exporter.
-    fn shutdown(&mut self) -> OTelSdkResult {
+    fn shutdown(&self) -> OTelSdkResult {
         Ok(())
     }
     #[cfg(feature = "spec_unstable_logs_enabled")]

opentelemetry-sdk/src/logs/in_memory_exporter.rs

@@ -211,7 +211,7 @@ impl LogExporter for InMemoryLogExporter {
         Ok(())
     }
 
-    fn shutdown(&mut self) -> OTelSdkResult {
+    fn shutdown(&self) -> OTelSdkResult {
         self.shutdown_called
             .store(true, std::sync::atomic::Ordering::Relaxed);
         if self.should_reset_on_shutdown {

opentelemetry-sdk/src/logs/log_processor.rs

@@ -92,7 +92,7 @@ pub(crate) mod tests {
             Ok(())
         }
 
-        fn shutdown(&mut self) -> OTelSdkResult {
+        fn shutdown(&self) -> OTelSdkResult {
             Ok(())
         }
 

opentelemetry-sdk/src/logs/log_processor_with_async_runtime.rs

@@ -321,7 +321,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&mut self) -> OTelSdkResult {
+        fn shutdown(&self) -> OTelSdkResult {
             Ok(())
         }
 

opentelemetry-sdk/src/logs/simple_log_processor.rs

@@ -117,7 +117,7 @@ impl<T: LogExporter> LogProcessor for SimpleLogProcessor<T> {
     fn shutdown(&self) -> OTelSdkResult {
         self.is_shutdown
             .store(true, std::sync::atomic::Ordering::Relaxed);
-        if let Ok(mut exporter) = self.exporter.lock() {
+        if let Ok(exporter) = self.exporter.lock() {
             exporter.shutdown()
         } else {
             Err(OTelSdkError::InternalFailure(

opentelemetry-stdout/src/logs/exporter.rs

@@ -57,7 +57,7 @@
         }
     }
 
-    fn shutdown(&mut self) -> OTelSdkResult {
+    fn shutdown(&self) -> OTelSdkResult {
         self.is_shutdown.store(true, atomic::Ordering::SeqCst);
         Ok(())
     }