Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import '../../../dartastic_opentelemetry.dart';
5 :
6 : /// ObservableCounter is an asynchronous instrument that reports monotonically
7 : /// increasing values when observed.
8 : ///
9 : /// An ObservableCounter is used to measure monotonically increasing values
10 : /// where measurements are made by a callback function. For example, CPU time,
11 : /// bytes received, or number of operations.
12 : class ObservableCounter<T extends num>
13 : implements APIObservableCounter<T>, SDKInstrument {
14 : /// The underlying API ObservableCounter.
15 : final APIObservableCounter<T> _apiCounter;
16 :
17 : /// The Meter that created this ObservableCounter.
18 : final Meter _meter;
19 :
20 : /// Storage for accumulating counter measurements.
21 : final SumStorage<T> _storage = SumStorage<T>(isMonotonic: true);
22 :
23 : /// The last observed values, for tracking and detecting resets.
24 : final Map<Attributes, T> _lastValues = {};
25 :
26 : /// Creates a new ObservableCounter instance.
27 10 : ObservableCounter({
28 : required APIObservableCounter<T> apiCounter,
29 : required Meter meter,
30 : }) : _apiCounter = apiCounter,
31 : _meter = meter;
32 :
33 10 : @override
34 20 : String get name => _apiCounter.name;
35 :
36 5 : @override
37 10 : String? get unit => _apiCounter.unit;
38 :
39 5 : @override
40 10 : String? get description => _apiCounter.description;
41 :
42 7 : @override
43 : bool get enabled {
44 : // In the SDK, metrics are enabled based on the meter provider's enabled state
45 21 : return _meter.provider.enabled;
46 : }
47 :
48 10 : @override
49 10 : APIMeter get meter => _meter;
50 :
51 7 : @override
52 14 : List<ObservableCallback<T>> get callbacks => _apiCounter.callbacks;
53 :
54 4 : @override
55 : APICallbackRegistration<T> addCallback(ObservableCallback<T> callback) {
56 : // Register with the API implementation first
57 8 : final registration = _apiCounter.addCallback(callback);
58 :
59 : // Return a registration that also unregisters from our list
60 4 : return _ObservableCounterCallbackRegistration<T>(
61 : apiRegistration: registration,
62 : counter: this,
63 : callback: callback,
64 : );
65 : }
66 :
67 2 : @override
68 : void removeCallback(ObservableCallback<T> callback) {
69 4 : _apiCounter.removeCallback(callback);
70 : }
71 :
72 : /// Gets the current value of the counter for a specific set of attributes.
73 : /// If no attributes are provided, returns the sum of all recorded values.
74 1 : T getValue([Attributes? attributes]) {
75 : final num value;
76 :
77 : if (attributes == null) {
78 : // For no attributes, sum all points
79 3 : value = _storage.collectPoints().fold<num>(
80 : 0,
81 3 : (sum, point) => sum + point.value,
82 : );
83 : } else {
84 : // For specific attributes, get that value
85 2 : value = _storage.getValue(attributes);
86 : }
87 :
88 : // Handle the cast to the generic type
89 2 : if (T == int) return value.toInt() as T;
90 2 : if (T == double) return value.toDouble() as T;
91 : return value as T;
92 : }
93 :
94 : /// Collects measurements from all registered callbacks.
95 7 : @override
96 : List<Measurement<T>> collect() {
97 7 : if (!enabled) {
98 2 : return [];
99 : }
100 :
101 7 : final result = <Measurement<T>>[];
102 :
103 : // Get a snapshot of callbacks to avoid concurrent modification issues
104 14 : final callbacksSnapshot = List<ObservableCallback<T>>.from(callbacks);
105 :
106 : // Return early if no callbacks registered
107 7 : if (callbacksSnapshot.isEmpty) {
108 : return result;
109 : }
110 :
111 : // First, clear previous values to prepare for fresh collection
112 : // This is necessary to avoid accumulating values from multiple collections
113 14 : _storage.reset();
114 :
115 : // Call all callbacks
116 14 : for (final callback in callbacksSnapshot) {
117 : try {
118 : // Create a new observable result for each callback
119 7 : final observableResult = ObservableResult<T>();
120 :
121 : // Call the callback with the observable result
122 : // Cast the parameter to ensure type safety
123 : try {
124 7 : callback(observableResult as APIObservableResult<T>);
125 : } catch (e) {
126 4 : print('Type error in callback: $e');
127 : continue;
128 : }
129 :
130 : // Process the measurements from the observable result
131 14 : for (final measurement in observableResult.measurements) {
132 : // Type checking for the generic parameter
133 7 : final dynamic rawValue = measurement.value;
134 7 : final value = (rawValue is num)
135 : ? rawValue
136 0 : : num.tryParse(rawValue.toString()) ?? 0;
137 : final attributes =
138 14 : measurement.attributes ?? OTelFactory.otelFactory!.attributes();
139 :
140 : // Check for monotonicity - current value should be >= last value
141 : final lastValue =
142 21 : (_lastValues[attributes] ?? (T == int ? 0 : 0.0)) as T;
143 :
144 : // If value decreased, it indicates a counter reset
145 7 : if (value < lastValue) {
146 : // Per spec, for a reset we just record the current value
147 : // For SDK storage, convert the num to the appropriate T type
148 3 : if (T == int) {
149 6 : _storage.record(value.toInt() as T, attributes);
150 1 : } else if (T == double) {
151 3 : _storage.record(value.toDouble() as T, attributes);
152 : } else {
153 0 : _storage.record(value as T, attributes);
154 : }
155 3 : result.add(measurement);
156 7 : } else if (value > lastValue) {
157 : // Only add measurements with positive deltas
158 : // For SDK storage, convert the num to the appropriate T type
159 7 : if (T == int) {
160 18 : _storage.record(value.toInt() as T, attributes);
161 3 : } else if (T == double) {
162 9 : _storage.record(value.toDouble() as T, attributes);
163 : } else {
164 0 : _storage.record(value as T, attributes);
165 : }
166 7 : result.add(measurement);
167 : } else {
168 : // For zero deltas, we still record the value in storage for cumulative reporting,
169 : // but don't include it in the returned measurements
170 : // For SDK storage, convert the num to the appropriate T type
171 5 : if (T == int) {
172 12 : _storage.record(value.toInt() as T, attributes);
173 2 : } else if (T == double) {
174 6 : _storage.record(value.toDouble() as T, attributes);
175 : } else {
176 0 : _storage.record(value as T, attributes);
177 : }
178 : // Note: The measurement is deliberately not added to the result list
179 : }
180 :
181 : // Store the latest value for next time
182 7 : if (T == int) {
183 18 : _lastValues[attributes] = value.toInt() as T;
184 3 : } else if (T == double) {
185 9 : _lastValues[attributes] = value.toDouble() as T;
186 : } else {
187 0 : _lastValues[attributes] = value as T;
188 : }
189 : }
190 : } catch (e) {
191 0 : print(
192 0 : 'Error collecting measurements from ObservableCounter callback: $e',
193 : );
194 : }
195 : }
196 :
197 : return result;
198 : }
199 :
200 : /// Collects metrics for the SDK metric export.
201 : ///
202 : /// This is called by the MeterProvider during metric collection.
203 : /// Per the OTel spec, observable instruments must invoke their
204 : /// registered callbacks on every collection cycle. Drive [collect]
205 : /// first so the callback runs and storage reflects the latest
206 : /// absolute counter value before we read it.
207 5 : @override
208 : List<Metric> collectMetrics() {
209 5 : if (!enabled) {
210 3 : return [];
211 : }
212 :
213 5 : collect();
214 :
215 : // Get the points from storage
216 5 : final points = collectPoints();
217 5 : if (points.isEmpty) {
218 1 : return [];
219 : }
220 :
221 : // Create the metric to export
222 5 : return [
223 5 : Metric.sum(
224 5 : name: name,
225 5 : description: description,
226 5 : unit: unit,
227 : temporality: AggregationTemporality.cumulative,
228 : points: points,
229 : isMonotonic: true, // Counters are monotonic
230 : ),
231 : ];
232 : }
233 :
234 : /// Gets the current points for this counter.
235 : /// This is used by the SDK to collect metrics.
236 5 : List<MetricPoint<T>> collectPoints() {
237 5 : if (!enabled) {
238 1 : return [];
239 : }
240 :
241 : // Then return points from storage
242 10 : return _storage.collectPoints();
243 : }
244 :
245 : /// Resets the counter. This is only used for testing.
246 3 : void reset() {
247 6 : _storage.reset();
248 6 : _lastValues.clear();
249 : }
250 : }
251 :
252 : /// Wrapper for APICallbackRegistration that also handles our internal state.
253 : class _ObservableCounterCallbackRegistration<T extends num>
254 : implements APICallbackRegistration<T> {
255 : /// The API registration.
256 : final APICallbackRegistration<T> apiRegistration;
257 :
258 : /// The counter this registration is for.
259 : final ObservableCounter<T> counter;
260 :
261 : /// The callback that was registered.
262 : final ObservableCallback<T> callback;
263 :
264 4 : _ObservableCounterCallbackRegistration({
265 : required this.apiRegistration,
266 : required this.counter,
267 : required this.callback,
268 : });
269 :
270 2 : @override
271 : void unregister() {
272 : // Unregister from the API implementation
273 4 : apiRegistration.unregister();
274 :
275 : // Also remove from our counter directly for redundancy
276 6 : counter.removeCallback(callback);
277 : }
278 : }
|
