Merge branch 'main' into cijothomas/metric-doc1

Author: cijothomas

.github/workflows/ci.yml

@@ -98,6 +98,8 @@ jobs:
         with:
           toolchain: nightly-2024-06-30
           components: rustfmt
+      - name: Patch dependencies versions
+        run: bash ./scripts/patch_dependencies.sh
       - name: external-type-check
         run: |
           cargo install [email protected]

deny.toml

@@ -15,19 +15,22 @@ exceptions = [
     { allow = ["Unicode-3.0"], crate = "icu_locid" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "icu_locid_transform" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "icu_locid_transform_data" }, # This crate gets used transitively by `reqwest`.
+    { allow = ["Unicode-3.0"], crate = "icu_locale_core" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "icu_normalizer" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "icu_normalizer_data" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "icu_properties" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "icu_properties_data" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "icu_provider" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "icu_provider_macros" }, # This crate gets used transitively by `reqwest`.
+    { allow = ["Unicode-3.0"], crate = "potential_utf" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "litemap" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "tinystr" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "writeable" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "unicode-ident" }, # This crate gets used transitively by `reqwest` and other crates.
     { allow = ["Unicode-3.0"], crate = "yoke" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "yoke-derive" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "zerovec" }, # This crate gets used transitively by `reqwest`.
+    { allow = ["Unicode-3.0"], crate = "zerotrie" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "zerovec-derive" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "zerofrom" }, # This crate gets used transitively by `reqwest`.
     { allow = ["Unicode-3.0"], crate = "zerofrom-derive" }, # This crate gets used transitively by `reqwest`.

opentelemetry-proto/src/transform/metrics.rs

@@ -115,7 +115,7 @@ pub mod tonic {
             ExportMetricsServiceRequest {
                 resource_metrics: vec![TonicResourceMetrics {
                     resource: Some((&rm.resource).into()),
-                    scope_metrics: rm.scope_metrics.iter().map(Into::into).collect(),
+                    scope_metrics: rm.scope_metrics().map(Into::into).collect(),
                     schema_url: rm.resource.schema_url().map(Into::into).unwrap_or_default(),
                 }],
             }
@@ -136,7 +136,7 @@ pub mod tonic {
         fn from(sm: &SdkScopeMetrics) -> Self {
             TonicScopeMetrics {
                 scope: Some((&sm.scope, None).into()),
-                metrics: sm.metrics.iter().map(Into::into).collect(),
+                metrics: sm.metrics().map(Into::into).collect(),
                 schema_url: sm
                     .scope
                     .schema_url()
@@ -273,8 +273,7 @@ pub mod tonic {
         fn from(sum: &SdkSum<T>) -> Self {
             TonicSum {
                 data_points: sum
-                    .data_points
-                    .iter()
+                    .data_points()
                     .map(|dp| TonicNumberDataPoint {
                         attributes: dp.attributes.iter().map(Into::into).collect(),
                         start_time_unix_nano: to_nanos(sum.start_time),

opentelemetry-sdk/benches/metric.rs

@@ -9,7 +9,6 @@ use opentelemetry_sdk::{
         data::ResourceMetrics, new_view, reader::MetricReader, Aggregation, Instrument,
         InstrumentKind, ManualReader, Pipeline, SdkMeterProvider, Stream, Temporality, View,
     },
-    Resource,
 };
 use rand::Rng;
 use std::sync::{Arc, Weak};
@@ -240,10 +239,7 @@ fn counters(c: &mut Criterion) {
     });
 
     let (rdr, cntr) = bench_counter(None, "cumulative");
-    let mut rm = ResourceMetrics {
-        resource: Resource::builder_empty().build(),
-        scope_metrics: Vec::new(),
-    };
+    let mut rm = ResourceMetrics::default();
 
     group.bench_function("CollectOneAttr", |b| {
         let mut v = 0;
@@ -337,10 +333,7 @@ fn benchmark_collect_histogram(b: &mut Bencher, n: usize) {
         h.record(1, &[]);
     }
 
-    let mut rm = ResourceMetrics {
-        resource: Resource::builder_empty().build(),
-        scope_metrics: Vec::new(),
-    };
+    let mut rm = ResourceMetrics::default();
 
     b.iter(|| {
         let _ = r.collect(&mut rm);

opentelemetry-sdk/src/metrics/aggregation.rs

@@ -151,10 +151,8 @@ impl Aggregation {
 #[cfg(test)]
 mod tests {
     use crate::metrics::error::{MetricError, MetricResult};
-    use crate::metrics::{
-        internal::{EXPO_MAX_SCALE, EXPO_MIN_SCALE},
-        Aggregation,
-    };
+    use crate::metrics::internal::{EXPO_MAX_SCALE, EXPO_MIN_SCALE};
+    use crate::metrics::Aggregation;
 
     #[test]
     fn validate_aggregation() {

opentelemetry-sdk/src/metrics/data/mod.rs

@@ -14,7 +14,23 @@ pub struct ResourceMetrics {
     /// The entity that collected the metrics.
     pub resource: Resource,
     /// The collection of metrics with unique [InstrumentationScope]s.
-    pub scope_metrics: Vec<ScopeMetrics>,
+    pub(crate) scope_metrics: Vec<ScopeMetrics>,
+}
+
+impl Default for ResourceMetrics {
+    fn default() -> Self {
+        Self {
+            resource: Resource::empty(),
+            scope_metrics: Vec::new(),
+        }
+    }
+}
+
+impl ResourceMetrics {
+    /// Returns an iterator over the [ScopeMetrics] in [ResourceMetrics].
+    pub fn scope_metrics(&self) -> impl Iterator<Item = &ScopeMetrics> {
+        self.scope_metrics.iter()
+    }
 }
 
 /// A collection of metrics produced by a meter.
@@ -23,7 +39,14 @@ pub struct ScopeMetrics {
     /// The [InstrumentationScope] that the meter was created with.
     pub scope: InstrumentationScope,
     /// The list of aggregations created by the meter.
-    pub metrics: Vec<Metric>,
+    pub(crate) metrics: Vec<Metric>,
+}
+
+impl ScopeMetrics {
+    /// Returns an iterator over the [Metric]s in [ScopeMetrics].
+    pub fn metrics(&self) -> impl Iterator<Item = &Metric> {
+        self.metrics.iter()
+    }
 }
 
 /// A collection of one or more aggregated time series from an [Instrument].
@@ -146,7 +169,7 @@ pub struct SumDataPoint<T> {
 #[derive(Debug, Clone)]
 pub struct Sum<T> {
     /// Represents individual aggregated measurements with unique attributes.
-    pub data_points: Vec<SumDataPoint<T>>,
+    pub(crate) data_points: Vec<SumDataPoint<T>>,
     /// The time when the time series was started.
     pub start_time: SystemTime,
     /// The time when the time series was recorded.
@@ -158,6 +181,13 @@ pub struct Sum<T> {
     pub is_monotonic: bool,
 }
 
+impl<T> Sum<T> {
+    /// Returns an iterator over the [SumDataPoint]s in [Sum].
+    pub fn data_points(&self) -> impl Iterator<Item = &SumDataPoint<T>> {
+        self.data_points.iter()
+    }
+}
+
 /// Represents the histogram of all measurements of values from an instrument.
 #[derive(Debug, Clone)]
 pub struct Histogram<T> {

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -156,6 +156,11 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
         };
         let cloned_reader = reader.clone();
 
+        let mut rm = ResourceMetrics {
+            resource: Resource::empty(),
+            scope_metrics: Vec::new(),
+        };
+
         let result_thread_creation = thread::Builder::new()
             .name("OpenTelemetry.Metrics.PeriodicReader".to_string())
             .spawn(move || {
@@ -175,7 +180,7 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
                             otel_debug!(
                                 name: "PeriodReaderThreadExportingDueToFlush"
                             );
-                            let export_result = cloned_reader.collect_and_export();
+                            let export_result = cloned_reader.collect_and_export(&mut rm);
                             otel_debug!(
                                 name: "PeriodReaderInvokedExport",
                                 export_result = format!("{:?}", export_result)
@@ -231,7 +236,7 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
                         Ok(Message::Shutdown(response_sender)) => {
                             // Perform final export and break out of loop and exit the thread
                             otel_debug!(name: "PeriodReaderThreadExportingDueToShutdown");
-                            let export_result = cloned_reader.collect_and_export();
+                            let export_result = cloned_reader.collect_and_export(&mut rm);
                             otel_debug!(
                                 name: "PeriodReaderInvokedExport",
                                 export_result = format!("{:?}", export_result)
@@ -279,7 +284,7 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
                                 name: "PeriodReaderThreadExportingDueToTimer"
                             );
 
-                            let export_result = cloned_reader.collect_and_export();
+                            let export_result = cloned_reader.collect_and_export(&mut rm);
                             otel_debug!(
                                 name: "PeriodReaderInvokedExport",
                                 export_result = format!("{:?}", export_result)
@@ -331,8 +336,8 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
         reader
     }
 
-    fn collect_and_export(&self) -> OTelSdkResult {
-        self.inner.collect_and_export()
+    fn collect_and_export(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
+        self.inner.collect_and_export(rm)
     }
 }
 
@@ -378,16 +383,9 @@ 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 = ResourceMetrics {
-            resource: Resource::empty(),
-            scope_metrics: Vec::new(),
-        };
-
+    fn collect_and_export(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
         let current_time = Instant::now();
-        let collect_result = self.collect(&mut rm);
+        let collect_result = self.collect(rm);
         let time_taken_for_collect = current_time.elapsed();
 
         #[allow(clippy::question_mark)]
@@ -411,7 +409,7 @@ impl<E: PushMetricExporter> PeriodicReaderInner<E> {
 
         // Relying on futures executor to execute async call.
         // TODO: Pass timeout to exporter
-        futures_executor::block_on(self.exporter.export(&rm))
+        futures_executor::block_on(self.exporter.export(rm))
     }
 
     fn force_flush(&self) -> OTelSdkResult {

opentelemetry-stdout/src/metrics/exporter.rs

@@ -55,7 +55,7 @@ impl PushMetricExporter for MetricExporter {
             metrics.resource.iter().for_each(|(k, v)| {
                 println!("\t ->  {}={:?}", k, v);
             });
-            print_metrics(&metrics.scope_metrics);
+            print_metrics(metrics.scope_metrics());
             Ok(())
         }
     }
@@ -79,8 +79,8 @@ impl PushMetricExporter for MetricExporter {
     }
 }
 
-fn print_metrics(metrics: &[ScopeMetrics]) {
-    for (i, metric) in metrics.iter().enumerate() {
+fn print_metrics<'a>(metrics: impl Iterator<Item = &'a ScopeMetrics>) {
+    for (i, metric) in metrics.enumerate() {
         println!("\tInstrumentation Scope #{}", i);
         println!("\t\tName         : {}", &metric.scope.name());
         if let Some(version) = &metric.scope.version() {
@@ -100,7 +100,7 @@ fn print_metrics(metrics: &[ScopeMetrics]) {
                 println!("\t\t\t ->  {}: {}", kv.key, kv.value);
             });
 
-        metric.metrics.iter().enumerate().for_each(|(i, metric)| {
+        metric.metrics().enumerate().for_each(|(i, metric)| {
             println!("Metric #{}", i);
             println!("\t\tName         : {}", &metric.name);
             println!("\t\tDescription  : {}", &metric.description);
@@ -155,7 +155,7 @@ fn print_sum<T: Debug>(sum: &Sum<T>) {
         "\t\tEndTime      : {}",
         datetime.format("%Y-%m-%d %H:%M:%S%.6f")
     );
-    print_sum_data_points(&sum.data_points);
+    print_sum_data_points(sum.data_points());
 }
 
 fn print_gauge<T: Debug>(gauge: &Gauge<T>) {
@@ -195,8 +195,10 @@ fn print_histogram<T: Debug>(histogram: &Histogram<T>) {
     print_hist_data_points(&histogram.data_points);
 }
 
-fn print_sum_data_points<T: Debug>(data_points: &[SumDataPoint<T>]) {
-    for (i, data_point) in data_points.iter().enumerate() {
+fn print_sum_data_points<'a, T: Debug + 'a>(
+    data_points: impl Iterator<Item = &'a SumDataPoint<T>>,
+) {
+    for (i, data_point) in data_points.enumerate() {
         println!("\t\tDataPoint #{}", i);
         println!("\t\t\tValue        : {:#?}", data_point.value);
         println!("\t\t\tAttributes   :");