Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import 'dart:async';
5 : import 'dart:math';
6 :
7 : import 'package:dartastic_opentelemetry_api/dartastic_opentelemetry_api.dart'
8 : show OTelLog;
9 : import 'package:grpc/grpc.dart';
10 :
11 : import '../../../../proto/opentelemetry_proto_dart.dart' as proto;
12 : import '../../../export/otlp_user_agent.dart';
13 : import '../../span.dart';
14 : import '../../span_logger.dart';
15 : import '../span_exporter.dart';
16 : import 'certificate_utils_io.dart';
17 : import 'otlp_grpc_span_exporter_config.dart';
18 : import 'span_transformer.dart';
19 :
20 : /// An OpenTelemetry span exporter that exports spans using OTLP over gRPC.
21 : ///
22 : /// This exporter sends trace data to an OpenTelemetry collector or compatible backend
23 : /// using the OpenTelemetry Protocol (OTLP) over gRPC. It supports features such as:
24 : /// - Retrying failed exports with exponential backoff
25 : /// - Secure and insecure connections
26 : /// - Custom headers and timeouts
27 : /// - Compression
28 : class OtlpGrpcSpanExporter implements SpanExporter {
29 : static const _retryableStatusCodes = [
30 : // Note: Don't retry on deadline exceeded as it indicates a timeout
31 : StatusCode.resourceExhausted, // Maps to HTTP 429
32 : StatusCode.unavailable, // Maps to HTTP 503
33 : ];
34 :
35 : final OtlpGrpcExporterConfig _config;
36 : ClientChannel? _channel;
37 : proto.TraceServiceClient? _traceService;
38 : bool _isShutdown = false;
39 : final Random _random = Random();
40 : final List<Future<void>> _pendingExports = [];
41 :
42 : /// Creates a new OtlpGrpcSpanExporter with the specified configuration.
43 : ///
44 : /// If no configuration is provided, default values will be used.
45 : ///
46 : /// @param config Optional configuration for the exporter
47 9 : OtlpGrpcSpanExporter([OtlpGrpcExporterConfig? config])
48 1 : : _config = config ?? OtlpGrpcExporterConfig();
49 : bool _initialized = false;
50 :
51 : /// Creates channel credentials based on configuration.
52 : ///
53 : /// If insecure is true, returns insecure credentials.
54 : /// Otherwise, creates secure credentials with optional custom certificates for mTLS.
55 5 : ChannelCredentials _createChannelCredentials() {
56 10 : if (_config.insecure) {
57 : return const ChannelCredentials.insecure();
58 : }
59 :
60 : // If no custom certificates are provided, use default secure credentials
61 4 : if (_config.certificate == null &&
62 4 : _config.clientKey == null &&
63 4 : _config.clientCertificate == null) {
64 : return const ChannelCredentials.secure();
65 : }
66 :
67 : try {
68 0 : final context = CertificateUtils.createSecurityContext(
69 0 : certificate: _config.certificate,
70 0 : clientKey: _config.clientKey,
71 0 : clientCertificate: _config.clientCertificate,
72 : );
73 :
74 : if (context == null) {
75 : return const ChannelCredentials.secure();
76 : }
77 :
78 : return const ChannelCredentials.secure(
79 : certificates: null, // We're using SecurityContext instead
80 : authority: null,
81 : onBadCertificate: null,
82 : );
83 : } catch (e) {
84 0 : if (OTelLog.isError()) {
85 0 : OTelLog.error('OtlpGrpcSpanExporter: Failed to load certificates: $e');
86 : }
87 : // Fall back to default secure credentials on error
88 : return const ChannelCredentials.secure();
89 : }
90 : }
91 :
92 : /// Cleanup the gRPC channel and release resources
93 9 : Future<void> _cleanupChannel() async {
94 9 : if (_channel != null) {
95 5 : if (OTelLog.isDebug()) {
96 4 : OTelLog.debug('OtlpGrpcSpanExporter: Shutting down existing channel');
97 : }
98 :
99 : try {
100 : // First try a graceful shutdown
101 : try {
102 5 : if (OTelLog.isDebug()) {
103 4 : OTelLog.debug(
104 : 'OtlpGrpcSpanExporter: Attempting graceful channel shutdown',
105 : );
106 : }
107 10 : await _channel!.shutdown();
108 5 : await Future<void>.delayed(
109 : const Duration(milliseconds: 100),
110 : ); // Brief delay for shutdown to complete
111 : } catch (e) {
112 0 : if (OTelLog.isDebug()) {
113 0 : OTelLog.debug(
114 0 : 'OtlpGrpcSpanExporter: Error during graceful shutdown: $e',
115 : );
116 : }
117 : }
118 :
119 : // Then try to terminate to ensure cleanup
120 : try {
121 5 : if (OTelLog.isDebug()) {
122 4 : OTelLog.debug('OtlpGrpcSpanExporter: Terminating channel');
123 : }
124 15 : unawaited(_channel!.terminate());
125 5 : await Future<void>.delayed(
126 : const Duration(milliseconds: 100),
127 : ); // Brief delay for termination to complete
128 : } catch (e) {
129 0 : if (OTelLog.isDebug()) {
130 0 : OTelLog.debug(
131 0 : 'OtlpGrpcSpanExporter: Error terminating channel: $e',
132 : );
133 : }
134 : }
135 : } catch (e) {
136 0 : if (OTelLog.isError()) {
137 0 : OTelLog.error(
138 0 : 'OtlpGrpcSpanExporter: Error shutting down existing channel: $e',
139 : );
140 : }
141 : }
142 :
143 : // Set to null to allow garbage collection
144 5 : _channel = null;
145 5 : _traceService = null;
146 :
147 : // Force garbage collection if possible
148 : try {
149 : // In Dart, we can't directly force garbage collection,
150 : // but we can suggest it by setting variables to null and
151 : // creating some memory pressure
152 5 : final temp = <int>[];
153 10 : for (var i = 0; i < 1000; i++) {
154 5 : temp.add(i);
155 : }
156 5 : temp.clear();
157 : } catch (e) {
158 : // Ignore any errors
159 : }
160 : }
161 : }
162 :
163 5 : Future<void> _setupChannel() async {
164 5 : if (_isShutdown) {
165 0 : if (OTelLog.isDebug()) {
166 0 : OTelLog.debug(
167 : 'OtlpGrpcSpanExporter: Not setting up channel - exporter is shut down',
168 : );
169 : }
170 : return;
171 : }
172 :
173 5 : if (OTelLog.isDebug()) {
174 4 : OTelLog.debug(
175 12 : 'OtlpGrpcSpanExporter: Setting up gRPC channel with endpoint ${_config.endpoint}',
176 : );
177 : }
178 :
179 : // First, clean up any existing channel
180 5 : await _cleanupChannel();
181 :
182 : String host;
183 : int port;
184 :
185 : try {
186 20 : final endpoint = _config.endpoint.trim().replaceAll(
187 5 : RegExp(r'^(http://|https://)'),
188 : '',
189 : );
190 5 : final parts = endpoint.split(':');
191 15 : host = parts[0].isEmpty ? '127.0.0.1' : parts[0];
192 20 : port = parts.length > 1 ? int.parse(parts[1]) : 4317;
193 :
194 : // Replace localhost with 127.0.0.1 for more reliable connections
195 5 : if (host == 'localhost') {
196 : host = '127.0.0.1';
197 : }
198 :
199 5 : if (OTelLog.isDebug()) {
200 4 : OTelLog.debug(
201 4 : 'OtlpGrpcSpanExporter: Setting up gRPC channel to $host:$port',
202 : );
203 : }
204 :
205 : // Create a channel
206 10 : _channel ??= ClientChannel(
207 : host,
208 : port: port,
209 5 : options: ChannelOptions(
210 5 : credentials: _createChannelCredentials(),
211 : userAgent: otlpUserAgent,
212 : connectTimeout: const Duration(seconds: 5),
213 : // Keep connection alive better
214 : idleTimeout: const Duration(seconds: 30),
215 5 : codecRegistry: CodecRegistry(
216 : codecs: const [GzipCodec(), IdentityCodec()],
217 : ),
218 : ),
219 : );
220 :
221 : try {
222 15 : _traceService = proto.TraceServiceClient(_channel!);
223 5 : if (OTelLog.isDebug()) {
224 4 : OTelLog.debug(
225 : 'OtlpGrpcSpanExporter: Successfully created TraceServiceClient',
226 : );
227 : }
228 : } catch (e) {
229 0 : if (OTelLog.isError()) {
230 0 : OTelLog.error(
231 0 : 'OtlpGrpcSpanExporter: Failed to create TraceServiceClient: $e',
232 : );
233 : }
234 : rethrow;
235 : }
236 5 : if (OTelLog.isDebug()) {
237 4 : OTelLog.debug(
238 : 'OtlpGrpcSpanExporter: Successfully created gRPC channel and trace service',
239 : );
240 : }
241 : } catch (e, stackTrace) {
242 0 : if (OTelLog.isError()) {
243 0 : OTelLog.error(
244 0 : 'OtlpGrpcSpanExporter: Failed to setup gRPC channel: $e',
245 : );
246 : }
247 0 : if (OTelLog.isError()) OTelLog.error('Stack trace: $stackTrace');
248 : rethrow;
249 : }
250 : }
251 :
252 5 : Future<void> _ensureChannel() async {
253 5 : if (_isShutdown) {
254 0 : if (OTelLog.isDebug()) {
255 0 : OTelLog.debug(
256 : 'OtlpGrpcSpanExporter: Not ensuring channel - exporter is shut down',
257 : );
258 : }
259 0 : throw StateError('Exporter is shutdown');
260 : }
261 :
262 8 : if (_initialized && _channel != null && _traceService != null) {
263 : return;
264 : }
265 :
266 5 : _initialized = true;
267 8 : if (_channel == null || _traceService == null) {
268 5 : await _setupChannel();
269 : }
270 : }
271 :
272 3 : Duration _calculateJitteredDelay(int retries) {
273 9 : final baseMs = _config.baseDelay.inMilliseconds;
274 6 : final delay = baseMs * pow(2, retries);
275 9 : final jitter = _random.nextDouble() * delay;
276 9 : return Duration(milliseconds: (delay + jitter).toInt());
277 : }
278 :
279 5 : Future<void> _tryExport(List<Span> spans) async {
280 5 : await _ensureChannel();
281 5 : if (_isShutdown) {
282 0 : throw StateError('Exporter is shutdown');
283 : }
284 5 : if (OTelLog.isLogSpans()) {
285 3 : logSpans(spans, 'Exporting spans.');
286 : }
287 :
288 5 : if (OTelLog.isDebug()) {
289 4 : OTelLog.debug(
290 8 : 'OtlpGrpcSpanExporter: Preparing to export ${spans.length} spans',
291 : );
292 8 : for (var span in spans) {
293 4 : OTelLog.debug(
294 24 : ' Span: ${span.name}, spanId: ${span.spanContext.spanId}, traceId: ${span.spanContext.traceId}',
295 : );
296 : }
297 : }
298 :
299 5 : if (OTelLog.isDebug()) {
300 12 : OTelLog.debug('OtlpGrpcSpanExporter: Transforming ${spans.length} spans');
301 : }
302 5 : final request = OtlpSpanTransformer.transformSpans(spans);
303 5 : if (OTelLog.isDebug()) {
304 4 : OTelLog.debug('OtlpGrpcSpanExporter: Successfully transformed spans');
305 : }
306 :
307 5 : if (OTelLog.isDebug()) {
308 8 : for (var rs in request.resourceSpans) {
309 4 : OTelLog.debug(' ResourceSpan:');
310 4 : if (rs.hasResource()) {
311 4 : OTelLog.debug(' Resource attributes:');
312 12 : for (var attr in rs.resource.attributes) {
313 16 : OTelLog.debug(' ${attr.key}: ${attr.value}');
314 : }
315 : }
316 8 : for (var ss in rs.scopeSpans) {
317 4 : OTelLog.debug(' ScopeSpan:');
318 8 : for (var span in ss.spans) {
319 12 : OTelLog.debug(' Span: ${span.name}');
320 12 : OTelLog.debug(' TraceId: ${span.traceId}');
321 12 : OTelLog.debug(' SpanId: ${span.spanId}');
322 : }
323 : }
324 : }
325 : }
326 :
327 : // Add compression header if configured
328 15 : final headers = Map<String, String>.from(_config.headers);
329 10 : if (_config.compression) {
330 1 : headers['grpc-encoding'] = 'gzip';
331 : }
332 :
333 5 : final options = CallOptions(
334 10 : timeout: _config.timeout,
335 : metadata: headers,
336 : );
337 :
338 5 : if (OTelLog.isDebug()) {
339 4 : OTelLog.debug(
340 12 : 'OtlpGrpcSpanExporter: Sending export request to ${_config.endpoint}',
341 : );
342 : }
343 : try {
344 5 : if (_traceService == null) {
345 0 : throw StateError(
346 : 'Trace service is null, channel may not be properly initialized',
347 : );
348 : }
349 :
350 : // gRPC's CallOptions deadline (set above) can fail to terminate a call
351 : // when the connection is torn down mid-flight — e.g. a half-open or
352 : // broken collector that accepts then drops the TCP connection — leaving
353 : // this await hung well past _config.timeout. This Dart-level backstop
354 : // guarantees export() is bounded by the configured timeout regardless of
355 : // the gRPC client's connection-state behavior. On timeout we tear the
356 : // channel down so the leaked RPC is aborted and the next attempt (if any)
357 : // reconnects cleanly.
358 : final response =
359 15 : await _traceService!.export(request, options: options).timeout(
360 10 : _config.timeout,
361 0 : onTimeout: () {
362 0 : unawaited(_cleanupChannel());
363 0 : throw TimeoutException(
364 : 'OtlpGrpcSpanExporter: export exceeded configured timeout',
365 0 : _config.timeout,
366 : );
367 : },
368 : );
369 4 : if (OTelLog.isDebug()) {
370 3 : OTelLog.debug(
371 : 'OtlpGrpcSpanExporter: Export request completed successfully',
372 : );
373 : }
374 4 : if (OTelLog.isDebug()) {
375 6 : OTelLog.debug('OtlpGrpcSpanExporter: Response: $response');
376 : }
377 : } catch (e, stackTrace) {
378 3 : if (OTelLog.isError()) {
379 6 : OTelLog.error('OtlpGrpcSpanExporter: Export request failed: $e');
380 6 : OTelLog.error('Stack trace: $stackTrace');
381 : }
382 :
383 : // If we have a channel error, try to recreate it
384 3 : if (e is GrpcError &&
385 6 : (e.code == StatusCode.unavailable ||
386 6 : e.code == StatusCode.unknown ||
387 4 : e.code == StatusCode.internal)) {
388 3 : if (OTelLog.isDebug()) {
389 3 : OTelLog.debug(
390 : 'OtlpGrpcSpanExporter: Channel error detected, recreating channel',
391 : );
392 : }
393 : // Force channel recreation
394 3 : await _cleanupChannel();
395 3 : _initialized = false;
396 : }
397 :
398 : rethrow;
399 : }
400 : }
401 :
402 6 : @override
403 : Future<void> export(List<Span> spans) async {
404 6 : if (_isShutdown) {
405 4 : throw StateError('Exporter is shutdown');
406 : }
407 :
408 6 : if (spans.isEmpty) {
409 3 : if (OTelLog.isDebug()) {
410 3 : OTelLog.debug('OtlpGrpcSpanExporter: No spans to export');
411 : }
412 : return;
413 : }
414 :
415 5 : if (OTelLog.isDebug()) {
416 4 : OTelLog.debug(
417 8 : 'OtlpGrpcSpanExporter: Beginning export of ${spans.length} spans',
418 : );
419 : }
420 5 : final exportFuture = _export(spans);
421 :
422 : // Track the pending export but don't throw if it fails during shutdown
423 10 : _pendingExports.add(exportFuture);
424 : try {
425 : await exportFuture;
426 4 : if (OTelLog.isDebug()) {
427 3 : OTelLog.debug('OtlpGrpcSpanExporter: Export completed successfully');
428 : }
429 : } catch (e) {
430 3 : if (_isShutdown &&
431 1 : e is StateError &&
432 2 : e.message.contains('shut down during')) {
433 : // Gracefully handle the case where shutdown interrupted the export
434 1 : if (OTelLog.isDebug()) {
435 1 : OTelLog.debug(
436 : 'OtlpGrpcSpanExporter: Export was interrupted by shutdown, suppressing error',
437 : );
438 : }
439 : } else {
440 : // Re-throw other errors
441 : rethrow;
442 : }
443 : } finally {
444 10 : _pendingExports.remove(exportFuture);
445 : }
446 : }
447 :
448 5 : Future<void> _export(List<Span> spans) async {
449 5 : if (_isShutdown) {
450 0 : throw StateError('Exporter was shut down during export');
451 : }
452 :
453 5 : if (OTelLog.isDebug()) {
454 4 : OTelLog.debug(
455 16 : 'OtlpGrpcSpanExporter: Attempting to export ${spans.length} spans to ${_config.endpoint}',
456 : );
457 : }
458 :
459 : var attempts = 0;
460 15 : final maxAttempts = _config.maxRetries + 1; // Initial attempt + retries
461 :
462 5 : while (attempts < maxAttempts) {
463 : // Allow the export to continue even during shutdown, so we complete in-flight requests
464 5 : final wasShutdownDuringRetry = _isShutdown;
465 :
466 : try {
467 : // Only check for shutdown on retry attempts to ensure in-progress exports can complete
468 1 : if (wasShutdownDuringRetry && attempts > 0) {
469 1 : if (OTelLog.isDebug()) {
470 1 : OTelLog.debug(
471 : 'OtlpGrpcSpanExporter: Export interrupted by shutdown',
472 : );
473 : }
474 1 : throw StateError('Exporter was shut down during export');
475 : }
476 :
477 5 : await _tryExport(spans);
478 4 : if (OTelLog.isDebug()) {
479 3 : OTelLog.debug('OtlpGrpcSpanExporter: Successfully exported spans');
480 : }
481 : return;
482 3 : } on GrpcError catch (e, stackTrace) {
483 3 : if (OTelLog.isError()) {
484 3 : OTelLog.error(
485 9 : 'OtlpGrpcSpanExporter: gRPC error during export: ${e.code} - ${e.message}',
486 : );
487 : }
488 9 : if (OTelLog.isError()) OTelLog.error('Stack trace: $stackTrace');
489 :
490 : // Check if the exporter was shut down while we were waiting
491 : if (wasShutdownDuringRetry) {
492 0 : if (OTelLog.isError()) {
493 0 : OTelLog.error(
494 : 'OtlpGrpcSpanExporter: Export interrupted by shutdown',
495 : );
496 : }
497 0 : throw StateError('Exporter was shut down during export');
498 : }
499 :
500 6 : if (!_retryableStatusCodes.contains(e.code)) {
501 3 : if (OTelLog.isError()) {
502 3 : OTelLog.error(
503 6 : 'OtlpGrpcSpanExporter: Non-retryable gRPC error (${e.code}), stopping retry attempts',
504 : );
505 : }
506 : rethrow;
507 : }
508 :
509 6 : if (attempts >= maxAttempts - 1) {
510 3 : if (OTelLog.isError()) {
511 3 : OTelLog.error(
512 3 : 'OtlpGrpcSpanExporter: Max attempts reached ($attempts out of $maxAttempts), giving up',
513 : );
514 : }
515 : rethrow;
516 : }
517 :
518 3 : final delay = _calculateJitteredDelay(attempts);
519 3 : if (OTelLog.isDebug()) {
520 3 : OTelLog.debug(
521 6 : 'OtlpGrpcSpanExporter: Retrying export after ${delay.inMilliseconds}ms...',
522 : );
523 : }
524 3 : await Future<void>.delayed(delay);
525 3 : if (!_isShutdown) {
526 : // Only recreate channel if not shut down
527 3 : await _setupChannel();
528 : }
529 3 : attempts++;
530 : } catch (e, stackTrace) {
531 1 : if (OTelLog.isError()) {
532 1 : OTelLog.error(
533 1 : 'OtlpGrpcSpanExporter: Unexpected error during export: $e',
534 : );
535 : }
536 3 : if (OTelLog.isError()) OTelLog.error('Stack trace: $stackTrace');
537 :
538 : // Check if we should stop retrying due to shutdown
539 : if (wasShutdownDuringRetry) {
540 1 : throw StateError('Exporter was shut down during export');
541 : }
542 :
543 0 : if (attempts >= maxAttempts - 1) {
544 : rethrow;
545 : }
546 :
547 0 : final delay = _calculateJitteredDelay(attempts);
548 0 : if (OTelLog.isDebug()) {
549 0 : OTelLog.debug(
550 0 : 'OtlpGrpcSpanExporter: Retrying export after ${delay.inMilliseconds}ms...',
551 : );
552 : }
553 0 : await Future<void>.delayed(delay);
554 0 : if (!_isShutdown) {
555 : // Only recreate channel if not shut down
556 0 : await _setupChannel();
557 : }
558 0 : attempts++;
559 : }
560 : }
561 : }
562 :
563 : /// Force flush any pending spans
564 5 : @override
565 : Future<void> forceFlush() async {
566 5 : if (OTelLog.isDebug()) {
567 5 : OTelLog.debug('OtlpGrpcSpanExporter: Force flush requested');
568 : }
569 5 : if (_isShutdown) {
570 3 : if (OTelLog.isDebug()) {
571 3 : OTelLog.debug(
572 : 'OtlpGrpcSpanExporter: Exporter is already shut down, nothing to flush',
573 : );
574 : }
575 : return;
576 : }
577 :
578 : // Wait for any pending export operations to complete
579 8 : if (_pendingExports.isNotEmpty) {
580 1 : if (OTelLog.isDebug()) {
581 1 : OTelLog.debug(
582 3 : 'OtlpGrpcSpanExporter: Waiting for ${_pendingExports.length} pending exports to complete',
583 : );
584 : }
585 : try {
586 2 : await Future.wait(_pendingExports);
587 1 : if (OTelLog.isDebug()) {
588 1 : OTelLog.debug('OtlpGrpcSpanExporter: All pending exports completed');
589 : }
590 : } catch (e) {
591 1 : if (OTelLog.isError()) {
592 2 : OTelLog.error('OtlpGrpcSpanExporter: Error during force flush: $e');
593 : }
594 : }
595 : } else {
596 3 : if (OTelLog.isDebug()) {
597 3 : OTelLog.debug('OtlpGrpcSpanExporter: No pending exports to flush');
598 : }
599 : }
600 : }
601 :
602 9 : @override
603 : Future<void> shutdown() async {
604 9 : if (OTelLog.isDebug()) {
605 6 : OTelLog.debug('OtlpGrpcSpanExporter: Shutdown requested');
606 : }
607 9 : if (_isShutdown) {
608 : return;
609 : }
610 9 : if (OTelLog.isDebug()) {
611 6 : OTelLog.debug(
612 18 : 'OtlpGrpcSpanExporter: Shutting down - waiting for ${_pendingExports.length} pending exports',
613 : );
614 : }
615 :
616 : // Set shutdown flag first
617 9 : _isShutdown = true;
618 :
619 : // Create a safe copy of pending exports to avoid concurrent modification
620 18 : final pendingExportsCopy = List<Future<void>>.of(_pendingExports);
621 :
622 : // Wait for pending exports but don't start any new ones
623 : // Use a timeout to prevent hanging if exports take too long
624 9 : if (pendingExportsCopy.isNotEmpty) {
625 1 : if (OTelLog.isDebug()) {
626 1 : OTelLog.debug(
627 2 : 'OtlpGrpcSpanExporter: Waiting for ${pendingExportsCopy.length} pending exports with timeout',
628 : );
629 : }
630 : try {
631 : // Use a generous timeout but don't wait forever
632 2 : await Future.wait(pendingExportsCopy).timeout(
633 : const Duration(seconds: 10),
634 0 : onTimeout: () {
635 0 : if (OTelLog.isDebug()) {
636 0 : OTelLog.debug(
637 : 'OtlpGrpcSpanExporter: Timeout waiting for exports to complete',
638 : );
639 : }
640 0 : return Future.value([]);
641 : },
642 : );
643 : } catch (e) {
644 1 : if (OTelLog.isDebug()) {
645 1 : OTelLog.debug(
646 1 : 'OtlpGrpcSpanExporter: Error during shutdown while waiting for exports: $e',
647 : );
648 : }
649 : }
650 : }
651 :
652 : // Clean up channel resources
653 9 : await _cleanupChannel();
654 :
655 9 : if (OTelLog.isDebug()) {
656 6 : OTelLog.debug('OtlpGrpcSpanExporter: Shutdown complete');
657 : }
658 : }
659 : }
|
