feat: add shutdown with timeout for metric reader and provider (#2890)

Co-authored-by: Lalit Kumar Bhasin <[email protected]>
Co-authored-by: Cijo Thomas <[email protected]>

Author: 3 people

opentelemetry-sdk/CHANGELOG.md

@@ -78,6 +78,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.
+- **Breaking** The `shutdown_with_timeout` method is added to MetricReader trait. This is breaking change for custom `MetricReader` authors.
 - Bug Fix: `BatchLogProcessor` now correctly calls `shutdown` on the exporter
   when its `shutdown` is invoked.
 

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,7 +32,7 @@ impl MetricReader for SharedReader {
         self.0.force_flush()
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         self.0.shutdown()
     }
 

opentelemetry-sdk/src/metrics/manual_reader.rs

@@ -1,10 +1,10 @@
+use opentelemetry::otel_debug;
+use std::time::Duration;
 use std::{
     fmt,
     sync::{Mutex, Weak},
 };
 
-use opentelemetry::otel_debug;
-
 use crate::{
     error::{OTelSdkError, OTelSdkResult},
     metrics::{MetricError, MetricResult, Temporality},
@@ -110,7 +110,7 @@ impl MetricReader for ManualReader {
     }
 
     /// Closes any connections and frees any resources used by the reader.
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         let mut inner = self
             .inner
             .lock()

opentelemetry-sdk/src/metrics/meter_provider.rs

@@ -1,4 +1,9 @@
 use core::fmt;
+use opentelemetry::{
+    metrics::{Meter, MeterProvider},
+    otel_debug, otel_error, otel_info, InstrumentationScope,
+};
+use std::time::Duration;
 use std::{
     collections::HashMap,
     sync::{
@@ -7,11 +12,6 @@ use std::{
     },
 };
 
-use opentelemetry::{
-    metrics::{Meter, MeterProvider},
-    otel_debug, otel_error, otel_info, InstrumentationScope,
-};
-
 use crate::error::OTelSdkResult;
 use crate::Resource;
 
@@ -109,13 +109,18 @@ impl SdkMeterProvider {
     ///
     /// There is no guaranteed that all telemetry be flushed or all resources have
     /// been released on error.
-    pub fn shutdown(&self) -> OTelSdkResult {
+    pub fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         otel_debug!(
             name: "MeterProvider.Shutdown",
             message = "User initiated shutdown of MeterProvider."
         );
         self.inner.shutdown()
     }
+
+    /// shutdown with default timeout
+    pub fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(Duration::from_secs(5))
+    }
 }
 
 impl SdkMeterProviderInner {
@@ -130,7 +135,7 @@ impl SdkMeterProviderInner {
         }
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         if self
             .shutdown_invoked
             .swap(true, std::sync::atomic::Ordering::SeqCst)
@@ -141,6 +146,10 @@ impl SdkMeterProviderInner {
             self.pipes.shutdown()
         }
     }
+
+    fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(Duration::from_secs(5))
+    }
 }
 
 impl Drop for SdkMeterProviderInner {

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -491,7 +491,7 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
     // completion, and avoid blocking the thread. The default shutdown on drop
     // can still use blocking call. If user already explicitly called shutdown,
     // drop won't call shutdown again.
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         self.inner.shutdown()
     }
 

opentelemetry-sdk/src/metrics/periodic_reader_with_async_runtime.rs

@@ -390,7 +390,7 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
             .and_then(|res| res)
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         let mut inner = self
             .inner
             .lock()

opentelemetry-sdk/src/metrics/reader.rs

@@ -1,7 +1,7 @@
 //! Interfaces for reading and producing metrics
-use std::{fmt, sync::Weak};
-
 use crate::{error::OTelSdkResult, metrics::MetricResult};
+use std::time::Duration;
+use std::{fmt, sync::Weak};
 
 use super::{data::ResourceMetrics, pipeline::Pipeline, InstrumentKind, Temporality};
 
@@ -46,7 +46,12 @@ pub trait MetricReader: fmt::Debug + Send + Sync + 'static {
     ///
     /// After `shutdown` is called, calls to `collect` 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 default timeout
+    fn shutdown(&self) -> OTelSdkResult {
+        self.shutdown_with_timeout(Duration::from_secs(5))
+    }
 
     /// The output temporality, a function of instrument kind.
     /// This SHOULD be obtained from the exporter.

opentelemetry-sdk/src/testing/metrics/metric_reader.rs

@@ -1,10 +1,10 @@
-use std::sync::{Arc, Mutex, Weak};
-
 use crate::error::{OTelSdkError, OTelSdkResult};
 use crate::metrics::{
     data::ResourceMetrics, pipeline::Pipeline, reader::MetricReader, InstrumentKind,
 };
 use crate::metrics::{MetricResult, Temporality};
+use std::sync::{Arc, Mutex, Weak};
+use std::time::Duration;
 
 #[derive(Debug, Clone)]
 pub struct TestMetricReader {
@@ -42,7 +42,7 @@ impl MetricReader for TestMetricReader {
         Ok(())
     }
 
-    fn shutdown(&self) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         let result = self.force_flush();
         {
             let mut is_shutdown = self.is_shutdown.lock().unwrap();