From 47247789addfd3fb0dc0d02495f9c0f15d1919e6 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 24 Apr 2026 01:02:26 +0800 Subject: [PATCH] feat(obs): support OTLP headers and timeout overrides for HTTP exporters (#2661) --- Cargo.lock | 1 + crates/config/src/observability/mod.rs | 16 ++++ crates/obs/Cargo.toml | 1 + crates/obs/src/config.rs | 45 +++++++++-- crates/obs/src/telemetry/otel.rs | 103 +++++++++++++++++++++++-- 5 files changed, 153 insertions(+), 13 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cbae72f83..800e5be14 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8488,6 +8488,7 @@ dependencies = [ "opentelemetry-semantic-conventions", "opentelemetry-stdout", "opentelemetry_sdk", + "percent-encoding", "pyroscope", "rustfs-audit", "rustfs-config", diff --git a/crates/config/src/observability/mod.rs b/crates/config/src/observability/mod.rs index 1ce6fae5f..9dd2ba654 100644 --- a/crates/config/src/observability/mod.rs +++ b/crates/config/src/observability/mod.rs @@ -22,6 +22,14 @@ pub const ENV_OBS_ENDPOINT: &str = "RUSTFS_OBS_ENDPOINT"; pub const ENV_OBS_TRACE_ENDPOINT: &str = "RUSTFS_OBS_TRACE_ENDPOINT"; pub const ENV_OBS_METRIC_ENDPOINT: &str = "RUSTFS_OBS_METRIC_ENDPOINT"; pub const ENV_OBS_LOG_ENDPOINT: &str = "RUSTFS_OBS_LOG_ENDPOINT"; +pub const ENV_OBS_ENDPOINT_HEADERS: &str = "RUSTFS_OBS_ENDPOINT_HEADERS"; +pub const ENV_OBS_ENDPOINT_TRACES_HEADERS: &str = "RUSTFS_OBS_ENDPOINT_TRACES_HEADERS"; +pub const ENV_OBS_ENDPOINT_METRICS_HEADERS: &str = "RUSTFS_OBS_ENDPOINT_METRICS_HEADERS"; +pub const ENV_OBS_ENDPOINT_LOGS_HEADERS: &str = "RUSTFS_OBS_ENDPOINT_LOGS_HEADERS"; +pub const ENV_OBS_ENDPOINT_TIMEOUT_MILLIS: &str = "RUSTFS_OBS_ENDPOINT_TIMEOUT_MILLIS"; +pub const ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS: &str = "RUSTFS_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS"; +pub const ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS: &str = "RUSTFS_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS"; +pub const ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS: &str = "RUSTFS_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS"; pub const ENV_OBS_PROFILING_ENDPOINT: &str = "RUSTFS_OBS_PROFILING_ENDPOINT"; pub const ENV_OBS_USE_STDOUT: &str = "RUSTFS_OBS_USE_STDOUT"; pub const ENV_OBS_SAMPLE_RATIO: &str = "RUSTFS_OBS_SAMPLE_RATIO"; @@ -108,6 +116,14 @@ mod tests { assert_eq!(ENV_OBS_TRACE_ENDPOINT, "RUSTFS_OBS_TRACE_ENDPOINT"); assert_eq!(ENV_OBS_METRIC_ENDPOINT, "RUSTFS_OBS_METRIC_ENDPOINT"); assert_eq!(ENV_OBS_LOG_ENDPOINT, "RUSTFS_OBS_LOG_ENDPOINT"); + assert_eq!(ENV_OBS_ENDPOINT_HEADERS, "RUSTFS_OBS_ENDPOINT_HEADERS"); + assert_eq!(ENV_OBS_ENDPOINT_TRACES_HEADERS, "RUSTFS_OBS_ENDPOINT_TRACES_HEADERS"); + assert_eq!(ENV_OBS_ENDPOINT_METRICS_HEADERS, "RUSTFS_OBS_ENDPOINT_METRICS_HEADERS"); + assert_eq!(ENV_OBS_ENDPOINT_LOGS_HEADERS, "RUSTFS_OBS_ENDPOINT_LOGS_HEADERS"); + assert_eq!(ENV_OBS_ENDPOINT_TIMEOUT_MILLIS, "RUSTFS_OBS_ENDPOINT_TIMEOUT_MILLIS"); + assert_eq!(ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS, "RUSTFS_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS"); + assert_eq!(ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS, "RUSTFS_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS"); + assert_eq!(ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS, "RUSTFS_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS"); assert_eq!(ENV_OBS_PROFILING_ENDPOINT, "RUSTFS_OBS_PROFILING_ENDPOINT"); assert_eq!(ENV_OBS_USE_STDOUT, "RUSTFS_OBS_USE_STDOUT"); assert_eq!(ENV_OBS_SAMPLE_RATIO, "RUSTFS_OBS_SAMPLE_RATIO"); diff --git a/crates/obs/Cargo.toml b/crates/obs/Cargo.toml index c4d5a02f2..1132a33cc 100644 --- a/crates/obs/Cargo.toml +++ b/crates/obs/Cargo.toml @@ -53,6 +53,7 @@ opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] } opentelemetry-stdout = { workspace = true } opentelemetry-otlp = { workspace = true } opentelemetry-semantic-conventions = { workspace = true, features = ["semconv_experimental"] } +percent-encoding = { workspace = true } serde = { workspace = true } tracing = { workspace = true, features = ["std", "attributes"] } tracing-appender = { workspace = true } diff --git a/crates/obs/src/config.rs b/crates/obs/src/config.rs index 9449f4ff8..b9fc834a9 100644 --- a/crates/obs/src/config.rs +++ b/crates/obs/src/config.rs @@ -28,11 +28,14 @@ use rustfs_config::observability::{ DEFAULT_OBS_LOG_DRY_RUN, DEFAULT_OBS_LOG_GZIP_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_MATCH_MODE, DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES, DEFAULT_OBS_LOG_MIN_FILE_AGE_SECONDS, DEFAULT_OBS_LOG_PARALLEL_COMPRESS, DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL, - DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_ENDPOINT, ENV_OBS_ENVIRONMENT, - ENV_OBS_LOG_CLEANUP_INTERVAL_SECONDS, ENV_OBS_LOG_COMPRESS_OLD_FILES, ENV_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS, - ENV_OBS_LOG_COMPRESSION_ALGORITHM, ENV_OBS_LOG_DELETE_EMPTY_FILES, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_DRY_RUN, - ENV_OBS_LOG_ENDPOINT, ENV_OBS_LOG_EXCLUDE_PATTERNS, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL, - ENV_OBS_LOG_KEEP_FILES, ENV_OBS_LOG_MATCH_MODE, ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES, + DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_ENDPOINT, ENV_OBS_ENDPOINT_HEADERS, + ENV_OBS_ENDPOINT_LOGS_HEADERS, ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_METRICS_HEADERS, + ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TIMEOUT_MILLIS, ENV_OBS_ENDPOINT_TRACES_HEADERS, + ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS, ENV_OBS_ENVIRONMENT, ENV_OBS_LOG_CLEANUP_INTERVAL_SECONDS, + ENV_OBS_LOG_COMPRESS_OLD_FILES, ENV_OBS_LOG_COMPRESSED_FILE_RETENTION_DAYS, ENV_OBS_LOG_COMPRESSION_ALGORITHM, + ENV_OBS_LOG_DELETE_EMPTY_FILES, ENV_OBS_LOG_DIRECTORY, ENV_OBS_LOG_DRY_RUN, ENV_OBS_LOG_ENDPOINT, + ENV_OBS_LOG_EXCLUDE_PATTERNS, ENV_OBS_LOG_FILENAME, ENV_OBS_LOG_GZIP_COMPRESSION_LEVEL, ENV_OBS_LOG_KEEP_FILES, + ENV_OBS_LOG_MATCH_MODE, ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES, ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES, ENV_OBS_LOG_MIN_FILE_AGE_SECONDS, ENV_OBS_LOG_PARALLEL_COMPRESS, ENV_OBS_LOG_PARALLEL_WORKERS, ENV_OBS_LOG_ROTATION_TIME, ENV_OBS_LOG_STDOUT_ENABLED, ENV_OBS_LOG_ZSTD_COMPRESSION_LEVEL, ENV_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, ENV_OBS_LOG_ZSTD_WORKERS, ENV_OBS_LOGGER_LEVEL, ENV_OBS_LOGS_EXPORT_ENABLED, ENV_OBS_METER_INTERVAL, ENV_OBS_METRIC_ENDPOINT, @@ -45,7 +48,7 @@ use rustfs_config::{ DEFAULT_OBS_PROFILING_EXPORT_ENABLED, DEFAULT_OBS_TRACES_EXPORT_ENABLED, ENVIRONMENT, METER_INTERVAL, SAMPLE_RATIO, SERVICE_VERSION, USE_STDOUT, }; -use rustfs_utils::{get_env_bool, get_env_f64, get_env_opt_str, get_env_str, get_env_u64, get_env_usize}; +use rustfs_utils::{get_env_bool, get_env_f64, get_env_opt_str, get_env_opt_u64, get_env_str, get_env_u64, get_env_usize}; use serde::{Deserialize, Serialize}; use std::env; @@ -101,6 +104,28 @@ pub struct OtelConfig { pub metric_endpoint: Option, /// Dedicated log endpoint; overrides `endpoint` + `/v1/logs` fallback. pub log_endpoint: Option, + /// Headers applied to all OTLP signals when using HTTP exporter. + /// Format: comma-separated `key=value` pairs with URL-encoded values, for + /// example: `Authorization=Bearer%20abc%20123,X-Scope-OrgID=my-tenant`. + /// URL-encode reserved characters in values such as spaces, commas, and `=`. + pub endpoint_headers: Option, + /// Additional headers for traces; merged on top of `endpoint_headers`. + /// Uses the same comma-separated `key=value` format with URL-encoded values. + pub trace_headers: Option, + /// Additional headers for metrics; merged on top of `endpoint_headers`. + /// Uses the same comma-separated `key=value` format with URL-encoded values. + pub metric_headers: Option, + /// Additional headers for logs; merged on top of `endpoint_headers`. + /// Uses the same comma-separated `key=value` format with URL-encoded values. + pub log_headers: Option, + /// Timeout (milliseconds) for all OTLP HTTP exports. + pub endpoint_timeout_millis: Option, + /// Timeout (milliseconds) for trace OTLP HTTP export. + pub trace_timeout_millis: Option, + /// Timeout (milliseconds) for metrics OTLP HTTP export. + pub metric_timeout_millis: Option, + /// Timeout (milliseconds) for log OTLP HTTP export. + pub log_timeout_millis: Option, /// Dedicated profiling endpoint. pub profiling_endpoint: Option, /// Whether to export distributed traces (default: `true`). @@ -245,6 +270,14 @@ impl OtelConfig { trace_endpoint: get_env_opt_str(ENV_OBS_TRACE_ENDPOINT), metric_endpoint: get_env_opt_str(ENV_OBS_METRIC_ENDPOINT), log_endpoint: get_env_opt_str(ENV_OBS_LOG_ENDPOINT), + endpoint_headers: get_env_opt_str(ENV_OBS_ENDPOINT_HEADERS), + trace_headers: get_env_opt_str(ENV_OBS_ENDPOINT_TRACES_HEADERS), + metric_headers: get_env_opt_str(ENV_OBS_ENDPOINT_METRICS_HEADERS), + log_headers: get_env_opt_str(ENV_OBS_ENDPOINT_LOGS_HEADERS), + endpoint_timeout_millis: get_env_opt_u64(ENV_OBS_ENDPOINT_TIMEOUT_MILLIS), + trace_timeout_millis: get_env_opt_u64(ENV_OBS_ENDPOINT_TRACES_TIMEOUT_MILLIS), + metric_timeout_millis: get_env_opt_u64(ENV_OBS_ENDPOINT_METRICS_TIMEOUT_MILLIS), + log_timeout_millis: get_env_opt_u64(ENV_OBS_ENDPOINT_LOGS_TIMEOUT_MILLIS), profiling_endpoint: get_env_opt_str(ENV_OBS_PROFILING_ENDPOINT), traces_export_enabled: Some(get_env_bool(ENV_OBS_TRACES_EXPORT_ENABLED, DEFAULT_OBS_TRACES_EXPORT_ENABLED)), metrics_export_enabled: Some(get_env_bool(ENV_OBS_METRICS_EXPORT_ENABLED, DEFAULT_OBS_METRICS_EXPORT_ENABLED)), diff --git a/crates/obs/src/telemetry/otel.rs b/crates/obs/src/telemetry/otel.rs index 736dc0dcd..d575da2fd 100644 --- a/crates/obs/src/telemetry/otel.rs +++ b/crates/obs/src/telemetry/otel.rs @@ -57,11 +57,13 @@ use opentelemetry_sdk::{ metrics::{PeriodicReader, SdkMeterProvider}, trace::{RandomIdGenerator, Sampler, SdkTracerProvider}, }; +use percent_encoding::percent_decode_str; use rustfs_config::observability::{DEFAULT_OBS_LOG_MATCH_MODE, DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES}; use rustfs_config::{ APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME, DEFAULT_OBS_LOG_STDOUT_ENABLED, DEFAULT_OBS_LOGS_EXPORT_ENABLED, DEFAULT_OBS_METRICS_EXPORT_ENABLED, DEFAULT_OBS_TRACES_EXPORT_ENABLED, METER_INTERVAL, SAMPLE_RATIO, }; +use std::collections::HashMap; use std::{fs, io::IsTerminal, time::Duration}; use tracing::info; use tracing_error::ErrorLayer; @@ -94,7 +96,7 @@ use tracing_subscriber::{ /// /// # Note /// This function is intentionally kept unchanged from the pre-refactor -/// implementation to preserve existing OTLP behaviour. +/// implementation to preserve existing OTLP behavior. pub(super) fn init_observability_http( config: &OtelConfig, logger_level: &str, @@ -339,11 +341,19 @@ fn build_tracer_provider( return Ok(None); } - let exporter = opentelemetry_otlp::SpanExporter::builder() + let mut exporter_builder = opentelemetry_otlp::SpanExporter::builder() .with_http() .with_endpoint(trace_ep) .with_protocol(Protocol::HttpBinary) - .with_compression(Compression::Gzip) + .with_compression(Compression::Gzip); + let trace_headers = resolve_signal_headers(config.endpoint_headers.as_deref(), config.trace_headers.as_deref()); + if !trace_headers.is_empty() { + exporter_builder = exporter_builder.with_headers(trace_headers); + } + if let Some(timeout) = resolve_signal_timeout(config.endpoint_timeout_millis, config.trace_timeout_millis) { + exporter_builder = exporter_builder.with_timeout(timeout); + } + let exporter = exporter_builder .build() .map_err(|e| TelemetryError::BuildSpanExporter(e.to_string()))?; @@ -398,12 +408,20 @@ fn build_meter_provider( return Ok(None); } - let exporter = opentelemetry_otlp::MetricExporter::builder() + let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder() .with_http() .with_endpoint(metric_ep) .with_temporality(opentelemetry_sdk::metrics::Temporality::default()) .with_protocol(Protocol::HttpBinary) - .with_compression(Compression::Gzip) + .with_compression(Compression::Gzip); + let metric_headers = resolve_signal_headers(config.endpoint_headers.as_deref(), config.metric_headers.as_deref()); + if !metric_headers.is_empty() { + exporter_builder = exporter_builder.with_headers(metric_headers); + } + if let Some(timeout) = resolve_signal_timeout(config.endpoint_timeout_millis, config.metric_timeout_millis) { + exporter_builder = exporter_builder.with_timeout(timeout); + } + let exporter = exporter_builder .build() .map_err(|e| TelemetryError::BuildMetricExporter(e.to_string()))?; @@ -444,11 +462,19 @@ fn build_logger_provider( return Ok(None); } - let exporter = opentelemetry_otlp::LogExporter::builder() + let mut exporter_builder = opentelemetry_otlp::LogExporter::builder() .with_http() .with_endpoint(log_ep) .with_protocol(Protocol::HttpBinary) - .with_compression(Compression::Gzip) + .with_compression(Compression::Gzip); + let log_headers = resolve_signal_headers(config.endpoint_headers.as_deref(), config.log_headers.as_deref()); + if !log_headers.is_empty() { + exporter_builder = exporter_builder.with_headers(log_headers); + } + if let Some(timeout) = resolve_signal_timeout(config.endpoint_timeout_millis, config.log_timeout_millis) { + exporter_builder = exporter_builder.with_timeout(timeout); + } + let exporter = exporter_builder .build() .map_err(|e| TelemetryError::BuildLogExporter(e.to_string()))?; @@ -513,6 +539,39 @@ fn create_periodic_reader(interval: u64) -> PeriodicReader, signal_headers: Option<&str>) -> HashMap { + let mut headers = HashMap::new(); + if let Some(raw_headers) = common_headers { + headers.extend(parse_otlp_headers(raw_headers)); + } + if let Some(raw_headers) = signal_headers { + headers.extend(parse_otlp_headers(raw_headers)); + } + headers +} + +fn parse_otlp_headers(raw_headers: &str) -> HashMap { + raw_headers + .split(',') + .filter_map(|entry| { + let (key, value) = entry.split_once('=')?; + let key = key.trim(); + if key.is_empty() { + return None; + } + let value = percent_decode_str(value.trim()).decode_utf8().ok()?; + Some((key.to_string(), value.into_owned())) + }) + .collect() +} + +fn resolve_signal_timeout(common_timeout_millis: Option, signal_timeout_millis: Option) -> Option { + signal_timeout_millis + .or(common_timeout_millis) + .filter(|timeout_millis| *timeout_millis > 0) + .map(Duration::from_millis) +} + #[cfg(test)] mod tests { use super::*; @@ -539,4 +598,34 @@ mod tests { let sampler = build_tracer_sampler(1.2); assert!(format!("{sampler:?}").contains("AlwaysOn")); } + + #[test] + fn test_parse_otlp_headers_ignores_invalid_entries() { + let headers = parse_otlp_headers("Authorization=Bearer%20abc,empty=,missing, =ignored,key=value,bad=%FF"); + assert_eq!(headers.len(), 3); + assert_eq!(headers.get("Authorization"), Some(&"Bearer abc".to_string())); + assert_eq!(headers.get("empty"), Some(&"".to_string())); + assert_eq!(headers.get("key"), Some(&"value".to_string())); + } + + #[test] + fn test_resolve_signal_headers_signal_overrides_common() { + let headers = resolve_signal_headers(Some("k1=v1,k2=common"), Some("k2=signal,k3=v3")); + assert_eq!(headers.get("k1"), Some(&"v1".to_string())); + assert_eq!(headers.get("k2"), Some(&"signal".to_string())); + assert_eq!(headers.get("k3"), Some(&"v3".to_string())); + } + + #[test] + fn test_resolve_signal_timeout_prefers_signal_value() { + assert_eq!(resolve_signal_timeout(Some(2_000), Some(5_000)), Some(Duration::from_millis(5_000))); + } + + #[test] + fn test_resolve_signal_timeout_falls_back_to_common() { + assert_eq!(resolve_signal_timeout(Some(3_000), None), Some(Duration::from_millis(3_000))); + assert_eq!(resolve_signal_timeout(None, None), None); + assert_eq!(resolve_signal_timeout(Some(0), None), None); + assert_eq!(resolve_signal_timeout(None, Some(0)), None); + } }