feat: Add shutdown with timeout for metric exporter (#2854)

Co-authored-by: Braden Steffaniak <[email protected]>
Co-authored-by: Cijo Thomas <[email protected]>
Co-authored-by: Cijo Thomas <[email protected]>

Author: 3 people

opentelemetry-otlp/src/metric.rs

@@ -21,6 +21,7 @@ use opentelemetry_sdk::metrics::{
     data::ResourceMetrics, exporter::PushMetricExporter, Temporality,
 };
 use std::fmt::{Debug, Formatter};
+use std::time::Duration;
 
 /// Target to which the exporter is going to send metrics, defaults to https://localhost:4317/v1/metrics.
 /// Learn about the relationship between this constant and default/spans/logs at
@@ -163,6 +164,10 @@ impl PushMetricExporter for MetricExporter {
     }
 
     fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(Duration::from_secs(5))
+    }
+
+    fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
         match &self.client {
             #[cfg(feature = "grpc-tonic")]
             SupportedTransportClient::Tonic(client) => client.shutdown(),

opentelemetry-sdk/CHANGELOG.md

@@ -68,6 +68,7 @@ Released 2025-Mar-21
 
   Custom exporters will need to internally synchronize any mutable state, if applicable.
 
+- **Breaking** The `shutdown_with_timeout` method is added to MetricExporter trait. This is breaking change for custom `MetricExporter` authors.
 - Bug Fix: `BatchLogProcessor` now correctly calls `shutdown` on the exporter
   when its `shutdown` is invoked.
 

opentelemetry-sdk/src/metrics/exporter.rs

@@ -1,5 +1,7 @@
 //! Interfaces for exporting metrics
+
 use crate::error::OTelSdkResult;
+use std::time::Duration;
 
 use crate::metrics::data::ResourceMetrics;
 
@@ -26,7 +28,12 @@ pub trait PushMetricExporter: Send + Sync + 'static {
     ///
     /// After Shutdown is called, calls to Export will perform no operation and
     /// instead will return an error indicating the shutdown state.
-    fn shutdown(&self) -> OTelSdkResult;
+    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult;
+
+    /// Shutdown with the default timeout of 5 seconds.
+    fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(Duration::from_secs(5))
+    }
 
     /// Access the [Temporality] of the MetricExporter.
     fn temporality(&self) -> Temporality;

opentelemetry-sdk/src/metrics/in_memory_exporter.rs

@@ -8,6 +8,7 @@ use crate::InMemoryExporterError;
 use std::collections::VecDeque;
 use std::fmt;
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
 
 use super::data::{AggregatedMetrics, Metric, ScopeMetrics};
 
@@ -253,6 +254,10 @@ impl PushMetricExporter for InMemoryMetricExporter {
         Ok(())
     }
 
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+        Ok(())
+    }
+
     fn temporality(&self) -> Temporality {
         self.temporality
     }

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -566,6 +566,10 @@ mod tests {
             Ok(())
         }
 
+        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+            Ok(())
+        }
+
         fn temporality(&self) -> Temporality {
             Temporality::Cumulative
         }
@@ -586,6 +590,10 @@ mod tests {
         }
 
         fn shutdown(&self) -> OTelSdkResult {
+            self.shutdown_with_timeout(Duration::from_secs(5))
+        }
+
+        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
             self.is_shutdown.store(true, Ordering::Relaxed);
             Ok(())
         }

opentelemetry-stdout/src/metrics/exporter.rs

@@ -14,6 +14,7 @@ use opentelemetry_sdk::{
 };
 use std::fmt::Debug;
 use std::sync::atomic;
+use std::time::Duration;
 
 /// An OpenTelemetry exporter that writes to stdout on export.
 pub struct MetricExporter {
@@ -65,6 +66,10 @@ impl PushMetricExporter for MetricExporter {
     }
 
     fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(Duration::from_secs(5))
+    }
+
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         self.is_shutdown.store(true, atomic::Ordering::SeqCst);
         Ok(())
     }