cqlite_core/observability/
config.rs

1//! Observability configuration, populated from the standard `CQLITE_OTEL_*`
2//! environment variables (epic #1031, issue #1032).
3//!
4//! The config type is always compiled (it carries no OpenTelemetry types), so
5//! callers and tests can construct and inspect it whether or not the
6//! `observability` feature is enabled. The actual exporter wiring lives in
7//! [`crate::observability`]'s feature-gated `init`.
8//!
9//! # Environment variables
10//!
11//! | Variable                       | Type    | Default                      | Meaning |
12//! |--------------------------------|---------|------------------------------|---------|
13//! | `CQLITE_OTEL_ENABLED`          | bool    | `false`                      | Master switch; when false, `init` is a no-op even with the feature on. |
14//! | `CQLITE_OTEL_ENDPOINT`         | string  | `http://localhost:4317`      | OTLP collector endpoint (gRPC) or base URL (HTTP). |
15//! | `CQLITE_OTEL_PROTOCOL`         | enum    | `grpc`                       | `grpc` or `http` (HTTP/protobuf). |
16//! | `CQLITE_OTEL_SERVICE_NAME`     | string  | `cqlite`                     | `service.name` resource attribute. |
17//! | `CQLITE_OTEL_SERVICE_VERSION`  | string  | crate version                | `service.version` resource attribute. |
18//! | `CQLITE_OTEL_SAMPLING_RATIO`   | f64     | `1.0`                        | Trace-ID-ratio sampling probability, clamped to `[0.0, 1.0]`. |
19//! | `CQLITE_OTEL_TIMEOUT_MS`       | u64     | `10000`                      | Exporter export timeout in milliseconds. |
20//!
21//! Booleans accept `1/0`, `true/false`, `yes/no`, `on/off` (case-insensitive).
22
23use std::time::Duration;
24
25/// OTLP wire protocol selection.
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
27pub enum OtelProtocol {
28    /// OTLP over gRPC (default; port 4317).
29    #[default]
30    Grpc,
31    /// OTLP over HTTP/protobuf (port 4318).
32    Http,
33}
34
35impl OtelProtocol {
36    /// Parse a protocol string. Recognises `grpc` and `http`
37    /// (and the alias `http/protobuf`), case-insensitively.
38    pub fn parse(s: &str) -> Option<Self> {
39        match s.trim().to_ascii_lowercase().as_str() {
40            "grpc" => Some(Self::Grpc),
41            "http" | "http/protobuf" | "http-protobuf" | "httpbinary" => Some(Self::Http),
42            _ => None,
43        }
44    }
45
46    /// Stable lowercase name (`"grpc"` / `"http"`).
47    pub fn as_str(self) -> &'static str {
48        match self {
49            Self::Grpc => "grpc",
50            Self::Http => "http",
51        }
52    }
53}
54
55/// Default OTLP gRPC endpoint.
56pub const DEFAULT_ENDPOINT: &str = "http://localhost:4317";
57/// Default service name resource attribute.
58pub const DEFAULT_SERVICE_NAME: &str = "cqlite";
59/// Default exporter timeout.
60pub const DEFAULT_TIMEOUT: Duration = Duration::from_millis(10_000);
61
62/// Runtime observability configuration.
63///
64/// Construct via [`ObservabilityConfig::from_env`] for the standard
65/// `CQLITE_OTEL_*` behaviour, or [`ObservabilityConfig::builder`] to set fields
66/// programmatically (e.g. in tests that want a disabled / no-export config).
67#[derive(Debug, Clone)]
68pub struct ObservabilityConfig {
69    /// Master enable switch. When `false`, `init` returns an inert guard and
70    /// installs no exporters or global providers, regardless of the feature.
71    pub enabled: bool,
72    /// OTLP endpoint (gRPC endpoint or HTTP base URL).
73    pub endpoint: String,
74    /// Wire protocol.
75    pub protocol: OtelProtocol,
76    /// `service.name` resource attribute.
77    pub service_name: String,
78    /// `service.version` resource attribute.
79    pub service_version: String,
80    /// Trace-ID-ratio sampling probability in `[0.0, 1.0]`. Wrapped in a
81    /// parent-based sampler so children of a sampled span are always recorded.
82    pub sampling_ratio: f64,
83    /// Exporter export timeout.
84    pub timeout: Duration,
85    /// Opt-in, default-OFF presence-oracle false-negative verification (issue
86    /// #2163). When `true`, an SSTable read whose bloom/BTI-trie reports a key
87    /// "definitely absent" runs an AUTHORITATIVE confirmation scan and increments
88    /// `cqlite.read.bloom.false_negatives` on a contradiction. Populated from
89    /// `CQLITE_VERIFY_PRESENCE_ORACLE`; the read path itself consults the
90    /// process-global switch in
91    /// [`crate::storage::sstable::reader::presence_verification`], which reads the
92    /// SAME environment variable, so this field mirrors that runtime state.
93    pub verify_presence_oracle: bool,
94}
95
96impl Default for ObservabilityConfig {
97    fn default() -> Self {
98        Self {
99            enabled: false,
100            endpoint: DEFAULT_ENDPOINT.to_string(),
101            protocol: OtelProtocol::Grpc,
102            service_name: DEFAULT_SERVICE_NAME.to_string(),
103            service_version: env!("CARGO_PKG_VERSION").to_string(),
104            sampling_ratio: 1.0,
105            timeout: DEFAULT_TIMEOUT,
106            verify_presence_oracle: false,
107        }
108    }
109}
110
111impl ObservabilityConfig {
112    /// Start from defaults and override fields fluently.
113    pub fn builder() -> ObservabilityConfigBuilder {
114        ObservabilityConfigBuilder {
115            config: Self::default(),
116        }
117    }
118
119    /// Populate from the `CQLITE_OTEL_*` environment variables, falling back to
120    /// the documented defaults for anything unset or unparseable.
121    ///
122    /// Unparseable values fall back to the default rather than erroring, so a
123    /// typo in an env var never crashes the host process.
124    pub fn from_env() -> Self {
125        let mut cfg = Self::default();
126
127        if let Some(v) = env_str("CQLITE_OTEL_ENABLED") {
128            if let Some(b) = parse_bool(&v) {
129                cfg.enabled = b;
130            }
131        }
132        if let Some(v) = env_str("CQLITE_OTEL_ENDPOINT") {
133            cfg.endpoint = v;
134        }
135        if let Some(v) = env_str("CQLITE_OTEL_PROTOCOL") {
136            if let Some(p) = OtelProtocol::parse(&v) {
137                cfg.protocol = p;
138            }
139        }
140        if let Some(v) = env_str("CQLITE_OTEL_SERVICE_NAME") {
141            cfg.service_name = v;
142        }
143        if let Some(v) = env_str("CQLITE_OTEL_SERVICE_VERSION") {
144            cfg.service_version = v;
145        }
146        if let Some(v) = env_str("CQLITE_OTEL_SAMPLING_RATIO") {
147            if let Ok(r) = v.trim().parse::<f64>() {
148                // Reject non-finite values (NaN, ±inf) — they survive `clamp`
149                // and would violate the documented [0.0, 1.0] contract. Keep the
150                // default in that case.
151                if r.is_finite() {
152                    cfg.sampling_ratio = r.clamp(0.0, 1.0);
153                }
154            }
155        }
156        if let Some(v) = env_str("CQLITE_OTEL_TIMEOUT_MS") {
157            if let Ok(ms) = v.trim().parse::<u64>() {
158                cfg.timeout = Duration::from_millis(ms);
159            }
160        }
161        // Issue #2163: presence-oracle false-negative verification switch. Mirrors
162        // the process-global runtime switch the read path consults (same env var).
163        if let Some(v) = env_str("CQLITE_VERIFY_PRESENCE_ORACLE") {
164            if let Some(b) = parse_bool(&v) {
165                cfg.verify_presence_oracle = b;
166            }
167        }
168
169        cfg.sampling_ratio = sanitize_ratio(cfg.sampling_ratio);
170        cfg
171    }
172}
173
174/// Clamp a sampling ratio into `[0.0, 1.0]`, falling back to full sampling for
175/// non-finite inputs (NaN / ±inf) which would otherwise bypass the clamp.
176fn sanitize_ratio(r: f64) -> f64 {
177    if r.is_finite() {
178        r.clamp(0.0, 1.0)
179    } else {
180        1.0
181    }
182}
183
184/// Builder for [`ObservabilityConfig`].
185#[derive(Debug, Clone)]
186pub struct ObservabilityConfigBuilder {
187    config: ObservabilityConfig,
188}
189
190impl ObservabilityConfigBuilder {
191    /// Set the master enable switch.
192    pub fn enabled(mut self, enabled: bool) -> Self {
193        self.config.enabled = enabled;
194        self
195    }
196    /// Set the OTLP endpoint.
197    pub fn endpoint(mut self, endpoint: impl Into<String>) -> Self {
198        self.config.endpoint = endpoint.into();
199        self
200    }
201    /// Set the wire protocol.
202    pub fn protocol(mut self, protocol: OtelProtocol) -> Self {
203        self.config.protocol = protocol;
204        self
205    }
206    /// Set the `service.name`.
207    pub fn service_name(mut self, name: impl Into<String>) -> Self {
208        self.config.service_name = name.into();
209        self
210    }
211    /// Set the `service.version`.
212    pub fn service_version(mut self, version: impl Into<String>) -> Self {
213        self.config.service_version = version.into();
214        self
215    }
216    /// Set the trace-ID-ratio sampling probability (clamped to `[0.0, 1.0]`;
217    /// non-finite values fall back to full sampling).
218    pub fn sampling_ratio(mut self, ratio: f64) -> Self {
219        self.config.sampling_ratio = sanitize_ratio(ratio);
220        self
221    }
222    /// Set the exporter timeout.
223    pub fn timeout(mut self, timeout: Duration) -> Self {
224        self.config.timeout = timeout;
225        self
226    }
227    /// Enable/disable presence-oracle false-negative verification (issue #2163).
228    pub fn verify_presence_oracle(mut self, verify: bool) -> Self {
229        self.config.verify_presence_oracle = verify;
230        self
231    }
232    /// Finish building.
233    pub fn build(self) -> ObservabilityConfig {
234        let mut cfg = self.config;
235        cfg.sampling_ratio = sanitize_ratio(cfg.sampling_ratio);
236        cfg
237    }
238}
239
240fn env_str(key: &str) -> Option<String> {
241    match std::env::var(key) {
242        Ok(v) if !v.trim().is_empty() => Some(v),
243        _ => None,
244    }
245}
246
247fn parse_bool(s: &str) -> Option<bool> {
248    match s.trim().to_ascii_lowercase().as_str() {
249        "1" | "true" | "yes" | "on" => Some(true),
250        "0" | "false" | "no" | "off" => Some(false),
251        _ => None,
252    }
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258
259    #[test]
260    fn defaults_are_disabled_and_sane() {
261        let cfg = ObservabilityConfig::default();
262        assert!(!cfg.enabled);
263        assert_eq!(cfg.endpoint, DEFAULT_ENDPOINT);
264        assert_eq!(cfg.protocol, OtelProtocol::Grpc);
265        assert_eq!(cfg.service_name, DEFAULT_SERVICE_NAME);
266        assert_eq!(cfg.sampling_ratio, 1.0);
267        assert_eq!(cfg.timeout, DEFAULT_TIMEOUT);
268        assert_eq!(cfg.service_version, env!("CARGO_PKG_VERSION"));
269        assert!(!cfg.verify_presence_oracle);
270    }
271
272    #[test]
273    fn builder_overrides_and_clamps() {
274        let cfg = ObservabilityConfig::builder()
275            .enabled(true)
276            .endpoint("http://collector:4317")
277            .protocol(OtelProtocol::Http)
278            .service_name("svc")
279            .service_version("9.9.9")
280            .sampling_ratio(2.0) // clamps to 1.0
281            .timeout(Duration::from_millis(500))
282            .build();
283        assert!(cfg.enabled);
284        assert_eq!(cfg.endpoint, "http://collector:4317");
285        assert_eq!(cfg.protocol, OtelProtocol::Http);
286        assert_eq!(cfg.service_name, "svc");
287        assert_eq!(cfg.service_version, "9.9.9");
288        assert_eq!(cfg.sampling_ratio, 1.0);
289        assert_eq!(cfg.timeout, Duration::from_millis(500));
290    }
291
292    #[test]
293    fn negative_ratio_clamps_to_zero() {
294        let cfg = ObservabilityConfig::builder().sampling_ratio(-1.0).build();
295        assert_eq!(cfg.sampling_ratio, 0.0);
296    }
297
298    #[test]
299    fn non_finite_ratio_falls_back_to_full_sampling() {
300        for bad in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
301            let cfg = ObservabilityConfig::builder().sampling_ratio(bad).build();
302            assert!(cfg.sampling_ratio.is_finite());
303            assert_eq!(cfg.sampling_ratio, 1.0);
304        }
305    }
306
307    #[test]
308    fn protocol_parse() {
309        assert_eq!(OtelProtocol::parse("grpc"), Some(OtelProtocol::Grpc));
310        assert_eq!(OtelProtocol::parse("GRPC"), Some(OtelProtocol::Grpc));
311        assert_eq!(OtelProtocol::parse("http"), Some(OtelProtocol::Http));
312        assert_eq!(
313            OtelProtocol::parse("http/protobuf"),
314            Some(OtelProtocol::Http)
315        );
316        assert_eq!(OtelProtocol::parse("nope"), None);
317        assert_eq!(OtelProtocol::Grpc.as_str(), "grpc");
318        assert_eq!(OtelProtocol::Http.as_str(), "http");
319    }
320
321    #[test]
322    fn bool_parsing() {
323        for t in ["1", "true", "TRUE", "yes", "on"] {
324            assert_eq!(parse_bool(t), Some(true), "{t}");
325        }
326        for f in ["0", "false", "no", "off"] {
327            assert_eq!(parse_bool(f), Some(false), "{f}");
328        }
329        assert_eq!(parse_bool("maybe"), None);
330    }
331
332    // from_env mutates process-global env, so all env-touching assertions live
333    // in ONE test to avoid cross-test races (tests share a process).
334    #[test]
335    fn from_env_parses_and_falls_back() {
336        let keys = [
337            "CQLITE_OTEL_ENABLED",
338            "CQLITE_OTEL_ENDPOINT",
339            "CQLITE_OTEL_PROTOCOL",
340            "CQLITE_OTEL_SERVICE_NAME",
341            "CQLITE_OTEL_SERVICE_VERSION",
342            "CQLITE_OTEL_SAMPLING_RATIO",
343            "CQLITE_OTEL_TIMEOUT_MS",
344            "CQLITE_VERIFY_PRESENCE_ORACLE",
345        ];
346        // Save + clear.
347        let saved: Vec<_> = keys.iter().map(|k| (*k, std::env::var(k).ok())).collect();
348        for k in keys {
349            std::env::remove_var(k);
350        }
351
352        // All unset -> defaults.
353        let cfg = ObservabilityConfig::from_env();
354        assert!(!cfg.enabled);
355        assert_eq!(cfg.endpoint, DEFAULT_ENDPOINT);
356        assert!(!cfg.verify_presence_oracle);
357
358        // Set every var.
359        std::env::set_var("CQLITE_OTEL_ENABLED", "true");
360        std::env::set_var("CQLITE_VERIFY_PRESENCE_ORACLE", "on");
361        std::env::set_var("CQLITE_OTEL_ENDPOINT", "http://c:4318");
362        std::env::set_var("CQLITE_OTEL_PROTOCOL", "http");
363        std::env::set_var("CQLITE_OTEL_SERVICE_NAME", "mysvc");
364        std::env::set_var("CQLITE_OTEL_SERVICE_VERSION", "1.2.3");
365        std::env::set_var("CQLITE_OTEL_SAMPLING_RATIO", "0.25");
366        std::env::set_var("CQLITE_OTEL_TIMEOUT_MS", "2500");
367        let cfg = ObservabilityConfig::from_env();
368        assert!(cfg.enabled);
369        assert_eq!(cfg.endpoint, "http://c:4318");
370        assert_eq!(cfg.protocol, OtelProtocol::Http);
371        assert_eq!(cfg.service_name, "mysvc");
372        assert_eq!(cfg.service_version, "1.2.3");
373        assert_eq!(cfg.sampling_ratio, 0.25);
374        assert_eq!(cfg.timeout, Duration::from_millis(2500));
375        assert!(cfg.verify_presence_oracle);
376
377        // Unparseable ratio / timeout fall back to defaults; bad protocol keeps default.
378        std::env::set_var("CQLITE_OTEL_SAMPLING_RATIO", "not-a-number");
379        std::env::set_var("CQLITE_OTEL_TIMEOUT_MS", "xyz");
380        std::env::set_var("CQLITE_OTEL_PROTOCOL", "carrier-pigeon");
381        let cfg = ObservabilityConfig::from_env();
382        assert_eq!(cfg.sampling_ratio, 1.0);
383        assert_eq!(cfg.timeout, DEFAULT_TIMEOUT);
384        assert_eq!(cfg.protocol, OtelProtocol::Grpc);
385
386        // Restore.
387        for (k, v) in saved {
388            match v {
389                Some(val) => std::env::set_var(k, val),
390                None => std::env::remove_var(k),
391            }
392        }
393    }
394}
cqlite_core/observability/config.rs

cqlite_core/observability/
config.rs
