Use dedicated ShutdownResult for Metric SDK shutdown

Author: cijothomas

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,13 @@ 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::Failed(format!("Internal Error. 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,13 @@ 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::Failed(format!("Internal Error. 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 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/Cargo.toml

@@ -42,7 +42,7 @@ temp-env = { workspace = true }
 pprof = { version = "0.14", features = ["flamegraph", "criterion"] }
 
 [features]
-default = ["trace", "metrics", "logs", "internal-logs"]
+default = ["trace", "metrics", "logs", "internal-logs", "experimental_metrics_periodicreader_with_async_runtime"]
 trace = ["opentelemetry/trace", "rand", "percent-encoding"]
 jaeger_remote_sampler = ["trace", "opentelemetry-http", "http", "serde", "serde_json", "url"]
 logs = ["opentelemetry/logs", "serde_json"]

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,31 @@
 //! 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 already invoked.
+    #[error("Shutdown already invoked")]
+    AlreadyShutdown,
+
+    /// Shutdown timed out before completing.
+    #[error("Shutdown timed out after {0:?}")]
+    Timeout(Duration),
+
+    /// Shutdown failed with an error.
+    /// This error is returned when the shutdown process failed with an error.
+    #[error("Shutdown failed: {0}")]
+    Failed(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::Failed(format!("Internal Error. 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()
         }