Use dedicated ShutdownResult for Metric SDK shutdown (#2573)

Author: cijothomas

examples/metrics-basic/src/main.rs

@@ -1,4 +1,5 @@
 use opentelemetry::{global, KeyValue};
+use opentelemetry_sdk::error::ShutdownError;
 use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
 use opentelemetry_sdk::Resource;
 use std::error::Error;
@@ -23,7 +24,7 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
 }
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+async fn main() -> Result<(), Box<dyn Error>> {
     // Initialize the MeterProvider with the stdout Exporter.
     let meter_provider = init_meter_provider();
 
@@ -137,9 +138,41 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         })
         .build();
 
-    // Metrics are exported by default every 30 seconds when using stdout exporter,
-    // however shutting down the MeterProvider here instantly flushes
-    // the metrics, instead of waiting for the 30 sec interval.
+    // Metrics are exported by default every 30 seconds when using stdout
+    // exporter, however shutting down the MeterProvider here instantly flushes
+    // the metrics, instead of waiting for the 30 sec interval. Shutdown returns
+    // a result, which is bubbled up to the caller The commented code below
+    // demonstrates handling the shutdown result, instead of bubbling up the
+    // error.
     meter_provider.shutdown()?;
+
+    // let shutdown_result = meter_provider.shutdown();
+
+    // Handle the shutdown result.
+    // match shutdown_result {
+    //     Ok(_) => println!("MeterProvider shutdown successfully"),
+    //     Err(e) => {
+    //         match e {
+    //             opentelemetry_sdk::error::ShutdownError::InternalFailure(message) => {
+    //                 // This indicates some internal failure during shutdown. The
+    //                 // error message is intended for logging purposes only and
+    //                 // should not be used to make programmatic decisions.
+    //                 println!("MeterProvider shutdown failed: {}", message)
+    //             }
+    //             opentelemetry_sdk::error::ShutdownError::AlreadyShutdown => {
+    //                 // This indicates some user code tried to shutdown
+    //                 // elsewhere. user need to review their code to ensure
+    //                 // shutdown is called only once.
+    //                 println!("MeterProvider already shutdown")
+    //             }
+    //             opentelemetry_sdk::error::ShutdownError::Timeout(e) => {
+    //                 // This indicates the shutdown timed out, and a good hint to
+    //                 // user to increase the timeout. (Shutdown method does not
+    //                 // allow custom timeout today, but that is temporary)
+    //                 println!("MeterProvider shutdown timed out after {:?}", e)
+    //             }
+    //         }
+    //     }
+    // }
     Ok(())
 }

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

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
+use opentelemetry_sdk::error::{ShutdownError, ShutdownResult};
 use opentelemetry_sdk::metrics::data::ResourceMetrics;
 use opentelemetry_sdk::metrics::{MetricError, MetricResult};
 
