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