difftreelog
feat otel-exporter-env
in: trunk
2 files changed
crates/opentelemetry-exporter-env/src/lib.rsdiffbeforeafterboth1use std::collections::HashMap;2use std::convert::Infallible;1use std::convert::Infallible;3use std::env::{self, VarError};2use std::env::{self, VarError};4use std::ffi::OsString;3use std::ffi::OsString;5use std::num::ParseIntError;4use std::num::ParseIntError;6use std::str::FromStr;5use std::str::FromStr;7use std::time::Duration;6use std::time::Duration;89use clap::Parser;10#[cfg(feature = "otlp")]11use opentelemetry_otlp::tonic_types::metadata::MetadataMap;12#[cfg(feature = "otlp")]13use opentelemetry_otlp::{14 ExporterBuildError, LogExporter, MetricExporter, SpanExporter, WithExportConfig,15 WithHttpConfig, WithTonicConfig,16};17718#[cfg(feature = "otlp")]8#[cfg(feature = "otlp")]19mod otlp;9mod otlp;201011#[derive(thiserror::Error, Debug)]21pub enum Error {12pub enum Error {13 #[error("environment variable {env} contains invalid UTF-8: {value:?}")]22 InvalidUtf8 {14 InvalidUtf8 {23 env: &'static str,15 env: &'static str,24 value: OsString,16 value: OsString,25 },17 },18 #[error("environment variable {env}={value:?}: {error}")]26 EnvParseError {19 EnvParse {27 env: &'static str,20 env: &'static str,28 value: String,21 value: String,29 error: &'static str,22 error: &'static str,30 },23 },24 #[error("environment variable {env}={value:?}: {error}")]31 EnvParseIntError {25 EnvParseInt {32 env: &'static str,26 env: &'static str,33 value: String,27 value: String,34 error: ParseIntError,28 error: ParseIntError,35 },29 },30 #[cfg(feature = "otlp")]31 #[error("failed to build exporter: {0}")]32 Exporter(#[from] opentelemetry_otlp::ExporterBuildError),36}33}3437impl From<(&'static str, &'static str, String)> for Error {35impl From<(&'static str, &'static str, String)> for Error {38 fn from((env, error, value): (&'static str, &'static str, String)) -> Self {36 fn from((env, error, value): (&'static str, &'static str, String)) -> Self {39 Self::EnvParseError { env, value, error }37 Self::EnvParse { env, value, error }40 }38 }41}39}42impl From<(&'static str, ParseIntError, String)> for Error {40impl From<(&'static str, ParseIntError, String)> for Error {43 fn from((env, error, value): (&'static str, ParseIntError, String)) -> Self {41 fn from((env, error, value): (&'static str, ParseIntError, String)) -> Self {44 Self::EnvParseIntError { env, value, error }42 Self::EnvParseInt { env, value, error }45 }43 }46}44}47impl From<(&'static str, Infallible, String)> for Error {45impl From<(&'static str, Infallible, String)> for Error {62 }60 }63}61}646265macro_rules! impl_settings {66 (67 #[name($env_prefix:literal, $long_prefix:literal)]68 struct $id:ident {69 $(70 $(#[doc = $doc:literal])*71 #[name($env:literal, $long:literal)]72 $(#[arg($($tt:tt)*)])?73 $name:ident: $ty:ty,74 )*75 }) => {76 #[derive(Parser)]77 pub struct $id {78 $(79 $(#[doc = $doc])*80 #[arg(long = concat!("otel-exporter-otlp-", $long_prefix, $long), env = concat!("OTEL_EXPORTER_OTLP_", $env_prefix, $env), $($($tt)*)?)]81 pub $name: Option<$ty>,82 )*83 }84 impl $id {85 pub fn from_env() -> Result<Self, Error> {86 Ok(Self {87 $(88 $name: load_env(concat!("OTEL_EXPORTER_OTLP_", $env_prefix, $env))?,89 )*90 })91 }92 }93 }94}95macro_rules! impl_enum {63macro_rules! impl_enum {96 (enum $id:ident {64 (enum $id:ident {97 $(65 $(115 $(83 $(116 $value => Self::$var,84 $value => Self::$var,117 )*85 )*118 _ => return Err("unsupported value, supported are")86 _ => return Err("unsupported value")119 })87 })120 }88 }121 }89 }122 };90 };123}91}9293impl_enum! {94 enum ExporterKind {95 #[name = "otlp"]96 Otlp,97 #[name = "none"]98 None,99 }100}101102#[derive(Default)]103#[cfg_attr(feature = "clap", derive(clap::Parser))]104pub struct SignalExporterSettings {105 /// Traces exporter to be used.106 #[cfg_attr(feature = "clap", arg(long = "otel-traces-exporter", env = "OTEL_TRACES_EXPORTER", value_enum))]107 pub traces: Option<ExporterKind>,108 /// Metrics exporter to be used.109 #[cfg_attr(feature = "clap", arg(long = "otel-metrics-exporter", env = "OTEL_METRICS_EXPORTER", value_enum))]110 pub metrics: Option<ExporterKind>,111 /// Logs exporter to be used.112 #[cfg_attr(feature = "clap", arg(long = "otel-logs-exporter", env = "OTEL_LOGS_EXPORTER", value_enum))]113 pub logs: Option<ExporterKind>,114}115116impl SignalExporterSettings {117 pub fn from_env() -> Result<Self, Error> {118 Ok(Self {119 traces: load_env("OTEL_TRACES_EXPORTER")?,120 metrics: load_env("OTEL_METRICS_EXPORTER")?,121 logs: load_env("OTEL_LOGS_EXPORTER")?,122 })123 }124125 pub fn traces_enabled(&self) -> bool {126 !matches!(self.traces, Some(ExporterKind::None))127 }128 pub fn metrics_enabled(&self) -> bool {129 !matches!(self.metrics, Some(ExporterKind::None))130 }131 pub fn logs_enabled(&self) -> bool {132 !matches!(self.logs, Some(ExporterKind::None))133 }134}124135125impl_enum! {136impl_enum! {126 enum Compression {137 enum Compression {161 }172 }162}173}174175pub trait OtlpSignalSettings {176 fn compression(&self) -> Option<Compression>;177 fn endpoint(&self) -> Option<&str>;178 fn headers(&self) -> Option<&str>;179 fn protocol(&self) -> Option<OtlpProtocol>;180 fn timeout(&self) -> Option<u64>;181}182183macro_rules! impl_settings {184 (185 #[name($env_prefix:literal, $long_prefix:literal)]186 struct $id:ident {187 $(188 $(#[doc = $doc:literal])*189 #[name($env:literal, $long:literal)]190 $(#[arg($($tt:tt)*)])?191 $name:ident: $ty:ty,192 )*193 }) => {194 #[derive(Default)]195 #[cfg_attr(feature = "clap", derive(clap::Parser))]196 pub struct $id {197 $(198 $(#[doc = $doc])*199 #[cfg_attr(feature = "clap", arg(long = concat!("otel-exporter-otlp-", $long_prefix, $long), env = concat!("OTEL_EXPORTER_OTLP_", $env_prefix, $env) $(, $($tt)*)?))]200 pub $name: Option<$ty>,201 )*202 }203 impl $id {204 pub fn from_env() -> Result<Self, Error> {205 Ok(Self {206 $(207 $name: load_env(concat!("OTEL_EXPORTER_OTLP_", $env_prefix, $env))?,208 )*209 })210 }211 }212 impl OtlpSignalSettings for $id {213 fn compression(&self) -> Option<Compression> { self.compression }214 fn endpoint(&self) -> Option<&str> { self.endpoint.as_deref() }215 fn headers(&self) -> Option<&str> { self.headers.as_deref() }216 fn protocol(&self) -> Option<OtlpProtocol> { self.protocol }217 fn timeout(&self) -> Option<u64> { self.timeout }218 }219 }220}163221164impl_settings! {222impl_settings! {165 #[name("", "")]223 #[name("", "")]168 #[name("COMPRESSION", "compression")]226 #[name("COMPRESSION", "compression")]169 #[arg(value_enum)]227 #[arg(value_enum)]170 compression: Compression,228 compression: Compression,171 /// A base endpoint URL for any signal type, with an optionally-specified port number. Helpful for when you’re sending more than one signal to the same endpoint and want one environment variable to control the endpoint.229 /// A base endpoint URL for any signal type, with an optionally-specified port number. Helpful for when you're sending more than one signal to the same endpoint and want one environment variable to control the endpoint.172 #[name("ENDPOINT", "endpoint")]230 #[name("ENDPOINT", "endpoint")]173 endpoint: String,231 endpoint: String,174 /// A list of headers to apply to all outgoing data (traces, metrics, and logs).232 /// A list of headers to apply to all outgoing data (traces, metrics, and logs).251 }308 }252}309}253310311pub struct ResolvedOtlpSettings {312 pub compression: Option<Compression>,313 pub endpoint: String,314 pub headers: Option<String>,315 pub protocol: OtlpProtocol,316 pub timeout: Duration,317}318319impl ResolvedOtlpSettings {320 const DEFAULT_TIMEOUT_MS: u64 = 10000;321 const DEFAULT_GRPC_ENDPOINT: &str = "http://localhost:4317";322 const DEFAULT_HTTP_ENDPOINT: &str = "http://localhost:4318";323324 pub fn traces(325 base: &impl OtlpSignalSettings,326 signal: &impl OtlpSignalSettings,327 ) -> Result<Self, Error> {254#[derive(thiserror::Error, Debug)]328 Self::resolve(base, signal, "/v1/traces")329 }330331 pub fn metrics(332 base: &impl OtlpSignalSettings,333 signal: &impl OtlpSignalSettings,334 ) -> Result<Self, Error> {335 Self::resolve(base, signal, "/v1/metrics")336 }337338 pub fn logs(339 base: &impl OtlpSignalSettings,340 signal: &impl OtlpSignalSettings,341 ) -> Result<Self, Error> {342 Self::resolve(base, signal, "/v1/logs")343 }344345 fn resolve(346 base: &impl OtlpSignalSettings,347 signal: &impl OtlpSignalSettings,348 signal_path: &str,255enum ProviderError {349 ) -> Result<Self, Error> {350 let protocol = signal351 .protocol()256 #[error("protocol is not set")]352 .or_else(|| base.protocol())353 .unwrap_or(OtlpProtocol::HttpProtobuf);354355 let endpoint = if let Some(ep) = signal.endpoint() {356 ep.to_owned()357 } else if let Some(ep) = base.endpoint() {358 match protocol {257 UnsetProtocol,359 OtlpProtocol::Grpc => ep.to_owned(),360 _ => format!("{ep}{signal_path}"),361 }362 } else {363 match protocol {364 OtlpProtocol::Grpc => Self::DEFAULT_GRPC_ENDPOINT.to_owned(),258 #[error("endpoint is not set")]365 _ => format!("{}{signal_path}", Self::DEFAULT_HTTP_ENDPOINT),259 EndpointUnset,366 }367 };368369 Ok(Self {260 #[cfg(feature = "otlp")]370 compression: signal.compression().or_else(|| base.compression()),371 endpoint,372 headers: signal373 .headers()374 .or_else(|| base.headers())261 #[error("failed to build exporter: {0}")]375 .map(str::to_owned),262 Exporter(#[from] ExporterBuildError),376 protocol,377 timeout: Duration::from_millis(378 signal379 .timeout()380 .or_else(|| base.timeout())381 .unwrap_or(Self::DEFAULT_TIMEOUT_MS),382 ),383 })263}384 }264type ProviderResult<T, E = ProviderError> = Result<T, E>;385}265386crates/opentelemetry-exporter-env/src/otlp.rsdiffbeforeafterboth--- a/crates/opentelemetry-exporter-env/src/otlp.rs
+++ b/crates/opentelemetry-exporter-env/src/otlp.rs
@@ -1,5 +1,4 @@
use std::collections::HashMap;
-use std::time::Duration;
use opentelemetry_otlp::tonic_types::metadata::MetadataMap;
use opentelemetry_otlp::{
@@ -7,14 +6,9 @@
WithTonicConfig as _,
};
-use crate::{
- OtlpBaseSettings, OtlpLogsSettings, OtlpMetricsSettings, OtlpProtocol, ProviderError,
- ProviderResult,
-};
+use crate::{Error, OtlpProtocol, ResolvedOtlpSettings};
-fn parse_headers<'a>(
- headers: &'a str,
-) -> std::iter::Map<std::str::Split<'a, char>, impl FnMut(&'a str) -> (&'a str, &'a str)> {
+fn parse_headers(headers: &str) -> impl Iterator<Item = (&str, &str)> {
headers.split(',').map(|header| {
let mut parts = header.splitn(2, '=');
let key = parts.next().unwrap();
@@ -23,7 +17,7 @@
})
}
-fn parse_headers_metadata_map(headers: Option<&str>) -> MetadataMap {
+fn to_metadata_map(headers: Option<&str>) -> MetadataMap {
headers
.map(|headers| {
MetadataMap::from_headers(
@@ -34,7 +28,8 @@
})
.unwrap_or_default()
}
-fn parse_headers_hashmap(headers: Option<&str>) -> HashMap<String, String> {
+
+fn to_hashmap(headers: Option<&str>) -> HashMap<String, String> {
headers
.map(|headers| {
parse_headers(headers)
@@ -44,132 +39,45 @@
.unwrap_or_default()
}
-fn logger_exporter(base: &OtlpBaseSettings, log: &OtlpLogsSettings) -> ProviderResult<LogExporter> {
- let endpoint = log
- .endpoint
- .clone()
- .or_else(|| Some(format!("{}/v1/logs", base.endpoint.as_ref()?)))
- .ok_or(ProviderError::EndpointUnset)?;
- let headers = log.headers.as_deref().or(base.headers.as_deref());
- let timeout = Duration::from_millis(log.timeout.or(base.timeout).unwrap_or(10000));
-
- let protocol = log
- .protocol
- .or(base.protocol)
- .ok_or(ProviderError::UnsetProtocol)?;
-
- match protocol {
- OtlpProtocol::Grpc => {
- let mut builder = LogExporter::builder()
- .with_tonic()
- .with_endpoint(endpoint)
- .with_metadata(parse_headers_metadata_map(headers))
- .with_protocol(protocol.into())
- .with_timeout(timeout);
- let compression = log.compression.or(base.compression);
- if let Some(compression) = compression {
- builder = builder.with_compression(compression.into());
+macro_rules! build_exporter {
+ ($exporter:ty, $settings:expr) => {{
+ let s: &ResolvedOtlpSettings = $settings;
+ match s.protocol {
+ OtlpProtocol::Grpc => {
+ let mut builder = <$exporter>::builder()
+ .with_tonic()
+ .with_endpoint(&s.endpoint)
+ .with_metadata(to_metadata_map(s.headers.as_deref()))
+ .with_protocol(s.protocol.into())
+ .with_timeout(s.timeout);
+ if let Some(compression) = s.compression {
+ builder = builder.with_compression(compression.into());
+ }
+ builder.build()
+ }
+ OtlpProtocol::HttpProtobuf | OtlpProtocol::HttpJson => {
+ <$exporter>::builder()
+ .with_http()
+ .with_endpoint(&s.endpoint)
+ .with_headers(to_hashmap(s.headers.as_deref()))
+ .with_protocol(s.protocol.into())
+ .with_timeout(s.timeout)
+ .build()
}
-
- Ok(builder.build()?)
}
- OtlpProtocol::HttpProtobuf | OtlpProtocol::HttpJson => {
- let builder = LogExporter::builder()
- .with_http()
- .with_endpoint(endpoint)
- .with_headers(parse_headers_hashmap(headers))
- .with_protocol(protocol.into())
- .with_timeout(timeout);
+ }};
+}
- Ok(builder.build()?)
- }
+impl ResolvedOtlpSettings {
+ pub fn span_exporter(&self) -> Result<SpanExporter, Error> {
+ Ok(build_exporter!(SpanExporter, self)?)
}
-}
-fn metric_exporter(
- base: &OtlpBaseSettings,
- metric: &OtlpMetricsSettings,
-) -> ProviderResult<MetricExporter> {
- let endpoint = metric
- .endpoint
- .clone()
- .or_else(|| Some(format!("{}/v1/metrics", base.endpoint.as_ref()?)))
- .ok_or(ProviderError::EndpointUnset)?;
- let headers = metric.headers.as_deref().or(base.headers.as_deref());
- let timeout = Duration::from_millis(metric.timeout.or(base.timeout).unwrap_or(10000));
- let protocol = metric
- .protocol
- .or(base.protocol)
- .ok_or(ProviderError::UnsetProtocol)?;
-
- match protocol {
- OtlpProtocol::Grpc => {
- let mut builder = MetricExporter::builder()
- .with_tonic()
- .with_endpoint(endpoint)
- .with_metadata(parse_headers_metadata_map(headers))
- .with_protocol(protocol.into())
- .with_timeout(timeout);
- let compression = metric.compression.or(base.compression);
- if let Some(compression) = compression {
- builder = builder.with_compression(compression.into());
- }
-
- Ok(builder.build()?)
- }
- OtlpProtocol::HttpProtobuf | OtlpProtocol::HttpJson => {
- let builder = MetricExporter::builder()
- .with_http()
- .with_endpoint(endpoint)
- .with_headers(parse_headers_hashmap(headers))
- .with_protocol(protocol.into())
- .with_timeout(timeout);
-
- Ok(builder.build()?)
- }
+ pub fn log_exporter(&self) -> Result<LogExporter, Error> {
+ Ok(build_exporter!(LogExporter, self)?)
}
-}
-fn span_exporter(
- base: &OtlpBaseSettings,
- trace: &OtlpMetricsSettings,
-) -> ProviderResult<SpanExporter> {
- let endpoint = trace
- .endpoint
- .clone()
- .or_else(|| Some(format!("{}/v1/traces", base.endpoint.as_ref()?)))
- .ok_or(ProviderError::EndpointUnset)?;
- let headers = trace.headers.as_deref().or(base.headers.as_deref());
- let timeout = Duration::from_millis(trace.timeout.or(base.timeout).unwrap_or(10000));
- let protocol = trace
- .protocol
- .or(base.protocol)
- .ok_or(ProviderError::UnsetProtocol)?;
-
- match protocol {
- OtlpProtocol::Grpc => {
- let mut builder = SpanExporter::builder()
- .with_tonic()
- .with_endpoint(endpoint)
- .with_metadata(parse_headers_metadata_map(headers))
- .with_protocol(protocol.into())
- .with_timeout(timeout);
- let compression = trace.compression.or(base.compression);
- if let Some(compression) = compression {
- builder = builder.with_compression(compression.into());
- }
-
- Ok(builder.build()?)
- }
- OtlpProtocol::HttpProtobuf | OtlpProtocol::HttpJson => {
- let builder = SpanExporter::builder()
- .with_http()
- .with_endpoint(endpoint)
- .with_headers(parse_headers_hashmap(headers))
- .with_protocol(protocol.into())
- .with_timeout(timeout);
-
- Ok(builder.build()?)
- }
+ pub fn metric_exporter(&self) -> Result<MetricExporter, Error> {
+ Ok(build_exporter!(MetricExporter, self)?)
}
}