Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import '../../../dartastic_opentelemetry.dart';
5 :
6 : /// ObservableUpDownCounter is an asynchronous instrument that reports additive
7 : /// values when observed.
8 : ///
9 : /// An ObservableUpDownCounter is used to measure a value that increases and
10 : /// decreases where measurements are made by a callback function. For example,
11 : /// number of active requests, queue size, pool size.
12 : class ObservableUpDownCounter<T extends num>
13 : implements APIObservableUpDownCounter<T>, SDKInstrument {
14 : /// The underlying API ObservableUpDownCounter.
15 : final APIObservableUpDownCounter<T> _apiCounter;
16 :
17 : /// The Meter that created this ObservableUpDownCounter.
18 : final Meter _meter;
19 :
20 : /// Storage for accumulating counter measurements.
21 : final SumStorage<T> _storage = SumStorage<T>(isMonotonic: false);
22 :
23 : /// The last observed values, for tracking changes.
24 : final Map<Attributes, T> _lastValues = {};
25 :
26 : /// Creates a new ObservableUpDownCounter instance.
27 7 : ObservableUpDownCounter({
28 : required APIObservableUpDownCounter<T> apiCounter,
29 : required Meter meter,
30 : }) : _apiCounter = apiCounter,
31 : _meter = meter;
32 :
33 7 : @override
34 14 : 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 5 : @override
43 : bool get enabled {
44 : // In the SDK, metrics are enabled based on the meter provider's enabled state
45 15 : return _meter.provider.enabled;
46 : }
47 :
48 7 : @override
49 7 : APIMeter get meter => _meter;
50 :
51 5 : @override
52 10 : List<ObservableCallback<T>> get callbacks => _apiCounter.callbacks;
53 :
54 3 : @override
55 : APICallbackRegistration<T> addCallback(ObservableCallback<T> callback) {
56 : // Register with the API implementation first
57 6 : final registration = _apiCounter.addCallback(callback);
58 :
59 : // Return a registration that also unregisters from our list
60 3 : return _ObservableUpDownCounterCallbackRegistration(
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 5 : @override
96 : List<Measurement<T>> collect() {
97 5 : if (!enabled) {
98 2 : return [];
99 : }
100 :
101 5 : final result = <Measurement<T>>[];
102 10 : final callbackList = List<ObservableCallback<T>>.from(callbacks);
103 :
104 : // Return early if no callbacks registered
105 5 : if (callbackList.isEmpty) {
106 : return result;
107 : }
108 :
109 : // First, clear previous values to prepare for fresh collection
110 : // This is necessary to avoid accumulating values from multiple collections
111 10 : _storage.reset();
112 :
113 : // Call all callbacks
114 10 : for (final callback in callbackList) {
115 : try {
116 : // Create a new observable result for each callback
117 5 : final observableResult = ObservableResult<T>();
118 :
119 : // Call the callback with the observable result
120 : // Cast the parameter to ensure type safety
121 : try {
122 5 : callback(observableResult as APIObservableResult<T>);
123 : } catch (e) {
124 4 : print('Type error in callback: $e');
125 : continue;
126 : }
127 :
128 : // Process the measurements from the observable result
129 10 : for (final measurement in observableResult.measurements) {
130 : // Type checking for the generic parameter
131 5 : final dynamic rawValue = measurement.value;
132 5 : final value = (rawValue is num)
133 : ? rawValue
134 0 : : num.tryParse(rawValue.toString()) ?? 0;
135 : final attributes =
136 10 : measurement.attributes ?? OTelFactory.otelFactory!.attributes();
137 :
138 : // Per the spec, for ObservableUpDownCounter we record the absolute value
139 : // directly - not the delta
140 : // For SDK storage, convert the num to the appropriate T type
141 5 : if (T == int) {
142 15 : _storage.record(value.toInt() as T, attributes);
143 2 : } else if (T == double) {
144 6 : _storage.record(value.toDouble() as T, attributes);
145 : } else {
146 0 : _storage.record(value as T, attributes);
147 : }
148 :
149 : // Add measurement with the absolute value to the result
150 5 : result.add(measurement);
151 :
152 : // Keep track of the last value for debugging and tracking
153 5 : if (T == int) {
154 15 : _lastValues[attributes] = value.toInt() as T;
155 2 : } else if (T == double) {
156 6 : _lastValues[attributes] = value.toDouble() as T;
157 : } else {
158 0 : _lastValues[attributes] = value as T;
159 : }
160 : }
161 : } catch (e) {
162 0 : print(
163 0 : 'Error collecting measurements from ObservableUpDownCounter callback: $e',
164 : );
165 : }
166 : }
167 :
168 : return result;
169 : }
170 :
171 : /// Gets the current points for this counter.
172 : /// This is used by the SDK to collect metrics.
173 5 : List<MetricPoint<T>> collectPoints() {
174 5 : if (!enabled) {
175 1 : return [];
176 : }
177 :
178 : // Then return points from storage
179 10 : return _storage.collectPoints();
180 : }
181 :
182 : /// Collects metrics for the SDK metric export.
183 : ///
184 : /// This is called by the MeterProvider during metric collection.
185 : /// Per the OTel spec, observable instruments must invoke their
186 : /// registered callbacks on every collection cycle. Drive [collect]
187 : /// first so the callback runs and storage reflects the latest
188 : /// value before we read it.
189 5 : @override
190 : List<Metric> collectMetrics() {
191 5 : if (!enabled) {
192 3 : return [];
193 : }
194 :
195 5 : collect();
196 :
197 : // Get the points from storage
198 5 : final points = collectPoints();
199 5 : if (points.isEmpty) {
200 1 : return [];
201 : }
202 :
203 : // Create the metric to export
204 5 : return [
205 5 : Metric.sum(
206 5 : name: name,
207 5 : description: description,
208 5 : unit: unit,
209 : temporality: AggregationTemporality.cumulative,
210 : points: points,
211 : isMonotonic: false, // Up/down counters are non-monotonic
212 : ),
213 : ];
214 : }
215 :
216 : /// Resets the counter for testing. This is not typically used in production.
217 1 : void reset() {
218 2 : _storage.reset();
219 2 : _lastValues.clear();
220 : }
221 : }
222 :
223 : /// Wrapper for APICallbackRegistration that also handles our internal state.
224 : class _ObservableUpDownCounterCallbackRegistration<T extends num>
225 : implements APICallbackRegistration<T> {
226 : /// The API registration.
227 : final APICallbackRegistration<T> apiRegistration;
228 :
229 : /// The counter this registration is for.
230 : final ObservableUpDownCounter<T> counter;
231 :
232 : /// The callback that was registered.
233 : final ObservableCallback<T> callback;
234 :
235 3 : _ObservableUpDownCounterCallbackRegistration({
236 : required this.apiRegistration,
237 : required this.counter,
238 : required this.callback,
239 : });
240 :
241 2 : @override
242 : void unregister() {
243 : // Unregister from the API implementation
244 4 : apiRegistration.unregister();
245 :
246 : // Also remove from our counter directly for redundancy
247 6 : counter.removeCallback(callback);
248 : }
249 : }
|
