Move telemetry configuration into YAML

src/config.rs

@@ -1,7 +1,9 @@
 use std::{collections::BTreeMap, fs, net::SocketAddr, path::Path, sync::Arc, time::Duration};
 
+use http::{HeaderName, HeaderValue, Uri};
 use serde::Deserialize;
 use thiserror::Error;
+use tracing_subscriber::EnvFilter;
 
 use crate::acl::{Acl, AclError};
 
@@ -13,6 +15,7 @@ pub struct Config {
     pub listen_addr: SocketAddr,
     pub connect_timeout: Duration,
     pub upstream_timeout: Duration,
+    pub telemetry: TelemetryConfig,
     pub(crate) acl: Arc<Acl>,
 }
 
@@ -34,11 +37,13 @@ impl Config {
         if file.upstream_timeout_ms == 0 {
             return Err(ConfigError::ZeroDuration("upstream_timeout_ms"));
         }
+        file.telemetry.validate()?;
 
         Ok(Self {
             listen_addr: file.listen_addr,
             connect_timeout: Duration::from_millis(file.connect_timeout_ms),
             upstream_timeout: Duration::from_millis(file.upstream_timeout_ms),
+            telemetry: file.telemetry,
             acl: Arc::new(Acl::from_config(file.acl)?),
         })
     }
@@ -53,6 +58,8 @@ struct FileConfig {
     connect_timeout_ms: u64,
     #[serde(default = "default_upstream_timeout_ms")]
     upstream_timeout_ms: u64,
+    #[serde(default)]
+    telemetry: TelemetryConfig,
     acl: AclConfig,
 }
 
@@ -68,6 +75,161 @@ const fn default_upstream_timeout_ms() -> u64 {
     DEFAULT_UPSTREAM_TIMEOUT_MS
 }
 
+#[derive(Clone, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct TelemetryConfig {
+    pub(crate) service_name: String,
+    pub(crate) log_filter: String,
+    pub(crate) resource_attributes: BTreeMap<String, String>,
+    pub(crate) otlp: OtlpConfig,
+}
+
+impl Default for TelemetryConfig {
+    fn default() -> Self {
+        Self {
+            service_name: "hermes".to_owned(),
+            log_filter: "info".to_owned(),
+            resource_attributes: BTreeMap::new(),
+            otlp: OtlpConfig::default(),
+        }
+    }
+}
+
+impl TelemetryConfig {
+    fn validate(&self) -> Result<(), ConfigError> {
+        if self.service_name.trim().is_empty() {
+            return Err(ConfigError::EmptyServiceName);
+        }
+        EnvFilter::try_new(&self.log_filter)
+            .map_err(|error| ConfigError::InvalidLogFilter(error.to_string()))?;
+        validate_endpoint(&self.otlp.endpoint, "telemetry.otlp.endpoint")?;
+        validate_endpoint_override(&self.otlp.traces.endpoint, "telemetry.otlp.traces.endpoint")?;
+        validate_endpoint_override(
+            &self.otlp.metrics.endpoint,
+            "telemetry.otlp.metrics.endpoint",
+        )?;
+        validate_endpoint_override(&self.otlp.logs.endpoint, "telemetry.otlp.logs.endpoint")?;
+
+        if self.otlp.timeout_ms == 0 {
+            return Err(ConfigError::ZeroDuration("telemetry.otlp.timeout_ms"));
+        }
+        if self.otlp.metrics.export_interval_ms == 0 {
+            return Err(ConfigError::ZeroDuration(
+                "telemetry.otlp.metrics.export_interval_ms",
+            ));
+        }
+        if !self.otlp.traces.sample_ratio.is_finite()
+            || !(0.0..=1.0).contains(&self.otlp.traces.sample_ratio)
+        {
+            return Err(ConfigError::InvalidSampleRatio(
+                self.otlp.traces.sample_ratio,
+            ));
+        }
+
+        for (name, value) in &self.otlp.headers {
+            if HeaderName::from_bytes(name.as_bytes()).is_err()
+                || HeaderValue::from_str(value).is_err()
+            {
+                return Err(ConfigError::InvalidHeader(name.clone()));
+            }
+        }
+
+        Ok(())
+    }
+}
+
+fn validate_endpoint_override(
+    endpoint: &Option<String>,
+    field: &'static str,
+) -> Result<(), ConfigError> {
+    if let Some(endpoint) = endpoint {
+        validate_endpoint(endpoint, field)?;
+    }
+    Ok(())
+}
+
+fn validate_endpoint(endpoint: &str, field: &'static str) -> Result<(), ConfigError> {
+    let uri = endpoint
+        .parse::<Uri>()
+        .map_err(|_| ConfigError::InvalidEndpoint(field))?;
+    if !matches!(uri.scheme_str(), Some("http" | "https")) || uri.authority().is_none() {
+        return Err(ConfigError::InvalidEndpoint(field));
+    }
+    Ok(())
+}
+
+#[derive(Clone, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub struct OtlpConfig {
+    pub(crate) endpoint: String,
+    pub(crate) timeout_ms: u64,
+    pub(crate) headers: BTreeMap<String, String>,
+    pub(crate) traces: TraceTelemetryConfig,
+    pub(crate) metrics: MetricTelemetryConfig,
+    pub(crate) logs: LogTelemetryConfig,
+}
+
+impl Default for OtlpConfig {
+    fn default() -> Self {
+        Self {
+            endpoint: "http://127.0.0.1:4318".to_owned(),
+            timeout_ms: 10_000,
+            headers: BTreeMap::new(),
+            traces: TraceTelemetryConfig::default(),
+            metrics: MetricTelemetryConfig::default(),
+            logs: LogTelemetryConfig::default(),
+        }
+    }
+}
+
+#[derive(Clone, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub(crate) struct TraceTelemetryConfig {
+    pub endpoint: Option<String>,
+    pub sample_ratio: f64,
+}
+
+impl Default for TraceTelemetryConfig {
+    fn default() -> Self {
+        Self {
+            endpoint: None,
+            sample_ratio: 1.0,
+        }
+    }
+}
+
+#[derive(Clone, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub(crate) struct MetricTelemetryConfig {
+    pub endpoint: Option<String>,
+    pub export_interval_ms: u64,
+    pub temporality: MetricTemporality,
+}
+
+impl Default for MetricTelemetryConfig {
+    fn default() -> Self {
+        Self {
+            endpoint: None,
+            export_interval_ms: 60_000,
+            temporality: MetricTemporality::Cumulative,
+        }
+    }
+}
+
+#[derive(Clone, Default, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+pub(crate) struct LogTelemetryConfig {
+    pub endpoint: Option<String>,
+}
+
+#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum MetricTemporality {
+    #[default]
+    Cumulative,
+    Delta,
+}
+
 #[derive(Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct AclConfig {
