Collect and export with zero allocations and clones

fraillt · fraillt · commit 246dff23e766 · 2025-04-12T16:03:00.000+03:00
diff --git a/opentelemetry-sdk/src/metrics/exporter.rs b/opentelemetry-sdk/src/metrics/exporter.rs
@@ -3,11 +3,15 @@
 use opentelemetry::InstrumentationScope;
 
 use crate::{error::OTelSdkResult, Resource};
-use std::{fmt::Debug, slice::Iter, time::Duration};
+use std::{
+    fmt::Debug,
+    slice::Iter,
+    time::{Duration, SystemTime},
+};
 
 use super::{
-    data::AggregatedMetrics,
-    reader::{MetricsData, ResourceMetricsData, ScopeMetricsData},
+    data::{AggregatedMetrics, Sum},
+    pipeline::InstrumentSync,
     InstrumentInfo, Temporality,
 };
 
@@ -23,7 +27,7 @@ pub struct ResourceMetrics<'a> {
 /// Iterator over libraries instrumentation scopes ([`InstrumentationScope`]) together with metrics.
 /// Doesn't implement standard [`Iterator`], because it returns borrowed items. AKA "LendingIterator".
 pub struct ScopeMetricsLendingIter<'a> {
-    iter: Iter<'a, ScopeMetricsData>,
+    iter: std::collections::hash_map::Iter<'a, InstrumentationScope, Vec<InstrumentSync>>,
 }
 
 /// A collection of metrics produced by a [`InstrumentationScope`] meter.
@@ -38,7 +42,9 @@ pub struct ScopeMetrics<'a> {
 /// Iterator over aggregations created by the meter.
 /// Doesn't implement standard [`Iterator`], because it returns borrowed items. AKA "LendingIterator".
 pub struct MetricsLendingIter<'a> {
-    iter: Iter<'a, MetricsData>,
+    // for  optimization purposes
+    aggr: AggregatedMetrics,
+    iter: Iter<'a, InstrumentSync>,
 }
 
 /// A collection of one or more aggregated time series from an [Instrument].
@@ -53,23 +59,13 @@ pub struct Metric<'a> {
 }
 
 impl<'a> ResourceMetrics<'a> {
-    pub(crate) fn new(rm: &'a ResourceMetricsData) -> Self {
-        Self {
-            resource: &rm.resource,
-            scope_metrics: ScopeMetricsLendingIter {
-                iter: rm.scope_metrics.iter(),
-            },
-        }
-    }
-}
-
-impl<'a> ScopeMetrics<'a> {
-    fn new(sm: &'a ScopeMetricsData) -> Self {
+    pub(crate) fn new(
+        resource: &'a Resource,
+        iter: std::collections::hash_map::Iter<'a, InstrumentationScope, Vec<InstrumentSync>>,
+    ) -> Self {
         Self {
-            scope: &sm.scope,
-            metrics: MetricsLendingIter {
-                iter: sm.metrics.iter(),
-            },
+            resource,
+            scope_metrics: ScopeMetricsLendingIter { iter },
         }
     }
 }
@@ -83,17 +79,40 @@ impl Debug for ScopeMetricsLendingIter<'_> {
 impl ScopeMetricsLendingIter<'_> {
     /// Advances the iterator and returns the next value.
     pub fn next(&mut self) -> Option<ScopeMetrics<'_>> {
-        self.iter.next().map(ScopeMetrics::new)
+        self.iter.next().map(|(scope, instruments)| ScopeMetrics {
+            scope,
+            metrics: MetricsLendingIter {
+                // doesn't matter what we initialize this with,
+                // it's purpose is to be reused between collections
+                aggr: AggregatedMetrics::F64(super::data::MetricData::Sum(Sum {
+                    is_monotonic: true,
+                    data_points: Vec::new(),
+                    start_time: SystemTime::now(),
+                    time: SystemTime::now(),
+                    temporality: Temporality::Cumulative,
+                })),
+                iter: instruments.iter(),
+            },
+        })
     }
 }
 
 impl MetricsLendingIter<'_> {
     /// Advances the iterator and returns the next value.
     pub fn next(&mut self) -> Option<Metric<'_>> {
