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