reddb_server/telemetry/
slow_query_logger.rs

1//! Dedicated slow-query sink — writes structured JSON lines to `red-slow.log`.
2//!
3//! Below-threshold calls pay only one relaxed atomic load and return
4//! immediately with zero allocations. Above-threshold calls go through
5//! a deterministic counter-based sampler before writing to the
6//! `NonBlocking` writer, which pushes the bytes onto a channel and
7//! returns without waiting for the disk write.
8
9use std::io::Write;
10use std::path::PathBuf;
11use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
12use std::sync::{Arc, Mutex};
13
14use tracing_appender::non_blocking::{NonBlocking, NonBlockingBuilder, WorkerGuard};
15
16use crate::runtime::EffectiveScope;
17
18// ---------------------------------------------------------------------------
19// Public types
20// ---------------------------------------------------------------------------
21
22pub struct SlowQueryOpts {
23    pub log_dir: PathBuf,
24    pub threshold_ms: u64,
25    /// 0..=100; values > 100 are clamped to 100.
26    pub sample_pct: u8,
27}
28
29/// Closed enum of query kinds emitted in the slow-query log.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum QueryKind {
32    Select,
33    Insert,
34    Update,
35    Delete,
36    Bulk,
37    Aggregate,
38    DDL,
39    Internal,
40}
41
42impl QueryKind {
43    /// Every variant, in stable order. The histogram substrate (#1241)
44    /// keys one fixed cell per variant off this array, so the order
45    /// here defines the bounded `kind` cardinality budget (ADR 0060 §4).
46    pub const ALL: [QueryKind; 8] = [
47        Self::Select,
48        Self::Insert,
49        Self::Update,
50        Self::Delete,
51        Self::Bulk,
52        Self::Aggregate,
53        Self::DDL,
54        Self::Internal,
55    ];
56
57    pub fn as_str(self) -> &'static str {
58        match self {
59            Self::Select => "select",
60            Self::Insert => "insert",
61            Self::Update => "update",
62            Self::Delete => "delete",
63            Self::Bulk => "bulk",
64            Self::Aggregate => "aggregate",
65            Self::DDL => "ddl",
66            Self::Internal => "internal",
67        }
68    }
69
70    /// Dense index into [`QueryKind::ALL`] — the cell offset used by the
71    /// latency histogram substrate.
72    pub fn index(self) -> usize {
73        match self {
74            Self::Select => 0,
75            Self::Insert => 1,
76            Self::Update => 2,
77            Self::Delete => 3,
78            Self::Bulk => 4,
79            Self::Aggregate => 5,
80            Self::DDL => 6,
81            Self::Internal => 7,
82        }
83    }
84}
85
86// ---------------------------------------------------------------------------
87// SlowQueryLogger
88// ---------------------------------------------------------------------------
89
90pub struct SlowQueryLogger {
91    /// Interior-mutable because `Write::write_all` takes `&mut self`.
92    writer: Mutex<NonBlocking>,
93    /// Keeps the background writer thread alive for the process lifetime.
94    _guard: WorkerGuard,
95    threshold_ms: AtomicU64,
96    /// 0..100; 100 means "emit every above-threshold query".
97    sample_pct: AtomicU8,
98    /// Monotonic counter across above-threshold calls — drives round-robin
99    /// sampling so exactly `sample_pct`% of above-threshold calls emit.
100    above_count: AtomicU64,
101    /// Durable telemetry substrate (ADR 0060). When attached, above-threshold
102    /// events are dual-written: file sink (existing behavior) + ring store.
103    store: std::sync::OnceLock<std::sync::Arc<super::slow_query_store::SlowQueryStore>>,
104}
105
106impl SlowQueryLogger {
107    pub fn new(opts: SlowQueryOpts) -> Arc<Self> {
108        let path = reddb_file::layout::legacy_slow_query_log_path(&opts.log_dir);
109        Self::open_at(path, opts.threshold_ms, opts.sample_pct)
110    }
111
112    /// Resolve a [`crate::storage::layout::LogDestination`] into a concrete
113    /// slow-query sink. `File(p)` writes to that exact path and `Stderr`
114    /// writes to stderr. `Syslog` falls back to a file until its dedicated
115    /// sink is wired (ADR 0018). Unavailable files fall back to stderr.
116    pub fn for_destination(
117        dest: &crate::storage::layout::LogDestination,
118        fallback_log_dir: &std::path::Path,
119        threshold_ms: u64,
120        sample_pct: u8,
121    ) -> Arc<Self> {
122        use crate::storage::layout::LogDestination;
123        let path = match dest {
124            LogDestination::File(p) => p.clone(),
125            LogDestination::Stderr => {
126                return Self::from_writer(std::io::stderr(), threshold_ms, sample_pct);
127            }
128            LogDestination::Syslog => {
129                tracing::warn!(
130                    target: "reddb::slow",
131                    "slow-query LogDestination::Syslog requested; sink not implemented, falling back to file"
132                );
133                reddb_file::layout::legacy_slow_query_log_path(fallback_log_dir)
134            }
135        };
136        Self::open_at(path, threshold_ms, sample_pct)
137    }
138
139    fn open_at(path: PathBuf, threshold_ms: u64, sample_pct: u8) -> Arc<Self> {
140        let file = (|| -> std::io::Result<std::fs::File> {
141            if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
142                std::fs::create_dir_all(parent)?;
143            }
144            std::fs::OpenOptions::new()
145                .create(true)
146                .append(true)
147                .open(&path)
148        })();
149        match file {
150            Ok(file) => Self::from_writer(file, threshold_ms, sample_pct),
151            Err(error) => {
152                tracing::warn!(
153                    target: "reddb::slow",
154                    path = %path.display(),
155                    %error,
156                    "slow-query file sink unavailable; using stderr"
157                );
158                Self::from_writer(std::io::stderr(), threshold_ms, sample_pct)
159            }
160        }
161    }
162
163    fn from_writer(
164        sink: impl Write + Send + 'static,
165        threshold_ms: u64,
166        sample_pct: u8,
167    ) -> Arc<Self> {
168        let (writer, guard) = NonBlockingBuilder::default()
169            .buffered_lines_limit(65_536)
170            .lossy(true)
171            .finish(sink);
172
173        Arc::new(Self {
174            writer: Mutex::new(writer),
175            _guard: guard,
176            threshold_ms: AtomicU64::new(threshold_ms),
177            sample_pct: AtomicU8::new(sample_pct.min(100)),
178            above_count: AtomicU64::new(0),
179            store: std::sync::OnceLock::new(),
180        })
181    }
182
183    /// Attach the operational telemetry substrate store (ADR 0060).
184    ///
185    /// Called once at runtime startup. Above-threshold, sampled events will
186    /// be written to both the file sink and the ring store. A second call
187    /// is a no-op (first registration wins).
188    pub fn attach_store(&self, store: std::sync::Arc<super::slow_query_store::SlowQueryStore>) {
189        let _ = self.store.set(store);
190    }
191
192    /// Record a completed query. Below-threshold: single relaxed load, no
193    /// allocation. Above-threshold + sampled: emit a JSON line to the sink.
194    pub fn record(
195        &self,
196        kind: QueryKind,
197        duration_ms: u64,
198        sql_redacted: String,
199        scope: &EffectiveScope,
200    ) {
201        // Hot path: threshold gate — single relaxed atomic load.
202        if duration_ms < self.threshold_ms.load(Ordering::Relaxed) {
203            return;
204        }
205
206        // Sampling: deterministic round-robin counter.
207        // `above_count % 100 < sample_pct` gives exactly sample_pct% over
208        // long runs without any floating-point or RNG overhead.
209        let pct = u64::from(self.sample_pct.load(Ordering::Relaxed));
210        if pct < 100 {
211            let n = self.above_count.fetch_add(1, Ordering::Relaxed);
212            if (n % 100) >= pct {
213                return;
214            }
215        }
216
217        self.emit(kind, duration_ms, sql_redacted, scope);
218    }
219
220    fn emit(
221        &self,
222        kind: QueryKind,
223        duration_ms: u64,
224        sql_redacted: String,
225        scope: &EffectiveScope,
226    ) {
227        let ts_ms = crate::utils::now_unix_millis();
228        let tenant = scope.tenant.as_deref().unwrap_or("").to_string();
229        let identity = scope
230            .identity
231            .as_ref()
232            .map(|(u, _)| u.as_str())
233            .unwrap_or("")
234            .to_string();
235
236        // Durable substrate (ADR 0060): push to the ring store with hashed
237        // tenant/identity before writing the file line. Store is optional;
238        // missing store → file-only behavior (backward-compatible).
239        if let Some(store) = self.store.get() {
240            store.push(super::slow_query_store::SlowQueryEvent {
241                ts_ms,
242                kind: kind.as_str(),
243                duration_ms,
244                sql_redacted: sql_redacted.clone(),
245                tenant_hash: super::slow_query_store::hash_label(&tenant),
246                identity_hash: super::slow_query_store::hash_label(&identity),
247            });
248        }
249
250        let mut map = std::collections::BTreeMap::new();
251        map.insert(
252            "ts_ms".to_string(),
253            crate::json::Value::Number(ts_ms as f64),
254        );
255        map.insert(
256            "kind".to_string(),
257            crate::json::Value::String(kind.as_str().to_string()),
258        );
259        map.insert(
260            "duration_ms".to_string(),
261            crate::json::Value::Number(duration_ms as f64),
262        );
263        map.insert("sql".to_string(), crate::json::Value::String(sql_redacted));
264        map.insert("tenant".to_string(), crate::json::Value::String(tenant));
265        map.insert("identity".to_string(), crate::json::Value::String(identity));
266
267        let obj = crate::json::Value::Object(map);
268        if let Ok(mut line) = crate::json::to_string(&obj) {
269            line.push('\n');
270            if let Ok(mut w) = self.writer.lock() {
271                let _ = w.write_all(line.as_bytes());
272            }
273        }
274    }
275}
276
277// ---------------------------------------------------------------------------
278// Tests
279// ---------------------------------------------------------------------------
280
281#[cfg(test)]
282mod tests {
283    use std::collections::HashSet;
284    use std::sync::mpsc;
285    use std::time::Duration;
286
287    use super::*;
288    use crate::runtime::EffectiveScope;
289    use crate::storage::transaction::snapshot::Snapshot;
290
291    struct TempSlowDir(PathBuf);
292
293    impl std::ops::Deref for TempSlowDir {
294        type Target = PathBuf;
295
296        fn deref(&self) -> &PathBuf {
297            &self.0
298        }
299    }
300
301    impl AsRef<std::path::Path> for TempSlowDir {
302        fn as_ref(&self) -> &std::path::Path {
303            &self.0
304        }
305    }
306
307    impl Drop for TempSlowDir {
308        fn drop(&mut self) {
309            let _ = std::fs::remove_dir_all(&self.0);
310        }
311    }
312
313    fn tmp_dir() -> TempSlowDir {
314        let mut d = std::env::temp_dir();
315        d.push(format!(
316            "reddb-slow-{}-{}",
317            std::process::id(),
318            crate::utils::now_unix_nanos()
319        ));
320        TempSlowDir(d)
321    }
322
323    fn logger(dir: &TempSlowDir, threshold_ms: u64, sample_pct: u8) -> Arc<SlowQueryLogger> {
324        SlowQueryLogger::new(SlowQueryOpts {
325            log_dir: dir.0.clone(),
326            threshold_ms,
327            sample_pct,
328        })
329    }
330
331    fn empty_scope() -> EffectiveScope {
332        EffectiveScope {
333            captured_snapshot: None,
334            tenant: None,
335            identity: None,
336            snapshot: Snapshot {
337                xid: 0,
338                in_progress: HashSet::new(),
339            },
340            visible_collections: None,
341        }
342    }
343
344    #[test]
345    fn unavailable_log_directory_does_not_panic_or_drop_store_events() {
346        let directory = tempfile::tempdir().expect("temporary directory");
347        let blocked = directory.path().join("not-a-directory");
348        std::fs::write(&blocked, b"file").expect("unavailable directory fixture");
349        let logger = SlowQueryLogger::new(SlowQueryOpts {
350            log_dir: blocked.join("slow"),
351            threshold_ms: 0,
352            sample_pct: 100,
353        });
354        let store = super::super::slow_query_store::SlowQueryStore::new(4);
355        logger.attach_store(Arc::clone(&store));
356        logger.record(QueryKind::Select, 42, "SELECT ?".into(), &empty_scope());
357        assert_eq!(store.len(), 1);
358        assert!(blocked.is_file());
359    }
360
361    #[test]
362    fn stderr_destination_does_not_create_a_fallback_file() {
363        let directory = tempfile::tempdir().expect("temporary directory");
364        let fallback = directory.path().join("unused");
365        let logger = SlowQueryLogger::for_destination(
366            &crate::storage::layout::LogDestination::Stderr,
367            &fallback,
368            100,
369            100,
370        );
371        drop(logger);
372        assert!(
373            !fallback.exists(),
374            "stderr logging created filesystem artifacts"
375        );
376    }
377
378    fn flush(log: Arc<SlowQueryLogger>) {
379        // Dropping the final WorkerGuard drains and joins the non-blocking
380        // writer. A fixed sleep is racy under parallel CI load.
381        drop(log);
382    }
383
384    fn read_log_lines(dir: &TempSlowDir) -> Vec<crate::json::Value> {
385        let path = reddb_file::layout::legacy_slow_query_log_path(dir);
386        let body = std::fs::read_to_string(&path).unwrap_or_default();
387        body.lines()
388            .filter(|l| !l.is_empty())
389            .map(|l| crate::json::from_str::<crate::json::Value>(l).expect("valid JSON"))
390            .collect()
391    }
392
393    // -----------------------------------------------------------------------
394    // Below-threshold: zero writes, fast
395    // -----------------------------------------------------------------------
396
397    #[test]
398    fn below_threshold_no_file_writes() {
399        let dir = tmp_dir();
400        let log = logger(&dir, 1000, 100);
401        let scope = empty_scope();
402
403        for _ in 0..10_000 {
404            log.record(QueryKind::Select, 5, "SELECT 1".into(), &scope);
405        }
406
407        flush(log);
408        let lines = read_log_lines(&dir);
409        assert!(
410            lines.is_empty(),
411            "expected zero writes, got {}",
412            lines.len()
413        );
414    }
415
416    #[test]
417    fn below_threshold_fast_path_does_not_wait_for_writer() {
418        let dir = tmp_dir();
419        let log = logger(&dir, 1000, 100);
420
421        // Hold the slow sink lock. The below-threshold path must complete
422        // without consulting it; this verifies the hot-path property without
423        // a wall-clock microbenchmark that flakes under parallel CI load.
424        let writer_guard = log.writer.lock().expect("writer lock");
425        let worker_log = Arc::clone(&log);
426        let (done_tx, done_rx) = mpsc::channel();
427        let worker = std::thread::spawn(move || {
428            let scope = empty_scope();
429            for _ in 0..10_000 {
430                worker_log.record(QueryKind::Select, 5, "SELECT 1".into(), &scope);
431            }
432            let _ = done_tx.send(());
433        });
434
435        let completed_without_writer = done_rx.recv_timeout(Duration::from_secs(1)).is_ok();
436        drop(writer_guard);
437        worker.join().expect("below-threshold worker");
438        assert!(
439            completed_without_writer,
440            "below-threshold calls waited for the slow-query writer"
441        );
442    }
443
444    // -----------------------------------------------------------------------
445    // Above-threshold: structured JSON line, parseable fields
446    // -----------------------------------------------------------------------
447
448    #[test]
449    fn above_threshold_emits_json_line() {
450        let dir = tmp_dir();
451        let log = logger(&dir, 10, 100);
452        let scope = empty_scope();
453
454        log.record(QueryKind::Select, 100, "SELECT * FROM t".into(), &scope);
455        flush(log);
456
457        let lines = read_log_lines(&dir);
458        assert_eq!(lines.len(), 1, "expected 1 line");
459        let v = &lines[0];
460        assert_eq!(v.get("kind").and_then(|x| x.as_str()), Some("select"));
461        assert_eq!(v.get("duration_ms").and_then(|x| x.as_i64()), Some(100));
462        let sql = v.get("sql").and_then(|x| x.as_str());
463        assert_eq!(sql, Some("SELECT * FROM t"));
464    }
465
466    #[test]
467    fn json_line_has_all_required_fields() {
468        let dir = tmp_dir();
469        let log = logger(&dir, 0, 100);
470        let scope = empty_scope();
471
472        log.record(
473            QueryKind::Insert,
474            42,
475            "INSERT INTO t VALUES (1)".into(),
476            &scope,
477        );
478        flush(log);
479
480        let lines = read_log_lines(&dir);
481        assert_eq!(lines.len(), 1);
482        let v = &lines[0];
483        assert!(v.get("ts_ms").is_some(), "missing ts_ms");
484        assert!(v.get("kind").is_some(), "missing kind");
485        assert!(v.get("duration_ms").is_some(), "missing duration_ms");
486        assert!(v.get("sql").is_some(), "missing sql");
487        assert!(v.get("tenant").is_some(), "missing tenant");
488        assert!(v.get("identity").is_some(), "missing identity");
489    }
490
491    #[test]
492    fn all_query_kinds_serialise() {
493        let kinds = [
494            QueryKind::Select,
495            QueryKind::Insert,
496            QueryKind::Update,
497            QueryKind::Delete,
498            QueryKind::Bulk,
499            QueryKind::Aggregate,
500            QueryKind::DDL,
501            QueryKind::Internal,
502        ];
503        for k in kinds {
504            assert!(!k.as_str().is_empty());
505        }
506    }
507
508    // -----------------------------------------------------------------------
509    // Sampling property test: sample_pct=10 → ~10% (±2pp) over 10_000 calls
510    // -----------------------------------------------------------------------
511
512    #[test]
513    fn sampling_property_10pct() {
514        let dir = tmp_dir();
515        // threshold=0 so every call is above-threshold
516        let log = logger(&dir, 0, 10);
517        let scope = empty_scope();
518
519        let calls = 10_000u64;
520        for i in 0..calls {
521            log.record(QueryKind::Select, 1, format!("SELECT {i}"), &scope);
522        }
523        flush(log);
524
525        let lines = read_log_lines(&dir);
526        let emitted = lines.len() as u64;
527        // Expected: 1000 ± 200 (2pp of 10_000)
528        assert!(
529            emitted >= 800 && emitted <= 1200,
530            "sample_pct=10 over {calls} calls emitted {emitted} (expected 800..=1200)"
531        );
532    }
533
534    // -----------------------------------------------------------------------
535    // Adversarial SQL: CRLF / NUL / quote → escape-safe JSON
536    // -----------------------------------------------------------------------
537
538    #[test]
539    fn adversarial_sql_is_escape_safe() {
540        let payloads: &[(&str, &str)] = &[
541            ("crlf", "SELECT 1\r\nDROP TABLE t--"),
542            ("nul", "SELECT '\x00'"),
543            ("quote", r#"SELECT "secret" FROM t"#),
544            ("json_inject", r#"SELECT 1},"pwned":true,{"x":"#),
545            ("low_ctrl", "SELECT \x01\x02\x07\x1f"),
546            ("backslash", "SELECT 'C:\\path\\file'"),
547        ];
548
549        for (label, sql) in payloads {
550            let dir = tmp_dir();
551            let log = logger(&dir, 0, 100);
552            let scope = empty_scope();
553
554            log.record(QueryKind::Select, 1, sql.to_string(), &scope);
555            flush(log);
556
557            let path = reddb_file::layout::legacy_slow_query_log_path(&dir);
558            let body = std::fs::read_to_string(&path).unwrap_or_default();
559            let line = body
560                .lines()
561                .find(|l| !l.is_empty())
562                .unwrap_or_else(|| panic!("{label}: no line emitted"));
563
564            // Must be a single JSONL row — no embedded newline.
565            assert!(
566                !line.contains('\n'),
567                "{label}: embedded newline in JSONL row"
568            );
569
570            // Must parse as valid JSON.
571            let v: crate::json::Value = crate::json::from_str(line)
572                .unwrap_or_else(|e| panic!("{label}: not valid JSON: {e}\n{line:?}"));
573
574            // SQL must round-trip correctly.
575            let recovered = v.get("sql").and_then(|x| x.as_str()).unwrap_or("");
576            assert_eq!(recovered, *sql, "{label}: SQL round-trip mismatch");
577        }
578    }
579
580    // -----------------------------------------------------------------------
581    // Exact-threshold boundary: duration == threshold emits
582    // -----------------------------------------------------------------------
583
584    #[test]
585    fn at_threshold_boundary_emits() {
586        let dir = tmp_dir();
587        let log = logger(&dir, 50, 100);
588        let scope = empty_scope();
589
590        // Exactly at threshold: duration_ms < threshold_ms is false → emits.
591        log.record(QueryKind::Select, 50, "SELECT 1".into(), &scope);
592        flush(log);
593        let lines = read_log_lines(&dir);
594        assert_eq!(lines.len(), 1, "duration == threshold should emit");
595    }
596
597    #[test]
598    fn just_below_threshold_does_not_emit() {
599        let dir = tmp_dir();
600        let log = logger(&dir, 50, 100);
601        let scope = empty_scope();
602
603        log.record(QueryKind::Select, 49, "SELECT 1".into(), &scope);
604        flush(log);
605        let lines = read_log_lines(&dir);
606        assert!(lines.is_empty(), "duration < threshold must not emit");
607    }
608}
reddb_server/telemetry/slow_query_logger.rs

reddb_server/telemetry/
slow_query_logger.rs