@@ -43,8 +44,11 @@ impl MetricsClient for OtlpHttpClient {
         Ok(())
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
-        let _ = self.client.lock()?.take();
+    fn shutdown(&self) -> ShutdownResult {
+        self.client
+            .lock()
+            .map_err(|e| ShutdownError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
+            .take();
 
         Ok(())
     }

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

@@ -6,6 +6,7 @@ use opentelemetry::otel_debug;
 use opentelemetry_proto::tonic::collector::metrics::v1::{
     metrics_service_client::MetricsServiceClient, ExportMetricsServiceRequest,
 };
+use opentelemetry_sdk::error::{ShutdownError, ShutdownResult};
 use opentelemetry_sdk::metrics::data::ResourceMetrics;
 use opentelemetry_sdk::metrics::{MetricError, MetricResult};
 use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
@@ -89,8 +90,11 @@ impl MetricsClient for TonicMetricsClient {
         Ok(())
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
-        let _ = self.inner.lock()?.take();
+    fn shutdown(&self) -> ShutdownResult {
+        self.inner
+            .lock()
+            .map_err(|e| ShutdownError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
+            .take();
 
         Ok(())
     }

opentelemetry-otlp/src/metric.rs

@@ -16,6 +16,7 @@ use crate::NoExporterBuilderSet;
 
 use async_trait::async_trait;
 use core::fmt;
+use opentelemetry_sdk::error::ShutdownResult;
 use opentelemetry_sdk::metrics::MetricResult;
 
 use opentelemetry_sdk::metrics::{
@@ -123,7 +124,7 @@ impl HasHttpConfig for MetricExporterBuilder<HttpExporterBuilderSet> {
 #[async_trait]
 pub(crate) trait MetricsClient: fmt::Debug + Send + Sync + 'static {
     async fn export(&self, metrics: &mut ResourceMetrics) -> MetricResult<()>;
-    fn shutdown(&self) -> MetricResult<()>;
+    fn shutdown(&self) -> ShutdownResult;
 }
 
 /// Export metrics in OTEL format.
@@ -149,7 +150,7 @@ impl PushMetricExporter for MetricExporter {
         Ok(())
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         self.client.shutdown()
     }
 

opentelemetry-sdk/benches/metric.rs

@@ -7,6 +7,7 @@ use opentelemetry::{
     Key, KeyValue,
 };
 use opentelemetry_sdk::{
+    error::ShutdownResult,
     metrics::{
         data::ResourceMetrics, new_view, reader::MetricReader, Aggregation, Instrument,
         InstrumentKind, ManualReader, MetricResult, Pipeline, SdkMeterProvider, Stream,
@@ -31,7 +32,7 @@ impl MetricReader for SharedReader {
         self.0.force_flush()
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         self.0.shutdown()
     }
 

opentelemetry-sdk/src/error.rs

@@ -1,7 +1,45 @@
 //! Wrapper for error from trace, logs and metrics part of open telemetry.
 
+use std::{result::Result, time::Duration};
+
+use thiserror::Error;
+
 /// Trait for errors returned by exporters
 pub trait ExportError: std::error::Error + Send + Sync + 'static {
     /// The name of exporter that returned this error
     fn exporter_name(&self) -> &'static str;
 }
+
+#[derive(Error, Debug)]
+/// Errors that can occur during shutdown.
+pub enum ShutdownError {
+    /// Shutdown has already been invoked.
+    ///
+    /// While shutdown is idempotent and calling it multiple times has no
+    /// impact, this error suggests that another part of the application is
+    /// invoking `shutdown` earlier than intended. Users should review their
+    /// code to identify unintended or duplicate shutdown calls and ensure it is
+    /// only triggered once at the correct place.
+    #[error("Shutdown already invoked")]
+    AlreadyShutdown,
+
+    /// Shutdown timed out before completing.
+    ///
+    /// This does not necessarily indicate a failure—shutdown may still be
+    /// complete. If this occurs frequently, consider increasing the timeout
+    /// duration to allow more time for completion.
+    #[error("Shutdown timed out after {0:?}")]
+    Timeout(Duration),
+
+    /// Shutdown failed due to an internal error.
+    ///
+    /// The error message is intended for logging purposes only and should not
+    /// be used to make programmatic decisions. It is implementation-specific
+    /// and subject to change without notice. Consumers of this error should not
+    /// rely on its content beyond logging.
+    #[error("Shutdown failed: {0}")]
+    InternalFailure(String),
+}
+
+/// A specialized `Result` type for Shutdown operations.
+pub type ShutdownResult = Result<(), ShutdownError>;

opentelemetry-sdk/src/metrics/exporter.rs

@@ -1,6 +1,7 @@
 //! Interfaces for exporting metrics
 use async_trait::async_trait;
 
+use crate::error::ShutdownResult;
 use crate::metrics::MetricResult;
 
 use crate::metrics::data::ResourceMetrics;
@@ -27,7 +28,7 @@ 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) -> MetricResult<()>;
+    fn shutdown(&self) -> ShutdownResult;
 
     /// Access the [Temporality] of the MetricExporter.
     fn temporality(&self) -> Temporality;

opentelemetry-sdk/src/metrics/in_memory_exporter.rs

@@ -1,3 +1,4 @@
+use crate::error::ShutdownResult;
 use crate::metrics::data::{self, Gauge, Sum};
 use crate::metrics::data::{Histogram, Metric, ResourceMetrics, ScopeMetrics};
 use crate::metrics::exporter::PushMetricExporter;
@@ -277,7 +278,7 @@ impl PushMetricExporter for InMemoryMetricExporter {
         Ok(()) // In this implementation, flush does nothing
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         Ok(())
     }
 

opentelemetry-sdk/src/metrics/manual_reader.rs

@@ -5,7 +5,10 @@ use std::{
 
 use opentelemetry::otel_debug;
 
-use crate::metrics::{MetricError, MetricResult, Temporality};
+use crate::{
+    error::{ShutdownError, ShutdownResult},
+    metrics::{MetricError, MetricResult, Temporality},
+};
 
 use super::{
     data::ResourceMetrics,
@@ -107,8 +110,10 @@ impl MetricReader for ManualReader {
     }
 
     /// Closes any connections and frees any resources used by the reader.
-    fn shutdown(&self) -> MetricResult<()> {
-        let mut inner = self.inner.lock()?;
+    fn shutdown(&self) -> ShutdownResult {
+        let mut inner = self.inner.lock().map_err(|e| {
+            ShutdownError::InternalFailure(format!("Failed to acquire lock: {}", e))
+        })?;
 
         // Any future call to collect will now return an error.
         inner.sdk_producer = None;

opentelemetry-sdk/src/metrics/meter_provider.rs

@@ -12,8 +12,11 @@ use opentelemetry::{
     otel_debug, otel_error, otel_info, InstrumentationScope,
 };
 
-use crate::metrics::{MetricError, MetricResult};
 use crate::Resource;
+use crate::{
+    error::ShutdownResult,
+    metrics::{MetricError, MetricResult},
+};
 
 use super::{
     meter::SdkMeter, noop::NoopMeter, pipeline::Pipelines, reader::MetricReader, view::View,
@@ -108,7 +111,7 @@ impl SdkMeterProvider {
     ///
     /// There is no guaranteed that all telemetry be flushed or all resources have
     /// been released on error.
-    pub fn shutdown(&self) -> MetricResult<()> {
+    pub fn shutdown(&self) -> ShutdownResult {
         otel_info!(
             name: "MeterProvider.Shutdown",
             message = "User initiated shutdown of MeterProvider."
@@ -131,15 +134,13 @@ impl SdkMeterProviderInner {
         }
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         if self
             .shutdown_invoked
             .swap(true, std::sync::atomic::Ordering::SeqCst)
         {
             // If the previous value was true, shutdown was already invoked.
-            Err(MetricError::Other(
-                "MeterProvider shutdown already invoked.".into(),
-            ))
+            Err(crate::error::ShutdownError::AlreadyShutdown)
         } else {
             self.pipes.shutdown()
         }

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -11,6 +11,7 @@ use std::{
 use opentelemetry::{otel_debug, otel_error, otel_info, otel_warn};
 
 use crate::{
+    error::{ShutdownError, ShutdownResult},
     metrics::{exporter::PushMetricExporter, reader::SdkProducer, MetricError, MetricResult},
     Resource,
 };
@@ -402,27 +403,27 @@ impl PeriodicReaderInner {
         }
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         // TODO: See if this is better to be created upfront.
         let (response_tx, response_rx) = mpsc::channel();
         self.message_sender
             .send(Message::Shutdown(response_tx))
-            .map_err(|e| MetricError::Other(e.to_string()))?;
+            .map_err(|e| ShutdownError::InternalFailure(e.to_string()))?;
 
         // TODO: Make this timeout configurable.
         match response_rx.recv_timeout(Duration::from_secs(5)) {
             Ok(response) => {
                 if response {
                     Ok(())
                 } else {
-                    Err(MetricError::Other("Failed to shutdown".into()))
+                    Err(ShutdownError::InternalFailure("Failed to shutdown".into()))
                 }
             }
-            Err(mpsc::RecvTimeoutError::Timeout) => Err(MetricError::Other(
-                "Failed to shutdown due to Timeout".into(),
-            )),
+            Err(mpsc::RecvTimeoutError::Timeout) => {
+                Err(ShutdownError::Timeout(Duration::from_secs(5)))
+            }
             Err(mpsc::RecvTimeoutError::Disconnected) => {
-                Err(MetricError::Other("Failed to shutdown".into()))
+                Err(ShutdownError::InternalFailure("Failed to shutdown".into()))
             }
         }
     }
@@ -451,7 +452,7 @@ impl MetricReader for PeriodicReader {
     // 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) -> MetricResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         self.inner.shutdown()
     }
 
@@ -471,10 +472,10 @@ impl MetricReader for PeriodicReader {
 mod tests {
     use super::PeriodicReader;
     use crate::{
-        metrics::InMemoryMetricExporter,
+        error::ShutdownResult,
         metrics::{
-            data::ResourceMetrics, exporter::PushMetricExporter, reader::MetricReader, MetricError,
-            MetricResult, SdkMeterProvider, Temporality,
+            data::ResourceMetrics, exporter::PushMetricExporter, reader::MetricReader,
+            InMemoryMetricExporter, MetricError, MetricResult, SdkMeterProvider, Temporality,
         },
         Resource,
     };
@@ -524,7 +525,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> MetricResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             Ok(())
         }
 
@@ -548,7 +549,7 @@ mod tests {
             Ok(())
         }
 
-        fn shutdown(&self) -> MetricResult<()> {
+        fn shutdown(&self) -> ShutdownResult {
             self.is_shutdown.store(true, Ordering::Relaxed);
             Ok(())
         }