Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import 'package:dartastic_opentelemetry_api/dartastic_opentelemetry_api.dart';
5 :
6 : import '../data/metric_point.dart';
7 : import '../exemplar_filter.dart';
8 : import '../exemplar_reservoir.dart';
9 : import 'metric_storage.dart';
10 :
11 : /// HistogramStorage is used for storing and accumulating histogram data.
12 : class HistogramStorage<T extends num> extends HistogramStorageBase<T>
13 : with ExemplarSampling<T> {
14 : /// Map of attribute sets to histogram data.
15 : final Map<Attributes, _HistogramPointData<T>> _points = {};
16 :
17 : /// The bucket boundaries for this histogram.
18 : final List<double> boundaries;
19 :
20 : /// Whether to record min and max values.
21 : final bool recordMinMax;
22 :
23 : /// The start time for all points.
24 : final DateTime _startTime = DateTime.now();
25 :
26 : /// The exemplar filter used by this storage.
27 : @override
28 : final ExemplarFilter exemplarFilter;
29 :
30 : /// Creates a new HistogramStorage instance.
31 9 : HistogramStorage(
32 : {required this.boundaries,
33 : this.recordMinMax = true,
34 : ExemplarFilter? exemplarFilter})
35 : : exemplarFilter = exemplarFilter ?? const TraceBasedExemplarFilter();
36 :
37 : /// Records a measurement with the given attributes and context.
38 6 : @override
39 : void record(T value,
40 : [Attributes? attributes, Context? context, DateTime? timestamp]) {
41 : // Create a normalized key for lookup
42 5 : final key = attributes ?? _emptyAttributes();
43 :
44 : _HistogramPointData<T> pointData;
45 : // Find matching attributes
46 6 : final existingKey = _findMatchingKey(key);
47 12 : var bucketIndex = boundaries.length;
48 : if (existingKey != null) {
49 : // Update existing point
50 10 : pointData = _points[existingKey]!;
51 5 : bucketIndex = pointData.record(value);
52 : } else {
53 : // Create new point
54 6 : pointData = _HistogramPointData<T>(
55 6 : boundaries: boundaries,
56 6 : recordMinMax: recordMinMax,
57 18 : reservoir: boundaries.length > 1
58 12 : ? AlignedHistogramBucketExemplarReservoir(boundaries)
59 0 : : SimpleFixedSizeExemplarReservoir(1),
60 : );
61 6 : bucketIndex = pointData.record(value);
62 12 : _points[key] = pointData;
63 : }
64 :
65 14 : maybeOffer(pointData.reservoir, value, key, context ?? Context.current,
66 6 : timestamp ?? DateTime.now(), bucketIndex);
67 : }
68 :
69 : /// Helper to get empty attributes safely
70 5 : Attributes _emptyAttributes() {
71 : // If OTelFactory is not initialized yet, create an empty attributes directly
72 5 : if (OTelFactory.otelFactory == null) {
73 0 : return OTelAPI.attributes(); // Use the API's static method instead
74 : }
75 10 : return OTelFactory.otelFactory!.attributes();
76 : }
77 :
78 : /// Finds a key in the points map that equals the given key
79 6 : Attributes? _findMatchingKey(Attributes key) {
80 18 : for (final existingKey in _points.keys) {
81 6 : if (existingKey == key) {
82 : // Using the == operator which should call equals
83 : return existingKey;
84 : }
85 : }
86 : return null;
87 : }
88 :
89 : /// Gets the current histogram value for the given attributes.
90 : /// If no attributes are provided, returns a combined HistogramValue across all attribute sets.
91 2 : @override
92 : HistogramValue getValue([Attributes? attributes]) {
93 : if (attributes == null) {
94 : // Combine across all attribute sets
95 3 : final totalSum = _points.values.fold<num>(
96 : 0,
97 3 : (sum, data) => sum + data.sum,
98 : );
99 3 : final totalCount = _points.values.fold<int>(
100 : 0,
101 3 : (count, data) => count + data.count,
102 : );
103 :
104 : // Combine bucket counts
105 1 : final combinedCounts = List<int>.filled(
106 3 : boundaries.length + 1,
107 : 0,
108 : );
109 3 : for (final data in _points.values) {
110 4 : for (var i = 0; i < data.counts.length; i++) {
111 4 : combinedCounts[i] += data.counts[i];
112 : }
113 : }
114 :
115 : // Find overall min and max
116 : num? overallMin;
117 : num? overallMax;
118 3 : if (recordMinMax && _points.isNotEmpty) {
119 2 : overallMin = _points.values
120 3 : .map((data) => data.min)
121 3 : .where((min) => min != double.infinity)
122 1 : .isEmpty
123 : ? null
124 2 : : _points.values
125 3 : .map((data) => data.min)
126 3 : .where((min) => min != double.infinity)
127 3 : .reduce((a, b) => a < b ? a : b);
128 2 : overallMax = _points.values
129 3 : .map((data) => data.max)
130 3 : .where((max) => max != double.negativeInfinity)
131 1 : .isEmpty
132 : ? null
133 2 : : _points.values
134 3 : .map((data) => data.max)
135 3 : .where((max) => max != double.negativeInfinity)
136 3 : .reduce((a, b) => a > b ? a : b);
137 : }
138 :
139 1 : return HistogramValue(
140 : sum: totalSum,
141 : count: totalCount,
142 1 : boundaries: boundaries,
143 : bucketCounts: combinedCounts,
144 : min: overallMin,
145 : max: overallMax,
146 : );
147 : }
148 :
149 : // Find matching attributes
150 2 : final existingKey = _findMatchingKey(attributes);
151 : if (existingKey != null) {
152 4 : final data = _points[existingKey]!;
153 2 : return HistogramValue(
154 2 : sum: data.sum,
155 2 : count: data.count,
156 2 : boundaries: boundaries,
157 2 : bucketCounts: data.counts,
158 8 : min: recordMinMax && data.min != double.infinity ? data.min : null,
159 6 : max: recordMinMax && data.max != double.negativeInfinity
160 2 : ? data.max
161 : : null,
162 : );
163 : } else {
164 : // Return empty histogram
165 0 : return HistogramValue(
166 : sum: 0,
167 : count: 0,
168 0 : boundaries: boundaries,
169 0 : bucketCounts: List<int>.filled(boundaries.length + 1, 0),
170 : min: null,
171 : max: null,
172 : );
173 : }
174 : }
175 :
176 : /// Collects the current set of metric points.
177 5 : @override
178 : List<MetricPoint<HistogramValue>> collectPoints() {
179 5 : final now = DateTime.now();
180 :
181 20 : return _points.entries.map((entry) {
182 5 : final data = entry.value;
183 :
184 : // Create a HistogramValue directly
185 5 : final histogramValue = HistogramValue(
186 5 : sum: data.sum,
187 5 : count: data.count,
188 5 : boundaries: boundaries,
189 5 : bucketCounts: data.counts,
190 20 : min: recordMinMax && data.min != double.infinity ? data.min : null,
191 15 : max: recordMinMax && data.max != double.negativeInfinity
192 5 : ? data.max
193 : : null,
194 : );
195 :
196 : // Create a MetricPoint<HistogramValue> - no type casting needed!
197 5 : return MetricPoint<HistogramValue>(
198 5 : attributes: entry.key,
199 5 : startTime: _startTime,
200 : endTime: now,
201 : value: histogramValue,
202 15 : exemplars: data.reservoir.collectAndReset(entry.key),
203 : );
204 5 : }).toList();
205 : }
206 :
207 : /// Resets all points (for delta temporality).
208 2 : @override
209 : void reset() {
210 4 : _points.clear();
211 : }
212 : }
213 :
214 : /// Data for a single histogram point.
215 : class _HistogramPointData<T extends num> {
216 : /// The total count of measurements.
217 : int count = 0;
218 :
219 : /// The sum of all measurements.
220 : num sum = 0;
221 :
222 : /// The minimum value recorded.
223 : num min = double.infinity;
224 :
225 : /// The maximum value recorded.
226 : num max = double.negativeInfinity;
227 :
228 : /// The counts per bucket.
229 : late List<int> counts;
230 :
231 : /// The bucket boundaries.
232 : final List<double> boundaries;
233 :
234 : /// Whether to record min and max values.
235 : final bool recordMinMax;
236 :
237 : /// Reservoir for this point.
238 : final ExemplarReservoir reservoir;
239 :
240 6 : _HistogramPointData(
241 : {required this.boundaries,
242 : required this.recordMinMax,
243 : required this.reservoir}) {
244 : // Initialize count array with one more than boundaries
245 : // (for the +Inf bucket)
246 30 : counts = List<int>.filled(boundaries.length + 1, 0);
247 : }
248 :
249 : /// Records a measurement.
250 : /// Returns the bucket index where the measurement fell.
251 6 : int record(T value) {
252 12 : count++;
253 12 : sum += value;
254 :
255 6 : if (recordMinMax) {
256 : final num numValue = value;
257 18 : if (numValue < min) min = numValue;
258 18 : if (numValue > max) max = numValue;
259 : }
260 :
261 : // Find the right bucket
262 12 : var bucketIndex = boundaries.length; // Default to the +Inf bucket
263 24 : for (var i = 0; i < boundaries.length; i++) {
264 18 : if (value <= boundaries[i]) {
265 : bucketIndex = i;
266 : break;
267 : }
268 : }
269 :
270 : // Increment the bucket count
271 18 : counts[bucketIndex]++;
272 : return bucketIndex;
273 : }
274 : }
|
