diff --git a/CHANGELOG.md b/CHANGELOG.md index c08d37e..f0df8e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,14 @@ its GitHub release, and Stackline shows it in the update dialog. ### 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 - **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 diff --git a/crates/stackline-core/src/collector.rs b/crates/stackline-core/src/collector.rs index a0e0916..49abe2d 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,66 @@ 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 4 seconds: a first read of hours of logs. + 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(4)).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 started = Instant::now(); + 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. + tokio::time::sleep(Duration::from_secs(3)).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/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 32f80f6..f73e8b7 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) }; @@ -328,14 +343,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.rs b/crates/stackline-core/src/store.rs index b6a530a..9781402 100644 --- a/crates/stackline-core/src/store.rs +++ b/crates/stackline-core/src/store.rs @@ -43,6 +43,9 @@ 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 SET_CURSOR: &str = "INSERT INTO cursors (conn, source, ts) VALUES (?1, ?2, ?3) + ON CONFLICT (conn, source) DO UPDATE SET ts = excluded.ts"; + const INSERT_TRIGGER: &str = "CREATE TRIGGER entries_ai AFTER INSERT ON entries BEGIN INSERT INTO entries_fts (rowid, text) VALUES (new.id, new.text); END;"; @@ -321,6 +324,36 @@ fn new_groups(db: &Connection, conn: &str, now: i64, oldest: Option) -> Res 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, +} + +/// How many lines each connection gives up so that `keep` lines are left: +/// the lines kept are shared out equally, and a connection with fewer lines +/// than its share keeps all of them, which leaves more for the others. +fn over_share(counts: &[(String, u64)], keep: u64) -> Vec<(String, u64)> { + let mut sorted: Vec<&(String, u64)> = counts.iter().collect(); + sorted.sort_by_key(|(_, n)| *n); + let mut left = keep; + let mut out = Vec::new(); + for (i, (conn, n)) in sorted.iter().enumerate() { + let share = left / (sorted.len() - i) as u64; + let kept = (*n).min(share); + left -= kept; + if *n > kept { + out.push((conn.clone(), n - kept)); + } + } + out +} + fn escape_like(s: &str) -> String { s.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_") } @@ -591,53 +624,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 +708,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 +768,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 +803,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!( @@ -822,31 +904,40 @@ impl Store { .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. + // 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 total: u64 = self.db.lock().unwrap().query_row("SELECT COUNT(*) FROM entries", [], |r| r.get(0))?; + 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 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; + 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()?; @@ -995,26 +1086,34 @@ impl Store { 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 { + /// 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()?; - if known.as_deref() == Some(identity) { - return Ok(false); + 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) + } } - let changed = known.is_some(); - drop(db); - if changed { + } + + /// 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(changed) + Ok(()) } pub fn stats(&self) -> Result { @@ -1037,7 +1136,12 @@ impl Store { 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); + // 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()?; @@ -1659,6 +1763,78 @@ mod tests { assert!(keys(&all, 10).is_empty()); } + #[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 +1932,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 +1940,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/src/api.rs b/crates/stackline/src/api.rs index c116405..9683d7e 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)) @@ -197,6 +198,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 +332,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/ui/src/App.svelte b/ui/src/App.svelte index d82dae0..8d7b279 100644 --- a/ui/src/App.svelte +++ b/ui/src/App.svelte @@ -127,6 +127,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..687f87c 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 }