-        self.iter.next().map(|metric| Metric {
-            instrument: &metric.instrument,
-            data: &metric.data,
-        })
+        loop {
+            let inst = self.iter.next()?;
+            let (len, data) = inst.comp_agg.call(Some(&mut self.aggr));
+            if len > 0 {
+                if let Some(new_aggr) = data {
+                    self.aggr = new_aggr;
+                }
+                return Some(Metric {
+                    instrument: &inst.info,
+                    data: &self.aggr,
+                });
+            }
+        }
     }
 }
 
diff --git a/opentelemetry-sdk/src/metrics/in_memory_exporter.rs b/opentelemetry-sdk/src/metrics/in_memory_exporter.rs
@@ -194,7 +194,7 @@ impl InMemoryMetricExporter {
             .map(|mut metrics_guard| metrics_guard.clear());
     }
 
-    fn clone_metrics(mut metric: ResourceMetrics<'_>) -> ResourceMetricsData {
+    fn clone_metrics(mut metric: ResourceMetrics<'_>) -> Option<ResourceMetricsData> {
         let mut scope_metrics = Vec::new();
         while let Some(mut scope_metric) = metric.scope_metrics.next() {
             let mut metrics = Vec::new();
@@ -204,14 +204,20 @@ impl InMemoryMetricExporter {
                     data: Self::clone_data(&metric.data),
                 });
             }
-            scope_metrics.push(ScopeMetricsData {
-                scope: scope_metric.scope.clone(),
-                metrics,
-            });
+            if !metrics.is_empty() {
+                scope_metrics.push(ScopeMetricsData {
+                    scope: scope_metric.scope.clone(),
+                    metrics,
+                });
+            }
         }
-        ResourceMetricsData {
-            resource: metric.resource.clone(),
-            scope_metrics,
+        if !scope_metrics.is_empty() {
+            Some(ResourceMetricsData {
+                resource: metric.resource.clone(),
+                scope_metrics,
+            })
+        } else {
+            None
         }
     }
 
@@ -261,7 +267,7 @@ impl PushMetricExporter for InMemoryMetricExporter {
         self.metrics
             .lock()
             .map(|mut metrics_guard| {
-                metrics_guard.push_back(InMemoryMetricExporter::clone_metrics(metrics))
+                metrics_guard.extend(InMemoryMetricExporter::clone_metrics(metrics))
             })
             .map_err(|_| OTelSdkError::InternalFailure("Failed to lock metrics".to_string()))
     }
diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
@@ -82,8 +82,7 @@ pub use periodic_reader::*;
 #[cfg(feature = "experimental_metrics_custom_reader")]
 pub use pipeline::Pipeline;
 
-#[cfg(feature = "experimental_metrics_custom_reader")]
-pub use instrument::InstrumentKind;
+pub use instrument::{InstrumentInfo, InstrumentKind};
 
 #[cfg(feature = "spec_unstable_metrics_views")]
 pub use instrument::*;
diff --git a/opentelemetry-sdk/src/metrics/periodic_reader.rs b/opentelemetry-sdk/src/metrics/periodic_reader.rs
@@ -16,7 +16,6 @@ use crate::{
         exporter::{PushMetricExporter, ResourceMetrics},
         reader::SdkProducer,
     },
