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/CHANGELOG.md b/CHANGELOG.md index c24d9bf..ddddffd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,24 @@ Release notes for each version. The section of a version becomes the text of its GitHub release, and Stackline shows it in the update dialog. +## Unreleased + +### Security +- **SSH connections no longer open a port on this computer.** Stackline reached a server's Docker through a local TCP port that any program on the computer could also use, and a web page could try to. Whoever reaches that port controls the server's Docker. On macOS and Linux the connection now goes through a socket file in a folder only you can open, removed when Stackline stops. Windows still uses a local port for now + +### Changed +- **A connection that reaches another server no longer deletes its logs by itself.** When the server behind a connection changes (its address was edited, an SSH alias points elsewhere, the swarm was set up again), Stackline stops collecting and asks: keep the old logs, or delete them. Before, they were deleted without asking +- **At the size limit, a busy server no longer pushes out the logs of a quiet one.** Each connection gets an equal share of the space, and only connections above their share lose their oldest lines + +### Fixed +- **"Clear stored logs" during a read could leave part of that read behind**, and those lines then kept the cleared range from filling again +- **A replica read for the first time held up all the others.** Reading hours of a busy replica can take minutes; other replicas and container events now go on being read meanwhile +- **Deploys could go missing from the Overview** when a service crashed thousands of times after them +- **`--host 0.0.0.0` answered nobody.** Other machines were turned away because they address Stackline by an IP address it did not know. Any IP address of the machine now works; names are still refused, which keeps DNS rebinding out. Stackline also warns that the link travels unencrypted +- **A settings file that could not be read kept Stackline from starting.** It is now set aside as `config.broken.json`, Stackline starts with the default settings and says so +- **A log line could run as a formula** when a CSV download was opened in a spreadsheet app. Cells that start with `=`, `+`, `-` or `@` now start with `'` +- **A Docker that accepted a request and then never answered stopped log collection for good.** Requests now give up after 30 seconds without an answer and are tried again on the next poll + ## 0.5.1 Logs no longer end without warning: the stream pages through every stored line, and long lines are stored whole. diff --git a/README.md b/README.md index b07ef4b..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 @@ -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/crates/stackline-core/src/collector.rs b/crates/stackline-core/src/collector.rs index a0e0916..69ab72a 100644 --- a/crates/stackline-core/src/collector.rs +++ b/crates/stackline-core/src/collector.rs @@ -53,6 +53,24 @@ pub struct Collector { store: Arc, states: HashMap, epoch: u64, + /// First reads under way, by source. They run apart from the regular + /// polls: reading hours of a busy replica can take minutes, and the other + /// replicas and the events must not wait for it. + first: HashMap>, +} + +/// How many first reads of one connection run at once. +const FIRST_READS: usize = 2; +/// How long a poll waits for first reads to finish, so that a quick one +/// (most of them) is stored by the poll that started it. +const FIRST_WAIT: Duration = Duration::from_secs(2); + +impl Drop for Collector { + fn drop(&mut self) { + for (_, h) in self.first.drain() { + h.abort(); + } + } } #[derive(Debug, Default, Clone)] @@ -65,7 +83,7 @@ pub struct PollReport { impl Collector { pub fn new(conn: &str, store: Arc) -> Self { let epoch = store.epoch(); - Self { conn: conn.to_string(), store, states: HashMap::new(), epoch } + Self { conn: conn.to_string(), store, states: HashMap::new(), epoch, first: HashMap::new() } } /// Reads new lines from every source. The first read of a new source goes @@ -79,10 +97,18 @@ impl Collector { if refill { self.states.clear(); self.epoch = epoch; + // What they read belongs to the lines just cleared. + for (_, h) in self.first.drain() { + h.abort(); + } } + let started = Instant::now(); let keys: std::collections::HashSet = sources.iter().map(|s| s.key.clone()).collect(); self.states.retain(|k, _| keys.contains(k)); for src in sources { + if self.first.contains_key(&src.key) { + continue; + } let st = self.states.entry(src.key.clone()).or_insert_with(|| SourceState { assembler: Assembler::new(&src.service), cursor: None, @@ -112,48 +138,90 @@ impl Collector { .filter(|(_, s)| s.src.running || s.last_read.is_none_or(|t| t.elapsed() > Duration::from_secs(60))) .map(|(k, _)| k.clone()) .collect(); - let mut taken: Vec = due.iter().filter_map(|k| self.states.remove(k)).collect(); + let taken: Vec = due.iter().filter_map(|k| self.states.remove(k)).collect(); let mut report = PollReport { sources: taken.len(), ..Default::default() }; + let (fresh, known): (Vec, Vec) = taken.into_iter().partition(|st| st.cursor.is_none()); + + // First reads start in the background, a few at a time; the rest wait for a later poll. + for st in fresh { + if self.first.len() >= FIRST_READS { + self.states.insert(st.src.key.clone(), st); + continue; + } + let (conn, store, client) = (self.conn.clone(), self.store.clone(), client.clone()); + let key = st.src.key.clone(); + self.first.insert(key, tokio::spawn(read_one(client, store, conn, st, now, retention_us))); + } let conn = self.conn.clone(); let store = self.store.clone(); - let results: Vec<(SourceState, usize)> = futures_util::stream::iter(taken.drain(..)) - .map(|mut st| { - let (conn, store, client) = (conn.clone(), store.clone(), client.clone()); - async move { - let n = match read_source(&client, &store, &conn, &mut st, now, retention_us).await { - Ok(n) => { - st.error = None; - n - } - Err(e) => { - st.error = Some(format!("{}: {e:#}", st.src.service)); - // Batches stored before the error stay stored: go on - // after them, or the next poll stores them twice. - // A half-built entry is dropped for the same reason. - if st.stored > st.cursor { - st.cursor = st.stored; - } - st.assembler = Assembler::new(&st.src.service); - 0 - } - }; - st.last_read = Some(Instant::now()); - (st, n) - } - }) + let results: Vec<(SourceState, usize)> = futures_util::stream::iter(known) + .map(|st| read_one(client.clone(), store.clone(), conn.clone(), st, now, retention_us)) .buffer_unordered(6) .collect() .await; for (st, n) in results { - report.entries += n; - if let Some(e) = &st.error { - report.errors.push(e.clone()); + self.done(st, n, &mut report); + } + + // First reads that finished, waiting a moment for quick ones. + let deadline = started + FIRST_WAIT; + let keys: Vec = self.first.keys().cloned().collect(); + for key in keys { + let Some(h) = self.first.get_mut(&key) else { continue }; + match tokio::time::timeout_at(deadline.into(), h).await { + Err(_) => continue, + Ok(r) => { + self.first.remove(&key); + // A read that panicked is started again from the stored cursor. + if let Ok((st, n)) = r + && st.epoch == self.epoch + { + self.done(st, n, &mut report); + } + } } - self.states.insert(st.src.key.clone(), st); } report } + + fn done(&mut self, st: SourceState, n: usize, report: &mut PollReport) { + report.entries += n; + if let Some(e) = &st.error { + report.errors.push(e.clone()); + } + self.states.insert(st.src.key.clone(), st); + } +} + +/// Reads one source and says how many entries it stored. +async fn read_one( + client: DockerClient, + store: Arc, + conn: String, + mut st: SourceState, + now: i64, + retention_us: i64, +) -> (SourceState, usize) { + let n = match read_source(&client, &store, &conn, &mut st, now, retention_us).await { + Ok(n) => { + st.error = None; + n + } + Err(e) => { + st.error = Some(format!("{}: {e:#}", st.src.service)); + // Batches stored before the error stay stored: go on + // after them, or the next poll stores them twice. + // A half-built entry is dropped for the same reason. + if st.stored > st.cursor { + st.cursor = st.stored; + } + st.assembler = Assembler::new(&st.src.service); + 0 + } + }; + st.last_read = Some(Instant::now()); + (st, n) } async fn read_source( @@ -306,19 +374,11 @@ async fn flush(store: &Arc, conn: &str, st: &mut SourceState, entries: &m (t, _) => t, }; let (store, conn, src, epoch) = (store.clone(), conn.to_string(), st.src.clone(), st.epoch); - tokio::task::spawn_blocking(move || -> Result<()> { - // The store was cleared during this read: the next poll starts over. - if store.epoch() != epoch { - return Ok(()); - } - store.insert(&conn, &src, &batch)?; - if let Some(ts) = last_ts { - store.set_cursor(&conn, &src.key, ts)?; - } - Ok(()) - }) - .await??; - st.stored = last_ts.or(st.stored); + // A store cleared during this read keeps nothing of it: the next poll starts over. + let stored = tokio::task::spawn_blocking(move || store.store_batch(&conn, &src, &batch, last_ts, epoch)).await??; + if stored { + st.stored = last_ts.or(st.stored); + } Ok(n) } @@ -485,6 +545,69 @@ mod tests { assert_eq!(stored(&store), 5); } + #[tokio::test] + async fn a_slow_first_read_does_not_hold_up_the_other_replicas() { + // Answers every log request at once with one new line, except those of + // the replica `slow`, which take 8 seconds: a first read of hours of logs. + // Two polls wait at most 2 × FIRST_WAIT for first reads, well short of that. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let line = Arc::new(Mutex::new(0i64)); + let l = line.clone(); + tokio::spawn(async move { + loop { + let Ok((mut sock, _)) = listener.accept().await else { return }; + let l = l.clone(); + tokio::spawn(async move { + let mut buf = Vec::new(); + let mut chunk = [0u8; 1024]; + while !buf.windows(4).any(|w| w == b"\r\n\r\n") { + match sock.read(&mut chunk).await { + Ok(0) | Err(_) => return, + Ok(n) => buf.extend_from_slice(&chunk[..n]), + } + } + let head = String::from_utf8_lossy(&buf).to_string(); + if head.contains("/slow/") { + tokio::time::sleep(Duration::from_secs(8)).await; + } + let ts = *l.lock().unwrap(); + let body = frames(&[(ts, "INFO hello".to_string())], "/"); + let res = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()); + let _ = sock.write_all(res.as_bytes()).await; + let _ = sock.write_all(&body).await; + }); + } + }); + let client = DockerClient::new(Endpoint::Tcp(addr)); + let store = Arc::new(Store::open_in_memory().unwrap()); + let now = chrono::Utc::now().timestamp_micros(); + let fast = LogSource { key: "fast".into(), ..source() }; + let slow = LogSource { key: "slow".into(), service: "shop_worker".into(), ..source() }; + let mut c = Collector::new("conn", store.clone()); + + *line.lock().unwrap() = now - 10_000_000; + let first = Instant::now(); + let started = first; + let r = c.poll(&client, vec![fast.clone(), slow.clone()], RETENTION).await; + assert!(started.elapsed() < Duration::from_secs(3), "the poll waited {:?} for the slow first read", started.elapsed()); + assert_eq!(r.entries, 1, "{r:?}"); + + // The next polls read the fast replica again while the slow one is still busy. + *line.lock().unwrap() = now - 5_000_000; + let started = Instant::now(); + let r = c.poll(&client, vec![fast.clone(), slow.clone()], RETENTION).await; + assert!(started.elapsed() < Duration::from_secs(3)); + assert_eq!(r.entries, 1, "{r:?}"); + + // Once done, the slow replica's lines are stored and it is polled like any other. + let left = Duration::from_secs(9).saturating_sub(first.elapsed()); + tokio::time::sleep(left).await; + c.poll(&client, vec![fast, slow], RETENTION).await; + assert!(store.cursor("conn", "slow").unwrap().is_some()); + assert_eq!(stored(&store), 3); + } + #[tokio::test] async fn a_new_replica_is_read_6_hours_back_and_a_clear_refills_the_retention() { let docker = FakeDocker::start().await; diff --git a/crates/stackline-core/src/config.rs b/crates/stackline-core/src/config.rs index c9b4fae..5931531 100644 --- a/crates/stackline-core/src/config.rs +++ b/crates/stackline-core/src/config.rs @@ -129,6 +129,31 @@ impl AppConfig { } } + /// Like [`AppConfig::load`], but a file that cannot be read as settings + /// (edited by hand, say) does not keep Stackline from starting: it is + /// moved aside as `config.broken.json` and the defaults are used. The + /// second value says what happened, for the UI to show. + pub fn load_or_set_aside(path: &Path) -> Result<(AppConfig, Option)> { + let bytes = match std::fs::read(path) { + Ok(b) => b, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((AppConfig::default(), None)), + Err(e) => return Err(e.into()), + }; + match serde_json::from_slice(&bytes) { + Ok(c) => Ok((c, None)), + Err(e) => { + let aside = path.with_file_name("config.broken.json"); + std::fs::rename(path, &aside)?; + tracing::warn!("{} could not be read ({e}); moved to {}", path.display(), aside.display()); + let msg = format!( + "Your settings file could not be read ({e}), so Stackline started with the default settings. The old file was kept as {}.", + aside.display() + ); + Ok((AppConfig::default(), Some(msg))) + } + } + } + pub fn save(&self, path: &Path) -> Result<()> { if let Some(dir) = path.parent() { std::fs::create_dir_all(dir)?; @@ -255,6 +280,24 @@ mod tests { assert!(!base.exists()); } + #[test] + fn a_broken_settings_file_is_set_aside_not_fatal() { + let dir = std::env::temp_dir().join(format!("stackline-broken-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("config.json"); + std::fs::write(&path, r#"{"connections": [ oops"#).unwrap(); + let (c, warning) = AppConfig::load_or_set_aside(&path).unwrap(); + assert_eq!(c.connections.len(), 1); + assert!(warning.unwrap().contains("config.broken.json")); + assert!(!path.exists()); + assert_eq!(std::fs::read_to_string(dir.join("config.broken.json")).unwrap(), r#"{"connections": [ oops"#); + // A good file loads as it is, with nothing to say. + AppConfig::default().save(&path).unwrap(); + assert!(AppConfig::load_or_set_aside(&path).unwrap().1.is_none()); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn collect_lists_exceptions_or_choices() { let all = Collect::default(); diff --git a/crates/stackline-core/src/docker/client.rs b/crates/stackline-core/src/docker/client.rs index ca6c339..43f9884 100644 --- a/crates/stackline-core/src/docker/client.rs +++ b/crates/stackline-core/src/docker/client.rs @@ -17,6 +17,9 @@ use serde::de::DeserializeOwned; use super::endpoint::Endpoint; const JSON_TIMEOUT: Duration = Duration::from_secs(20); +/// How long a request may wait for the response headers. Log and event +/// streams send them at once and the body later. +const HEAD_TIMEOUT: Duration = Duration::from_secs(30); #[derive(Clone, Debug)] pub struct DockerClient { @@ -33,7 +36,16 @@ impl DockerClient { } /// Sends a GET request and returns the response with its body still streaming. + /// Fails when Docker does not answer with headers within [`HEAD_TIMEOUT`]: + /// a daemon or tunnel that accepts the connection and then hangs would + /// otherwise stop log collection for good. pub async fn get(&self, path_and_query: &str) -> Result> { + tokio::time::timeout(HEAD_TIMEOUT, self.get_inner(path_and_query)) + .await + .map_err(|_| anyhow!("Docker did not start answering {path_and_query} within {HEAD_TIMEOUT:?}"))? + } + + async fn get_inner(&self, path_and_query: &str) -> Result> { let stream = self.endpoint.connect().await.with_context(|| format!("could not reach Docker at {}", self.endpoint.describe()))?; let (mut sender, conn) = hyper::client::conn::http1::handshake(TokioIo::new(stream)).await.context("HTTP handshake with Docker failed")?; diff --git a/crates/stackline-core/src/docker/ssh.rs b/crates/stackline-core/src/docker/ssh.rs index 0d370d4..b0b7efc 100644 --- a/crates/stackline-core/src/docker/ssh.rs +++ b/crates/stackline-core/src/docker/ssh.rs @@ -1,9 +1,14 @@ //! SSH access to a remote Docker host. //! //! Stackline runs your system `ssh` so that `~/.ssh/config`, the agent, keys and -//! ProxyJump all work exactly as in a terminal. One SSH connection forwards a -//! local TCP port to the remote Docker socket; every API request then goes -//! through that single connection. Nothing is installed or opened on the server. +//! ProxyJump all work exactly as in a terminal. One SSH connection forwards the +//! remote Docker socket; every API request then goes through that single +//! connection. Nothing is installed or opened on the server. +//! +//! Whoever can reach the local end of the forward controls the server's Docker, +//! so on macOS and Linux it is a socket file in a folder only this user can +//! open, never a TCP port that every program on the computer (or a web page in +//! a browser) could connect to. Windows still forwards a loopback port. use std::process::Stdio; use std::sync::{Arc, Mutex}; @@ -14,6 +19,8 @@ use serde::{Deserialize, Serialize}; use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::process::{Child, Command}; +use super::Endpoint; + #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct SshTarget { /// `host`, `user@host` or an alias from `~/.ssh/config`. @@ -37,14 +44,54 @@ pub fn default_socket() -> String { pub struct SshTunnel { child: Child, - local_addr: String, + local: Local, stderr: Arc>, } +/// The local end of the forward. +enum Local { + /// A socket file in a private folder, removed with the tunnel. + #[cfg(unix)] + Socket(PrivateDir), + #[cfg(not(unix))] + Port(String), +} + +impl Local { + fn new() -> Result { + #[cfg(unix)] + { + Ok(Local::Socket(PrivateDir::new()?)) + } + #[cfg(not(unix))] + { + Ok(Local::Port(format!("127.0.0.1:{}", free_port()?))) + } + } + + /// What `ssh -L` listens on. + fn listen_spec(&self) -> String { + match self { + #[cfg(unix)] + Local::Socket(d) => d.socket().display().to_string(), + #[cfg(not(unix))] + Local::Port(a) => a.clone(), + } + } + + fn endpoint(&self) -> Endpoint { + match self { + #[cfg(unix)] + Local::Socket(d) => Endpoint::Unix(d.socket()), + #[cfg(not(unix))] + Local::Port(a) => Endpoint::Tcp(a.clone()), + } + } +} + impl SshTunnel { pub async fn open(t: &SshTarget) -> Result { - let port = free_port()?; - let local_addr = format!("127.0.0.1:{port}"); + let local = Local::new()?; let mut cmd = Command::new(ssh_binary()); cmd.args([ "-N", @@ -69,7 +116,10 @@ impl SshTunnel { if let Some(j) = t.jump.as_deref().filter(|s| !s.is_empty()) { cmd.arg("-J").arg(j); } - cmd.arg("-L").arg(format!("{local_addr}:{}", t.socket)); + #[cfg(unix)] + // Replace a stale socket file, and keep the new one to this user. + cmd.args(["-o", "StreamLocalBindUnlink=yes", "-o", "StreamLocalBindMask=0177"]); + cmd.arg("-L").arg(format!("{}:{}", local.listen_spec(), t.socket)); cmd.arg(&t.target); cmd.stdin(Stdio::null()).stdout(Stdio::null()).stderr(Stdio::piped()).kill_on_drop(true); cmd.env_remove("STACKLINE_TOKEN").env_remove("SWARMDASH_TOKEN"); @@ -104,7 +154,7 @@ impl SshTunnel { let msg = stderr.lock().unwrap().trim().to_string(); bail!("ssh to {} exited ({status}){}", t.target, if msg.is_empty() { String::new() } else { format!(": {msg}") }); } - if tokio::net::TcpStream::connect(&local_addr).await.is_ok() { + if local.endpoint().connect().await.is_ok() { break; } if started.elapsed() > Duration::from_secs(30) { @@ -112,11 +162,12 @@ impl SshTunnel { } tokio::time::sleep(Duration::from_millis(150)).await; } - Ok(SshTunnel { child, local_addr, stderr }) + Ok(SshTunnel { child, local, stderr }) } - pub fn local_addr(&self) -> &str { - &self.local_addr + /// Where the Docker API of the server is reached through this tunnel. + pub fn endpoint(&self) -> Endpoint { + self.local.endpoint() } pub fn is_alive(&mut self) -> bool { @@ -126,6 +177,10 @@ impl SshTunnel { /// Stops ssh right away, e.g. before Stackline exits or restarts. pub fn kill(&mut self) { let _ = self.child.start_kill(); + #[cfg(unix)] + let Local::Socket(dir) = &self.local; + #[cfg(unix)] + let _ = std::fs::remove_dir_all(&dir.0); } /// The last thing ssh printed, e.g. a permission error on the remote socket. @@ -135,6 +190,54 @@ impl SshTunnel { } } +/// A folder only this user can open, holding the tunnel's socket. Removed +/// when the tunnel goes away. +#[cfg(unix)] +struct PrivateDir(std::path::PathBuf); + +#[cfg(unix)] +impl PrivateDir { + fn new() -> Result { + use std::os::unix::fs::DirBuilderExt; + use std::sync::atomic::{AtomicU64, Ordering}; + static NEXT: AtomicU64 = AtomicU64::new(0); + // The runtime folder is private already; the temporary folder is shared, + // so the new folder must not exist yet (no one can have prepared it). + let base = dirs::runtime_dir().filter(|d| d.is_dir()).unwrap_or_else(std::env::temp_dir); + let nanos = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.subsec_nanos()).unwrap_or(0); + for _ in 0..20 { + let n = NEXT.fetch_add(1, Ordering::Relaxed); + let dir = base.join(format!("stackline-{}-{nanos:x}-{n}", std::process::id())); + match std::fs::DirBuilder::new().mode(0o700).create(&dir) { + Ok(()) => { + let d = PrivateDir(dir); + // A socket path longer than about 100 bytes cannot be bound (macOS: 104). + let len = d.socket().as_os_str().len(); + if len > 100 { + bail!("the folder for the SSH socket has too long a path ({len} bytes): {}", d.0.display()); + } + return Ok(d); + } + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(e) => return Err(e).with_context(|| format!("could not create a folder in {}", base.display())), + } + } + bail!("could not create a private folder in {}", base.display()) + } + + fn socket(&self) -> std::path::PathBuf { + self.0.join("docker.sock") + } +} + +#[cfg(unix)] +impl Drop for PrivateDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +#[cfg(not(unix))] fn free_port() -> Result { let l = std::net::TcpListener::bind("127.0.0.1:0")?; Ok(l.local_addr()?.port()) @@ -153,3 +256,70 @@ fn expand_tilde(p: &str) -> String { } p.to_string() } + +#[cfg(all(test, unix))] +mod tests { + use super::*; + use crate::docker::DockerClient; + + /// Stands in for `ssh`: serves the `-L` socket and answers every request + /// like Docker's `/version`, and writes down the arguments it got. + const FAKE_SSH: &str = r#"#!/usr/bin/env python3 +import os, socket, sys +args = sys.argv[1:] +open(os.environ["FAKE_SSH_ARGS"], "w").write("\n".join(args)) +local = args[args.index("-L") + 1].rsplit(":", 1)[0] +s = socket.socket(socket.AF_UNIX) +s.bind(local) +s.listen(8) +while True: + c, _ = s.accept() + data = b"" + while b"\r\n\r\n" not in data: + chunk = c.recv(4096) + if not chunk: + break + data += chunk + body = b'{"Version":"fake"}' + try: + if data: + c.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: %d\r\n\r\n%s" % (len(body), body)) + except OSError: + pass + c.close() +"#; + + #[tokio::test] + async fn the_tunnel_is_a_private_socket_not_a_port() { + use std::os::unix::fs::PermissionsExt; + let dir = std::env::temp_dir().join(format!("stackline-fake-ssh-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let ssh = dir.join("ssh"); + std::fs::write(&ssh, FAKE_SSH).unwrap(); + std::fs::set_permissions(&ssh, std::fs::Permissions::from_mode(0o755)).unwrap(); + let args_file = dir.join("args"); + // Only this test runs ssh, so changing the environment is safe here. + unsafe { + std::env::set_var("STACKLINE_SSH", &ssh); + std::env::set_var("FAKE_SSH_ARGS", &args_file); + } + let target = SshTarget { target: "me@example".into(), port: None, identity: None, jump: None, socket: default_socket() }; + let mut tunnel = SshTunnel::open(&target).await.unwrap(); + + let Endpoint::Unix(sock) = tunnel.endpoint() else { panic!("an SSH tunnel must not listen on a TCP port") }; + let folder = sock.parent().unwrap().to_path_buf(); + assert_eq!(std::fs::metadata(&folder).unwrap().permissions().mode() & 0o777, 0o700); + let args = std::fs::read_to_string(&args_file).unwrap(); + assert!(args.contains(&format!("{}:/var/run/docker.sock", sock.display())), "{args}"); + assert!(args.contains("StreamLocalBindMask=0177") && args.contains("BatchMode=yes"), "{args}"); + + let v: serde_json::Value = DockerClient::new(tunnel.endpoint()).get_json("/version").await.unwrap(); + assert_eq!(v["Version"], "fake"); + + tunnel.kill(); + drop(tunnel); + assert!(!folder.exists(), "the socket folder is removed with the tunnel"); + let _ = std::fs::remove_dir_all(&dir); + } +} diff --git a/crates/stackline-core/src/events.rs b/crates/stackline-core/src/events.rs index 909c8e7..8ce9686 100644 --- a/crates/stackline-core/src/events.rs +++ b/crates/stackline-core/src/events.rs @@ -132,16 +132,9 @@ impl EventPoller { events.extend(snap.task_failures.iter().filter(|t| t.node != names.local || t.container.is_empty()).map(from_task)); let (store, conn) = (self.store.clone(), self.conn.clone()); - tokio::task::spawn_blocking(move || { - // A clear while reading: what was read belongs to the lines just deleted. - if store.epoch() != epoch { - return Ok(0); - } - let n = store.insert_events(&conn, &events)?; - store.set_cursor(&conn, CURSOR, now)?; - Ok(n) - }) - .await? + // A clear while reading: what was read belongs to the lines just deleted. + let n = tokio::task::spawn_blocking(move || store.store_events(&conn, &events, Some(now), epoch)).await??; + Ok(n.unwrap_or(0)) } } diff --git a/crates/stackline-core/src/runtime.rs b/crates/stackline-core/src/runtime.rs index 3e112dd..1cc2017 100644 --- a/crates/stackline-core/src/runtime.rs +++ b/crates/stackline-core/src/runtime.rs @@ -17,7 +17,7 @@ use crate::docker::{DockerClient, Endpoint}; use crate::events::EventPoller; use crate::model::Snapshot; use crate::snapshot::{self, Options}; -use crate::store::Store; +use crate::store::{HostCheck, Store}; #[derive(Clone, Debug, Serialize, PartialEq, Eq)] #[serde(rename_all = "lowercase")] @@ -37,6 +37,10 @@ pub struct ConnState { pub updated_at: Option, pub logs_synced_at: Option, pub log_errors: Vec, + /// The connection now reaches another host or swarm than its stored + /// lines come from. Nothing is collected until the user decides whether + /// the old lines stay ([`crate::store::Store::accept_host`]). + pub other_host: bool, } /// One opened connection. @@ -65,6 +69,7 @@ impl Runtime { updated_at: None, logs_synced_at: None, log_errors: vec![], + other_host: false, }), client: tokio::sync::Mutex::new(None), tasks: Mutex::new(vec![]), @@ -81,6 +86,16 @@ impl Runtime { self.state.read().unwrap().clone() } + /// The host this connection reaches now, as the store remembers hosts. + pub fn identity(&self) -> Option { + self.snapshot()?.identity() + } + + /// The user answered the question about another host: collecting goes on. + pub fn host_accepted(&self) { + self.state.write().unwrap().other_host = false; + } + /// Reads the full configuration of a service by its full name. pub async fn details(&self, name: &str) -> Result> { let Some(snap) = self.snapshot() else { return Ok(None) }; @@ -161,7 +176,7 @@ pub async fn open(cfg: &ConnectionConfig) -> Result<(DockerClient, Option { let tunnel = SshTunnel::open(t).await?; - let client = DockerClient::new(Endpoint::Tcp(tunnel.local_addr().to_string())); + let client = DockerClient::new(tunnel.endpoint()); Ok((client, Some(tunnel))) } } @@ -209,11 +224,13 @@ pub struct Manager { deleting: Arc>>, /// Why the last cleanup of the store failed, until one succeeds again. maintenance_error: Mutex>, + /// The settings file could not be read at startup and was set aside. + pub config_warning: Option, } impl Manager { pub fn new(paths: Paths) -> Result> { - let config = AppConfig::load(&paths.config)?; + let (config, config_warning) = AppConfig::load_or_set_aside(&paths.config)?; let store = Arc::new(Store::open(&paths.store)?); let m = Arc::new(Manager { paths, @@ -223,6 +240,7 @@ impl Manager { runtimes: Mutex::new(HashMap::new()), deleting: Arc::default(), maintenance_error: Mutex::default(), + config_warning, }); let store = m.store.clone(); tokio::task::spawn_blocking(move || { @@ -328,14 +346,18 @@ impl Manager { // Checked when the identity is first seen or changes, not on every poll. if let Some(identity) = snap.identity().filter(|i| bound.as_ref() != Some(i)) { let (store, conn, id2) = (m.store.clone(), r.cfg.id.clone(), identity.clone()); - // Clearing a large store takes a while; keep it off the async workers. - match tokio::task::spawn_blocking(move || store.bind_host(&conn, &id2)).await { - Ok(Ok(cleared)) => { - if cleared { - tracing::info!("{} now reaches another host; its stored logs were cleared", r.cfg.name); - collector = Collector::new(&r.cfg.id, m.store.clone()); - events = EventPoller::new(&r.cfg.id, m.store.clone()); + match tokio::task::spawn_blocking(move || store.check_host(&conn, &id2)).await { + Ok(Ok(HostCheck::Changed)) => { + // Nothing is read until the user says what happens to the old + // lines; asked again on every poll until then. + if !std::mem::replace(&mut r.state.write().unwrap().other_host, true) { + tracing::warn!("{} now reaches another host; collecting waits for your answer", r.cfg.name); } + tokio::time::sleep(Duration::from_secs(settings.log_poll_secs.max(1) as u64)).await; + continue; + } + Ok(Ok(_)) => { + r.state.write().unwrap().other_host = false; bound = Some(identity); } Ok(Err(e)) => tracing::warn!("could not check the host of {}: {e:#}", r.cfg.name), 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 59% rename from crates/stackline-core/src/store.rs rename to crates/stackline-core/src/store/mod.rs index b6a530a..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,108 +51,9 @@ 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. -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)); -"#; +/// 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"; /// Filters shared by queries and exports. Times are Unix microseconds. #[derive(Clone, Debug, Default)] @@ -287,38 +198,15 @@ 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 { + /// The host the stored lines come from. + Same, + /// The first host seen for the connection; now remembered. + First, + /// Another host than the stored lines come from. + Changed, } fn escape_like(s: &str) -> String { @@ -591,53 +479,83 @@ impl Store { } let mut db = self.db.lock().unwrap(); let tx = db.transaction()?; - { - let mut ins = tx.prepare_cached( - "INSERT INTO entries (conn, ts, stack, service, replica, node, source, stream, level, exc, lines, fp, text, raw) + insert_entries(&tx, conn, src, entries)?; + tx.commit()?; + Ok(()) + } + + /// Stores lines of a read and moves the source's cursor in one step, + /// unless the lines of the connection were cleared since `epoch` (see + /// [`Store::epoch`]): then nothing is stored and false comes back. The + /// check holds the write lock, so a clear either comes first and the + /// batch is dropped, or comes after and deletes it with the rest. + pub fn store_batch(&self, conn: &str, src: &LogSource, entries: &[LogEntry], cursor: Option, epoch: u64) -> Result { + let mut db = self.db.lock().unwrap(); + if self.epoch() != epoch { + return Ok(false); + } + let tx = db.transaction()?; + insert_entries(&tx, conn, src, entries)?; + if let Some(ts) = cursor { + tx.execute(SET_CURSOR, params![conn, src.key, ts])?; + } + tx.commit()?; + Ok(true) + } +} + +/// Writes lines, their error patterns and their per-minute counts. +fn insert_entries(tx: &rusqlite::Transaction<'_>, conn: &str, src: &LogSource, entries: &[LogEntry]) -> Result<()> { + if entries.is_empty() { + return Ok(()); + } + { + let mut ins = tx.prepare_cached( + "INSERT INTO entries (conn, ts, stack, service, replica, node, source, stream, level, exc, lines, fp, text, raw) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)", - )?; - let mut pat = tx.prepare_cached( - "INSERT INTO patterns (conn, fp, pattern) VALUES (?1, ?2, ?3) + )?; + let mut pat = tx.prepare_cached( + "INSERT INTO patterns (conn, fp, pattern) VALUES (?1, ?2, ?3) ON CONFLICT (conn, fp) DO UPDATE SET pattern = excluded.pattern WHERE pattern != excluded.pattern", - )?; - // Counts per minute: debug, info, warn, error, then the same with a stack trace. - let mut minutes: HashMap = HashMap::new(); - for e in entries { - let m = minutes.entry(e.ts.div_euclid(MINUTE)).or_default(); - let l = e.level.as_i64() as usize - 1; - m[l] += 1; - m[4 + l] += e.exc as i64; - ins.execute(params![ - conn, - e.ts, - src.stack, - src.service, - src.replica, - src.node, - src.key, - if e.stream == StreamKind::Stderr { 2 } else { 1 }, - e.level.as_i64(), - e.exc as i64, - e.lines, - e.fp, - e.text, - e.raw, - ])?; - if let (Some(fp), Some(p)) = (&e.fp, &e.pattern) { - pat.execute(params![conn, fp, p])?; - } + )?; + // Counts per minute: debug, info, warn, error, then the same with a stack trace. + let mut minutes: HashMap = HashMap::new(); + for e in entries { + let m = minutes.entry(e.ts.div_euclid(MINUTE)).or_default(); + let l = e.level.as_i64() as usize - 1; + m[l] += 1; + m[4 + l] += e.exc as i64; + ins.execute(params![ + conn, + e.ts, + src.stack, + src.service, + src.replica, + src.node, + src.key, + if e.stream == StreamKind::Stderr { 2 } else { 1 }, + e.level.as_i64(), + e.exc as i64, + e.lines, + e.fp, + e.text, + e.raw, + ])?; + if let (Some(fp), Some(p)) = (&e.fp, &e.pattern) { + pat.execute(params![conn, fp, p])?; } - let mut add = tx.prepare_cached(ADD_MINUTE)?; - for (minute, c) in minutes { - for (service, stack) in [(src.service.as_str(), src.stack.as_str()), ("", "")] { - add.execute(params![conn, minute, service, stack, c[0], c[1], c[2], c[3], c[4], c[5], c[6], c[7]])?; - } + } + let mut add = tx.prepare_cached(ADD_MINUTE)?; + for (minute, c) in minutes { + for (service, stack) in [(src.service.as_str(), src.stack.as_str()), ("", "")] { + add.execute(params![conn, minute, service, stack, c[0], c[1], c[2], c[3], c[4], c[5], c[6], c[7]])?; } } - tx.commit()?; - Ok(()) } + Ok(()) +} +impl Store { pub fn cursor(&self, conn: &str, source: &str) -> Result> { let db = self.db.lock().unwrap(); Ok(db.query_row("SELECT ts FROM cursors WHERE conn = ?1 AND source = ?2", params![conn, source], |r| r.get(0)).optional()?) @@ -645,11 +563,7 @@ impl Store { pub fn set_cursor(&self, conn: &str, source: &str, ts: i64) -> Result<()> { let db = self.db.lock().unwrap(); - db.execute( - "INSERT INTO cursors (conn, source, ts) VALUES (?1, ?2, ?3) - ON CONFLICT (conn, source) DO UPDATE SET ts = excluded.ts", - params![conn, source, ts], - )?; + db.execute(SET_CURSOR, params![conn, source, ts])?; Ok(()) } @@ -709,7 +623,18 @@ impl Store { /// Stores events read from Docker; ones already stored are skipped. /// Returns how many were new. pub fn insert_events(&self, conn: &str, events: &[Event]) -> Result { + let epoch = self.epoch(); + Ok(self.store_events(conn, events, None, epoch)?.unwrap_or(0)) + } + + /// Like [`Store::insert_events`], and moves the event cursor to `cursor` + /// in the same step, unless the connection was cleared since `epoch` + /// (then None comes back and nothing is stored). + pub fn store_events(&self, conn: &str, events: &[Event], cursor: Option, epoch: u64) -> Result> { let mut db = self.db.lock().unwrap(); + if self.epoch() != epoch { + return Ok(None); + } let tx = db.transaction()?; let mut n = 0; { @@ -733,15 +658,27 @@ impl Store { ])?; } } + if let Some(ts) = cursor { + tx.execute(SET_CURSOR, params![conn, crate::events::CURSOR, ts])?; + } tx.commit()?; - Ok(n) + Ok(Some(n)) } /// The newest events in the filter's time range and sources, oldest /// first. Levels, error groups and search words do not apply to events. pub fn events(&self, conn: &str, f: &LogFilter, limit: usize) -> Result> { + self.events_of(conn, f, &[], limit) + } + + /// Like [`Store::events`], only events of these kinds (all when empty). + pub fn events_of(&self, conn: &str, f: &LogFilter, kinds: &[Kind], limit: usize) -> Result> { let f = LogFilter { fps: vec![], text: vec![], levels: vec![], no_levels: false, ..f.clone() }; - let (w, mut p) = f.where_clause(conn, false); + let (mut w, mut p) = f.where_clause(conn, false); + if !kinds.is_empty() { + w.push_str(&format!(" AND kind IN ({})", vec!["?"; kinds.len()].join(","))); + p.extend(kinds.iter().map(|k| Sql::Text(k.as_str().into()))); + } p.push(Sql::Integer(limit as i64)); self.read(|db| { let mut stmt = db.prepare(&format!( @@ -794,708 +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 first, down to 90% of the - // limit so this does not run again a minute later. Ids grow with - // arrival, so the rowid order finds them without sorting the table. - // Deleted lines stay in the search index until it is merged, so the - // index is optimized before the size is measured again. - for _ in 0..4 { - let live = live_bytes(&self.db.lock().unwrap())?; - if live <= max_bytes { - break; - } - let total: u64 = self.db.lock().unwrap().query_row("SELECT COUNT(*) FROM entries", [], |r| r.get(0))?; - if total == 0 { - break; - } - let target = max_bytes / 10 * 9; - let mut left = ((total as u128 * (live - target) as u128).div_ceil(live as u128) as u64).clamp(1, total); - while left > 0 { - let n = self.db.lock().unwrap().execute( - "DELETE FROM entries WHERE id IN (SELECT id FROM entries ORDER BY id LIMIT ?1)", - params![left.min(BATCH as u64)], - )? as u64; - deleted += n; - left = left.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)) - } - - /// Remembers which host or swarm a connection's logs come from. When a - /// connection now reaches a different one (its address was edited, or an - /// alias points elsewhere), the old lines are deleted so the two never mix. - /// Returns true when lines were deleted. - pub fn bind_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()?; - if known.as_deref() == Some(identity) { - return Ok(false); - } - let changed = known.is_some(); - drop(db); - if changed { - 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(changed) - } - - 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. - 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 { - if s.contains([',', '"', '\n', '\r']) { format!("\"{}\"", s.replace('"', "\"\"")) } else { s.to_string() } -} - #[cfg(test)] mod tests { use super::*; @@ -1659,6 +894,87 @@ mod tests { assert!(keys(&all, 10).is_empty()); } + #[test] + fn csv_cells_never_start_a_formula() { + assert_eq!(csv("=HYPERLINK(\"http://x\")"), "\"'=HYPERLINK(\"\"http://x\"\")\""); + assert_eq!(csv("-1 retries left"), "'-1 retries left"); + assert_eq!(csv("@admin logged in"), "'@admin logged in"); + assert_eq!(csv("plain, text"), "\"plain, text\""); + assert_eq!(csv("ok"), "ok"); + } + + #[test] + fn deploys_are_found_among_many_other_events() { + let store = Store::open_in_memory().unwrap(); + let ev = |key: String, ts: i64, kind: Kind| Event { + key, + ts, + stack: "demo".into(), + service: "demo_app".into(), + replica: "1".into(), + node: "box".into(), + source: "c0".into(), + kind, + exit_code: None, + detail: None, + }; + let mut batch = vec![ev("deploy".into(), 10, Kind::Deploy)]; + batch.extend((0..6000).map(|i| ev(format!("crash{i}"), 100 + i, Kind::Crash))); + store.insert_events("c1", &batch).unwrap(); + let all = LogFilter { from: 0, to: i64::MAX, ..Default::default() }; + let deploys = store.events_of("c1", &all, &[Kind::Deploy], 50).unwrap(); + assert_eq!(deploys.iter().map(|e| e.key.as_str()).collect::>(), ["deploy"]); + assert_eq!(store.events_of("c1", &all, &[], 10).unwrap().len(), 10); + } + + #[test] + fn a_batch_read_before_a_clear_is_not_stored_after_it() { + let store = Store::open_in_memory().unwrap(); + let epoch = store.epoch(); + assert!(store.store_batch("c1", &src("shop_api", "1"), &entries(&[(10, "before")]), Some(10), epoch).unwrap()); + // "Clear stored logs" while a read of the same source is under way. + store.clear(Some("c1")).unwrap(); + let stored = store.store_batch("c1", &src("shop_api", "1"), &entries(&[(20, "read before the clear")]), Some(20), epoch).unwrap(); + assert!(!stored); + let all = LogFilter { from: 0, to: 100, ..Default::default() }; + assert_eq!(store.query("c1", &all, 1, 10).unwrap().entries.len(), 0); + // No cursor either, so the next read starts over and fills the range again. + assert_eq!(store.cursor("c1", "shop_api-1").unwrap(), None); + assert_eq!(store.store_events("c1", &[], Some(20), epoch).unwrap(), None); + assert_eq!(store.cursor("c1", crate::events::CURSOR).unwrap(), None); + } + + #[test] + fn a_chatty_connection_does_not_push_out_a_quiet_one() { + assert_eq!(over_share(&[("a".into(), 900), ("b".into(), 100)], 500), [("a".to_string(), 500)]); + assert_eq!(over_share(&[("a".into(), 900), ("b".into(), 300)], 400), [("b".to_string(), 100), ("a".to_string(), 700)]); + let dir = std::env::temp_dir().join(format!("stackline-fair-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let store = Store::open(&dir.join("logs.db")).unwrap(); + // The quiet connection logged first, long ago; the chatty one fills the store. + store.insert("quiet", &src("db", "1"), &entries(&[(1, "the quiet server's oldest line")])).unwrap(); + let mut t = 1_000; + for round in 0..40 { + let lines: Vec<(i64, String)> = (0..2000) + .map(|i| { + t += 1; + (t, format!("GET /api/items/{round}-{i} 200 id={:016x}", (round * 2000 + i) as u64 * 2654435761)) + }) + .collect(); + let refs: Vec<(i64, &str)> = lines.iter().map(|(t, l)| (*t, l.as_str())).collect(); + store.insert("chatty", &src("api", "1"), &entries(&refs)).unwrap(); + } + store.retain(0, 2 << 20).unwrap(); + let all = LogFilter { from: 0, to: i64::MAX, ..Default::default() }; + let quiet = store.query("quiet", &all, 1, 10).unwrap(); + assert_eq!(quiet.entries.len(), 1, "the quiet server lost its history"); + let chatty = store.query("chatty", &all, 1, 1).unwrap(); + assert!(chatty.total < 80_000 && chatty.entries[0].ts == t, "the chatty server keeps its newest lines"); + drop(store); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn picked_stacks_and_services_combine_with_or() { let store = Store::open_in_memory().unwrap(); @@ -1756,7 +1072,7 @@ mod tests { } #[test] - fn a_connection_that_reaches_another_host_drops_the_old_lines() { + fn another_host_keeps_the_old_lines_until_the_user_decides() { let store = Store::open_in_memory().unwrap(); let now = 1_700_000_000_000_000i64; let s1 = src("shop_api", "1"); @@ -1764,17 +1080,30 @@ mod tests { store.insert("c2", &s1, &entries(&[(now - 1_000_000, "2026-10-27 ERROR boom")])).unwrap(); store.set_cursor("c1", &s1.key, now).unwrap(); let all = LogFilter { from: 0, to: now, ..Default::default() }; + let total = |c: &str| store.query(c, &all, 1, 10).unwrap().total; // The first host seen keeps what is stored. - assert!(!store.bind_host("c1", "engine:A").unwrap()); - assert!(!store.bind_host("c1", "engine:A").unwrap()); - assert_eq!(store.query("c1", &all, 1, 10).unwrap().total, 1); - - // Another host under the same connection: its lines start from scratch. - assert!(store.bind_host("c1", "engine:B").unwrap()); - assert_eq!(store.query("c1", &all, 1, 10).unwrap().total, 0); + assert_eq!(store.check_host("c1", "engine:A").unwrap(), HostCheck::First); + assert_eq!(store.check_host("c1", "engine:A").unwrap(), HostCheck::Same); + assert_eq!(total("c1"), 1); + + // Another host under the same connection: reported, nothing deleted. + assert_eq!(store.check_host("c1", "engine:B").unwrap(), HostCheck::Changed); + assert_eq!(store.check_host("c1", "engine:B").unwrap(), HostCheck::Changed); + assert_eq!(total("c1"), 1); + + // "Keep them": the new host is remembered, the old lines stay. + store.accept_host("c1", "engine:B", true).unwrap(); + assert_eq!(store.check_host("c1", "engine:B").unwrap(), HostCheck::Same); + assert_eq!(total("c1"), 1); + + // "Delete them": lines and cursors of that connection go, others stay. + assert_eq!(store.check_host("c1", "engine:C").unwrap(), HostCheck::Changed); + store.accept_host("c1", "engine:C", false).unwrap(); + assert_eq!(store.check_host("c1", "engine:C").unwrap(), HostCheck::Same); + assert_eq!(total("c1"), 0); assert_eq!(store.cursor("c1", &s1.key).unwrap(), None); - assert_eq!(store.query("c2", &all, 1, 10).unwrap().total, 1, "other connections are untouched"); + assert_eq!(total("c2"), 1, "other connections are untouched"); } #[test] 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)); +"#; diff --git a/crates/stackline/src/api.rs b/crates/stackline/src/api.rs index c116405..03b095b 100644 --- a/crates/stackline/src/api.rs +++ b/crates/stackline/src/api.rs @@ -50,6 +50,7 @@ pub fn router() -> Router> { .route("/c/{id}/snapshot", get(snapshot)) .route("/c/{id}/pins", put(put_pins)) .route("/c/{id}/collect", put(put_collect)) + .route("/c/{id}/host", post(accept_host)) .route("/c/{id}/usage", get(usage)) .route("/c/{id}/usage/clear", post(clear_usage)) .route("/c/{id}/logs", get(logs)) @@ -76,6 +77,7 @@ async fn meta(st: St) -> Json { "desktop": st.desktop, "storePath": st.manager.store_path().display().to_string(), "platform": std::env::consts::OS, + "configWarning": st.manager.config_warning, })) } @@ -197,6 +199,24 @@ async fn put_pins(st: St, Path(id): Path, Json(pins): Json>) Ok(Json(clean)) } +#[derive(Deserialize)] +struct HostInput { + /// Keep the lines stored from the host before, next to the new ones. + keep: bool, +} + +/// Answers the question the UI asks when a connection reaches another host +/// than its stored lines come from: delete those lines, or keep them. +async fn accept_host(st: St, Path(id): Path, Json(input): Json) -> Result { + let rt = st.manager.runtime(&id).ok_or_else(not_found)?; + let identity = rt.identity().ok_or_else(|| ApiError(StatusCode::SERVICE_UNAVAILABLE, "Not connected yet.".into()))?; + let store = st.manager.store.clone(); + // Deleting a large store takes a while; keep it off the async workers. + tokio::task::spawn_blocking(move || store.accept_host(&id, &identity, input.keep)).await.map_err(|e| anyhow::anyhow!(e))??; + rt.host_accepted(); + Ok(StatusCode::NO_CONTENT) +} + /// Chooses which services of a connection have their logs collected. async fn put_collect(st: St, Path(id): Path, Json(c): Json) -> Result, ApiError> { if !st.manager.config().connections.iter().any(|x| x.id == id) { @@ -313,8 +333,7 @@ async fn deploys(st: St, Path(id): Path, Query(p): Query) -> let store = st.manager.store.clone(); let now = chrono::Utc::now().timestamp_micros(); let out = tokio::task::spawn_blocking(move || -> anyhow::Result> { - let mut events: Vec<_> = - store.events(&id, &f, 5000)?.into_iter().filter(|e| e.kind == stackline_core::events::Kind::Deploy).collect(); + let mut events: Vec<_> = store.events_of(&id, &f, &[stackline_core::events::Kind::Deploy], 50)?; events.sort_by_key(|e| std::cmp::Reverse(e.ts)); events.truncate(50); let mut next: std::collections::HashMap = Default::default(); diff --git a/crates/stackline/src/lib.rs b/crates/stackline/src/lib.rs index c0e3b2d..8400ac8 100644 --- a/crates/stackline/src/lib.rs +++ b/crates/stackline/src/lib.rs @@ -44,6 +44,8 @@ pub struct AppState { pub token: String, pub desktop: bool, pub allowed_hosts: Vec, + /// Also answer any `Host` that is an IP address with this port. + pub ip_hosts: Option, pub updater: Arc, } @@ -71,6 +73,10 @@ pub async fn start(opts: ServeOptions) -> Result { if !addr.ip().is_loopback() { allowed_hosts.push(addr.to_string()); } + // Listening beyond this computer (`--host`): other machines address it by + // one of its IP addresses, which cannot be known in advance for 0.0.0.0. + // Names stay refused, which is what keeps DNS rebinding out. + let ip_hosts = (!addr.ip().is_loopback()).then_some(addr.port()); manager.open_all(); let updater = Updater::new(opts.installer); updater.spawn_checks(manager.clone()); @@ -79,6 +85,7 @@ pub async fn start(opts: ServeOptions) -> Result { token: token.clone(), desktop: opts.desktop, allowed_hosts, + ip_hosts, updater: updater.clone(), }); let app = web::router(state); diff --git a/crates/stackline/src/main.rs b/crates/stackline/src/main.rs index 68f659b..08cdb97 100644 --- a/crates/stackline/src/main.rs +++ b/crates/stackline/src/main.rs @@ -54,6 +54,13 @@ async fn main() -> Result<()> { println!("Stackline {} is running at {url}", env!("CARGO_PKG_VERSION")); } else { println!("Stackline is running. Open this link to use it:\n\n {url}\n\nPress Ctrl+C to stop."); + if !host.is_loopback() { + println!( + "\nStackline listens beyond this computer ({host}). Other machines open it with this computer's IP address\n\ + in place of {}. The link carries the sign-in and travels unencrypted: share it only on a network you trust.", + server.addr.ip() + ); + } if !no_open { let _ = webbrowser::open(&url); } diff --git a/crates/stackline/src/web.rs b/crates/stackline/src/web.rs index d6cc314..b36a5a5 100644 --- a/crates/stackline/src/web.rs +++ b/crates/stackline/src/web.rs @@ -32,7 +32,8 @@ pub fn router(state: Arc) -> Router { /// Only answers requests addressed to this machine, and only API calls that carry the session token. async fn guard(State(st): State>, req: Request, next: Next) -> Response { let host = req.headers().get(header::HOST).and_then(|h| h.to_str().ok()).unwrap_or(""); - if !st.allowed_hosts.iter().any(|h| h == host) { + let ip_host = st.ip_hosts.is_some_and(|port| host.parse::().is_ok_and(|a| a.port() == port)); + if !ip_host && !st.allowed_hosts.iter().any(|h| h == host) { return (StatusCode::FORBIDDEN, "Stackline only answers on its own address").into_response(); } let path = req.uri().path().to_string(); diff --git a/crates/stackline/tests/api.rs b/crates/stackline/tests/api.rs index 5af22e3..3b3e5ef 100644 --- a/crates/stackline/tests/api.rs +++ b/crates/stackline/tests/api.rs @@ -30,14 +30,25 @@ impl Drop for App { } fn app() -> App { + app_with(None) +} + +/// An app listening beyond this computer (`--host`) on this port. +fn app_with(ip_hosts: Option) -> App { // A counter, because tests run in parallel and macOS clocks tick in microseconds. static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); let nanos = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos(); let dir = std::env::temp_dir().join(format!("stackline-api-test-{}-{nanos}-{n}", std::process::id())); let manager = Manager::new(Paths::in_dir(&dir)).unwrap(); - let state = - Arc::new(AppState { manager, token: TOKEN.into(), desktop: false, allowed_hosts: vec![HOST.into()], updater: Updater::new(None) }); + let state = Arc::new(AppState { + manager, + token: TOKEN.into(), + desktop: false, + allowed_hosts: vec![HOST.into()], + ip_hosts, + updater: Updater::new(None), + }); App { router: router(state.clone()), state, dir } } @@ -109,6 +120,27 @@ async fn only_answers_on_its_own_address() { assert_eq!(send(&app, elsewhere).await.0, StatusCode::FORBIDDEN); } +#[tokio::test] +async fn listening_on_the_network_answers_ip_addresses_but_no_names() { + let at = |app: &App, host: &str| { + let r = Request::builder() + .uri("/api/meta") + .header(header::HOST, host) + .header(header::COOKIE, format!("stackline_token={TOKEN}")) + .body(Body::empty()) + .unwrap(); + let router = app.router.clone(); + async move { router.oneshot(r).await.unwrap().status() } + }; + let app = app_with(Some(7070)); + assert_eq!(at(&app, "192.168.1.20:7070").await, StatusCode::OK); + assert_eq!(at(&app, "[fd00::5]:7070").await, StatusCode::OK); + assert_eq!(at(&app, "192.168.1.20:8080").await, StatusCode::FORBIDDEN); + assert_eq!(at(&app, "attacker.example:7070").await, StatusCode::FORBIDDEN); + // On this computer only, an IP address of the network is refused like a name. + assert_eq!(at(&app_with(None), "192.168.1.20:7070").await, StatusCode::FORBIDDEN); +} + #[tokio::test] async fn the_link_with_the_token_signs_the_browser_in() { let app = app(); 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 }) +}) diff --git a/ui/src/App.svelte b/ui/src/App.svelte index d82dae0..4d7e1e9 100644 --- a/ui/src/App.svelte +++ b/ui/src/App.svelte @@ -61,6 +61,8 @@ gAt = e.key === 'g' ? Date.now() : 0 } } + /** The note about a reset settings file was read. */ + let configSeen = $state(false) @@ -112,6 +114,13 @@
+ {#if app.meta?.configWarning && !configSeen} + + {/if} {#if app.fatal}
{app.fatal}
{:else if app.view === 'connections'} @@ -127,6 +136,14 @@ {#if app.conn?.status === 'error'}
Connection lost.{app.conn.error} Showing the last known state.
{/if} + {#if app.conn?.otherHost} + + {/if} {#key app.activeId} {#if app.view === 'overview'} diff --git a/ui/src/app.css b/ui/src/app.css index eb1d521..d496b88 100644 --- a/ui/src/app.css +++ b/ui/src/app.css @@ -314,6 +314,7 @@ button:disabled{opacity:.5;cursor:default} a{color:var(--accent)} .err-banner{display:flex;gap:12px;align-items:flex-start;padding:14px 18px;border-radius:var(--r);background:color-mix(in srgb,var(--err) 9%,var(--panel));border:1px solid color-mix(in srgb,var(--err) 30%,transparent);white-space:pre-wrap} .warn-banner{display:flex;gap:12px;align-items:flex-start;padding:12px 18px;border-radius:var(--r);background:color-mix(in srgb,var(--warn) 10%,var(--panel));border:1px solid color-mix(in srgb,var(--warn) 30%,transparent)} +.host-banner{align-items:center;flex-wrap:wrap;margin-bottom:12px} .center-empty{display:grid;place-items:center;gap:10px;padding:60px 20px;text-align:center;color:var(--muted)} .spinner{width:22px;height:22px;border-radius:50%;border:3px solid var(--line);border-top-color:var(--accent);animation:spin .8s linear infinite} @keyframes spin{to{transform:rotate(360deg)}} diff --git a/ui/src/lib/state.svelte.ts b/ui/src/lib/state.svelte.ts index d39eb12..9f3400d 100644 --- a/ui/src/lib/state.svelte.ts +++ b/ui/src/lib/state.svelte.ts @@ -512,6 +512,21 @@ class AppState { this.#timer = setTimeout(() => this.poll(), document.hidden ? Math.max(every, 30000) : every) } + /** + * Answers the question about another host behind the active connection: + * keep the logs stored from the host before, or delete them. + */ + async acceptHost(keep: boolean) { + const id = this.activeId + try { + await post(`/c/${encodeURIComponent(id)}/host`, { keep }) + if (this.conn && id === this.activeId) this.conn = { ...this.conn, otherHost: false } + this.notify(keep ? 'Kept the old logs. Collecting goes on.' : 'Deleted the old logs. Collecting starts over.', 'ok') + } catch (e) { + this.notify(this.describe(e), 'err') + } + } + async reloadConnections() { this.connections = await api('/connections') if (!this.connections.some((c) => c.id === this.activeId) && this.connections[0]) this.select(this.connections[0].id) diff --git a/ui/src/lib/types.ts b/ui/src/lib/types.ts index dbf43c7..a1bfcab 100644 --- a/ui/src/lib/types.ts +++ b/ui/src/lib/types.ts @@ -37,6 +37,8 @@ export interface Snapshot { export interface ConnState { status: 'connecting' | 'connected' | 'error'; error: string | null; latencyMs: number | null endpoint: string | null; updatedAt: number | null; logsSyncedAt: number | null; logErrors: string[] + /** The connection reaches another host than its stored logs come from; collecting waits for an answer. */ + otherHost?: boolean } export interface SshFields { target: string; port?: number | null; identity?: string | null; jump?: string | null; socket: string } @@ -50,7 +52,11 @@ export interface Settings { alertEnabled: boolean; alertWindowMins: number; alertThreshold: number } -export interface Meta { version: string; desktop: boolean; storePath: string; platform: string } +export interface Meta { + version: string; desktop: boolean; storePath: string; platform: string + /** The settings file could not be read at startup and was set aside. */ + configWarning?: string | null +} export interface Bucket { error: number; warn: number; info: number; debug: number; exc: number } export interface Group {