Collect and export with zero allocations and clones

fraillt · fraillt · commit fc6205d3b83b · 2025-04-13T20:12:14.000+03:00
diff --git a/opentelemetry-proto/src/transform/metrics.rs b/opentelemetry-proto/src/transform/metrics.rs
@@ -113,7 +113,7 @@ pub mod tonic {
     impl From<ResourceMetrics<'_>> for ExportMetricsServiceRequest {
         fn from(mut rm: ResourceMetrics<'_>) -> Self {
             let mut scope_metrics = Vec::new();
-            while let Some(scope_metric) = rm.scope_metrics.next() {
+            while let Some(scope_metric) = rm.scope_metrics.next_scope_metric() {
                 scope_metrics.push(scope_metric.into());
             }
             ExportMetricsServiceRequest {
@@ -138,7 +138,7 @@ pub mod tonic {
     impl From<ScopeMetrics<'_>> for TonicScopeMetrics {
         fn from(mut sm: ScopeMetrics<'_>) -> Self {
             let mut metrics = Vec::new();
-            while let Some(metric) = sm.metrics.next() {
+            while let Some(metric) = sm.metrics.next_metric() {
                 metrics.push(metric.into());
             }
             TonicScopeMetrics {
diff --git a/opentelemetry-sdk/benches/metric.rs b/opentelemetry-sdk/benches/metric.rs
@@ -6,8 +6,10 @@ use opentelemetry::{
 use opentelemetry_sdk::{
     error::OTelSdkResult,
     metrics::{
-        data::ResourceMetricsData, new_view, reader::MetricReader, Aggregation, Instrument,
-        InstrumentKind, ManualReader, Pipeline, SdkMeterProvider, Stream, Temporality, View,
+        new_view,
+        reader::{MetricReader, ResourceMetricsData},
+        Aggregation, Instrument, InstrumentKind, ManualReader, Pipeline, SdkMeterProvider, Stream,
+        Temporality, View,
     },
     Resource,
 };
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 },
         }
     }
 }
@@ -81,19 +77,42 @@ 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)
+    /// Advances the iterator and returns the next value.    
+    pub fn next_scope_metric(&mut self) -> Option<ScopeMetrics<'_>> {
+        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,
-        })
+    /// Advances the iterator and returns the next value.    
+    pub fn next_metric(&mut self) -> Option<Metric<'_>> {
+        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,24 +194,30 @@ 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() {
+        while let Some(mut scope_metric) = metric.scope_metrics.next_scope_metric() {
             let mut metrics = Vec::new();
-            while let Some(metric) = scope_metric.metrics.next() {
+            while let Some(metric) = scope_metric.metrics.next_metric() {
                 metrics.push(MetricsData {
                     instrument: metric.instrument.clone(),
-                    data: Self::clone_data(&metric.data),
+                    data: Self::clone_data(metric.data),
+                });
+            }
+            if !metrics.is_empty() {
+                scope_metrics.push(ScopeMetricsData {
+                    scope: scope_metric.scope.clone(),
+                    metrics,
                 });
             }
-            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,48 @@ 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 {
+        if let Err(err) = &res {
             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)))
+        res
     }
 
     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
diff --git a/opentelemetry-sdk/src/metrics/pipeline.rs b/opentelemetry-sdk/src/metrics/pipeline.rs
diff --git a/opentelemetry-stdout/src/metrics/exporter.rs b/opentelemetry-stdout/src/metrics/exporter.rs
