faucet_cli/
obs.rs

1//! Shared observability setup (Prometheus + tracing) used by `run` and
2//! `schedule`. `install_observability` is idempotent, so calling this once per
3//! process is safe even though `main.rs` already installed a basic subscriber.
4
5use crate::config::PipelineConfig;
6use crate::error::CliResult;
7use faucet_core::{ObservabilityConfig, PrometheusConfig, TracingConfig, install_observability};
8
9/// Resolve the effective tracing level using the documented precedence:
10/// `FAUCET_LOG` > `RUST_LOG` > YAML `observability.tracing.level` > `None`.
11/// (`cli_flag` is reserved for a future call-site that forwards `--log-level`.)
12pub fn resolve_tracing_level(cli_flag: Option<&str>, yaml_level: Option<&str>) -> Option<String> {
13    if let Some(l) = cli_flag {
14        return Some(l.to_string());
15    }
16    if let Ok(l) = std::env::var("FAUCET_LOG")
17        && !l.is_empty()
18    {
19        return Some(l);
20    }
21    if let Ok(l) = std::env::var("RUST_LOG")
22        && !l.is_empty()
23    {
24        return Some(l);
25    }
26    yaml_level.map(|s| s.to_string())
27}
28
29/// Build the core `ObservabilityConfig` from the CLI config, including OTLP when
30/// the `otel` feature is compiled in. Logs a one-shot warning if an `otel:` block
31/// is present but the binary was built without `--features otel`.
32pub fn build_observability_config(cfg: &PipelineConfig) -> ObservabilityConfig {
33    let level = resolve_tracing_level(
34        None,
35        cfg.observability
36            .as_ref()
37            .and_then(|o| o.tracing.as_ref())
38            .and_then(|t| t.level.as_deref()),
39    );
40    // `mut` is only needed when the `otel` feature populates `obs.otel` below.
41    #[cfg_attr(not(feature = "otel"), allow(unused_mut))]
42    let mut obs = ObservabilityConfig {
43        prometheus: cfg
44            .observability
45            .as_ref()
46            .and_then(|o| o.prometheus.as_ref())
47            .map(|p| PrometheusConfig {
48                listen: p.listen.clone(),
49                buckets: p.buckets.clone(),
50            }),
51        tracing: level.map(|l| TracingConfig { level: l }),
52        ..Default::default()
53    };
54
55    let otel_present = cfg
56        .observability
57        .as_ref()
58        .and_then(|o| o.otel.as_ref())
59        .is_some();
60    #[cfg(feature = "otel")]
61    {
62        if let Some(spec) = cfg.observability.as_ref().and_then(|o| o.otel.as_ref()) {
63            match spec.to_core() {
64                Ok(c) => obs.otel = Some(c),
65                Err(e) => tracing::warn!("ignoring invalid otel config: {e}"),
66            }
67        }
68    }
69    #[cfg(not(feature = "otel"))]
70    {
71        if otel_present {
72            tracing::warn!(
73                "observability.otel is configured but this binary was built without --features otel; OTLP export is disabled"
74            );
75        }
76    }
77    let _ = otel_present;
78    obs
79}
80
81/// Install Prometheus + tracing from the config's `observability:` block. Logs
82/// (does not fail) when a recorder/subscriber is already installed.
83pub fn install(cfg: &PipelineConfig) -> CliResult<()> {
84    let obs_cfg = build_observability_config(cfg);
85    let report = install_observability(&obs_cfg)?;
86    if let Some(addr) = report.prometheus_listen.as_deref() {
87        tracing::info!("Prometheus /metrics listening on {addr}");
88    }
89    if report.prometheus_already_installed {
90        tracing::warn!(
91            "Prometheus recorder already installed; metrics route through the existing recorder"
92        );
93    }
94    if report.tracing_already_installed {
95        tracing::warn!(
96            "tracing subscriber already installed; logs route through the existing subscriber"
97        );
98    }
99    if report.otel_installed {
100        tracing::info!("OTLP export enabled: {}", report.otel_signals.join(", "));
101    }
102    Ok(())
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108    use std::sync::Mutex;
109    static ENV_LOCK: Mutex<()> = Mutex::new(());
110
111    fn with_clean_env<F: FnOnce()>(f: F) {
112        let _g = ENV_LOCK.lock().unwrap();
113        unsafe {
114            std::env::remove_var("FAUCET_LOG");
115            std::env::remove_var("RUST_LOG");
116        }
117        f();
118    }
119
120    #[test]
121    fn cli_flag_beats_env_and_yaml() {
122        with_clean_env(|| {
123            unsafe {
124                std::env::set_var("FAUCET_LOG", "debug");
125                std::env::set_var("RUST_LOG", "trace");
126            }
127            assert_eq!(
128                resolve_tracing_level(Some("error"), Some("info")).as_deref(),
129                Some("error")
130            );
131        });
132    }
133
134    #[test]
135    fn faucet_log_beats_rust_log_and_yaml() {
136        with_clean_env(|| {
137            unsafe {
138                std::env::set_var("FAUCET_LOG", "debug");
139                std::env::set_var("RUST_LOG", "trace");
140            }
141            assert_eq!(
142                resolve_tracing_level(None, Some("info")).as_deref(),
143                Some("debug")
144            );
145        });
146    }
147
148    #[test]
149    fn rust_log_beats_yaml() {
150        with_clean_env(|| {
151            unsafe {
152                std::env::set_var("RUST_LOG", "trace");
153            }
154            assert_eq!(
155                resolve_tracing_level(None, Some("info")).as_deref(),
156                Some("trace")
157            );
158        });
159    }
160
161    #[test]
162    fn yaml_used_when_no_flag_or_env() {
163        with_clean_env(|| {
164            assert_eq!(
165                resolve_tracing_level(None, Some("info")).as_deref(),
166                Some("info")
167            );
168        });
169    }
170
171    #[test]
172    fn none_returned_when_nothing_set() {
173        with_clean_env(|| {
174            assert_eq!(resolve_tracing_level(None, None), None);
175        });
176    }
177
178    #[test]
179    fn maps_otel_spec_into_observability_config() {
180        let yaml = r#"
181version: 1
182pipeline:
183  source: { type: rest, config: { base_url: "http://x" } }
184  sink: { type: stdout, config: {} }
185observability:
186  otel: { endpoint: "http://c:4317" }
187"#;
188        let cfg = crate::config::parse_with_extension(yaml, "yaml").unwrap();
189        let obs = build_observability_config(&cfg);
190        #[cfg(feature = "otel")]
191        assert!(obs.otel.is_some());
192        let _ = obs;
193    }
194}
faucet_cli/obs.rs

faucet_cli/
obs.rs
