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