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