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