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>, SDKObservableInstrument {
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 : /// Observations queued by batch callbacks, drained on the next [collect].
27 : final List<Measurement<T>> _batchObservations = [];
28 :
29 : /// Creates a new ObservableCounter instance.
30 11 : ObservableCounter({
31 : required APIObservableCounter<T> apiCounter,
32 : required Meter meter,
33 : }) : _apiCounter = apiCounter,
34 : _meter = meter,
35 11 : _storage = SumStorage<T>(
36 : isMonotonic: true,
37 22 : exemplarFilter: meter.provider.exemplarFilter,
38 : );
39 :
40 11 : @override
41 22 : 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 8 : @override
53 : bool isEnabled() {
54 : // In the SDK, metrics are enabled based on the meter provider's enabled state
55 24 : return _meter.provider.enabled;
56 : }
57 :
58 11 : @override
59 11 : APIMeter get meter => _meter;
60 :
61 8 : @override
62 16 : List<ObservableCallback<T>> get callbacks => _apiCounter.callbacks;
63 :
64 4 : @override
65 : APICallbackRegistration<T> addCallback(ObservableCallback<T> callback) {
66 : // Register with the API implementation first
67 8 : final registration = _apiCounter.addCallback(callback);
68 :
69 : // Return a registration that also unregisters from our list
70 4 : return _ObservableCounterCallbackRegistration<T>(
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 8 : @override
115 : List<Measurement<T>> collect() {
116 8 : if (!isEnabled()) {
117 4 : _batchObservations.clear();
118 2 : return [];
119 : }
120 :
121 8 : final result = <Measurement<T>>[];
122 :
123 : // Get a snapshot of callbacks to avoid concurrent modification issues
124 16 : final callbacksSnapshot = List<ObservableCallback<T>>.from(callbacks);
125 :
126 : // Return early if nothing can observe
127 14 : if (callbacksSnapshot.isEmpty && _batchObservations.isEmpty) {
128 : return result;
129 : }
130 :
131 : // First, clear previous values to prepare for fresh collection
132 : // This is necessary to avoid accumulating values from multiple collections
133 16 : _storage.reset();
134 :
135 : // Call all callbacks
136 15 : for (final callback in callbacksSnapshot) {
137 : try {
138 : // Create a new observable result for each callback
139 7 : final observableResult = ObservableResult<T>();
140 :
141 : // Call the callback with the observable result
142 : // Cast the parameter to ensure type safety
143 : try {
144 7 : callback(observableResult as APIObservableResult<T>);
145 : } catch (e) {
146 4 : print('Type error in callback: $e');
147 : continue;
148 : }
149 :
150 : // Process the measurements from the observable result
151 14 : for (final measurement in observableResult.measurements) {
152 14 : if (_record(measurement)) result.add(measurement);
153 : }
154 : } catch (e) {
155 0 : print(
156 0 : 'Error collecting measurements from ObservableCounter callback: $e',
157 : );
158 : }
159 : }
160 :
161 : // Batch callbacks ran before this instrument collected; drain what they
162 : // observed through the same monotonicity logic.
163 9 : for (final measurement in _batchObservations) {
164 2 : if (_record(measurement)) result.add(measurement);
165 : }
166 16 : _batchObservations.clear();
167 :
168 : return result;
169 : }
170 :
171 : /// Records one observed absolute value into storage.
172 : ///
173 : /// Returns true when the measurement carries a change (a positive delta
174 : /// or a counter reset) that belongs in the collected result; a zero delta
175 : /// is still stored for cumulative reporting but not returned.
176 8 : bool _record(Measurement<T> measurement) {
177 16 : final value = castNum<T>(measurement.value);
178 : final attributes =
179 24 : measurement.attributes ?? OTelFactory.otelFactory!.attributes();
180 :
181 : // Check for monotonicity - current value should be >= last value.
182 : // A decrease indicates a counter reset; per spec we just record the
183 : // current value.
184 24 : final lastValue = _lastValues[attributes] ?? castNum<T>(0);
185 24 : _storage.record(value, attributes, Context.current);
186 :
187 : // Store the latest value for next time
188 16 : _lastValues[attributes] = value;
189 :
190 8 : return value != lastValue;
191 : }
192 :
193 : /// Collects metrics for the SDK metric export.
194 : ///
195 : /// This is called by the MeterProvider during metric collection.
196 : /// Per the OTel spec, observable instruments must invoke their
197 : /// registered callbacks on every collection cycle. Drive [collect]
198 : /// first so the callback runs and storage reflects the latest
199 : /// absolute counter value before we read it.
200 6 : @override
201 : List<Metric> collectMetrics() {
202 6 : if (!isEnabled()) {
203 3 : return [];
204 : }
205 :
206 6 : collect();
207 :
208 : // Get the points from storage
209 6 : final points = collectPoints();
210 6 : if (points.isEmpty) {
211 1 : return [];
212 : }
213 :
214 : // Create the metric to export
215 6 : return [
216 6 : Metric.sum(
217 6 : name: name,
218 6 : description: description,
219 6 : unit: unit,
220 : temporality: AggregationTemporality.cumulative,
221 : points: points,
222 : isMonotonic: true, // Counters are monotonic
223 : ),
224 : ];
225 : }
226 :
227 : /// Gets the current points for this counter.
228 : /// This is used by the SDK to collect metrics.
229 6 : List<MetricPoint<T>> collectPoints() {
230 6 : if (!isEnabled()) {
231 1 : return [];
232 : }
233 :
234 : // Then return points from storage
235 12 : return _storage.collectPoints();
236 : }
237 :
238 : /// Resets the counter. This is only used for testing.
239 3 : void reset() {
240 6 : _storage.reset();
241 6 : _lastValues.clear();
242 : }
243 : }
244 :
245 : /// Wrapper for APICallbackRegistration that also handles our internal state.
246 : class _ObservableCounterCallbackRegistration<T extends num>
247 : implements APICallbackRegistration<T> {
248 : /// The API registration.
249 : final APICallbackRegistration<T> apiRegistration;
250 :
251 : /// The counter this registration is for.
252 : final ObservableCounter<T> counter;
253 :
254 : /// The callback that was registered.
255 : final ObservableCallback<T> callback;
256 :
257 4 : _ObservableCounterCallbackRegistration({
258 : required this.apiRegistration,
259 : required this.counter,
260 : required this.callback,
261 : });
262 :
263 2 : @override
264 : void unregister() {
265 : // Unregister from the API implementation
266 4 : apiRegistration.unregister();
267 :
268 : // Also remove from our counter directly for redundancy
269 6 : counter.removeCallback(callback);
270 : }
271 : }
|