-    Resource,
 };
 
 use super::{
@@ -350,11 +349,11 @@ impl<E: PushMetricExporter> fmt::Debug for PeriodicReader<E> {
 struct PeriodicReaderInner<E: PushMetricExporter> {
     exporter: Arc<E>,
     message_sender: mpsc::Sender<Message>,
-    producer: Mutex<Option<Weak<dyn SdkProducer>>>,
+    producer: Mutex<Option<Weak<Pipeline>>>,
 }
 
 impl<E: PushMetricExporter> PeriodicReaderInner<E> {
-    fn register_pipeline(&self, producer: Weak<dyn SdkProducer>) {
+    fn register_pipeline(&self, producer: Weak<Pipeline>) {
         let mut inner = self.producer.lock().expect("lock poisoned");
         *inner = Some(producer);
     }
@@ -384,39 +383,47 @@ impl<E: PushMetricExporter> PeriodicReaderInner<E> {
     }
 
     fn collect_and_export(&self) -> OTelSdkResult {
-        // TODO: Reuse the internal vectors. Or refactor to avoid needing any
-        // owned data structures to be passed to exporters.
-        let mut rm = ResourceMetricsData {
-            resource: Resource::empty(),
-            scope_metrics: Vec::new(),
+        let producer = self.producer.lock().expect("lock poisoned");
+        let pipeline = if let Some(p) = producer.as_ref() {
+            p.upgrade().ok_or(OTelSdkError::AlreadyShutdown)?
+        } else {
+            otel_warn!(
+            name: "PeriodReader.MeterProviderNotRegistered",
+            message = "PeriodicReader is not registered with MeterProvider. Metrics will not be collected. \
+                   This occurs when a periodic reader is created but not associated with a MeterProvider \
+                   by calling `.with_reader(reader)` on MeterProviderBuilder."
+            );
+            return Err(OTelSdkError::InternalFailure(
+                "MeterProvider is not registered".into(),
+            ));
         };
-
+        drop(producer);
+        let Ok(inner) = pipeline.inner.lock() else {
+            otel_warn!(
+                name: "PeriodReader.PipelineLockPoisoned",
+                message = "Failed to acquire lock for collect and export"
+            );
+            return Err(OTelSdkError::InternalFailure(
+                "Paniced while holding a pipeline lock".into(),
+            ));
+        };
+        for cb in &inner.callbacks {
+            cb();
+        }
         let current_time = Instant::now();
-        let collect_result = self.collect(&mut rm);
-        let time_taken_for_collect = current_time.elapsed();
+        // Relying on futures executor to execute async call.
+        let res = futures_executor::block_on(self.exporter.export(ResourceMetrics::new(
+            &pipeline.resource,
+            inner.aggregations.iter(),
+        )));
+        otel_debug!(name: "PeriodicReaderMetricsCollected", time_taken_in_millis = current_time.elapsed().as_millis());
 
-        #[allow(clippy::question_mark)]
-        if let Err(e) = collect_result {
+        res.inspect_err(|err| {
             otel_warn!(
                 name: "PeriodReaderCollectError",
-                error = format!("{:?}", e)
+                error = format!("{:?}", err)
             );
-            return Err(OTelSdkError::InternalFailure(e.to_string()));
-        }
-
-        if rm.scope_metrics.is_empty() {
-            otel_debug!(name: "NoMetricsCollected");
-            return Ok(());
-        }
-
-        let metrics_count = rm.scope_metrics.iter().fold(0, |count, scope_metrics| {
-            count + scope_metrics.metrics.len()
-        });
-        otel_debug!(name: "PeriodicReaderMetricsCollected", count = metrics_count, time_taken_in_millis = time_taken_for_collect.as_millis());
-
-        // Relying on futures executor to execute async call.
-        // TODO: Pass timeout to exporter
-        futures_executor::block_on(self.exporter.export(ResourceMetrics::new(&rm)))
+        })
     }
 
     fn force_flush(&self) -> OTelSdkResult {
diff --git a/opentelemetry-sdk/src/metrics/periodic_reader_with_async_runtime.rs b/opentelemetry-sdk/src/metrics/periodic_reader_with_async_runtime.rs
@@ -6,8 +6,6 @@ use std::{
 
 use futures_channel::{mpsc, oneshot};
 use futures_util::{
-    future::{self, Either},
-    pin_mut,
     stream::{self, FusedStream},
     StreamExt,
 };
@@ -19,7 +17,6 @@ use crate::runtime::{to_interval_stream, Runtime};
 use crate::{
     error::{OTelSdkError, OTelSdkResult},
     metrics::{exporter::PushMetricExporter, reader::SdkProducer},
-    Resource,
 };
 
 use super::{instrument::InstrumentKind, pipeline::Pipeline, reader::MetricReader};
@@ -116,17 +113,7 @@ where
                     .skip(1) // The ticker is fired immediately, so we should skip the first one to align with the interval.
                     .map(|_| Message::Export);
                 let messages = Box::pin(stream::select(message_receiver, ticker));
-                PeriodicReaderWorker {
-                    reader,
-                    timeout: self.timeout,
-                    runtime,
-                    rm: ResourceMetricsData {
-                        resource: Resource::empty(),
-                        scope_metrics: Vec::new(),
-                    },
-                }
-                .run(messages)
-                .await
+                PeriodicReaderWorker { reader }.run(messages).await
             });
         };
 
@@ -229,47 +216,50 @@ enum Message {
 }
 
 enum ProducerOrWorker<E: PushMetricExporter> {
-    Producer(Weak<dyn SdkProducer>),
+    Producer(Weak<Pipeline>),
     #[allow(clippy::type_complexity)]
     Worker(Box<dyn FnOnce(&PeriodicReader<E>) + Send + Sync>),
 }
 
-struct PeriodicReaderWorker<E: PushMetricExporter, RT: Runtime> {
+struct PeriodicReaderWorker<E: PushMetricExporter> {
     reader: PeriodicReader<E>,
-    timeout: Duration,
-    runtime: RT,
-    rm: ResourceMetricsData,
 }
 
-impl<E: PushMetricExporter, RT: Runtime> PeriodicReaderWorker<E, RT> {
+impl<E: PushMetricExporter> PeriodicReaderWorker<E> {
     async fn collect_and_export(&mut self) -> OTelSdkResult {
-        self.reader
-            .collect(&mut self.rm)
-            .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
-        if self.rm.scope_metrics.is_empty() {
-            otel_debug!(
-                name: "PeriodicReaderWorker.NoMetricsToExport",
-            );
-            // No metrics to export.
-            return Ok(());
+        let inner = self
+            .reader
+            .inner
+            .lock()
+            .map_err(|_| OTelSdkError::InternalFailure("Failed to lock pipeline".into()))?;
+
+        if inner.is_shutdown {
+            return Err(OTelSdkError::AlreadyShutdown);
         }
 
-        otel_debug!(
-            name: "PeriodicReaderWorker.InvokeExporter",
-            message = "Calling exporter's export method with collected metrics.",
-            count = self.rm.scope_metrics.len(),
-        );
-        let export = self.reader.exporter.export(ResourceMetrics::new(&self.rm));
-        let timeout = self.runtime.delay(self.timeout);
-        pin_mut!(export);
-        pin_mut!(timeout);
-
-        match future::select(export, timeout).await {
-            Either::Left((res, _)) => {
-                res // return the status of export.
-            }
-            Either::Right(_) => Err(OTelSdkError::Timeout(self.timeout)),
+        let producer = match &inner.sdk_producer_or_worker {
+            ProducerOrWorker::Producer(sdk_producer) => sdk_producer.upgrade(),
+            ProducerOrWorker::Worker(_) => None,
+        }
+        .ok_or(OTelSdkError::InternalFailure(
+            "reader is not registered".into(),
+        ))?;
+        drop(inner);
+
+        let Ok(producer_inner) = producer.inner.lock() else {
+            return Err(OTelSdkError::InternalFailure(
+                "Paniced while holding a pipeline lock".into(),
+            ));
+        };
+        for cb in &producer_inner.callbacks {
+            cb();
         }
+        // unfortunatelly we need to block here, because runtime require future to be "Send",
+        // but we hold a lock for PipelineInner.
+        futures_executor::block_on(self.reader.exporter.export(ResourceMetrics::new(
+            &producer.resource,
+            producer_inner.aggregations.iter(),
+        )))
     }
 
     async fn process_message(&mut self, message: Message) -> bool {
diff --git a/opentelemetry-sdk/src/metrics/pipeline.rs b/opentelemetry-sdk/src/metrics/pipeline.rs