@@ -118,6 +280,16 @@ pub enum ConfigError {
     Yaml(#[from] serde_norway::Error),
     #[error("{0} must be greater than zero")]
     ZeroDuration(&'static str),
+    #[error("{0} must be an absolute http:// or https:// endpoint")]
+    InvalidEndpoint(&'static str),
+    #[error("telemetry.otlp.traces.sample_ratio must be between 0 and 1, got {0}")]
+    InvalidSampleRatio(f64),
+    #[error("invalid telemetry.log_filter: {0}")]
+    InvalidLogFilter(String),
+    #[error("invalid OTLP header {0:?}")]
+    InvalidHeader(String),
+    #[error("telemetry.service_name must not be empty")]
+    EmptyServiceName,
     #[error(transparent)]
     Acl(#[from] AclError),
 }
@@ -142,6 +314,108 @@ acl:
         assert_eq!(config.listen_addr, default_listen_addr());
         assert_eq!(config.connect_timeout, Duration::from_secs(10));
         assert_eq!(config.upstream_timeout, Duration::from_secs(30));
+        assert_eq!(config.telemetry.service_name, "hermes");
+        assert_eq!(config.telemetry.log_filter, "info");
+        assert_eq!(config.telemetry.otlp.endpoint, "http://127.0.0.1:4318");
+        assert_eq!(config.telemetry.otlp.traces.sample_ratio, 1.0);
+        assert_eq!(
+            config.telemetry.otlp.metrics.temporality,
+            MetricTemporality::Cumulative
+        );
+    }
+
+    #[test]
+    fn loads_telemetry_configuration() {
+        let config = Config::from_yaml(
+            r#"
+telemetry:
+  service_name: "hermes-test"
+  log_filter: "hermes=debug"
+  resource_attributes:
+    deployment.environment.name: "test"
+  otlp:
+    endpoint: "https://collector.example:4318"
+    timeout_ms: 5000
+    headers:
+      Authorization: "Bearer secret"
+    traces:
+      endpoint: "https://traces.example:4318"
+      sample_ratio: 0.25
+    metrics:
+      export_interval_ms: 1000
+      temporality: "delta"
+    logs:
+      endpoint: "https://logs.example:4318"
+acl: {}
+"#,
+        )
+        .unwrap();
+
+        assert_eq!(config.telemetry.service_name, "hermes-test");
+        assert_eq!(config.telemetry.log_filter, "hermes=debug");
+        assert_eq!(
+            config.telemetry.resource_attributes["deployment.environment.name"],
+            "test"
+        );
+        assert_eq!(config.telemetry.otlp.timeout_ms, 5000);
+        assert_eq!(
+            config.telemetry.otlp.headers["Authorization"],
+            "Bearer secret"
+        );
+        assert_eq!(
+            config.telemetry.otlp.traces.endpoint.as_deref(),
+            Some("https://traces.example:4318")
+        );
+        assert_eq!(config.telemetry.otlp.traces.sample_ratio, 0.25);
+        assert_eq!(config.telemetry.otlp.metrics.export_interval_ms, 1000);
+        assert_eq!(
+            config.telemetry.otlp.metrics.temporality,
+            MetricTemporality::Delta
+        );
+        assert_eq!(
+            config.telemetry.otlp.logs.endpoint.as_deref(),
+            Some("https://logs.example:4318")
+        );
+    }
+
+    #[test]
+    fn rejects_invalid_telemetry_configuration() {
+        for yaml in [
+            "telemetry:\n  otlp:\n    endpoint: \"collector:4318\"\nacl: {}\n",
+            "telemetry:\n  otlp:\n    endpoint: \"ftp://collector.example\"\nacl: {}\n",
+            "telemetry:\n  otlp:\n    traces:\n      endpoint: \"not-a-url\"\nacl: {}\n",
+        ] {
+            assert!(matches!(
+                Config::from_yaml(yaml),
+                Err(ConfigError::InvalidEndpoint(_))
+            ));
+        }
+
+        for yaml in [
+            "telemetry:\n  otlp:\n    timeout_ms: 0\nacl: {}\n",
+            "telemetry:\n  otlp:\n    metrics:\n      export_interval_ms: 0\nacl: {}\n",
+        ] {
+            assert!(matches!(
+                Config::from_yaml(yaml),
+                Err(ConfigError::ZeroDuration(_))
+            ));
+        }
+
+        let ratio = Config::from_yaml(
+            "telemetry:\n  otlp:\n    traces:\n      sample_ratio: 1.1\nacl: {}\n",
+        );
+        assert!(matches!(ratio, Err(ConfigError::InvalidSampleRatio(_))));
+
+        let filter = Config::from_yaml("telemetry:\n  log_filter: \"hermes=verbose\"\nacl: {}\n");
+        assert!(matches!(filter, Err(ConfigError::InvalidLogFilter(_))));
+
+        let header = Config::from_yaml(
+            "telemetry:\n  otlp:\n    headers:\n      Authorization: \"bad\\nvalue\"\nacl: {}\n",
+        );
+        assert!(matches!(header, Err(ConfigError::InvalidHeader(_))));
+
+        let service = Config::from_yaml("telemetry:\n  service_name: \"\"\nacl: {}\n");
+        assert!(matches!(service, Err(ConfigError::EmptyServiceName)));
     }
 
     #[test]

src/main.rs

@@ -1,4 +1,4 @@
-use std::path::PathBuf;
+use std::{env, path::PathBuf};
 
 use clap::{Parser, Subcommand};
 use hermes::{BoxError, acl, config::Config, telemetry};
@@ -10,8 +10,13 @@ use zeroize::Zeroizing;
     about = "A minimal HTTP forward proxy with OpenTelemetry observability"
 )]
 struct Cli {
-    #[arg(short, long, default_value = "hermes.yaml", value_name = "PATH")]
-    config: PathBuf,
+    #[arg(
+        short,
+        long,
+        value_name = "PATH",
+        help = "Configuration file path (overrides HERMES_CONFIG_PATH)"
+    )]
+    config: Option<PathBuf>,
     #[command(subcommand)]
     command: Option<Command>,
 }
