Skip to content

Commit 825d181

Browse files
frailltMindaugas Vinkelis
fraillt
authored and
Mindaugas Vinkelis
committed
ValueMap interface change
1 parent 7ab5e0f commit 825d181

File tree

5 files changed

+231
-214
lines changed

5 files changed

+231
-214
lines changed

‎opentelemetry-sdk/src/metrics/internal/histogram.rs‎

Lines changed: 85 additions & 100 deletions
Original file line numberDiff line numberDiff line change
@@ -7,45 +7,16 @@ use crate::metrics::data::HistogramDataPoint;
77
use crate::metrics::data::{self, Aggregation, Temporality};
88
use opentelemetry::KeyValue;
99

10-
use super::Number;
11-
use super::{AtomicTracker, AtomicallyUpdate, Operation, ValueMap};
10+
use super::ValueMap;
11+
use super::{Aggregator, Number};
1212

13-
struct HistogramUpdate;
14-
15-
impl Operation for HistogramUpdate {
16-
fn update_tracker<T: Default, AT: AtomicTracker<T>>(tracker: &AT, value: T, index: usize) {
17-
tracker.update_histogram(index, value);
18-
}
19-
}
20-
21-
struct HistogramTracker<T> {
22-
buckets: Mutex<Buckets<T>>,
23-
}
24-
25-
impl<T: Number> AtomicTracker<T> for HistogramTracker<T> {
26-
fn update_histogram(&self, index: usize, value: T) {
27-
let mut buckets = match self.buckets.lock() {
28-
Ok(guard) => guard,
29-
Err(_) => return,
30-
};
31-
32-
buckets.bin(index, value);
33-
buckets.sum(value);
34-
}
35-
}
36-
37-
impl<T: Number> AtomicallyUpdate<T> for HistogramTracker<T> {
38-
type AtomicTracker = HistogramTracker<T>;
39-
40-
fn new_atomic_tracker(buckets_count: Option<usize>) -> Self::AtomicTracker {
41-
let count = buckets_count.unwrap();
42-
HistogramTracker {
43-
buckets: Mutex::new(Buckets::<T>::new(count)),
44-
}
45-
}
13+
struct BucketsConfig {
14+
bounds: Vec<f64>,
15+
record_min_max: bool,
16+
record_sum: bool,
4617
}
4718

48-
#[derive(Default)]
19+
#[derive(Default, Debug, Clone)]
4920
struct Buckets<T> {
5021
counts: Vec<u64>,
5122
count: u64,
@@ -54,30 +25,37 @@ struct Buckets<T> {
5425
max: T,
5526
}
5627

57-
impl<T: Number> Buckets<T> {
58-
/// returns buckets with `n` bins.
59-
fn new(n: usize) -> Buckets<T> {
60-
Buckets {
61-
counts: vec![0; n],
28+
impl<T> Buckets<T>
29+
where
30+
T: Number,
31+
{
32+
fn new(size: usize) -> Self {
33+
Self {
34+
counts: vec![0; size],
6235
min: T::max(),
6336
max: T::min(),
6437
..Default::default()
6538
}
6639
}
67-
68-
fn sum(&mut self, value: T) {
69-
self.total += value;
70-
}
71-
72-
fn bin(&mut self, idx: usize, value: T) {
40+
fn update(&mut self, config: &BucketsConfig, f_value: f64, measurement: T) {
41+
// This search will return an index in the range `[0, bounds.len()]`, where
42+
// it will return `bounds.len()` if value is greater than the last element
43+
// of `bounds`. This aligns with the buckets in that the length of buckets
44+
// is `bounds.len()+1`, with the last bucket representing:
45+
// `(bounds[bounds.len()-1], +∞)`.
46+
let idx = config.bounds.partition_point(|&x| x < f_value);
7347
self.counts[idx] += 1;
7448
self.count += 1;
75-
if value < self.min {
76-
self.min = value;
77-
}
78-
if value > self.max {
79-
self.max = value
49+
if config.record_min_max {
50+
if measurement < self.min {
51+
self.min = measurement;
52+
}
53+
if measurement > self.max {
54+
self.max = measurement
55+
}
8056
}
57+
// it's very cheap to update it, even if it is not configured to record_sum
58+
self.total += measurement;
8159
}
8260

8361
fn reset(&mut self) {
@@ -91,45 +69,52 @@ impl<T: Number> Buckets<T> {
9169
}
9270
}
9371

72+
impl<T> Aggregator<T> for Mutex<Buckets<T>>
73+
where
74+
T: Number,
75+
{
76+
type Config = BucketsConfig;
77+
78+
fn create(config: &BucketsConfig) -> Self {
79+
let size = config.bounds.len() + 1;
80+
Mutex::new(Buckets::new(size))
81+
}
82+
83+
fn update(&self, config: &BucketsConfig, measurement: T) {
84+
let f_value = measurement.into_float();
85+
// Ignore NaN and infinity.
86+
if f_value.is_infinite() || f_value.is_nan() {
87+
return;
88+
}
89+
if let Ok(mut this) = self.lock() {
90+
this.update(config, f_value, measurement);
91+
}
92+
}
93+
}
94+
9495
/// Summarizes a set of measurements as a histogram with explicitly defined
9596
/// buckets.
9697
pub(crate) struct Histogram<T: Number> {
97-
value_map: ValueMap<HistogramTracker<T>, T, HistogramUpdate>,
98-
bounds: Vec<f64>,
99-
record_min_max: bool,
100-
record_sum: bool,
98+
value_map: ValueMap<T, Mutex<Buckets<T>>>,
10199
start: Mutex<SystemTime>,
102100
}
103101

104102
impl<T: Number> Histogram<T> {
105-
pub(crate) fn new(boundaries: Vec<f64>, record_min_max: bool, record_sum: bool) -> Self {
106-
let buckets_count = boundaries.len() + 1;
107-
let mut histogram = Histogram {
108-
value_map: ValueMap::new_with_buckets_count(buckets_count),
109-
bounds: boundaries,
110-
record_min_max,
111-
record_sum,
103+
pub(crate) fn new(mut bounds: Vec<f64>, record_min_max: bool, record_sum: bool) -> Self {
104+
bounds.retain(|v| !v.is_nan());
105+
bounds.sort_by(|a, b| a.partial_cmp(b).expect("NaNs filtered out"));
106+
Self {
107+
value_map: ValueMap::new(BucketsConfig {
108+
record_min_max,
109+
record_sum,
110+
bounds,
111+
}),
112112
start: Mutex::new(SystemTime::now()),
113-
};
114-
115-
histogram.bounds.retain(|v| !v.is_nan());
116-
histogram
117-
.bounds
118-
.sort_by(|a, b| a.partial_cmp(b).expect("NaNs filtered out"));
119-
120-
histogram
113+
}
121114
}
122115

123116
pub(crate) fn measure(&self, measurement: T, attrs: &[KeyValue]) {
124-
let f = measurement.into_float();
125-
126-
// This search will return an index in the range `[0, bounds.len()]`, where
127-
// it will return `bounds.len()` if value is greater than the last element
128-
// of `bounds`. This aligns with the buckets in that the length of buckets
129-
// is `bounds.len()+1`, with the last bucket representing:
130-
// `(bounds[bounds.len()-1], +∞)`.
131-
let index = self.bounds.partition_point(|&x| x < f);
132-
self.value_map.measure(measurement, attrs, index);
117+
self.value_map.measure(measurement, attrs);
133118
}
134119

135120
pub(crate) fn delta(
@@ -167,25 +152,25 @@ impl<T: Number> Histogram<T> {
167152
.has_no_attribute_value
168153
.swap(false, Ordering::AcqRel)
169154
{
170-
if let Ok(ref mut b) = self.value_map.no_attribute_tracker.buckets.lock() {
155+
if let Ok(ref mut b) = self.value_map.no_attribute_tracker.lock() {
171156
h.data_points.push(HistogramDataPoint {
172157
attributes: vec![],
173158
start_time: start,
174159
time: t,
175160
count: b.count,
176-
bounds: self.bounds.clone(),
161+
bounds: self.value_map.config.bounds.clone(),
177162
bucket_counts: b.counts.clone(),
178-
sum: if self.record_sum {
163+
sum: if self.value_map.config.record_sum {
179164
b.total
180165
} else {
181166
T::default()
182167
},
183-
min: if self.record_min_max {
168+
min: if self.value_map.config.record_min_max {
184169
Some(b.min)
185170
} else {
186171
None
187172
},
188-
max: if self.record_min_max {
173+
max: if self.value_map.config.record_min_max {
189174
Some(b.max)
190175
} else {
191176
None
@@ -205,25 +190,25 @@ impl<T: Number> Histogram<T> {
205190
let mut seen = HashSet::new();
206191
for (attrs, tracker) in trackers.drain() {
207192
if seen.insert(Arc::as_ptr(&tracker)) {
208-
if let Ok(b) = tracker.buckets.lock() {
193+
if let Ok(b) = tracker.lock() {
209194
h.data_points.push(HistogramDataPoint {
210195
attributes: attrs.clone(),
211196
start_time: start,
212197
time: t,
213198
count: b.count,
214-
bounds: self.bounds.clone(),
199+
bounds: self.value_map.config.bounds.clone(),
215200
bucket_counts: b.counts.clone(),
216-
sum: if self.record_sum {
201+
sum: if self.value_map.config.record_sum {
217202
b.total
218203
} else {
219204
T::default()
220205
},
221-
min: if self.record_min_max {
206+
min: if self.value_map.config.record_min_max {
222207
Some(b.min)
223208
} else {
224209
None
225210
},
226-
max: if self.record_min_max {
211+
max: if self.value_map.config.record_min_max {
227212
Some(b.max)
228213
} else {
229214
None
@@ -278,25 +263,25 @@ impl<T: Number> Histogram<T> {
278263
.has_no_attribute_value
279264
.load(Ordering::Acquire)
280265
{
281-
if let Ok(b) = &self.value_map.no_attribute_tracker.buckets.lock() {
266+
if let Ok(b) = &self.value_map.no_attribute_tracker.lock() {
282267
h.data_points.push(HistogramDataPoint {
283268
attributes: vec![],
284269
start_time: start,
285270
time: t,
286271
count: b.count,
287-
bounds: self.bounds.clone(),
272+
bounds: self.value_map.config.bounds.clone(),
288273
bucket_counts: b.counts.clone(),
289-
sum: if self.record_sum {
274+
sum: if self.value_map.config.record_sum {
290275
b.total
291276
} else {
292277
T::default()
293278
},
294-
min: if self.record_min_max {
279+
min: if self.value_map.config.record_min_max {
295280
Some(b.min)
296281
} else {
297282
None
298283
},
299-
max: if self.record_min_max {
284+
max: if self.value_map.config.record_min_max {
300285
Some(b.max)
301286
} else {
302287
None
@@ -318,25 +303,25 @@ impl<T: Number> Histogram<T> {
318303
let mut seen = HashSet::new();
319304
for (attrs, tracker) in trackers.iter() {
320305
if seen.insert(Arc::as_ptr(tracker)) {
321-
if let Ok(b) = tracker.buckets.lock() {
306+
if let Ok(b) = tracker.lock() {
322307
h.data_points.push(HistogramDataPoint {
323308
attributes: attrs.clone(),
324309
start_time: start,
325310
time: t,
326311
count: b.count,
327-
bounds: self.bounds.clone(),
312+
bounds: self.value_map.config.bounds.clone(),
328313
bucket_counts: b.counts.clone(),
329-
sum: if self.record_sum {
314+
sum: if self.value_map.config.record_sum {
330315
b.total
331316
} else {
332317
T::default()
333318
},
334-
min: if self.record_min_max {
319+
min: if self.value_map.config.record_min_max {
335320
Some(b.min)
336321
} else {
337322
None
338323
},
339-
max: if self.record_min_max {
324+
max: if self.value_map.config.record_min_max {
340325
Some(b.max)
341326
} else {
342327
None

0 commit comments

Comments
 (0)