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