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            auth_principal: None,
337            snapshot: Snapshot {
338                xid: 0,
339                in_progress: HashSet::new(),
340            },
341            visible_collections: None,
342        }
343    }
344
345    #[test]
346    fn unavailable_log_directory_does_not_panic_or_drop_store_events() {
347        let directory = tempfile::tempdir().expect("temporary directory");
348        let blocked = directory.path().join("not-a-directory");
349        std::fs::write(&blocked, b"file").expect("unavailable directory fixture");
350        let logger = SlowQueryLogger::new(SlowQueryOpts {
351            log_dir: blocked.join("slow"),
352            threshold_ms: 0,
353            sample_pct: 100,
354        });
355        let store = super::super::slow_query_store::SlowQueryStore::new(4);
356        logger.attach_store(Arc::clone(&store));
357        logger.record(QueryKind::Select, 42, "SELECT ?".into(), &empty_scope());
358        assert_eq!(store.len(), 1);
359        assert!(blocked.is_file());
360    }
361
362    #[test]
363    fn stderr_destination_does_not_create_a_fallback_file() {
364        let directory = tempfile::tempdir().expect("temporary directory");
365        let fallback = directory.path().join("unused");
366        let logger = SlowQueryLogger::for_destination(
367            &crate::storage::layout::LogDestination::Stderr,
368            &fallback,
369            100,
370            100,
371        );
372        drop(logger);
373        assert!(
374            !fallback.exists(),
375            "stderr logging created filesystem artifacts"
376        );
377    }
378
379    fn flush(log: Arc<SlowQueryLogger>) {
380        // Dropping the final WorkerGuard drains and joins the non-blocking
381        // writer. A fixed sleep is racy under parallel CI load.
382        drop(log);
383    }
384
385    fn read_log_lines(dir: &TempSlowDir) -> Vec<crate::json::Value> {
386        let path = reddb_file::layout::legacy_slow_query_log_path(dir);
387        let body = std::fs::read_to_string(&path).unwrap_or_default();
388        body.lines()
389            .filter(|l| !l.is_empty())
390            .map(|l| crate::json::from_str::<crate::json::Value>(l).expect("valid JSON"))
391            .collect()
392    }
393
394    // -----------------------------------------------------------------------
395    // Below-threshold: zero writes, fast
396    // -----------------------------------------------------------------------
397
398    #[test]
399    fn below_threshold_no_file_writes() {
400        let dir = tmp_dir();
401        let log = logger(&dir, 1000, 100);
402        let scope = empty_scope();
403
404        for _ in 0..10_000 {
405            log.record(QueryKind::Select, 5, "SELECT 1".into(), &scope);
406        }
407
408        flush(log);
409        let lines = read_log_lines(&dir);
410        assert!(
411            lines.is_empty(),
412            "expected zero writes, got {}",
413            lines.len()
414        );
415    }
416
417    #[test]
418    fn below_threshold_fast_path_does_not_wait_for_writer() {
419        let dir = tmp_dir();
420        let log = logger(&dir, 1000, 100);
421
422        // Hold the slow sink lock. The below-threshold path must complete
423        // without consulting it; this verifies the hot-path property without
424        // a wall-clock microbenchmark that flakes under parallel CI load.
425        let writer_guard = log.writer.lock().expect("writer lock");
426        let worker_log = Arc::clone(&log);
427        let (done_tx, done_rx) = mpsc::channel();
428        let worker = std::thread::spawn(move || {
429            let scope = empty_scope();
430            for _ in 0..10_000 {
431                worker_log.record(QueryKind::Select, 5, "SELECT 1".into(), &scope);
432            }
433            let _ = done_tx.send(());
434        });
435
436        let completed_without_writer = done_rx.recv_timeout(Duration::from_secs(1)).is_ok();
437        drop(writer_guard);
438        worker.join().expect("below-threshold worker");
439        assert!(
440            completed_without_writer,
441            "below-threshold calls waited for the slow-query writer"
442        );
443    }
444
445    // -----------------------------------------------------------------------
446    // Above-threshold: structured JSON line, parseable fields
447    // -----------------------------------------------------------------------
448
449    #[test]
450    fn above_threshold_emits_json_line() {
451        let dir = tmp_dir();
452        let log = logger(&dir, 10, 100);
453        let scope = empty_scope();
454
455        log.record(QueryKind::Select, 100, "SELECT * FROM t".into(), &scope);
456        flush(log);
457
458        let lines = read_log_lines(&dir);
459        assert_eq!(lines.len(), 1, "expected 1 line");
460        let v = &lines[0];
461        assert_eq!(v.get("kind").and_then(|x| x.as_str()), Some("select"));
462        assert_eq!(v.get("duration_ms").and_then(|x| x.as_i64()), Some(100));
463        let sql = v.get("sql").and_then(|x| x.as_str());
464        assert_eq!(sql, Some("SELECT * FROM t"));
465    }
466
467    #[test]
468    fn json_line_has_all_required_fields() {
469        let dir = tmp_dir();
470        let log = logger(&dir, 0, 100);
471        let scope = empty_scope();
472
473        log.record(
474            QueryKind::Insert,
475            42,
476            "INSERT INTO t VALUES (1)".into(),
477            &scope,
478        );
479        flush(log);
480
481        let lines = read_log_lines(&dir);
482        assert_eq!(lines.len(), 1);
483        let v = &lines[0];
484        assert!(v.get("ts_ms").is_some(), "missing ts_ms");
485        assert!(v.get("kind").is_some(), "missing kind");
486        assert!(v.get("duration_ms").is_some(), "missing duration_ms");
487        assert!(v.get("sql").is_some(), "missing sql");
488        assert!(v.get("tenant").is_some(), "missing tenant");
489        assert!(v.get("identity").is_some(), "missing identity");
490    }
491
492    #[test]
493    fn all_query_kinds_serialise() {
494        let kinds = [
495            QueryKind::Select,
496            QueryKind::Insert,
497            QueryKind::Update,
498            QueryKind::Delete,
499            QueryKind::Bulk,
500            QueryKind::Aggregate,
501            QueryKind::DDL,
502            QueryKind::Internal,
503        ];
504        for k in kinds {
505            assert!(!k.as_str().is_empty());
506        }
507    }
508
509    // -----------------------------------------------------------------------
510    // Sampling property test: sample_pct=10 → ~10% (±2pp) over 10_000 calls
511    // -----------------------------------------------------------------------
512
513    #[test]
514    fn sampling_property_10pct() {
515        let dir = tmp_dir();
516        // threshold=0 so every call is above-threshold
517        let log = logger(&dir, 0, 10);
518        let scope = empty_scope();
519
520        let calls = 10_000u64;
521        for i in 0..calls {
522            log.record(QueryKind::Select, 1, format!("SELECT {i}"), &scope);
523        }
524        flush(log);
525
526        let lines = read_log_lines(&dir);
527        let emitted = lines.len() as u64;
528        // Expected: 1000 ± 200 (2pp of 10_000)
529        assert!(
530            emitted >= 800 && emitted <= 1200,
531            "sample_pct=10 over {calls} calls emitted {emitted} (expected 800..=1200)"
532        );
533    }
534
535    // -----------------------------------------------------------------------
536    // Adversarial SQL: CRLF / NUL / quote → escape-safe JSON
537    // -----------------------------------------------------------------------
538
539    #[test]
540    fn adversarial_sql_is_escape_safe() {
541        let payloads: &[(&str, &str)] = &[
542            ("crlf", "SELECT 1\r\nDROP TABLE t--"),
543            ("nul", "SELECT '\x00'"),
544            ("quote", r#"SELECT "secret" FROM t"#),
545            ("json_inject", r#"SELECT 1},"pwned":true,{"x":"#),
546            ("low_ctrl", "SELECT \x01\x02\x07\x1f"),
547            ("backslash", "SELECT 'C:\\path\\file'"),
548        ];
549
550        for (label, sql) in payloads {
551            let dir = tmp_dir();
552            let log = logger(&dir, 0, 100);
553            let scope = empty_scope();
554
555            log.record(QueryKind::Select, 1, sql.to_string(), &scope);
556            flush(log);
557
558            let path = reddb_file::layout::legacy_slow_query_log_path(&dir);
559            let body = std::fs::read_to_string(&path).unwrap_or_default();
560            let line = body
561                .lines()
562                .find(|l| !l.is_empty())
563                .unwrap_or_else(|| panic!("{label}: no line emitted"));
564
565            // Must be a single JSONL row — no embedded newline.
566            assert!(
567                !line.contains('\n'),
568                "{label}: embedded newline in JSONL row"
569            );
570
571            // Must parse as valid JSON.
572            let v: crate::json::Value = crate::json::from_str(line)
573                .unwrap_or_else(|e| panic!("{label}: not valid JSON: {e}\n{line:?}"));
574
575            // SQL must round-trip correctly.
576            let recovered = v.get("sql").and_then(|x| x.as_str()).unwrap_or("");
577            assert_eq!(recovered, *sql, "{label}: SQL round-trip mismatch");
578        }
579    }
580
581    // -----------------------------------------------------------------------
582    // Exact-threshold boundary: duration == threshold emits
583    // -----------------------------------------------------------------------
584
585    #[test]
586    fn at_threshold_boundary_emits() {
587        let dir = tmp_dir();
588        let log = logger(&dir, 50, 100);
589        let scope = empty_scope();
590
591        // Exactly at threshold: duration_ms < threshold_ms is false → emits.
592        log.record(QueryKind::Select, 50, "SELECT 1".into(), &scope);
593        flush(log);
594        let lines = read_log_lines(&dir);
595        assert_eq!(lines.len(), 1, "duration == threshold should emit");
596    }
597
598    #[test]
599    fn just_below_threshold_does_not_emit() {
600        let dir = tmp_dir();
601        let log = logger(&dir, 50, 100);
602        let scope = empty_scope();
603
604        log.record(QueryKind::Select, 49, "SELECT 1".into(), &scope);
605        flush(log);
606        let lines = read_log_lines(&dir);
607        assert!(lines.is_empty(), "duration < threshold must not emit");
608    }
609}
reddb_server/telemetry/slow_query_logger.rs

reddb_server/telemetry/
slow_query_logger.rs
