git.delta.rocks / fleet / refs/commits / 9766610438d2

1use std::convert::Infallible;2use std::env::{self, VarError};3use std::ffi::OsString;4use std::num::ParseIntError;5use std::str::FromStr;6use std::time::Duration;78#[cfg(feature = "otlp")]9mod otlp;1011#[derive(thiserror::Error, Debug)]12pub enum Error {13	#[error("environment variable {env} contains invalid UTF-8: {value:?}")]14	InvalidUtf8 { env: &'static str, value: OsString },15	#[error("environment variable {env}={value:?}: {error}")]16	EnvParse {17		env: &'static str,18		value: String,19		error: &'static str,20	},21	#[error("environment variable {env}={value:?}: {error}")]22	EnvParseInt {23		env: &'static str,24		value: String,25		error: ParseIntError,26	},27	#[cfg(feature = "otlp")]28	#[error("failed to build exporter: {0}")]29	Exporter(#[from] opentelemetry_otlp::ExporterBuildError),30}3132impl From<(&'static str, &'static str, String)> for Error {33	fn from((env, error, value): (&'static str, &'static str, String)) -> Self {34		Self::EnvParse { env, value, error }35	}36}37impl From<(&'static str, ParseIntError, String)> for Error {38	fn from((env, error, value): (&'static str, ParseIntError, String)) -> Self {39		Self::EnvParseInt { env, value, error }40	}41}42impl From<(&'static str, Infallible, String)> for Error {43	fn from(_v: (&'static str, Infallible, String)) -> Self {44		unreachable!()45	}46}4748fn load_env<T>(env: &'static str) -> Result<Option<T>, Error>49where50	T: FromStr,51	Error: From<(&'static str, <T as FromStr>::Err, String)>,52{53	match env::var(env) {54		Ok(v) => Ok(Some(T::from_str(&v).map_err(|err| (env, err, v))?)),55		Err(VarError::NotPresent) => Ok(None),56		Err(VarError::NotUnicode(value)) => Err(Error::InvalidUtf8 { env, value }),57	}58}5960macro_rules! impl_enum {61	(enum $id:ident {62		$(63			#[name = $value:literal]64			$var:ident,65		)*66	}) => {67		#[derive(Clone, Copy)]68		#[cfg_attr(feature = "clap", derive(clap::ValueEnum))]69		pub enum $id {70			$(71				#[cfg_attr(feature = "clap", value(name = $value))]72				$var,73			)*74		}75		impl FromStr for $id {76			type Err = &'static str;7778			fn from_str(s: &str) -> Result<Self, Self::Err> {79				Ok(match s {80					$(81						$value => Self::$var,82					)*83					_ => return Err("unsupported value")84				})85			}86		}87	};88}8990impl_enum! {91	enum ExporterKind {92		#[name = "otlp"]93		Otlp,94		#[name = "none"]95		None,96	}97}9899#[derive(Default)]100#[cfg_attr(feature = "clap", derive(clap::Parser))]101pub struct SignalExporterSettings {102	/// Traces exporter to be used.103	#[cfg_attr(104		feature = "clap",105		arg(106			long = "otel-traces-exporter",107			env = "OTEL_TRACES_EXPORTER",108			value_enum109		)110	)]111	pub traces: Option<ExporterKind>,112	/// Metrics exporter to be used.113	#[cfg_attr(114		feature = "clap",115		arg(116			long = "otel-metrics-exporter",117			env = "OTEL_METRICS_EXPORTER",118			value_enum119		)120	)]121	pub metrics: Option<ExporterKind>,122	/// Logs exporter to be used.123	#[cfg_attr(124		feature = "clap",125		arg(long = "otel-logs-exporter", env = "OTEL_LOGS_EXPORTER", value_enum)126	)]127	pub logs: Option<ExporterKind>,128}129130impl SignalExporterSettings {131	pub fn from_env() -> Result<Self, Error> {132		Ok(Self {133			traces: load_env("OTEL_TRACES_EXPORTER")?,134			metrics: load_env("OTEL_METRICS_EXPORTER")?,135			logs: load_env("OTEL_LOGS_EXPORTER")?,136		})137	}138139	pub fn traces_enabled(&self) -> bool {140		!matches!(self.traces, Some(ExporterKind::None))141	}142	pub fn metrics_enabled(&self) -> bool {143		!matches!(self.metrics, Some(ExporterKind::None))144	}145	pub fn logs_enabled(&self) -> bool {146		!matches!(self.logs, Some(ExporterKind::None))147	}148}149150impl_enum! {151	enum Compression {152		#[name = "gzip"]153		Gzip,154		#[name = "zstd"]155		Zstd,156	}157}158#[cfg(feature = "otlp")]159impl From<Compression> for opentelemetry_otlp::Compression {160	fn from(value: Compression) -> Self {161		match value {162			Compression::Gzip => opentelemetry_otlp::Compression::Gzip,163			Compression::Zstd => opentelemetry_otlp::Compression::Zstd,164		}165	}166}167168impl_enum! {169	enum OtlpProtocol {170		#[name = "grpc"]171		Grpc,172		#[name = "http/protobuf"]173		HttpProtobuf,174		#[name = "http/json"]175		HttpJson,176	}177}178#[cfg(feature = "otlp")]179impl From<OtlpProtocol> for opentelemetry_otlp::Protocol {180	fn from(value: OtlpProtocol) -> Self {181		match value {182			OtlpProtocol::Grpc => opentelemetry_otlp::Protocol::Grpc,183			OtlpProtocol::HttpProtobuf => opentelemetry_otlp::Protocol::HttpBinary,184			OtlpProtocol::HttpJson => opentelemetry_otlp::Protocol::HttpJson,185		}186	}187}188189pub trait OtlpSignalSettings {190	fn compression(&self) -> Option<Compression>;191	fn endpoint(&self) -> Option<&str>;192	fn headers(&self) -> Option<&str>;193	fn protocol(&self) -> Option<OtlpProtocol>;194	fn timeout(&self) -> Option<u64>;195}196197macro_rules! impl_settings {198	(199	#[name($env_prefix:literal, $long_prefix:literal)]200	struct $id:ident {201		$(202			$(#[doc = $doc:literal])*203			#[name($env:literal, $long:literal)]204			$(#[arg($($tt:tt)*)])?205			$name:ident: $ty:ty,206		)*207	}) => {208		#[derive(Default)]209		#[cfg_attr(feature = "clap", derive(clap::Parser))]210		pub struct $id {211			$(212				$(#[doc = $doc])*213				#[cfg_attr(feature = "clap", arg(214						long = concat!("otel-exporter-otlp-", $long_prefix, $long),215						id = concat!("otel-exporter-otlp-", $long_prefix, $long),216						env = concat!("OTEL_EXPORTER_OTLP_", $env_prefix, $env)217						$(, $($tt)*)?)218				)]219				pub $name: Option<$ty>,220			)*221		}222		impl $id {223			pub fn from_env() -> Result<Self, Error> {224				Ok(Self {225					$(226						$name: load_env(concat!("OTEL_EXPORTER_OTLP_", $env_prefix, $env))?,227					)*228				})229			}230		}231		impl OtlpSignalSettings for $id {232			fn compression(&self) -> Option<Compression> { self.compression }233			fn endpoint(&self) -> Option<&str> { self.endpoint.as_deref() }234			fn headers(&self) -> Option<&str> { self.headers.as_deref() }235			fn protocol(&self) -> Option<OtlpProtocol> { self.protocol }236			fn timeout(&self) -> Option<u64> { self.timeout }237		}238	}239}240241impl_settings! {242	#[name("", "")]243	struct OtlpBaseSettings {244		/// Specifies the OTLP transport compression to be used for all telemetry data.245		#[name("COMPRESSION", "compression")]246		#[arg(value_enum)]247		compression: Compression,248		/// 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.249		#[name("ENDPOINT", "endpoint")]250		endpoint: String,251		/// A list of headers to apply to all outgoing data (traces, metrics, and logs).252		#[name("HEADERS", "headers")]253		headers: String,254		/// Specifies the OTLP transport protocol to be used for all telemetry data.255		#[name("PROTOCOL", "protocol")]256		#[arg(value_enum)]257		protocol: OtlpProtocol,258		/// The timeout value for all outgoing data (traces, metrics, and logs) in milliseconds.259		#[name("TIMEOUT", "timeout")]260		timeout: u64,261	}262}263impl_settings! {264	#[name("LOGS_", "logs-")]265	struct OtlpLogsSettings {266		/// Specifies the OTLP transport compression to be used for log data.267		#[name("COMPRESSION", "compression")]268		#[arg(value_enum)]269		compression: Compression,270		/// Endpoint URL for log data only, with an optionally-specified port number. Typically ends with `v1/logs` when using OTLP/HTTP.271		#[name("ENDPOINT", "endpoint")]272		endpoint: String,273		/// A list of headers to apply to all outgoing logs.274		#[name("HEADERS", "headers")]275		headers: String,276		/// Specifies the OTLP transport protocol to be used for log data.277		#[name("PROTOCOL", "protocol")]278		#[arg(value_enum)]279		protocol: OtlpProtocol,280		/// The timeout value for all outgoing logs in milliseconds.281		#[name("TIMEOUT", "timeout")]282		timeout: u64,283	}284}285impl_settings! {286	#[name("METRICS_", "metrics-")]287	struct OtlpMetricsSettings {288		/// Specifies the OTLP transport compression to be used for metrics data.289		#[name("COMPRESSION", "compression")]290		#[arg(value_enum)]291		compression: Compression,292		/// Endpoint URL for metric data only, with an optionally-specified port number. Typically ends with `v1/metrics` when using OTLP/HTTP.293		#[name("ENDPOINT", "endpoint")]294		endpoint: String,295		/// A list of headers to apply to all outgoing metrics.296		#[name("HEADERS", "headers")]297		headers: String,298		/// Specifies the OTLP transport protocol to be used for metrics data.299		#[name("PROTOCOL", "protocol")]300		#[arg(value_enum)]301		protocol: OtlpProtocol,302		/// The timeout value for all outgoing metrics in milliseconds.303		#[name("TIMEOUT", "timeout")]304		timeout: u64,305	}306}307impl_settings! {308	#[name("TRACES_", "traces-")]309	struct OtlpTracesSettings {310		/// Specifies the OTLP transport compression to be used for trace data.311		#[name("COMPRESSION", "compression")]312		#[arg(value_enum)]313		compression: Compression,314		/// Endpoint URL for trace data only, with an optionally-specified port number. Typically ends with `v1/traces` when using OTLP/HTTP.315		#[name("ENDPOINT", "endpoint")]316		endpoint: String,317		/// A list of headers to apply to all outgoing traces.318		#[name("HEADERS", "headers")]319		headers: String,320		/// Specifies the OTLP transport protocol to be used for trace data.321		#[name("PROTOCOL", "protocol")]322		#[arg(value_enum)]323		protocol: OtlpProtocol,324		/// The timeout value for all outgoing traces in milliseconds.325		#[name("TIMEOUT", "timeout")]326		timeout: u64,327	}328}329330pub struct ResolvedOtlpSettings {331	pub compression: Option<Compression>,332	pub endpoint: String,333	pub headers: Option<String>,334	pub protocol: OtlpProtocol,335	pub timeout: Duration,336}337338impl ResolvedOtlpSettings {339	const DEFAULT_TIMEOUT_MS: u64 = 10000;340	const DEFAULT_GRPC_ENDPOINT: &str = "http://localhost:4317";341	const DEFAULT_HTTP_ENDPOINT: &str = "http://localhost:4318";342343	pub fn traces(344		base: &impl OtlpSignalSettings,345		signal: &impl OtlpSignalSettings,346	) -> Result<Self, Error> {347		Self::resolve(base, signal, "/v1/traces")348	}349350	pub fn metrics(351		base: &impl OtlpSignalSettings,352		signal: &impl OtlpSignalSettings,353	) -> Result<Self, Error> {354		Self::resolve(base, signal, "/v1/metrics")355	}356357	pub fn logs(358		base: &impl OtlpSignalSettings,359		signal: &impl OtlpSignalSettings,360	) -> Result<Self, Error> {361		Self::resolve(base, signal, "/v1/logs")362	}363364	fn resolve(365		base: &impl OtlpSignalSettings,366		signal: &impl OtlpSignalSettings,367		signal_path: &str,368	) -> Result<Self, Error> {369		let protocol = signal370			.protocol()371			.or_else(|| base.protocol())372			.unwrap_or(OtlpProtocol::HttpProtobuf);373374		let endpoint = if let Some(ep) = signal.endpoint() {375			ep.to_owned()376		} else if let Some(ep) = base.endpoint() {377			match protocol {378				OtlpProtocol::Grpc => ep.to_owned(),379				_ => format!("{ep}{signal_path}"),380			}381		} else {382			match protocol {383				OtlpProtocol::Grpc => Self::DEFAULT_GRPC_ENDPOINT.to_owned(),384				_ => format!("{}{signal_path}", Self::DEFAULT_HTTP_ENDPOINT),385			}386		};387388		Ok(Self {389			compression: signal.compression().or_else(|| base.compression()),390			endpoint,391			headers: signal392				.headers()393				.or_else(|| base.headers())394				.map(str::to_owned),395			protocol,396			timeout: Duration::from_millis(397				signal398					.timeout()399					.or_else(|| base.timeout())400					.unwrap_or(Self::DEFAULT_TIMEOUT_MS),401			),402		})403	}404}