add shutdown with timeout for metric provider

Author: mohammadVatandoost

opentelemetry-appender-tracing/benches/logs.rs

@@ -12,7 +12,6 @@
     | ot_layer_disabled           | 19 ns       |
     | ot_layer_enabled            | 196 ns      |
 */
-
 use criterion::{criterion_group, criterion_main, Criterion};
 use opentelemetry::InstrumentationScope;
 use opentelemetry_appender_tracing::layer as tracing_layer;
@@ -21,6 +20,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;
@@ -44,7 +44,7 @@ impl LogProcessor for NoopProcessor {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 

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

@@ -1,5 +1,3 @@
-use std::time::Duration;
-
 use super::OtlpHttpClient;
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
@@ -48,7 +46,7 @@ impl LogExporter for OtlpHttpClient {
         Ok(())
     }
 
-    fn shutdown(&mut self, _timeout: Duration) -> OTelSdkResult {
+    fn shutdown(&mut 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

@@ -1,5 +1,4 @@
 use core::fmt;
-use std::time::Duration;
 use opentelemetry::otel_debug;
 use opentelemetry_proto::tonic::collector::logs::v1::{
     logs_service_client::LogsServiceClient, ExportLogsServiceRequest,
@@ -87,7 +86,7 @@ impl LogExporter for TonicLogsClient {
         Ok(())
     }
 
-    fn shutdown(&mut self, _timeout: Duration) -> OTelSdkResult {
+    fn shutdown(&mut self) -> OTelSdkResult {
         match self.inner.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-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_processor.rs

@@ -14,7 +14,8 @@
 use opentelemetry::time::now;
 use std::{
     sync::{Arc, Mutex},
-    thread::sleep, time::Duration,
+    thread::sleep,
+    time::Duration,
 };
 
 use criterion::{criterion_group, criterion_main, Criterion};

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

@@ -435,7 +435,7 @@ impl BatchLogProcessor {
                                 &current_batch_size,
                                 &config,
                             );
-                            let _ = exporter.shutdown(Duration::from_secs(5));
+                            let _ = exporter.shutdown();
                             let _ = sender.send(result);
 
                             otel_debug!(

opentelemetry-sdk/src/logs/export.rs

@@ -6,7 +6,6 @@ use crate::Resource;
 use opentelemetry::logs::Severity;
 use opentelemetry::InstrumentationScope;
 use std::fmt::Debug;
-use std::time::Duration;
 
 /// A batch of log records to be exported by a `LogExporter`.
 ///
@@ -137,7 +136,7 @@ pub trait LogExporter: Send + Sync + Debug {
     ) -> impl std::future::Future<Output = OTelSdkResult> + Send;
 
     /// Shuts down the exporter.
-    fn shutdown(&mut self, _timeout: Duration) -> OTelSdkResult {
+    fn shutdown(&mut self) -> OTelSdkResult {
         Ok(())
     }
     #[cfg(feature = "spec_unstable_logs_enabled")]

opentelemetry-sdk/src/logs/in_memory_exporter.rs

@@ -6,7 +6,6 @@ use opentelemetry::InstrumentationScope;
 use std::borrow::Cow;
 use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
 
 type LogResult<T> = Result<T, OTelSdkError>;
 
@@ -212,7 +211,7 @@ impl LogExporter for InMemoryLogExporter {
         Ok(())
     }
 
-    fn shutdown(&mut self, _timeout: Duration) -> OTelSdkResult {
+    fn shutdown(&mut 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

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