diff --git a/Cargo.toml b/Cargo.toml index ea9f105..9d75407 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,11 +13,46 @@ time = { version = "0.3.36", optional = true } tracing = "0.1.40" url = { version = "2.5.2", features = ["serde"] } +# telemetry deps +async-trait = { version = "0.1.85", optional = true } +http = { version = "1.2.0", optional = true } +once_cell = { version = "1.20.2", optional = true } +opentelemetry = { version = "0.27.1", optional = true } +opentelemetry-appender-tracing = { version = "0.27.0", optional = true } +opentelemetry-http = { version = "0.27.0", optional = true } +opentelemetry-otlp = { version = "0.27.0", optional = true } +opentelemetry-semantic-conventions = { version = "0.27.0", optional = true } +opentelemetry_sdk = { version = "0.27.1", features = [ + "rt-tokio", +], optional = true } +reqwest = { version = "0.12.12", optional = true } +reqwest-middleware = { version = "0.4.0", optional = true } +tracing-opentelemetry = { version = "0.28.0", optional = true } +tracing-subscriber = { version = "0.3.19", features = [ + "env-filter", +], optional = true } + + [dev-dependencies] serde_json = "1.0.128" [features] time = ["dep:time"] +telemetry = [ + "async-trait", + "http", + "once_cell", + "opentelemetry", + "opentelemetry-appender-tracing", + "opentelemetry-http", + "opentelemetry-otlp", + "opentelemetry-semantic-conventions", + "opentelemetry_sdk", + "reqwest", + "reqwest-middleware", + "tracing-opentelemetry", + "tracing-subscriber", +] [lints.rust] dead_code = "warn" diff --git a/README.md b/README.md index a2936e5..b2fc7af 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,56 @@ Random rust utility functions and types +## Telemetry + +For using this module the feature flag `telemetry` need to be added. +This module contains a set of helpers to work with OpenTelemetry logs, traces and metrics. + +### Setup + +For setup all that's needed it to run the function `famedly_rust_utils::famedly_rust_utils::telemetry::init_otel`. + +If no configuration is present the exporting of logs traces and metrics is disable and the stdout logging is enable. + +The opentelemetry information is exported using gRPC to and opentelemetry compatible collector. By default the expected endpoint is `http://localhots:4317` + +The functions on the crate exporting opentelemetry traces should be annotated with `tracing::instrument` to generate a new span for that function. Documentation on this macro can be found on the [here](https://docs.rs/tracing/latest/tracing/attr.instrument.html) + +### Propagate the context + +A context can be propagated to allow linking the traces from two different services. This is done by injecting the context information on the request and retrieving it on the other service. + +#### reqwest + +For injecting the current context using the reqwest client we can warp a client on a [reqwest-middleware](https://crates.io/crates/reqwest-middleware) and use the `OtelMiddleware` middleware present on the crate. + +```rust +use famedly_rust_utils::telemetry::OtelMiddleware; + +let reqwest_client = reqwest::Client::builder().build().unwrap(); +let client = reqwest_middleware::ClientBuilder::new(reqwest_client) + // Insert the tracing middleware + .with(OtelMiddleware::default()) + .build(); +client.get("http://localhost").send().await; +``` + +### axum + +For retrieving a context using axum we can use the `OtelAxumLayer` from [axum_tracing_opentelemetry](https://crates.io/crates/axum-tracing-opentelemetry) + +> [!WARNING] +> This only seems to be working using the feature flag `tracing_level_info`. See the [issue](https://github.com/davidB/tracing-opentelemetry-instrumentation-sdk/issues/148) + +This layer should run as soon as possible + +```rust +use axum_tracing_opentelemetry::middleware::OtelAxumLayer; + +Router::new().layer(OtelAxumLayer::default()) + +``` + ## Lints ```sh diff --git a/src/lib.rs b/src/lib.rs index 84a45f9..95e3aa5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,9 @@ mod base_url; pub mod duration; /// [serde::Deserialize] impl for [tracing::level_filters::LevelFilter] mod level_filter; +#[cfg(feature = "telemetry")] +/// Function to setup the telemetry tools +pub mod telemetry; pub use base_url::{BaseUrl, BaseUrlParseError}; pub use level_filter::LevelFilter; diff --git a/src/telemetry/config.rs b/src/telemetry/config.rs new file mode 100644 index 0000000..df280fd --- /dev/null +++ b/src/telemetry/config.rs @@ -0,0 +1,155 @@ +use std::str::FromStr as _; + +use serde::Deserialize; +use url::Url; + +use crate::LevelFilter; + +const DEFAULT_FILTER: &str = "opentelemetry=off,tonic=off,h2=off,reqwest=off,axum=info,hyper=info,hyper-tls=info,tokio=info,tower=info,josekit=info,openssl=info"; +const DEFAULT_LEVEL: &str = "info"; +const DEFAULT_ENDPOINT: &str = "http://localhost:4317"; + +/// OpenTelemetry configuration +#[derive(Debug, Deserialize, Clone)] +pub struct OtelConfig { + /// Enables logs on stdout + pub stdout: Option, + /// Configurations for exporting traces, metrics and logs + pub exporter: Option, +} + +/// Configuration for exporting OpenTelemetry data +#[derive(Debug, Deserialize, Clone)] +pub struct ExporterConfig { + /// gRPC endpoint for exporting using OTELP + pub endpoint: Option, + /// Application service name + pub service_name: String, + /// Application version + pub version: String, + + /// Logs exporting config + pub logger: Option, + /// Traces exporting config + pub tracer: Option, + /// Metrics exporting config + pub meter: Option, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct StdoutLogsConfig { + enable: bool, + level: Option, + filter_directives: Option, +} + +/// Provider configuration for OpenTelemetry export +#[derive(Debug, Deserialize, Clone)] +#[allow(missing_docs)] +pub struct ProviderConfig { + pub enable: bool, + pub level: Option, + pub filter_directives: Option, +} + +impl ProviderConfig { + #[allow(clippy::expect_used)] + pub(crate) fn get_filter(&self) -> String { + format!( + "{},{}", + self.level.unwrap_or( + LevelFilter::from_str(DEFAULT_LEVEL).expect("Error parsing default level") + ), + self.filter_directives.as_ref().unwrap_or(&DEFAULT_FILTER.to_owned()) + ) + } +} + +impl StdoutLogsConfig { + #[allow(clippy::expect_used)] + pub(crate) fn get_filter(&self) -> String { + format!( + "{},{}", + self.level.unwrap_or( + LevelFilter::from_str(DEFAULT_LEVEL).expect("Error parsing default level") + ), + self.filter_directives.as_ref().unwrap_or(&DEFAULT_FILTER.to_owned()) + ) + } +} + +impl Default for StdoutLogsConfig { + fn default() -> Self { + Self { enable: true, level: None, filter_directives: None } + } +} + +impl OtelConfig { + pub(crate) fn stdout_enable(&self) -> bool { + self.stdout.as_ref().is_none_or(|config| config.enable) + } + pub(crate) fn traces_enable(&self) -> bool { + self.exporter + .as_ref() + .is_some_and(|config| config.tracer.as_ref().is_some_and(|tracer| tracer.enable)) + } + pub(crate) fn metrics_enable(&self) -> bool { + self.exporter + .as_ref() + .is_some_and(|config| config.meter.as_ref().is_some_and(|meter| meter.enable)) + } + pub(crate) fn logs_enable(&self) -> bool { + self.exporter + .as_ref() + .is_some_and(|config| config.logger.as_ref().is_some_and(|logger| logger.enable)) + } + pub(crate) fn get_traces_config(&self) -> Result { + self.exporter + .as_ref() + .and_then(|exporter| exporter.tracer.clone()) + .ok_or(MissingConfigError::Traces) + } + pub(crate) fn get_metrics_config(&self) -> Result { + self.exporter + .as_ref() + .and_then(|exporter| exporter.meter.clone()) + .ok_or(MissingConfigError::Metrics) + } + pub(crate) fn get_logs_config(&self) -> Result { + self.exporter + .as_ref() + .and_then(|exporter| exporter.logger.clone()) + .ok_or(MissingConfigError::Logs) + } + pub(crate) fn get_stdout_config(&self) -> StdoutLogsConfig { + self.stdout.clone().unwrap_or_default() + } + #[allow(clippy::expect_used)] + pub(crate) fn get_endpoint(&self) -> Url { + self.exporter + .as_ref() + .and_then(|exporter| exporter.clone().endpoint) + .unwrap_or(Url::from_str(DEFAULT_ENDPOINT).expect("Error parsing default endpoint")) + } + pub(crate) fn get_service_name(&self) -> String { + self.exporter + .as_ref() + .map_or(env!("CARGO_PKG_NAME").to_owned(), |exporter| exporter.service_name.clone()) + } + pub(crate) fn get_version(&self) -> String { + self.exporter + .as_ref() + .map_or(env!("CARGO_PKG_VERSION").to_owned(), |exporter| exporter.service_name.clone()) + } +} + +/// Missing configurations errors +#[derive(Debug, thiserror::Error)] +pub enum MissingConfigError { + #[error("Traces export configuration is missing")] + Traces, + #[error("Metrics export configuration is missing")] + Metrics, + #[error("Logs export configuration is missing")] + Logs, +} diff --git a/src/telemetry/mod.rs b/src/telemetry/mod.rs new file mode 100644 index 0000000..2a74a91 --- /dev/null +++ b/src/telemetry/mod.rs @@ -0,0 +1,207 @@ +//! OpenTelemetry initialization +//! +//! Lib containing the definitions and initializations of the OpenTelemetry +//! tools +use std::str::FromStr as _; + +use config::{MissingConfigError, OtelConfig}; +use opentelemetry::{ + trace::{TraceError, TracerProvider as _}, + KeyValue, +}; +use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge; +use opentelemetry_otlp::{LogExporter, SpanExporter, WithExportConfig as _}; +use opentelemetry_sdk::{ + logs::{LogError, LoggerProvider}, + metrics::{MeterProviderBuilder, MetricError, PeriodicReader, SdkMeterProvider}, + propagation::TraceContextPropagator, + runtime, + trace::{RandomIdGenerator, TracerProvider}, + Resource, +}; +use opentelemetry_semantic_conventions::{ + resource::{SERVICE_NAME, SERVICE_VERSION}, + SCHEMA_URL, +}; +use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer}; +use tracing_subscriber::{ + layer::SubscriberExt as _, util::SubscriberInitExt as _, EnvFilter, Layer, +}; +use url::Url; + +pub mod config; +mod reqwest_middleware; +pub use reqwest_middleware::OtelMiddleware; + +fn resource(service_name: String, version: String) -> Resource { + Resource::from_schema_url( + [KeyValue::new(SERVICE_NAME, service_name), KeyValue::new(SERVICE_VERSION, version)], + SCHEMA_URL, + ) +} + +fn init_traces( + endpoint: Url, + service_name: String, + version: String, +) -> Result { + let exporter = SpanExporter::builder().with_tonic().with_endpoint(endpoint).build()?; + let tracer_provider = TracerProvider::builder() + .with_id_generator(RandomIdGenerator::default()) + .with_resource(resource(service_name, version)) + // .with_simple_exporter(exporter) + .with_batch_exporter(exporter, runtime::Tokio) + .build(); + + opentelemetry::global::set_tracer_provider(tracer_provider.clone()); + Ok(tracer_provider) +} + +fn init_metrics( + endpoint: Url, + service_name: String, + version: String, +) -> Result { + let exporter = opentelemetry_otlp::MetricExporter::builder() + .with_tonic() + .with_endpoint(endpoint) + .with_temporality(opentelemetry_sdk::metrics::Temporality::default()) + .build()?; + + let reader = PeriodicReader::builder(exporter, runtime::Tokio) + // TODO: Should this be configurable or not? + .with_interval(std::time::Duration::from_secs(1)) + .build(); + + let meter_provider = MeterProviderBuilder::default() + .with_resource(resource(service_name, version)) + .with_reader(reader) + .build(); + + Ok(meter_provider) +} + +fn init_logs( + endpoint: Url, + service_name: String, + version: String, +) -> Result { + let exporter = LogExporter::builder().with_tonic().with_endpoint(endpoint).build()?; + + Ok(LoggerProvider::builder() + .with_resource(resource(service_name, version)) + .with_batch_exporter(exporter, runtime::Tokio) + .build()) +} + +/// Initializes the OpenTelemetry +pub fn init_otel(config: OtelConfig) -> Result { + opentelemetry::global::set_text_map_propagator(TraceContextPropagator::default()); + + let stdout_layer = if config.stdout_enable() { + let logger_config = config.get_stdout_config(); + let filter_fmt = EnvFilter::from_str(&logger_config.get_filter())?; + + Some(tracing_subscriber::fmt::layer().with_thread_names(true).with_filter(filter_fmt)) + } else { + None + }; + + let (logger_provider, logs_layer) = if config.logs_enable() { + let logger_config = config.get_logs_config()?; + let filter_otel = EnvFilter::from_str(&logger_config.get_filter())?; + let logger_provider = + init_logs(config.get_endpoint(), config.get_service_name(), config.get_version())?; + + // Create a new OpenTelemetryTracingBridge using the above LoggerProvider. + let logs_layer = OpenTelemetryTracingBridge::new(&logger_provider); + let logs_layer = logs_layer.with_filter(filter_otel); + + (Some(logger_provider), Some(logs_layer)) + } else { + (None, None) + }; + + let (tracer_provider, tracer_layer) = if config.traces_enable() { + let tracer_config = config.get_traces_config()?; + + let trace_filter = EnvFilter::from_str(&tracer_config.get_filter())?; + let tracer_provider = + init_traces(config.get_endpoint(), config.get_service_name(), config.get_version())?; + let tracer = tracer_provider.tracer(config.get_service_name()); + let tracer_layer = OpenTelemetryLayer::new(tracer).with_filter(trace_filter); + + (Some(tracer_provider), Some(tracer_layer)) + } else { + (None, None) + }; + + let (meter_provider, meter_layer) = if config.metrics_enable() { + let meter_config = config.get_metrics_config()?; + + let metrics_filter = EnvFilter::from_str(&meter_config.get_filter())?; + let meter_provider = + init_metrics(config.get_endpoint(), config.get_service_name(), config.get_version())?; + let meter_layer = MetricsLayer::new(meter_provider.clone()).with_filter(metrics_filter); + + (Some(meter_provider), Some(meter_layer)) + } else { + (None, None) + }; + + // Initialize the tracing subscriber with the OpenTelemetry layer, the + // stdout layer, traces and metrics. + tracing_subscriber::registry() + .with(logs_layer) + .with(stdout_layer) + .with(meter_layer) + .with(tracer_layer) + .init(); + + Ok(ProvidersGuard { logger_provider, tracer_provider, meter_provider }) +} + +/// Guarding object to make sure the providers are properly shutdown +#[derive(Debug)] +pub struct ProvidersGuard { + logger_provider: Option, + tracer_provider: Option, + meter_provider: Option, +} + +impl Drop for ProvidersGuard { + #[allow(clippy::print_stderr)] + fn drop(&mut self) { + self.logger_provider.as_ref().inspect(|logger_provider| { + if let Err(err) = logger_provider.shutdown() { + eprintln!("{err:?}"); + } + }); + self.tracer_provider.as_ref().inspect(|tracer_provider| { + if let Err(err) = tracer_provider.shutdown() { + eprintln!("{err:?}"); + } + }); + self.meter_provider.as_ref().inspect(|meter_provider| { + if let Err(err) = meter_provider.shutdown() { + eprintln!("{err:?}"); + } + }); + } +} + +/// OpenTelemetry setup errors +#[allow(missing_docs)] +#[derive(Debug, thiserror::Error)] +pub enum OtelInitError { + #[error("Logger initialization error: {0}")] + LoggerInitError(#[from] LogError), + #[error("Tracer initialization error: {0}")] + TracerInitError(#[from] TraceError), + #[error("Meter initialization error: {0}")] + MeterInitError(#[from] MetricError), + #[error("Parsing EnvFilter directives error: {0}")] + EnvFilterError(#[from] tracing_subscriber::filter::ParseError), + #[error("Otel configuration is missing: {0}")] + MissingConfig(#[from] MissingConfigError), +} diff --git a/src/telemetry/reqwest_middleware.rs b/src/telemetry/reqwest_middleware.rs new file mode 100644 index 0000000..8453439 --- /dev/null +++ b/src/telemetry/reqwest_middleware.rs @@ -0,0 +1,35 @@ +use http::Extensions; +use opentelemetry_http::HeaderInjector; +use reqwest::{Request, Response}; +use reqwest_middleware::{Middleware, Next, Result}; +use tracing::Span; +use tracing_opentelemetry::OpenTelemetrySpanExt as _; + +/// Middleware for [reqwest-middleware] to propagate the Otel context +/// +/// Example +/// +/// let reqwest_client = reqwest::Client::builder().build().unwrap(); +/// let client = ClientBuilder::new(reqwest_client) +/// // Insert the tracing middleware +/// .with(OtelMiddleware::default()) +/// .build(); +/// client.get("http://localhost").send().await; +#[derive(Debug, Default)] +pub struct OtelMiddleware; + +#[async_trait::async_trait] +impl Middleware for OtelMiddleware { + async fn handle( + &self, + mut req: Request, + extensions: &mut Extensions, + next: Next<'_>, + ) -> Result { + opentelemetry::global::get_text_map_propagator(|propagator| { + let cx = Span::current().context(); + propagator.inject_context(&cx, &mut HeaderInjector(req.headers_mut())); + }); + next.run(req, extensions).await + } +}