From aee66607cde5f60d5057efa3f8a3556a3646c93c Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 17:16:09 -0700
Subject: [PATCH 01/10] Refactor Metrics tests and add more

---
 opentelemetry-sdk/src/metrics/mod.rs | 423 ++++++++++++---------------
 1 file changed, 190 insertions(+), 233 deletions(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index b7b6e23f05..faba7bf7be 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -60,7 +60,7 @@ pub use view::*;
 
 #[cfg(all(test, feature = "testing"))]
 mod tests {
-    use self::data::ScopeMetrics;
+    use self::data::{DataPoint, ScopeMetrics};
     use super::*;
     use crate::metrics::data::{ResourceMetrics, Temporality};
     use crate::metrics::reader::TemporalitySelector;
@@ -73,98 +73,38 @@ mod tests {
     };
     use std::borrow::Cow;
 
+    // Note for all tests in this mod:
     // "multi_thread" tokio flavor must be used else flush won't
     // be able to make progress!
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    async fn counter_aggregation() {
+    async fn counter_aggregation_cumulative() {
         // Run this test with stdout enabled to see output.
-        // cargo test counter --features=metrics,testing -- --nocapture
-
-        // Arrange
-        let exporter = InMemoryMetricsExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
-        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
-
-        // Act
-        let meter = meter_provider.meter("test");
-        let counter = meter
-            .u64_counter("my_counter")
-            .with_unit(Unit::new("my_unit"))
-            .init();
-        counter.add(1, &[KeyValue::new("key1", "value1")]);
-        counter.add(1, &[KeyValue::new("key1", "value1")]);
-        counter.add(1, &[KeyValue::new("key1", "value1")]);
-        counter.add(1, &[KeyValue::new("key1", "value1")]);
-        counter.add(1, &[KeyValue::new("key1", "value1")]);
-
-        counter.add(1, &[KeyValue::new("key1", "value2")]);
-        counter.add(1, &[KeyValue::new("key1", "value2")]);
-        counter.add(1, &[KeyValue::new("key1", "value2")]);
-
-        meter_provider.force_flush().unwrap();
-
-        // Assert
-        let resource_metrics = exporter
-            .get_finished_metrics()
-            .expect("metrics are expected to be exported.");
-        assert!(!resource_metrics.is_empty());
-        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
-        assert_eq!(metric.name, "my_counter");
-        assert_eq!(metric.unit.as_str(), "my_unit");
-        let sum = metric
-            .data
-            .as_any()
-            .downcast_ref::<data::Sum<u64>>()
-            .expect("Sum aggregation expected for Counter instruments by default");
+        // cargo test counter_aggregation_cumulative --features=metrics,testing -- --nocapture
+        counter_aggregation(Temporality::Cumulative);
+    }
 
-        // Expecting 2 time-series.
-        assert_eq!(sum.data_points.len(), 2);
-        assert!(sum.is_monotonic, "Counter should produce monotonic.");
-        assert_eq!(
-            sum.temporality,
-            data::Temporality::Cumulative,
-            "Should produce cumulative by default."
-        );
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn counter_aggregation_delta() {
+        // Run this test with stdout enabled to see output.
+        // cargo test counter_aggregation_delta --features=metrics,testing -- --nocapture
+        counter_aggregation(Temporality::Delta);
+    }
 
-        // find and validate key1=value1 datapoint
-        let mut data_point1 = None;
-        for datapoint in &sum.data_points {
-            if datapoint
-                .attributes
-                .iter()
-                .any(|(k, v)| k.as_str() == "key1" && v.as_str() == "value1")
-            {
-                data_point1 = Some(datapoint);
-            }
-        }
-        assert_eq!(
-            data_point1
-                .expect("datapoint with key1=value1 expected")
-                .value,
-            5
-        );
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn updown_counter_aggregation_cumulative() {
+        // Run this test with stdout enabled to see output.
+        // cargo test counter_aggregation_cumulative --features=metrics,testing -- --nocapture
+        updown_counter_aggregation(Temporality::Cumulative);
+    }
 
-        // find and validate key1=value2 datapoint
-        let mut data_point1 = None;
-        for datapoint in &sum.data_points {
-            if datapoint
-                .attributes
-                .iter()
-                .any(|(k, v)| k.as_str() == "key1" && v.as_str() == "value2")
-            {
-                data_point1 = Some(datapoint);
-            }
-        }
-        assert_eq!(
-            data_point1
-                .expect("datapoint with key1=value2 expected")
-                .value,
-            3
-        );
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn updown_counter_aggregation_delta() {
+        // Run this test with stdout enabled to see output.
+        // cargo test counter_aggregation_delta --features=metrics,testing -- --nocapture
+        updown_counter_aggregation(Temporality::Delta);
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn observable_counter_aggregation() {
         // Run this test with stdout enabled to see output.
@@ -248,8 +188,6 @@ mod tests {
         );
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn counter_duplicate_instrument_merge() {
         // Arrange
@@ -301,8 +239,6 @@ mod tests {
         assert_eq!(datapoint.value, 15);
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn counter_duplicate_instrument_different_meter_no_merge() {
         // Arrange
@@ -392,8 +328,6 @@ mod tests {
         }
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn instrumentation_scope_identity_test() {
         // Arrange
@@ -474,8 +408,6 @@ mod tests {
         assert_eq!(datapoint.value, 15);
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn histogram_aggregation_with_invalid_aggregation_should_proceed_as_if_view_not_exist() {
         // Run this test with stdout enabled to see output.
@@ -527,8 +459,6 @@ mod tests {
         );
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     // #[ignore = "Spatial aggregation is not yet implemented."]
     async fn spatial_aggregation_when_view_drops_attributes_observable_counter() {
@@ -605,8 +535,6 @@ mod tests {
         assert_eq!(data_point.value, 300);
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn spatial_aggregation_when_view_drops_attributes_counter() {
         // cargo test spatial_aggregation_when_view_drops_attributes_counter --features=metrics,testing
@@ -683,21 +611,16 @@ mod tests {
         assert_eq!(data_point.value, 30);
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn counter_aggregation_attribute_order() {
         // Run this test with stdout enabled to see output.
         // cargo test counter_aggregation_attribute_order --features=metrics,testing -- --nocapture
 
         // Arrange
-        let exporter = InMemoryMetricsExporter::default();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
-        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
+        let mut test_context = TestContext::new(Temporality::Delta);
+        let counter = test_context.u64_counter("test", "my_counter", None);
 
         // Act
-        let meter = meter_provider.meter("test");
-        let counter = meter.u64_counter("my_counter").init();
         // Add the same set of attributes in different order. (they are expected
         // to be treated as same attributes)
         counter.add(
@@ -748,33 +671,12 @@ mod tests {
                 KeyValue::new("B", "b"),
             ],
         );
+        test_context.flush_metrics();
 
-        meter_provider.force_flush().unwrap();
-
-        // Assert
-        let resource_metrics = exporter
-            .get_finished_metrics()
-            .expect("metrics are expected to be exported.");
-        assert!(!resource_metrics.is_empty());
-        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
-        assert_eq!(metric.name, "my_counter");
-        let sum = metric
-            .data
-            .as_any()
-            .downcast_ref::<data::Sum<u64>>()
-            .expect("Sum aggregation expected for Counter instruments by default");
+        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         // Expecting 1 time-series.
-        assert_eq!(
-            sum.data_points.len(),
-            1,
-            "Expected only one data point as attributes are same, but just reordered."
-        );
-        assert_eq!(
-            sum.temporality,
-            data::Temporality::Cumulative,
-            "Should produce cumulative by default."
-        );
+        assert_eq!(sum.data_points.len(), 1);
 
         // validate the sole datapoint
         let data_point1 = &sum.data_points[0];
@@ -783,13 +685,13 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn no_attr_cumulative_counter() {
-        let mut test_context = TestContext::new(Some(Temporality::Cumulative));
-        let counter = test_context.u64_counter("test", "my_counter", "my_unit");
+        let mut test_context = TestContext::new(Temporality::Cumulative);
+        let counter = test_context.u64_counter("test", "my_counter", None);
 
         counter.add(50, &[]);
         test_context.flush_metrics();
 
-        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", "my_unit");
+        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
         assert!(sum.is_monotonic, "Should produce monotonic.");
@@ -806,13 +708,13 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn no_attr_delta_counter() {
-        let mut test_context = TestContext::new(Some(Temporality::Delta));
-        let counter = test_context.u64_counter("test", "my_counter", "my_unit");
+        let mut test_context = TestContext::new(Temporality::Delta);
+        let counter = test_context.u64_counter("test", "my_counter", None);
 
         counter.add(50, &[]);
         test_context.flush_metrics();
 
-        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", "my_unit");
+        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
         assert!(sum.is_monotonic, "Should produce monotonic.");
@@ -825,13 +727,13 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn no_attr_cumulative_up_down_counter() {
-        let mut test_context = TestContext::new(Some(Temporality::Cumulative));
-        let counter = test_context.i64_up_down_counter("test", "my_counter", "my_unit");
+        let mut test_context = TestContext::new(Temporality::Cumulative);
+        let counter = test_context.i64_up_down_counter("test", "my_counter", Some("my_unit"));
 
         counter.add(50, &[]);
         test_context.flush_metrics();
 
-        let sum = test_context.get_aggregation::<data::Sum<i64>>("my_counter", "my_unit");
+        let sum = test_context.get_aggregation::<data::Sum<i64>>("my_counter", Some("my_unit"));
 
         assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
         assert!(!sum.is_monotonic, "Should not produce monotonic.");
@@ -848,13 +750,13 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn no_attr_delta_up_down_counter() {
-        let mut test_context = TestContext::new(Some(Temporality::Delta));
-        let counter = test_context.i64_up_down_counter("test", "my_counter", "my_unit");
+        let mut test_context = TestContext::new(Temporality::Delta);
+        let counter = test_context.i64_up_down_counter("test", "my_counter", Some("my_unit"));
 
         counter.add(50, &[]);
         test_context.flush_metrics();
 
-        let sum = test_context.get_aggregation::<data::Sum<i64>>("my_counter", "my_unit");
+        let sum = test_context.get_aggregation::<data::Sum<i64>>("my_counter", Some("my_unit"));
 
         assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
         assert!(!sum.is_monotonic, "Should not produce monotonic.");
@@ -867,16 +769,16 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn no_attr_cumulative_counter_value_added_after_export() {
-        let mut test_context = TestContext::new(Some(Temporality::Cumulative));
-        let counter = test_context.u64_counter("test", "my_counter", "my_unit");
+        let mut test_context = TestContext::new(Temporality::Cumulative);
+        let counter = test_context.u64_counter("test", "my_counter", None);
 
         counter.add(50, &[]);
         test_context.flush_metrics();
-        let _ = test_context.get_aggregation::<data::Sum<u64>>("my_counter", "my_unit");
+        let _ = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         counter.add(5, &[]);
         test_context.flush_metrics();
-        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", "my_unit");
+        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
         assert!(sum.is_monotonic, "Should produce monotonic.");
@@ -893,16 +795,16 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn no_attr_delta_counter_value_reset_after_export() {
-        let mut test_context = TestContext::new(Some(Temporality::Delta));
-        let counter = test_context.u64_counter("test", "my_counter", "my_unit");
+        let mut test_context = TestContext::new(Temporality::Delta);
+        let counter = test_context.u64_counter("test", "my_counter", None);
 
         counter.add(50, &[]);
         test_context.flush_metrics();
-        let _ = test_context.get_aggregation::<data::Sum<u64>>("my_counter", "my_unit");
+        let _ = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         counter.add(5, &[]);
         test_context.flush_metrics();
-        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", "my_unit");
+        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
         assert!(sum.is_monotonic, "Should produce monotonic.");
@@ -919,16 +821,16 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn second_delta_export_does_not_give_no_attr_value_if_add_not_called() {
-        let mut test_context = TestContext::new(Some(Temporality::Delta));
-        let counter = test_context.u64_counter("test", "my_counter", "my_unit");
+        let mut test_context = TestContext::new(Temporality::Delta);
+        let counter = test_context.u64_counter("test", "my_counter", None);
 
         counter.add(50, &[]);
         test_context.flush_metrics();
-        let _ = test_context.get_aggregation::<data::Sum<u64>>("my_counter", "my_unit");
+        let _ = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         counter.add(50, &[KeyValue::new("a", "b")]);
         test_context.flush_metrics();
-        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", "my_unit");
+        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
         let no_attr_data_point = sum.data_points.iter().find(|x| x.attributes.is_empty());
 
@@ -938,8 +840,6 @@ mod tests {
         );
     }
 
-    // "multi_thread" tokio flavor must be used else flush won't
-    // be able to make progress!
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     #[ignore = "Known bug: https://github.com/open-telemetry/opentelemetry-rust/issues/1598"]
     async fn delta_memory_efficiency_test() {
@@ -947,18 +847,10 @@ mod tests {
         // cargo test delta_memory_efficiency_test --features=metrics,testing -- --nocapture
 
         // Arrange
-        let exporter = InMemoryMetricsExporterBuilder::new()
-            .with_temporality_selector(DeltaTemporalitySelector())
-            .build();
-        let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
-        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
+        let mut test_context = TestContext::new(Temporality::Delta);
+        let counter = test_context.u64_counter("test", "my_counter", None);
 
         // Act
-        let meter = meter_provider.meter("test");
-        let counter = meter
-            .u64_counter("my_counter")
-            .with_unit(Unit::new("my_unit"))
-            .init();
         counter.add(1, &[KeyValue::new("key1", "value1")]);
         counter.add(1, &[KeyValue::new("key1", "value1")]);
         counter.add(1, &[KeyValue::new("key1", "value1")]);
@@ -968,76 +860,139 @@ mod tests {
         counter.add(1, &[KeyValue::new("key1", "value2")]);
         counter.add(1, &[KeyValue::new("key1", "value2")]);
         counter.add(1, &[KeyValue::new("key1", "value2")]);
+        test_context.flush_metrics();
 
-        meter_provider.force_flush().unwrap();
+        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
 
-        // Assert
-        let resource_metrics = exporter
+        // Expecting 2 time-series.
+        assert_eq!(sum.data_points.len(), 2);
+
+        // find and validate key1=value1 datapoint
+        let data_point1 = find_datapoint_with_key_value(&sum.data_points, "key1", "value1")
+            .expect("datapoint with key1=value1 expected");
+        assert_eq!(data_point1.value, 5);
+
+        // find and validate key1=value2 datapoint
+        let data_point1 = find_datapoint_with_key_value(&sum.data_points, "key1", "value2")
+            .expect("datapoint with key1=value2 expected");
+        assert_eq!(data_point1.value, 3);
+
+        // flush again, and validate that nothing is flushed
+        // as delta temporality.
+        test_context.flush_metrics();
+
+        let resource_metrics = test_context
+            .exporter
             .get_finished_metrics()
             .expect("metrics are expected to be exported.");
-        assert!(!resource_metrics.is_empty());
-        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
-        assert_eq!(metric.name, "my_counter");
-        assert_eq!(metric.unit.as_str(), "my_unit");
-        let sum = metric
-            .data
-            .as_any()
-            .downcast_ref::<data::Sum<u64>>()
-            .expect("Sum aggregation expected for Counter instruments by default");
+        println!("resource_metrics: {:?}", resource_metrics);
+        assert!(resource_metrics.is_empty(), "No metrics should be exported as no new measurements were recorded since last collect.");
+    }
+
+    fn counter_aggregation(temporality: Temporality) {
+        // Arrange
+        let mut test_context = TestContext::new(temporality);
+        let counter = test_context.u64_counter("test", "my_counter", None);
+
+        // Act
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+
+        counter.add(1, &[KeyValue::new("key1", "value2")]);
+        counter.add(1, &[KeyValue::new("key1", "value2")]);
+        counter.add(1, &[KeyValue::new("key1", "value2")]);
+
+        test_context.flush_metrics();
 
+        // Assert
+        let sum = test_context.get_aggregation::<data::Sum<u64>>("my_counter", None);
         // Expecting 2 time-series.
         assert_eq!(sum.data_points.len(), 2);
         assert!(sum.is_monotonic, "Counter should produce monotonic.");
-        assert_eq!(
-            sum.temporality,
-            data::Temporality::Delta,
-            "Should produce Delta as configured"
-        );
-
-        // find and validate key1=value1 datapoint
-        let mut data_point1 = None;
-        for datapoint in &sum.data_points {
-            if datapoint
-                .attributes
-                .iter()
-                .any(|(k, v)| k.as_str() == "key1" && v.as_str() == "value1")
-            {
-                data_point1 = Some(datapoint);
-            }
+        if let Temporality::Cumulative = temporality {
+            assert_eq!(
+                sum.temporality,
+                Temporality::Cumulative,
+                "Should produce cumulative"
+            );
+        } else {
+            assert_eq!(sum.temporality, Temporality::Delta, "Should produce delta");
         }
-        assert_eq!(
-            data_point1
-                .expect("datapoint with key1=value1 expected")
-                .value,
-            5
+
+        // find and validate key1=value2 datapoint
+        let data_point1 = find_datapoint_with_key_value(&sum.data_points, "key1", "value1")
+            .expect("datapoint with key1=value1 expected");
+        assert_eq!(data_point1.value, 5);
+
+        let data_point1 = find_datapoint_with_key_value(&sum.data_points, "key1", "value2")
+            .expect("datapoint with key1=value2 expected");
+        assert_eq!(data_point1.value, 3);
+    }
+
+    fn updown_counter_aggregation(temporality: Temporality) {
+        // Arrange
+        let mut test_context = TestContext::new(temporality);
+        let counter = test_context.i64_up_down_counter("test", "my_counter", None);
+
+        // Act
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+        counter.add(1, &[KeyValue::new("key1", "value1")]);
+
+        counter.add(1, &[KeyValue::new("key1", "value2")]);
+        counter.add(1, &[KeyValue::new("key1", "value2")]);
+        counter.add(1, &[KeyValue::new("key1", "value2")]);
+
+        test_context.flush_metrics();
+
+        // Assert
+        let sum = test_context.get_aggregation::<data::Sum<i64>>("my_counter", None);
+        // Expecting 2 time-series.
+        assert_eq!(sum.data_points.len(), 2);
+        assert!(
+            !sum.is_monotonic,
+            "UpDownCounter should produce non-monotonic."
         );
+        if let Temporality::Cumulative = temporality {
+            assert_eq!(
+                sum.temporality,
+                Temporality::Cumulative,
+                "Should produce cumulative"
+            );
+        } else {
+            assert_eq!(sum.temporality, Temporality::Delta, "Should produce delta");
+        }
 
         // find and validate key1=value2 datapoint
-        let mut data_point1 = None;
-        for datapoint in &sum.data_points {
+        let data_point1 = find_datapoint_with_key_value(&sum.data_points, "key1", "value1")
+            .expect("datapoint with key1=value1 expected");
+        assert_eq!(data_point1.value, 5);
+
+        let data_point1 = find_datapoint_with_key_value(&sum.data_points, "key1", "value2")
+            .expect("datapoint with key1=value2 expected");
+        assert_eq!(data_point1.value, 3);
+    }
+
+    fn find_datapoint_with_key_value<'a, T>(
+        data_points: &'a Vec<DataPoint<T>>,
+        key: &str,
+        value: &str,
+    ) -> Option<&'a DataPoint<T>> {
+        for datapoint in data_points {
             if datapoint
                 .attributes
                 .iter()
-                .any(|(k, v)| k.as_str() == "key1" && v.as_str() == "value2")
+                .any(|(k, v)| k.as_str() == key && v.as_str() == value)
             {
-                data_point1 = Some(datapoint);
+                return Some(&datapoint);
             }
         }
-        assert_eq!(
-            data_point1
-                .expect("datapoint with key1=value2 expected")
-                .value,
-            3
-        );
-
-        // flush again, and validate that nothing is flushed
-        // as delta temporality.
-        meter_provider.force_flush().unwrap();
-        let resource_metrics = exporter
-            .get_finished_metrics()
-            .expect("metrics are expected to be exported.");
-        println!("resource_metrics: {:?}", resource_metrics);
-        assert!(resource_metrics.is_empty(), "No metrics should be exported as no new measurements were recorded since last collect.");
+        None
     }
 
     fn find_scope_metric<'a>(
@@ -1065,7 +1020,7 @@ mod tests {
     }
 
     impl TestContext {
-        fn new(temporality: Option<Temporality>) -> Self {
+        fn new(temporality: Temporality) -> Self {
             struct TestTemporalitySelector(Temporality);
             impl TemporalitySelector for TestTemporalitySelector {
                 fn temporality(&self, _kind: InstrumentKind) -> Temporality {
@@ -1074,9 +1029,7 @@ mod tests {
             }
 
             let mut exporter = InMemoryMetricsExporterBuilder::new();
-            if let Some(temporality) = temporality {
-                exporter = exporter.with_temporality_selector(TestTemporalitySelector(temporality));
-            }
+            exporter = exporter.with_temporality_selector(TestTemporalitySelector(temporality));
 
             let exporter = exporter.build();
             let reader = PeriodicReader::builder(exporter.clone(), runtime::Tokio).build();
@@ -1093,26 +1046,28 @@ mod tests {
             &self,
             meter_name: &'static str,
             counter_name: &'static str,
-            unit_name: &'static str,
+            unit: Option<&'static str>,
         ) -> Counter<u64> {
-            self.meter_provider
-                .meter(meter_name)
-                .u64_counter(counter_name)
-                .with_unit(Unit::new(unit_name))
-                .init()
+            let meter = self.meter_provider.meter(meter_name);
+            let mut counter_builder = meter.u64_counter(counter_name);
+            if let Some(unit_name) = unit {
+                counter_builder = counter_builder.with_unit(Unit::new(unit_name));
+            }
+            counter_builder.init()
         }
 
         fn i64_up_down_counter(
             &self,
             meter_name: &'static str,
             counter_name: &'static str,
-            unit_name: &'static str,
+            unit: Option<&'static str>,
         ) -> UpDownCounter<i64> {
-            self.meter_provider
-                .meter(meter_name)
-                .i64_up_down_counter(counter_name)
-                .with_unit(Unit::new(unit_name))
-                .init()
+            let meter = self.meter_provider.meter(meter_name);
+            let mut updown_counter_builder = meter.i64_up_down_counter(counter_name);
+            if let Some(unit_name) = unit {
+                updown_counter_builder = updown_counter_builder.with_unit(Unit::new(unit_name));
+            }
+            updown_counter_builder.init()
         }
 
         fn flush_metrics(&self) {
@@ -1122,7 +1077,7 @@ mod tests {
         fn get_aggregation<T: data::Aggregation>(
             &mut self,
             counter_name: &str,
-            unit_name: &str,
+            unit_name: Option<&str>,
         ) -> &T {
             self.resource_metrics = self
                 .exporter
@@ -1145,7 +1100,9 @@ mod tests {
 
             let metric = &resource_metric.scope_metrics[0].metrics[0];
             assert_eq!(metric.name, counter_name);
-            assert_eq!(metric.unit.as_str(), unit_name);
+            if let Some(expected_unit) = unit_name {
+                assert_eq!(metric.unit.as_str(), expected_unit);
+            }
 
             metric
                 .data

From ebae05e1c26c38a3b8cc6f43323ef2496d6bc272 Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 17:18:54 -0700
Subject: [PATCH 02/10] comment

---
 opentelemetry-sdk/src/metrics/mod.rs | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index faba7bf7be..2635c3b0a7 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -73,6 +73,9 @@ mod tests {
     };
     use std::borrow::Cow;
 
+    // Run all tests in this mod
+    // cargo test metrics::tests --features=metrics,testing
+
     // Note for all tests in this mod:
     // "multi_thread" tokio flavor must be used else flush won't
     // be able to make progress!
@@ -81,28 +84,28 @@ mod tests {
     async fn counter_aggregation_cumulative() {
         // Run this test with stdout enabled to see output.
         // cargo test counter_aggregation_cumulative --features=metrics,testing -- --nocapture
-        counter_aggregation(Temporality::Cumulative);
+        counter_aggregation_helper(Temporality::Cumulative);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn counter_aggregation_delta() {
         // Run this test with stdout enabled to see output.
         // cargo test counter_aggregation_delta --features=metrics,testing -- --nocapture
-        counter_aggregation(Temporality::Delta);
+        counter_aggregation_helper(Temporality::Delta);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn updown_counter_aggregation_cumulative() {
         // Run this test with stdout enabled to see output.
         // cargo test counter_aggregation_cumulative --features=metrics,testing -- --nocapture
-        updown_counter_aggregation(Temporality::Cumulative);
+        updown_counter_aggregation_helper(Temporality::Cumulative);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn updown_counter_aggregation_delta() {
         // Run this test with stdout enabled to see output.
         // cargo test counter_aggregation_delta --features=metrics,testing -- --nocapture
-        updown_counter_aggregation(Temporality::Delta);
+        updown_counter_aggregation_helper(Temporality::Delta);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -889,7 +892,7 @@ mod tests {
         assert!(resource_metrics.is_empty(), "No metrics should be exported as no new measurements were recorded since last collect.");
     }
 
-    fn counter_aggregation(temporality: Temporality) {
+    fn counter_aggregation_helper(temporality: Temporality) {
         // Arrange
         let mut test_context = TestContext::new(temporality);
         let counter = test_context.u64_counter("test", "my_counter", None);
@@ -932,7 +935,7 @@ mod tests {
         assert_eq!(data_point1.value, 3);
     }
 
-    fn updown_counter_aggregation(temporality: Temporality) {
+    fn updown_counter_aggregation_helper(temporality: Temporality) {
         // Arrange
         let mut test_context = TestContext::new(temporality);
         let counter = test_context.i64_up_down_counter("test", "my_counter", None);

From 4c4fae8e61bc8d93c09e11040960429e752bcdd8 Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 17:27:29 -0700
Subject: [PATCH 03/10] unused

---
 opentelemetry-sdk/src/metrics/mod.rs | 7 -------
 1 file changed, 7 deletions(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index 2635c3b0a7..d7685df371 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -1007,13 +1007,6 @@ mod tests {
             .find(|&scope_metric| scope_metric.scope.name == name)
     }
 
-    struct DeltaTemporalitySelector();
-    impl TemporalitySelector for DeltaTemporalitySelector {
-        fn temporality(&self, _kind: InstrumentKind) -> Temporality {
-            Temporality::Delta
-        }
-    }
-
     struct TestContext {
         exporter: InMemoryMetricsExporter,
         meter_provider: SdkMeterProvider,

From 33e88ffffef4651f0310940b31eb66f3b079996b Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 17:28:08 -0700
Subject: [PATCH 04/10] clip

---
 opentelemetry-sdk/src/metrics/mod.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index d7685df371..1e24a53b11 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -992,7 +992,7 @@ mod tests {
                 .iter()
                 .any(|(k, v)| k.as_str() == key && v.as_str() == value)
             {
-                return Some(&datapoint);
+                return Some(datapoint);
             }
         }
         None

From ee8790a7d6c02c24a56500b1eadb7f951ab7877f Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 17:28:22 -0700
Subject: [PATCH 05/10] clipp

---
 opentelemetry-sdk/src/metrics/mod.rs | 10 ++--------
 1 file changed, 2 insertions(+), 8 deletions(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index 1e24a53b11..b9ee1a9f93 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -986,16 +986,10 @@ mod tests {
         key: &str,
         value: &str,
     ) -> Option<&'a DataPoint<T>> {
-        for datapoint in data_points {
-            if datapoint
+        data_points.iter().find(|&datapoint| datapoint
                 .attributes
                 .iter()
-                .any(|(k, v)| k.as_str() == key && v.as_str() == value)
-            {
-                return Some(datapoint);
-            }
-        }
-        None
+                .any(|(k, v)| k.as_str() == key && v.as_str() == value))
     }
 
     fn find_scope_metric<'a>(

From 5fbdc92532f0dbca60a62b46c5a70083ebb4664c Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 17:29:15 -0700
Subject: [PATCH 06/10] slice

---
 opentelemetry-sdk/src/metrics/mod.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index b9ee1a9f93..ebf5a8ba67 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -982,7 +982,7 @@ mod tests {
     }
 
     fn find_datapoint_with_key_value<'a, T>(
-        data_points: &'a Vec<DataPoint<T>>,
+        data_points: &'a DataPoint<T>,
         key: &str,
         value: &str,
     ) -> Option<&'a DataPoint<T>> {

From 7df7ff6d111ef1d67be8fd6a6a9b79e24ffbc2a3 Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 17:31:53 -0700
Subject: [PATCH 07/10] nit fix

---
 opentelemetry-sdk/src/metrics/mod.rs | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index ebf5a8ba67..d7685df371 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -982,14 +982,20 @@ mod tests {
     }
 
     fn find_datapoint_with_key_value<'a, T>(
-        data_points: &'a DataPoint<T>,
+        data_points: &'a Vec<DataPoint<T>>,
         key: &str,
         value: &str,
     ) -> Option<&'a DataPoint<T>> {
-        data_points.iter().find(|&datapoint| datapoint
+        for datapoint in data_points {
+            if datapoint
                 .attributes
                 .iter()
-                .any(|(k, v)| k.as_str() == key && v.as_str() == value))
+                .any(|(k, v)| k.as_str() == key && v.as_str() == value)
+            {
+                return Some(&datapoint);
+            }
+        }
+        None
     }
 
     fn find_scope_metric<'a>(

From 174cdac30faca7a928f3e5f95bd73d53b157e3be Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 17:54:46 -0700
Subject: [PATCH 08/10] fix lint

---
 opentelemetry-sdk/src/metrics/mod.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index d7685df371..1e24a53b11 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -992,7 +992,7 @@ mod tests {
                 .iter()
                 .any(|(k, v)| k.as_str() == key && v.as_str() == value)
             {
-                return Some(&datapoint);
+                return Some(datapoint);
             }
         }
         None

From 1608f8ea338f58484cf695063fc1ff9683b57ab6 Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 18:10:50 -0700
Subject: [PATCH 09/10] use iter

---
 opentelemetry-sdk/src/metrics/mod.rs | 10 +++-------
 1 file changed, 3 insertions(+), 7 deletions(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index 1e24a53b11..0263a4d428 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -986,16 +986,12 @@ mod tests {
         key: &str,
         value: &str,
     ) -> Option<&'a DataPoint<T>> {
-        for datapoint in data_points {
-            if datapoint
+        data_points.iter().find(|&datapoint| {
+            datapoint
                 .attributes
                 .iter()
                 .any(|(k, v)| k.as_str() == key && v.as_str() == value)
-            {
-                return Some(datapoint);
-            }
-        }
-        None
+        })
     }
 
     fn find_scope_metric<'a>(

From a72110d10669921b44be61c817cc95a8551b0da3 Mon Sep 17 00:00:00 2001
From: Cijo Thomas <cijo.thomas@gmail.com>
Date: Fri, 10 May 2024 18:19:55 -0700
Subject: [PATCH 10/10] slice

---
 opentelemetry-sdk/src/metrics/mod.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
index 0263a4d428..557cfc2611 100644
--- a/opentelemetry-sdk/src/metrics/mod.rs
+++ b/opentelemetry-sdk/src/metrics/mod.rs
@@ -982,7 +982,7 @@ mod tests {
     }
 
     fn find_datapoint_with_key_value<'a, T>(
-        data_points: &'a Vec<DataPoint<T>>,
+        data_points: &'a [DataPoint<T>],
         key: &str,
         value: &str,
     ) -> Option<&'a DataPoint<T>> {