@@ -35,12 +40,13 @@ async fn main() -> Result<(), BoxError> {
         return Ok(());
     }
 
-    let telemetry = telemetry::init()?;
-    let result: Result<(), BoxError> = async {
-        let config = Config::load(&cli.config)?;
-        hermes::run(config).await
-    }
-    .await;
+    let config_path = cli
+        .config
+        .or_else(|| env::var_os("HERMES_CONFIG_PATH").map(PathBuf::from))
+        .unwrap_or_else(|| PathBuf::from("hermes.yaml"));
+    let config = Config::load(config_path)?;
+    let telemetry = telemetry::init(&config.telemetry)?;
+    let result = hermes::run(config).await;
 
     telemetry.shutdown();
     result

src/proxy.rs

@@ -617,7 +617,7 @@ mod tests {
     };
 
     use super::*;
-    use crate::config::{AclConfig, Action, RuleConfig, UserConfig};
+    use crate::config::{AclConfig, Action, RuleConfig, TelemetryConfig, UserConfig};
 
     #[test]
     fn validates_absolute_http_targets() {
@@ -871,6 +871,7 @@ mod tests {
             listen_addr: address,
             connect_timeout: Duration::from_secs(2),
             upstream_timeout: Duration::from_secs(2),
+            telemetry: TelemetryConfig::default(),
             acl: Arc::new(acl),
         });
         let task = tokio::spawn(async move {

src/telemetry.rs

@@ -1,19 +1,30 @@
-use std::env;
+use std::{collections::HashMap, time::Duration};
 
 use http::{HeaderMap, HeaderName, HeaderValue, header};
 use opentelemetry::{
-    global,
+    KeyValue, global,
     propagation::{Extractor, Injector},
 };
 use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
-use opentelemetry_otlp::{LogExporter, MetricExporter, Protocol, SpanExporter, WithExportConfig};
+use opentelemetry_otlp::{
+    LogExporter, MetricExporter, Protocol, SpanExporter, WithExportConfig, WithHttpConfig,
+};
 use opentelemetry_sdk::{
-    Resource, logs::SdkLoggerProvider, metrics::SdkMeterProvider,
-    propagation::TraceContextPropagator, trace::SdkTracerProvider,
+    Resource,
+    logs::{BatchConfigBuilder as LogBatchConfigBuilder, BatchLogProcessor, SdkLoggerProvider},
+    metrics::{PeriodicReader, SdkMeterProvider, Temporality},
+    propagation::TraceContextPropagator,
+    trace::{
+        BatchConfigBuilder as TraceBatchConfigBuilder, BatchSpanProcessor, Sampler,
+        SdkTracerProvider,
+    },
 };
 use tracing_subscriber::{EnvFilter, Layer, layer::SubscriberExt, util::SubscriberInitExt};
 
-use crate::BoxError;
+use crate::{
+    BoxError,
+    config::{MetricTemporality, OtlpConfig, TelemetryConfig},
+};
 
 pub struct TelemetryGuard {
     tracer_provider: SdkTracerProvider,
@@ -35,18 +46,43 @@ impl TelemetryGuard {
     }
 }
 
-pub fn init() -> Result<TelemetryGuard, BoxError> {
+pub fn init(config: &TelemetryConfig) -> Result<TelemetryGuard, BoxError> {
     global::set_text_map_propagator(TraceContextPropagator::new());
 
-    let service_name = env::var("OTEL_SERVICE_NAME").unwrap_or_else(|_| "hermes".to_owned());
-    let resource = Resource::builder().with_service_name(service_name).build();
+    let resource = Resource::builder_empty()
+        .with_service_name(config.service_name.clone())
+        .with_attributes(
+            config
+                .resource_attributes
+                .iter()
+                .map(|(key, value)| KeyValue::new(key.clone(), value.clone())),
+        )
+        .build();
+    let timeout = Duration::from_millis(config.otlp.timeout_ms);
+    let headers = otlp_headers(&config.otlp);
 
     let metric_exporter = MetricExporter::builder()
         .with_http()
         .with_protocol(Protocol::HttpBinary)
+        .with_endpoint(endpoint(
+            &config.otlp,
+            &config.otlp.metrics.endpoint,
+            "metrics",
+        ))
+        .with_timeout(timeout)
+        .with_headers(headers.clone())
+        .with_temporality(match config.otlp.metrics.temporality {
+            MetricTemporality::Cumulative => Temporality::Cumulative,
+            MetricTemporality::Delta => Temporality::Delta,
+        })
         .build()?;
+    let metric_reader = PeriodicReader::builder(metric_exporter)
+        .with_interval(Duration::from_millis(
+            config.otlp.metrics.export_interval_ms,
+        ))
+        .build();
     let meter_provider = SdkMeterProvider::builder()
-        .with_periodic_exporter(metric_exporter)
+        .with_reader(metric_reader)
         .with_resource(resource.clone())
         .build();
     global::set_meter_provider(meter_provider.clone());
@@ -54,24 +90,55 @@ pub fn init() -> Result<TelemetryGuard, BoxError> {
     let log_exporter = LogExporter::builder()
         .with_http()
         .with_protocol(Protocol::HttpBinary)
+        .with_endpoint(endpoint(&config.otlp, &config.otlp.logs.endpoint, "logs"))
+        .with_timeout(timeout)
+        .with_headers(headers.clone())
         .build()?;
+    let log_processor = BatchLogProcessor::builder(log_exporter)
+        .with_batch_config(
+            LogBatchConfigBuilder::default()
+                .with_max_queue_size(2048)
+                .with_scheduled_delay(Duration::from_secs(1))
+                .with_max_export_batch_size(512)
+                .build(),
+        )
+        .build();
     let logger_provider = SdkLoggerProvider::builder()
-        .with_batch_exporter(log_exporter)
+        .with_log_processor(log_processor)
         .with_resource(resource.clone())
         .build();
 
     let span_exporter = SpanExporter::builder()
         .with_http()
         .with_protocol(Protocol::HttpBinary)
+        .with_endpoint(endpoint(
+            &config.otlp,
+            &config.otlp.traces.endpoint,
+            "traces",
+        ))
+        .with_timeout(timeout)
+        .with_headers(headers)
         .build()?;
+    let span_processor = BatchSpanProcessor::builder(span_exporter)
+        .with_batch_config(
+            TraceBatchConfigBuilder::default()
+                .with_max_queue_size(2048)
+                .with_scheduled_delay(Duration::from_secs(5))
+                .with_max_export_batch_size(512)
+                .build(),
+        )
+        .build();
     let tracer_provider = SdkTracerProvider::builder()
-        .with_batch_exporter(span_exporter)
+        .with_span_processor(span_processor)
+        .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
+            config.otlp.traces.sample_ratio,
+        ))))
         .with_resource(resource)
         .build();
     global::set_tracer_provider(tracer_provider.clone());
 
-    let fmt_filter = env_filter();
-    let otel_log_filter = exporter_log_filter(env_filter());
+    let fmt_filter = EnvFilter::try_new(&config.log_filter)?;
+    let otel_log_filter = exporter_log_filter(EnvFilter::try_new(&config.log_filter)?);
     let tracer = global::tracer("hermes");
 
     tracing_subscriber::registry()
@@ -87,8 +154,18 @@ pub fn init() -> Result<TelemetryGuard, BoxError> {
     })
 }
 
