@@ -18,6 +18,7 @@ use std::sync::OnceLock;
1818
1919use base64:: Engine ;
2020use base64:: prelude:: BASE64_STANDARD_NO_PAD ;
21+ use ginepro:: LoadBalancedChannel ;
2122use hyper:: http:: Response ;
2223use nativelink_error:: { Code , ResultExt , make_err} ;
2324use nativelink_proto:: build:: bazel:: remote:: execution:: v2:: RequestMetadata ;
@@ -26,7 +27,9 @@ use opentelemetry::trace::{TraceContextExt, Tracer, TracerProvider};
2627use opentelemetry:: { KeyValue , global} ;
2728use opentelemetry_appender_tracing:: layer:: OpenTelemetryTracingBridge ;
2829use opentelemetry_http:: HeaderExtractor ;
29- use opentelemetry_otlp:: { LogExporter , MetricExporter , Protocol , SpanExporter , WithExportConfig } ;
30+ use opentelemetry_otlp:: {
31+ LogExporter , MetricExporter , Protocol , SpanExporter , WithExportConfig , WithTonicConfig ,
32+ } ;
3033use opentelemetry_sdk:: Resource ;
3134use opentelemetry_sdk:: logs:: SdkLoggerProvider ;
3235use opentelemetry_sdk:: metrics:: SdkMeterProvider ;
@@ -103,7 +106,7 @@ fn tracing_stdout_layer() -> impl Layer<Registry> {
103106///
104107/// Returns `Err` if logging was already initialized or if the exporters can't
105108/// be initialized.
106- pub fn init_tracing ( ) -> Result < ( ) , nativelink_error:: Error > {
109+ pub async fn init_tracing ( ) -> Result < ( ) , nativelink_error:: Error > {
107110 static INITIALIZED : OnceLock < ( ) > = OnceLock :: new ( ) ;
108111
109112 if INITIALIZED . get ( ) . is_some ( ) {
@@ -128,13 +131,18 @@ pub fn init_tracing() -> Result<(), nativelink_error::Error> {
128131 ] ) ;
129132 global:: set_text_map_propagator ( propagator) ;
130133
134+ let maybe_channel = maybe_load_balanced_channel ( ) . await ;
135+
131136 // Logs
137+ let mut log_exporter_builder = LogExporter :: builder ( ) . with_tonic ( ) ;
138+ if let Some ( channel) = maybe_channel. clone ( ) {
139+ log_exporter_builder = log_exporter_builder. with_channel ( channel. into ( ) ) ;
140+ }
132141 let otlp_log_layer = OpenTelemetryTracingBridge :: new (
133142 & SdkLoggerProvider :: builder ( )
134143 . with_resource ( resource. clone ( ) )
135144 . with_batch_exporter (
136- LogExporter :: builder ( )
137- . with_tonic ( )
145+ log_exporter_builder
138146 . with_protocol ( Protocol :: Grpc )
139147 . build ( )
140148 . map_err ( |e| make_err ! ( Code :: Internal , "{e}" ) )
@@ -145,13 +153,16 @@ pub fn init_tracing() -> Result<(), nativelink_error::Error> {
145153 . with_filter ( otlp_filter ( ) ) ;
146154
147155 // Traces
156+ let mut span_exporter_builder = SpanExporter :: builder ( ) . with_tonic ( ) ;
157+ if let Some ( channel) = maybe_channel. clone ( ) {
158+ span_exporter_builder = span_exporter_builder. with_channel ( channel. into ( ) ) ;
159+ }
148160 let otlp_trace_layer = layer ( )
149161 . with_tracer (
150162 SdkTracerProvider :: builder ( )
151163 . with_resource ( resource. clone ( ) )
152164 . with_batch_exporter (
153- SpanExporter :: builder ( )
154- . with_tonic ( )
165+ span_exporter_builder
155166 . with_protocol ( Protocol :: Grpc )
156167 . build ( )
157168 . map_err ( |e| make_err ! ( Code :: Internal , "{e}" ) )
@@ -163,11 +174,14 @@ pub fn init_tracing() -> Result<(), nativelink_error::Error> {
163174 . with_filter ( otlp_filter ( ) ) ;
164175
165176 // Metrics
177+ let mut metric_exporter_builder = MetricExporter :: builder ( ) . with_tonic ( ) ;
178+ if let Some ( channel) = maybe_channel {
179+ metric_exporter_builder = metric_exporter_builder. with_channel ( channel. into ( ) ) ;
180+ }
166181 let meter_provider = SdkMeterProvider :: builder ( )
167182 . with_resource ( resource)
168183 . with_periodic_exporter (
169- MetricExporter :: builder ( )
170- . with_tonic ( )
184+ metric_exporter_builder
171185 . with_protocol ( Protocol :: Grpc )
172186 . build ( )
173187 . map_err ( |e| make_err ! ( Code :: Internal , "{e}" ) )
@@ -191,6 +205,38 @@ pub fn init_tracing() -> Result<(), nativelink_error::Error> {
191205 Ok ( ( ) )
192206}
193207
208+ const NL_OTEL_ENDPOINT : & str = "NL_OTEL_ENDPOINT" ;
209+
210+ async fn maybe_load_balanced_channel ( ) -> Option < LoadBalancedChannel > {
211+ match env:: var ( NL_OTEL_ENDPOINT ) {
212+ Ok ( endpoint) => {
213+ let url = Url :: parse ( endpoint. as_str ( ) )
214+ . map_err ( |e| {
215+ make_err ! ( Code :: Internal , "Unable to parse endpoint {endpoint}: {e:?}" )
216+ } )
217+ . unwrap ( ) ;
218+
219+ let host = url
220+ . host ( )
221+ . err_tip ( || format ! ( "Unable to get host from endpoint {endpoint}" ) )
222+ . unwrap ( ) ;
223+ let port = url
224+ . port ( )
225+ . err_tip ( || format ! ( "Unable to get port from endpoint {endpoint}" ) )
226+ . unwrap ( ) ;
227+
228+ Some (
229+ LoadBalancedChannel :: builder ( ( host. to_string ( ) , port) )
230+ . channel ( )
231+ . await
232+ . map_err ( |e| make_err ! ( Code :: Internal , "Invalid hostname '{endpoint}': {e}" ) )
233+ . unwrap ( ) ,
234+ )
235+ }
236+ Err ( _) => None ,
237+ }
238+ }
239+
194240/// Custom metadata key field for Bazel metadata.
195241const BAZEL_METADATA_KEY : & str = "bazel.metadata" ;
196242
@@ -201,6 +247,7 @@ const BAZEL_REQUESTMETADATA_HEADER: &str = "build.bazel.remote.execution.v2.requ
201247
202248use opentelemetry:: baggage:: BaggageExt ;
203249use opentelemetry:: context:: FutureExt ;
250+ use url:: Url ;
204251
205252#[ derive( Debug , Clone ) ]
206253pub struct OtlpMiddleware < S > {
0 commit comments