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::time::Instant;
285
286    use super::*;
287    use crate::runtime::EffectiveScope;
288    use crate::storage::transaction::snapshot::Snapshot;
289
290    struct TempSlowDir(PathBuf);
291
292    impl std::ops::Deref for TempSlowDir {
293        type Target = PathBuf;
294
295        fn deref(&self) -> &PathBuf {
296            &self.0
297        }
298    }
299
300    impl AsRef<std::path::Path> for TempSlowDir {
301        fn as_ref(&self) -> &std::path::Path {
302            &self.0
303        }
304    }
305
306    impl Drop for TempSlowDir {
307        fn drop(&mut self) {
308            let _ = std::fs::remove_dir_all(&self.0);
309        }
310    }
311
312    fn tmp_dir() -> TempSlowDir {
313        let mut d = std::env::temp_dir();
314        d.push(format!(
315            "reddb-slow-{}-{}",
316            std::process::id(),
317            crate::utils::now_unix_nanos()
318        ));
319        TempSlowDir(d)
320    }
321
322    fn logger(dir: &TempSlowDir, threshold_ms: u64, sample_pct: u8) -> Arc<SlowQueryLogger> {
323        SlowQueryLogger::new(SlowQueryOpts {
324            log_dir: dir.0.clone(),
325            threshold_ms,
326            sample_pct,
327        })
328    }
329
330    fn empty_scope() -> EffectiveScope {
331        EffectiveScope {
332            captured_snapshot: None,
333            tenant: None,
334            identity: None,
335            snapshot: Snapshot {
336                xid: 0,
337                in_progress: HashSet::new(),
338            },
339            visible_collections: None,
340        }
341    }
342
343    #[test]
344    fn unavailable_log_directory_does_not_panic_or_drop_store_events() {
345        let directory = tempfile::tempdir().expect("temporary directory");
346        let blocked = directory.path().join("not-a-directory");
347        std::fs::write(&blocked, b"file").expect("unavailable directory fixture");
348        let logger = SlowQueryLogger::new(SlowQueryOpts {
349            log_dir: blocked.join("slow"),
350            threshold_ms: 0,
351            sample_pct: 100,
352        });
353        let store = super::super::slow_query_store::SlowQueryStore::new(4);
354        logger.attach_store(Arc::clone(&store));
355        logger.record(QueryKind::Select, 42, "SELECT ?".into(), &empty_scope());
356        assert_eq!(store.len(), 1);
357        assert!(blocked.is_file());
358    }
359
360    #[test]
361    fn stderr_destination_does_not_create_a_fallback_file() {
362        let directory = tempfile::tempdir().expect("temporary directory");
363        let fallback = directory.path().join("unused");
364        let logger = SlowQueryLogger::for_destination(
365            &crate::storage::layout::LogDestination::Stderr,
366            &fallback,
367            100,
368            100,
369        );
370        drop(logger);
371        assert!(
372            !fallback.exists(),
373            "stderr logging created filesystem artifacts"
374        );
375    }
376
377    fn flush(_log: &Arc<SlowQueryLogger>) {
378        // NonBlocking sends bytes to a background thread; sleep gives the
379        // worker time to drain the channel and write to disk.
380        std::thread::sleep(std::time::Duration::from_millis(50));
381    }
382
383    fn read_log_lines(dir: &TempSlowDir) -> Vec<crate::json::Value> {
384        let path = reddb_file::layout::legacy_slow_query_log_path(dir);
385        let body = std::fs::read_to_string(&path).unwrap_or_default();
386        body.lines()
387            .filter(|l| !l.is_empty())
388            .map(|l| crate::json::from_str::<crate::json::Value>(l).expect("valid JSON"))
389            .collect()
390    }
391
392    // -----------------------------------------------------------------------
393    // Below-threshold: zero writes, fast
394    // -----------------------------------------------------------------------
395
396    #[test]
397    fn below_threshold_no_file_writes() {
398        let dir = tmp_dir();
399        let log = logger(&dir, 1000, 100);
400        let scope = empty_scope();
401
402        for _ in 0..10_000 {
403            log.record(QueryKind::Select, 5, "SELECT 1".into(), &scope);
404        }
405
406        flush(&log);
407        let lines = read_log_lines(&dir);
408        assert!(
409            lines.is_empty(),
410            "expected zero writes, got {}",
411            lines.len()
412        );
413    }
414
415    #[test]
416    fn below_threshold_wall_time_under_10ms() {
417        let dir = tmp_dir();
418        let log = logger(&dir, 1000, 100);
419        let scope = empty_scope();
420
421        let start = Instant::now();
422        for _ in 0..10_000 {
423            log.record(QueryKind::Select, 5, "SELECT 1".into(), &scope);
424        }
425        let elapsed = start.elapsed();
426        assert!(
427            elapsed.as_millis() < 10,
428            "10k below-threshold calls took {}ms (>10ms budget)",
429            elapsed.as_millis()
430        );
431    }
432
433    // -----------------------------------------------------------------------
434    // Above-threshold: structured JSON line, parseable fields
435    // -----------------------------------------------------------------------
436
437    #[test]
438    fn above_threshold_emits_json_line() {
439        let dir = tmp_dir();
440        let log = logger(&dir, 10, 100);
441        let scope = empty_scope();
442
443        log.record(QueryKind::Select, 100, "SELECT * FROM t".into(), &scope);
444        flush(&log);
445
446        let lines = read_log_lines(&dir);
447        assert_eq!(lines.len(), 1, "expected 1 line");
448        let v = &lines[0];
449        assert_eq!(v.get("kind").and_then(|x| x.as_str()), Some("select"));
450        assert_eq!(v.get("duration_ms").and_then(|x| x.as_i64()), Some(100));
451        let sql = v.get("sql").and_then(|x| x.as_str());
452        assert_eq!(sql, Some("SELECT * FROM t"));
453    }
454
455    #[test]
456    fn json_line_has_all_required_fields() {
457        let dir = tmp_dir();
458        let log = logger(&dir, 0, 100);
459        let scope = empty_scope();
460
461        log.record(
462            QueryKind::Insert,
463            42,
464            "INSERT INTO t VALUES (1)".into(),
465            &scope,
466        );
467        flush(&log);
468
469        let lines = read_log_lines(&dir);
470        assert_eq!(lines.len(), 1);
471        let v = &lines[0];
472        assert!(v.get("ts_ms").is_some(), "missing ts_ms");
473        assert!(v.get("kind").is_some(), "missing kind");
474        assert!(v.get("duration_ms").is_some(), "missing duration_ms");
475        assert!(v.get("sql").is_some(), "missing sql");
476        assert!(v.get("tenant").is_some(), "missing tenant");
477        assert!(v.get("identity").is_some(), "missing identity");
478    }
479
480    #[test]
481    fn all_query_kinds_serialise() {
482        let kinds = [
483            QueryKind::Select,
484            QueryKind::Insert,
485            QueryKind::Update,
486            QueryKind::Delete,
487            QueryKind::Bulk,
488            QueryKind::Aggregate,
489            QueryKind::DDL,
490            QueryKind::Internal,
491        ];
492        for k in kinds {
493            assert!(!k.as_str().is_empty());
494        }
495    }
496
497    // -----------------------------------------------------------------------
498    // Sampling property test: sample_pct=10 → ~10% (±2pp) over 10_000 calls
499    // -----------------------------------------------------------------------
500
501    #[test]
502    fn sampling_property_10pct() {
503        let dir = tmp_dir();
504        // threshold=0 so every call is above-threshold
505        let log = logger(&dir, 0, 10);
506        let scope = empty_scope();
507
508        let calls = 10_000u64;
509        for i in 0..calls {
510            log.record(QueryKind::Select, 1, format!("SELECT {i}"), &scope);
511        }
512        flush(&log);
513
514        let lines = read_log_lines(&dir);
515        let emitted = lines.len() as u64;
516        // Expected: 1000 ± 200 (2pp of 10_000)
517        assert!(
518            emitted >= 800 && emitted <= 1200,
519            "sample_pct=10 over {calls} calls emitted {emitted} (expected 800..=1200)"
520        );
521    }
522
523    // -----------------------------------------------------------------------
524    // Adversarial SQL: CRLF / NUL / quote → escape-safe JSON
525    // -----------------------------------------------------------------------
526
527    #[test]
528    fn adversarial_sql_is_escape_safe() {
529        let payloads: &[(&str, &str)] = &[
530            ("crlf", "SELECT 1\r\nDROP TABLE t--"),
531            ("nul", "SELECT '\x00'"),
532            ("quote", r#"SELECT "secret" FROM t"#),
533            ("json_inject", r#"SELECT 1},"pwned":true,{"x":"#),
534            ("low_ctrl", "SELECT \x01\x02\x07\x1f"),
535            ("backslash", "SELECT 'C:\\path\\file'"),
536        ];
537
538        for (label, sql) in payloads {
539            let dir = tmp_dir();
540            let log = logger(&dir, 0, 100);
541            let scope = empty_scope();
542
543            log.record(QueryKind::Select, 1, sql.to_string(), &scope);
544            flush(&log);
545
546            let path = reddb_file::layout::legacy_slow_query_log_path(&dir);
547            let body = std::fs::read_to_string(&path).unwrap_or_default();
548            let line = body
549                .lines()
550                .find(|l| !l.is_empty())
551                .unwrap_or_else(|| panic!("{label}: no line emitted"));
552
553            // Must be a single JSONL row — no embedded newline.
554            assert!(
555                !line.contains('\n'),
556                "{label}: embedded newline in JSONL row"
557            );
558
559            // Must parse as valid JSON.
560            let v: crate::json::Value = crate::json::from_str(line)
561                .unwrap_or_else(|e| panic!("{label}: not valid JSON: {e}\n{line:?}"));
562
563            // SQL must round-trip correctly.
564            let recovered = v.get("sql").and_then(|x| x.as_str()).unwrap_or("");
565            assert_eq!(recovered, *sql, "{label}: SQL round-trip mismatch");
566        }
567    }
568
569    // -----------------------------------------------------------------------
570    // Exact-threshold boundary: duration == threshold emits
571    // -----------------------------------------------------------------------
572
573    #[test]
574    fn at_threshold_boundary_emits() {
575        let dir = tmp_dir();
576        let log = logger(&dir, 50, 100);
577        let scope = empty_scope();
578
579        // Exactly at threshold: duration_ms < threshold_ms is false → emits.
580        log.record(QueryKind::Select, 50, "SELECT 1".into(), &scope);
581        flush(&log);
582        let lines = read_log_lines(&dir);
583        assert_eq!(lines.len(), 1, "duration == threshold should emit");
584    }
585
586    #[test]
587    fn just_below_threshold_does_not_emit() {
588        let dir = tmp_dir();
589        let log = logger(&dir, 50, 100);
590        let scope = empty_scope();
591
592        log.record(QueryKind::Select, 49, "SELECT 1".into(), &scope);
593        flush(&log);
594        let lines = read_log_lines(&dir);
595        assert!(lines.is_empty(), "duration < threshold must not emit");
596    }
597}
reddb_server/telemetry/slow_query_logger.rs

reddb_server/telemetry/
slow_query_logger.rs
