Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import '../../../dartastic_opentelemetry.dart';
5 :
6 : /// ObservableGauge is an asynchronous instrument which reports non-additive value(s)
7 : /// when the instrument is being observed.
8 : ///
9 : /// An ObservableGauge is used to asynchronously measure a non-additive current value
10 : /// that cannot be calculated synchronously.
11 : class ObservableGauge<T extends num>
12 : implements APIObservableGauge<T>, SDKObservableInstrument {
13 : /// The underlying API ObservableGauge.
14 : final APIObservableGauge<T> _apiGaugeDelegate;
15 :
16 : /// The Meter that created this ObservableGauge.
17 : final Meter _meter;
18 :
19 : /// Storage for gauge measurements.
20 : final GaugeStorage<T> _storage;
21 :
22 : /// Observations queued by batch callbacks, drained on the next [collect].
23 : final List<Measurement<T>> _batchObservations = [];
24 :
25 : /// Creates a new ObservableGauge instance.
26 8 : ObservableGauge({
27 : required APIObservableGauge<T> apiGauge,
28 : required Meter meter,
29 : }) : _apiGaugeDelegate = apiGauge,
30 : _meter = meter,
31 8 : _storage = GaugeStorage<T>(
32 16 : exemplarFilter: meter.provider.exemplarFilter,
33 : );
34 :
35 8 : @override
36 16 : String get name => _apiGaugeDelegate.name;
37 :
38 6 : @override
39 12 : String? get unit => _apiGaugeDelegate.unit;
40 :
41 6 : @override
42 12 : String? get description => _apiGaugeDelegate.description;
43 :
44 1 : @override
45 2 : InstrumentAdvisory? get advisory => _apiGaugeDelegate.advisory;
46 :
47 6 : @override
48 : bool isEnabled() {
49 18 : return _meter.provider.enabled;
50 : }
51 :
52 8 : @override
53 8 : APIMeter get meter => _meter;
54 :
55 6 : @override
56 12 : List<ObservableCallback<T>> get callbacks => _apiGaugeDelegate.callbacks;
57 :
58 3 : @override
59 : APICallbackRegistration<T> addCallback(ObservableCallback<T> callback) {
60 : // Register with the API implementation first
61 6 : final registration = _apiGaugeDelegate.addCallback(callback);
62 :
63 : // Return a registration that handles unregistering properly
64 3 : return _ObservableGaugeCallbackRegistration<T>(
65 : apiRegistration: registration,
66 : gauge: this,
67 : callback: callback,
68 : );
69 : }
70 :
71 2 : @override
72 : void removeCallback(ObservableCallback<T> callback) {
73 4 : _apiGaugeDelegate.removeCallback(callback);
74 : }
75 :
76 : /// Gets the current value of the gauge for a specific set of attributes.
77 : /// If no attributes are provided, returns the average of all recorded values.
78 1 : T getValue([Attributes? attributes]) {
79 : final num value;
80 :
81 : if (attributes == null) {
82 : // For gauges without attributes, we return the average of all values
83 2 : final points = _storage.collectPoints();
84 1 : if (points.isEmpty) {
85 : value = 0;
86 : } else {
87 : value =
88 5 : points.fold<num>(0, (sum, point) => sum + (point.value as num)) /
89 1 : points.length;
90 : }
91 : } else {
92 : // For specific attributes, get that value
93 2 : value = _storage.getValue(attributes);
94 : }
95 :
96 : // Handle the cast to the generic type
97 2 : if (T == int) return value.toInt() as T;
98 2 : if (T == double) return value.toDouble() as T;
99 : return value as T;
100 : }
101 :
102 1 : @override
103 : void observeFromBatch(num value, Attributes? attributes) {
104 2 : _batchObservations.add(
105 1 : OTelFactory.otelFactory!
106 2 : .createMeasurement<T>(castNum<T>(value), attributes),
107 : );
108 : }
109 :
110 : /// Collects measurements from all registered callbacks and from any
111 : /// batch callbacks that observed this instrument since the last collection.
112 6 : @override
113 : List<Measurement<T>> collect() {
114 6 : if (!isEnabled()) {
115 4 : _batchObservations.clear();
116 2 : return [];
117 : }
118 :
119 6 : final result = <Measurement<T>>[];
120 :
121 : // Get a snapshot of callbacks to avoid concurrent modification issues
122 12 : final callbacksSnapshot = List<ObservableCallback<T>>.from(callbacks);
123 :
124 : // Call all callbacks
125 12 : for (final callback in callbacksSnapshot) {
126 : try {
127 : // Create a new observable result for each callback
128 6 : final observableResult = ObservableResult<T>();
129 :
130 : // Call the callback with the observable result
131 : // Cast the parameter to ensure type safety
132 : try {
133 6 : callback(observableResult as APIObservableResult<T>);
134 : } catch (e) {
135 4 : print('Type error in callback: $e');
136 : continue;
137 : }
138 :
139 : // Process the measurements from the observable result
140 12 : for (final measurement in observableResult.measurements) {
141 6 : _record(measurement);
142 6 : result.add(measurement);
143 : }
144 : } catch (e) {
145 0 : print(
146 0 : 'Error collecting measurements from ObservableGauge callback: $e',
147 : );
148 : }
149 : }
150 :
151 : // Batch callbacks ran before this instrument collected; drain what they
152 : // observed into the same storage.
153 7 : for (final measurement in _batchObservations) {
154 1 : _record(measurement);
155 1 : result.add(measurement);
156 : }
157 12 : _batchObservations.clear();
158 :
159 : return result;
160 : }
161 :
162 : /// Records one observed value; a gauge keeps only the latest per attributes.
163 6 : void _record(Measurement<T> measurement) {
164 : final attributes =
165 18 : measurement.attributes ?? OTelFactory.otelFactory!.attributes();
166 30 : _storage.record(castNum<T>(measurement.value), attributes, Context.current);
167 : }
168 :
169 : /// Collects metrics for the SDK metric export.
170 : ///
171 : /// This is called by the MeterProvider during metric collection.
172 : /// Per the OTel spec, observable instruments must invoke their
173 : /// registered callbacks on every collection cycle and report the
174 : /// values the callback observes — that's the whole point of being
175 : /// "observable" vs sync. Drive [collect] first so the callback
176 : /// runs and storage is fresh; discard the returned measurements
177 : /// (collect already pushed them into [_storage]).
178 6 : @override
179 : List<Metric> collectMetrics() {
180 6 : if (!isEnabled()) {
181 3 : return [];
182 : }
183 :
184 6 : collect();
185 :
186 : // Get the points from storage
187 6 : final points = collectPoints();
188 6 : if (points.isEmpty) {
189 2 : return [];
190 : }
191 :
192 : // Create the metric to export
193 6 : return [
194 6 : Metric.gauge(
195 6 : name: name,
196 6 : description: description,
197 6 : unit: unit,
198 : points: points,
199 : ),
200 : ];
201 : }
202 :
203 : /// Gets the current points for this gauge.
204 : /// This is used by the SDK to collect metrics.
205 6 : List<MetricPoint<T>> collectPoints() {
206 6 : if (!isEnabled()) {
207 1 : return [];
208 : }
209 :
210 : // Return points from storage
211 12 : return _storage.collectPoints();
212 : }
213 : }
214 :
215 : /// Wrapper for APICallbackRegistration that also handles our internal state.
216 : class _ObservableGaugeCallbackRegistration<T extends num>
217 : implements APICallbackRegistration<T> {
218 : /// The API registration.
219 : final APICallbackRegistration<T> apiRegistration;
220 :
221 : /// The gauge this registration is for.
222 : final ObservableGauge<T> gauge;
223 :
224 : /// The callback that was registered.
225 : final ObservableCallback<T> callback;
226 :
227 3 : _ObservableGaugeCallbackRegistration({
228 : required this.apiRegistration,
229 : required this.gauge,
230 : required this.callback,
231 : });
232 :
233 2 : @override
234 : void unregister() {
235 : // Unregister from the API implementation
236 4 : apiRegistration.unregister();
237 :
238 : // Also remove from our gauge directly for redundancy
239 6 : gauge.removeCallback(callback);
240 : }
241 : }
|
