Merge branch 'main' into cijothomas/metric-shutdown

Author: cijothomas

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -149,7 +149,7 @@ impl PeriodicReader {
         let exporter_arc = Arc::new(exporter);
         let reader = PeriodicReader {
             inner: Arc::new(PeriodicReaderInner {
-                message_sender: Arc::new(message_sender),
+                message_sender,
                 producer: Mutex::new(None),
                 exporter: exporter_arc.clone(),
             }),
@@ -295,7 +295,7 @@ impl fmt::Debug for PeriodicReader {
 
 struct PeriodicReaderInner {
     exporter: Arc<dyn PushMetricExporter>,
-    message_sender: Arc<mpsc::Sender<Message>>,
+    message_sender: mpsc::Sender<Message>,
     producer: Mutex<Option<Weak<dyn SdkProducer>>>,
 }
 
@@ -321,15 +321,23 @@ impl PeriodicReaderInner {
         }
     }
 
-    fn collect_and_export(&self, _timeout: Duration) -> MetricResult<()> {
+    fn collect_and_export(&self, timeout: Duration) -> MetricResult<()> {
         // TODO: Reuse the internal vectors. Or refactor to avoid needing any
         // owned data structures to be passed to exporters.
         let mut rm = ResourceMetrics {
             resource: Resource::empty(),
             scope_metrics: Vec::new(),
         };
 
+        // Measure time taken for collect, and subtract it from the timeout.
+        let current_time = Instant::now();
         let collect_result = self.collect(&mut rm);
+        let time_taken_for_collect = current_time.elapsed();
+        let _timeout = if time_taken_for_collect > timeout {
+            Duration::from_secs(0)
+        } else {
+            timeout - time_taken_for_collect
+        };
         #[allow(clippy::question_mark)]
         if let Err(e) = collect_result {
             otel_warn!(
@@ -347,15 +355,10 @@ impl PeriodicReaderInner {
         let metrics_count = rm.scope_metrics.iter().fold(0, |count, scope_metrics| {
             count + scope_metrics.metrics.len()
         });
-        otel_debug!(name: "PeriodicReaderMetricsCollected", count = metrics_count);
+        otel_debug!(name: "PeriodicReaderMetricsCollected", count = metrics_count, time_taken_in_millis = time_taken_for_collect.as_millis());
 
-        // TODO: subtract the time taken for collect from the timeout. collect
-        // involves observable callbacks too, which are user defined and can
-        // take arbitrary time.
-        //
         // Relying on futures executor to execute async call.
-        // TODO: Add timeout and pass it to exporter or consider alternative
-        // design to enforce timeout here.
+        // TODO: Pass timeout to exporter
         let exporter_result = futures_executor::block_on(self.exporter.export(&mut rm));
         #[allow(clippy::question_mark)]
         if let Err(e) = exporter_result {