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