Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ValueMap interface change #2117

Merged
merged 5 commits into from
Nov 1, 2024
Merged

ValueMap interface change #2117

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
606d126
ValueMap interface change
fraillt Sep 14, 2024
5ff18ca
Merge branch 'main' into value-map-interface-change
lalitb Oct 15, 2024
1f54f90
Merge branch 'main' into value-map-interface-change
cijothomas Oct 23, 2024
ffe0d4a
Merge branch 'main' into value-map-interface-change
lalitb Oct 28, 2024
133f317
Merge branch 'main' into value-map-interface-change
cijothomas Nov 1, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions opentelemetry-sdk/src/metrics/internal/exponential_histogram.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,6 +352,7 @@ impl<T: Number> ExpoHistogram<T> {
pub(crate) fn measure(&self, value: T, attrs: &[KeyValue]) {
let f_value = value.into_float();
// Ignore NaN and infinity.
// Only makes sense if T is f64, maybe this could be no-op for other cases?
if f_value.is_infinite() || f_value.is_nan() {
return;
}
Expand Down
114 changes: 91 additions & 23 deletions opentelemetry-sdk/src/metrics/internal/histogram.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,23 +7,22 @@ use crate::metrics::data::HistogramDataPoint;
use crate::metrics::data::{self, Aggregation, Temporality};
use opentelemetry::KeyValue;

use super::Number;
use super::{AtomicTracker, AtomicallyUpdate, Operation, ValueMap};

struct HistogramUpdate;

impl Operation for HistogramUpdate {
fn update_tracker<T: Default, AT: AtomicTracker<T>>(tracker: &AT, value: T, index: usize) {
tracker.update_histogram(index, value);
}
}
use super::ValueMap;
use super::{Aggregator, Number};

struct HistogramTracker<T> {
buckets: Mutex<Buckets<T>>,
}

impl<T: Number> AtomicTracker<T> for HistogramTracker<T> {
fn update_histogram(&self, index: usize, value: T) {
impl<T> Aggregator<T> for HistogramTracker<T>
where
T: Number,
{
type InitConfig = usize;
/// Value and bucket index
type PreComputedValue = (T, usize);

fn update(&self, (value, index): (T, usize)) {
let mut buckets = match self.buckets.lock() {
Ok(guard) => guard,
Err(_) => return,
Expand All @@ -32,15 +31,10 @@ impl<T: Number> AtomicTracker<T> for HistogramTracker<T> {
buckets.bin(index, value);
buckets.sum(value);
}
}

impl<T: Number> AtomicallyUpdate<T> for HistogramTracker<T> {
type AtomicTracker = HistogramTracker<T>;

fn new_atomic_tracker(buckets_count: Option<usize>) -> Self::AtomicTracker {
let count = buckets_count.unwrap();
fn create(count: &usize) -> Self {
HistogramTracker {
buckets: Mutex::new(Buckets::<T>::new(count)),
buckets: Mutex::new(Buckets::<T>::new(*count)),
fraillt marked this conversation as resolved.
Show resolved Hide resolved
}
}
}
Expand Down Expand Up @@ -94,7 +88,7 @@ impl<T: Number> Buckets<T> {
/// Summarizes a set of measurements as a histogram with explicitly defined
/// buckets.
pub(crate) struct Histogram<T: Number> {
value_map: ValueMap<HistogramTracker<T>, T, HistogramUpdate>,
value_map: ValueMap<T, HistogramTracker<T>>,
bounds: Vec<f64>,
record_min_max: bool,
record_sum: bool,
Expand All @@ -103,9 +97,11 @@ pub(crate) struct Histogram<T: Number> {

impl<T: Number> Histogram<T> {
pub(crate) fn new(boundaries: Vec<f64>, record_min_max: bool, record_sum: bool) -> Self {
// TODO fix the bug, by first removing NaN and only then getting buckets_count
// once we know the reason for performance degradation
let buckets_count = boundaries.len() + 1;
let mut histogram = Histogram {
value_map: ValueMap::new_with_buckets_count(buckets_count),
value_map: ValueMap::new(buckets_count),
bounds: boundaries,
record_min_max,
record_sum,
Expand All @@ -122,14 +118,20 @@ impl<T: Number> Histogram<T> {

pub(crate) fn measure(&self, measurement: T, attrs: &[KeyValue]) {
let f = measurement.into_float();

// Ignore NaN and infinity.
// Only makes sense if T is f64, maybe this could be no-op for other cases?
// TODO: uncomment once we know the reason for performance degradation
// if f.is_infinite() || f.is_nan() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@cijothomas Let's add this check as well?

// return;
// }
// This search will return an index in the range `[0, bounds.len()]`, where
// it will return `bounds.len()` if value is greater than the last element
// of `bounds`. This aligns with the buckets in that the length of buckets
// is `bounds.len()+1`, with the last bucket representing:
// `(bounds[bounds.len()-1], +∞)`.
let index = self.bounds.partition_point(|&x| x < f);
self.value_map.measure(measurement, attrs, index);

self.value_map.measure((measurement, index), attrs);
}

pub(crate) fn delta(
Expand Down Expand Up @@ -350,3 +352,69 @@ impl<T: Number> Histogram<T> {
(h.data_points.len(), new_agg.map(|a| Box::new(a) as Box<_>))
}
}

// TODO: uncomment once we know the reason for performance degradation
// #[cfg(test)]
// mod tests {

// use super::*;

// #[test]
// fn when_f64_is_nan_or_infinity_then_ignore() {
// struct Expected {
// min: f64,
// max: f64,
// sum: f64,
// count: u64,
// }
// impl Expected {
// fn new(min: f64, max: f64, sum: f64, count: u64) -> Self {
// Expected {
// min,
// max,
// sum,
// count,
// }
// }
// }
// struct TestCase {
// values: Vec<f64>,
// expected: Expected,
// }

// let test_cases = vec![
// TestCase {
// values: vec![2.0, 4.0, 1.0],
// expected: Expected::new(1.0, 4.0, 7.0, 3),
// },
// TestCase {
// values: vec![2.0, 4.0, 1.0, f64::INFINITY],
// expected: Expected::new(1.0, 4.0, 7.0, 3),
// },
// TestCase {
// values: vec![2.0, 4.0, 1.0, -f64::INFINITY],
// expected: Expected::new(1.0, 4.0, 7.0, 3),
// },
// TestCase {
// values: vec![2.0, f64::NAN, 4.0, 1.0],
// expected: Expected::new(1.0, 4.0, 7.0, 3),
// },
// TestCase {
// values: vec![4.0, 4.0, 4.0, 2.0, 16.0, 1.0],
// expected: Expected::new(1.0, 16.0, 31.0, 6),
// },
// ];

// for test in test_cases {
// let h = Histogram::new(vec![], true, true);
// for v in test.values {
// h.measure(v, &[]);
// }
// let res = h.value_map.no_attribute_tracker.buckets.lock().unwrap();
// assert_eq!(test.expected.max, res.max);
// assert_eq!(test.expected.min, res.min);
// assert_eq!(test.expected.sum, res.total);
// assert_eq!(test.expected.count, res.count);
// }
// }
// }
46 changes: 38 additions & 8 deletions opentelemetry-sdk/src/metrics/internal/last_value.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,25 +7,51 @@ use std::{
use crate::metrics::data::DataPoint;
use opentelemetry::KeyValue;

use super::{Assign, AtomicTracker, Number, ValueMap};
use super::{Aggregator, AtomicTracker, AtomicallyUpdate, Number, ValueMap};

/// this is reused by PrecomputedSum
pub(crate) struct Assign<T>
where
T: AtomicallyUpdate<T>,
{
pub(crate) value: T::AtomicTracker,
}

impl<T> Aggregator<T> for Assign<T>
where
T: Number,
{
type InitConfig = ();
type PreComputedValue = T;

fn create(_init: &()) -> Self {
Self {
value: T::new_atomic_tracker(T::default()),
}
}

fn update(&self, value: T) {
self.value.store(value)
}
}

/// Summarizes a set of measurements as the last one made.
pub(crate) struct LastValue<T: Number> {
value_map: ValueMap<T, T, Assign>,
value_map: ValueMap<T, Assign<T>>,
start: Mutex<SystemTime>,
}

impl<T: Number> LastValue<T> {
pub(crate) fn new() -> Self {
LastValue {
value_map: ValueMap::new(),
value_map: ValueMap::new(()),
start: Mutex::new(SystemTime::now()),
}
}

pub(crate) fn measure(&self, measurement: T, attrs: &[KeyValue]) {
// The argument index is not applicable to LastValue.
self.value_map.measure(measurement, attrs, 0);
self.value_map.measure(measurement, attrs);
}

pub(crate) fn compute_aggregation_delta(&self, dest: &mut Vec<DataPoint<T>>) {
Expand All @@ -49,7 +75,11 @@ impl<T: Number> LastValue<T> {
attributes: vec![],
start_time: Some(prev_start),
time: Some(t),
value: self.value_map.no_attribute_tracker.get_and_reset_value(),
value: self
.value_map
.no_attribute_tracker
.value
.get_and_reset_value(),
exemplars: vec![],
});
}
Expand All @@ -66,7 +96,7 @@ impl<T: Number> LastValue<T> {
attributes: attrs.clone(),
start_time: Some(prev_start),
time: Some(t),
value: tracker.get_value(),
value: tracker.value.get_value(),
exemplars: vec![],
});
}
Expand Down Expand Up @@ -101,7 +131,7 @@ impl<T: Number> LastValue<T> {
attributes: vec![],
start_time: Some(prev_start),
time: Some(t),
value: self.value_map.no_attribute_tracker.get_value(),
value: self.value_map.no_attribute_tracker.value.get_value(),
exemplars: vec![],
});
}
Expand All @@ -118,7 +148,7 @@ impl<T: Number> LastValue<T> {
attributes: attrs.clone(),
start_time: Some(prev_start),
time: Some(t),
value: tracker.get_value(),
value: tracker.value.get_value(),
exemplars: vec![],
});
}
Expand Down
Loading
Loading