1#[cfg(not(any(feature = "grpc", feature = "http")))]
27compile_error!("at least one transport feature must be enabled: `grpc` or `http`");
28
29#[cfg(feature = "testing")]
30pub mod testing;
31
32#[cfg(feature = "axum")]
33pub mod axum_middleware;
34
35#[cfg(feature = "tonic-tracing")]
36pub mod grpc_middleware;
37
38#[cfg(feature = "profiling")]
39pub mod profiling;
40mod runtime_metrics;
41
42pub mod instrumented_port;
43pub mod log_bridge;
44pub mod span_enrichment;
45pub mod spanned;
46
47pub use instrumented_port::{Instrumented, InstrumentedArc};
48pub use log_bridge::{
49 PROPAGATED_SPAN_FIELDS, SpanLogAttrs, record_span_log_attr, record_span_log_attr_on,
50};
51pub use spanned::{Spanned, in_span};
52
53use opentelemetry::KeyValue;
54use opentelemetry::propagation::TextMapCompositePropagator;
55use opentelemetry_otlp::WithExportConfig;
56use opentelemetry_sdk::{
57 Resource,
58 logs::SdkLoggerProvider,
59 metrics::{MeterProviderBuilder, PeriodicReader, SdkMeterProvider},
60 propagation::{BaggagePropagator, TraceContextPropagator},
61 trace::{BatchConfigBuilder, BatchSpanProcessor, Sampler, SdkTracer, SdkTracerProvider},
62};
63use opentelemetry_semantic_conventions::attribute::{
64 DEPLOYMENT_ENVIRONMENT_NAME, HOST_NAME, PROCESS_PID, SERVICE_VERSION,
65};
66use std::error::Error;
67use std::time::Duration;
68use tracing_subscriber::layer::SubscriberExt;
69use tracing_subscriber::util::SubscriberInitExt;
70
71fn tracing_bridge_tracer(provider: &SdkTracerProvider) -> SdkTracer {
72 use opentelemetry::trace::TracerProvider as _;
73
74 provider.tracer(env!("CARGO_PKG_NAME"))
75}
76
77#[derive(Debug, Clone)]
92pub enum TraceSampler {
93 AlwaysOn,
95 AlwaysOff,
97 TraceIdRatio(f64),
99 ParentBased(Box<TraceSampler>),
102}
103
104#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
106pub enum LogFormat {
107 #[default]
109 Pretty,
110 Json,
112}
113
114impl TraceSampler {
115 fn into_sdk_sampler(self) -> Sampler {
117 match self {
118 TraceSampler::AlwaysOn => Sampler::AlwaysOn,
119 TraceSampler::AlwaysOff => Sampler::AlwaysOff,
120 TraceSampler::TraceIdRatio(r) => Sampler::TraceIdRatioBased(r),
121 TraceSampler::ParentBased(inner) => {
122 Sampler::ParentBased(Box::new(inner.into_sdk_sampler()))
123 }
124 }
125 }
126}
127
128fn sampler_from_env() -> Result<Option<TraceSampler>, Box<dyn Error>> {
136 let name = match std::env::var("OTEL_TRACES_SAMPLER") {
137 Ok(v) => v,
138 Err(_) => return Ok(None),
139 };
140 let arg = std::env::var("OTEL_TRACES_SAMPLER_ARG").ok();
141 let sampler = match name.as_str() {
142 "always_on" => TraceSampler::AlwaysOn,
143 "always_off" => TraceSampler::AlwaysOff,
144 "traceidratio" => {
145 let ratio = arg
146 .as_deref()
147 .unwrap_or("1.0")
148 .parse::<f64>()
149 .unwrap_or(1.0);
150 TraceSampler::TraceIdRatio(ratio)
151 }
152 "parentbased_always_on" => TraceSampler::ParentBased(Box::new(TraceSampler::AlwaysOn)),
153 "parentbased_always_off" => TraceSampler::ParentBased(Box::new(TraceSampler::AlwaysOff)),
154 "parentbased_traceidratio" => {
155 let ratio = arg
156 .as_deref()
157 .unwrap_or("1.0")
158 .parse::<f64>()
159 .unwrap_or(1.0);
160 TraceSampler::ParentBased(Box::new(TraceSampler::TraceIdRatio(ratio)))
161 }
162 unknown => {
163 return Err(format!(
164 "OTEL_TRACES_SAMPLER: unrecognised sampler name '{unknown}'. \
165 Valid values: always_on, always_off, traceidratio, \
166 parentbased_always_on, parentbased_always_off, parentbased_traceidratio"
167 )
168 .into());
169 }
170 };
171 Ok(Some(sampler))
172}
173
174const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
176
177pub struct TelemetryHandles {
198 pub tracer_provider: SdkTracerProvider,
199 pub meter_provider: Option<SdkMeterProvider>,
200 pub logger_provider: Option<SdkLoggerProvider>,
201 shutdown_timeout: Duration,
202 #[cfg(feature = "profiling")]
203 pub profiling_handle: Option<profiling::ProfilingHandle>,
204}
205
206impl TelemetryHandles {
207 pub fn shutdown(&self) -> Result<(), Box<dyn Error>> {
238 if let Err(e) = self.tracer_provider.shutdown() {
239 tracing::warn!("tracer provider shutdown error: {e}");
240 }
241 if let Some(mp) = &self.meter_provider
242 && let Err(e) = mp.shutdown()
243 {
244 tracing::warn!("meter provider shutdown error: {e}");
245 }
246 if let Some(lp) = &self.logger_provider
247 && let Err(e) = lp.shutdown()
248 {
249 tracing::warn!("logger provider shutdown error: {e}");
250 }
251 Ok(())
252 }
253}
254
255impl Drop for TelemetryHandles {
256 fn drop(&mut self) {
257 let tracer_provider = self.tracer_provider.clone();
258 let meter_provider = self.meter_provider.clone();
259 let logger_provider = self.logger_provider.clone();
260 let timeout = self.shutdown_timeout;
261
262 let (tx, rx) = std::sync::mpsc::channel();
263 std::thread::spawn(move || {
264 if let Err(e) = tracer_provider.shutdown() {
265 tracing::warn!("tracer provider shutdown error: {e}");
266 }
267 if let Some(mp) = meter_provider
268 && let Err(e) = mp.shutdown()
269 {
270 tracing::warn!("meter provider shutdown error: {e}");
271 }
272 if let Some(lp) = logger_provider
273 && let Err(e) = lp.shutdown()
274 {
275 tracing::warn!("logger provider shutdown error: {e}");
276 }
277 let _ = tx.send(());
278 });
279
280 if rx.recv_timeout(timeout).is_err() {
281 tracing::warn!(
282 "telemetry shutdown did not complete within {timeout:?}; \
283 some spans/metrics may not have been exported"
284 );
285 }
286 }
287}
288
289#[derive(Debug, Clone, Copy, PartialEq, Eq)]
312pub enum ExportProtocol {
313 #[cfg(feature = "grpc")]
315 Grpc,
316 #[cfg(feature = "http")]
318 HttpProtobuf,
319}
320
321#[cfg(feature = "grpc-mtls")]
331#[derive(Clone)]
332pub struct MtlsMaterial {
333 pub client_cert_chain_pem: Vec<u8>,
335 pub client_key_pem: Vec<u8>,
337 pub trust_bundle_pem: Vec<u8>,
339}
340
341#[cfg(feature = "grpc-mtls")]
342impl std::fmt::Debug for MtlsMaterial {
343 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
344 f.debug_struct("MtlsMaterial")
345 .field("client_cert_chain_pem", &"<redacted>")
346 .field("client_key_pem", &"<redacted>")
347 .field("trust_bundle_pem", &"<redacted>")
348 .finish()
349 }
350}
351
352fn protocol_from_env() -> Option<ExportProtocol> {
354 let val = std::env::var("OTEL_EXPORTER_OTLP_PROTOCOL").ok()?;
355 match val.trim() {
356 #[cfg(feature = "grpc")]
357 "grpc" => Some(ExportProtocol::Grpc),
358 #[cfg(feature = "http")]
359 "http/protobuf" => Some(ExportProtocol::HttpProtobuf),
360 _ => None,
361 }
362}
363
364pub struct Telemetry;
380
381impl Telemetry {
382 pub fn builder(service_name: &str) -> TelemetryBuilder {
386 TelemetryBuilder {
387 service_name: Some(service_name.to_string()),
388 service_version: None,
389 deployment_environment: None,
390 sampler: None,
391 metrics: true,
392 logs: false,
393 protocol: None,
394 max_export_batch_size: None,
395 metric_export_interval: None,
396 export_timeout: None,
397 shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
398 log_filter: None,
399 log_format: LogFormat::default(),
400 extra_layers: Vec::new(),
401 extra_metric_readers: Vec::new(),
402 runtime_metrics: true,
403 #[cfg(feature = "grpc-mtls")]
404 mtls: None,
405 propagated_span_fields: crate::log_bridge::PROPAGATED_SPAN_FIELDS,
406 #[cfg(feature = "profiling")]
407 pyroscope_endpoint: None,
408 }
409 }
410
411 pub fn from_env() -> TelemetryBuilder {
421 TelemetryBuilder {
422 service_name: None,
423 service_version: None,
424 deployment_environment: None,
425 sampler: None,
426 metrics: true,
427 logs: false,
428 protocol: None,
429 max_export_batch_size: None,
430 metric_export_interval: None,
431 export_timeout: None,
432 shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
433 log_filter: None,
434 log_format: LogFormat::default(),
435 extra_layers: Vec::new(),
436 extra_metric_readers: Vec::new(),
437 runtime_metrics: true,
438 #[cfg(feature = "grpc-mtls")]
439 mtls: None,
440 propagated_span_fields: crate::log_bridge::PROPAGATED_SPAN_FIELDS,
441 #[cfg(feature = "profiling")]
442 pyroscope_endpoint: None,
443 }
444 }
445}
446
447#[must_use = "a TelemetryBuilder does nothing until .init() is called"]
465pub struct TelemetryBuilder {
466 service_name: Option<String>,
467 service_version: Option<String>,
468 deployment_environment: Option<String>,
469 sampler: Option<TraceSampler>,
470 metrics: bool,
471 logs: bool,
472 protocol: Option<ExportProtocol>,
473 max_export_batch_size: Option<usize>,
474 metric_export_interval: Option<Duration>,
475 export_timeout: Option<Duration>,
476 shutdown_timeout: Duration,
477 log_filter: Option<String>,
478 log_format: LogFormat,
479 extra_layers: Vec<
480 Box<dyn tracing_subscriber::Layer<tracing_subscriber::Registry> + Send + Sync + 'static>,
481 >,
482 extra_metric_readers: Vec<MeterProviderInstaller>,
483 runtime_metrics: bool,
484 #[cfg(feature = "grpc-mtls")]
485 mtls: Option<MtlsMaterial>,
486 propagated_span_fields: &'static [&'static str],
487 #[cfg(feature = "profiling")]
488 pyroscope_endpoint: Option<String>,
489}
490
491type MeterProviderInstaller =
496 Box<dyn FnOnce(MeterProviderBuilder) -> MeterProviderBuilder + Send + Sync>;
497
498impl TelemetryBuilder {
499 pub fn with_log_filter(mut self, directive: impl Into<String>) -> Self {
504 self.log_filter = Some(directive.into());
505 self
506 }
507
508 pub fn with_log_format(mut self, format: LogFormat) -> Self {
510 self.log_format = format;
511 self
512 }
513
514 pub fn with_version(mut self, version: &str) -> Self {
516 self.service_version = Some(version.to_string());
517 self
518 }
519
520 pub fn with_environment(mut self, environment: &str) -> Self {
522 self.deployment_environment = Some(environment.to_string());
523 self
524 }
525
526 #[cfg(feature = "grpc-mtls")]
551 pub fn with_mtls(mut self, material: MtlsMaterial) -> Self {
552 self.mtls = Some(material);
553 self.protocol = Some(ExportProtocol::Grpc);
554 self
555 }
556
557 pub fn with_sampler(mut self, sampler: TraceSampler) -> Self {
560 self.sampler = Some(sampler);
561 self
562 }
563
564 pub fn with_metrics(mut self, enabled: bool) -> Self {
566 self.metrics = enabled;
567 self
568 }
569
570 pub fn with_runtime_metrics(mut self, enabled: bool) -> Self {
586 self.runtime_metrics = enabled;
587 self
588 }
589
590 pub fn with_protocol(mut self, protocol: ExportProtocol) -> Self {
594 self.protocol = Some(protocol);
595 self
596 }
597
598 pub fn with_max_export_batch_size(mut self, size: usize) -> Self {
603 self.max_export_batch_size = Some(size);
604 self
605 }
606
607 pub fn with_metric_export_interval(mut self, interval: Duration) -> Self {
612 self.metric_export_interval = Some(interval);
613 self
614 }
615
616 pub fn with_logs(mut self, enabled: bool) -> Self {
623 self.logs = enabled;
624 self
625 }
626
627 pub fn with_propagated_span_fields(mut self, fields: &'static [&'static str]) -> Self {
640 self.propagated_span_fields = fields;
641 self
642 }
643
644 pub fn with_export_timeout(mut self, timeout: Duration) -> Self {
648 self.export_timeout = Some(timeout);
649 self
650 }
651
652 pub fn with_shutdown_timeout(mut self, timeout: Duration) -> Self {
659 self.shutdown_timeout = timeout;
660 self
661 }
662
663 #[cfg(feature = "profiling")]
681 pub fn with_profiling(mut self, endpoint: &str) -> Self {
682 self.pyroscope_endpoint = Some(endpoint.to_string());
683 self
684 }
685
686 pub fn with_meter_provider_setup<F>(mut self, setup: F) -> Self
743 where
744 F: FnOnce(MeterProviderBuilder) -> MeterProviderBuilder + Send + Sync + 'static,
745 {
746 self.extra_metric_readers.push(Box::new(setup));
747 self
748 }
749
750 pub fn with_layer<L>(mut self, layer: L) -> Self
751 where
752 L: tracing_subscriber::Layer<tracing_subscriber::Registry> + Send + Sync + 'static,
753 {
754 self.extra_layers.push(Box::new(layer));
755 self
756 }
757
758 pub fn init(self) -> Result<TelemetryHandles, Box<dyn Error>> {
773 let log_filter = match self.log_filter.as_deref() {
774 Some(directive) => tracing_subscriber::EnvFilter::try_new(directive)?,
775 None => tracing_subscriber::EnvFilter::from_default_env(),
776 };
777
778 if let Some(interval) = self.metric_export_interval
779 && interval.is_zero()
780 {
781 return Err("metric_export_interval must be greater than zero".into());
782 }
783
784 let protocol = self.protocol.or_else(protocol_from_env).unwrap_or({
785 #[cfg(feature = "grpc")]
786 {
787 ExportProtocol::Grpc
788 }
789 #[cfg(all(not(feature = "grpc"), feature = "http"))]
790 {
791 ExportProtocol::HttpProtobuf
792 }
793 });
794
795 let default_endpoint = match protocol {
796 #[cfg(feature = "grpc")]
797 ExportProtocol::Grpc => "http://localhost:4317",
798 #[cfg(feature = "http")]
799 ExportProtocol::HttpProtobuf => "http://localhost:4318",
800 };
801 let endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
802 .unwrap_or_else(|_| default_endpoint.to_string());
803
804 let export_timeout = self.export_timeout.or_else(timeout_from_env);
806
807 let service_name = self.service_name.unwrap_or_else(|| {
809 std::env::var("OTEL_SERVICE_NAME").unwrap_or_else(|_| "unknown_service".to_string())
810 });
811
812 let resource = build_resource(
813 &service_name,
814 self.service_version.as_deref(),
815 self.deployment_environment.as_deref(),
816 );
817
818 let sampler = match self.sampler {
819 Some(s) => s,
820 None => sampler_from_env()?.unwrap_or(TraceSampler::AlwaysOn),
821 };
822
823 let trace_exporter = build_span_exporter(
825 protocol,
826 &endpoint,
827 export_timeout,
828 #[cfg(feature = "grpc-mtls")]
829 self.mtls.as_ref(),
830 )?;
831
832 let batch_processor = if let Some(size) = self.max_export_batch_size {
833 BatchSpanProcessor::builder(trace_exporter)
834 .with_batch_config(
835 BatchConfigBuilder::default()
836 .with_max_export_batch_size(size)
837 .build(),
838 )
839 .build()
840 } else {
841 BatchSpanProcessor::builder(trace_exporter).build()
842 };
843
844 let tracer_provider = SdkTracerProvider::builder()
845 .with_resource(resource.clone())
846 .with_sampler(sampler.into_sdk_sampler())
847 .with_span_processor(batch_processor)
848 .build();
849
850 opentelemetry::global::set_tracer_provider(tracer_provider.clone());
851
852 let propagator = TextMapCompositePropagator::new(vec![
854 Box::new(TraceContextPropagator::new()),
855 Box::new(BaggagePropagator::new()),
856 ]);
857 opentelemetry::global::set_text_map_propagator(propagator);
858
859 let meter_provider = if self.metrics {
861 let metric_exporter = build_metric_exporter(
862 protocol,
863 &endpoint,
864 export_timeout,
865 #[cfg(feature = "grpc-mtls")]
866 self.mtls.as_ref(),
867 )?;
868
869 let periodic_reader = if let Some(interval) = self.metric_export_interval {
870 PeriodicReader::builder(metric_exporter)
871 .with_interval(interval)
872 .build()
873 } else {
874 PeriodicReader::builder(metric_exporter).build()
875 };
876
877 let mut mp_builder = SdkMeterProvider::builder()
878 .with_resource(resource.clone())
879 .with_reader(periodic_reader);
880 for installer in self.extra_metric_readers {
881 mp_builder = installer(mp_builder);
882 }
883 let mp = mp_builder.build();
884
885 opentelemetry::global::set_meter_provider(mp.clone());
886
887 if self.runtime_metrics {
891 crate::runtime_metrics::install();
892 }
893
894 Some(mp)
895 } else {
896 None
897 };
898
899 let logger_provider = if self.logs {
901 let log_exporter = build_log_exporter(
902 protocol,
903 &endpoint,
904 export_timeout,
905 #[cfg(feature = "grpc-mtls")]
906 self.mtls.as_ref(),
907 )?;
908
909 let lp = SdkLoggerProvider::builder()
910 .with_resource(resource)
911 .with_batch_exporter(log_exporter)
912 .build();
913
914 Some(lp)
915 } else {
916 None
917 };
918
919 #[cfg(feature = "profiling")]
921 let profiling_handle = if let Some(ref endpoint) = self.pyroscope_endpoint {
922 let identity = profiling::ProfilingIdentity {
927 host_name: hostname::get()
928 .ok()
929 .and_then(|h| h.into_string().ok())
930 .filter(|h| !h.is_empty()),
931 deployment_environment: self.deployment_environment.clone(),
932 service_version: self.service_version.clone(),
933 };
934 profiling::start_pyroscope_bridge(&service_name, endpoint, &identity)?
935 } else {
936 None
937 };
938 #[cfg(not(feature = "profiling"))]
939 let _profiling_handle: Option<()> = None;
940
941 let extra = if self.extra_layers.is_empty() {
947 None
948 } else {
949 Some(self.extra_layers)
950 };
951
952 macro_rules! install_subscriber {
953 ($fmt_layer:expr) => {{
954 let otel_layer = tracing_opentelemetry::layer()
958 .with_tracer(tracing_bridge_tracer(&tracer_provider));
959 let registry = tracing_subscriber::registry()
960 .with(extra)
961 .with(log_filter)
962 .with($fmt_layer)
963 .with(otel_layer);
964
965 #[cfg(feature = "profiling-bridge-pyroscope-rs")]
966 let registry = registry.with(crate::profiling::ProfilingTagLayer);
967
968 if let Some(lp) = &logger_provider {
969 if let Err(e) = registry
970 .with(crate::log_bridge::SpanAwareLogBridge::new(
971 lp,
972 self.propagated_span_fields,
973 ))
974 .try_init()
975 {
976 eprintln!(
977 "otel-bootstrap: global tracing subscriber already installed — \
978 OTLP log records will NOT be exported to the collector: {e}"
979 );
980 }
981 } else if let Err(e) = registry.try_init() {
982 eprintln!(
983 "otel-bootstrap: global tracing subscriber already installed — \
984 OTLP telemetry will NOT be exported to the collector: {e}"
985 );
986 }
987 }};
988 }
989
990 match self.log_format {
991 LogFormat::Pretty => install_subscriber!(tracing_subscriber::fmt::layer()),
992 LogFormat::Json => install_subscriber!(tracing_subscriber::fmt::layer().json()),
993 }
994
995 Ok(TelemetryHandles {
996 tracer_provider,
997 meter_provider,
998 logger_provider,
999 shutdown_timeout: self.shutdown_timeout,
1000 #[cfg(feature = "profiling")]
1001 profiling_handle,
1002 })
1003 }
1004}
1005
1006pub fn init_telemetry(service_name: &str) -> Result<TelemetryHandles, Box<dyn Error>> {
1020 Telemetry::builder(service_name).init()
1021}
1022
1023pub fn init_telemetry_with_sampler(
1040 service_name: &str,
1041 sampler: Option<TraceSampler>,
1042) -> Result<TelemetryHandles, Box<dyn Error>> {
1043 let builder = Telemetry::builder(service_name);
1044 match sampler {
1045 Some(s) => builder.with_sampler(s),
1046 None => builder, }
1048 .init()
1049}
1050
1051fn timeout_from_env() -> Option<Duration> {
1053 let ms = std::env::var("OTEL_EXPORTER_OTLP_TIMEOUT").ok()?;
1054 let ms: u64 = ms.trim().parse().ok()?;
1055 Some(Duration::from_millis(ms))
1056}
1057
1058#[cfg(feature = "grpc-mtls")]
1065fn build_tls_config(material: &MtlsMaterial) -> tonic::transport::ClientTlsConfig {
1066 use tonic::transport::{Certificate, ClientTlsConfig, Identity};
1067 ClientTlsConfig::new()
1068 .ca_certificate(Certificate::from_pem(&material.trust_bundle_pem))
1069 .identity(Identity::from_pem(
1070 &material.client_cert_chain_pem,
1071 &material.client_key_pem,
1072 ))
1073}
1074
1075fn build_span_exporter(
1076 protocol: ExportProtocol,
1077 endpoint: &str,
1078 timeout: Option<Duration>,
1079 #[cfg(feature = "grpc-mtls")] mtls: Option<&MtlsMaterial>,
1080) -> Result<opentelemetry_otlp::SpanExporter, Box<dyn Error>> {
1081 match protocol {
1082 #[cfg(feature = "grpc")]
1083 ExportProtocol::Grpc => {
1084 let mut b = opentelemetry_otlp::SpanExporter::builder()
1085 .with_tonic()
1086 .with_endpoint(endpoint);
1087 if let Some(t) = timeout {
1088 b = b.with_timeout(t);
1089 }
1090 #[cfg(feature = "grpc-mtls")]
1091 if let Some(m) = mtls {
1092 use opentelemetry_otlp::WithTonicConfig as _;
1093 b = b.with_tls_config(build_tls_config(m));
1094 }
1095 Ok(b.build()?)
1096 }
1097 #[cfg(feature = "http")]
1098 ExportProtocol::HttpProtobuf => {
1099 let mut b = opentelemetry_otlp::SpanExporter::builder()
1100 .with_http()
1101 .with_endpoint(endpoint);
1102 if let Some(t) = timeout {
1103 b = b.with_timeout(t);
1104 }
1105 Ok(b.build()?)
1106 }
1107 }
1108}
1109
1110fn build_metric_exporter(
1111 protocol: ExportProtocol,
1112 endpoint: &str,
1113 timeout: Option<Duration>,
1114 #[cfg(feature = "grpc-mtls")] mtls: Option<&MtlsMaterial>,
1115) -> Result<opentelemetry_otlp::MetricExporter, Box<dyn Error>> {
1116 match protocol {
1117 #[cfg(feature = "grpc")]
1118 ExportProtocol::Grpc => {
1119 let mut b = opentelemetry_otlp::MetricExporter::builder()
1120 .with_tonic()
1121 .with_endpoint(endpoint);
1122 if let Some(t) = timeout {
1123 b = b.with_timeout(t);
1124 }
1125 #[cfg(feature = "grpc-mtls")]
1126 if let Some(m) = mtls {
1127 use opentelemetry_otlp::WithTonicConfig as _;
1128 b = b.with_tls_config(build_tls_config(m));
1129 }
1130 Ok(b.build()?)
1131 }
1132 #[cfg(feature = "http")]
1133 ExportProtocol::HttpProtobuf => {
1134 let mut b = opentelemetry_otlp::MetricExporter::builder()
1135 .with_http()
1136 .with_endpoint(endpoint);
1137 if let Some(t) = timeout {
1138 b = b.with_timeout(t);
1139 }
1140 Ok(b.build()?)
1141 }
1142 }
1143}
1144
1145fn build_log_exporter(
1146 protocol: ExportProtocol,
1147 endpoint: &str,
1148 timeout: Option<Duration>,
1149 #[cfg(feature = "grpc-mtls")] mtls: Option<&MtlsMaterial>,
1150) -> Result<opentelemetry_otlp::LogExporter, Box<dyn Error>> {
1151 match protocol {
1152 #[cfg(feature = "grpc")]
1153 ExportProtocol::Grpc => {
1154 let mut b = opentelemetry_otlp::LogExporter::builder()
1155 .with_tonic()
1156 .with_endpoint(endpoint);
1157 if let Some(t) = timeout {
1158 b = b.with_timeout(t);
1159 }
1160 #[cfg(feature = "grpc-mtls")]
1161 if let Some(m) = mtls {
1162 use opentelemetry_otlp::WithTonicConfig as _;
1163 b = b.with_tls_config(build_tls_config(m));
1164 }
1165 Ok(b.build()?)
1166 }
1167 #[cfg(feature = "http")]
1168 ExportProtocol::HttpProtobuf => {
1169 let mut b = opentelemetry_otlp::LogExporter::builder()
1170 .with_http()
1171 .with_endpoint(endpoint);
1172 if let Some(t) = timeout {
1173 b = b.with_timeout(t);
1174 }
1175 Ok(b.build()?)
1176 }
1177 }
1178}
1179
1180pub fn build_resource(
1195 service_name: &str,
1196 service_version: Option<&str>,
1197 deployment_environment: Option<&str>,
1198) -> Resource {
1199 let hostname = hostname::get()
1200 .ok()
1201 .and_then(|h| h.into_string().ok())
1202 .unwrap_or_default();
1203
1204 let mut builder = Resource::builder()
1205 .with_service_name(service_name.to_string())
1206 .with_attributes([
1207 KeyValue::new(HOST_NAME, hostname),
1208 KeyValue::new(PROCESS_PID, std::process::id() as i64),
1209 ]);
1210
1211 if let Some(version) = service_version {
1212 builder = builder.with_attribute(KeyValue::new(SERVICE_VERSION, version.to_string()));
1213 }
1214
1215 if let Some(env) = deployment_environment {
1216 builder =
1217 builder.with_attribute(KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, env.to_string()));
1218 }
1219
1220 builder.build()
1221}
1222
1223#[cfg(feature = "axum")]
1241pub fn axum_layer() -> axum_middleware::OtelTraceLayer {
1242 axum_middleware::OtelTraceLayer
1243}
1244
1245#[cfg(feature = "axum")]
1276pub fn span_enricher_layer<T>() -> axum_middleware::SpanEnricherLayer<T>
1277where
1278 T: span_enrichment::EnrichSpan + Clone + Send + Sync + 'static,
1279{
1280 axum_middleware::SpanEnricherLayer::default()
1281}
1282
1283#[cfg(feature = "tonic-tracing")]
1305pub fn grpc_client_layer() -> grpc_middleware::GrpcClientTraceLayer {
1306 grpc_middleware::GrpcClientTraceLayer
1307}
1308
1309#[cfg(feature = "tonic-tracing")]
1324pub fn grpc_server_layer() -> grpc_middleware::GrpcServerTraceLayer {
1325 grpc_middleware::GrpcServerTraceLayer
1326}
1327
1328#[cfg(test)]
1329mod tests {
1330 use super::*;
1331
1332 #[test]
1334 fn runtime_metrics_can_be_disabled() {
1335 assert!(
1336 Telemetry::builder("rm-default").runtime_metrics,
1337 "runtime metrics are on by default"
1338 );
1339 assert!(
1340 !Telemetry::builder("rm-off")
1341 .with_runtime_metrics(false)
1342 .runtime_metrics
1343 );
1344 }
1345
1346 #[tokio::test]
1354 async fn shutdown_absorbs_provider_errors() {
1355 let handles = TelemetryHandles {
1356 tracer_provider: SdkTracerProvider::builder().build(),
1357 meter_provider: Some(SdkMeterProvider::builder().build()),
1358 logger_provider: Some(SdkLoggerProvider::builder().build()),
1359 shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
1360 #[cfg(feature = "profiling")]
1361 profiling_handle: None,
1362 };
1363
1364 handles.shutdown().expect("first shutdown succeeds");
1365 handles
1366 .shutdown()
1367 .expect("second shutdown absorbs the already-shut-down errors");
1368 }
1369 use opentelemetry::trace::{Span as _, Tracer as _};
1370 use std::sync::Mutex;
1371
1372 static ENV_LOCK: Mutex<()> = Mutex::new(());
1373
1374 #[test]
1375 fn tracing_bridge_uses_sdk_tracer() {
1376 let provider = SdkTracerProvider::builder().build();
1377 let tracer = tracing_bridge_tracer(&provider);
1378 let span = tracer.start("bridge-regression");
1379
1380 assert!(span.span_context().is_valid());
1381
1382 provider.shutdown().expect("provider shutdown");
1383 }
1384
1385 #[test]
1386 fn resource_contains_all_attributes_when_provided() {
1387 let resource = build_resource("test-svc", Some("1.2.3"), Some("staging"));
1388
1389 assert_eq!(
1390 resource.get(&opentelemetry::Key::new("service.name")),
1391 Some(opentelemetry::Value::from("test-svc")),
1392 );
1393 assert_eq!(
1394 resource.get(&opentelemetry::Key::new(SERVICE_VERSION)),
1395 Some(opentelemetry::Value::from("1.2.3")),
1396 );
1397 assert_eq!(
1398 resource.get(&opentelemetry::Key::new(DEPLOYMENT_ENVIRONMENT_NAME)),
1399 Some(opentelemetry::Value::from("staging")),
1400 );
1401 assert!(resource.get(&opentelemetry::Key::new(HOST_NAME)).is_some());
1402 assert!(
1403 resource
1404 .get(&opentelemetry::Key::new(PROCESS_PID))
1405 .is_some()
1406 );
1407 }
1408
1409 #[test]
1410 fn resource_graceful_when_optional_values_omitted() {
1411 let resource = build_resource("test-svc", None, None);
1412
1413 assert_eq!(
1414 resource.get(&opentelemetry::Key::new("service.name")),
1415 Some(opentelemetry::Value::from("test-svc")),
1416 );
1417 assert!(
1418 resource
1419 .get(&opentelemetry::Key::new(SERVICE_VERSION))
1420 .is_none()
1421 );
1422 assert!(
1423 resource
1424 .get(&opentelemetry::Key::new(DEPLOYMENT_ENVIRONMENT_NAME))
1425 .is_none()
1426 );
1427 assert!(resource.get(&opentelemetry::Key::new(HOST_NAME)).is_some());
1429 assert!(
1430 resource
1431 .get(&opentelemetry::Key::new(PROCESS_PID))
1432 .is_some()
1433 );
1434 }
1435
1436 #[test]
1437 fn trace_sampler_ratio_converts_to_sdk() {
1438 let sampler = TraceSampler::TraceIdRatio(0.5);
1439 let sdk = sampler.into_sdk_sampler();
1440 assert_eq!(format!("{sdk:?}"), "TraceIdRatioBased(0.5)");
1441 }
1442
1443 #[test]
1444 fn trace_sampler_parent_based_converts_to_sdk() {
1445 let sampler = TraceSampler::ParentBased(Box::new(TraceSampler::TraceIdRatio(0.25)));
1446 let sdk = sampler.into_sdk_sampler();
1447 let debug = format!("{sdk:?}");
1448 assert!(debug.contains("ParentBased"));
1449 assert!(debug.contains("0.25"));
1450 }
1451
1452 unsafe fn set_env(key: &str, val: &str) {
1454 unsafe {
1455 std::env::set_var(key, val);
1456 }
1457 }
1458
1459 unsafe fn remove_env(key: &str) {
1460 unsafe {
1461 std::env::remove_var(key);
1462 }
1463 }
1464
1465 #[test]
1466 fn sampler_from_env_reads_traceidratio() {
1467 let _lock = ENV_LOCK.lock().unwrap();
1468 unsafe {
1469 set_env("OTEL_TRACES_SAMPLER", "traceidratio");
1470 set_env("OTEL_TRACES_SAMPLER_ARG", "0.42");
1471 }
1472
1473 let sampler = sampler_from_env()
1474 .expect("should not error")
1475 .expect("should return Some");
1476 assert!(
1477 matches!(sampler, TraceSampler::TraceIdRatio(r) if (r - 0.42).abs() < f64::EPSILON)
1478 );
1479
1480 unsafe {
1481 remove_env("OTEL_TRACES_SAMPLER");
1482 remove_env("OTEL_TRACES_SAMPLER_ARG");
1483 }
1484 }
1485
1486 #[test]
1487 fn sampler_from_env_returns_none_when_unset() {
1488 let _lock = ENV_LOCK.lock().unwrap();
1489 unsafe {
1490 remove_env("OTEL_TRACES_SAMPLER");
1491 }
1492 assert!(sampler_from_env().expect("should not error").is_none());
1493 }
1494
1495 #[test]
1496 fn sampler_from_env_reads_parentbased_traceidratio() {
1497 let _lock = ENV_LOCK.lock().unwrap();
1498 unsafe {
1499 set_env("OTEL_TRACES_SAMPLER", "parentbased_traceidratio");
1500 set_env("OTEL_TRACES_SAMPLER_ARG", "0.1");
1501 }
1502
1503 let sampler = sampler_from_env()
1504 .expect("should not error")
1505 .expect("should return Some");
1506 assert!(
1507 matches!(sampler, TraceSampler::ParentBased(inner) if matches!(*inner, TraceSampler::TraceIdRatio(r) if (r - 0.1).abs() < f64::EPSILON))
1508 );
1509
1510 unsafe {
1511 remove_env("OTEL_TRACES_SAMPLER");
1512 remove_env("OTEL_TRACES_SAMPLER_ARG");
1513 }
1514 }
1515
1516 #[test]
1517 fn sampler_from_env_parentbased_always_on() {
1518 let _lock = ENV_LOCK.lock().unwrap();
1519 unsafe {
1520 set_env("OTEL_TRACES_SAMPLER", "parentbased_always_on");
1521 }
1522 let sampler = sampler_from_env()
1523 .expect("should not error")
1524 .expect("should return Some");
1525 assert!(
1526 matches!(sampler, TraceSampler::ParentBased(inner) if matches!(*inner, TraceSampler::AlwaysOn))
1527 );
1528 unsafe {
1529 remove_env("OTEL_TRACES_SAMPLER");
1530 }
1531 }
1532
1533 #[test]
1534 fn sampler_from_env_parentbased_always_off() {
1535 let _lock = ENV_LOCK.lock().unwrap();
1536 unsafe {
1537 set_env("OTEL_TRACES_SAMPLER", "parentbased_always_off");
1538 }
1539 let sampler = sampler_from_env()
1540 .expect("should not error")
1541 .expect("should return Some");
1542 assert!(
1543 matches!(sampler, TraceSampler::ParentBased(inner) if matches!(*inner, TraceSampler::AlwaysOff))
1544 );
1545 unsafe {
1546 remove_env("OTEL_TRACES_SAMPLER");
1547 }
1548 }
1549
1550 #[test]
1551 fn sampler_from_env_always_on() {
1552 let _lock = ENV_LOCK.lock().unwrap();
1553 unsafe {
1554 set_env("OTEL_TRACES_SAMPLER", "always_on");
1555 }
1556 let sampler = sampler_from_env()
1557 .expect("should not error")
1558 .expect("should return Some");
1559 assert!(matches!(sampler, TraceSampler::AlwaysOn));
1560 unsafe {
1561 remove_env("OTEL_TRACES_SAMPLER");
1562 }
1563 }
1564
1565 #[test]
1566 fn sampler_from_env_always_off() {
1567 let _lock = ENV_LOCK.lock().unwrap();
1568 unsafe {
1569 set_env("OTEL_TRACES_SAMPLER", "always_off");
1570 }
1571 let sampler = sampler_from_env()
1572 .expect("should not error")
1573 .expect("should return Some");
1574 assert!(matches!(sampler, TraceSampler::AlwaysOff));
1575 unsafe {
1576 remove_env("OTEL_TRACES_SAMPLER");
1577 }
1578 }
1579
1580 #[test]
1581 fn sampler_from_env_unknown_returns_error() {
1582 let _lock = ENV_LOCK.lock().unwrap();
1583 unsafe {
1584 set_env("OTEL_TRACES_SAMPLER", "unknown_sampler");
1585 }
1586 let err = sampler_from_env().expect_err("unknown sampler should produce an error");
1587 assert!(
1588 err.to_string().contains("unknown_sampler"),
1589 "error message should include the unknown name, got: {err}"
1590 );
1591 unsafe {
1592 remove_env("OTEL_TRACES_SAMPLER");
1593 }
1594 }
1595
1596 #[test]
1597 fn trace_sampler_always_on_converts_to_sdk() {
1598 let sdk = TraceSampler::AlwaysOn.into_sdk_sampler();
1599 assert_eq!(format!("{sdk:?}"), "AlwaysOn");
1600 }
1601
1602 #[test]
1603 fn trace_sampler_always_off_converts_to_sdk() {
1604 let sdk = TraceSampler::AlwaysOff.into_sdk_sampler();
1605 assert_eq!(format!("{sdk:?}"), "AlwaysOff");
1606 }
1607
1608 #[test]
1609 fn builder_has_sensible_defaults() {
1610 let builder = Telemetry::builder("test-svc");
1611 assert_eq!(builder.service_name.as_deref(), Some("test-svc"));
1612 assert!(builder.service_version.is_none());
1613 assert!(builder.deployment_environment.is_none());
1614 assert!(builder.sampler.is_none());
1615 assert!(builder.metrics);
1616 assert!(!builder.logs);
1617 assert!(builder.protocol.is_none());
1618 assert!(builder.max_export_batch_size.is_none());
1619 assert!(builder.metric_export_interval.is_none());
1620 assert!(builder.export_timeout.is_none());
1621 }
1622
1623 #[test]
1624 fn from_env_builder_has_no_service_name() {
1625 let builder = Telemetry::from_env();
1626 assert!(builder.service_name.is_none());
1627 }
1628
1629 #[test]
1630 fn with_export_timeout_stores_value() {
1631 let timeout = Duration::from_secs(5);
1632 let builder = Telemetry::builder("test-svc").with_export_timeout(timeout);
1633 assert_eq!(builder.export_timeout, Some(timeout));
1634 }
1635
1636 #[test]
1637 fn timeout_from_env_reads_milliseconds() {
1638 let _lock = ENV_LOCK.lock().unwrap();
1639 unsafe {
1640 set_env("OTEL_EXPORTER_OTLP_TIMEOUT", "5000");
1641 }
1642 let t = timeout_from_env();
1643 assert_eq!(t, Some(Duration::from_millis(5000)));
1644 unsafe {
1645 remove_env("OTEL_EXPORTER_OTLP_TIMEOUT");
1646 }
1647 }
1648
1649 #[test]
1650 fn timeout_from_env_returns_none_when_unset() {
1651 let _lock = ENV_LOCK.lock().unwrap();
1652 unsafe {
1653 remove_env("OTEL_EXPORTER_OTLP_TIMEOUT");
1654 }
1655 assert_eq!(timeout_from_env(), None);
1656 }
1657
1658 #[test]
1659 fn service_name_from_env_used_when_none_given() {
1660 let builder = Telemetry::from_env();
1661 assert!(builder.service_name.is_none());
1662 }
1663
1664 #[test]
1665 fn explicit_service_name_overrides_env_var() {
1666 let builder = Telemetry::builder("explicit-svc");
1667 assert_eq!(builder.service_name.as_deref(), Some("explicit-svc"));
1668 }
1669
1670 #[test]
1671 fn from_env_builder_service_name_is_none() {
1672 let builder = Telemetry::from_env();
1673 assert!(builder.service_name.is_none());
1674 }
1675
1676 #[test]
1677 fn init_returns_error_for_unknown_otel_traces_sampler() {
1678 let _lock = ENV_LOCK.lock().unwrap();
1679 unsafe {
1680 set_env("OTEL_TRACES_SAMPLER", "not_a_real_sampler");
1681 }
1682 let result = Telemetry::builder("test-svc").with_metrics(false).init();
1683 let err = result
1684 .err()
1685 .expect("unknown sampler env var should cause init to fail");
1686 assert!(
1687 err.to_string().contains("not_a_real_sampler"),
1688 "error should name the unknown sampler, got: {err}"
1689 );
1690 unsafe {
1691 remove_env("OTEL_TRACES_SAMPLER");
1692 }
1693 }
1694
1695 #[test]
1696 fn with_max_export_batch_size_stores_value() {
1697 let builder = Telemetry::builder("test-svc").with_max_export_batch_size(1024);
1698 assert_eq!(builder.max_export_batch_size, Some(1024));
1699 }
1700
1701 #[test]
1702 fn with_metric_export_interval_stores_value() {
1703 let interval = Duration::from_secs(30);
1704 let builder = Telemetry::builder("test-svc").with_metric_export_interval(interval);
1705 assert_eq!(builder.metric_export_interval, Some(interval));
1706 }
1707
1708 #[test]
1709 fn init_rejects_zero_metric_export_interval() {
1710 let err = Telemetry::builder("test-svc")
1711 .with_metric_export_interval(Duration::ZERO)
1712 .with_metrics(false)
1713 .init()
1714 .err()
1715 .expect("expected error for zero interval");
1716 assert!(
1717 err.to_string().contains("metric_export_interval"),
1718 "error message should mention metric_export_interval, got: {err}"
1719 );
1720 }
1721
1722 #[test]
1723 fn builder_with_custom_values() {
1724 let builder = Telemetry::builder("test-svc")
1725 .with_version("2.0.0")
1726 .with_environment("production")
1727 .with_sampler(TraceSampler::TraceIdRatio(0.5))
1728 .with_metrics(false);
1729
1730 assert_eq!(builder.service_name.as_deref(), Some("test-svc"));
1731 assert_eq!(builder.service_version.as_deref(), Some("2.0.0"));
1732 assert_eq!(
1733 builder.deployment_environment.as_deref(),
1734 Some("production")
1735 );
1736 assert!(
1737 matches!(builder.sampler, Some(TraceSampler::TraceIdRatio(r)) if (r - 0.5).abs() < f64::EPSILON)
1738 );
1739 assert!(!builder.metrics);
1740 }
1741
1742 #[test]
1743 fn builder_stores_programmatic_log_configuration() {
1744 let builder = Telemetry::builder("test-svc")
1745 .with_log_filter("info,opentelemetry_sdk=warn")
1746 .with_log_format(LogFormat::Json);
1747
1748 assert_eq!(
1749 builder.log_filter.as_deref(),
1750 Some("info,opentelemetry_sdk=warn")
1751 );
1752 assert_eq!(builder.log_format, LogFormat::Json);
1753 }
1754
1755 #[test]
1756 fn init_rejects_invalid_programmatic_log_filter_before_provider_setup() {
1757 let setup_ran = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
1758 let setup_ran_in_closure = std::sync::Arc::clone(&setup_ran);
1759
1760 let error = Telemetry::builder("test-svc")
1761 .with_log_filter("[")
1762 .with_meter_provider_setup(move |builder| {
1763 setup_ran_in_closure.store(true, std::sync::atomic::Ordering::SeqCst);
1764 builder
1765 })
1766 .init()
1767 .err()
1768 .expect("invalid filter must fail initialization");
1769
1770 assert!(error.to_string().contains("invalid filter directive"));
1771 assert!(!setup_ran.load(std::sync::atomic::Ordering::SeqCst));
1772 }
1773
1774 #[test]
1775 #[cfg(feature = "grpc")]
1776 fn builder_with_protocol_grpc() {
1777 let builder = Telemetry::builder("test-svc").with_protocol(ExportProtocol::Grpc);
1778 assert_eq!(builder.protocol, Some(ExportProtocol::Grpc));
1779 }
1780
1781 #[test]
1782 #[cfg(feature = "http")]
1783 fn builder_with_protocol_http() {
1784 let builder = Telemetry::builder("test-svc").with_protocol(ExportProtocol::HttpProtobuf);
1785 assert_eq!(builder.protocol, Some(ExportProtocol::HttpProtobuf));
1786 }
1787
1788 #[test]
1789 #[cfg(feature = "grpc")]
1790 fn protocol_from_env_reads_grpc() {
1791 let _lock = ENV_LOCK.lock().unwrap();
1792 unsafe {
1793 set_env("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc");
1794 }
1795 assert_eq!(protocol_from_env(), Some(ExportProtocol::Grpc));
1796 unsafe {
1797 remove_env("OTEL_EXPORTER_OTLP_PROTOCOL");
1798 }
1799 }
1800
1801 #[test]
1802 #[cfg(feature = "http")]
1803 fn protocol_from_env_reads_http_protobuf() {
1804 let _lock = ENV_LOCK.lock().unwrap();
1805 unsafe {
1806 set_env("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf");
1807 }
1808 assert_eq!(protocol_from_env(), Some(ExportProtocol::HttpProtobuf));
1809 unsafe {
1810 remove_env("OTEL_EXPORTER_OTLP_PROTOCOL");
1811 }
1812 }
1813
1814 #[test]
1815 fn protocol_from_env_returns_none_when_unset() {
1816 let _lock = ENV_LOCK.lock().unwrap();
1817 unsafe {
1818 remove_env("OTEL_EXPORTER_OTLP_PROTOCOL");
1819 }
1820 assert_eq!(protocol_from_env(), None);
1821 }
1822
1823 #[test]
1824 fn protocol_from_env_returns_none_for_unknown() {
1825 let _lock = ENV_LOCK.lock().unwrap();
1826 unsafe {
1827 set_env("OTEL_EXPORTER_OTLP_PROTOCOL", "websocket");
1828 }
1829 assert_eq!(protocol_from_env(), None);
1830 unsafe {
1831 remove_env("OTEL_EXPORTER_OTLP_PROTOCOL");
1832 }
1833 }
1834
1835 #[test]
1836 fn builder_is_send_and_sync() {
1837 fn assert_send_sync<T: Send + Sync>() {}
1838 assert_send_sync::<TelemetryBuilder>();
1839 }
1840
1841 #[test]
1842 fn with_shutdown_timeout_stores_value() {
1843 let timeout = Duration::from_secs(10);
1844 let builder = Telemetry::builder("test-svc").with_shutdown_timeout(timeout);
1845 assert_eq!(builder.shutdown_timeout, timeout);
1846 }
1847
1848 #[test]
1849 fn default_shutdown_timeout_is_five_seconds() {
1850 let builder = Telemetry::builder("test-svc");
1851 assert_eq!(builder.shutdown_timeout, Duration::from_secs(5));
1852 }
1853
1854 #[cfg(feature = "testing")]
1862 #[test]
1863 fn drop_completes_within_shutdown_timeout() {
1864 let mut handles = crate::Telemetry::testing("drop-timeout-test");
1866 handles.shutdown_timeout = Duration::from_millis(100);
1868
1869 let start = std::time::Instant::now();
1870 drop(handles);
1871 let elapsed = start.elapsed();
1872
1873 assert!(
1875 elapsed < Duration::from_millis(500),
1876 "drop took {elapsed:?}, expected < 500 ms"
1877 );
1878 }
1879}