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