Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import 'dart:math';
5 :
6 : import 'package:dartastic_opentelemetry_api/dartastic_opentelemetry_api.dart';
7 :
8 : import 'data/exemplar.dart';
9 :
10 : /// An ExemplarReservoir receives measurements from instruments and samples them
11 : /// to provide Exemplars.
12 : ///
13 : /// Note for future extension (Issues #151, #153): The reservoir choice should be
14 : /// user-configurable via a View. Currently, storage constructors hardcode the reservoir
15 : /// types. Once View parameters are fully wired up, this should be driven by the resolved View.
16 : abstract class ExemplarReservoir {
17 : /// Offers a measurement to the reservoir.
18 : ///
19 : /// The reservoir decides whether to retain the measurement as an exemplar.
20 : ///
21 : /// @param value The value of the measurement.
22 : /// @param attributes The attributes associated with the measurement.
23 : /// @param context The Context associated with the measurement.
24 : /// @param timestamp The time the measurement was recorded.
25 : /// @param bucketIndex Optional pre-computed histogram bucket index.
26 : void offerMeasurement(
27 : num value, Attributes attributes, Context context, DateTime timestamp,
28 : [int? bucketIndex]);
29 :
30 : /// Returns the currently sampled exemplars and clears the reservoir.
31 : ///
32 : /// @param pointAttributes The attributes associated with the metric point.
33 : /// @return A list of the sampled exemplars.
34 : List<Exemplar> collectAndReset(Attributes pointAttributes);
35 : }
36 :
37 : /// A reservoir that retains a fixed number of exemplars.
38 : ///
39 : /// If more measurements are offered than the fixed size, this implementation
40 : /// uses a simple random sampling algorithm to decide which exemplars to replace.
41 : class SimpleFixedSizeExemplarReservoir implements ExemplarReservoir {
42 : final int _size;
43 : final Random _random;
44 : int _measurementsSeen = 0;
45 : final List<_MeasurementData?> _storage;
46 :
47 30 : SimpleFixedSizeExemplarReservoir(this._size, {Random? random})
48 29 : : _random = random ?? Random(),
49 30 : _storage = List.filled(_size, null);
50 :
51 3 : @override
52 : void offerMeasurement(
53 : num value, Attributes attributes, Context context, DateTime timestamp,
54 : [int? bucketIndex]) {
55 : int bucket;
56 9 : if (_measurementsSeen < _size) {
57 3 : bucket = _measurementsSeen;
58 : } else {
59 8 : bucket = _random.nextInt(_measurementsSeen + 1);
60 : }
61 :
62 6 : if (bucket < _size) {
63 9 : _storage[bucket] = _MeasurementData(
64 : value: value,
65 : attributes: attributes,
66 5 : traceId: context.spanContext?.traceId,
67 5 : spanId: context.spanContext?.spanId,
68 : timestamp: timestamp,
69 : );
70 : }
71 6 : _measurementsSeen++;
72 : }
73 :
74 23 : @override
75 : List<Exemplar> collectAndReset(Attributes pointAttributes) {
76 23 : final exemplars = <Exemplar>[];
77 92 : for (var i = 0; i < _storage.length; i++) {
78 46 : final data = _storage[i];
79 : if (data != null) {
80 3 : exemplars.add(
81 3 : Exemplar.fromMeasurement(
82 9 : measurement: OTelAPI.createMeasurement(data.value, data.attributes),
83 3 : timestamp: data.timestamp,
84 : aggregationAttributes: pointAttributes,
85 3 : spanId: data.spanId,
86 3 : traceId: data.traceId,
87 : ),
88 : );
89 6 : _storage[i] = null;
90 : }
91 : }
92 23 : _measurementsSeen = 0;
93 : return exemplars;
94 : }
95 : }
96 :
97 : /// A reservoir that is aligned to histogram buckets.
98 : ///
99 : /// It MUST store at most one measurement that falls within a histogram bucket,
100 : /// and SHOULD use a uniformly-weighted sampling algorithm based on the number
101 : /// of measurements the bucket has seen so far.
102 : class AlignedHistogramBucketExemplarReservoir implements ExemplarReservoir {
103 : final List<double> _boundaries;
104 : final List<_MeasurementData?> _storage;
105 : final List<int> _counts;
106 : final Random _random;
107 :
108 7 : AlignedHistogramBucketExemplarReservoir(this._boundaries, {Random? random})
109 21 : : _storage = List.filled(_boundaries.length + 1, null),
110 21 : _counts = List.filled(_boundaries.length + 1, 0),
111 7 : _random = random ?? Random();
112 :
113 1 : @override
114 : void offerMeasurement(
115 : num value, Attributes attributes, Context context, DateTime timestamp,
116 : [int? bucketIndex]) {
117 : int index;
118 : if (bucketIndex != null) {
119 : index = bucketIndex;
120 : } else {
121 2 : index = _boundaries.length;
122 4 : for (var i = 0; i < _boundaries.length; i++) {
123 3 : if (value <= _boundaries[i]) {
124 : index = i;
125 : break;
126 : }
127 : }
128 : }
129 :
130 2 : final measurementsSeenBucket = _counts[index];
131 1 : if (measurementsSeenBucket == 0 ||
132 4 : _random.nextInt(measurementsSeenBucket + 1) == 0) {
133 3 : _storage[index] = _MeasurementData(
134 : value: value,
135 : attributes: attributes,
136 1 : traceId: context.spanContext?.traceId,
137 1 : spanId: context.spanContext?.spanId,
138 : timestamp: timestamp,
139 : );
140 : }
141 3 : _counts[index]++;
142 : }
143 :
144 6 : @override
145 : List<Exemplar> collectAndReset(Attributes pointAttributes) {
146 6 : final exemplars = <Exemplar>[];
147 24 : for (var i = 0; i < _storage.length; i++) {
148 12 : final data = _storage[i];
149 : if (data != null) {
150 1 : exemplars.add(
151 1 : Exemplar.fromMeasurement(
152 3 : measurement: OTelAPI.createMeasurement(data.value, data.attributes),
153 1 : timestamp: data.timestamp,
154 : aggregationAttributes: pointAttributes,
155 1 : spanId: data.spanId,
156 1 : traceId: data.traceId,
157 : ),
158 : );
159 2 : _storage[i] = null;
160 : }
161 12 : _counts[i] = 0;
162 : }
163 : return exemplars;
164 : }
165 : }
166 :
167 : class _MeasurementData {
168 : final num value;
169 : final Attributes attributes;
170 : final TraceId? traceId;
171 : final SpanId? spanId;
172 : final DateTime timestamp;
173 :
174 3 : _MeasurementData({
175 : required this.value,
176 : required this.attributes,
177 : this.traceId,
178 : this.spanId,
179 : required this.timestamp,
180 : });
181 : }
|
