fix: Cleanup MetricError and use OTelSdkResult instead (#2905)

Author: cijothomas

opentelemetry-sdk/CHANGELOG.md

@@ -21,6 +21,10 @@ also modified to suppress telemetry before invoking exporters.
   - TODO/Placeholder: Add ability to configure cardinality limits via Instrument
     advisory.
 
+- *Breaking* change for custom `MetricReader` authors.
+  The `shutdown_with_timeout` method is added to `MetricReader` trait.
+  `collect` method on `MetricReader` modified to return `OTelSdkResult`.
+
 ## 0.29.0
 
 Released 2025-Mar-21
@@ -78,7 +82,6 @@ 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

@@ -7,8 +7,7 @@ use opentelemetry_sdk::{
     error::OTelSdkResult,
     metrics::{
         data::ResourceMetrics, new_view, reader::MetricReader, Aggregation, Instrument,
-        InstrumentKind, ManualReader, MetricResult, Pipeline, SdkMeterProvider, Stream,
-        Temporality, View,
+        InstrumentKind, ManualReader, Pipeline, SdkMeterProvider, Stream, Temporality, View,
     },
     Resource,
 };
@@ -24,7 +23,7 @@ impl MetricReader for SharedReader {
         self.0.register_pipeline(pipeline)
     }
 
-    fn collect(&self, rm: &mut ResourceMetrics) -> MetricResult<()> {
+    fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
         self.0.collect(rm)
     }
 

opentelemetry-sdk/src/metrics/manual_reader.rs

@@ -7,7 +7,7 @@ use std::{
 
 use crate::{
     error::{OTelSdkError, OTelSdkResult},
-    metrics::{MetricError, MetricResult, Temporality},
+    metrics::Temporality,
 };
 
 use super::{
@@ -90,12 +90,16 @@ impl MetricReader for ManualReader {
     /// callbacks necessary and returning the results.
     ///
     /// Returns an error if called after shutdown.
-    fn collect(&self, rm: &mut ResourceMetrics) -> MetricResult<()> {
-        let inner = self.inner.lock()?;
+    fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
+        let inner = self
+            .inner
+            .lock()
+            .map_err(|_| OTelSdkError::InternalFailure("Failed to lock pipeline".into()))?;
+
         match &inner.sdk_producer.as_ref().and_then(|w| w.upgrade()) {
             Some(producer) => producer.produce(rm)?,
             None => {
-                return Err(MetricError::Other(
+                return Err(OTelSdkError::InternalFailure(
                     "reader is shut down or not registered".into(),
                 ))
             }

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -12,7 +12,7 @@ use opentelemetry::{otel_debug, otel_error, otel_info, otel_warn, Context};
 
 use crate::{
     error::{OTelSdkError, OTelSdkResult},
-    metrics::{exporter::PushMetricExporter, reader::SdkProducer, MetricError, MetricResult},
+    metrics::{exporter::PushMetricExporter, reader::SdkProducer},
     Resource,
 };
 
@@ -357,11 +357,11 @@ impl<E: PushMetricExporter> PeriodicReaderInner<E> {
         self.exporter.temporality()
     }
 
-    fn collect(&self, rm: &mut ResourceMetrics) -> MetricResult<()> {
+    fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
         let producer = self.producer.lock().expect("lock poisoned");
         if let Some(p) = producer.as_ref() {
             p.upgrade()
-                .ok_or_else(|| MetricError::Other("pipeline is dropped".into()))?
+                .ok_or(OTelSdkError::AlreadyShutdown)?
                 .produce(rm)?;
             Ok(())
         } else {
@@ -371,7 +371,9 @@ impl<E: PushMetricExporter> PeriodicReaderInner<E> {
                    This occurs when a periodic reader is created but not associated with a MeterProvider \
                    by calling `.with_reader(reader)` on MeterProviderBuilder."
             );
-            Err(MetricError::Other("MeterProvider is not registered".into()))
+            Err(OTelSdkError::InternalFailure(
+                "MeterProvider is not registered".into(),
+            ))
         }
     }
 
@@ -479,7 +481,7 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
         self.inner.register_pipeline(pipeline);
     }
 
-    fn collect(&self, rm: &mut ResourceMetrics) -> MetricResult<()> {
+    fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
         self.inner.collect(rm)
     }
 

opentelemetry-sdk/src/metrics/periodic_reader_with_async_runtime.rs

@@ -16,7 +16,7 @@ use opentelemetry::{otel_debug, otel_error};
 use crate::runtime::{to_interval_stream, Runtime};
 use crate::{
     error::{OTelSdkError, OTelSdkResult},
-    metrics::{exporter::PushMetricExporter, reader::SdkProducer, MetricError, MetricResult},
+    metrics::{exporter::PushMetricExporter, reader::SdkProducer},
     Resource,
 };
 
@@ -351,10 +351,14 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
         worker(self);
     }
 
-    fn collect(&self, rm: &mut ResourceMetrics) -> MetricResult<()> {
-        let inner = self.inner.lock()?;
+    fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
+        let inner = self
+            .inner
+            .lock()
+            .map_err(|_| OTelSdkError::InternalFailure("Failed to lock pipeline".into()))?;
+
         if inner.is_shutdown {
-            return Err(MetricError::Other("reader is shut down".into()));
+            return Err(OTelSdkError::AlreadyShutdown);
         }
 
         if let Some(producer) = match &inner.sdk_producer_or_worker {
@@ -363,7 +367,9 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
         } {
             producer.produce(rm)?;
         } else {
-            return Err(MetricError::Other("reader is not registered".into()));
+            return Err(OTelSdkError::InternalFailure(
+                "reader is not registered".into(),
+            ));
         }
 
         Ok(())
@@ -434,8 +440,8 @@ impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
 #[cfg(all(test, feature = "testing"))]
 mod tests {
     use super::PeriodicReader;
+    use crate::error::OTelSdkError;
     use crate::metrics::reader::MetricReader;
-    use crate::metrics::MetricError;
     use crate::{
         metrics::data::ResourceMetrics, metrics::InMemoryMetricExporter, metrics::SdkMeterProvider,
         runtime, Resource,
@@ -496,7 +502,7 @@ mod tests {
 
         // Assert
         assert!(
-            matches!(result.unwrap_err(), MetricError::Other(err) if err == "reader is not registered")
+            matches!(result.unwrap_err(), OTelSdkError::InternalFailure(err) if err == "reader is not registered")
         );
     }
 

opentelemetry-sdk/src/metrics/pipeline.rs

@@ -100,8 +100,11 @@ impl Pipeline {
 
 impl SdkProducer for Pipeline {
     /// Returns aggregated metrics from a single collection.
-    fn produce(&self, rm: &mut ResourceMetrics) -> MetricResult<()> {
-        let inner = self.inner.lock()?;
+    fn produce(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
+        let inner = self
+            .inner
+            .lock()
+            .map_err(|_| OTelSdkError::InternalFailure("Failed to lock pipeline".into()))?;
         otel_debug!(
             name: "MeterProviderInvokingObservableCallbacks",
             count =  inner.callbacks.len(),

opentelemetry-sdk/src/metrics/reader.rs

@@ -1,5 +1,5 @@
 //! Interfaces for reading and producing metrics
-use crate::{error::OTelSdkResult, metrics::MetricResult};
+use crate::error::OTelSdkResult;
 use std::time::Duration;
 use std::{fmt, sync::Weak};
 
@@ -30,7 +30,7 @@ pub trait MetricReader: fmt::Debug + Send + Sync + 'static {
     /// SDK and stores it in the provided [ResourceMetrics] reference.
     ///
     /// An error is returned if this is called after shutdown.
-    fn collect(&self, rm: &mut ResourceMetrics) -> MetricResult<()>;
+    fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult;
 
     /// Flushes all metric measurements held in an export pipeline.
     ///
@@ -63,5 +63,5 @@ pub trait MetricReader: fmt::Debug + Send + Sync + 'static {
 /// Produces metrics for a [MetricReader].
 pub(crate) trait SdkProducer: fmt::Debug + Send + Sync {
     /// Returns aggregated metrics from a single collection.
-    fn produce(&self, rm: &mut ResourceMetrics) -> MetricResult<()>;
+    fn produce(&self, rm: &mut ResourceMetrics) -> OTelSdkResult;
 }

opentelemetry-sdk/src/metrics/view.rs

@@ -170,6 +170,7 @@ pub fn new_view(criteria: Instrument, mask: Stream) -> MetricResult<Box<dyn View
 }
 
 #[cfg(test)]
+#[cfg(feature = "spec_unstable_metrics_views")]
 mod tests {
     use super::*;
     #[test]

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

@@ -1,8 +1,8 @@
 use crate::error::{OTelSdkError, OTelSdkResult};
+use crate::metrics::Temporality;
 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;
 
@@ -34,7 +34,7 @@ impl Default for TestMetricReader {
 impl MetricReader for TestMetricReader {
     fn register_pipeline(&self, _pipeline: Weak<Pipeline>) {}
 
-    fn collect(&self, _rm: &mut ResourceMetrics) -> MetricResult<()> {
+    fn collect(&self, _rm: &mut ResourceMetrics) -> OTelSdkResult {
         Ok(())
     }