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