Merge branch 'main' into re-export-tonic

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

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 _},
@@ -10,11 +7,13 @@ 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,
 };
+use rand::Rng;
+use std::sync::{Arc, Weak};
+use std::time::Duration;
 
 #[derive(Clone, Debug)]
 struct SharedReader(Arc<dyn MetricReader>);
@@ -24,15 +23,15 @@ 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)
     }
 
     fn force_flush(&self) -> OTelSdkResult {
         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,13 +1,13 @@
+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},
+    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(),
                 ))
             }
@@ -110,7 +114,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

@@ -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)
     }
 
@@ -491,7 +493,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

@@ -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(())
@@ -390,7 +396,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()
@@ -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,8 +1,8 @@
 //! Interfaces for reading and producing metrics
+use crate::error::OTelSdkResult;
+use std::time::Duration;
 use std::{fmt, sync::Weak};
 
-use crate::{error::OTelSdkResult, metrics::MetricResult};
-
 use super::{data::ResourceMetrics, pipeline::Pipeline, InstrumentKind, Temporality};
 
 /// The interface used between the SDK and an exporter.
@@ -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.
     ///
@@ -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.
@@ -58,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,10 +1,10 @@
-use std::sync::{Arc, Mutex, Weak};
-
 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;
 
 #[derive(Debug, Clone)]
 pub struct TestMetricReader {
@@ -34,15 +34,15 @@ 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(())
     }
 
     fn force_flush(&self) -> OTelSdkResult {
         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();