Passed down the ignored timeout from MeterProvider to reader in pipeline

Author: partg952

opentelemetry-sdk/src/metrics/meter_provider.rs

@@ -110,12 +110,12 @@ impl SdkMeterProvider {
     ///
     /// There is no guaranteed that all telemetry be flushed or all resources have
     /// been released on error.
-    pub fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+    pub fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
         otel_debug!(
             name: "MeterProvider.Shutdown",
             message = "User initiated shutdown of MeterProvider."
         );
-        self.inner.shutdown()
+        self.inner.shutdown_with_timeout(timeout)
     }
 
     /// shutdown with default timeout
@@ -136,15 +136,15 @@ impl SdkMeterProviderInner {
         }
     }
 
-    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
         if self
             .shutdown_invoked
             .swap(true, std::sync::atomic::Ordering::SeqCst)
         {
             // If the previous value was true, shutdown was already invoked.
             Err(crate::error::OTelSdkError::AlreadyShutdown)
         } else {
-            self.pipes.shutdown()
+            self.pipes.shutdown_with_timeout(timeout)
         }
     }
 

opentelemetry-sdk/src/metrics/pipeline.rs

@@ -3,6 +3,7 @@ use std::{
     borrow::Cow,
     collections::{HashMap, HashSet},
     sync::{Arc, Mutex},
+    time::Duration,
 };
 
 use opentelemetry::{otel_debug, otel_warn, InstrumentationScope, KeyValue};
@@ -96,6 +97,11 @@ impl Pipeline {
     fn shutdown(&self) -> OTelSdkResult {
         self.reader.shutdown()
     }
+
+    /// Shut down pipeline with timeout
+    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
+        self.reader.shutdown_with_timeout(timeout)
+    }
 }
 
 impl SdkProducer for Pipeline {
@@ -703,6 +709,24 @@ impl Pipelines {
         }
     }
 
+    /// Shut down all pipelines
+    pub(crate) fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
+        let mut errs = vec![];
+        for pipeline in &self.0 {
+            if let Err(err) = pipeline.shutdown_with_timeout(timeout) {
+                errs.push(err);
+            }
+        }
+
+        if errs.is_empty() {
+            Ok(())
+        } else {
+            Err(crate::error::OTelSdkError::InternalFailure(format!(
+                "{errs:?}"
+            )))
+        }
+    }
+
     /// Shut down all pipelines
     pub(crate) fn shutdown(&self) -> OTelSdkResult {
         let mut errs = vec![];