@@ -7,79 +7,41 @@ use crate::metrics::data::HistogramDataPoint;
7
7
use crate :: metrics:: data:: { self , Aggregation , Temporality } ;
8
8
use opentelemetry:: KeyValue ;
9
9
10
- use super :: Number ;
11
- use super :: { AtomicTracker , AtomicallyUpdate , Operation , ValueMap } ;
10
+ use super :: ValueMap ;
11
+ use super :: { Aggregator , Number } ;
12
12
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 ,
46
17
}
47
18
48
- #[ derive( Default ) ]
49
- struct Buckets < T > {
19
+ #[ derive( Default , Debug , Clone ) ]
20
+ struct BucketsData < T > {
50
21
counts : Vec < u64 > ,
51
22
count : u64 ,
52
23
total : T ,
53
24
min : T ,
54
25
max : T ,
55
26
}
56
27
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
+ struct Buckets < T > {
29
+ data : Mutex < BucketsData < T > > ,
30
+ }
31
+
32
+ impl < T > BucketsData < T >
33
+ where
34
+ T : Number ,
35
+ {
36
+ fn new ( size : usize ) -> Self {
37
+ Self {
38
+ counts : vec ! [ 0 ; size] ,
62
39
min : T :: max ( ) ,
63
40
max : T :: min ( ) ,
64
41
..Default :: default ( )
65
42
}
66
43
}
67
44
68
- fn sum ( & mut self , value : T ) {
69
- self . total += value;
70
- }
71
-
72
- fn bin ( & mut self , idx : usize , value : T ) {
73
- self . counts [ idx] += 1 ;
74
- self . count += 1 ;
75
- if value < self . min {
76
- self . min = value;
77
- }
78
- if value > self . max {
79
- self . max = value
80
- }
81
- }
82
-
83
45
fn reset ( & mut self ) {
84
46
for item in & mut self . counts {
85
47
* item = 0 ;
@@ -91,45 +53,71 @@ impl<T: Number> Buckets<T> {
91
53
}
92
54
}
93
55
56
+ impl < T > Aggregator < T > for Buckets < T >
57
+ where
58
+ T : Number ,
59
+ {
60
+ type Config = BucketsConfig ;
61
+
62
+ fn create ( config : & BucketsConfig ) -> Self {
63
+ let size = config. bounds . len ( ) + 1 ;
64
+ Buckets {
65
+ data : Mutex :: new ( BucketsData :: new ( size) ) ,
66
+ }
67
+ }
68
+
69
+ fn update ( & self , config : & BucketsConfig , measurement : T ) {
70
+ let f_value = measurement. into_float ( ) ;
71
+ // Ignore NaN and infinity.
72
+ if f_value. is_infinite ( ) || f_value. is_nan ( ) {
73
+ return ;
74
+ }
75
+ // This search will return an index in the range `[0, bounds.len()]`, where
76
+ // it will return `bounds.len()` if value is greater than the last element
77
+ // of `bounds`. This aligns with the buckets in that the length of buckets
78
+ // is `bounds.len()+1`, with the last bucket representing:
79
+ // `(bounds[bounds.len()-1], +∞)`.
80
+ let idx = config. bounds . partition_point ( |& x| x < f_value) ;
81
+ if let Ok ( mut data) = self . data . lock ( ) {
82
+ data. counts [ idx] += 1 ;
83
+ data. count += 1 ;
84
+ if config. record_min_max {
85
+ if measurement < data. min {
86
+ data. min = measurement;
87
+ }
88
+ if measurement > data. max {
89
+ data. max = measurement
90
+ }
91
+ }
92
+ // it's very cheap to update it, even if it is not configured to record_sum
93
+ data. total += measurement;
94
+ }
95
+ }
96
+ }
97
+
94
98
/// Summarizes a set of measurements as a histogram with explicitly defined
95
99
/// buckets.
96
100
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 ,
101
+ value_map : ValueMap < T , Buckets < T > > ,
101
102
start : Mutex < SystemTime > ,
102
103
}
103
104
104
105
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,
106
+ pub ( crate ) fn new ( mut bounds : Vec < f64 > , record_min_max : bool , record_sum : bool ) -> Self {
107
+ bounds. retain ( |v| !v. is_nan ( ) ) ;
108
+ bounds. sort_by ( |a, b| a. partial_cmp ( b) . expect ( "NaNs filtered out" ) ) ;
109
+ Self {
110
+ value_map : ValueMap :: new ( BucketsConfig {
111
+ record_min_max,
112
+ record_sum,
113
+ bounds,
114
+ } ) ,
112
115
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
116
+ }
121
117
}
122
118
123
119
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) ;
120
+ self . value_map . measure ( measurement, attrs) ;
133
121
}
134
122
135
123
pub ( crate ) fn delta (
@@ -167,25 +155,25 @@ impl<T: Number> Histogram<T> {
167
155
. has_no_attribute_value
168
156
. swap ( false , Ordering :: AcqRel )
169
157
{
170
- if let Ok ( ref mut b) = self . value_map . no_attribute_tracker . buckets . lock ( ) {
158
+ if let Ok ( ref mut b) = self . value_map . no_attribute_tracker . data . lock ( ) {
171
159
h. data_points . push ( HistogramDataPoint {
172
160
attributes : vec ! [ ] ,
173
161
start_time : start,
174
162
time : t,
175
163
count : b. count ,
176
- bounds : self . bounds . clone ( ) ,
164
+ bounds : self . value_map . config . bounds . clone ( ) ,
177
165
bucket_counts : b. counts . clone ( ) ,
178
- sum : if self . record_sum {
166
+ sum : if self . value_map . config . record_sum {
179
167
b. total
180
168
} else {
181
169
T :: default ( )
182
170
} ,
183
- min : if self . record_min_max {
171
+ min : if self . value_map . config . record_min_max {
184
172
Some ( b. min )
185
173
} else {
186
174
None
187
175
} ,
188
- max : if self . record_min_max {
176
+ max : if self . value_map . config . record_min_max {
189
177
Some ( b. max )
190
178
} else {
191
179
None
@@ -205,25 +193,25 @@ impl<T: Number> Histogram<T> {
205
193
let mut seen = HashSet :: new ( ) ;
206
194
for ( attrs, tracker) in trackers. drain ( ) {
207
195
if seen. insert ( Arc :: as_ptr ( & tracker) ) {
208
- if let Ok ( b) = tracker. buckets . lock ( ) {
196
+ if let Ok ( b) = tracker. data . lock ( ) {
209
197
h. data_points . push ( HistogramDataPoint {
210
198
attributes : attrs. clone ( ) ,
211
199
start_time : start,
212
200
time : t,
213
201
count : b. count ,
214
- bounds : self . bounds . clone ( ) ,
202
+ bounds : self . value_map . config . bounds . clone ( ) ,
215
203
bucket_counts : b. counts . clone ( ) ,
216
- sum : if self . record_sum {
204
+ sum : if self . value_map . config . record_sum {
217
205
b. total
218
206
} else {
219
207
T :: default ( )
220
208
} ,
221
- min : if self . record_min_max {
209
+ min : if self . value_map . config . record_min_max {
222
210
Some ( b. min )
223
211
} else {
224
212
None
225
213
} ,
226
- max : if self . record_min_max {
214
+ max : if self . value_map . config . record_min_max {
227
215
Some ( b. max )
228
216
} else {
229
217
None
@@ -278,25 +266,25 @@ impl<T: Number> Histogram<T> {
278
266
. has_no_attribute_value
279
267
. load ( Ordering :: Acquire )
280
268
{
281
- if let Ok ( b) = & self . value_map . no_attribute_tracker . buckets . lock ( ) {
269
+ if let Ok ( b) = & self . value_map . no_attribute_tracker . data . lock ( ) {
282
270
h. data_points . push ( HistogramDataPoint {
283
271
attributes : vec ! [ ] ,
284
272
start_time : start,
285
273
time : t,
286
274
count : b. count ,
287
- bounds : self . bounds . clone ( ) ,
275
+ bounds : self . value_map . config . bounds . clone ( ) ,
288
276
bucket_counts : b. counts . clone ( ) ,
289
- sum : if self . record_sum {
277
+ sum : if self . value_map . config . record_sum {
290
278
b. total
291
279
} else {
292
280
T :: default ( )
293
281
} ,
294
- min : if self . record_min_max {
282
+ min : if self . value_map . config . record_min_max {
295
283
Some ( b. min )
296
284
} else {
297
285
None
298
286
} ,
299
- max : if self . record_min_max {
287
+ max : if self . value_map . config . record_min_max {
300
288
Some ( b. max )
301
289
} else {
302
290
None
@@ -318,25 +306,25 @@ impl<T: Number> Histogram<T> {
318
306
let mut seen = HashSet :: new ( ) ;
319
307
for ( attrs, tracker) in trackers. iter ( ) {
320
308
if seen. insert ( Arc :: as_ptr ( tracker) ) {
321
- if let Ok ( b) = tracker. buckets . lock ( ) {
309
+ if let Ok ( b) = tracker. data . lock ( ) {
322
310
h. data_points . push ( HistogramDataPoint {
323
311
attributes : attrs. clone ( ) ,
324
312
start_time : start,
325
313
time : t,
326
314
count : b. count ,
327
- bounds : self . bounds . clone ( ) ,
315
+ bounds : self . value_map . config . bounds . clone ( ) ,
328
316
bucket_counts : b. counts . clone ( ) ,
329
- sum : if self . record_sum {
317
+ sum : if self . value_map . config . record_sum {
330
318
b. total
331
319
} else {
332
320
T :: default ( )
333
321
} ,
334
- min : if self . record_min_max {
322
+ min : if self . value_map . config . record_min_max {
335
323
Some ( b. min )
336
324
} else {
337
325
None
338
326
} ,
339
- max : if self . record_min_max {
327
+ max : if self . value_map . config . record_min_max {
340
328
Some ( b. max )
341
329
} else {
342
330
None
0 commit comments