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 : /// Storage implementation for sum-based metrics like Counter and UpDownCounter.
12 : ///
13 : /// SumStorage accumulates measurements for sum-based instruments. It maintains
14 : /// separate accumulated values for each unique set of attributes, and provides
15 : /// methods to collect the current state as metric points.
16 : ///
17 : /// This storage implementation supports both monotonic sums (like Counter)
18 : /// and non-monotonic sums (like UpDownCounter).
19 : ///
20 : /// More information:
21 : /// https://opentelemetry.io/docs/specs/otel/metrics/sdk/#the-temporality-of-instruments
22 : class SumStorage<T extends num> extends NumericStorage<T>
23 : with ExemplarSampling<T> {
24 : /// Map of attribute sets to accumulated values.
25 : final Map<Attributes?, _SumPointData<T>> _points = {};
26 :
27 : /// Whether the sum is monotonic (only increases).
28 : ///
29 : /// Monotonic sums only accept positive increments and are
30 : /// appropriate for counters that never decrease.
31 : final bool isMonotonic;
32 :
33 : /// The start time for all points.
34 : ///
35 : /// This is used for cumulative temporality reporting.
36 : final DateTime _startTime = DateTime.now();
37 :
38 : /// The exemplar filter used by this storage.
39 : @override
40 : final ExemplarFilter exemplarFilter;
41 :
42 : /// Creates a new SumStorage instance.
43 : ///
44 : /// @param isMonotonic Whether this storage is for a monotonic sum
45 : /// @param exemplarFilter Optional filter for exemplars
46 27 : SumStorage({required this.isMonotonic, ExemplarFilter? exemplarFilter})
47 : : exemplarFilter = exemplarFilter ?? const TraceBasedExemplarFilter();
48 :
49 : /// Records a measurement with the given attributes and context.
50 : ///
51 : /// For synchronous instruments, this is a delta that gets added to the existing value.
52 : /// For asynchronous instruments, this should be the absolute value.
53 : ///
54 : /// @param value The value to record
55 : /// @param attributes Optional attributes to associate with this measurement
56 : /// @param context Optional context associated with this measurement
57 25 : @override
58 : void record(T value,
59 : [Attributes? attributes, Context? context, DateTime? timestamp]) {
60 : // Check constraints for monotonic counters
61 47 : if (isMonotonic && value < 0) {
62 2 : print(
63 : 'Warning: Negative value $value provided to monotonic sum storage. '
64 : 'This will be ignored.',
65 : );
66 : return;
67 : }
68 :
69 : _SumPointData<T> pointData;
70 : // Check if we already have an entry for these attributes
71 50 : if (_points.containsKey(attributes)) {
72 : // Add to existing data point
73 20 : pointData = _points[attributes]!;
74 10 : pointData.add(value);
75 : } else {
76 : // Create new data point
77 25 : pointData = _SumPointData<T>(
78 : value: value,
79 25 : lastUpdateTime: DateTime.now(),
80 : reservoir:
81 25 : SimpleFixedSizeExemplarReservoir(1), // Simple fixed size of 1
82 : );
83 50 : _points[attributes] = pointData;
84 : }
85 :
86 25 : maybeOffer(
87 25 : pointData.reservoir,
88 : value,
89 36 : attributes ?? OTelFactory.otelFactory!.attributes(),
90 3 : context ?? Context.current,
91 25 : timestamp ?? DateTime.now());
92 : }
93 :
94 : /// Gets the current value for the given attributes.
95 : ///
96 : /// If no attributes are provided, returns the sum across all attribute sets.
97 : ///
98 : /// @param attributes Optional attributes to filter by
99 : /// @return The current accumulated value
100 10 : @override
101 : T getValue([Attributes? attributes]) {
102 : num result;
103 :
104 : if (attributes == null) {
105 : // Sum of all values across all attribute sets
106 48 : result = _points.values.fold<num>(0, (sum, data) => sum + data.value);
107 18 : } else if (_points.containsKey(attributes)) {
108 : // Return the value for the specific attributes
109 24 : result = _points[attributes]!.value;
110 : } else {
111 : // No entry for these attributes
112 : result = 0;
113 : }
114 :
115 : // Convert to the appropriate generic type
116 10 : if (T == int) {
117 8 : return result.toInt() as T;
118 6 : } else if (T == double) {
119 5 : return result.toDouble() as T;
120 : } else {
121 : return result as T;
122 : }
123 : }
124 :
125 : /// Collects the current set of metric points.
126 : ///
127 : /// This method is used by the instrument to collect all current
128 : /// sum values as metric points for export.
129 : ///
130 : /// @return A list of metric points containing the current values
131 18 : @override
132 : List<MetricPoint<T>> collectPoints() {
133 18 : final now = DateTime.now();
134 :
135 72 : return _points.entries.map((entry) {
136 : // Convert null attributes to empty attributes for MetricPoint
137 44 : final attributes = entry.key ?? OTelFactory.otelFactory!.attributes();
138 :
139 : // Convert numeric value to the specific generic type T
140 : final T typedValue;
141 18 : if (T == int) {
142 51 : typedValue = entry.value.value.toInt() as T;
143 4 : } else if (T == double) {
144 9 : typedValue = entry.value.value.toDouble() as T;
145 : } else {
146 2 : typedValue = entry.value.value;
147 : }
148 :
149 18 : return MetricPoint<T>.sum(
150 : attributes: attributes,
151 18 : startTime: _startTime,
152 : time: now,
153 : value: typedValue,
154 18 : isMonotonic: isMonotonic,
155 54 : exemplars: entry.value.reservoir.collectAndReset(attributes),
156 : );
157 18 : }).toList();
158 : }
159 :
160 : /// Resets all points (for delta temporality).
161 : ///
162 : /// This method clears all accumulated values. It is used when
163 : /// reporting with delta temporality to reset the accumulation
164 : /// after each export.
165 11 : @override
166 : void reset() {
167 22 : _points.clear();
168 : }
169 : }
170 :
171 : /// Internal class representing data for a single sum point.
172 : ///
173 : /// This class tracks the accumulated value, last update time,
174 : /// and exemplars for a specific combination of attributes.
175 : class _SumPointData<T extends num> {
176 : /// The accumulated value.
177 : T value;
178 :
179 : /// The time this point was last updated.
180 : DateTime lastUpdateTime;
181 :
182 : /// Reservoir for this point.
183 : final ExemplarReservoir reservoir;
184 :
185 : /// Creates a new _SumPointData instance.
186 : ///
187 : /// @param value The initial value
188 : /// @param lastUpdateTime The time of the initial value
189 : /// @param reservoir The exemplar reservoir for this point
190 25 : _SumPointData(
191 : {required this.value,
192 : required this.lastUpdateTime,
193 : required this.reservoir});
194 :
195 : /// Adds a value to this point (for synchronous counters).
196 : ///
197 : /// @param delta The value to add to the accumulated value
198 10 : void add(T delta) {
199 : // Handle the addition with proper type conversion
200 10 : if (T == int) {
201 36 : value = (value + delta).toInt() as T;
202 4 : } else if (T == double) {
203 12 : value = (value + delta).toDouble() as T;
204 : } else {
205 3 : value = (value + delta) as T;
206 : }
207 :
208 20 : lastUpdateTime = DateTime.now();
209 : }
210 :
211 : /// Sets the value directly (for asynchronous counters).
212 : ///
213 : /// @param newValue The new absolute value to set
214 0 : void setValue(T newValue) {
215 0 : value = newValue;
216 0 : lastUpdateTime = DateTime.now();
217 : }
218 :
219 0 : @override
220 0 : String toString() => 'SumPointData(value: $value)';
221 : }
|
