@@ -23,6 +23,11 @@ //! 30th day after it (UTC), so no DID is kept longer than 30 days.//! Before they go, each funnel's day is kept as two counts with no one//! in them: how many accounts started it, and how many of those went on.//! - What tells an operator why, by the hour, and for those same 30//! days: each of the index's statements slow enough to log (its SQL as//! written, no value it was given), why an event failed (what it//! answered, which can name a repository or an account), and the//! `User-Agent` of each crawler or script, never a browser's.//!//! Requests are counted in memory and written now and then//! ([`Usage::flush`]); the salt is in the database, so every serve@@ -38,7 +43,9 @@ use axum::http::{HeaderMap, StatusCode, header}; use axum::middleware::Next;use axum::response::{IntoResponse, Response};use knotless::telemetry::{self, Fields, Level};use knotless::usage::{Count, Day, Event, Funnel, Hour, Network, Report, Route, Stats};use knotless::usage::{ Count, Day, Event, Failure, Funnel, Hour, Network, Report, Route, Statement, Stats,};use rand_core::{OsRng, RngCore};use rusqlite::{Connection, params};use serde::Deserialize;@@ -46,7 +53,7 @@ use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, BTreeSet, HashMap};use std::net::IpAddr;use std::path::Path;use std::sync::{Arc, Mutex};use std::sync::{Arc, Mutex, RwLock, Weak};use std::time::{Duration, Instant, SystemTime};/// How often the counted requests are written.@@ -63,6 +70,17 @@ pub const ACCOUNT_DAYS: u64 = 30; /// The longest route or referrer a page may send.const NAME_MAX: usize = 200;/// The most of a failure's error, or a crawler's `User-Agent`, kept.const ERROR_MAX: usize = 300;const AGENT_MAX: usize = 200;/// The most failures, or crawlers, kept apart in an hour; past it, the/// rest are counted together as [`OVERFLOW`], so a crawler naming itself/// anew each time, or a statement built to its input, cannot grow the/// table without end.const KEPT_PER_HOUR: i64 = 500;const OVERFLOW: &str = "(others)";pub struct Usage { db: Mutex<Connection>, /// Requests counted since the last flush.@@ -72,6 +90,41 @@ pub struct Usage { /// The accounts seen in each, by day, in order: an account's first /// day of an event is written before its next. actors: Mutex<BTreeSet<(String, &'static str, String)>>, /// Slow statements counted since the last flush, by hour and SQL. statements: Mutex<HashMap<(String, String), Slow>>, /// Failures counted since the last flush. failures: Mutex<HashMap<FailureKey, u64>>, /// Crawlers' requests counted since the last flush, by hour and agent. agents: Mutex<HashMap<(String, String), u64>>,}/// A failure's hour, event name, subject and error.type FailureKey = (String, &'static str, String, String);/// A statement's slow runs: how many, their milliseconds, and the most.#[derive(Default, Debug, PartialEq)]struct Slow { n: u64, ms: u64, max: u64,}/// The usage slow statements are counted in: the index's statements are/// watched by a function SQLite calls, which can take no state.static SINK: RwLock<Weak<Usage>> = RwLock::new(Weak::new());/// Has the index's slow statements counted in `usage`.pub fn sink(usage: &Arc<Usage>) { *SINK.write().unwrap() = Arc::downgrade(usage);}/// Counts the slow statement `sql`, which took `took`, in the usage/// [`sink`] named, if it named one.pub fn slow_statement(sql: &str, took: Duration) { let usage = SINK.read().unwrap().upgrade(); if let Some(usage) = usage { usage.statement(sql, took); }}/// The funnels the report follows: of the accounts that first did the one@@ -138,6 +191,17 @@ impl Usage { PRIMARY KEY (day, start, goal)) WITHOUT ROWID; -- Kept before as `actor`, for good: moved, and folded at the -- next flush. CREATE TABLE IF NOT EXISTS statement ( hour TEXT NOT NULL, sql TEXT NOT NULL, n INTEGER NOT NULL, ms INTEGER NOT NULL, max INTEGER NOT NULL, PRIMARY KEY (hour, sql)) WITHOUT ROWID; CREATE TABLE IF NOT EXISTS failure ( hour TEXT NOT NULL, name TEXT NOT NULL, subject TEXT NOT NULL, error TEXT NOT NULL, n INTEGER NOT NULL, PRIMARY KEY (hour, name, subject, error)) WITHOUT ROWID; CREATE TABLE IF NOT EXISTS agent ( hour TEXT NOT NULL, agent TEXT NOT NULL, n INTEGER NOT NULL, PRIMARY KEY (hour, agent)) WITHOUT ROWID; CREATE TABLE IF NOT EXISTS actor ( day TEXT NOT NULL, name TEXT NOT NULL, did TEXT NOT NULL, PRIMARY KEY (day, name, did)) WITHOUT ROWID;@@ -153,9 +217,32 @@ impl Usage { pending: Mutex::default(), events: Mutex::default(), actors: Mutex::default(), statements: Mutex::default(), failures: Mutex::default(), agents: Mutex::default(), }) } /// Counts the index's statement `sql`, which took `took`: one slow /// enough to log. pub fn statement(&self, sql: &str, took: Duration) { let key = (hour(SystemTime::now()), telemetry::sql_text(sql)); let ms = took.as_millis() as u64; let mut statements = self.statements.lock().unwrap(); let slow = statements.entry(key).or_default(); slow.n += 1; slow.ms += ms; slow.max = slow.max.max(ms); } /// Counts why the event `name`, of `subject`, failed: `error`, what it /// answered. pub fn failed(&self, name: &'static str, subject: &str, error: &str) { let error: String = fingerprint(error).chars().take(ERROR_MAX).collect(); let key = (hour(SystemTime::now()), name, subject.to_string(), error); *self.failures.lock().unwrap().entry(key).or_default() += 1; } /// Counts an event a handler saw: `name`, of `subject` (a tool, a /// repository, a version, or nothing), whether it worked, how long it /// took, and the account, if signed in.@@ -193,12 +280,18 @@ impl Usage { path: &str, method: &str, status: u16, client: &'static str, agent: &str, took: Duration, ) { let Some((surface, route)) = surface(route, path) else { return; }; let client = client(agent); if client == "bot" { let agent: String = agent.chars().take(AGENT_MAX).collect(); let key = (hour(SystemTime::now()), agent); *self.agents.lock().unwrap().entry(key).or_default() += 1; } // A long poll's wait is what it is for, not what anyone waited on. let slow = u64::from(!crate::telemetry::waits(Some(route.as_str()))); let key = Key {@@ -282,6 +375,7 @@ impl Usage { } drop(stmt); tx.execute("DELETE FROM account WHERE day < ?1", [&kept])?; self.flush_detail(&tx, &kept)?; tx.execute( "INSERT INTO visitors SELECT day, count(*) FROM visitor WHERE day < ?1 GROUP BY day ON CONFLICT DO UPDATE SET n = n + excluded.n",@@ -292,6 +386,57 @@ impl Usage { tx.commit() } /// Writes the statements, failures and crawlers counted, and deletes /// those of the days before `kept`, as the accounts are. fn flush_detail(&self, tx: &rusqlite::Transaction, kept: &str) -> rusqlite::Result<()> { // A new statement, failure or crawler past the hour's // [`KEPT_PER_HOUR`] is counted as [`OVERFLOW`]; one already kept // goes on being. Statements are as written, so few, but one built // to its input, as an `IN` list, is new at each size. let statements = std::mem::take(&mut *self.statements.lock().unwrap()); let mut stmt = tx.prepare_cached( "INSERT INTO statement SELECT ?1, CASE WHEN EXISTS (SELECT 1 FROM statement WHERE hour = ?1 AND sql = ?2) OR (SELECT count(*) FROM statement WHERE hour = ?1) < ?6 THEN ?2 ELSE ?7 END, ?3, ?4, ?5 ON CONFLICT DO UPDATE SET n = n + excluded.n, ms = ms + excluded.ms, max = max(max, excluded.max)", )?; for ((hour, sql), s) in &statements { let p = params![hour, sql, s.n, s.ms, s.max, KEPT_PER_HOUR, OVERFLOW]; stmt.execute(p)?; } let failures = std::mem::take(&mut *self.failures.lock().unwrap()); let mut stmt = tx.prepare_cached( "INSERT INTO failure SELECT ?1, ?2, ?3, CASE WHEN EXISTS (SELECT 1 FROM failure WHERE hour = ?1 AND name = ?2 AND subject = ?3 AND error = ?4) OR (SELECT count(*) FROM failure WHERE hour = ?1) < ?6 THEN ?4 ELSE ?7 END, ?5 ON CONFLICT DO UPDATE SET n = n + excluded.n", )?; for ((hour, name, subject, error), n) in &failures { let p = params![hour, name, subject, error, n, KEPT_PER_HOUR, OVERFLOW]; stmt.execute(p)?; } let agents = std::mem::take(&mut *self.agents.lock().unwrap()); let mut stmt = tx.prepare_cached( "INSERT INTO agent SELECT ?1, CASE WHEN EXISTS (SELECT 1 FROM agent WHERE hour = ?1 AND agent = ?2) OR (SELECT count(*) FROM agent WHERE hour = ?1) < ?4 THEN ?2 ELSE ?5 END, ?3 ON CONFLICT DO UPDATE SET n = n + excluded.n", )?; for ((hour, agent), n) in &agents { stmt.execute(params![hour, agent, n, KEPT_PER_HOUR, OVERFLOW])?; } // An hour, `YYYY-MM-DDTHH`, sorts after its day and before the next. for table in ["statement", "failure", "agent"] { tx.execute(&format!("DELETE FROM {table} WHERE hour < ?1"), [kept])?; } Ok(()) } /// Counts a page drawn in a browser: the route that drew it, the host /// that sent the visitor there, and the visitor, by `ip` and `agent`. fn view(@@ -435,10 +580,71 @@ impl Usage { let funnel = stmt.query_row(params![from, to, since], funnel)?; report.funnels.push(funnel); } let mut stmt = db.prepare( "SELECT sql, sum(n), sum(ms) / sum(n), max(max) FROM statement WHERE hour >= ?1 GROUP BY 1 ORDER BY sum(ms) DESC, 1 LIMIT 50", )?; report.statements = stmt .query_map([&since], statement)? .collect::<rusqlite::Result<_>>()?; let mut stmt = db.prepare( "SELECT name, subject, error, sum(n) FROM failure WHERE hour >= ?1 GROUP BY 1, 2, 3 ORDER BY 4 DESC, 1, 2, 3 LIMIT 100", )?; report.failures = stmt .query_map([&since], failure)? .collect::<rusqlite::Result<_>>()?; let mut stmt = db.prepare( "SELECT agent, sum(n) FROM agent WHERE hour >= ?1 GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT 20", )?; report.agents = stmt .query_map([&since], count)? .collect::<rusqlite::Result<_>>()?; Ok(()) }}/// `error` with what differs from one failure to the next put as what/// it is, so like failures count together: a DID as/// `<did>`, an `at://` URI as `<uri>`, a hash or other hex of seven/// digits or more as `<hex>`, and a number as `<n>`.fn fingerprint(error: &str) -> String { let words = error.split_whitespace().map(|word| { let edge = |c: char| !c.is_alphanumeric() && !matches!(c, ':' | '/' | '.' | '_' | '-'); let core = word.trim_matches(edge).trim_end_matches(['.', ':']); let is = |pred: fn(char) -> bool| !core.is_empty() && core.chars().all(pred); let kind = match () { _ if core.starts_with("did:") => "<did>", _ if core.starts_with("at://") => "<uri>", _ if core.len() >= 7 && is(|c| c.is_ascii_hexdigit()) && !is(char::is_alphabetic) => { "<hex>" } _ if is(|c| c.is_ascii_digit()) => "<n>", _ => return word.to_string(), }; word.replacen(core, kind, 1) }); words.collect::<Vec<_>>().join(" ")}fn statement(row: &rusqlite::Row) -> rusqlite::Result<Statement> { Ok(Statement { sql: row.get(0)?, calls: row.get(1)?, mean_ms: row.get(2)?, max_ms: row.get(3)?, })}fn failure(row: &rusqlite::Row) -> rusqlite::Result<Failure> { Ok(Failure { name: row.get(0)?, subject: row.get(1)?, error: row.get(2)?, n: row.get(3)?, })}fn counts(row: &rusqlite::Row) -> rusqlite::Result<(String, u64, u64)> { Ok((row.get(0)?, row.get(1)?, row.get(2)?))}@@ -613,8 +819,7 @@ pub async fn count_requests(usage: Arc<Usage>, req: Request, next: Next) -> Resp .map(|m| m.as_str().to_string()); let path = req.uri().path().to_string(); let method = req.method().to_string(); let agent = agent(req.headers()); let client = client(agent); let agent = agent(req.headers()).to_string(); // The CLI's version, which says how far each release has reached. let version = agent .strip_prefix("knotless/")@@ -628,7 +833,7 @@ pub async fn count_requests(usage: Arc<Usage>, req: Request, next: Next) -> Resp &path, &method, status, client, &agent, started.elapsed(), ); res@@ -678,9 +883,14 @@ impl App { }}/// MCP's sign-ins and tool calls, counted./// MCP's sign-ins and tool calls, counted, and why those that failed did.pub fn mcp_events(usage: Arc<Usage>) -> knotless_mcp::http::Events { Arc::new(move |e| usage.event(e.name, e.subject, e.ok, e.took, e.did)) Arc::new(move |e| { usage.event(e.name, e.subject, e.ok, e.took, e.did); if let Some(error) = e.error { usage.failed(e.name, e.subject, error); } })}/// A page drawn, as the page tells of it.@@ -883,6 +1093,82 @@ mod tests { assert_eq!(host(""), ""); } /// The index's slow statements reach the usage named, and none when /// it is gone. #[test] fn sinks_slow_statements() { let dir = tempfile::tempdir().unwrap(); let usage = Arc::new(Usage::open(&dir.path().join("usage.db")).unwrap()); sink(&usage); let sql = "SELECT 'sinks_slow_statements'"; slow_statement(sql, Duration::from_millis(150)); let counted = usage .statements .lock() .unwrap() .keys() .any(|(_, s)| s == sql); assert!(counted); drop(usage); slow_statement(sql, Duration::from_millis(150)); } #[test] fn fingerprints() { for (error, print) in [ ( "did:plc:abc is not a member of at://did:plc:x/dev.knotless.repo/r.", "<did> is not a member of <uri>.", ), ("no commit 5bf7b54f503f (yet)", "no commit <hex> (yet)"), ("took 1500 ms, (3) times", "took <n> ms, (<n>) times"), ("deadbeef and facade", "deadbeef and facade"), ("", ""), ] { assert_eq!(fingerprint(error), print, "{error}"); } } /// Past an hour's kept failures and crawlers, the new ones count /// together; one already kept goes on being. #[test] fn caps_each_hour() { let dir = tempfile::tempdir().unwrap(); let usage = Usage::open(&dir.path().join("usage.db")).unwrap(); let fast = Duration::ZERO; for i in 0..KEPT_PER_HOUR { let [a, b] = [i / 26, i % 26].map(|d| char::from(b'a' + d as u8)); usage.failed("mcp.tool", "x", &format!("failed as {a}{b}")); usage.tally(None, "/", "GET", 200, &format!("bot{i}"), fast); usage.statement(&format!("SELECT {a}{b}"), Duration::from_millis(100)); } usage.flush().unwrap(); usage.failed("mcp.tool", "x", " failed as aa "); usage.failed("mcp.tool", "x", "new"); usage.failed("mcp.tool", "x", "newer"); usage.tally(None, "/", "GET", 200, "bot0", fast); usage.tally(None, "/", "GET", 200, "botnew", fast); usage.statement("SELECT aa", Duration::from_millis(300)); usage.statement("SELECT new", Duration::from_millis(200)); usage.flush().unwrap(); let db = usage.db.lock().unwrap(); let n = |sql: &str| db.query_row(sql, [], |r| r.get::<_, i64>(0)).unwrap(); assert_eq!(n("SELECT count(*) FROM failure"), KEPT_PER_HOUR + 1); assert_eq!(n("SELECT n FROM failure WHERE error = '(others)'"), 2); assert_eq!(n("SELECT n FROM failure WHERE error = 'failed as aa'"), 2); assert_eq!(n("SELECT count(*) FROM agent"), KEPT_PER_HOUR + 1); assert_eq!(n("SELECT n FROM agent WHERE agent = '(others)'"), 1); assert_eq!(n("SELECT n FROM agent WHERE agent = 'bot0'"), 2); assert_eq!(n("SELECT count(*) FROM statement"), KEPT_PER_HOUR + 1); let row = |sql: &str| { let q = "SELECT n, ms, max FROM statement WHERE sql = ?1"; db.query_row(q, [sql], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))) .unwrap() }; assert_eq!(row("SELECT aa"), (2i64, 400i64, 300i64)); assert_eq!(row("(others)"), (1, 200, 200)); } #[test] fn says_what_was_not_written() { wrote(Ok(()));@@ -921,6 +1207,25 @@ mod tests { // Opened again, as each deploy does: nothing moves twice. drop(usage); let usage = Usage::open(&path).unwrap(); // What tells why, past the month and at its edge. { let db = usage.db.lock().unwrap(); for d in [ago(30), ago(29)] { let h = format!("{d}T23"); db.execute( "INSERT INTO statement VALUES (?1, 'SELECT 1', 1, 100, 100)", [&h], ) .unwrap(); db.execute( "INSERT INTO failure VALUES (?1, 'mcp.tool', 'x', 'no', 1)", [&h], ) .unwrap(); db.execute("INSERT INTO agent VALUES (?1, 'curl/8', 1)", [&h]) .unwrap(); } } usage.event("gateway.served", "", true, Duration::ZERO, Some("did:c")); usage.event("gateway.delegated", "", true, Duration::ZERO, Some("did:c")); usage.event("gateway.delegated", "", true, Duration::ZERO, Some("did:d"));@@ -941,6 +1246,12 @@ mod tests { .query_row("SELECT min(day) FROM account", [], |r| r.get(0)) .unwrap(); assert_eq!(oldest, ago(29)); for table in ["statement", "failure", "agent"] { let oldest: String = db .query_row(&format!("SELECT min(hour) FROM {table}"), [], |r| r.get(0)) .unwrap(); assert_eq!(oldest, format!("{}T23", ago(29)), "{table}"); } // Their funnels, counted: a and b started 40 days ago and a went // on; a again 35 days ago was not a start. let mut stmt = db@@ -1036,7 +1347,7 @@ mod tests { let usage = Usage::open(&dir.path().join("usage.db")).unwrap(); let route = "/xrpc/dev.knotless.awaitSpaceWrites"; let waited = Duration::from_secs(25); usage.tally(Some(route), route, "GET", 200, "cli", waited); usage.tally(Some(route), route, "GET", 200, "knotless/0.4.0", waited); let pending = usage.pending.lock().unwrap(); let tally = pending.values().next().unwrap(); assert_eq!((tally.n, tally.over_half, tally.over_two), (1, 0, 0));@@ -1052,7 +1363,7 @@ mod tests { "/api/a/r/tree", "GET", 200, "cli", "knotless/0.4.0", fast, ); usage.tally(@@ -1060,22 +1371,37 @@ mod tests { "/api/b/r/tree", "GET", 502, "cli", "knotless/0.4.0", Duration::from_secs(3), ); usage.tally(Some("/healthz"), "/healthz", "GET", 200, "other", fast); usage.tally(Some("/healthz"), "/healthz", "GET", 200, "curl/8.0", fast); usage.tally( None, "/a.test", "GET", 200, "browser", "Mozilla/5.0", Duration::from_millis(600), ); // A crawler, by its agent; the health check's was not counted. let crawler = format!("Googlebot/2.1 {}", "x".repeat(300)); usage.tally(None, "/b.test", "GET", 200, &crawler, fast); usage.tally(None, "/b.test", "GET", 200, &crawler, fast); // Slow statements, as written; and why a tool failed. usage.statement("SELECT a\n FROM b", Duration::from_millis(100)); usage.statement("SELECT a FROM b", Duration::from_millis(500)); usage.statement("SELECT c", Duration::from_millis(200)); usage.failed( "mcp.tool", "read_file", &format!(" no such file {}", "y".repeat(400)), ); usage.failed("mcp.tool", "rerun", "not the owner"); usage.failed("mcp.tool", "rerun", "not the owner"); { let pending = usage.pending.lock().unwrap(); // Apart by the status's class. assert_eq!(pending.len(), 3); // Apart by the status's class, and by client. assert_eq!(pending.len(), 4); let failed = pending.iter().find(|(k, _)| k.class == 5).unwrap().1; assert_eq!( *failed,@@ -1151,24 +1477,20 @@ mod tests { let now = &days[&today]; assert_eq!( (now.visitors, now.views, now.requests, now.errors), (2, 3, 3, 1) ); assert_eq!(report.routes[0].route, "/api/{authority}/{repo}/tree"); assert_eq!( ( report.routes[0].requests, report.routes[0].errors, report.routes[0].mean_ms ), (2, 1, 1501) (2, 3, 5, 1) ); // The pages, a browser's and the crawler's, then the API. assert_eq!(report.routes[0].route, "*"); let api = &report.routes[1]; assert_eq!(api.route, "/api/{authority}/{repo}/tree"); assert_eq!((api.requests, api.errors, api.mean_ms), (2, 1, 1501)); // All in this hour: the one that failed took three seconds. let hours: Vec<_> = report .hours .iter() .map(|h| (h.requests, h.errors, h.over_two)) .collect(); assert_eq!(hours, [(3, 1, 1)]); assert_eq!(hours, [(5, 1, 1)]); assert!(report.hours[0].hour.starts_with(&today)); let names = |c: &[Count]| c.iter().map(|c| (c.name.clone(), c.n)).collect::<Vec<_>>(); assert_eq!(@@ -1181,8 +1503,28 @@ mod tests { assert_eq!(names(&report.referrers), [("bsky.app".into(), 1)]); assert_eq!( names(&report.clients), [("cli".into(), 2), ("browser".into(), 1)] [("bot".into(), 2), ("cli".into(), 2), ("browser".into(), 1)] ); let agent = &report.agents[0]; assert_eq!((agent.name.len(), agent.n), (AGENT_MAX, 2)); assert!(agent.name.starts_with("Googlebot/2.1 x")); assert_eq!(report.agents.len(), 1); let statements: Vec<_> = report .statements .iter() .map(|s| (s.sql.as_str(), s.calls, s.mean_ms, s.max_ms)) .collect(); assert_eq!( statements, [("SELECT a FROM b", 2, 300, 500), ("SELECT c", 1, 200, 200)] ); let failures: Vec<_> = report .failures .iter() .map(|f| (f.subject.as_str(), f.error.len(), f.n)) .collect(); assert_eq!(failures, [("rerun", 13, 2), ("read_file", ERROR_MAX, 1)]); assert!(report.failures[1].error.starts_with("no such file y")); assert_eq!(hour(SystemTime::UNIX_EPOCH), "1970-01-01T00"); }}
chadtmiller.com/knotless