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 :
8 : import '../../../../dartastic_opentelemetry.dart';
9 : import '../../../../proto/collector/metrics/v1/metrics_service.pbgrpc.dart';
10 : import '../../../../proto/common/v1/common.pb.dart' as common_proto;
11 : import '../../../../proto/metrics/v1/metrics.pb.dart' as proto;
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 : bool _shutdown = false;
19 :
20 : // Static channel reference to allow shutdown
21 1 : static late ClientChannel _channel;
22 :
23 : /// Creates a new OtlpGrpcMetricExporter with the given configuration.
24 1 : OtlpGrpcMetricExporter(OtlpGrpcMetricExporterConfig config)
25 1 : : _client = _createClient(config);
26 :
27 : /// Creates channel credentials based on configuration.
28 : ///
29 : /// If insecure is true, returns insecure credentials.
30 : /// Otherwise, creates secure credentials with optional custom certificates for mTLS.
31 1 : static ChannelCredentials _createChannelCredentials(
32 : OtlpGrpcMetricExporterConfig config,
33 : ) {
34 1 : if (config.insecure) {
35 : return const ChannelCredentials.insecure();
36 : }
37 :
38 : // If no custom certificates are provided, use default secure credentials
39 0 : if (config.certificate == null &&
40 0 : config.clientKey == null &&
41 0 : config.clientCertificate == null) {
42 : return const ChannelCredentials.secure();
43 : }
44 :
45 : try {
46 0 : final context = CertificateUtils.createSecurityContext(
47 0 : certificate: config.certificate,
48 0 : clientKey: config.clientKey,
49 0 : clientCertificate: config.clientCertificate,
50 : );
51 :
52 : if (context == null) {
53 : return const ChannelCredentials.secure();
54 : }
55 :
56 : return const ChannelCredentials.secure(
57 : certificates: null, // We're using SecurityContext instead
58 : authority: null,
59 : onBadCertificate: null,
60 : );
61 : } catch (e) {
62 0 : if (OTelLog.isError()) {
63 0 : OTelLog.error(
64 0 : 'OtlpGrpcMetricExporter: Failed to load certificates: $e',
65 : );
66 : }
67 : // Fall back to default secure credentials on error
68 : return const ChannelCredentials.secure();
69 : }
70 : }
71 :
72 1 : static MetricsServiceClient _createClient(
73 : OtlpGrpcMetricExporterConfig config,
74 : ) {
75 1 : final channelOptions = ChannelOptions(
76 1 : credentials: _createChannelCredentials(config),
77 1 : codecRegistry: CodecRegistry(codecs: const [GzipCodec()]),
78 : );
79 :
80 : // Parse host and port from endpoint
81 2 : final uri = Uri.parse(config.endpoint);
82 1 : final host = uri.host;
83 3 : final port = uri.port > 0 ? uri.port : (uri.scheme == 'https' ? 443 : 80);
84 :
85 1 : if (OTelLog.isLogExport()) {
86 1 : OTelLog.logExport(
87 1 : 'OtlpGrpcMetricExporter: Creating client for $host:$port',
88 : );
89 : }
90 :
91 : // We store the channel separately to be able to shut it down later
92 1 : _channel = ClientChannel(host, port: port, options: channelOptions);
93 :
94 : // Build call options with headers and compression
95 1 : final callOptionsBuilder = CallOptions(
96 2 : timeout: Duration(milliseconds: config.timeoutMillis),
97 : );
98 :
99 : // Add custom headers if provided
100 1 : final metadata = <String, String>{};
101 1 : if (config.headers != null) {
102 2 : metadata.addAll(config.headers!);
103 : }
104 :
105 : // Add compression header if enabled
106 1 : if (config.compression) {
107 1 : metadata['grpc-encoding'] = 'gzip';
108 : }
109 :
110 1 : return MetricsServiceClient(
111 1 : _channel,
112 1 : options: metadata.isNotEmpty
113 2 : ? callOptionsBuilder.mergedWith(CallOptions(metadata: metadata))
114 : : callOptionsBuilder,
115 : );
116 : }
117 :
118 1 : @override
119 : Future<bool> export(MetricData data) async {
120 1 : if (_shutdown) {
121 1 : if (OTelLog.isLogExport()) {
122 1 : OTelLog.logExport(
123 : 'OtlpGrpcMetricExporter: Cannot export after shutdown',
124 : );
125 : }
126 : return false;
127 : }
128 :
129 2 : if (data.metrics.isEmpty) {
130 1 : if (OTelLog.isLogExport()) {
131 1 : OTelLog.logExport('OtlpGrpcMetricExporter: No metrics to export');
132 : }
133 : return true;
134 : }
135 :
136 : try {
137 1 : if (OTelLog.isLogExport()) {
138 1 : OTelLog.logExport(
139 3 : 'OtlpGrpcMetricExporter: Exporting ${data.metrics.length} metrics',
140 : );
141 2 : for (final metric in data.metrics) {
142 1 : OTelLog.logExport(
143 5 : ' - ${metric.name} (${metric.type}): ${metric.points.length} data points',
144 : );
145 : }
146 : }
147 :
148 : // Transform metrics data to protocol buffers
149 1 : final request = _buildExportRequest(data);
150 :
151 : // Export to the collector
152 2 : await _client.export(request);
153 :
154 1 : if (OTelLog.isLogExport()) {
155 1 : OTelLog.logExport('OtlpGrpcMetricExporter: Export successful');
156 : }
157 : return true;
158 : } catch (e, stackTrace) {
159 1 : if (OTelLog.isLogExport()) {
160 2 : OTelLog.logExport('OtlpGrpcMetricExporter: Export failed: $e');
161 2 : OTelLog.logExport('Stack trace: $stackTrace');
162 : }
163 : return false;
164 : }
165 : }
166 :
167 : /// Builds the export request from the given metrics data.
168 1 : ExportMetricsServiceRequest _buildExportRequest(MetricData data) {
169 1 : final request = ExportMetricsServiceRequest();
170 1 : final resourceMetrics = proto.ResourceMetrics();
171 :
172 : // Add resource
173 1 : if (data.resource != null) {
174 2 : resourceMetrics.resource = MetricTransformer.transformResource(
175 1 : data.resource!,
176 : );
177 : } else {
178 : // Create empty resource if none provided
179 2 : resourceMetrics.resource = MetricTransformer.transformResource(
180 1 : OTel.resource(null),
181 : );
182 : }
183 :
184 : // Add scope metrics
185 1 : final scopeMetrics = proto.ScopeMetrics();
186 2 : scopeMetrics.metrics.addAll(
187 2 : data.metrics.map(MetricTransformer.transformMetric),
188 : );
189 :
190 : // Add instrumentation scope (hardcoded for now)
191 1 : final scope = common_proto.InstrumentationScope();
192 1 : scope.name = '@dart/dartastic_opentelemetry';
193 1 : scope.version = '1.0.0';
194 1 : scopeMetrics.scope = scope;
195 :
196 2 : resourceMetrics.scopeMetrics.add(scopeMetrics);
197 2 : request.resourceMetrics.add(resourceMetrics);
198 :
199 : return request;
200 : }
201 :
202 1 : @override
203 : Future<bool> forceFlush() async {
204 : // No-op for this exporter
205 : return true;
206 : }
207 :
208 1 : @override
209 : Future<bool> shutdown() async {
210 1 : if (_shutdown) {
211 : return true;
212 : }
213 :
214 1 : _shutdown = true;
215 : try {
216 : // Close the gRPC channel
217 : // Shutdown the stored channel
218 2 : await _channel.shutdown();
219 :
220 1 : if (OTelLog.isLogExport()) {
221 1 : OTelLog.logExport('OtlpGrpcMetricExporter: Channel shutdown completed');
222 : }
223 : return true;
224 : } catch (e) {
225 0 : if (OTelLog.isLogExport()) {
226 0 : OTelLog.logExport('OtlpGrpcMetricExporter: Shutdown failed: $e');
227 : }
228 : return false;
229 : }
230 : }
231 : }
|
