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>, SDKObservableInstrument {
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 : /// Observations queued by batch callbacks, drained on the next [collect].
27 : final List<Measurement<T>> _batchObservations = [];
28 :
29 : /// Creates a new ObservableUpDownCounter instance.
30 8 : ObservableUpDownCounter({
31 : required APIObservableUpDownCounter<T> apiCounter,
32 : required Meter meter,
33 : }) : _apiCounter = apiCounter,
34 : _meter = meter,
35 8 : _storage = SumStorage<T>(
36 : isMonotonic: false,
37 16 : exemplarFilter: meter.provider.exemplarFilter,
38 : );
39 :
40 8 : @override
41 16 : String get name => _apiCounter.name;
42 :
43 6 : @override
44 12 : String? get unit => _apiCounter.unit;
45 :
46 6 : @override
47 12 : String? get description => _apiCounter.description;
48 :
49 1 : @override
50 2 : InstrumentAdvisory? get advisory => _apiCounter.advisory;
51 :
52 6 : @override
53 : bool isEnabled() {
54 : // In the SDK, metrics are enabled based on the meter provider's enabled state
55 18 : return _meter.provider.enabled;
56 : }
57 :
58 8 : @override
59 8 : APIMeter get meter => _meter;
60 :
61 6 : @override
62 12 : List<ObservableCallback<T>> get callbacks => _apiCounter.callbacks;
63 :
64 3 : @override
65 : APICallbackRegistration<T> addCallback(ObservableCallback<T> callback) {
66 : // Register with the API implementation first
67 6 : final registration = _apiCounter.addCallback(callback);
68 :
69 : // Return a registration that also unregisters from our list
70 3 : return _ObservableUpDownCounterCallbackRegistration(
71 : apiRegistration: registration,
72 : counter: this,
73 : callback: callback,
74 : );
75 : }
76 :
77 2 : @override
78 : void removeCallback(ObservableCallback<T> callback) {
79 4 : _apiCounter.removeCallback(callback);
80 : }
81 :
82 : /// Gets the current value of the counter for a specific set of attributes.
83 : /// If no attributes are provided, returns the sum of all recorded values.
84 1 : T getValue([Attributes? attributes]) {
85 : final num value;
86 :
87 : if (attributes == null) {
88 : // For no attributes, sum all points
89 3 : value = _storage.collectPoints().fold<num>(
90 : 0,
91 3 : (sum, point) => sum + point.value,
92 : );
93 : } else {
94 : // For specific attributes, get that value
95 2 : value = _storage.getValue(attributes);
96 : }
97 :
98 : // Handle the cast to the generic type
99 2 : if (T == int) return value.toInt() as T;
100 2 : if (T == double) return value.toDouble() as T;
101 : return value as T;
102 : }
103 :
104 1 : @override
105 : void observeFromBatch(num value, Attributes? attributes) {
106 2 : _batchObservations.add(
107 1 : OTelFactory.otelFactory!
108 2 : .createMeasurement<T>(castNum<T>(value), attributes),
109 : );
110 : }
111 :
112 : /// Collects measurements from all registered callbacks and from any
113 : /// batch callbacks that observed this instrument since the last collection.
114 6 : @override
115 : List<Measurement<T>> collect() {
116 6 : if (!isEnabled()) {
117 4 : _batchObservations.clear();
118 2 : return [];
119 : }
120 :
121 6 : final result = <Measurement<T>>[];
122 12 : final callbackList = List<ObservableCallback<T>>.from(callbacks);
123 :
124 : // Return early if nothing can observe
125 12 : if (callbackList.isEmpty && _batchObservations.isEmpty) {
126 : return result;
127 : }
128 :
129 : // First, clear previous values to prepare for fresh collection
130 : // This is necessary to avoid accumulating values from multiple collections
131 12 : _storage.reset();
132 :
133 : // Call all callbacks
134 11 : for (final callback in callbackList) {
135 : try {
136 : // Create a new observable result for each callback
137 5 : final observableResult = ObservableResult<T>();
138 :
139 : // Call the callback with the observable result
140 : // Cast the parameter to ensure type safety
141 : try {
142 5 : callback(observableResult as APIObservableResult<T>);
143 : } catch (e) {
144 4 : print('Type error in callback: $e');
145 : continue;
146 : }
147 :
148 : // Process the measurements from the observable result
149 10 : for (final measurement in observableResult.measurements) {
150 5 : _record(measurement);
151 5 : result.add(measurement);
152 : }
153 : } catch (e) {
154 0 : print(
155 0 : 'Error collecting measurements from ObservableUpDownCounter callback: $e',
156 : );
157 : }
158 : }
159 :
160 : // Batch callbacks ran before this instrument collected; drain what they
161 : // observed into the same storage.
162 7 : for (final measurement in _batchObservations) {
163 1 : _record(measurement);
164 1 : result.add(measurement);
165 : }
166 12 : _batchObservations.clear();
167 :
168 : return result;
169 : }
170 :
171 : /// Records one observed absolute value into storage.
172 : ///
173 : /// Per the spec, for ObservableUpDownCounter we record the absolute value
174 : /// directly - not the delta.
175 6 : void _record(Measurement<T> measurement) {
176 12 : final value = castNum<T>(measurement.value);
177 : final attributes =
178 18 : measurement.attributes ?? OTelFactory.otelFactory!.attributes();
179 18 : _storage.record(value, attributes, Context.current);
180 :
181 : // Keep track of the last value for debugging and tracking
182 12 : _lastValues[attributes] = value;
183 : }
184 :
185 : /// Gets the current points for this counter.
186 : /// This is used by the SDK to collect metrics.
187 6 : List<MetricPoint<T>> collectPoints() {
188 6 : if (!isEnabled()) {
189 1 : return [];
190 : }
191 :
192 : // Then return points from storage
193 12 : return _storage.collectPoints();
194 : }
195 :
196 : /// Collects metrics for the SDK metric export.
197 : ///
198 : /// This is called by the MeterProvider during metric collection.
199 : /// Per the OTel spec, observable instruments must invoke their
200 : /// registered callbacks on every collection cycle. Drive [collect]
201 : /// first so the callback runs and storage reflects the latest
202 : /// value before we read it.
203 6 : @override
204 : List<Metric> collectMetrics() {
205 6 : if (!isEnabled()) {
206 3 : return [];
207 : }
208 :
209 6 : collect();
210 :
211 : // Get the points from storage
212 6 : final points = collectPoints();
213 6 : if (points.isEmpty) {
214 1 : return [];
215 : }
216 :
217 : // Create the metric to export
218 6 : return [
219 6 : Metric.sum(
220 6 : name: name,
221 6 : description: description,
222 6 : unit: unit,
223 : temporality: AggregationTemporality.cumulative,
224 : points: points,
225 : isMonotonic: false, // Up/down counters are non-monotonic
226 : ),
227 : ];
228 : }
229 :
230 : /// Resets the counter for testing. This is not typically used in production.
231 1 : void reset() {
232 2 : _storage.reset();
233 2 : _lastValues.clear();
234 : }
235 : }
236 :
237 : /// Wrapper for APICallbackRegistration that also handles our internal state.
238 : class _ObservableUpDownCounterCallbackRegistration<T extends num>
239 : implements APICallbackRegistration<T> {
240 : /// The API registration.
241 : final APICallbackRegistration<T> apiRegistration;
242 :
243 : /// The counter this registration is for.
244 : final ObservableUpDownCounter<T> counter;
245 :
246 : /// The callback that was registered.
247 : final ObservableCallback<T> callback;
248 :
249 3 : _ObservableUpDownCounterCallbackRegistration({
250 : required this.apiRegistration,
251 : required this.counter,
252 : required this.callback,
253 : });
254 :
255 2 : @override
256 : void unregister() {
257 : // Unregister from the API implementation
258 4 : apiRegistration.unregister();
259 :
260 : // Also remove from our counter directly for redundancy
261 6 : counter.removeCallback(callback);
262 : }
263 : }
|
