Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import 'dart:async';
5 : import 'dart:convert';
6 : import 'dart:math';
7 : import 'dart:typed_data';
8 :
9 : import 'package:http/http.dart' as http;
10 :
11 : import '../../../../../dartastic_opentelemetry.dart';
12 : import '../../../../export/otlp_json.dart';
13 : import '../../../../export/otlp_user_agent.dart';
14 : import '../../../../trace/export/otlp/http/http_client_factory.dart';
15 : import '../../../../util/zip/gzip.dart';
16 : import '../metric_transformer.dart';
17 :
18 : /// An OpenTelemetry metric exporter that exports metrics using OTLP over HTTP/protobuf
19 : class OtlpHttpMetricExporter implements MetricExporter {
20 : static const _retryableStatusCodes = [
21 : 429, // Too Many Requests
22 : 503, // Service Unavailable
23 : ];
24 :
25 : final OtlpHttpMetricExporterConfig _config;
26 : bool _isShutdown = false;
27 : final Random _random = Random();
28 : final List<Future<void>> _pendingExports = [];
29 : late final http.Client _client;
30 :
31 : /// Creates a new OTLP HTTP metric exporter with the specified configuration.
32 : /// If no configuration is provided, default settings will be used.
33 : ///
34 : /// @param config Optional configuration for the exporter
35 129 : OtlpHttpMetricExporter([OtlpHttpMetricExporterConfig? config])
36 1 : : _config = config ?? OtlpHttpMetricExporterConfig() {
37 258 : _client = _createHttpClient();
38 : }
39 :
40 : /// Creates an HTTP client with custom certificates if configured.
41 : /// Delegated to a platform-conditional factory: native gets an
42 : /// `IOClient` wrapping an `HttpClient` with a custom `SecurityContext`;
43 : /// web gets a `BrowserClient` (the browser handles TLS).
44 258 : http.Client _createHttpClient() => createOtlpHttpClient(
45 : exporterName: 'OtlpHttpMetricExporter',
46 258 : certificate: _config.certificate,
47 258 : clientKey: _config.clientKey,
48 258 : clientCertificate: _config.clientCertificate,
49 : );
50 :
51 4 : Duration _calculateJitteredDelay(int retries) {
52 12 : final baseMs = _config.baseDelay.inMilliseconds;
53 8 : final delay = baseMs * pow(2, retries);
54 12 : final jitter = _random.nextDouble() * delay;
55 12 : return Duration(milliseconds: (delay + jitter).toInt());
56 : }
57 :
58 8 : String _getEndpointUrl() {
59 : // Ensure the endpoint ends with /v1/metrics
60 16 : var endpoint = _config.endpoint;
61 8 : if (!endpoint.endsWith('/v1/metrics')) {
62 : // Ensure there's no trailing slash before adding path
63 8 : if (endpoint.endsWith('/')) {
64 6 : endpoint = endpoint.substring(0, endpoint.length - 1);
65 : }
66 8 : endpoint = '$endpoint/v1/metrics';
67 : }
68 : return endpoint;
69 : }
70 :
71 9 : @override
72 : Future<bool> export(MetricData metrics) async {
73 9 : if (_isShutdown) {
74 6 : throw StateError('Exporter is shutdown');
75 : }
76 :
77 18 : if (metrics.metrics.isEmpty) {
78 4 : if (OTelLog.isDebug()) {
79 4 : OTelLog.debug('OtlpHttpMetricExporter: No metrics to export');
80 : }
81 : return true;
82 : }
83 :
84 8 : if (OTelLog.isDebug()) {
85 7 : OTelLog.debug(
86 21 : 'OtlpHttpMetricExporter: Beginning export of ${metrics.metrics.length} metrics',
87 : );
88 : }
89 :
90 8 : final exportFuture = _export(metrics);
91 :
92 : // Register before awaiting: forceFlush() and shutdown() both drain
93 : // _pendingExports, so an export that is not in the list is invisible to
94 : // them and they return while it is still in flight.
95 16 : _pendingExports.add(exportFuture);
96 : try {
97 : final result = await exportFuture;
98 8 : if (OTelLog.isDebug()) {
99 7 : OTelLog.debug('OtlpHttpMetricExporter: Export completed successfully');
100 : }
101 : return result;
102 : } catch (e) {
103 2 : if (_isShutdown &&
104 2 : e is StateError &&
105 4 : e.message.contains('shut down during')) {
106 : // Gracefully handle the case where shutdown interrupted the export
107 2 : if (OTelLog.isDebug()) {
108 2 : OTelLog.debug(
109 : 'OtlpHttpMetricExporter: Export was interrupted by shutdown, suppressing error',
110 : );
111 : }
112 : return false;
113 : } else {
114 : // Re-throw other errors
115 : rethrow;
116 : }
117 : } finally {
118 16 : _pendingExports.remove(exportFuture);
119 : }
120 : }
121 :
122 8 : Future<bool> _export(MetricData metrics) async {
123 8 : if (_isShutdown) {
124 0 : throw StateError('Exporter was shut down during export');
125 : }
126 :
127 8 : if (OTelLog.isDebug()) {
128 7 : OTelLog.debug(
129 35 : 'OtlpHttpMetricExporter: Attempting to export ${metrics.metrics.length} metrics to ${_config.endpoint}',
130 : );
131 : }
132 :
133 : var attempts = 0;
134 24 : final maxAttempts = _config.maxRetries + 1; // Initial attempt + retries
135 :
136 8 : while (attempts < maxAttempts) {
137 : // Allow the export to continue even during shutdown, so we complete in-flight requests
138 8 : final wasShutdownDuringRetry = _isShutdown;
139 :
140 : try {
141 : // Only check for shutdown on retry attempts to ensure in-progress exports can complete
142 2 : if (wasShutdownDuringRetry && attempts > 0) {
143 2 : if (OTelLog.isDebug()) {
144 2 : OTelLog.debug(
145 : 'OtlpHttpMetricExporter: Export interrupted by shutdown',
146 : );
147 : }
148 2 : throw StateError('Exporter was shut down during export');
149 : }
150 :
151 8 : final success = await _tryExport(metrics);
152 6 : if (OTelLog.isDebug()) {
153 5 : OTelLog.debug(
154 : 'OtlpHttpMetricExporter: Successfully exported metrics',
155 : );
156 : }
157 : return success;
158 7 : } on http.ClientException catch (e, stackTrace) {
159 7 : if (OTelLog.isError()) {
160 14 : OTelLog.error('OtlpHttpMetricExporter: HTTP error during export: $e');
161 : }
162 21 : if (OTelLog.isError()) OTelLog.error('Stack trace: $stackTrace');
163 :
164 : // Check if the exporter was shut down while we were waiting
165 : if (wasShutdownDuringRetry) {
166 0 : if (OTelLog.isError()) {
167 0 : OTelLog.error(
168 : 'OtlpHttpMetricExporter: Export interrupted by shutdown',
169 : );
170 : }
171 0 : throw StateError('Exporter was shut down during export');
172 : }
173 :
174 : // Handle status code-based retries
175 : var shouldRetry = false;
176 14 : if (e.message.contains('status code')) {
177 8 : for (final code in _retryableStatusCodes) {
178 12 : if (e.message.contains('status code $code')) {
179 : shouldRetry = true;
180 : break;
181 : }
182 : }
183 : }
184 :
185 : if (!shouldRetry) {
186 6 : if (OTelLog.isError()) {
187 6 : OTelLog.error(
188 : 'OtlpHttpMetricExporter: Non-retryable HTTP error, stopping retry attempts',
189 : );
190 : }
191 : return false;
192 : }
193 :
194 8 : if (attempts >= maxAttempts - 1) {
195 2 : if (OTelLog.isError()) {
196 2 : OTelLog.error(
197 2 : 'OtlpHttpMetricExporter: Max attempts reached ($attempts out of $maxAttempts), giving up',
198 : );
199 : }
200 : return false;
201 : }
202 :
203 4 : final delay = _calculateJitteredDelay(attempts);
204 4 : if (OTelLog.isDebug()) {
205 4 : OTelLog.debug(
206 8 : 'OtlpHttpMetricExporter: Retrying export after ${delay.inMilliseconds}ms...',
207 : );
208 : }
209 4 : await Future<void>.delayed(delay);
210 4 : attempts++;
211 : } catch (e, stackTrace) {
212 2 : if (OTelLog.isError()) {
213 2 : OTelLog.error(
214 2 : 'OtlpHttpMetricExporter: Unexpected error during export: $e',
215 : );
216 : }
217 6 : if (OTelLog.isError()) OTelLog.error('Stack trace: $stackTrace');
218 :
219 : // Check if we should stop retrying due to shutdown
220 : if (wasShutdownDuringRetry) {
221 2 : throw StateError('Exporter was shut down during export');
222 : }
223 :
224 0 : if (attempts >= maxAttempts - 1) {
225 : return false;
226 : }
227 :
228 0 : final delay = _calculateJitteredDelay(attempts);
229 0 : if (OTelLog.isDebug()) {
230 0 : OTelLog.debug(
231 0 : 'OtlpHttpMetricExporter: Retrying export after ${delay.inMilliseconds}ms...',
232 : );
233 : }
234 0 : await Future<void>.delayed(delay);
235 0 : attempts++;
236 : }
237 : }
238 :
239 : return false;
240 : }
241 :
242 8 : Future<bool> _tryExport(MetricData metrics) async {
243 8 : if (_isShutdown) {
244 0 : throw StateError('Exporter is shutdown');
245 : }
246 :
247 8 : if (OTelLog.isLogMetrics()) {
248 7 : OTelLog.logMetric(
249 21 : 'Exporting metrics via HTTP: ${metrics.metrics.length} metrics',
250 : );
251 : }
252 :
253 8 : if (OTelLog.isDebug()) {
254 7 : OTelLog.debug(
255 21 : 'OtlpHttpMetricExporter: Preparing to export ${metrics.metrics.length} metrics',
256 : );
257 : }
258 :
259 8 : if (OTelLog.isDebug()) {
260 7 : OTelLog.debug('OtlpHttpMetricExporter: Transforming metrics');
261 : }
262 :
263 : // Create the export request — the shared one-shot builds the same
264 : // request this exporter used to assemble inline (same scope constant,
265 : // same OTel.resource(null) fallback), so the wire output is unchanged.
266 8 : final request = MetricTransformer.transformMetrics(
267 : metrics,
268 8 : fallbackResource: OTel.resource(null),
269 16 : exemplarFilter: _config.exemplarFilter,
270 : );
271 :
272 8 : if (OTelLog.isDebug()) {
273 7 : OTelLog.debug('OtlpHttpMetricExporter: Successfully transformed metrics');
274 : }
275 :
276 : // Prepare headers + body. Wire format is selected by config.protocol —
277 : // protobuf (default) or JSON via proto3-JSON mapping. See
278 : // `OtlpHttpProtocol` for the conformance rationale.
279 24 : final headers = Map<String, String>.from(_config.headers);
280 : // Default User-Agent per the OTLP exporter spec ("User agent"): the
281 : // exporter's default string is always present. A caller-supplied value
282 : // (typically a distribution identifier) is prepended to it, e.g.
283 : // "MyDistribution/1.0 OTel-OTLP-Exporter-Dart/1.1.0-beta.14-wip".
284 : // `headers` is a copy, and http's header map is case-insensitive, so the
285 : // User-Agent assignment below overrides the copied user-agent entry.
286 24 : final userAgent = _config.headers['user-agent'];
287 8 : headers['User-Agent'] =
288 1 : userAgent == null ? otlpUserAgent : '$userAgent $otlpUserAgent';
289 : Uint8List messageBytes;
290 24 : if (_config.protocol == OtlpHttpProtocol.httpJson) {
291 1 : headers['Content-Type'] = 'application/json';
292 1 : final jsonValue = otlpProto3JsonWithHexIds(request);
293 3 : messageBytes = Uint8List.fromList(utf8.encode(jsonEncode(jsonValue)));
294 : } else {
295 8 : headers['Content-Type'] = 'application/x-protobuf';
296 8 : messageBytes = request.writeToBuffer();
297 : }
298 :
299 16 : if (_config.compression) {
300 4 : headers['Content-Encoding'] = 'gzip';
301 : }
302 :
303 : var bodyBytes = messageBytes;
304 :
305 : // Apply gzip compression if configured
306 16 : if (_config.compression) {
307 4 : final gzip = GZip();
308 4 : final compressedBytes = await gzip.compress(messageBytes);
309 4 : bodyBytes = Uint8List.fromList(compressedBytes);
310 : }
311 :
312 : // Get the endpoint URL with the correct path
313 8 : final endpointUrl = _getEndpointUrl();
314 8 : if (OTelLog.isDebug()) {
315 7 : OTelLog.debug(
316 7 : 'OtlpHttpMetricExporter: Sending export request to $endpointUrl',
317 : );
318 : }
319 :
320 : try {
321 8 : final response = await _client
322 16 : .post(Uri.parse(endpointUrl), headers: headers, body: bodyBytes)
323 24 : .timeout(_config.timeout);
324 :
325 24 : if (response.statusCode >= 200 && response.statusCode < 300) {
326 6 : if (OTelLog.isDebug()) {
327 5 : OTelLog.debug(
328 : 'OtlpHttpMetricExporter: Export request completed successfully',
329 : );
330 : }
331 : return true;
332 : } else {
333 : final errorMessage =
334 8 : 'OtlpHttpMetricExporter: Export request failed with status code ${response.statusCode}';
335 8 : if (OTelLog.isError()) OTelLog.error(errorMessage);
336 4 : throw http.ClientException(errorMessage);
337 : }
338 7 : } on http.ClientException {
339 : // Let ClientException propagate to _export for retry handling
340 : rethrow;
341 : } catch (e, stackTrace) {
342 0 : if (OTelLog.isError()) {
343 0 : OTelLog.error('OtlpHttpMetricExporter: Export request failed: $e');
344 0 : OTelLog.error('Stack trace: $stackTrace');
345 : }
346 : return false;
347 : }
348 : }
349 :
350 8 : @override
351 : Future<bool> forceFlush() async {
352 8 : if (OTelLog.isDebug()) {
353 7 : OTelLog.debug('OtlpHttpMetricExporter: Force flush requested');
354 : }
355 8 : if (_isShutdown) {
356 4 : if (OTelLog.isDebug()) {
357 4 : OTelLog.debug(
358 : 'OtlpHttpMetricExporter: Exporter is already shut down, nothing to flush',
359 : );
360 : }
361 : return true;
362 : }
363 :
364 : // Wait for any pending export operations to complete
365 16 : if (_pendingExports.isNotEmpty) {
366 3 : if (OTelLog.isDebug()) {
367 3 : OTelLog.debug(
368 9 : 'OtlpHttpMetricExporter: Waiting for ${_pendingExports.length} pending exports to complete',
369 : );
370 : }
371 : try {
372 6 : await Future.wait(_pendingExports);
373 3 : if (OTelLog.isDebug()) {
374 3 : OTelLog.debug(
375 : 'OtlpHttpMetricExporter: All pending exports completed',
376 : );
377 : }
378 : return true;
379 : } catch (e) {
380 0 : if (OTelLog.isError()) {
381 0 : OTelLog.error('OtlpHttpMetricExporter: Error during force flush: $e');
382 : }
383 : return false;
384 : }
385 : } else {
386 6 : if (OTelLog.isDebug()) {
387 5 : OTelLog.debug('OtlpHttpMetricExporter: No pending exports to flush');
388 : }
389 : return true;
390 : }
391 : }
392 :
393 114 : @override
394 : Future<bool> shutdown() async {
395 114 : if (OTelLog.isDebug()) {
396 109 : OTelLog.debug('OtlpHttpMetricExporter: Shutdown requested');
397 : }
398 114 : if (_isShutdown) {
399 : return true;
400 : }
401 114 : if (OTelLog.isDebug()) {
402 109 : OTelLog.debug(
403 327 : 'OtlpHttpMetricExporter: Shutting down - waiting for ${_pendingExports.length} pending exports',
404 : );
405 : }
406 :
407 : // Set shutdown flag first
408 114 : _isShutdown = true;
409 :
410 : // Create a safe copy of pending exports to avoid concurrent modification
411 228 : final pendingExportsCopy = List<Future<void>>.of(_pendingExports);
412 :
413 : // Wait for pending exports but don't start any new ones
414 : // Use a timeout to prevent hanging if exports take too long
415 114 : if (pendingExportsCopy.isNotEmpty) {
416 4 : if (OTelLog.isDebug()) {
417 4 : OTelLog.debug(
418 8 : 'OtlpHttpMetricExporter: Waiting for ${pendingExportsCopy.length} pending exports with timeout',
419 : );
420 : }
421 : try {
422 : // Use a generous timeout but don't wait forever
423 8 : await Future.wait(pendingExportsCopy).timeout(
424 : const Duration(seconds: 10),
425 0 : onTimeout: () {
426 0 : if (OTelLog.isDebug()) {
427 0 : OTelLog.debug(
428 : 'OtlpHttpMetricExporter: Timeout waiting for exports to complete',
429 : );
430 : }
431 0 : return Future.value([]);
432 : },
433 : );
434 : } catch (e) {
435 2 : if (OTelLog.isDebug()) {
436 2 : OTelLog.debug(
437 2 : 'OtlpHttpMetricExporter: Error during shutdown while waiting for exports: $e',
438 : );
439 : }
440 : // Don't return false here - we still want to close the client
441 : }
442 : }
443 :
444 : // Close the HTTP client to release resources
445 228 : _client.close();
446 :
447 114 : if (OTelLog.isDebug()) {
448 109 : OTelLog.debug('OtlpHttpMetricExporter: Shutdown complete');
449 : }
450 : return true;
451 : }
452 : }
|
