prosa_utils/config/
observability.rs

1//! Definition of Opentelemetry configuration
2
3use opentelemetry::{KeyValue, trace::TracerProvider as _};
4use opentelemetry_otlp::{ExporterBuildError, Protocol, WithExportConfig, WithHttpConfig};
5use opentelemetry_sdk::{
6    logs::SdkLoggerProvider,
7    metrics::SdkMeterProvider,
8    trace::{SdkTracerProvider, Tracer},
9};
10use serde::{Deserialize, Serialize};
11use std::{collections::HashMap, fmt, time::Duration};
12use tracing_subscriber::{filter, prelude::*};
13use tracing_subscriber::{layer::SubscriberExt, util::TryInitError};
14use url::Url;
15
16use crate::config::url::{get_safe_url, url_authentication};
17
18use super::tracing::{TelemetryFilter, TelemetryLevel};
19
20/// Configuration struct of an **O**pen **T**e**l**emetry **P**rotocol Exporter
21#[derive(Deserialize, Serialize, Clone)]
22pub(crate) struct OTLPExporterCfg {
23    pub(crate) level: Option<TelemetryLevel>,
24    endpoint: Url,
25    #[serde(skip_serializing)]
26    timeout_sec: Option<u64>,
27}
28
29impl OTLPExporterCfg {
30    /// Get the sanitized endpoint string (without credentials)
31    pub(crate) fn get_endpoint(&self) -> String {
32        let mut endpoint = self.endpoint.clone();
33        if !endpoint.username().is_empty() {
34            let _ = endpoint.set_username("");
35        }
36        if endpoint.password().is_some() {
37            let _ = endpoint.set_password(None);
38        }
39        endpoint.to_string()
40    }
41
42    /// Get the timeout duration if configured
43    pub(crate) fn get_timeout(&self) -> Option<Duration> {
44        self.timeout_sec.map(Duration::from_secs)
45    }
46
47    pub(crate) fn get_protocol(&self) -> Protocol {
48        match self.endpoint.scheme().to_lowercase().as_str() {
49            "grpc" => Protocol::Grpc,
50            "http/json" => Protocol::HttpJson,
51            _ => Protocol::HttpBinary,
52        }
53    }
54
55    pub(crate) fn get_header(&self) -> HashMap<String, String> {
56        let mut headers = HashMap::with_capacity(1);
57        if let Some(authorization) = url_authentication(&self.endpoint) {
58            // `WithHttpConfig::with_headers` URL-decodes values, so preserve literal `%`.
59            headers.insert(
60                "Authorization".to_string(),
61                authorization.replace('%', "%25"),
62            );
63        }
64        headers
65    }
66
67    pub(crate) fn get_resource(
68        &self,
69        attr: Vec<KeyValue>,
70    ) -> opentelemetry_sdk::resource::Resource {
71        opentelemetry_sdk::resource::Resource::builder()
72            .with_attributes(attr)
73            .with_attribute(opentelemetry::KeyValue::new(
74                "process.creation.time",
75                chrono::Utc::now().to_rfc3339(),
76            ))
77            .with_attribute(opentelemetry::KeyValue::new(
78                "process.pid",
79                opentelemetry::Value::I64(std::process::id() as i64),
80            ))
81            .build()
82    }
83}
84
85impl Default for OTLPExporterCfg {
86    fn default() -> Self {
87        Self {
88            level: None,
89            endpoint: Url::parse("grpc://localhost:4317").expect("default OTLP address is invalid"),
90            timeout_sec: None,
91        }
92    }
93}
94
95impl fmt::Debug for OTLPExporterCfg {
96    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
97        f.debug_struct("OTLPExporterCfg")
98            .field("level", &self.level)
99            .field(
100                "endpoint",
101                &get_safe_url(&self.endpoint).without_credentials(),
102            )
103            .field("timeout_sec", &self.timeout_sec)
104            .finish()
105    }
106}
107
108#[cfg(feature = "config-observability-prometheus")]
109/// Configuration struct of a prometheus metric exporter
110#[derive(Default, Debug, Deserialize, Serialize, Clone)]
111pub struct PrometheusExporterCfg {
112    endpoint: Option<String>,
113}
114
115#[cfg(feature = "config-observability-prometheus")]
116impl PrometheusExporterCfg {
117    /// Start an HTTP server to expose the metrics if needed
118    pub(crate) fn init_prometheus_server(
119        &self,
120        registry: &prometheus::Registry,
121    ) -> Result<(), ExporterBuildError> {
122        if let Some(endpoint) = self.endpoint.clone() {
123            let registry = registry.clone();
124            tokio::task::spawn(async move {
125                match tokio::net::TcpListener::bind(endpoint).await {
126                    Ok(listener) => {
127                        loop {
128                            if let Ok((stream, _)) = listener.accept().await {
129                                let io = hyper_util::rt::TokioIo::new(stream);
130                                let registry = registry.clone();
131                                tokio::task::spawn(async move {
132                                    if let Err(err) = hyper::server::conn::http1::Builder::new()
133                                    .serve_connection(
134                                        io,
135                                        #[allow(unused)]
136                                        hyper::service::service_fn(|req| {
137                                            let registry = registry.clone();
138                                            async move {
139                                                let metric_families = registry.gather();
140                                                let encoder = prometheus::TextEncoder::new();
141                                                if let Ok(metric_data) =
142                                                    encoder.encode_to_string(&metric_families)
143                                                {
144                                                    let response = hyper::Response::builder()
145                                                        .header(hyper::header::SERVER, concat!("ProSA/", env!("CARGO_PKG_VERSION")))
146                                                        .header(
147                                                            hyper::header::CONTENT_TYPE,
148                                                            "text/plain; version=1.0.0",
149                                                        );
150
151                                                    #[cfg(feature = "config-observability-gzip")]
152                                                    if req.headers().get(hyper::header::ACCEPT_ENCODING).is_some_and(|a| a.to_str().is_ok_and(|v| v.contains("gzip"))) {
153                                                        let mut gz_encoder = flate2::write::GzEncoder::new(Vec::with_capacity(2048), flate2::Compression::fast());
154                                                        if std::io::Write::write_all(&mut gz_encoder, metric_data.as_bytes()).is_ok()
155                                                            && let Ok(compressed_data) = gz_encoder.finish()
156                                                        {
157                                                            return response
158                                                                .header(hyper::header::CONTENT_ENCODING, "gzip")
159                                                                .body(http_body_util::Full::new(
160                                                                    bytes::Bytes::from(compressed_data)),
161                                                                )
162                                                                .map_err(|e| e.to_string());
163                                                        }
164                                                    }
165
166                                                    response
167                                                        .body(http_body_util::Full::new(
168                                                            bytes::Bytes::from(metric_data),
169                                                        ))
170                                                        .map_err(|e| e.to_string())
171                                                } else {
172                                                    Err("Can't serialize metrics".to_string())
173                                                }
174                                            }
175                                        }),
176                                    )
177                                    .await
178                                {
179                                    log::debug!(target: "prosa::observability::prometheus_server", "Error serving prometheus connection: {err:?}");
180                                }
181                                });
182                            }
183                        }
184                    }
185                    Err(e) => {
186                        log::error!(target: "prosa::observability::prometheus_server", "Failed to bind Prometheus metrics server: {e}");
187                    }
188                }
189            });
190        }
191
192        Ok(())
193    }
194
195    pub(crate) fn get_resource(
196        &self,
197        attr: Vec<KeyValue>,
198    ) -> opentelemetry_sdk::resource::Resource {
199        opentelemetry_sdk::resource::Resource::builder()
200            .with_attributes(attr)
201            .build()
202    }
203}
204
205/// Configuration struct of an stdout exporter
206#[derive(Default, Debug, Deserialize, Serialize, Copy, Clone)]
207pub(crate) struct StdoutExporterCfg {
208    #[serde(default)]
209    pub(crate) level: Option<TelemetryLevel>,
210}
211
212/// Telemetry data define for metrics
213#[derive(Default, Debug, Deserialize, Serialize, Clone)]
214pub struct TelemetryMetrics {
215    otlp: Option<OTLPExporterCfg>,
216    #[cfg(feature = "config-observability-prometheus")]
217    prometheus: Option<PrometheusExporterCfg>,
218    stdout: Option<StdoutExporterCfg>,
219}
220
221impl TelemetryMetrics {
222    /// Build a meter provider based on the self configuration
223    fn build_provider(
224        &self,
225        #[cfg(feature = "config-observability-prometheus")] resource_attr: Vec<KeyValue>,
226        #[cfg(feature = "config-observability-prometheus")] registry: &prometheus::Registry,
227    ) -> Result<SdkMeterProvider, ExporterBuildError> {
228        let mut meter_provider = SdkMeterProvider::builder();
229        if let Some(s) = &self.otlp {
230            let exporter = if s.get_protocol() == Protocol::Grpc {
231                let mut builder = opentelemetry_otlp::MetricExporter::builder()
232                    .with_tonic()
233                    .with_endpoint(s.get_endpoint())
234                    .with_protocol(s.get_protocol());
235                if let Some(timeout) = s.get_timeout() {
236                    builder = builder.with_timeout(timeout);
237                }
238                builder.build()
239            } else {
240                let mut builder = opentelemetry_otlp::MetricExporter::builder()
241                    .with_http()
242                    .with_headers(s.get_header())
243                    .with_endpoint(s.get_endpoint())
244                    .with_protocol(s.get_protocol());
245                if let Some(timeout) = s.get_timeout() {
246                    builder = builder.with_timeout(timeout);
247                }
248                builder.build()
249            }?;
250            meter_provider = meter_provider.with_periodic_exporter(exporter);
251        }
252
253        #[cfg(feature = "config-observability-prometheus")]
254        if let Some(prom) = &self.prometheus {
255            // configure OpenTelemetry to use this registry
256            let exporter = opentelemetry_prometheus::exporter()
257                .with_registry(registry.clone())
258                .with_resource_selector(opentelemetry_prometheus::ResourceSelector::All)
259                .without_target_info()
260                .build()
261                .map_err(|e| ExporterBuildError::InternalFailure(e.to_string()))?;
262            meter_provider = meter_provider
263                .with_resource(prom.get_resource(resource_attr))
264                .with_reader(exporter);
265
266            // Initialize the Prometheus server if needed
267            prom.init_prometheus_server(registry)?;
268        }
269
270        if self.stdout.is_some() {
271            let exporter = opentelemetry_stdout::MetricExporter::default();
272            meter_provider = meter_provider.with_periodic_exporter(exporter);
273        }
274
275        Ok(meter_provider.build())
276    }
277}
278
279/// Telemetry data define for metrics, logs, traces
280#[derive(Debug, Deserialize, Serialize, Clone)]
281pub struct TelemetryData {
282    otlp: Option<OTLPExporterCfg>,
283    stdout: Option<StdoutExporterCfg>,
284}
285
286impl TelemetryData {
287    /// Get the greater log level of the configuration (log level that include both OpenTelemetry and stdout)
288    fn get_max_level(&self) -> TelemetryLevel {
289        if let Some(otlp_level) = self.otlp.as_ref().and_then(|o| o.level) {
290            if let Some(stdout_level) = self.stdout.as_ref().and_then(|l| l.level) {
291                if otlp_level > stdout_level {
292                    otlp_level
293                } else {
294                    stdout_level
295                }
296            } else {
297                otlp_level
298            }
299        } else if let Some(stdout_level) = self.stdout.as_ref().and_then(|l| l.level) {
300            stdout_level
301        } else {
302            TelemetryLevel::TRACE
303        }
304    }
305
306    /// Build a logger provider based on the self configuration
307    fn build_logger_provider(
308        &self,
309        resource_attr: Vec<KeyValue>,
310    ) -> Result<(SdkLoggerProvider, TelemetryLevel), ExporterBuildError> {
311        let logs_provider = SdkLoggerProvider::builder();
312        if let Some(s) = &self.otlp {
313            let exporter = if s.get_protocol() == Protocol::Grpc {
314                let mut builder = opentelemetry_otlp::LogExporter::builder()
315                    .with_tonic()
316                    .with_endpoint(s.get_endpoint())
317                    .with_protocol(s.get_protocol());
318                if let Some(timeout) = s.get_timeout() {
319                    builder = builder.with_timeout(timeout);
320                }
321                builder.build()
322            } else {
323                let mut builder = opentelemetry_otlp::LogExporter::builder()
324                    .with_http()
325                    .with_headers(s.get_header())
326                    .with_endpoint(s.get_endpoint())
327                    .with_protocol(s.get_protocol());
328                if let Some(timeout) = s.get_timeout() {
329                    builder = builder.with_timeout(timeout);
330                }
331                builder.build()
332            }?;
333            Ok((
334                logs_provider
335                    .with_resource(s.get_resource(resource_attr))
336                    .with_batch_exporter(exporter)
337                    .build(),
338                s.level.unwrap_or_default(),
339            ))
340        } else if let Some(stdout) = &self.stdout {
341            Ok((
342                logs_provider
343                    .with_simple_exporter(opentelemetry_stdout::LogExporter::default())
344                    .build(),
345                stdout.level.unwrap_or_default(),
346            ))
347        } else {
348            Ok((logs_provider.build(), TelemetryLevel::OFF))
349        }
350    }
351
352    /// Build a tracer provider based on the self configuration
353    fn build_tracer_provider(
354        &self,
355        resource_attr: Vec<KeyValue>,
356    ) -> Result<SdkTracerProvider, ExporterBuildError> {
357        let mut trace_provider = SdkTracerProvider::builder();
358        if let Some(s) = &self.otlp {
359            let exporter = if s.get_protocol() == Protocol::Grpc {
360                let mut builder = opentelemetry_otlp::SpanExporter::builder()
361                    .with_tonic()
362                    .with_endpoint(s.get_endpoint())
363                    .with_protocol(s.get_protocol());
364                if let Some(timeout) = s.get_timeout() {
365                    builder = builder.with_timeout(timeout);
366                }
367                builder.build()
368            } else {
369                let mut builder = opentelemetry_otlp::SpanExporter::builder()
370                    .with_http()
371                    .with_headers(s.get_header())
372                    .with_endpoint(s.get_endpoint())
373                    .with_protocol(s.get_protocol());
374                if let Some(timeout) = s.get_timeout() {
375                    builder = builder.with_timeout(timeout);
376                }
377                builder.build()
378            }?;
379
380            trace_provider = trace_provider
381                .with_resource(s.get_resource(resource_attr))
382                .with_batch_exporter(exporter);
383        }
384
385        Ok(trace_provider.build())
386    }
387
388    /// Build a tracer provider based on the self configuration
389    fn build_tracer(
390        &self,
391        name: &str,
392        resource_attr: Vec<KeyValue>,
393    ) -> Result<Tracer, ExporterBuildError> {
394        self.build_tracer_provider(resource_attr)
395            .map(|p| p.tracer(name.to_string()))
396    }
397}
398
399impl Default for TelemetryData {
400    fn default() -> Self {
401        TelemetryData {
402            otlp: None,
403            stdout: Some(StdoutExporterCfg::default()),
404        }
405    }
406}
407
408/// Open telemetry settings of a ProSA
409///
410/// See [`TelemetryFilter`] to configure a specific filter for ProSA processors.
411///
412/// ```
413/// use opentelemetry::global;
414/// use prosa_utils::config::observability::Observability;
415/// use prosa_utils::config::tracing::TelemetryFilter;
416///
417/// #[tokio::main]
418/// async fn main() {
419///     let observability_settings = Observability::default();
420///
421///     // trace
422///     let filter = TelemetryFilter::default();
423///     observability_settings.tracing_init(&filter);
424/// }
425/// ```
426#[derive(Debug, Deserialize, Serialize, Clone)]
427pub struct Observability {
428    /// Additional attributes for all telemetry data
429    #[serde(default)]
430    attributes: HashMap<String, String>,
431    /// Global level for observability
432    #[serde(default)]
433    level: TelemetryLevel,
434    /// Metrics settings of a ProSA
435    metrics: Option<TelemetryMetrics>,
436    /// Logs settings of a ProSA
437    logs: Option<TelemetryData>,
438    /// Traces settings of a ProSA
439    traces: Option<TelemetryData>,
440}
441
442impl Observability {
443    pub(crate) fn common_scope_attributes(service_name: String, capacity: usize) -> Vec<KeyValue> {
444        let mut scope_attributes = Vec::with_capacity(capacity + 3);
445        scope_attributes.push(KeyValue::new("service.name", service_name));
446
447        match std::env::consts::ARCH {
448            "x86_64" => scope_attributes.push(KeyValue::new("host.arch", "amd64")),
449            "aarch64" => scope_attributes.push(KeyValue::new("host.arch", "arm64")),
450            "arm" => scope_attributes.push(KeyValue::new("host.arch", "arm32")),
451            _ => {}
452        }
453
454        match std::env::consts::OS {
455            "linux" => scope_attributes.push(KeyValue::new("os.type", "linux")),
456            "macos" => scope_attributes.push(KeyValue::new("os.type", "darwin")),
457            "freebsd" => scope_attributes.push(KeyValue::new("os.type", "freebsd")),
458            "openbsd" => scope_attributes.push(KeyValue::new("os.type", "openbsd")),
459            "netbsd" => scope_attributes.push(KeyValue::new("os.type", "netbsd")),
460            "windows" => scope_attributes.push(KeyValue::new("os.type", "windows")),
461            _ => {}
462        }
463
464        scope_attributes
465    }
466
467    /// Create an observability object with inline parameter instead of getting it from an external configuration
468    pub fn new(level: TelemetryLevel) -> Observability {
469        Observability {
470            attributes: HashMap::new(),
471            level,
472            metrics: Some(TelemetryMetrics::default()),
473            logs: Some(TelemetryData::default()),
474            traces: Some(TelemetryData::default()),
475        }
476    }
477
478    /// Getter of the observability `service.name` attributes
479    pub fn get_service_name(&self) -> &str {
480        self.attributes
481            .get("service.name")
482            .map(|s| s.as_ref())
483            .unwrap_or("prosa")
484    }
485
486    /// Setter of the ProSA name for all observability `service.name` attributes
487    pub fn set_prosa_name(&mut self, name: &str) {
488        self.attributes
489            .entry("service.name".to_string())
490            .or_insert_with(|| name.to_string());
491    }
492
493    /// Getter of the common scope attributes
494    pub fn get_scope_attributes(&self) -> Vec<KeyValue> {
495        // start with common attributes
496        let mut scope_attr = Self::common_scope_attributes(
497            self.get_service_name().to_string(),
498            self.attributes.len() + 3,
499        );
500
501        if !self.attributes.contains_key("host.name")
502            && let Some(hostname) = super::hostname()
503        {
504            scope_attr.push(KeyValue::new("host.name", hostname));
505        }
506
507        if !self.attributes.contains_key("service.instance.id") {
508            scope_attr.push(KeyValue::new("service.instance.id", super::hostid()));
509        }
510
511        if !self.attributes.contains_key("service.version") {
512            scope_attr.push(KeyValue::new("service.version", env!("CARGO_PKG_VERSION")));
513        }
514
515        // append custom attributes from configuration
516        scope_attr.append(
517            self.attributes
518                .iter()
519                .map(|(k, v)| {
520                    KeyValue::new(k.clone(), opentelemetry::Value::String(v.clone().into()))
521                })
522                .collect::<Vec<KeyValue>>()
523                .as_mut(),
524        );
525
526        scope_attr
527    }
528
529    /// Getter of the log level (max value)
530    pub fn get_logger_level(&self) -> TelemetryLevel {
531        if let Some(logs) = &self.logs {
532            let logs_level = logs.get_max_level();
533            if logs_level > self.level {
534                logs_level
535            } else {
536                self.level
537            }
538        } else {
539            self.level
540        }
541    }
542
543    /// Getter of the global telemetry level
544    pub fn get_level(&self) -> TelemetryLevel {
545        self.level
546    }
547
548    /// Meter provider builder
549    #[cfg(feature = "config-observability-prometheus")]
550    pub fn build_meter_provider(&self, registry: &prometheus::Registry) -> SdkMeterProvider {
551        if let Some(settings) = &self.metrics {
552            settings
553                .build_provider(self.get_scope_attributes(), registry)
554                .unwrap_or_default()
555        } else {
556            SdkMeterProvider::default()
557        }
558    }
559
560    /// Meter provider builder
561    #[cfg(not(feature = "config-observability-prometheus"))]
562    pub fn build_meter_provider(&self) -> SdkMeterProvider {
563        if let Some(settings) = &self.metrics {
564            settings.build_provider().unwrap_or_default()
565        } else {
566            SdkMeterProvider::default()
567        }
568    }
569
570    /// Logger provider builder
571    pub fn build_logger_provider(&self) -> (SdkLoggerProvider, TelemetryLevel) {
572        if let Some(settings) = &self.logs {
573            match settings.build_logger_provider(self.get_scope_attributes()) {
574                Ok(m) => m,
575                Err(_) => (
576                    SdkLoggerProvider::builder().build(),
577                    TelemetryLevel::default(),
578                ),
579            }
580        } else {
581            (
582                SdkLoggerProvider::builder().build(),
583                TelemetryLevel::default(),
584            )
585        }
586    }
587
588    /// Tracer provider builder
589    ///
590    /// ```
591    /// use opentelemetry::{global, trace::TracerProvider};
592    /// use prosa_utils::config::observability::Observability;
593    ///
594    /// let otel_settings = Observability::default();
595    /// let tracer = otel_settings
596    ///     .build_tracer_provider()
597    ///     .tracer("prosa_proc_example");
598    /// ```
599    pub fn build_tracer_provider(&self) -> SdkTracerProvider {
600        if let Some(settings) = &self.traces {
601            settings
602                .build_tracer_provider(self.get_scope_attributes())
603                .unwrap_or_default()
604        } else {
605            SdkTracerProvider::default()
606        }
607    }
608
609    /// Tracer builder
610    ///
611    /// ```
612    /// use opentelemetry::{global, trace::Tracer};
613    /// use prosa_utils::config::observability::Observability;
614    ///
615    /// let otel_settings = Observability::default();
616    /// let tracer = otel_settings
617    ///     .build_tracer();
618    /// ```
619    pub fn build_tracer(&self) -> Tracer {
620        if let Some(settings) = &self.traces {
621            match settings.build_tracer(self.get_service_name(), self.get_scope_attributes()) {
622                Ok(m) => m,
623                Err(_) => SdkTracerProvider::default().tracer(self.get_service_name().to_string()),
624            }
625        } else {
626            SdkTracerProvider::default().tracer(self.get_service_name().to_string())
627        }
628    }
629
630    /// Method to init `tracing`
631    pub fn tracing_init(&self, filter: &TelemetryFilter) -> Result<(), TryInitError> {
632        filter.set_level(self.level.into());
633        let subscriber = tracing_subscriber::registry().with(filter::LevelFilter::TRACE);
634
635        if let Some(traces) = &self.traces {
636            if let Some(otlp) = &traces.otlp {
637                let tracer = self.build_tracer();
638                let subscriber_filter = filter.clone_with_level(otlp.level.unwrap_or_default());
639                let subscriber = subscriber.with(
640                    tracing_opentelemetry::layer()
641                        .with_tracer(tracer)
642                        .with_filter(subscriber_filter),
643                );
644
645                if let Some(stdout) = traces.stdout {
646                    let subscriber_filter =
647                        filter.clone_with_level(stdout.level.unwrap_or_default());
648                    subscriber
649                        .with(tracing_subscriber::fmt::Layer::new().with_filter(subscriber_filter))
650                        .try_init()
651                } else {
652                    subscriber.try_init()
653                }
654            } else if let Some(stdout) = traces.stdout {
655                let subscriber_filter = filter.clone_with_level(stdout.level.unwrap_or_default());
656                subscriber
657                    .with(tracing_subscriber::fmt::Layer::new().with_filter(subscriber_filter))
658                    .try_init()
659            } else {
660                subscriber.try_init()
661            }
662        } else if let Some(logs) = &self.logs
663            && let Ok((logger_provider, level)) =
664                logs.build_logger_provider(self.get_scope_attributes())
665            && level > TelemetryLevel::OFF
666        {
667            let logger_filter = filter.clone_with_level(level);
668            subscriber
669                .with(
670                    opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge::new(
671                        &logger_provider,
672                    )
673                    .with_filter(logger_filter),
674                )
675                .try_init()
676        } else {
677            subscriber.try_init()
678        }
679    }
680}
681
682impl Default for Observability {
683    fn default() -> Self {
684        Self {
685            attributes: HashMap::new(),
686            level: TelemetryLevel::default(),
687            metrics: Some(TelemetryMetrics::default()),
688            logs: Some(TelemetryData {
689                otlp: None,
690                stdout: Some(StdoutExporterCfg {
691                    level: Some(TelemetryLevel::DEBUG),
692                }),
693            }),
694            traces: Some(TelemetryData {
695                otlp: None,
696                stdout: Some(StdoutExporterCfg {
697                    level: Some(TelemetryLevel::DEBUG),
698                }),
699            }),
700        }
701    }
702}
703
704#[cfg(test)]
705mod tests {
706    use super::*;
707
708    #[test]
709    fn otlp_http_authorization_preserves_literal_percent_triplets() {
710        let config = OTLPExporterCfg {
711            level: None,
712            endpoint: Url::parse("http://:token%2541@localhost:4318")
713                .expect("OTLP endpoint should be valid"),
714            timeout_sec: None,
715        };
716
717        assert_eq!(
718            Some("Bearer token%2541"),
719            config.get_header().get("Authorization").map(String::as_str)
720        );
721    }
722
723    #[test]
724    fn otlp_debug_redacts_url_secrets() {
725        let config = OTLPExporterCfg {
726            level: None,
727            endpoint: Url::parse(
728                "http://user:password@localhost:4318/v1?token=secret#access_token=secret",
729            )
730            .expect("OTLP endpoint should be valid"),
731            timeout_sec: None,
732        };
733
734        let debug = format!("{config:?}");
735        assert!(debug.contains("http://localhost:4318/v1"));
736        assert!(!debug.contains("user"));
737        assert!(!debug.contains("password"));
738        assert!(!debug.contains("secret"));
739    }
740}
prosa_utils/config/observability.rs

prosa_utils/config/
observability.rs
