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