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; `Stderr` and
114    /// `Syslog` fall back to `<fallback_log_dir>/red-slow.log` until the
115    /// dedicated sinks are wired (ADR 0018).
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                reddb_file::layout::legacy_slow_query_log_path(fallback_log_dir)
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        if let Some(parent) = path.parent() {
141            let _ = std::fs::create_dir_all(parent);
142        }
143        let file = std::fs::OpenOptions::new()
144            .create(true)
145            .append(true)
146            .open(&path)
147            .unwrap_or_else(|e| panic!("SlowQueryLogger: cannot open {}: {e}", path.display()));
148
149        let (writer, guard) = NonBlockingBuilder::default()
150            .buffered_lines_limit(65_536)
151            .lossy(true)
152            .finish(file);
153
154        Arc::new(Self {
155            writer: Mutex::new(writer),
156            _guard: guard,
157            threshold_ms: AtomicU64::new(threshold_ms),
158            sample_pct: AtomicU8::new(sample_pct.min(100)),
159            above_count: AtomicU64::new(0),
160            store: std::sync::OnceLock::new(),
161        })
162    }
163
164    /// Attach the operational telemetry substrate store (ADR 0060).
165    ///
166    /// Called once at runtime startup. Above-threshold, sampled events will
167    /// be written to both the file sink and the ring store. A second call
168    /// is a no-op (first registration wins).
169    pub fn attach_store(&self, store: std::sync::Arc<super::slow_query_store::SlowQueryStore>) {
170        let _ = self.store.set(store);
171    }
172
173    /// Record a completed query. Below-threshold: single relaxed load, no
174    /// allocation. Above-threshold + sampled: emit a JSON line to the sink.
175    pub fn record(
176        &self,
177        kind: QueryKind,
178        duration_ms: u64,
179        sql_redacted: String,
180        scope: &EffectiveScope,
181    ) {
182        // Hot path: threshold gate — single relaxed atomic load.
183        if duration_ms < self.threshold_ms.load(Ordering::Relaxed) {
184            return;
185        }
186
187        // Sampling: deterministic round-robin counter.
188        // `above_count % 100 < sample_pct` gives exactly sample_pct% over
189        // long runs without any floating-point or RNG overhead.
190        let pct = u64::from(self.sample_pct.load(Ordering::Relaxed));
191        if pct < 100 {
192            let n = self.above_count.fetch_add(1, Ordering::Relaxed);
193            if (n % 100) >= pct {
194                return;
195            }
196        }
197
198        self.emit(kind, duration_ms, sql_redacted, scope);
199    }
200
201    fn emit(
202        &self,
203        kind: QueryKind,
204        duration_ms: u64,
205        sql_redacted: String,
206        scope: &EffectiveScope,
207    ) {
208        let ts_ms = crate::utils::now_unix_millis();
209        let tenant = scope.tenant.as_deref().unwrap_or("").to_string();
210        let identity = scope
211            .identity
212            .as_ref()
213            .map(|(u, _)| u.as_str())
214            .unwrap_or("")
215            .to_string();
216
217        // Durable substrate (ADR 0060): push to the ring store with hashed
218        // tenant/identity before writing the file line. Store is optional;
219        // missing store → file-only behavior (backward-compatible).
220        if let Some(store) = self.store.get() {
221            store.push(super::slow_query_store::SlowQueryEvent {
222                ts_ms,
223                kind: kind.as_str(),
224                duration_ms,
225                sql_redacted: sql_redacted.clone(),
226                tenant_hash: super::slow_query_store::hash_label(&tenant),
227                identity_hash: super::slow_query_store::hash_label(&identity),
228            });
229        }
230
231        let mut map = std::collections::BTreeMap::new();
232        map.insert(
233            "ts_ms".to_string(),
234            crate::json::Value::Number(ts_ms as f64),
235        );
236        map.insert(
237            "kind".to_string(),
238            crate::json::Value::String(kind.as_str().to_string()),
239        );
240        map.insert(
241            "duration_ms".to_string(),
242            crate::json::Value::Number(duration_ms as f64),
243        );
244        map.insert("sql".to_string(), crate::json::Value::String(sql_redacted));
245        map.insert("tenant".to_string(), crate::json::Value::String(tenant));
246        map.insert("identity".to_string(), crate::json::Value::String(identity));
247
248        let obj = crate::json::Value::Object(map);
249        if let Ok(mut line) = crate::json::to_string(&obj) {
250            line.push('\n');
251            if let Ok(mut w) = self.writer.lock() {
252                let _ = w.write_all(line.as_bytes());
253            }
254        }
255    }
256}
257
258// ---------------------------------------------------------------------------
259// Tests
260// ---------------------------------------------------------------------------
261
262#[cfg(test)]
263mod tests {
264    use std::collections::HashSet;
265    use std::time::Instant;
266
267    use super::*;
268    use crate::runtime::EffectiveScope;
269    use crate::storage::transaction::snapshot::Snapshot;
270
271    struct TempSlowDir(PathBuf);
272
273    impl std::ops::Deref for TempSlowDir {
274        type Target = PathBuf;
275
276        fn deref(&self) -> &PathBuf {
277            &self.0
278        }
279    }
280
281    impl AsRef<std::path::Path> for TempSlowDir {
282        fn as_ref(&self) -> &std::path::Path {
283            &self.0
284        }
285    }
286
287    impl Drop for TempSlowDir {
288        fn drop(&mut self) {
289            let _ = std::fs::remove_dir_all(&self.0);
290        }
291    }
292
293    fn tmp_dir() -> TempSlowDir {
294        let mut d = std::env::temp_dir();
295        d.push(format!(
296            "reddb-slow-{}-{}",
297            std::process::id(),
298            crate::utils::now_unix_nanos()
299        ));
300        TempSlowDir(d)
301    }
302
303    fn logger(dir: &TempSlowDir, threshold_ms: u64, sample_pct: u8) -> Arc<SlowQueryLogger> {
304        SlowQueryLogger::new(SlowQueryOpts {
305            log_dir: dir.0.clone(),
306            threshold_ms,
307            sample_pct,
308        })
309    }
310
311    fn empty_scope() -> EffectiveScope {
312        EffectiveScope {
313            tenant: None,
314            identity: None,
315            snapshot: Snapshot {
316                xid: 0,
317                in_progress: HashSet::new(),
318            },
319            visible_collections: None,
320        }
321    }
322
323    fn flush(_log: &Arc<SlowQueryLogger>) {
324        // NonBlocking sends bytes to a background thread; sleep gives the
325        // worker time to drain the channel and write to disk.
326        std::thread::sleep(std::time::Duration::from_millis(50));
327    }
328
329    fn read_log_lines(dir: &TempSlowDir) -> Vec<crate::json::Value> {
330        let path = reddb_file::layout::legacy_slow_query_log_path(dir);
331        let body = std::fs::read_to_string(&path).unwrap_or_default();
332        body.lines()
333            .filter(|l| !l.is_empty())
334            .map(|l| crate::json::from_str::<crate::json::Value>(l).expect("valid JSON"))
335            .collect()
336    }
337
338    // -----------------------------------------------------------------------
339    // Below-threshold: zero writes, fast
340    // -----------------------------------------------------------------------
341
342    #[test]
343    fn below_threshold_no_file_writes() {
344        let dir = tmp_dir();
345        let log = logger(&dir, 1000, 100);
346        let scope = empty_scope();
347
348        for _ in 0..10_000 {
349            log.record(QueryKind::Select, 5, "SELECT 1".into(), &scope);
350        }
351
352        flush(&log);
353        let lines = read_log_lines(&dir);
354        assert!(
355            lines.is_empty(),
356            "expected zero writes, got {}",
357            lines.len()
358        );
359    }
360
361    #[test]
362    fn below_threshold_wall_time_under_10ms() {
363        let dir = tmp_dir();
364        let log = logger(&dir, 1000, 100);
365        let scope = empty_scope();
366
367        let start = Instant::now();
368        for _ in 0..10_000 {
369            log.record(QueryKind::Select, 5, "SELECT 1".into(), &scope);
370        }
371        let elapsed = start.elapsed();
372        assert!(
373            elapsed.as_millis() < 10,
374            "10k below-threshold calls took {}ms (>10ms budget)",
375            elapsed.as_millis()
376        );
377    }
378
379    // -----------------------------------------------------------------------
380    // Above-threshold: structured JSON line, parseable fields
381    // -----------------------------------------------------------------------
382
383    #[test]
384    fn above_threshold_emits_json_line() {
385        let dir = tmp_dir();
386        let log = logger(&dir, 10, 100);
387        let scope = empty_scope();
388
389        log.record(QueryKind::Select, 100, "SELECT * FROM t".into(), &scope);
390        flush(&log);
391
392        let lines = read_log_lines(&dir);
393        assert_eq!(lines.len(), 1, "expected 1 line");
394        let v = &lines[0];
395        assert_eq!(v.get("kind").and_then(|x| x.as_str()), Some("select"));
396        assert_eq!(v.get("duration_ms").and_then(|x| x.as_i64()), Some(100));
397        let sql = v.get("sql").and_then(|x| x.as_str());
398        assert_eq!(sql, Some("SELECT * FROM t"));
399    }
400
401    #[test]
402    fn json_line_has_all_required_fields() {
403        let dir = tmp_dir();
404        let log = logger(&dir, 0, 100);
405        let scope = empty_scope();
406
407        log.record(
408            QueryKind::Insert,
409            42,
410            "INSERT INTO t VALUES (1)".into(),
411            &scope,
412        );
413        flush(&log);
414
415        let lines = read_log_lines(&dir);
416        assert_eq!(lines.len(), 1);
417        let v = &lines[0];
418        assert!(v.get("ts_ms").is_some(), "missing ts_ms");
419        assert!(v.get("kind").is_some(), "missing kind");
420        assert!(v.get("duration_ms").is_some(), "missing duration_ms");
421        assert!(v.get("sql").is_some(), "missing sql");
422        assert!(v.get("tenant").is_some(), "missing tenant");
423        assert!(v.get("identity").is_some(), "missing identity");
424    }
425
426    #[test]
427    fn all_query_kinds_serialise() {
428        let kinds = [
429            QueryKind::Select,
430            QueryKind::Insert,
431            QueryKind::Update,
432            QueryKind::Delete,
433            QueryKind::Bulk,
434            QueryKind::Aggregate,
435            QueryKind::DDL,
436            QueryKind::Internal,
437        ];
438        for k in kinds {
439            assert!(!k.as_str().is_empty());
440        }
441    }
442
443    // -----------------------------------------------------------------------
444    // Sampling property test: sample_pct=10 → ~10% (±2pp) over 10_000 calls
445    // -----------------------------------------------------------------------
446
447    #[test]
448    fn sampling_property_10pct() {
449        let dir = tmp_dir();
450        // threshold=0 so every call is above-threshold
451        let log = logger(&dir, 0, 10);
452        let scope = empty_scope();
453
454        let calls = 10_000u64;
455        for i in 0..calls {
456            log.record(QueryKind::Select, 1, format!("SELECT {i}"), &scope);
457        }
458        flush(&log);
459
460        let lines = read_log_lines(&dir);
461        let emitted = lines.len() as u64;
462        // Expected: 1000 ± 200 (2pp of 10_000)
463        assert!(
464            emitted >= 800 && emitted <= 1200,
465            "sample_pct=10 over {calls} calls emitted {emitted} (expected 800..=1200)"
466        );
467    }
468
469    // -----------------------------------------------------------------------
470    // Adversarial SQL: CRLF / NUL / quote → escape-safe JSON
471    // -----------------------------------------------------------------------
472
473    #[test]
474    fn adversarial_sql_is_escape_safe() {
475        let payloads: &[(&str, &str)] = &[
476            ("crlf", "SELECT 1\r\nDROP TABLE t--"),
477            ("nul", "SELECT '\x00'"),
478            ("quote", r#"SELECT "secret" FROM t"#),
479            ("json_inject", r#"SELECT 1},"pwned":true,{"x":"#),
480            ("low_ctrl", "SELECT \x01\x02\x07\x1f"),
481            ("backslash", "SELECT 'C:\\path\\file'"),
482        ];
483
484        for (label, sql) in payloads {
485            let dir = tmp_dir();
486            let log = logger(&dir, 0, 100);
487            let scope = empty_scope();
488
489            log.record(QueryKind::Select, 1, sql.to_string(), &scope);
490            flush(&log);
491
492            let path = reddb_file::layout::legacy_slow_query_log_path(&dir);
493            let body = std::fs::read_to_string(&path).unwrap_or_default();
494            let line = body
495                .lines()
496                .find(|l| !l.is_empty())
497                .unwrap_or_else(|| panic!("{label}: no line emitted"));
498
499            // Must be a single JSONL row — no embedded newline.
500            assert!(
501                !line.contains('\n'),
502                "{label}: embedded newline in JSONL row"
503            );
504
505            // Must parse as valid JSON.
506            let v: crate::json::Value = crate::json::from_str(line)
507                .unwrap_or_else(|e| panic!("{label}: not valid JSON: {e}\n{line:?}"));
508
509            // SQL must round-trip correctly.
510            let recovered = v.get("sql").and_then(|x| x.as_str()).unwrap_or("");
511            assert_eq!(recovered, *sql, "{label}: SQL round-trip mismatch");
512        }
513    }
514
515    // -----------------------------------------------------------------------
516    // Exact-threshold boundary: duration == threshold emits
517    // -----------------------------------------------------------------------
518
519    #[test]
520    fn at_threshold_boundary_emits() {
521        let dir = tmp_dir();
522        let log = logger(&dir, 50, 100);
523        let scope = empty_scope();
524
525        // Exactly at threshold: duration_ms < threshold_ms is false → emits.
526        log.record(QueryKind::Select, 50, "SELECT 1".into(), &scope);
527        flush(&log);
528        let lines = read_log_lines(&dir);
529        assert_eq!(lines.len(), 1, "duration == threshold should emit");
530    }
531
532    #[test]
533    fn just_below_threshold_does_not_emit() {
534        let dir = tmp_dir();
535        let log = logger(&dir, 50, 100);
536        let scope = empty_scope();
537
538        log.record(QueryKind::Select, 49, "SELECT 1".into(), &scope);
539        flush(&log);
540        let lines = read_log_lines(&dir);
541        assert!(lines.is_empty(), "duration < threshold must not emit");
542    }
543}
reddb_server/telemetry/slow_query_logger.rs

reddb_server/telemetry/
slow_query_logger.rs
