Merge branch 'main' into http-json-example

TommyCpp · web-flow · commit 61a6e5e08c37 · 2024-05-28T23:29:25.000-07:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -71,7 +71,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@nightly
         with:
-          toolchain: nightly-2024-02-07
+          toolchain: nightly-2024-05-01
           components: rustfmt
       - name: external-type-check
         run: |
diff --git a/opentelemetry-prometheus/tests/integration_test.rs b/opentelemetry-prometheus/tests/integration_test.rs
@@ -406,18 +406,18 @@ fn gather_and_compare(registry: prometheus::Registry, expected: String, name: &'
     let metric_families = registry.gather();
     encoder.encode(&metric_families, &mut output).unwrap();
 
+    let expected = get_platform_specific_string(expected);
     let output_string = get_platform_specific_string(String::from_utf8(output).unwrap());
 
     assert_eq!(output_string, expected, "{name}");
 }
 
 ///  Returns a String which uses the platform specific new line feed character.
 fn get_platform_specific_string(input: String) -> String {
-    if cfg!(windows) {
-        input.replace('\n', "\r\n")
-    } else {
-        input
+    if cfg!(windows) && !input.ends_with("\r\n") && input.ends_with('\n') {
+        return input.replace('\n', "\r\n");
     }
+    input
 }
 
 #[test]
@@ -812,6 +812,7 @@ fn duplicate_metrics() {
             .expected_files
             .into_iter()
             .map(|f| fs::read_to_string(Path::new("./tests/data").join(f)).expect(f))
+            .map(get_platform_specific_string)
             .collect();
         gather_and_compare_multi(registry, possible_matches, tc.name);
     }
diff --git a/opentelemetry-proto/tests/grpc_build.rs b/opentelemetry-proto/tests/grpc_build.rs
@@ -18,7 +18,7 @@ const TONIC_INCLUDES: &[&str] = &["src/proto/opentelemetry-proto", "src/proto"];
 
 #[test]
 fn build_tonic() {
-    let before_build = build_content_map(TONIC_OUT_DIR, false);
+    let before_build = build_content_map(TONIC_OUT_DIR, true);
 
     let out_dir = TempDir::new().expect("failed to create temp dir to store the generated files");
 
@@ -130,13 +130,12 @@ fn build_content_map(path: impl AsRef<Path>, normalize_line_feed: bool) -> HashM
         .collect()
 }
 
-///  Returns a String with the platform specific new line feed character.
+///  Returns a String which uses the platform specific new line feed character.
 fn get_platform_specific_string(input: String) -> String {
-    if cfg!(windows) {
-        input.replace('\n', "\r\n")
-    } else {
-        input
+    if cfg!(windows) && !input.ends_with("\r\n") && input.ends_with('\n') {
+        return input.replace('\n', "\r\n");
     }
+    input
 }
 
 fn ensure_files_are_same(
diff --git a/opentelemetry-sdk/CHANGELOG.md b/opentelemetry-sdk/CHANGELOG.md
@@ -13,6 +13,8 @@
 - Removed `XrayIdGenerator`, which was marked deprecated since 0.21.3. Use
   [`opentelemetry-aws`](https://crates.io/crates/opentelemetry-aws), version
   0.10.0 or newer.
+- Performance Improvement - Counter/UpDownCounter instruments internally use
+  `RwLock` instead of `Mutex` to reduce contention.
 
 - **Breaking** [1726](https://github.com/open-telemetry/opentelemetry-rust/pull/1726)
   Update `LogProcessor::emit() method to take mutable reference to LogData. This is breaking
diff --git a/opentelemetry-sdk/src/metrics/internal/sum.rs b/opentelemetry-sdk/src/metrics/internal/sum.rs
@@ -1,8 +1,8 @@
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::vec;
 use std::{
-    collections::{hash_map::Entry, HashMap},
-    sync::Mutex,
+    collections::HashMap,
+    sync::{Mutex, RwLock},
     time::SystemTime,
 };
 
@@ -18,7 +18,7 @@ use super::{
 
 /// The storage for sums.
 struct ValueMap<T: Number<T>> {
-    values: Mutex<HashMap<AttributeSet, T>>,
+    values: RwLock<HashMap<AttributeSet, T::AtomicTracker>>,
     has_no_value_attribute_value: AtomicBool,
     no_attribute_value: T::AtomicTracker,
 }
@@ -32,7 +32,7 @@ impl<T: Number<T>> Default for ValueMap<T> {
 impl<T: Number<T>> ValueMap<T> {
     fn new() -> Self {
         ValueMap {
-            values: Mutex::new(HashMap::new()),
+            values: RwLock::new(HashMap::new()),
             has_no_value_attribute_value: AtomicBool::new(false),
             no_attribute_value: T::new_atomic_tracker(),
         }
@@ -45,21 +45,31 @@ impl<T: Number<T>> ValueMap<T> {
             self.no_attribute_value.add(measurement);
             self.has_no_value_attribute_value
                 .store(true, Ordering::Release);
-        } else if let Ok(mut values) = self.values.lock() {
-            let size = values.len();
-            match values.entry(attrs) {
-                Entry::Occupied(mut occupied_entry) => {
-                    let sum = occupied_entry.get_mut();
-                    *sum += measurement;
-                }
-                Entry::Vacant(vacant_entry) => {
-                    if is_under_cardinality_limit(size) {
-                        vacant_entry.insert(measurement);
-                    } else if let Some(val) = values.get_mut(&STREAM_OVERFLOW_ATTRIBUTE_SET) {
-                        *val += measurement;
+        } else if let Ok(values) = self.values.read() {
+            if let Some(value_to_update) = values.get(&attrs) {
+                value_to_update.add(measurement);
+                return;
+            } else {
+                drop(values);
+                if let Ok(mut values) = self.values.write() {
+                    // Recheck after acquiring write lock, in case another
+                    // thread has added the value.
+                    if let Some(value_to_update) = values.get(&attrs) {
+                        value_to_update.add(measurement);
+                        return;
+                    } else if is_under_cardinality_limit(values.len()) {
+                        let new_value = T::new_atomic_tracker();
+                        new_value.add(measurement);
+                        values.insert(attrs, new_value);
+                    } else if let Some(overflow_value) =
+                        values.get_mut(&STREAM_OVERFLOW_ATTRIBUTE_SET)
+                    {
+                        overflow_value.add(measurement);
                         return;
                     } else {
-                        values.insert(STREAM_OVERFLOW_ATTRIBUTE_SET.clone(), measurement);
+                        let new_value = T::new_atomic_tracker();
+                        new_value.add(measurement);
+                        values.insert(STREAM_OVERFLOW_ATTRIBUTE_SET.clone(), new_value);
                         global::handle_error(MetricsError::Other("Warning: Maximum data points for metric stream exceeded. Entry added to overflow. Subsequent overflows to same metric until next collect will not be logged.".into()));
                     }
                 }
@@ -114,7 +124,7 @@ impl<T: Number<T>> Sum<T> {
         s_data.is_monotonic = self.monotonic;
         s_data.data_points.clear();
 
-        let mut values = match self.value_map.values.lock() {
+        let mut values = match self.value_map.values.write() {
             Ok(v) => v,
             Err(_) => return (0, None),
         };
@@ -149,7 +159,7 @@ impl<T: Number<T>> Sum<T> {
                     .collect(),
                 start_time: Some(prev_start),
                 time: Some(t),
-                value,
+                value: value.get_value(),
                 exemplars: vec![],
             });
         }
@@ -186,7 +196,7 @@ impl<T: Number<T>> Sum<T> {
         s_data.is_monotonic = self.monotonic;
         s_data.data_points.clear();
 
-        let values = match self.value_map.values.lock() {
+        let values = match self.value_map.values.write() {
             Ok(v) => v,
             Err(_) => return (0, None),
         };
@@ -226,7 +236,7 @@ impl<T: Number<T>> Sum<T> {
                     .collect(),
                 start_time: Some(prev_start),
                 time: Some(t),
-                value: *value,
+                value: value.get_value(),
                 exemplars: vec![],
             });
         }
@@ -282,7 +292,7 @@ impl<T: Number<T>> PrecomputedSum<T> {
         s_data.temporality = Temporality::Delta;
         s_data.is_monotonic = self.monotonic;
 
-        let mut values = match self.value_map.values.lock() {
+        let mut values = match self.value_map.values.write() {
             Ok(v) => v,
             Err(_) => return (0, None),
         };
@@ -315,9 +325,9 @@ impl<T: Number<T>> PrecomputedSum<T> {
 
         let default = T::default();
         for (attrs, value) in values.drain() {
-            let delta = value - *reported.get(&attrs).unwrap_or(&default);
+            let delta = value.get_value() - *reported.get(&attrs).unwrap_or(&default);
             if delta != default {
-                new_reported.insert(attrs.clone(), value);
+                new_reported.insert(attrs.clone(), value.get_value());
             }
             s_data.data_points.push(DataPoint {
                 attributes: attrs
@@ -367,7 +377,7 @@ impl<T: Number<T>> PrecomputedSum<T> {
         s_data.temporality = Temporality::Cumulative;
         s_data.is_monotonic = self.monotonic;
 
-        let values = match self.value_map.values.lock() {
+        let values = match self.value_map.values.write() {
             Ok(v) => v,
             Err(_) => return (0, None),
         };
@@ -400,9 +410,9 @@ impl<T: Number<T>> PrecomputedSum<T> {
 
         let default = T::default();
         for (attrs, value) in values.iter() {
-            let delta = *value - *reported.get(attrs).unwrap_or(&default);
+            let delta = value.get_value() - *reported.get(attrs).unwrap_or(&default);
             if delta != default {
-                new_reported.insert(attrs.clone(), *value);
+                new_reported.insert(attrs.clone(), value.get_value());
             }
             s_data.data_points.push(DataPoint {
                 attributes: attrs
diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/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,
diff --git a/stress/Cargo.toml b/stress/Cargo.toml
@@ -4,9 +4,14 @@ 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"
+path = "src/metrics_counter.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
diff --git a/stress/src/metrics_counter.rs b/stress/src/metrics_counter.rs
@@ -3,7 +3,7 @@
     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
-    3M /sec
+    35 M /sec
 */
 
 use lazy_static::lazy_static;
@@ -16,7 +16,7 @@ use rand::{
     rngs::{self},
     Rng, SeedableRng,
 };
-use std::{borrow::Cow, cell::RefCell};
+use std::cell::RefCell;
 
 mod throughput;
 
@@ -28,10 +28,7 @@ lazy_static! {
         "value1", "value2", "value3", "value4", "value5", "value6", "value7", "value8", "value9",
         "value10"
     ];
-    static ref COUNTER: Counter<u64> = PROVIDER
-        .meter(<&str as Into<Cow<'static, str>>>::into("test"))
-        .u64_counter("hello")
-        .init();
+    static ref COUNTER: Counter<u64> = PROVIDER.meter("test").u64_counter("hello").init();
 }
 
 thread_local! {
diff --git a/stress/src/metrics_histogram.rs b/stress/src/metrics_histogram.rs
diff --git a/stress/src/metrics_overflow.rs b/stress/src/metrics_overflow.rs
