From 0d21db83d79ac90f5c645cac78c8b21775bf0be0 Mon Sep 17 00:00:00 2001 From: dawn Date: Sat, 26 Sep 2026 04:07:29 +0300 Subject: [PATCH] bobbin: OTLP log export Signed-off-by: dawn --- Cargo.lock | 7 ++ bobbin/crates/bobbin/Cargo.toml | 6 + bobbin/crates/bobbin/src/config.rs | 41 ++++++ bobbin/crates/bobbin/src/main.rs | 85 +++++++------ bobbin/crates/bobbin/src/otlp.rs | 195 +++++++++++++++++++++++++++++ bobbin/example.toml | 20 +++ 6 files changed, 319 insertions(+), 35 deletions(-) create mode 100644 bobbin/crates/bobbin/src/otlp.rs diff --git a/Cargo.lock b/Cargo.lock index 1346c9796..0f73bc882 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -751,6 +751,7 @@ version = "0.0.1" dependencies = [ "anyhow", "axum", + "base64", "bobbin-edge-index", "bobbin-ingest", "bobbin-knot-ingest", @@ -766,6 +767,10 @@ dependencies = [ "futures", "jacquard-common", "jacquard-identity", + "opentelemetry", + "opentelemetry-appender-tracing", + "opentelemetry-otlp", + "opentelemetry_sdk", "rustls", "serde", "socket2", @@ -6471,6 +6476,7 @@ dependencies = [ "js-sys", "pin-project-lite", "thiserror 2.0.18", + "tracing", ] [[package]] @@ -6540,6 +6546,7 @@ dependencies = [ "portable-atomic", "rand 0.9.4", "thiserror 2.0.18", + "tokio", ] [[package]] diff --git a/bobbin/crates/bobbin/Cargo.toml b/bobbin/crates/bobbin/Cargo.toml index 5fffa1d58..adb6d2009 100644 --- a/bobbin/crates/bobbin/Cargo.toml +++ b/bobbin/crates/bobbin/Cargo.toml @@ -34,6 +34,10 @@ tokio-util = { workspace = true } url = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +opentelemetry_sdk = { workspace = true } +opentelemetry-otlp = { workspace = true, features = ["internal-logs"] } +opentelemetry-appender-tracing = { workspace = true } +base64 = { workspace = true } anyhow = { workspace = true } clap = { workspace = true } confique = { workspace = true } @@ -43,4 +47,6 @@ tikv-jemallocator = { version = "0.7", features = ["unprefixed_malloc_on_support tikv-jemalloc-ctl = { version = "0.7", features = ["stats", "profiling"] } [dev-dependencies] +opentelemetry = { workspace = true } +opentelemetry_sdk = { workspace = true, features = ["testing"] } tokio = { workspace = true, features = ["test-util"] } diff --git a/bobbin/crates/bobbin/src/config.rs b/bobbin/crates/bobbin/src/config.rs index a993c5ef3..ab07a6c78 100644 --- a/bobbin/crates/bobbin/src/config.rs +++ b/bobbin/crates/bobbin/src/config.rs @@ -10,6 +10,8 @@ use jacquard_common::types::did::Did; use trusted_proxies::{ProxyNetError, TrustedProxies}; use url::Url; +use crate::otlp::OtlpAuth; + const SYSTEM_CONFIG_PATH: &str = "/etc/bobbin/config.toml"; const ENV_PREFIX: &str = "BOBBIN_"; @@ -40,6 +42,9 @@ const KNOWN_KEYS: &[&str] = &[ "pocket.url", "log.format", "log.filter", + "otlp.endpoint", + "otlp.auth", + "otlp.service_name", ]; const KNOWN_ENVS: &[&str] = &[ @@ -70,6 +75,9 @@ const KNOWN_ENVS: &[&str] = &[ "BOBBIN_POCKET_URL", "BOBBIN_LOG_FORMAT", "BOBBIN_LOG", + "BOBBIN_OTLP_ENDPOINT", + "BOBBIN_OTLP_AUTH", + "BOBBIN_OTLP_SERVICE_NAME", ]; #[derive(Debug, Config)] @@ -115,6 +123,9 @@ pub struct BobbinConfig { #[config(nested)] pub log: LogConfig, + + #[config(nested)] + pub otlp: OtlpConfig, } #[derive(Debug, Config)] @@ -344,6 +355,36 @@ pub struct LogConfig { pub filter: String, } +#[derive(Debug, Config)] +pub struct OtlpConfig { + /// base url of an otlp/http collector for logs. + /// unset disables opentelemetry export. + #[config(env = "BOBBIN_OTLP_ENDPOINT")] + pub endpoint: Option, + + /// basic auth credentials in `user:pass` format for the collector. + /// unset sends requests without authorization. + #[config(env = "BOBBIN_OTLP_AUTH")] + pub auth: Option, + + /// service name reported in otlp resource attributes. + #[config( + env = "BOBBIN_OTLP_SERVICE_NAME", + default = "bobbin", + validate = non_empty + )] + pub service_name: String, +} + +#[allow(clippy::ptr_arg)] +fn non_empty(s: &String) -> Result<(), &'static str> { + if s.is_empty() { + Err("must not be empty") + } else { + Ok(()) + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum LogFormat { Text, diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs index d9f717571..8609a7475 100644 --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -35,10 +35,11 @@ use tokio::signal::unix::{SignalKind, signal}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tracing::level_filters::LevelFilter; -use tracing_subscriber::EnvFilter; +use tracing_subscriber::{EnvFilter, layer::SubscriberExt as _, util::SubscriberInitExt as _}; mod config; mod mem; +mod otlp; use config::{BobbinConfig, LogFormat}; @@ -94,22 +95,34 @@ async fn main() -> ExitCode { } }; - if matches!(cli.command, Some(Command::Validate)) { - if let Err(e) = init_tracing(&cfg, Arc::new(SystemClock::new())) { + let export = match init_tracing(&cfg, Arc::new(SystemClock::new())) { + Ok(export) => export, + Err(e) => { eprintln!("failed to install tracing subscriber: {e}"); return ExitCode::FAILURE; } - println!("configuration is valid"); - return ExitCode::SUCCESS; - } + }; - match run(cfg).await { - Ok(()) => ExitCode::SUCCESS, - Err(e) => { - tracing::error!(error = ?e, "fatal"); - ExitCode::FAILURE + let code = match cli.command { + Some(Command::Validate) => { + println!("configuration is valid"); + ExitCode::SUCCESS } + _ => match run(cfg).await { + Ok(()) => ExitCode::SUCCESS, + Err(e) => { + tracing::error!(error = ?e, "fatal"); + ExitCode::FAILURE + } + }, + }; + + // flush queued telemetry before process exit so trailing events aren't dropped + if let Some(export) = export { + export.flush(); } + + code } struct ClockTimer(Arc); @@ -120,37 +133,39 @@ impl tracing_subscriber::fmt::time::FormatTime for ClockTimer { } } -fn init_tracing(cfg: &BobbinConfig, clock: Arc) -> Result<(), String> { +fn init_tracing( + cfg: &BobbinConfig, + clock: Arc, +) -> Result, String> { let combined = format!("{},{}", LevelFilter::INFO, cfg.log.filter); + let filter = EnvFilter::try_new(&combined) + .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?; let format: LogFormat = cfg.log.format.parse()?; let timer = ClockTimer(clock); - match format { - LogFormat::Text => { - let filter = EnvFilter::try_new(&combined) - .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?; - tracing_subscriber::fmt() - .with_env_filter(filter) - .with_timer(timer) - .try_init() - .map_err(|e| e.to_string()) - } - LogFormat::Json => { - let filter = EnvFilter::try_new(&combined) - .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?; - tracing_subscriber::fmt() - .with_env_filter(filter) - .json() - .with_timer(timer) - .try_init() - .map_err(|e| e.to_string()) - } - } + let (text, json) = match format { + LogFormat::Text => ( + Some(tracing_subscriber::fmt::layer().with_timer(timer)), + None, + ), + LogFormat::Json => ( + None, + Some(tracing_subscriber::fmt::layer().json().with_timer(timer)), + ), + }; + let export = otlp::LogExport::from_config(&cfg.otlp)?; + + tracing_subscriber::registry() + .with(filter) + .with(text) + .with(json) + .with(export.as_ref().map(otlp::LogExport::layer)) + .try_init() + .map_err(|e| e.to_string())?; + Ok(export) } async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { let clock: Arc = Arc::new(SystemClock::new()); - init_tracing(&cfg, clock.clone()) - .map_err(|e| anyhow!("failed to install tracing subscriber: {e}"))?; let (budget, budget_source) = mem::detect_budget(); tracing::info!( diff --git a/bobbin/crates/bobbin/src/otlp.rs b/bobbin/crates/bobbin/src/otlp.rs new file mode 100644 index 000000000..ef5ad65b2 --- /dev/null +++ b/bobbin/crates/bobbin/src/otlp.rs @@ -0,0 +1,195 @@ +use std::collections::HashMap; +use std::fmt; + +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD as Base64; +use opentelemetry_appender_tracing::layer::{OpenTelemetryTracingBridge, TracingSpanAttributes}; +use opentelemetry_otlp::{LogExporter, Protocol, WithExportConfig as _, WithHttpConfig as _}; +use opentelemetry_sdk::{Resource, logs::SdkLoggerProvider}; +use serde::Deserialize; +use tracing::Subscriber; +use tracing_subscriber::Layer; +use tracing_subscriber::filter::{LevelFilter, Targets}; +use tracing_subscriber::registry::LookupSpan; +use url::Url; + +use crate::config::OtlpConfig; + +// suppress http client and otlp logs so exporting a record cannot trigger another export loop +const EXPORT_QUIET: [&str; 4] = ["h2", "hyper", "reqwest", "opentelemetry"]; + +#[derive(Deserialize)] +#[serde(try_from = "String")] +pub struct OtlpAuth(String); + +impl TryFrom for OtlpAuth { + type Error = &'static str; + + fn try_from(credentials: String) -> Result { + match credentials.split_once(':') { + Some((user, _)) if !user.is_empty() => { + Ok(Self(format!("Basic {}", Base64.encode(credentials)))) + } + _ => Err("otlp.auth must be `user:pass`"), + } + } +} + +impl fmt::Debug for OtlpAuth { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("OtlpAuth()") + } +} + +impl OtlpAuth { + fn headers(&self) -> HashMap { + HashMap::from([("Authorization".to_owned(), self.0.clone())]) + } +} + +pub struct LogExport { + provider: SdkLoggerProvider, +} + +impl LogExport { + pub fn from_config(cfg: &OtlpConfig) -> Result, String> { + let Some(base) = &cfg.endpoint else { + return Ok(None); + }; + let endpoint = signal_url(base, "logs"); + let exporter = LogExporter::builder() + .with_http() + .with_protocol(Protocol::HttpBinary) + .with_endpoint(endpoint.as_str()) + .with_headers(cfg.auth.as_ref().map(OtlpAuth::headers).unwrap_or_default()) + .build() + .map_err(|e| format!("OTLP log exporter for `{endpoint}`: {e}"))?; + Ok(Some(Self::batched(exporter, &cfg.service_name))) + } + + fn batched( + exporter: impl opentelemetry_sdk::logs::LogExporter + 'static, + service_name: &str, + ) -> Self { + let resource = Resource::builder() + .with_service_name(service_name.to_owned()) + .build(); + Self { + provider: SdkLoggerProvider::builder() + .with_batch_exporter(exporter) + .with_resource(resource) + .build(), + } + } + + pub fn layer(&self) -> impl Layer + use + where + S: Subscriber + for<'a> LookupSpan<'a>, + { + OpenTelemetryTracingBridge::builder(&self.provider) + .with_tracing_span_attributes(TracingSpanAttributes::all()) + .build() + .with_filter( + Targets::new() + .with_default(LevelFilter::TRACE) + .with_targets(EXPORT_QUIET.map(|target| (target, LevelFilter::OFF))), + ) + } + + pub fn flush(self) { + if let Err(error) = self.provider.shutdown() { + eprintln!("could not flush OTLP log export: {error}"); + } + } +} + +fn signal_url(base: &Url, signal: &str) -> Url { + let mut url = base.clone(); + let path = format!("{}/v1/{signal}", base.path().trim_end_matches('/')); + url.set_path(&path); + url +} + +#[cfg(test)] +mod tests { + use opentelemetry::logs::AnyValue; + use opentelemetry::{Key, Value}; + use opentelemetry_sdk::logs::InMemoryLogExporter; + use tracing_subscriber::layer::SubscriberExt as _; + + use super::*; + + #[test] + fn unset_endpoint_disables_export() { + let cfg = OtlpConfig { + endpoint: None, + auth: None, + service_name: "bobbin-test".to_owned(), + }; + assert!(LogExport::from_config(&cfg).unwrap().is_none()); + } + + #[test] + fn signals_go_under_the_base_path() { + let url = |base: &str| signal_url(&base.parse().unwrap(), "logs").to_string(); + assert_eq!( + url("https://otel.example.com"), + "https://otel.example.com/v1/logs" + ); + assert_eq!( + url("https://example.com/otel/"), + "https://example.com/otel/v1/logs" + ); + assert_eq!( + url("http://collector:4318/otel"), + "http://collector:4318/otel/v1/logs" + ); + } + + #[test] + fn export_ships_records_with_service_name_and_fields() { + let exporter = InMemoryLogExporter::default(); + let export = LogExport::batched(exporter.clone(), "bobbin-test"); + let subscriber = tracing_subscriber::registry().with(export.layer()); + tracing::subscriber::with_default(subscriber, || { + tracing::warn!(target: "bobbin_xrpc", did = "did:plc:dawn", "slow request"); + tracing::warn!(target: "h2", "stream reset"); + tracing::warn!(target: "hyper_util::client::legacy", "connect failed"); + tracing::warn!(target: "reqwest::connect", "connect failed"); + tracing::error!(target: "opentelemetry_sdk", "export failed"); + tracing::warn!(target: "opentelemetry-otlp", "retries exhausted"); + }); + export.provider.force_flush().unwrap(); + + let logs = exporter.get_emitted_logs().unwrap(); + let [log] = logs.as_slice() else { + panic!("expected only the bobbin record, got {logs:?}"); + }; + assert_eq!( + log.resource.get(&Key::from_static_str("service.name")), + Some(Value::from("bobbin-test")), + ); + assert_eq!(log.record.severity_text(), Some("WARN")); + assert_eq!(log.record.body(), Some(&AnyValue::from("slow request"))); + assert!( + log.record + .attributes_iter() + .any(|(k, v)| { k.as_str() == "did" && *v == AnyValue::from("did:plc:dawn") }) + ); + + export.flush(); + assert!(exporter.is_shutdown_called()); + } + + #[test] + fn auth_is_validated_encoded_and_redacted() { + let auth = OtlpAuth::try_from("user:secretpass".to_owned()).unwrap(); + assert_eq!( + auth.headers().get("Authorization").map(String::as_str), + Some("Basic dXNlcjpzZWNyZXRwYXNz"), + ); + assert_eq!(format!("{auth:?}"), "OtlpAuth()"); + assert!(OtlpAuth::try_from("no-colon".to_owned()).is_err()); + assert!(OtlpAuth::try_from(":pass".to_owned()).is_err()); + } +} diff --git a/bobbin/example.toml b/bobbin/example.toml index f02c68343..a3af6a18b 100644 --- a/bobbin/example.toml +++ b/bobbin/example.toml @@ -219,3 +219,23 @@ # # Default value: "info" #filter = "info" + +[otlp] +# base url of an otlp/http collector for logs. +# unset disables opentelemetry export. +# +# Can also be specified via environment variable `BOBBIN_OTLP_ENDPOINT`. +#endpoint = + +# basic auth credentials in `user:pass` format for the collector. +# unset sends requests without authorization. +# +# Can also be specified via environment variable `BOBBIN_OTLP_AUTH`. +#auth = + +# service name reported in otlp resource attributes. +# +# Can also be specified via environment variable `BOBBIN_OTLP_SERVICE_NAME`. +# +# Default value: "bobbin" +#service_name = "bobbin" -- 2.51.2