Add tests for Histogram metrics

Author: cijothomas

opentelemetry-sdk/src/metrics/mod.rs

@@ -139,14 +139,15 @@ impl Hash for AttributeSet {
 
 #[cfg(all(test, feature = "testing"))]
 mod tests {
-    use self::data::{DataPoint, ScopeMetrics};
+    use self::data::{DataPoint, HistogramDataPoint, ScopeMetrics};
     use super::*;
     use crate::metrics::data::{ResourceMetrics, Temporality};
     use crate::metrics::reader::TemporalitySelector;
     use crate::testing::metrics::InMemoryMetricsExporterBuilder;
     use crate::{runtime, testing::metrics::InMemoryMetricsExporter};
     use opentelemetry::metrics::{Counter, Meter, UpDownCounter};
     use opentelemetry::{metrics::MeterProvider as _, KeyValue};
+    use rand::{rngs, Rng, SeedableRng};
     use std::borrow::Cow;
     use std::sync::{Arc, Mutex};
 
@@ -199,6 +200,20 @@ mod tests {
         counter_aggregation_helper(Temporality::Delta);
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn histogram_aggregation_cumulative() {
+        // Run this test with stdout enabled to see output.
+        // cargo test histogram_aggregation_cumulative --features=metrics,testing -- --nocapture
+        histogram_aggregation_helper(Temporality::Cumulative);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn histogram_aggregation_delta() {
+        // Run this test with stdout enabled to see output.
+        // cargo test histogram_aggregation_delta --features=metrics,testing -- --nocapture
+        histogram_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.
@@ -1007,6 +1022,65 @@ mod tests {
         assert!(resource_metrics.is_empty(), "No metrics should be exported as no new measurements were recorded since last collect.");
     }
 
+    fn histogram_aggregation_helper(temporality: Temporality) {
+        // Arrange
+        let mut test_context = TestContext::new(temporality);
+        let histogram = test_context.meter().u64_histogram("my_histogram").init();
+
+        // Act
+        let mut rand = rngs::SmallRng::from_entropy();
+        let values_kv1 = (0..50)
+            .map(|_| rand.gen_range(0..100))
+            .collect::<Vec<u64>>();
+        for value in values_kv1.iter() {
+            histogram.record(*value, &[KeyValue::new("key1", "value1")]);
+        }
+
+        let values_kv2 = (0..30)
+            .map(|_| rand.gen_range(0..100))
+            .collect::<Vec<u64>>();
+        for value in values_kv2.iter() {
+            histogram.record(*value, &[KeyValue::new("key1", "value2")]);
+        }
+
+        test_context.flush_metrics();
+
+        // Assert
+        let histogram = test_context.get_aggregation::<data::Histogram<u64>>("my_histogram", None);
+        // Expecting 2 time-series.
+        assert_eq!(histogram.data_points.len(), 2);
+        if let Temporality::Cumulative = temporality {
+            assert_eq!(
+                histogram.temporality,
+                Temporality::Cumulative,
+                "Should produce cumulative"
+            );
+        } else {
+            assert_eq!(
+                histogram.temporality,
+                Temporality::Delta,
+                "Should produce delta"
+            );
+        }
+
+        // find and validate key1=value2 datapoint
+        let data_point1 =
+            find_histogram_datapoint_with_key_value(&histogram.data_points, "key1", "value1")
+                .expect("datapoint with key1=value1 expected");
+        assert_eq!(data_point1.count, values_kv1.len() as u64);
+        assert_eq!(data_point1.sum, values_kv1.iter().sum::<u64>());
+        assert_eq!(data_point1.min.unwrap(), *values_kv1.iter().min().unwrap());
+        assert_eq!(data_point1.max.unwrap(), *values_kv1.iter().max().unwrap());
+
+        let data_point2 =
+            find_histogram_datapoint_with_key_value(&histogram.data_points, "key1", "value2")
+                .expect("datapoint with key1=value2 expected");
+        assert_eq!(data_point2.count, values_kv2.len() as u64);
+        assert_eq!(data_point2.sum, values_kv2.iter().sum::<u64>());
+        assert_eq!(data_point2.min.unwrap(), *values_kv2.iter().min().unwrap());
+        assert_eq!(data_point2.max.unwrap(), *values_kv2.iter().max().unwrap());
+    }
+
     fn counter_aggregation_helper(temporality: Temporality) {
         // Arrange
         let mut test_context = TestContext::new(temporality);
@@ -1109,6 +1183,19 @@ mod tests {
         })
     }
 
+    fn find_histogram_datapoint_with_key_value<'a, T>(
+        data_points: &'a [HistogramDataPoint<T>],
+        key: &str,
+        value: &str,
+    ) -> Option<&'a HistogramDataPoint<T>> {
+        data_points.iter().find(|&datapoint| {
+            datapoint
+                .attributes
+                .iter()
+                .any(|kv| kv.key.as_str() == key && kv.value.as_str() == value)
+        })
+    }
+
     fn find_scope_metric<'a>(
         metrics: &'a [ScopeMetrics],
         name: &'a str,

stress/Cargo.toml

@@ -4,11 +4,16 @@ version = "0.1.0"
 edition = "2021"
 publish = false
 
-[[bin]] # Bin to run the metrics stress tests
+[[bin]] # Bin to run the metrics stress tests for Counter
 name = "metrics"
 path = "src/metrics.rs"
 doc = false
 
+[[bin]] # Bin to run the metrics stress tests for Histogram
+name = "metrics_histogram"
+path = "src/metrics_histogram.rs"
+doc = false
+
 [[bin]] # Bin to run the metrics overflow stress tests
 name = "metrics_overflow"
 path = "src/metrics_overflow.rs"

stress/src/metrics_histogram.rs

@@ -0,0 +1,69 @@
+/*
+    Stress test results:
+    OS: Ubuntu 22.04.3 LTS (5.15.146.1-microsoft-standard-WSL2)
+    Hardware: AMD EPYC 7763 64-Core Processor - 2.44 GHz, 16vCPUs,
+    RAM: 64.0 GB
+    4.6M /sec
+*/
+
+use lazy_static::lazy_static;
+use opentelemetry::{
+    metrics::{Histogram, MeterProvider as _},
+    KeyValue,
+};
+use opentelemetry_sdk::metrics::{ManualReader, SdkMeterProvider};
+use rand::{
+    rngs::{self},
+    Rng, SeedableRng,
+};
+use std::{borrow::Cow, cell::RefCell};
+
+mod throughput;
+
+lazy_static! {
+    static ref PROVIDER: SdkMeterProvider = SdkMeterProvider::builder()
+        .with_reader(ManualReader::builder().build())
+        .build();
+    static ref ATTRIBUTE_VALUES: [&'static str; 10] = [
+        "value1", "value2", "value3", "value4", "value5", "value6", "value7", "value8", "value9",
+        "value10"
+    ];
+    static ref HISTOGRAM: Histogram<u64> = PROVIDER
+        .meter(<&str as Into<Cow<'static, str>>>::into("test"))
+        .u64_histogram("hello")
+        .init();
+}
+
+thread_local! {
+    /// Store random number generator for each thread
+    static CURRENT_RNG: RefCell<rngs::SmallRng> = RefCell::new(rngs::SmallRng::from_entropy());
+}
+
+fn main() {
+    throughput::test_throughput(test_counter);
+}
+
+fn test_counter() {
+    let len = ATTRIBUTE_VALUES.len();
+    let rands = CURRENT_RNG.with(|rng| {
+        let mut rng = rng.borrow_mut();
+        [
+            rng.gen_range(0..len),
+            rng.gen_range(0..len),
+            rng.gen_range(0..len),
+        ]
+    });
+    let index_first_attribute = rands[0];
+    let index_second_attribute = rands[1];
+    let index_third_attribute = rands[2];
+
+    // each attribute has 10 possible values, so there are 1000 possible combinations (time-series)
+    HISTOGRAM.record(
+        1,
+        &[
+            KeyValue::new("attribute1", ATTRIBUTE_VALUES[index_first_attribute]),
+            KeyValue::new("attribute2", ATTRIBUTE_VALUES[index_second_attribute]),
+            KeyValue::new("attribute3", ATTRIBUTE_VALUES[index_third_attribute]),
+        ],
+    );
+}