From 9519d2366948559b37fd6a92d0f802c373787716 Mon Sep 17 00:00:00 2001 From: "snake.tech" <9079973+snaketech-tu@users.noreply.github.com> Date: Tue, 10 Nov 2026 01:37:29 +0000 Subject: [PATCH 1/3] Split the log store into schema, query, retention and export Only moves code: store.rs becomes store/mod.rs with four modules beside it. Nothing changes in what the store does. --- crates/stackline-core/src/store/export.rs | 58 ++ .../src/{store.rs => store/mod.rs} | 894 +----------------- crates/stackline-core/src/store/query.rs | 390 ++++++++ crates/stackline-core/src/store/retention.rs | 342 +++++++ crates/stackline-core/src/store/schema.rs | 104 ++ 5 files changed, 905 insertions(+), 883 deletions(-) create mode 100644 crates/stackline-core/src/store/export.rs rename crates/stackline-core/src/{store.rs => store/mod.rs} (63%) create mode 100644 crates/stackline-core/src/store/query.rs create mode 100644 crates/stackline-core/src/store/retention.rs create mode 100644 crates/stackline-core/src/store/schema.rs diff --git a/crates/stackline-core/src/store/export.rs b/crates/stackline-core/src/store/export.rs new file mode 100644 index 0000000..9a3af9f --- /dev/null +++ b/crates/stackline-core/src/store/export.rs @@ -0,0 +1,58 @@ +//! Writing stored lines out as text, JSON lines or CSV. + +use super::*; + +pub(super) fn export(db: &Connection, conn: &str, f: &LogFilter, format: ExportFormat, out: &mut dyn Write) -> Result { + let (w, p) = f.where_clause(conn, true); + let mut st = db.prepare(&format!( + "SELECT id, ts, stack, service, replica, node, source, stream, level, exc, lines, fp, text, raw + FROM entries WHERE {w} ORDER BY ts ASC, id ASC" + ))?; + let mut rows = st.query(params_from_iter(p.iter()))?; + let mut n = 0; + if format == ExportFormat::Csv { + writeln!(out, "time,level,stack,service,replica,node,stream,message")?; + } + while let Some(r) = rows.next()? { + let e = row_to_entry(r)?; + let time = chrono::DateTime::from_timestamp_micros(e.ts) + .map(|d| d.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)) + .unwrap_or_default(); + match format { + ExportFormat::Log => { + writeln!(out, "{time} {:<5} {}.{} {}", e.level.name(), e.service, e.replica, e.text)?; + } + ExportFormat::Jsonl => { + let v = serde_json::json!({ + "time": time, "level": e.level.name(), "stack": e.stack, "service": e.service, + "replica": e.replica, "node": e.node, "stream": e.stream, "message": e.text, + "raw": e.raw.as_deref().and_then(|r| serde_json::from_str::(r).ok()), + }); + writeln!(out, "{v}")?; + } + ExportFormat::Csv => { + writeln!( + out, + "{},{},{},{},{},{},{},{}", + time, + e.level.name(), + csv(&e.stack), + csv(&e.service), + csv(&e.replica), + csv(&e.node), + e.stream, + csv(&e.text) + )?; + } + } + n += 1; + } + Ok(n) +} + +pub(super) fn csv(s: &str) -> String { + // A cell starting like a formula is run by spreadsheet apps when the file + // is opened; a log line must never do that, so it starts with ' instead. + let s = if s.starts_with(['=', '+', '-', '@', '\t', '\r']) { format!("'{s}") } else { s.to_string() }; + if s.contains([',', '"', '\n', '\r']) { format!("\"{}\"", s.replace('"', "\"\"")) } else { s } +} diff --git a/crates/stackline-core/src/store.rs b/crates/stackline-core/src/store/mod.rs similarity index 63% rename from crates/stackline-core/src/store.rs rename to crates/stackline-core/src/store/mod.rs index 1a8ff44..a2977a3 100644 --- a/crates/stackline-core/src/store.rs +++ b/crates/stackline-core/src/store/mod.rs @@ -18,6 +18,16 @@ use crate::events::{Event, Kind}; use crate::logs::{Level, LogEntry}; use crate::model::LogSource; +mod export; +mod query; +mod retention; +mod schema; + +use export::*; +use query::*; +use retention::*; +use schema::*; + pub struct Store { /// The one connection that writes. Its lock is held only briefly, so new /// lines are never stuck behind a slow read. @@ -41,112 +51,10 @@ pub struct Store { /// Idle read connections kept open; more are opened while busy. const IDLE_READERS: usize = 4; -/// Keeps the search index in step with new lines. Their per-minute counts -/// are added by [`Store::insert`], once per minute instead of once per line. +/// Remembers how far a source has been read. const SET_CURSOR: &str = "INSERT INTO cursors (conn, source, ts) VALUES (?1, ?2, ?3) ON CONFLICT (conn, source) DO UPDATE SET ts = excluded.ts"; -const INSERT_TRIGGER: &str = "CREATE TRIGGER entries_ai AFTER INSERT ON entries BEGIN - INSERT INTO entries_fts (rowid, text) VALUES (new.id, new.text); -END;"; - -/// Adds to the counts of a minute: for the service, and for the whole connection (service ''). -const ADD_MINUTE: &str = - "INSERT INTO minutes (conn, minute, service, stack, debug, info, warn, error, debug_exc, info_exc, warn_exc, error_exc) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) - ON CONFLICT (conn, minute, service) DO UPDATE SET - debug = debug + excluded.debug, info = info + excluded.info, warn = warn + excluded.warn, error = error + excluded.error, - debug_exc = debug_exc + excluded.debug_exc, info_exc = info_exc + excluded.info_exc, - warn_exc = warn_exc + excluded.warn_exc, error_exc = error_exc + excluded.error_exc"; - -/// Keeps the search index and the per-minute counts in step with deleted -/// lines. Deleting every line drops it for a moment, so the table can be -/// emptied at once. -const DELETE_TRIGGER: &str = "CREATE TRIGGER entries_ad AFTER DELETE ON entries BEGIN - INSERT INTO entries_fts (entries_fts, rowid, text) VALUES ('delete', old.id, old.text); - UPDATE minutes SET - debug = debug - (old.level = 1), info = info - (old.level NOT IN (1, 3, 4)), - warn = warn - (old.level = 3), error = error - (old.level = 4), - debug_exc = debug_exc - (old.exc != 0 AND old.level = 1), info_exc = info_exc - (old.exc != 0 AND old.level NOT IN (1, 3, 4)), - warn_exc = warn_exc - (old.exc != 0 AND old.level = 3), error_exc = error_exc - (old.exc != 0 AND old.level = 4) - WHERE conn = old.conn AND minute = old.ts / 60000000 AND service IN (old.service, ''); -END;"; - -const MINUTE: i64 = 60_000_000; - -const SCHEMA: &str = r#" -PRAGMA auto_vacuum = INCREMENTAL; -PRAGMA journal_mode = WAL; -PRAGMA synchronous = NORMAL; -PRAGMA journal_size_limit = 67108864; -CREATE TABLE IF NOT EXISTS entries ( - id INTEGER PRIMARY KEY, - conn TEXT NOT NULL, - ts INTEGER NOT NULL, - stack TEXT NOT NULL, - service TEXT NOT NULL, - replica TEXT NOT NULL, - node TEXT NOT NULL, - source TEXT NOT NULL, - stream INTEGER NOT NULL, - level INTEGER NOT NULL, - exc INTEGER NOT NULL, - lines INTEGER NOT NULL, - fp TEXT, - text TEXT NOT NULL, - raw TEXT -); --- Covers the time graph and level counts, so they never read the rows themselves. -CREATE INDEX IF NOT EXISTS entries_conn_ts_level ON entries (conn, ts, level, exc); -DROP INDEX IF EXISTS entries_conn_ts; --- Only lines that belong to an error group, with what the group list needs. -CREATE INDEX IF NOT EXISTS entries_conn_ts_fp ON entries (conn, ts, fp, level, exc, service, node, source) WHERE fp IS NOT NULL; -CREATE INDEX IF NOT EXISTS entries_conn_fp ON entries (conn, fp, ts) WHERE fp IS NOT NULL; -CREATE VIRTUAL TABLE IF NOT EXISTS entries_fts USING fts5 (text, content = 'entries', content_rowid = 'id', tokenize = 'trigram'); --- Lines per minute, service and level, and per minute of the whole --- connection (service ''): the time graph and level counts of long ranges --- read these instead of every line. -CREATE TABLE IF NOT EXISTS minutes ( - conn TEXT NOT NULL, - minute INTEGER NOT NULL, - service TEXT NOT NULL, - stack TEXT NOT NULL, - debug INTEGER NOT NULL, - info INTEGER NOT NULL, - warn INTEGER NOT NULL, - error INTEGER NOT NULL, - debug_exc INTEGER NOT NULL, - info_exc INTEGER NOT NULL, - warn_exc INTEGER NOT NULL, - error_exc INTEGER NOT NULL, - PRIMARY KEY (conn, minute, service) -) WITHOUT ROWID; -CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value INTEGER NOT NULL); -CREATE TABLE IF NOT EXISTS patterns (conn TEXT NOT NULL, fp TEXT NOT NULL, pattern TEXT NOT NULL, PRIMARY KEY (conn, fp)); -CREATE TABLE IF NOT EXISTS cursors (conn TEXT NOT NULL, source TEXT NOT NULL, ts INTEGER NOT NULL, PRIMARY KEY (conn, source)); -CREATE TABLE IF NOT EXISTS hosts (conn TEXT PRIMARY KEY, identity TEXT NOT NULL); --- Container starts, stops and crashes; see events.rs. The columns the log --- filters use are named as in `entries`, so the same filters apply. -CREATE TABLE IF NOT EXISTS events ( - id INTEGER PRIMARY KEY, - conn TEXT NOT NULL, - key TEXT NOT NULL, - ts INTEGER NOT NULL, - stack TEXT NOT NULL, - service TEXT NOT NULL, - replica TEXT NOT NULL, - node TEXT NOT NULL, - source TEXT NOT NULL, - kind TEXT NOT NULL, - exit_code INTEGER, - detail TEXT, - UNIQUE (conn, key) -); -CREATE INDEX IF NOT EXISTS events_conn_ts ON events (conn, ts); --- What each service ran at its last deploy; see deploys.rs. -CREATE TABLE IF NOT EXISTS deploy_state (conn TEXT NOT NULL, service TEXT NOT NULL, ts INTEGER NOT NULL, config TEXT NOT NULL, PRIMARY KEY (conn, service)); -"#; - /// Filters shared by queries and exports. Times are Unix microseconds. #[derive(Clone, Debug, Default)] pub struct LogFilter { @@ -290,40 +198,6 @@ impl LogFilter { } } -/// Error groups whose first entry is in the last 15 minutes. Needs at least an -/// hour of stored history, or every group would look new after a fresh start. -fn new_groups(db: &Connection, conn: &str, now: i64, oldest: Option) -> Result> { - const WINDOW: i64 = 15 * 60_000_000; - const HISTORY: i64 = 3_600_000_000; - if oldest.is_none_or(|o| o > now - HISTORY) { - return Ok(Vec::new()); - } - let since = now - WINDOW; - let mut st = db.prepare_cached( - "SELECT fp, COUNT(*), MIN(ts), MIN(service), MAX(id) FROM entries - WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4 AND fp IS NOT NULL GROUP BY fp", - )?; - let recent = st - .query_map(params![conn, since, now, Level::Error.as_i64()], |r| { - Ok((r.get::<_, String>(0)?, r.get::<_, u64>(1)?, r.get::<_, i64>(2)?, r.get::<_, String>(3)?, r.get::<_, i64>(4)?)) - })? - .collect::>>()?; - let mut before = db.prepare_cached("SELECT 1 FROM entries WHERE conn = ?1 AND fp = ?2 AND ts < ?3 LIMIT 1")?; - let mut pat = db.prepare_cached("SELECT pattern FROM patterns WHERE conn = ?1 AND fp = ?2")?; - let mut text = db.prepare_cached("SELECT text FROM entries WHERE id = ?1")?; - let mut out = Vec::new(); - for (fp, count, first, service, last_id) in recent { - if before.exists(params![conn, fp, since])? { - continue; - } - let pattern = pat.query_row(params![conn, fp], |r| r.get(0)).optional()?.unwrap_or_default(); - let sample = text.query_row(params![last_id], |r| r.get(0)).optional()?.unwrap_or_default(); - out.push(NewGroup { fp, pattern, count, service, first, sample }); - } - out.sort_by_key(|g| std::cmp::Reverse(g.count)); - Ok(out) -} - /// What [`Store::check_host`] found. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum HostCheck { @@ -335,25 +209,6 @@ pub enum HostCheck { Changed, } -/// How many lines each connection gives up so that `keep` lines are left: -/// the lines kept are shared out equally, and a connection with fewer lines -/// than its share keeps all of them, which leaves more for the others. -fn over_share(counts: &[(String, u64)], keep: u64) -> Vec<(String, u64)> { - let mut sorted: Vec<&(String, u64)> = counts.iter().collect(); - sorted.sort_by_key(|(_, n)| *n); - let mut left = keep; - let mut out = Vec::new(); - for (i, (conn, n)) in sorted.iter().enumerate() { - let share = left / (sorted.len() - i) as u64; - let kept = (*n).min(share); - left -= kept; - if *n > kept { - out.push((conn.clone(), n - kept)); - } - } - out -} - fn escape_like(s: &str) -> String { s.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_") } @@ -876,733 +731,6 @@ impl Store { } } -impl Store { - /// Deletes entries older than `before` (µs) and the oldest entries beyond - /// `max_bytes`, then gives the space back to the system. - pub fn retain(&self, before: i64, max_bytes: u64) -> Result { - const BATCH: usize = 20_000; - let mut deleted = 0u64; - // Per connection, so the (conn, ts) index finds old lines without - // reading the table. The lock is taken per batch, so queries and new - // lines get through between batches. - let conns = conns(&self.db.lock().unwrap())?; - for conn in conns { - loop { - let n = self.db.lock().unwrap().execute( - "DELETE FROM entries WHERE id IN (SELECT id FROM entries WHERE conn = ?1 AND ts < ?2 LIMIT ?3)", - params![conn, before, BATCH], - )?; - deleted += n as u64; - if n < BATCH { - break; - } - } - // Counts of minutes with no lines left. - self.db - .lock() - .unwrap() - .execute("DELETE FROM minutes WHERE conn = ?1 AND minute < ?2", params![conn, before.div_euclid(MINUTE)])?; - } - self.db.lock().unwrap().execute("DELETE FROM events WHERE ts < ?1", params![before])?; - // Over the size limit: drop the oldest lines, down to 90% of the - // limit so this does not run again a minute later. Every connection - // gets an equal share of what is kept, and only connections above - // their share give up lines: a chatty server must not push out the - // whole history of a quiet one. Deleted lines stay in the search index - // until it is merged, so the index is merged before the size is - // measured again. - for _ in 0..4 { - let live = live_bytes(&self.db.lock().unwrap())?; - if live <= max_bytes { - break; - } - let counts: Vec<(String, u64)> = { - let db = self.db.lock().unwrap(); - let mut st = db.prepare("SELECT conn, COUNT(*) FROM entries GROUP BY conn")?; - st.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?.collect::>()? - }; - let total: u64 = counts.iter().map(|(_, n)| n).sum(); - if total == 0 { - break; - } - let target = max_bytes / 10 * 9; - let left = ((total as u128 * (live - target) as u128).div_ceil(live as u128) as u64).clamp(1, total); - for (conn, mut drop) in over_share(&counts, total - left) { - while drop > 0 { - let n = self.db.lock().unwrap().execute( - "DELETE FROM entries WHERE id IN (SELECT id FROM entries WHERE conn = ?1 ORDER BY ts LIMIT ?2)", - params![conn, drop.min(BATCH as u64)], - )? as u64; - deleted += n; - drop = drop.saturating_sub(n); - if n == 0 { - break; - } - } - } - self.merge_index()?; - } - { - let db = self.db.lock().unwrap(); - let mode: i64 = db.query_row("PRAGMA auto_vacuum", [], |r| r.get(0))?; - if mode != 2 { - // Stores made before 0.3 could never shrink: switch them over once. - tracing::info!("rebuilding the log store once so it can give space back"); - db.execute_batch("PRAGMA auto_vacuum = INCREMENTAL; VACUUM;")?; - } - } - // Also when nothing was deleted now: up to 0.3.1 free space was never - // given back, so stores made by those versions shrink here once. - self.shrink()?; - Ok(deleted) - } - - /// Gives free pages back to the system and moves the write-ahead log - /// into the file, so the file shrinks on disk. In steps that release the - /// lock in between, so queries and new lines get through. - fn shrink(&self) -> Result<()> { - const STEP: u64 = 4096; - let mut last = u64::MAX; - loop { - let db = self.db.lock().unwrap(); - let free: u64 = db.query_row("PRAGMA freelist_count", [], |r| r.get(0))?; - if free == 0 || free >= last { - return checkpoint(&db); - } - last = free; - // Each step of this pragma frees one page, so it is run to the end: - // run once, as `execute_batch` does, it freed a single page. - let mut st = db.prepare(&format!("PRAGMA incremental_vacuum({})", free.min(STEP)))?; - let mut rows = st.query([])?; - while rows.next()?.is_some() {} - } - } - - /// Merges the search index so deleted lines leave it, like FTS5's - /// `optimize`, but in steps that release the lock in between: on a large - /// store a full optimize would block queries and new lines for minutes. - fn merge_index(&self) -> Result<()> { - loop { - let db = self.db.lock().unwrap(); - let before = db.total_changes(); - db.execute("INSERT INTO entries_fts (entries_fts, rank) VALUES ('merge', -500)", [])?; - if db.total_changes() - before <= 1 { - return Ok(()); - } - } - } - - /// Space on disk: the database file plus its write-ahead log (kept under 64 MB). - fn used_bytes(&self, db: &Connection) -> Result { - let file = - |suffix: &str| self.path.as_ref().and_then(|p| std::fs::metadata(format!("{}{suffix}", p.display())).ok()).map(|m| m.len()); - match file("") { - Some(main) => Ok(main + file("-wal").unwrap_or(0)), - None => live_bytes(db), - } - } - - /// Lines and stored text per service of a connection, largest first. - pub fn usage(&self, conn: &str) -> Result> { - self.read(|db| usage(db, conn)) - } - - /// Deletes the stored lines of one service, in batches. - pub fn clear_service(&self, conn: &str, service: &str) -> Result { - let deleted = self.delete_entries(Some(conn), Some(service))?; - if deleted > 0 { - self.merge_index()?; - self.shrink()?; - } - Ok(deleted) - } - - /// Deletes the lines of a connection (all when None), or of one of its - /// services. Every deleted line also has to leave the search index, which - /// is the slow part, so it runs in small batches and releases the lock in - /// between. When nothing would be left, the table and the index are - /// emptied at once instead. - fn delete_entries(&self, conn: Option<&str>, service: Option<&str>) -> Result { - const BATCH: i64 = 5_000; - { - let db = self.db.lock().unwrap(); - let everything = match (conn, service) { - (None, _) => true, - (Some(c), None) => conns(&db)? == [c], - (Some(_), Some(_)) => false, - }; - if everything { - let n: u64 = db.query_row("SELECT COUNT(*) FROM entries", [], |r| r.get(0))?; - db.execute_batch(&format!( - "BEGIN; - INSERT INTO entries_fts (entries_fts) VALUES ('delete-all'); - DROP TRIGGER entries_ad; - DELETE FROM entries; - DELETE FROM minutes; - UPDATE meta SET value = 0 WHERE key = 'minutes_since'; - {DELETE_TRIGGER} - COMMIT;" - ))?; - // Nothing older is left, so every range can use the counts. - self.minutes_since.store(0, std::sync::atomic::Ordering::Relaxed); - return Ok(n); - } - } - let (conn, mut deleted, mut after) = (conn.unwrap_or_default(), 0u64, 0i64); - loop { - let db = self.db.lock().unwrap(); - // Walks the rows in id order, so each batch starts where the last - // one stopped instead of reading the table from the start again. - let n = match service { - Some(svc) => { - let last: Option = db.query_row( - "SELECT MAX(id) FROM (SELECT id FROM entries WHERE id > ?1 AND conn = ?2 AND service = ?3 ORDER BY id LIMIT ?4)", - params![after, conn, svc, BATCH], - |r| r.get(0), - )?; - let Some(last) = last else { break }; - let n = db.execute( - "DELETE FROM entries WHERE id > ?1 AND id <= ?2 AND conn = ?3 AND service = ?4", - params![after, last, conn, svc], - )?; - after = last; - n - } - // The (conn, ts) index finds them; deleted rows leave it. - None => { - db.execute("DELETE FROM entries WHERE id IN (SELECT id FROM entries WHERE conn = ?1 LIMIT ?2)", params![conn, BATCH])? - } - }; - deleted += n as u64; - if service.is_none() && (n as i64) < BATCH { - break; - } - } - Ok(deleted) - } - - /// Counts the error entries of a connection in `[from, to)` (µs). - pub fn error_rate(&self, conn: &str, from: i64, to: i64) -> Result { - self.read(|db| error_rate(db, conn, from, to)) - } - - /// Checks which host or swarm a connection's logs come from against the - /// one remembered. The first one seen is remembered. A different one is - /// only reported: its lines must not mix with the old ones, and whether - /// the old ones go is the user's call ([`Store::accept_host`]). - pub fn check_host(&self, conn: &str, identity: &str) -> Result { - let db = self.db.lock().unwrap(); - let known: Option = db.query_row("SELECT identity FROM hosts WHERE conn = ?1", params![conn], |r| r.get(0)).optional()?; - match known { - Some(k) if k == identity => Ok(HostCheck::Same), - Some(_) => Ok(HostCheck::Changed), - None => { - db.execute("INSERT INTO hosts (conn, identity) VALUES (?1, ?2)", params![conn, identity])?; - Ok(HostCheck::First) - } - } - } - - /// Makes `identity` the host of a connection. With `keep_old` false the - /// lines from the host before are deleted first. - pub fn accept_host(&self, conn: &str, identity: &str, keep_old: bool) -> Result<()> { - if !keep_old { - self.clear(Some(conn))?; - } - self.db.lock().unwrap().execute( - "INSERT INTO hosts (conn, identity) VALUES (?1, ?2) ON CONFLICT (conn) DO UPDATE SET identity = excluded.identity", - params![conn, identity], - )?; - Ok(()) - } - - pub fn stats(&self) -> Result { - self.read(|db| { - let entries = db.query_row("SELECT COUNT(*) FROM entries", [], |r| r.get(0))?; - let mut oldest: Option = None; - for c in conns(db)? { - let t: Option = db.query_row("SELECT MIN(ts) FROM entries WHERE conn = ?1", params![c], |r| r.get(0))?; - oldest = oldest.into_iter().chain(t).min(); - } - Ok(StoreStats { bytes: self.used_bytes(db)?, entries, oldest }) - }) - } - - /// Changes whenever stored lines of a whole connection are cleared. - pub fn epoch(&self) -> u64 { - self.epoch.load(std::sync::atomic::Ordering::SeqCst) - } - - pub fn clear(&self, conn: Option<&str>) -> Result<()> { - // First, so lines a collector is reading right now are not stored - // after the clear with a cursor that would skip everything before them. - // Under the write lock: a batch being stored right now either lands - // before this and is deleted below, or sees the new epoch and is dropped. - { - let _db = self.db.lock().unwrap(); - self.epoch.fetch_add(1, std::sync::atomic::Ordering::SeqCst); - } - let deleted = self.delete_entries(conn, None)?; - if deleted > 0 { - self.merge_index()?; - } - let db = self.db.lock().unwrap(); - match conn { - Some(c) => { - db.execute("DELETE FROM minutes WHERE conn = ?1", params![c])?; - db.execute("DELETE FROM patterns WHERE conn = ?1", params![c])?; - db.execute("DELETE FROM cursors WHERE conn = ?1", params![c])?; - db.execute("DELETE FROM hosts WHERE conn = ?1", params![c])?; - db.execute("DELETE FROM events WHERE conn = ?1", params![c])?; - db.execute("DELETE FROM deploy_state WHERE conn = ?1", params![c])?; - } - None => { - db.execute_batch("DELETE FROM minutes; DELETE FROM patterns; DELETE FROM cursors; DELETE FROM hosts; DELETE FROM events; DELETE FROM deploy_state;")? - } - } - drop(db); - self.shrink() - } -} - -#[allow(clippy::too_many_arguments)] -fn query( - db: &Connection, - conn: &str, - f: &LogFilter, - buckets: usize, - limit: usize, - minutes_since: i64, - indexed: bool, - page: Option, -) -> Result { - let buckets = buckets.clamp(1, 500) as i64; - // A page of older lines is only lines: the counts, the graph and the - // groups are those of the first page, which the caller has already. - let lines_only = matches!(page, Some(Page::Before(..))); - let span = (f.to - f.from).max(1); - // Picked stacks become their services, so the service index finds them. - let resolved; - let f = match indexed && !f.pick_stacks.is_empty() { - true => { - let mut g = f.clone(); - let found = services_of_stacks(db, conn, &g.pick_stacks)?; - // With none found the stacks have no lines, and the filter stays as it is. - if !found.is_empty() { - g.exact_services.extend(found.into_iter().filter(|s| !f.exact_services.contains(s))); - g.pick_stacks.clear(); - } - resolved = g; - &resolved - } - false => f, - }; - - // The level counts (which ignore the level filter, so the chips show what - // each would add) and the histogram. Whole minutes come from the - // per-minute counts when the filter only uses what they know; the rest is - // one pass over (ts, level, exc), read from a covering index. Bucketing - // here is about ten times faster than a GROUP BY, which makes SQLite sort - // every row of the range. - let wanted = |l: Level| !f.no_levels && (f.levels.is_empty() || f.levels.contains(&l)); - let mut levels = LevelCounts::default(); - let mut histogram = vec![Bucket::default(); buckets as usize]; - let mut total = 0u64; - let mut count = |ts: i64, level: Level, n: u64, exc: u64| { - *levels.get_mut(level) += n; - if wanted(level) { - let bk = &mut histogram[((ts - f.from) * buckets / span).clamp(0, buckets - 1) as usize]; - total += n; - bk.exc += exc; - *bk.get_mut(level) += n; - } - }; - let countable = - !lines_only && f.text.is_empty() && f.fps.is_empty() && f.nodes.is_empty() && f.replicas.is_empty() && f.sources.is_empty(); - let lo = (f.from.div_euclid(MINUTE) + (f.from.rem_euclid(MINUTE) != 0) as i64).max(minutes_since.div_euclid(MINUTE)); - let hi = f.to.div_euclid(MINUTE); - let mut raw = if lines_only { vec![] } else { vec![(f.from, f.to)] }; - if countable && hi - lo >= 30 { - raw = vec![(f.from, f.from.max(lo * MINUTE)), (f.to.min(hi * MINUTE), f.to)]; - let (mut w, p) = f.where_clause_at(conn, false, "minute", lo, hi); - // Without a service filter the rows of the whole connection are enough. - let whole = f.services.is_empty() && f.exact_services.is_empty() && f.pick_stacks.is_empty() && f.stacks.is_empty(); - w.push_str(if whole { " AND service = ''" } else { " AND service != ''" }); - let mut st = db.prepare_cached(&format!( - "SELECT minute, debug, info, warn, error, debug_exc, info_exc, warn_exc, error_exc FROM minutes WHERE {w}" - ))?; - let mut rows = st.query(params_from_iter(p.iter()))?; - while let Some(r) = rows.next()? { - let ts = r.get::<_, i64>(0)? * MINUTE; - for (i, l) in [Level::Debug, Level::Info, Level::Warn, Level::Error].into_iter().enumerate() { - let (n, x): (i64, i64) = (r.get(1 + i)?, r.get(5 + i)?); - if n > 0 { - count(ts, l, n as u64, x.max(0) as u64); - } - } - } - } - for (from, to) in raw.into_iter().filter(|(a, b)| a < b) { - let (w, p) = f.where_clause_at(conn, false, "ts", from, to); - let mut st = db.prepare_cached(&format!("SELECT ts, level, exc FROM entries WHERE {w}"))?; - let mut rows = st.query(params_from_iter(p.iter()))?; - while let Some(r) = rows.next()? { - let (ts, level, exc): (i64, i64, i64) = (r.get(0)?, r.get(1)?, r.get(2)?); - count(ts, Level::from_i64(level), 1, (exc != 0) as u64); - } - } - - // Error groups, aggregated the same way. The partial index holds only lines with a fingerprint. - let (w, p) = f.where_clause(conn, true); - // The sparklines (24 buckets per group) come from the same pass. - struct Agg { - g: Group, - sources: std::collections::HashSet, - max_id: i64, - } - let mut by_fp: HashMap = HashMap::new(); - let mut st = db.prepare(&format!( - "SELECT fp, level, ts, exc, service, node, source, id FROM entries WHERE {w} AND fp IS NOT NULL AND level >= 3{}", - if lines_only { " AND 0" } else { "" } - ))?; - let mut rows = st.query(params_from_iter(p.iter()))?; - while let Some(r) = rows.next()? { - // Borrowed strings: most rows only bump counters of a group that exists already. - let fp = r.get_ref(0)?.as_str()?; - let (level, ts, exc, id): (i64, i64, i64, i64) = (r.get(1)?, r.get(2)?, r.get(3)?, r.get(7)?); - let level = Level::from_i64(level); - let a = match by_fp.get_mut(fp) { - Some(a) => a, - None => by_fp.entry(fp.to_string()).or_insert(Agg { - g: Group { - fp: fp.to_string(), - pattern: String::new(), - count: 0, - level, - first: ts, - last: ts, - services: vec![], - nodes: vec![], - sources: 0, - exc: false, - sample: String::new(), - spark: vec![0; 24], - }, - sources: Default::default(), - max_id: id, - }), - }; - let g = &mut a.g; - g.count += 1; - g.level = g.level.max(level); - g.first = g.first.min(ts); - g.last = g.last.max(ts); - g.exc |= exc != 0; - g.spark[((ts - f.from) * 24 / span).clamp(0, 23) as usize] += 1; - let service = r.get_ref(4)?.as_str()?; - if !g.services.iter().any(|x| x == service) { - g.services.push(service.to_string()); - } - let node = r.get_ref(5)?.as_str()?; - if !g.nodes.iter().any(|x| x == node) { - g.nodes.push(node.to_string()); - } - let source = r.get_ref(6)?.as_str()?; - if !a.sources.contains(source) { - a.sources.insert(source.to_string()); - } - a.max_id = a.max_id.max(id); - } - drop(rows); - drop(st); - let mut groups: Vec = by_fp - .into_values() - .map(|mut a| { - a.g.sources = a.sources.len() as u64; - a.g.sample = a.max_id.to_string(); - a.g - }) - .collect(); - groups.sort_by(|a, b| b.count.cmp(&a.count).then(b.last.cmp(&a.last))); - groups.truncate(50); - if !groups.is_empty() { - let mut pat = db.prepare_cached("SELECT pattern FROM patterns WHERE conn = ?1 AND fp = ?2")?; - let mut sample = db.prepare_cached("SELECT text FROM entries WHERE id = ?1")?; - for g in &mut groups { - g.pattern = pat.query_row(params![conn, g.fp], |r| r.get(0)).optional()?.unwrap_or_default(); - let id: i64 = g.sample.parse().unwrap_or(0); - g.sample = sample.query_row(params![id], |r| r.get(0)).optional()?.unwrap_or_default(); - } - } - - // The newest lines. With picked services each one walks its own part of - // the service index, newest first, and the results are merged: a quiet - // service would otherwise mean reading the whole range. - // One line more than asked for says whether there are more. - let limit = limit.clamp(1, 5000); - let per_service = indexed && f.pick_stacks.is_empty() && (1..=50).contains(&f.exact_services.len()); - let parts: Vec<(String, Vec)> = if per_service { - f.exact_services.iter().map(|s| LogFilter { exact_services: vec![s.clone()], ..f.clone() }.where_clause(conn, true)).collect() - } else { - vec![(w, p)] - }; - let mut entries = Vec::new(); - for (mut w, mut p) in parts { - // The (ts, id) order of the stream; `ts` alone first, so the index finds the start. - match page { - Some(Page::Before(ts, id)) => { - w.push_str(" AND ts <= ? AND (ts < ? OR id < ?)"); - p.extend([Sql::Integer(ts), Sql::Integer(ts), Sql::Integer(id)]); - } - Some(Page::After(ts, id)) => { - w.push_str(" AND ts >= ? AND (ts > ? OR id > ?)"); - p.extend([Sql::Integer(ts), Sql::Integer(ts), Sql::Integer(id)]); - } - None => {} - } - let mut st = db.prepare_cached(&format!( - "SELECT id, ts, stack, service, replica, node, source, stream, level, exc, lines, fp, text, raw - FROM entries WHERE {w} ORDER BY ts DESC, id DESC LIMIT {}", - limit + 1 - ))?; - entries.extend(st.query_map(params_from_iter(p.iter()), row_to_entry)?.collect::>>()?); - } - if per_service && f.exact_services.len() > 1 { - entries.sort_by(|a, b| b.ts.cmp(&a.ts).then(b.id.cmp(&a.id))); - } - let more = entries.len() > limit; - entries.truncate(limit); - - Ok(QueryResult { total, levels, histogram, groups, entries, more }) -} - -/// The services with stored lines in these stacks, found by skipping through -/// the service index. -fn services_of_stacks(db: &Connection, conn: &str, stacks: &[String]) -> Result> { - let mut st = db.prepare_cached( - "WITH RECURSIVE s(service) AS ( - SELECT MIN(service) FROM entries WHERE conn = ?1 - UNION ALL SELECT (SELECT MIN(service) FROM entries WHERE conn = ?1 AND service > s.service) FROM s WHERE s.service IS NOT NULL - ) SELECT service, (SELECT stack FROM entries WHERE conn = ?1 AND service = s.service LIMIT 1) FROM s WHERE service IS NOT NULL", - )?; - let rows = st.query_map(params![conn], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?; - let mut out = Vec::new(); - for row in rows { - let (service, stack) = row?; - if stacks.contains(&stack) { - out.push(service); - } - } - Ok(out) -} - -fn error_summary(db: &Connection, conn: &str, now: i64) -> Result { - const HOUR: i64 = 3_600_000_000; - let err = Level::Error.as_i64(); - let mut by_service = HashMap::new(); - // Without the hint SQLite walks the service index of the whole store to skip a sort. - let mut st = db.prepare_cached( - "SELECT service, COUNT(*) FROM entries INDEXED BY entries_conn_ts_level - WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4 GROUP BY service", - )?; - let rows = st.query_map(params![conn, now - HOUR, now, err], |r| Ok((r.get::<_, String>(0)?, r.get::<_, u64>(1)?)))?; - for row in rows { - let (svc, n) = row?; - by_service.insert(svc, n); - } - let last_hour = by_service.values().sum(); - let prev_hour: u64 = db.query_row( - "SELECT COUNT(*) FROM entries WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4", - params![conn, now - 2 * HOUR, now - HOUR, err], - |r| r.get(0), - )?; - let mut spark = vec![0u64; 24]; - let from = now - 12 * HOUR; - let mut st = db.prepare_cached( - "SELECT (ts - ?2) * 24 / ?3 AS b, COUNT(*) FROM entries WHERE conn = ?1 AND ts >= ?2 AND ts < ?4 AND level = ?5 GROUP BY b", - )?; - let rows = st.query_map(params![conn, from, 12 * HOUR, now, err], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, u64>(1)?)))?; - for row in rows { - let (b, n) = row?; - spark[b.clamp(0, 23) as usize] += n; - } - let oldest: Option = db.query_row("SELECT MIN(ts) FROM entries WHERE conn = ?1", params![conn], |r| r.get(0))?; - let new_groups = new_groups(db, conn, now, oldest)?; - Ok(ErrorSummary { by_service, last_hour, prev_hour, spark, oldest, new_groups, quiet: Vec::new() }) -} - -/// Services that went quiet: lines in at least half the minutes of two hours, -/// then none for ten minutes. Read from the per-minute counts, so it needs -/// two hours and ten minutes of counted history. -fn quiet(db: &Connection, conn: &str, now: i64, minutes_since: i64) -> Result> { - const SILENT: i64 = 10; - const BASELINE: i64 = 120; - let cut = now.div_euclid(MINUTE) - SILENT; - let lo = cut - BASELINE; - if minutes_since > lo * MINUTE { - return Ok(Vec::new()); - } - let mut st = db.prepare_cached( - "SELECT service, SUM(minute < ?3), MAX(minute) FROM minutes - WHERE conn = ?1 AND service != '' AND minute >= ?2 AND debug + info + warn + error > 0 - GROUP BY service HAVING SUM(minute < ?3) >= ?4 AND MAX(minute) < ?3", - )?; - let rows = st.query_map(params![conn, lo, cut, BASELINE / 2], |r| { - Ok(Quiet { service: r.get(0)?, busy_minutes: r.get(1)?, since: (r.get::<_, i64>(2)? + 1) * MINUTE }) - })?; - let mut out = rows.collect::>>()?; - out.sort_by_key(|q| q.since); - Ok(out) -} - -fn export(db: &Connection, conn: &str, f: &LogFilter, format: ExportFormat, out: &mut dyn Write) -> Result { - let (w, p) = f.where_clause(conn, true); - let mut st = db.prepare(&format!( - "SELECT id, ts, stack, service, replica, node, source, stream, level, exc, lines, fp, text, raw - FROM entries WHERE {w} ORDER BY ts ASC, id ASC" - ))?; - let mut rows = st.query(params_from_iter(p.iter()))?; - let mut n = 0; - if format == ExportFormat::Csv { - writeln!(out, "time,level,stack,service,replica,node,stream,message")?; - } - while let Some(r) = rows.next()? { - let e = row_to_entry(r)?; - let time = chrono::DateTime::from_timestamp_micros(e.ts) - .map(|d| d.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)) - .unwrap_or_default(); - match format { - ExportFormat::Log => { - writeln!(out, "{time} {:<5} {}.{} {}", e.level.name(), e.service, e.replica, e.text)?; - } - ExportFormat::Jsonl => { - let v = serde_json::json!({ - "time": time, "level": e.level.name(), "stack": e.stack, "service": e.service, - "replica": e.replica, "node": e.node, "stream": e.stream, "message": e.text, - "raw": e.raw.as_deref().and_then(|r| serde_json::from_str::(r).ok()), - }); - writeln!(out, "{v}")?; - } - ExportFormat::Csv => { - writeln!( - out, - "{},{},{},{},{},{},{},{}", - time, - e.level.name(), - csv(&e.stack), - csv(&e.service), - csv(&e.replica), - csv(&e.node), - e.stream, - csv(&e.text) - )?; - } - } - n += 1; - } - Ok(n) -} - -fn error_rate(db: &Connection, conn: &str, from: i64, to: i64) -> Result { - let err = Level::Error.as_i64(); - let count: u64 = db.query_row( - "SELECT COUNT(*) FROM entries WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4", - params![conn, from, to, err], - |r| r.get(0), - )?; - let mut rate = ErrorRate { count, ..Default::default() }; - if count == 0 { - return Ok(rate); - } - // The service comes from the group's newest line: a bare column next to - // GROUP BY would be any line's, and a group can span several services. - let top: Option<(String, i64)> = db - .query_row( - "SELECT fp, MAX(id) FROM entries WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4 AND fp IS NOT NULL - GROUP BY fp ORDER BY COUNT(*) DESC LIMIT 1", - params![conn, from, to, err], - |r| Ok((r.get(0)?, r.get(1)?)), - ) - .optional()?; - if let Some((fp, id)) = top { - let pattern: Option = - db.query_row("SELECT pattern FROM patterns WHERE conn = ?1 AND fp = ?2", params![conn, fp], |r| r.get(0)).optional()?; - let (service, sample): (String, String) = - db.query_row("SELECT service, text FROM entries WHERE id = ?1", params![id], |r| Ok((r.get(0)?, r.get(1)?)))?; - let text = pattern.filter(|p| !p.is_empty()).unwrap_or(sample); - rate.top = Some(text.lines().next().unwrap_or("").chars().take(200).collect()); - rate.top_service = Some(service); - } - Ok(rate) -} - -/// Lines and stored text per service of a connection, largest first. -fn usage(db: &Connection, conn: &str) -> Result> { - let mut st = db.prepare( - "SELECT service, COUNT(*), SUM(LENGTH(text) + COALESCE(LENGTH(raw), 0)) AS b FROM entries WHERE conn = ?1 - GROUP BY service ORDER BY b DESC", - )?; - let rows = st.query_map(params![conn], |r| Ok(ServiceUsage { service: r.get(0)?, entries: r.get(1)?, text_bytes: r.get(2)? }))?; - Ok(rows.collect::>>()?) -} - -/// The connections that have stored lines, found by skipping through the -/// index instead of reading every row. -fn conns(db: &Connection) -> Result> { - let mut st = db.prepare_cached( - "WITH RECURSIVE c(conn) AS ( - SELECT MIN(conn) FROM entries - UNION ALL SELECT (SELECT MIN(conn) FROM entries WHERE conn > c.conn) FROM c WHERE c.conn IS NOT NULL - ) SELECT conn FROM c WHERE conn IS NOT NULL", - )?; - let rows = st.query_map([], |r| r.get(0))?; - Ok(rows.collect::>>()?) -} - -/// Moves the write-ahead log into the database and empties it. A read still -/// running (an export, say) keeps part of the log in use: then it is left -/// for the next time instead of waiting, which would hold up new lines. -fn checkpoint(db: &Connection) -> Result<()> { - db.busy_timeout(std::time::Duration::ZERO)?; - let r = db.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);"); - db.busy_timeout(std::time::Duration::from_secs(5))?; - Ok(r?) -} - -/// Pages that hold data; free pages are given back by `incremental_vacuum`. -fn live_bytes(db: &Connection) -> Result { - let pages: u64 = db.query_row("PRAGMA page_count", [], |r| r.get(0))?; - let free: u64 = db.query_row("PRAGMA freelist_count", [], |r| r.get(0))?; - let size: u64 = db.query_row("PRAGMA page_size", [], |r| r.get(0))?; - Ok(pages.saturating_sub(free) * size) -} - -fn row_to_entry(r: &rusqlite::Row<'_>) -> rusqlite::Result { - Ok(EntryOut { - id: r.get(0)?, - ts: r.get(1)?, - stack: r.get(2)?, - service: r.get(3)?, - replica: r.get(4)?, - node: r.get(5)?, - source: r.get(6)?, - stream: if r.get::<_, i64>(7)? == 2 { "stderr" } else { "stdout" }, - level: Level::from_i64(r.get(8)?), - exc: r.get::<_, i64>(9)? != 0, - lines: r.get(10)?, - fp: r.get(11)?, - text: r.get(12)?, - raw: r.get(13)?, - }) -} - -fn csv(s: &str) -> String { - // A cell starting like a formula is run by spreadsheet apps when the file - // is opened; a log line must never do that, so it starts with ' instead. - let s = if s.starts_with(['=', '+', '-', '@', '\t', '\r']) { format!("'{s}") } else { s.to_string() }; - if s.contains([',', '"', '\n', '\r']) { format!("\"{}\"", s.replace('"', "\"\"")) } else { s } -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/stackline-core/src/store/query.rs b/crates/stackline-core/src/store/query.rs new file mode 100644 index 0000000..cebd15a --- /dev/null +++ b/crates/stackline-core/src/store/query.rs @@ -0,0 +1,390 @@ +//! Searches, counts, the time graph and error groups: everything the logs +//! view asks of the store. + +use super::*; + +/// Error groups whose first entry is in the last 15 minutes. Needs at least an +/// hour of stored history, or every group would look new after a fresh start. +pub(super) fn new_groups(db: &Connection, conn: &str, now: i64, oldest: Option) -> Result> { + const WINDOW: i64 = 15 * 60_000_000; + const HISTORY: i64 = 3_600_000_000; + if oldest.is_none_or(|o| o > now - HISTORY) { + return Ok(Vec::new()); + } + let since = now - WINDOW; + let mut st = db.prepare_cached( + "SELECT fp, COUNT(*), MIN(ts), MIN(service), MAX(id) FROM entries + WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4 AND fp IS NOT NULL GROUP BY fp", + )?; + let recent = st + .query_map(params![conn, since, now, Level::Error.as_i64()], |r| { + Ok((r.get::<_, String>(0)?, r.get::<_, u64>(1)?, r.get::<_, i64>(2)?, r.get::<_, String>(3)?, r.get::<_, i64>(4)?)) + })? + .collect::>>()?; + let mut before = db.prepare_cached("SELECT 1 FROM entries WHERE conn = ?1 AND fp = ?2 AND ts < ?3 LIMIT 1")?; + let mut pat = db.prepare_cached("SELECT pattern FROM patterns WHERE conn = ?1 AND fp = ?2")?; + let mut text = db.prepare_cached("SELECT text FROM entries WHERE id = ?1")?; + let mut out = Vec::new(); + for (fp, count, first, service, last_id) in recent { + if before.exists(params![conn, fp, since])? { + continue; + } + let pattern = pat.query_row(params![conn, fp], |r| r.get(0)).optional()?.unwrap_or_default(); + let sample = text.query_row(params![last_id], |r| r.get(0)).optional()?.unwrap_or_default(); + out.push(NewGroup { fp, pattern, count, service, first, sample }); + } + out.sort_by_key(|g| std::cmp::Reverse(g.count)); + Ok(out) +} + +#[allow(clippy::too_many_arguments)] +pub(super) fn query( + db: &Connection, + conn: &str, + f: &LogFilter, + buckets: usize, + limit: usize, + minutes_since: i64, + indexed: bool, + page: Option, +) -> Result { + let buckets = buckets.clamp(1, 500) as i64; + // A page of older lines is only lines: the counts, the graph and the + // groups are those of the first page, which the caller has already. + let lines_only = matches!(page, Some(Page::Before(..))); + let span = (f.to - f.from).max(1); + // Picked stacks become their services, so the service index finds them. + let resolved; + let f = match indexed && !f.pick_stacks.is_empty() { + true => { + let mut g = f.clone(); + let found = services_of_stacks(db, conn, &g.pick_stacks)?; + // With none found the stacks have no lines, and the filter stays as it is. + if !found.is_empty() { + g.exact_services.extend(found.into_iter().filter(|s| !f.exact_services.contains(s))); + g.pick_stacks.clear(); + } + resolved = g; + &resolved + } + false => f, + }; + + // The level counts (which ignore the level filter, so the chips show what + // each would add) and the histogram. Whole minutes come from the + // per-minute counts when the filter only uses what they know; the rest is + // one pass over (ts, level, exc), read from a covering index. Bucketing + // here is about ten times faster than a GROUP BY, which makes SQLite sort + // every row of the range. + let wanted = |l: Level| !f.no_levels && (f.levels.is_empty() || f.levels.contains(&l)); + let mut levels = LevelCounts::default(); + let mut histogram = vec![Bucket::default(); buckets as usize]; + let mut total = 0u64; + let mut count = |ts: i64, level: Level, n: u64, exc: u64| { + *levels.get_mut(level) += n; + if wanted(level) { + let bk = &mut histogram[((ts - f.from) * buckets / span).clamp(0, buckets - 1) as usize]; + total += n; + bk.exc += exc; + *bk.get_mut(level) += n; + } + }; + let countable = + !lines_only && f.text.is_empty() && f.fps.is_empty() && f.nodes.is_empty() && f.replicas.is_empty() && f.sources.is_empty(); + let lo = (f.from.div_euclid(MINUTE) + (f.from.rem_euclid(MINUTE) != 0) as i64).max(minutes_since.div_euclid(MINUTE)); + let hi = f.to.div_euclid(MINUTE); + let mut raw = if lines_only { vec![] } else { vec![(f.from, f.to)] }; + if countable && hi - lo >= 30 { + raw = vec![(f.from, f.from.max(lo * MINUTE)), (f.to.min(hi * MINUTE), f.to)]; + let (mut w, p) = f.where_clause_at(conn, false, "minute", lo, hi); + // Without a service filter the rows of the whole connection are enough. + let whole = f.services.is_empty() && f.exact_services.is_empty() && f.pick_stacks.is_empty() && f.stacks.is_empty(); + w.push_str(if whole { " AND service = ''" } else { " AND service != ''" }); + let mut st = db.prepare_cached(&format!( + "SELECT minute, debug, info, warn, error, debug_exc, info_exc, warn_exc, error_exc FROM minutes WHERE {w}" + ))?; + let mut rows = st.query(params_from_iter(p.iter()))?; + while let Some(r) = rows.next()? { + let ts = r.get::<_, i64>(0)? * MINUTE; + for (i, l) in [Level::Debug, Level::Info, Level::Warn, Level::Error].into_iter().enumerate() { + let (n, x): (i64, i64) = (r.get(1 + i)?, r.get(5 + i)?); + if n > 0 { + count(ts, l, n as u64, x.max(0) as u64); + } + } + } + } + for (from, to) in raw.into_iter().filter(|(a, b)| a < b) { + let (w, p) = f.where_clause_at(conn, false, "ts", from, to); + let mut st = db.prepare_cached(&format!("SELECT ts, level, exc FROM entries WHERE {w}"))?; + let mut rows = st.query(params_from_iter(p.iter()))?; + while let Some(r) = rows.next()? { + let (ts, level, exc): (i64, i64, i64) = (r.get(0)?, r.get(1)?, r.get(2)?); + count(ts, Level::from_i64(level), 1, (exc != 0) as u64); + } + } + + // Error groups, aggregated the same way. The partial index holds only lines with a fingerprint. + let (w, p) = f.where_clause(conn, true); + // The sparklines (24 buckets per group) come from the same pass. + struct Agg { + g: Group, + sources: std::collections::HashSet, + max_id: i64, + } + let mut by_fp: HashMap = HashMap::new(); + let mut st = db.prepare(&format!( + "SELECT fp, level, ts, exc, service, node, source, id FROM entries WHERE {w} AND fp IS NOT NULL AND level >= 3{}", + if lines_only { " AND 0" } else { "" } + ))?; + let mut rows = st.query(params_from_iter(p.iter()))?; + while let Some(r) = rows.next()? { + // Borrowed strings: most rows only bump counters of a group that exists already. + let fp = r.get_ref(0)?.as_str()?; + let (level, ts, exc, id): (i64, i64, i64, i64) = (r.get(1)?, r.get(2)?, r.get(3)?, r.get(7)?); + let level = Level::from_i64(level); + let a = match by_fp.get_mut(fp) { + Some(a) => a, + None => by_fp.entry(fp.to_string()).or_insert(Agg { + g: Group { + fp: fp.to_string(), + pattern: String::new(), + count: 0, + level, + first: ts, + last: ts, + services: vec![], + nodes: vec![], + sources: 0, + exc: false, + sample: String::new(), + spark: vec![0; 24], + }, + sources: Default::default(), + max_id: id, + }), + }; + let g = &mut a.g; + g.count += 1; + g.level = g.level.max(level); + g.first = g.first.min(ts); + g.last = g.last.max(ts); + g.exc |= exc != 0; + g.spark[((ts - f.from) * 24 / span).clamp(0, 23) as usize] += 1; + let service = r.get_ref(4)?.as_str()?; + if !g.services.iter().any(|x| x == service) { + g.services.push(service.to_string()); + } + let node = r.get_ref(5)?.as_str()?; + if !g.nodes.iter().any(|x| x == node) { + g.nodes.push(node.to_string()); + } + let source = r.get_ref(6)?.as_str()?; + if !a.sources.contains(source) { + a.sources.insert(source.to_string()); + } + a.max_id = a.max_id.max(id); + } + drop(rows); + drop(st); + let mut groups: Vec = by_fp + .into_values() + .map(|mut a| { + a.g.sources = a.sources.len() as u64; + a.g.sample = a.max_id.to_string(); + a.g + }) + .collect(); + groups.sort_by(|a, b| b.count.cmp(&a.count).then(b.last.cmp(&a.last))); + groups.truncate(50); + if !groups.is_empty() { + let mut pat = db.prepare_cached("SELECT pattern FROM patterns WHERE conn = ?1 AND fp = ?2")?; + let mut sample = db.prepare_cached("SELECT text FROM entries WHERE id = ?1")?; + for g in &mut groups { + g.pattern = pat.query_row(params![conn, g.fp], |r| r.get(0)).optional()?.unwrap_or_default(); + let id: i64 = g.sample.parse().unwrap_or(0); + g.sample = sample.query_row(params![id], |r| r.get(0)).optional()?.unwrap_or_default(); + } + } + + // The newest lines. With picked services each one walks its own part of + // the service index, newest first, and the results are merged: a quiet + // service would otherwise mean reading the whole range. + // One line more than asked for says whether there are more. + let limit = limit.clamp(1, 5000); + let per_service = indexed && f.pick_stacks.is_empty() && (1..=50).contains(&f.exact_services.len()); + let parts: Vec<(String, Vec)> = if per_service { + f.exact_services.iter().map(|s| LogFilter { exact_services: vec![s.clone()], ..f.clone() }.where_clause(conn, true)).collect() + } else { + vec![(w, p)] + }; + let mut entries = Vec::new(); + for (mut w, mut p) in parts { + // The (ts, id) order of the stream; `ts` alone first, so the index finds the start. + match page { + Some(Page::Before(ts, id)) => { + w.push_str(" AND ts <= ? AND (ts < ? OR id < ?)"); + p.extend([Sql::Integer(ts), Sql::Integer(ts), Sql::Integer(id)]); + } + Some(Page::After(ts, id)) => { + w.push_str(" AND ts >= ? AND (ts > ? OR id > ?)"); + p.extend([Sql::Integer(ts), Sql::Integer(ts), Sql::Integer(id)]); + } + None => {} + } + let mut st = db.prepare_cached(&format!( + "SELECT id, ts, stack, service, replica, node, source, stream, level, exc, lines, fp, text, raw + FROM entries WHERE {w} ORDER BY ts DESC, id DESC LIMIT {}", + limit + 1 + ))?; + entries.extend(st.query_map(params_from_iter(p.iter()), row_to_entry)?.collect::>>()?); + } + if per_service && f.exact_services.len() > 1 { + entries.sort_by(|a, b| b.ts.cmp(&a.ts).then(b.id.cmp(&a.id))); + } + let more = entries.len() > limit; + entries.truncate(limit); + + Ok(QueryResult { total, levels, histogram, groups, entries, more }) +} + +/// The services with stored lines in these stacks, found by skipping through +/// the service index. +pub(super) fn services_of_stacks(db: &Connection, conn: &str, stacks: &[String]) -> Result> { + let mut st = db.prepare_cached( + "WITH RECURSIVE s(service) AS ( + SELECT MIN(service) FROM entries WHERE conn = ?1 + UNION ALL SELECT (SELECT MIN(service) FROM entries WHERE conn = ?1 AND service > s.service) FROM s WHERE s.service IS NOT NULL + ) SELECT service, (SELECT stack FROM entries WHERE conn = ?1 AND service = s.service LIMIT 1) FROM s WHERE service IS NOT NULL", + )?; + let rows = st.query_map(params![conn], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?; + let mut out = Vec::new(); + for row in rows { + let (service, stack) = row?; + if stacks.contains(&stack) { + out.push(service); + } + } + Ok(out) +} + +pub(super) fn error_summary(db: &Connection, conn: &str, now: i64) -> Result { + const HOUR: i64 = 3_600_000_000; + let err = Level::Error.as_i64(); + let mut by_service = HashMap::new(); + // Without the hint SQLite walks the service index of the whole store to skip a sort. + let mut st = db.prepare_cached( + "SELECT service, COUNT(*) FROM entries INDEXED BY entries_conn_ts_level + WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4 GROUP BY service", + )?; + let rows = st.query_map(params![conn, now - HOUR, now, err], |r| Ok((r.get::<_, String>(0)?, r.get::<_, u64>(1)?)))?; + for row in rows { + let (svc, n) = row?; + by_service.insert(svc, n); + } + let last_hour = by_service.values().sum(); + let prev_hour: u64 = db.query_row( + "SELECT COUNT(*) FROM entries WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4", + params![conn, now - 2 * HOUR, now - HOUR, err], + |r| r.get(0), + )?; + let mut spark = vec![0u64; 24]; + let from = now - 12 * HOUR; + let mut st = db.prepare_cached( + "SELECT (ts - ?2) * 24 / ?3 AS b, COUNT(*) FROM entries WHERE conn = ?1 AND ts >= ?2 AND ts < ?4 AND level = ?5 GROUP BY b", + )?; + let rows = st.query_map(params![conn, from, 12 * HOUR, now, err], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, u64>(1)?)))?; + for row in rows { + let (b, n) = row?; + spark[b.clamp(0, 23) as usize] += n; + } + let oldest: Option = db.query_row("SELECT MIN(ts) FROM entries WHERE conn = ?1", params![conn], |r| r.get(0))?; + let new_groups = new_groups(db, conn, now, oldest)?; + Ok(ErrorSummary { by_service, last_hour, prev_hour, spark, oldest, new_groups, quiet: Vec::new() }) +} + +/// Services that went quiet: lines in at least half the minutes of two hours, +/// then none for ten minutes. Read from the per-minute counts, so it needs +/// two hours and ten minutes of counted history. +pub(super) fn quiet(db: &Connection, conn: &str, now: i64, minutes_since: i64) -> Result> { + const SILENT: i64 = 10; + const BASELINE: i64 = 120; + let cut = now.div_euclid(MINUTE) - SILENT; + let lo = cut - BASELINE; + if minutes_since > lo * MINUTE { + return Ok(Vec::new()); + } + let mut st = db.prepare_cached( + "SELECT service, SUM(minute < ?3), MAX(minute) FROM minutes + WHERE conn = ?1 AND service != '' AND minute >= ?2 AND debug + info + warn + error > 0 + GROUP BY service HAVING SUM(minute < ?3) >= ?4 AND MAX(minute) < ?3", + )?; + let rows = st.query_map(params![conn, lo, cut, BASELINE / 2], |r| { + Ok(Quiet { service: r.get(0)?, busy_minutes: r.get(1)?, since: (r.get::<_, i64>(2)? + 1) * MINUTE }) + })?; + let mut out = rows.collect::>>()?; + out.sort_by_key(|q| q.since); + Ok(out) +} + +pub(super) fn error_rate(db: &Connection, conn: &str, from: i64, to: i64) -> Result { + let err = Level::Error.as_i64(); + let count: u64 = db.query_row( + "SELECT COUNT(*) FROM entries WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4", + params![conn, from, to, err], + |r| r.get(0), + )?; + let mut rate = ErrorRate { count, ..Default::default() }; + if count == 0 { + return Ok(rate); + } + // The service comes from the group's newest line: a bare column next to + // GROUP BY would be any line's, and a group can span several services. + let top: Option<(String, i64)> = db + .query_row( + "SELECT fp, MAX(id) FROM entries WHERE conn = ?1 AND ts >= ?2 AND ts < ?3 AND level = ?4 AND fp IS NOT NULL + GROUP BY fp ORDER BY COUNT(*) DESC LIMIT 1", + params![conn, from, to, err], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .optional()?; + if let Some((fp, id)) = top { + let pattern: Option = + db.query_row("SELECT pattern FROM patterns WHERE conn = ?1 AND fp = ?2", params![conn, fp], |r| r.get(0)).optional()?; + let (service, sample): (String, String) = + db.query_row("SELECT service, text FROM entries WHERE id = ?1", params![id], |r| Ok((r.get(0)?, r.get(1)?)))?; + let text = pattern.filter(|p| !p.is_empty()).unwrap_or(sample); + rate.top = Some(text.lines().next().unwrap_or("").chars().take(200).collect()); + rate.top_service = Some(service); + } + Ok(rate) +} + +/// Lines and stored text per service of a connection, largest first. +pub(super) fn usage(db: &Connection, conn: &str) -> Result> { + let mut st = db.prepare( + "SELECT service, COUNT(*), SUM(LENGTH(text) + COALESCE(LENGTH(raw), 0)) AS b FROM entries WHERE conn = ?1 + GROUP BY service ORDER BY b DESC", + )?; + let rows = st.query_map(params![conn], |r| Ok(ServiceUsage { service: r.get(0)?, entries: r.get(1)?, text_bytes: r.get(2)? }))?; + Ok(rows.collect::>>()?) +} + +pub(super) fn row_to_entry(r: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(EntryOut { + id: r.get(0)?, + ts: r.get(1)?, + stack: r.get(2)?, + service: r.get(3)?, + replica: r.get(4)?, + node: r.get(5)?, + source: r.get(6)?, + stream: if r.get::<_, i64>(7)? == 2 { "stderr" } else { "stdout" }, + level: Level::from_i64(r.get(8)?), + exc: r.get::<_, i64>(9)? != 0, + lines: r.get(10)?, + fp: r.get(11)?, + text: r.get(12)?, + raw: r.get(13)?, + }) +} diff --git a/crates/stackline-core/src/store/retention.rs b/crates/stackline-core/src/store/retention.rs new file mode 100644 index 0000000..3989d06 --- /dev/null +++ b/crates/stackline-core/src/store/retention.rs @@ -0,0 +1,342 @@ +//! Keeping the store within its age and size limits, and clearing it. + +use super::*; + +/// How many lines each connection gives up so that `keep` lines are left: +/// the lines kept are shared out equally, and a connection with fewer lines +/// than its share keeps all of them, which leaves more for the others. +pub(super) fn over_share(counts: &[(String, u64)], keep: u64) -> Vec<(String, u64)> { + let mut sorted: Vec<&(String, u64)> = counts.iter().collect(); + sorted.sort_by_key(|(_, n)| *n); + let mut left = keep; + let mut out = Vec::new(); + for (i, (conn, n)) in sorted.iter().enumerate() { + let share = left / (sorted.len() - i) as u64; + let kept = (*n).min(share); + left -= kept; + if *n > kept { + out.push((conn.clone(), n - kept)); + } + } + out +} + +impl Store { + /// Deletes entries older than `before` (µs) and the oldest entries beyond + /// `max_bytes`, then gives the space back to the system. + pub fn retain(&self, before: i64, max_bytes: u64) -> Result { + const BATCH: usize = 20_000; + let mut deleted = 0u64; + // Per connection, so the (conn, ts) index finds old lines without + // reading the table. The lock is taken per batch, so queries and new + // lines get through between batches. + let conns = conns(&self.db.lock().unwrap())?; + for conn in conns { + loop { + let n = self.db.lock().unwrap().execute( + "DELETE FROM entries WHERE id IN (SELECT id FROM entries WHERE conn = ?1 AND ts < ?2 LIMIT ?3)", + params![conn, before, BATCH], + )?; + deleted += n as u64; + if n < BATCH { + break; + } + } + // Counts of minutes with no lines left. + self.db + .lock() + .unwrap() + .execute("DELETE FROM minutes WHERE conn = ?1 AND minute < ?2", params![conn, before.div_euclid(MINUTE)])?; + } + self.db.lock().unwrap().execute("DELETE FROM events WHERE ts < ?1", params![before])?; + // Over the size limit: drop the oldest lines, down to 90% of the + // limit so this does not run again a minute later. Every connection + // gets an equal share of what is kept, and only connections above + // their share give up lines: a chatty server must not push out the + // whole history of a quiet one. Deleted lines stay in the search index + // until it is merged, so the index is merged before the size is + // measured again. + for _ in 0..4 { + let live = live_bytes(&self.db.lock().unwrap())?; + if live <= max_bytes { + break; + } + let counts: Vec<(String, u64)> = { + let db = self.db.lock().unwrap(); + let mut st = db.prepare("SELECT conn, COUNT(*) FROM entries GROUP BY conn")?; + st.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?.collect::>()? + }; + let total: u64 = counts.iter().map(|(_, n)| n).sum(); + if total == 0 { + break; + } + let target = max_bytes / 10 * 9; + let left = ((total as u128 * (live - target) as u128).div_ceil(live as u128) as u64).clamp(1, total); + for (conn, mut drop) in over_share(&counts, total - left) { + while drop > 0 { + let n = self.db.lock().unwrap().execute( + "DELETE FROM entries WHERE id IN (SELECT id FROM entries WHERE conn = ?1 ORDER BY ts LIMIT ?2)", + params![conn, drop.min(BATCH as u64)], + )? as u64; + deleted += n; + drop = drop.saturating_sub(n); + if n == 0 { + break; + } + } + } + self.merge_index()?; + } + { + let db = self.db.lock().unwrap(); + let mode: i64 = db.query_row("PRAGMA auto_vacuum", [], |r| r.get(0))?; + if mode != 2 { + // Stores made before 0.3 could never shrink: switch them over once. + tracing::info!("rebuilding the log store once so it can give space back"); + db.execute_batch("PRAGMA auto_vacuum = INCREMENTAL; VACUUM;")?; + } + } + // Also when nothing was deleted now: up to 0.3.1 free space was never + // given back, so stores made by those versions shrink here once. + self.shrink()?; + Ok(deleted) + } + + /// Gives free pages back to the system and moves the write-ahead log + /// into the file, so the file shrinks on disk. In steps that release the + /// lock in between, so queries and new lines get through. + pub(super) fn shrink(&self) -> Result<()> { + const STEP: u64 = 4096; + let mut last = u64::MAX; + loop { + let db = self.db.lock().unwrap(); + let free: u64 = db.query_row("PRAGMA freelist_count", [], |r| r.get(0))?; + if free == 0 || free >= last { + return checkpoint(&db); + } + last = free; + // Each step of this pragma frees one page, so it is run to the end: + // run once, as `execute_batch` does, it freed a single page. + let mut st = db.prepare(&format!("PRAGMA incremental_vacuum({})", free.min(STEP)))?; + let mut rows = st.query([])?; + while rows.next()?.is_some() {} + } + } + + /// Merges the search index so deleted lines leave it, like FTS5's + /// `optimize`, but in steps that release the lock in between: on a large + /// store a full optimize would block queries and new lines for minutes. + pub(super) fn merge_index(&self) -> Result<()> { + loop { + let db = self.db.lock().unwrap(); + let before = db.total_changes(); + db.execute("INSERT INTO entries_fts (entries_fts, rank) VALUES ('merge', -500)", [])?; + if db.total_changes() - before <= 1 { + return Ok(()); + } + } + } + + /// Space on disk: the database file plus its write-ahead log (kept under 64 MB). + pub(super) fn used_bytes(&self, db: &Connection) -> Result { + let file = + |suffix: &str| self.path.as_ref().and_then(|p| std::fs::metadata(format!("{}{suffix}", p.display())).ok()).map(|m| m.len()); + match file("") { + Some(main) => Ok(main + file("-wal").unwrap_or(0)), + None => live_bytes(db), + } + } + + /// Lines and stored text per service of a connection, largest first. + pub fn usage(&self, conn: &str) -> Result> { + self.read(|db| usage(db, conn)) + } + + /// Deletes the stored lines of one service, in batches. + pub fn clear_service(&self, conn: &str, service: &str) -> Result { + let deleted = self.delete_entries(Some(conn), Some(service))?; + if deleted > 0 { + self.merge_index()?; + self.shrink()?; + } + Ok(deleted) + } + + /// Deletes the lines of a connection (all when None), or of one of its + /// services. Every deleted line also has to leave the search index, which + /// is the slow part, so it runs in small batches and releases the lock in + /// between. When nothing would be left, the table and the index are + /// emptied at once instead. + pub(super) fn delete_entries(&self, conn: Option<&str>, service: Option<&str>) -> Result { + const BATCH: i64 = 5_000; + { + let db = self.db.lock().unwrap(); + let everything = match (conn, service) { + (None, _) => true, + (Some(c), None) => conns(&db)? == [c], + (Some(_), Some(_)) => false, + }; + if everything { + let n: u64 = db.query_row("SELECT COUNT(*) FROM entries", [], |r| r.get(0))?; + db.execute_batch(&format!( + "BEGIN; + INSERT INTO entries_fts (entries_fts) VALUES ('delete-all'); + DROP TRIGGER entries_ad; + DELETE FROM entries; + DELETE FROM minutes; + UPDATE meta SET value = 0 WHERE key = 'minutes_since'; + {DELETE_TRIGGER} + COMMIT;" + ))?; + // Nothing older is left, so every range can use the counts. + self.minutes_since.store(0, std::sync::atomic::Ordering::Relaxed); + return Ok(n); + } + } + let (conn, mut deleted, mut after) = (conn.unwrap_or_default(), 0u64, 0i64); + loop { + let db = self.db.lock().unwrap(); + // Walks the rows in id order, so each batch starts where the last + // one stopped instead of reading the table from the start again. + let n = match service { + Some(svc) => { + let last: Option = db.query_row( + "SELECT MAX(id) FROM (SELECT id FROM entries WHERE id > ?1 AND conn = ?2 AND service = ?3 ORDER BY id LIMIT ?4)", + params![after, conn, svc, BATCH], + |r| r.get(0), + )?; + let Some(last) = last else { break }; + let n = db.execute( + "DELETE FROM entries WHERE id > ?1 AND id <= ?2 AND conn = ?3 AND service = ?4", + params![after, last, conn, svc], + )?; + after = last; + n + } + // The (conn, ts) index finds them; deleted rows leave it. + None => { + db.execute("DELETE FROM entries WHERE id IN (SELECT id FROM entries WHERE conn = ?1 LIMIT ?2)", params![conn, BATCH])? + } + }; + deleted += n as u64; + if service.is_none() && (n as i64) < BATCH { + break; + } + } + Ok(deleted) + } + + /// Counts the error entries of a connection in `[from, to)` (µs). + pub fn error_rate(&self, conn: &str, from: i64, to: i64) -> Result { + self.read(|db| error_rate(db, conn, from, to)) + } + + /// Checks which host or swarm a connection's logs come from against the + /// one remembered. The first one seen is remembered. A different one is + /// only reported: its lines must not mix with the old ones, and whether + /// the old ones go is the user's call ([`Store::accept_host`]). + pub fn check_host(&self, conn: &str, identity: &str) -> Result { + let db = self.db.lock().unwrap(); + let known: Option = db.query_row("SELECT identity FROM hosts WHERE conn = ?1", params![conn], |r| r.get(0)).optional()?; + match known { + Some(k) if k == identity => Ok(HostCheck::Same), + Some(_) => Ok(HostCheck::Changed), + None => { + db.execute("INSERT INTO hosts (conn, identity) VALUES (?1, ?2)", params![conn, identity])?; + Ok(HostCheck::First) + } + } + } + + /// Makes `identity` the host of a connection. With `keep_old` false the + /// lines from the host before are deleted first. + pub fn accept_host(&self, conn: &str, identity: &str, keep_old: bool) -> Result<()> { + if !keep_old { + self.clear(Some(conn))?; + } + self.db.lock().unwrap().execute( + "INSERT INTO hosts (conn, identity) VALUES (?1, ?2) ON CONFLICT (conn) DO UPDATE SET identity = excluded.identity", + params![conn, identity], + )?; + Ok(()) + } + + pub fn stats(&self) -> Result { + self.read(|db| { + let entries = db.query_row("SELECT COUNT(*) FROM entries", [], |r| r.get(0))?; + let mut oldest: Option = None; + for c in conns(db)? { + let t: Option = db.query_row("SELECT MIN(ts) FROM entries WHERE conn = ?1", params![c], |r| r.get(0))?; + oldest = oldest.into_iter().chain(t).min(); + } + Ok(StoreStats { bytes: self.used_bytes(db)?, entries, oldest }) + }) + } + + /// Changes whenever stored lines of a whole connection are cleared. + pub fn epoch(&self) -> u64 { + self.epoch.load(std::sync::atomic::Ordering::SeqCst) + } + + pub fn clear(&self, conn: Option<&str>) -> Result<()> { + // First, so lines a collector is reading right now are not stored + // after the clear with a cursor that would skip everything before them. + // Under the write lock: a batch being stored right now either lands + // before this and is deleted below, or sees the new epoch and is dropped. + { + let _db = self.db.lock().unwrap(); + self.epoch.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + let deleted = self.delete_entries(conn, None)?; + if deleted > 0 { + self.merge_index()?; + } + let db = self.db.lock().unwrap(); + match conn { + Some(c) => { + db.execute("DELETE FROM minutes WHERE conn = ?1", params![c])?; + db.execute("DELETE FROM patterns WHERE conn = ?1", params![c])?; + db.execute("DELETE FROM cursors WHERE conn = ?1", params![c])?; + db.execute("DELETE FROM hosts WHERE conn = ?1", params![c])?; + db.execute("DELETE FROM events WHERE conn = ?1", params![c])?; + db.execute("DELETE FROM deploy_state WHERE conn = ?1", params![c])?; + } + None => { + db.execute_batch("DELETE FROM minutes; DELETE FROM patterns; DELETE FROM cursors; DELETE FROM hosts; DELETE FROM events; DELETE FROM deploy_state;")? + } + } + drop(db); + self.shrink() + } +} + +/// The connections that have stored lines, found by skipping through the +/// index instead of reading every row. +pub(super) fn conns(db: &Connection) -> Result> { + let mut st = db.prepare_cached( + "WITH RECURSIVE c(conn) AS ( + SELECT MIN(conn) FROM entries + UNION ALL SELECT (SELECT MIN(conn) FROM entries WHERE conn > c.conn) FROM c WHERE c.conn IS NOT NULL + ) SELECT conn FROM c WHERE conn IS NOT NULL", + )?; + let rows = st.query_map([], |r| r.get(0))?; + Ok(rows.collect::>>()?) +} + +/// Moves the write-ahead log into the database and empties it. A read still +/// running (an export, say) keeps part of the log in use: then it is left +/// for the next time instead of waiting, which would hold up new lines. +pub(super) fn checkpoint(db: &Connection) -> Result<()> { + db.busy_timeout(std::time::Duration::ZERO)?; + let r = db.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);"); + db.busy_timeout(std::time::Duration::from_secs(5))?; + Ok(r?) +} + +/// Pages that hold data; free pages are given back by `incremental_vacuum`. +pub(super) fn live_bytes(db: &Connection) -> Result { + let pages: u64 = db.query_row("PRAGMA page_count", [], |r| r.get(0))?; + let free: u64 = db.query_row("PRAGMA freelist_count", [], |r| r.get(0))?; + let size: u64 = db.query_row("PRAGMA page_size", [], |r| r.get(0))?; + Ok(pages.saturating_sub(free) * size) +} diff --git a/crates/stackline-core/src/store/schema.rs b/crates/stackline-core/src/store/schema.rs new file mode 100644 index 0000000..ba6b309 --- /dev/null +++ b/crates/stackline-core/src/store/schema.rs @@ -0,0 +1,104 @@ +//! The tables, indexes and triggers of the store. + +/// Keeps the search index in step with new lines. Their per-minute counts +/// are added by [`Store::insert`](super::Store::insert), once per minute instead of once per line. +pub(super) const INSERT_TRIGGER: &str = "CREATE TRIGGER entries_ai AFTER INSERT ON entries BEGIN + INSERT INTO entries_fts (rowid, text) VALUES (new.id, new.text); +END;"; + +/// Adds to the counts of a minute: for the service, and for the whole connection (service ''). +pub(super) const ADD_MINUTE: &str = + "INSERT INTO minutes (conn, minute, service, stack, debug, info, warn, error, debug_exc, info_exc, warn_exc, error_exc) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) + ON CONFLICT (conn, minute, service) DO UPDATE SET + debug = debug + excluded.debug, info = info + excluded.info, warn = warn + excluded.warn, error = error + excluded.error, + debug_exc = debug_exc + excluded.debug_exc, info_exc = info_exc + excluded.info_exc, + warn_exc = warn_exc + excluded.warn_exc, error_exc = error_exc + excluded.error_exc"; + +/// Keeps the search index and the per-minute counts in step with deleted +/// lines. Deleting every line drops it for a moment, so the table can be +/// emptied at once. +pub(super) const DELETE_TRIGGER: &str = "CREATE TRIGGER entries_ad AFTER DELETE ON entries BEGIN + INSERT INTO entries_fts (entries_fts, rowid, text) VALUES ('delete', old.id, old.text); + UPDATE minutes SET + debug = debug - (old.level = 1), info = info - (old.level NOT IN (1, 3, 4)), + warn = warn - (old.level = 3), error = error - (old.level = 4), + debug_exc = debug_exc - (old.exc != 0 AND old.level = 1), info_exc = info_exc - (old.exc != 0 AND old.level NOT IN (1, 3, 4)), + warn_exc = warn_exc - (old.exc != 0 AND old.level = 3), error_exc = error_exc - (old.exc != 0 AND old.level = 4) + WHERE conn = old.conn AND minute = old.ts / 60000000 AND service IN (old.service, ''); +END;"; + +pub(super) const MINUTE: i64 = 60_000_000; + +pub(super) const SCHEMA: &str = r#" +PRAGMA auto_vacuum = INCREMENTAL; +PRAGMA journal_mode = WAL; +PRAGMA synchronous = NORMAL; +PRAGMA journal_size_limit = 67108864; +CREATE TABLE IF NOT EXISTS entries ( + id INTEGER PRIMARY KEY, + conn TEXT NOT NULL, + ts INTEGER NOT NULL, + stack TEXT NOT NULL, + service TEXT NOT NULL, + replica TEXT NOT NULL, + node TEXT NOT NULL, + source TEXT NOT NULL, + stream INTEGER NOT NULL, + level INTEGER NOT NULL, + exc INTEGER NOT NULL, + lines INTEGER NOT NULL, + fp TEXT, + text TEXT NOT NULL, + raw TEXT +); +-- Covers the time graph and level counts, so they never read the rows themselves. +CREATE INDEX IF NOT EXISTS entries_conn_ts_level ON entries (conn, ts, level, exc); +DROP INDEX IF EXISTS entries_conn_ts; +-- Only lines that belong to an error group, with what the group list needs. +CREATE INDEX IF NOT EXISTS entries_conn_ts_fp ON entries (conn, ts, fp, level, exc, service, node, source) WHERE fp IS NOT NULL; +CREATE INDEX IF NOT EXISTS entries_conn_fp ON entries (conn, fp, ts) WHERE fp IS NOT NULL; +CREATE VIRTUAL TABLE IF NOT EXISTS entries_fts USING fts5 (text, content = 'entries', content_rowid = 'id', tokenize = 'trigram'); +-- Lines per minute, service and level, and per minute of the whole +-- connection (service ''): the time graph and level counts of long ranges +-- read these instead of every line. +CREATE TABLE IF NOT EXISTS minutes ( + conn TEXT NOT NULL, + minute INTEGER NOT NULL, + service TEXT NOT NULL, + stack TEXT NOT NULL, + debug INTEGER NOT NULL, + info INTEGER NOT NULL, + warn INTEGER NOT NULL, + error INTEGER NOT NULL, + debug_exc INTEGER NOT NULL, + info_exc INTEGER NOT NULL, + warn_exc INTEGER NOT NULL, + error_exc INTEGER NOT NULL, + PRIMARY KEY (conn, minute, service) +) WITHOUT ROWID; +CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value INTEGER NOT NULL); +CREATE TABLE IF NOT EXISTS patterns (conn TEXT NOT NULL, fp TEXT NOT NULL, pattern TEXT NOT NULL, PRIMARY KEY (conn, fp)); +CREATE TABLE IF NOT EXISTS cursors (conn TEXT NOT NULL, source TEXT NOT NULL, ts INTEGER NOT NULL, PRIMARY KEY (conn, source)); +CREATE TABLE IF NOT EXISTS hosts (conn TEXT PRIMARY KEY, identity TEXT NOT NULL); +-- Container starts, stops and crashes; see events.rs. The columns the log +-- filters use are named as in `entries`, so the same filters apply. +CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY, + conn TEXT NOT NULL, + key TEXT NOT NULL, + ts INTEGER NOT NULL, + stack TEXT NOT NULL, + service TEXT NOT NULL, + replica TEXT NOT NULL, + node TEXT NOT NULL, + source TEXT NOT NULL, + kind TEXT NOT NULL, + exit_code INTEGER, + detail TEXT, + UNIQUE (conn, key) +); +CREATE INDEX IF NOT EXISTS events_conn_ts ON events (conn, ts); +-- What each service ran at its last deploy; see deploys.rs. +CREATE TABLE IF NOT EXISTS deploy_state (conn TEXT NOT NULL, service TEXT NOT NULL, ts INTEGER NOT NULL, config TEXT NOT NULL, PRIMARY KEY (conn, service)); +"#; From 5a96d11c9e9500a723190001469cd21549f8422a Mon Sep 17 00:00:00 2001 From: "snake.tech" <9079973+snaketech-tu@users.noreply.github.com> Date: Tue, 10 Nov 2026 02:03:41 +0000 Subject: [PATCH 2/3] Test a swarm stack in CI, and describe how Stackline keeps Docker access private - New CI job e2e-swarm: starts a single-node swarm, deploys the demo stack and checks replicas, global and crashing services, lines of every replica and whole stack traces (e2e/tests/swarm.spec.ts, project "swarm"). It runs on its own because the crashing service fills Docker's short event history, which the Compose tests read. - README: a Security section (reads only, private SSH tunnel, sign-in token and Host check, --host 0.0.0.0, signed updates, how to report). --- .github/workflows/ci.yml | 42 +++++++++++++++++++++++++++++++++++- README.md | 39 +++++++++++++++++++++++++++++++++- e2e/playwright.config.ts | 4 +++- e2e/tests/swarm.spec.ts | 46 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 128 insertions(+), 3 deletions(-) create mode 100644 e2e/tests/swarm.spec.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0ac8931..45e13ac 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -107,7 +107,7 @@ jobs: working-directory: e2e - name: Wait until the demo has logged every kind of error run: timeout 90 sh -c 'until docker logs demo-worker-1 2>&1 | grep -q import-catalog; do sleep 2; done' - - run: npx playwright test + - run: npx playwright test --project ui --project read-only working-directory: e2e - uses: actions/upload-artifact@v7 if: failure() @@ -117,3 +117,43 @@ jobs: e2e/playwright-report e2e/test-results retention-days: 7 + + e2e-swarm: + needs: ui + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/download-artifact@v8 + with: + name: ui-dist + path: ui/dist + - name: Start the swarm demo stack + working-directory: dev/demo + run: | + docker pull -q mirror.gcr.io/library/alpine:3.20 && docker tag mirror.gcr.io/library/alpine:3.20 alpine:3.20 || docker pull -q alpine:3.20 + docker swarm init + docker stack deploy --resolve-image never -c stack.yaml shop + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + - run: cargo build -p stackline + - uses: actions/setup-node@v7 + with: + node-version: 24 + cache: npm + cache-dependency-path: e2e/package-lock.json + - run: npm ci + working-directory: e2e + - run: npx playwright install --with-deps chromium + working-directory: e2e + - name: Wait until every replica runs + run: timeout 120 sh -c 'until docker service ls --filter name=shop_api --format "{{.Replicas}}" | grep -q "3/3"; do sleep 2; done' + - run: npx playwright test --project swarm + working-directory: e2e + - uses: actions/upload-artifact@v7 + if: failure() + with: + name: e2e-swarm-report + path: | + e2e/playwright-report + e2e/test-results + retention-days: 7 diff --git a/README.md b/README.md index b07ef4b..99c2d41 100644 --- a/README.md +++ b/README.md @@ -231,6 +231,38 @@ When an update is available, Stackline shows the release notes and lets you choose when to install it. Every update is verified against the app's built-in public key before installation. +## Security + +Stackline can reach the Docker of every server you connect, and whoever +controls a server's Docker controls the server. This is how it keeps that +access to you: + +- **Only reads by default.** Stackline sends Docker requests that read: + lists, logs and events. Log downloads are files written on your computer. +- **SSH connections stay private.** Stackline reaches a server's Docker + socket through your own `ssh`. On macOS and Linux the tunnel ends in a + socket file in a folder only your user can open, which is removed when + Stackline stops. On Windows it still ends in a port on `127.0.0.1` that + other programs on the computer could use, so do not run Stackline on a + Windows machine you share with people you do not trust. Your private keys + are never read or copied. +- **Only your browser tab can use it.** The standalone server listens on + `127.0.0.1`. Every API call needs the random sign-in token from the link it + prints; the link stores it in a cookie that only Stackline's own pages send + (`HttpOnly`, `SameSite=Strict`). Requests addressed to any other name than + `localhost` or `127.0.0.1` are refused, which keeps web pages from reaching + Stackline through DNS rebinding. +- **`--host 0.0.0.0` makes it reachable from the network**, for anyone who has + the link. The connection is plain HTTP, so the token can be read on the way. + Only do this on a network you trust, or put Stackline behind a reverse proxy + that adds TLS. Stackline prints a warning when it listens beyond loopback. +- **Updates are signed.** An update is installed only when its signature + matches the public key built into the app. + +To report a security problem, open a private +[security advisory](https://github.com/snaketech-tu/stackline/security/advisories/new) +on GitHub instead of an issue. + ## Development You need **Rust stable** and **Node.js 24**. From a fresh clone: @@ -315,13 +347,18 @@ start the demo project, build the UI and Stackline, then: cd e2e npm ci npx playwright install chromium -npx playwright test +npx playwright test --project ui --project read-only ``` They start their own Stackline on port 7171 with an empty data folder (`e2e/.run`), so your own connections and stored logs are not touched. Set `STACKLINE_BIN` to test another build, such as `target/release/stackline`. +The [swarm tests](e2e/tests/swarm.spec.ts) need the demo Swarm stack instead +(`docker swarm init`, then `docker stack deploy -c stack.yaml shop` in +`dev/demo`) and run with `npx playwright test --project swarm`. CI runs them in +a job of their own. + ## License Licensed under [Apache-2.0](LICENSE). diff --git a/e2e/playwright.config.ts b/e2e/playwright.config.ts index f08b6ad..ae8e360 100644 --- a/e2e/playwright.config.ts +++ b/e2e/playwright.config.ts @@ -20,9 +20,11 @@ export default defineConfig({ screenshot: 'only-on-failure', }, projects: [ - { name: 'ui', testIgnore: /read-only/, use: { ...devices['Desktop Chrome'], viewport: { width: 1400, height: 900 } } }, + { name: 'ui', testIgnore: /read-only|swarm\.spec/, use: { ...devices['Desktop Chrome'], viewport: { width: 1400, height: 900 } } }, // Last, so it sees every Docker call the other tests caused. { name: 'read-only', testMatch: /read-only/, dependencies: ['ui'] }, + // Only with the swarm stack of dev/demo deployed: `--project swarm`, as the CI swarm job runs it. + { name: 'swarm', testMatch: /swarm\.spec/ }, ], webServer: { command: 'node serve.mjs', diff --git a/e2e/tests/swarm.spec.ts b/e2e/tests/swarm.spec.ts new file mode 100644 index 0000000..344e22b --- /dev/null +++ b/e2e/tests/swarm.spec.ts @@ -0,0 +1,46 @@ +import { expect, test } from '@playwright/test' +import { TOKEN } from '../playwright.config' + +// Needs the swarm stack of dev/demo/stack.yaml deployed as "shop": `docker +// swarm init`, then `docker stack deploy -c stack.yaml shop` in dev/demo. Run +// with `npx playwright test --project swarm`; CI does this in its own job, +// because the crashing service of the stack fills Docker's short event +// history and would push out the events the other tests look for. + +const headers = { 'x-stackline-token': TOKEN } +const hour = () => { + const to = Date.now() + 60e3 + return `from=${to - 3600e3}&to=${to}` +} + +type Service = { fullName: string; mode: string; desired: number; running: number } + +test('a swarm stack shows its services with their mode and replicas', async ({ request }) => { + await expect(async () => { + const { snapshot } = await (await request.get('/api/c/local/snapshot', { headers })).json() + expect(snapshot.mode).toBe('swarm') + const shop: Service[] = snapshot.stacks.find((s: { name: string }) => s.name === 'shop').services + const by = Object.fromEntries(shop.map((s) => [s.fullName, s])) + expect(by.shop_api).toMatchObject({ mode: 'replicated', desired: 3, running: 3 }) + expect(by.shop_agent).toMatchObject({ mode: 'global', running: 1 }) + expect(by.shop_broken).toMatchObject({ desired: 1, running: 0 }) + }).toPass({ timeout: 30_000 }) +}) + +test('lines of every replica are stored, stack traces in one piece', async ({ request }) => { + await expect(async () => { + const r = await request.get(`/api/c/local/logs?service=shop_api&${hour()}&limit=500`, { headers }) + const { entries } = (await r.json()).result as { entries: { replica: string; text: string }[] } + expect(new Set(entries.map((e) => e.replica))).toEqual(new Set(['1', '2', '3'])) + const trace = entries.find((e) => e.text.includes('NullPointerException')) + expect(trace?.text).toMatch(/ERROR[\s\S]*\n\s+at com\.acme\./) + }).toPass({ timeout: 60_000 }) +}) + +test('a service that keeps failing has its crashes and exit code listed', async ({ request }) => { + await expect(async () => { + const r = await request.get(`/api/c/local/events?service=shop_broken&${hour()}&limit=100`, { headers }) + const { events } = await r.json() + expect(events.some((e: { kind: string; exitCode: number | null }) => e.kind === 'crash' && e.exitCode === 1)).toBe(true) + }).toPass({ timeout: 60_000 }) +}) From ac07bfd87a6a23b59bb2743c96bdda0e6dba617d Mon Sep 17 00:00:00 2001 From: "snake.tech" <9079973+snaketech-tu@users.noreply.github.com> Date: Tue, 10 Nov 2026 02:03:47 +0000 Subject: [PATCH 3/3] Link the Security section from the top of the README --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 99c2d41..a650446 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ that opens the dashboard in your browser. [Download](https://github.com/snaketech-tu/stackline/releases/latest) · [Get started](#get-started) · [Connect a host](#connect-a-host) · -[Development](#development) · [Changelog](CHANGELOG.md) +[Security](#security) · [Development](#development) · [Changelog](CHANGELOG.md) ## Features