Line data Source code
1 : // Copyright The OpenTelemetry Authors
2 : // SPDX-License-Identifier: Apache-2.0
3 :
4 : import 'dart:async';
5 :
6 : import 'package:grpc/grpc.dart';
7 : import 'package:meta/meta.dart';
8 :
9 : import '../../../../dartastic_opentelemetry.dart';
10 : import '../../../../proto/collector/metrics/v1/metrics_service.pbgrpc.dart';
11 : import '../../../export/otlp_user_agent.dart';
12 : import '../../../trace/export/otlp/certificate_utils_io.dart';
13 : import 'metric_transformer.dart';
14 :
15 : /// OtlpGrpcMetricExporter exports metrics to the OpenTelemetry collector via gRPC.
16 : class OtlpGrpcMetricExporter implements MetricExporter {
17 : final MetricsServiceClient _client;
18 :
19 : /// The configuration this exporter was created with. Exposed for
20 : /// tests asserting the resolved connection security (#253).
21 : @visibleForTesting
22 : final OtlpGrpcMetricExporterConfig config;
23 :
24 : bool _shutdown = false;
25 :
26 : // Static channel reference to allow shutdown
27 5 : static late ClientChannel _channel;
28 :
29 : /// Creates a new OtlpGrpcMetricExporter with the given configuration.
30 10 : OtlpGrpcMetricExporter(this.config) : _client = _createClient(config);
31 :
32 : /// Creates channel credentials based on configuration.
33 : ///
34 : /// If insecure is true, returns insecure credentials.
35 : /// Otherwise, creates secure credentials with optional custom certificates for mTLS.
36 5 : static ChannelCredentials _createChannelCredentials(
37 : OtlpGrpcMetricExporterConfig config,
38 : ) {
39 5 : if (config.insecure) {
40 : return const ChannelCredentials.insecure();
41 : }
42 :
43 : // If no custom certificates are provided, use default secure credentials
44 1 : if (config.certificate == null &&
45 1 : config.clientKey == null &&
46 1 : config.clientCertificate == null) {
47 : return const ChannelCredentials.secure();
48 : }
49 :
50 : try {
51 0 : final context = CertificateUtils.createSecurityContext(
52 0 : certificate: config.certificate,
53 0 : clientKey: config.clientKey,
54 0 : clientCertificate: config.clientCertificate,
55 : );
56 :
57 : if (context == null) {
58 : return const ChannelCredentials.secure();
59 : }
60 :
61 : return const ChannelCredentials.secure(
62 : certificates: null, // We're using SecurityContext instead
63 : authority: null,
64 : onBadCertificate: null,
65 : );
66 : } catch (e) {
67 0 : if (OTelLog.isError()) {
68 0 : OTelLog.error(
69 0 : 'OtlpGrpcMetricExporter: Failed to load certificates: $e',
70 : );
71 : }
72 : // Fall back to default secure credentials on error
73 : return const ChannelCredentials.secure();
74 : }
75 : }
76 :
77 5 : static MetricsServiceClient _createClient(
78 : OtlpGrpcMetricExporterConfig config,
79 : ) {
80 5 : final channelOptions = ChannelOptions(
81 5 : credentials: _createChannelCredentials(config),
82 : userAgent: otlpUserAgent,
83 5 : codecRegistry: CodecRegistry(codecs: const [GzipCodec()]),
84 : );
85 :
86 : // Parse host and port from endpoint
87 10 : final uri = Uri.parse(config.endpoint);
88 5 : final host = uri.host;
89 17 : final port = uri.port > 0 ? uri.port : (uri.scheme == 'https' ? 443 : 80);
90 :
91 5 : if (OTelLog.isLogExport()) {
92 4 : OTelLog.logExport(
93 4 : 'OtlpGrpcMetricExporter: Creating client for $host:$port',
94 : );
95 : }
96 :
97 : // We store the channel separately to be able to shut it down later
98 5 : _channel = ClientChannel(host, port: port, options: channelOptions);
99 :
100 : // Build call options with headers and compression
101 5 : final callOptionsBuilder = CallOptions(
102 10 : timeout: Duration(milliseconds: config.timeoutMillis),
103 : );
104 :
105 : // Add custom headers if provided
106 5 : final metadata = <String, String>{};
107 5 : if (config.headers != null) {
108 10 : metadata.addAll(config.headers!);
109 : }
110 :
111 : // Add compression header if enabled
112 5 : if (config.compression) {
113 1 : metadata['grpc-encoding'] = 'gzip';
114 : }
115 :
116 5 : return MetricsServiceClient(
117 5 : _channel,
118 5 : options: metadata.isNotEmpty
119 2 : ? callOptionsBuilder.mergedWith(CallOptions(metadata: metadata))
120 : : callOptionsBuilder,
121 : );
122 : }
123 :
124 1 : @override
125 : Future<bool> export(MetricData data) async {
126 1 : if (_shutdown) {
127 1 : if (OTelLog.isLogExport()) {
128 1 : OTelLog.logExport(
129 : 'OtlpGrpcMetricExporter: Cannot export after shutdown',
130 : );
131 : }
132 : return false;
133 : }
134 :
135 2 : if (data.metrics.isEmpty) {
136 1 : if (OTelLog.isLogExport()) {
137 1 : OTelLog.logExport('OtlpGrpcMetricExporter: No metrics to export');
138 : }
139 : return true;
140 : }
141 :
142 : try {
143 1 : if (OTelLog.isLogExport()) {
144 1 : OTelLog.logExport(
145 3 : 'OtlpGrpcMetricExporter: Exporting ${data.metrics.length} metrics',
146 : );
147 2 : for (final metric in data.metrics) {
148 1 : OTelLog.logExport(
149 5 : ' - ${metric.name} (${metric.type}): ${metric.points.length} data points',
150 : );
151 : }
152 : }
153 :
154 : // Transform metrics data to protocol buffers
155 1 : final request = _buildExportRequest(data);
156 :
157 : // Export to the collector
158 2 : await _client.export(request);
159 :
160 1 : if (OTelLog.isLogExport()) {
161 1 : OTelLog.logExport('OtlpGrpcMetricExporter: Export successful');
162 : }
163 : return true;
164 : } catch (e, stackTrace) {
165 1 : if (OTelLog.isLogExport()) {
166 2 : OTelLog.logExport('OtlpGrpcMetricExporter: Export failed: $e');
167 2 : OTelLog.logExport('Stack trace: $stackTrace');
168 : }
169 : return false;
170 : }
171 : }
172 :
173 : /// Builds the export request from the given metrics data — via the
174 : /// shared one-shot, which builds the same request this exporter used to
175 : /// assemble inline (same scope constant, same OTel.resource(null)
176 : /// fallback), so the wire output is unchanged.
177 1 : ExportMetricsServiceRequest _buildExportRequest(MetricData data) =>
178 1 : MetricTransformer.transformMetrics(
179 : data,
180 1 : fallbackResource: OTel.resource(null),
181 2 : exemplarFilter: config.exemplarFilter,
182 : );
183 :
184 1 : @override
185 : Future<bool> forceFlush() async {
186 : // No-op for this exporter
187 : return true;
188 : }
189 :
190 5 : @override
191 : Future<bool> shutdown() async {
192 5 : if (_shutdown) {
193 : return true;
194 : }
195 :
196 5 : _shutdown = true;
197 : try {
198 : // Close the gRPC channel
199 : // Shutdown the stored channel
200 10 : await _channel.shutdown();
201 :
202 5 : if (OTelLog.isLogExport()) {
203 4 : OTelLog.logExport('OtlpGrpcMetricExporter: Channel shutdown completed');
204 : }
205 : return true;
206 : } catch (e) {
207 0 : if (OTelLog.isLogExport()) {
208 0 : OTelLog.logExport('OtlpGrpcMetricExporter: Shutdown failed: $e');
209 : }
210 : return false;
211 : }
212 : }
213 : }
|