-fn env_filter() -> EnvFilter {
-    EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"))
+fn endpoint(config: &OtlpConfig, override_endpoint: &Option<String>, signal: &str) -> String {
+    override_endpoint
+        .clone()
+        .unwrap_or_else(|| format!("{}/v1/{signal}", config.endpoint.trim_end_matches('/')))
+}
+
+fn otlp_headers(config: &OtlpConfig) -> HashMap<String, String> {
+    config
+        .headers
+        .iter()
+        .map(|(key, value)| (key.clone(), value.clone()))
+        .collect()
 }
 
 fn exporter_log_filter(filter: EnvFilter) -> EnvFilter {
@@ -177,4 +254,28 @@ mod tests {
         assert!(!headers.contains_key("proxy-authorization"));
         assert_eq!(headers["x-keep"], "value");
     }
+
+    #[test]
+    fn signal_endpoints_inherit_the_common_endpoint() {
+        let config = OtlpConfig::default();
+        assert_eq!(
+            endpoint(&config, &None, "traces"),
+            "http://127.0.0.1:4318/v1/traces"
+        );
+        assert_eq!(
+            endpoint(
+                &config,
+                &Some("http://signals.example:4318/custom/traces".to_owned()),
+                "traces",
+            ),
+            "http://signals.example:4318/custom/traces"
+        );
+
+        let mut with_slash = OtlpConfig::default();
+        with_slash.endpoint.push('/');
+        assert_eq!(
+            endpoint(&with_slash, &None, "metrics"),
+            "http://127.0.0.1:4318/v1/metrics"
+        );
+    }
 }