johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! How this server is used, for its operator: counted, and who only for
//! a month.
//!
//! - Every request but the health checks is counted by the hour, by its
//! surface (the pages, the API, git, xrpc, MCP, the
//! installer), the route axum matched (its pattern, not the path), its
//! method, whether it failed, how long it took, and the kind of client
//! that asked, from its `User-Agent`.
//! - The pages, once drawn in a browser, send `/_hit` the route that drew
//! them and the site that sent the visitor, if another. Each visitor is
//! counted once a day by a hash of their address, the last byte of it
//! dropped, and their `User-Agent`, salted with a key made for the day
//! and deleted once it is over: no address is kept, and no one can be
//! followed from one day to the next. A browser asking not to be
//! (`Sec-GPC: 1`) is not counted.
//! - What the handlers saw is counted by the hour, by its name and subject
//! (a tool, a repository, a version), never by who. Where the asker is
//! signed in (the pages' logins, MCP's sign-ins and tool calls, the
//! gateway's delegations, pushes and commits), their DID is kept with
//! the event's name and the day, to count how many accounts did each
//! and the funnels. It is deleted once that day is no longer one of the
//! last [`ACCOUNT_DAYS`] (30), today the last: at the first flush on the
//! 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
//! running that day, as two do for a moment in each deploy, hashes alike.
//! `dev.knotless.getUsage`, for an operator, answers it with what the index says
//! the network did, as [`knotless::usage::Report`].
use crate::{App, AppError};
use axum::Json;
use axum::body::Bytes;
use axum::extract::{MatchedPath, Query, Request, State};
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, Failure, Funnel, Hour, Network, Report, Route, Statement, Stats,
};
use rand_core::{OsRng, RngCore};
use rusqlite::{Connection, params};
use serde::Deserialize;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::net::IpAddr;
use std::path::Path;
use std::sync::{Arc, Mutex, RwLock, Weak};
use std::time::{Duration, Instant, SystemTime};
/// How often the counted requests are written.
pub const FLUSH: Duration = Duration::from_secs(15);
/// The most days `dev.knotless.getUsage` answers for.
const DAYS_MAX: u32 = 400;
/// How many days, today the last, the accounts in the events are kept:
/// the month `getUsage` counts each event's accounts over at most, as the
/// index counts the network's (`authorsMonth`).
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.
pending: Mutex<HashMap<Key, Tally>>,
/// Events counted since the last flush.
events: Mutex<HashMap<EventKey, Tally>>,
/// 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
/// in the days asked, how many went on to the other.
const FUNNELS: [(&str, &str); 2] = [
("gateway.delegated", "gateway.served"),
("mcp.signin", "mcp.tool"),
];
#[derive(Hash, PartialEq, Eq, Clone, Debug)]
struct Key {
hour: String,
surface: &'static str,
route: String,
method: String,
/// The status's class: 2 for 2xx, and so on.
class: u16,
client: &'static str,
}
/// An event's hour, name, subject, and whether it worked.
type EventKey = (String, &'static str, String, bool);
#[derive(Default, Debug, PartialEq)]
struct Tally {
n: u64,
ms: u64,
over_half: u64,
over_two: u64,
}
impl Usage {
pub fn open(path: &Path) -> rusqlite::Result<Usage> {
let db = Connection::open(path)?;
db.busy_timeout(Duration::from_secs(5))?;
let made = db.execute_batch(
"PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
CREATE TABLE IF NOT EXISTS salt (day TEXT PRIMARY KEY, value BLOB NOT NULL);
CREATE TABLE IF NOT EXISTS visitor (
day TEXT NOT NULL, hash BLOB NOT NULL, PRIMARY KEY (day, hash)) WITHOUT ROWID;
CREATE TABLE IF NOT EXISTS visitors (day TEXT PRIMARY KEY, n INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS request (
hour TEXT NOT NULL, surface TEXT NOT NULL, route TEXT NOT NULL,
method TEXT NOT NULL, class INTEGER NOT NULL, client TEXT NOT NULL,
n INTEGER NOT NULL, ms INTEGER NOT NULL,
over_half INTEGER NOT NULL, over_two INTEGER NOT NULL,
PRIMARY KEY (hour, surface, route, method, class, client)) WITHOUT ROWID;
CREATE TABLE IF NOT EXISTS view (
hour TEXT NOT NULL, route TEXT NOT NULL, referrer TEXT NOT NULL,
n INTEGER NOT NULL, PRIMARY KEY (hour, route, referrer)) WITHOUT ROWID;
CREATE TABLE IF NOT EXISTS event (
hour TEXT NOT NULL, name TEXT NOT NULL, subject TEXT NOT NULL,
ok INTEGER NOT NULL, n INTEGER NOT NULL, ms INTEGER NOT NULL,
PRIMARY KEY (hour, name, subject, ok)) WITHOUT ROWID;
CREATE TABLE IF NOT EXISTS account (
day TEXT NOT NULL, name TEXT NOT NULL, did TEXT NOT NULL,
first INTEGER NOT NULL,
PRIMARY KEY (day, name, did)) WITHOUT ROWID;
CREATE INDEX IF NOT EXISTS account_did ON account (name, did, day);
CREATE TABLE IF NOT EXISTS cohort (
day TEXT NOT NULL, start TEXT NOT NULL, goal TEXT NOT NULL,
started INTEGER NOT NULL, reached INTEGER NOT NULL,
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;
INSERT OR IGNORE INTO account SELECT day, name, did,
NOT EXISTS (SELECT 1 FROM actor b
WHERE b.name = a.name AND b.did = a.did AND b.day < a.day)
FROM actor a;
DROP TABLE actor;",
);
made?;
Ok(Usage {
db: Mutex::new(db),
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.
pub fn event(
&self,
name: &'static str,
subject: &str,
ok: bool,
took: Duration,
did: Option<&str>,
) {
let now = SystemTime::now();
let key = (hour(now), name, subject.to_string(), ok);
let mut events = self.events.lock().unwrap();
let tally = events.entry(key).or_default();
tally.n += 1;
tally.ms += took.as_millis() as u64;
drop(events);
let actor = did.map(|did| (day(now), name, did.to_string()));
self.actors.lock().unwrap().extend(actor);
}
/// Counts a request from the CLI at `version`.
fn cli(&self, version: Option<&str>) {
if let Some(v) = version {
self.event("cli.version", v, true, Duration::ZERO, None);
}
}
/// Counts a request to `route` (or, matching none, `path`) that took
/// `took` and answered `status`.
fn tally(
&self,
route: Option<&str>,
path: &str,
method: &str,
status: u16,
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 {
hour: hour(SystemTime::now()),
surface,
route,
method: method.to_string(),
class: status / 100,
client,
};
let ms = took.as_millis() as u64;
let mut pending = self.pending.lock().unwrap();
let tally = pending.entry(key).or_default();
tally.n += 1;
tally.ms += ms;
tally.over_half += u64::from(ms > 500) * slow;
tally.over_two += u64::from(ms > 2000) * slow;
}
/// Writes the requests counted; folds the days gone by into their
/// count of visitors, deleting their hashes and their salts; and folds
/// the accounts' days past [`ACCOUNT_DAYS`] into the funnels' counts,
/// deleting them.
pub fn flush(&self) -> rusqlite::Result<()> {
let pending = std::mem::take(&mut *self.pending.lock().unwrap());
let mut db = self.db.lock().unwrap();
let tx = db.transaction()?;
let mut stmt = tx.prepare_cached(
"INSERT INTO request VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
ON CONFLICT DO UPDATE SET n = n + excluded.n, ms = ms + excluded.ms,
over_half = over_half + excluded.over_half, over_two = over_two + excluded.over_two",
)?;
for (k, t) in &pending {
stmt.execute(params![
k.hour,
k.surface,
k.route,
k.method,
k.class,
k.client,
t.n,
t.ms,
t.over_half,
t.over_two
])?;
}
drop(stmt);
let events = std::mem::take(&mut *self.events.lock().unwrap());
let mut stmt = tx.prepare_cached(
"INSERT INTO event VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT DO UPDATE SET n = n + excluded.n, ms = ms + excluded.ms",
)?;
for ((hour, name, subject, ok), t) in &events {
stmt.execute(params![hour, name, subject, ok, t.n, t.ms])?;
}
drop(stmt);
let actors = std::mem::take(&mut *self.actors.lock().unwrap());
// An account's first of an event in the days kept starts a funnel.
let mut stmt = tx.prepare_cached(
"INSERT OR IGNORE INTO account SELECT ?1, ?2, ?3, NOT EXISTS (SELECT 1 FROM account
WHERE name = ?2 AND did = ?3 AND day < ?1)",
)?;
for (day, name, did) in &actors {
stmt.execute(params![day, name, did])?;
}
drop(stmt);
// Two serves may both fold: each adds what it finds, and the
// second finds the hashes, and the accounts, already gone.
let now = SystemTime::now();
let today = day(now);
let kept = day(now - Duration::from_secs(86_400 * (ACCOUNT_DAYS - 1)));
let mut stmt = tx.prepare_cached(
"INSERT INTO cohort SELECT day, ?1, ?2, count(*), sum(EXISTS (SELECT 1 FROM account b
WHERE b.name = ?2 AND b.did = a.did AND b.day >= a.day))
FROM account a WHERE name = ?1 AND first AND day < ?3 GROUP BY day
ON CONFLICT DO UPDATE SET started = started + excluded.started,
reached = reached + excluded.reached",
)?;
for (from, to) in FUNNELS {
stmt.execute(params![from, to, kept])?;
}
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",
[&today],
)?;
tx.execute("DELETE FROM visitor WHERE day < ?1", [&today])?;
tx.execute("DELETE FROM salt WHERE day < ?1", [&today])?;
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(
&self,
ip: Option<IpAddr>,
agent: &str,
route: &str,
referrer: &str,
) -> rusqlite::Result<()> {
let now = SystemTime::now();
let today = day(now);
let db = self.db.lock().unwrap();
let mut fresh = [0u8; 16];
OsRng.fill_bytes(&mut fresh);
db.execute(
"INSERT OR IGNORE INTO salt VALUES (?1, ?2)",
params![today, fresh],
)?;
let salt = db.query_row("SELECT value FROM salt WHERE day = ?1", [&today], |r| {
r.get::<_, Vec<u8>>(0)
});
let salt = salt?;
let hash = visitor(&salt, ip, agent);
db.execute(
"INSERT OR IGNORE INTO visitor VALUES (?1, ?2)",
params![today, hash],
)?;
db.execute(
"INSERT INTO view VALUES (?1, ?2, ?3, 1) ON CONFLICT DO UPDATE SET n = n + 1",
params![hour(now), route, referrer],
)?;
Ok(())
}
/// What was counted over the `days` days to today, into `report`.
/// Fills `days` with each one's requests, failures, views and visitors.
fn days(&self, days: &mut BTreeMap<String, Day>) -> rusqlite::Result<()> {
self.flush()?;
let since = days.keys().next().cloned().unwrap_or_default();
let db = self.db.lock().unwrap();
let mut stmt = db.prepare(
"SELECT substr(hour, 1, 10), sum(n), sum(CASE WHEN class = 5 THEN n ELSE 0 END)
FROM request WHERE hour >= ?1 GROUP BY 1",
)?;
for row in stmt.query_map([&since], counts)? {
let (d, requests, errors) = row?;
days.entry(d)
.and_modify(|d| (d.requests, d.errors) = (requests, errors));
}
let mut stmt = db.prepare(
"SELECT substr(hour, 1, 10), sum(n), 0 FROM view WHERE hour >= ?1 GROUP BY 1",
)?;
for row in stmt.query_map([&since], counts)? {
let (d, views, _) = row?;
days.entry(d).and_modify(|d| d.views = views);
}
// Today's, still hashes; the days before, folded.
let mut stmt = db.prepare(
"SELECT day, n, 0 FROM visitors WHERE day >= ?1
UNION ALL SELECT day, count(*), 0 FROM visitor WHERE day >= ?1 GROUP BY day",
)?;
for row in stmt.query_map([&since], counts)? {
let (d, visitors, _) = row?;
days.entry(d).and_modify(|d| d.visitors += visitors);
}
Ok(())
}
/// Fills `days`, as [`Usage::days`] does, and the rest of `report`.
fn read(&self, days: &mut BTreeMap<String, Day>, report: &mut Report) -> rusqlite::Result<()> {
self.days(days)?;
let since = days.keys().next().cloned().unwrap_or_default();
let db = self.db.lock().unwrap();
// The busiest hundred, and with them every write and every route
// that was slow, however few its requests: what someone waited on.
let mut stmt = db.prepare(
"SELECT surface, method, route, n, errors, mean, over_half, over_two FROM (
SELECT surface, method, route, sum(n) AS n,
sum(CASE WHEN class = 5 THEN n ELSE 0 END) AS errors,
sum(ms) / sum(n) AS mean, sum(over_half) AS over_half,
sum(over_two) AS over_two,
row_number() OVER (ORDER BY sum(n) DESC, route) AS rank
FROM request WHERE hour >= ?1 GROUP BY 1, 2, 3)
WHERE rank <= 100 OR method = 'POST' OR over_two > 0
ORDER BY n DESC, route",
)?;
report.routes = stmt
.query_map([&since], route)?
.collect::<rusqlite::Result<_>>()?;
let mut stmt = db.prepare(
"SELECT hour, sum(n), sum(CASE WHEN class = 5 THEN n ELSE 0 END), sum(over_two)
FROM request WHERE hour >= ?1 GROUP BY 1 ORDER BY 1",
)?;
report.hours = stmt
.query_map([&since], hour_row)?
.collect::<rusqlite::Result<_>>()?;
let tops = [
(
"SELECT route, sum(n) FROM view WHERE hour >= ?1 GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT 50",
&mut report.pages,
),
(
"SELECT referrer, sum(n) FROM view WHERE hour >= ?1 AND referrer != ''
GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT 50",
&mut report.referrers,
),
(
"SELECT client, sum(n) FROM request WHERE hour >= ?1 GROUP BY 1 ORDER BY 2 DESC, 1",
&mut report.clients,
),
];
for (sql, into) in tops {
let mut stmt = db.prepare(sql)?;
*into = stmt
.query_map([&since], count)?
.collect::<rusqlite::Result<_>>()?;
}
let mut stmt = db.prepare(
"SELECT e.name, e.subject, e.n, e.failed, e.ms / e.n,
(SELECT count(DISTINCT did) FROM account a WHERE a.name = e.name AND a.day >= ?1)
FROM (SELECT name, subject, sum(n) n, sum(CASE WHEN ok THEN 0 ELSE n END) failed,
sum(ms) ms FROM event WHERE hour >= ?1 GROUP BY 1, 2) e
ORDER BY 1, 3 DESC, 2 LIMIT 200",
)?;
report.events = stmt
.query_map([&since], event)?
.collect::<rusqlite::Result<_>>()?;
// An account's first day of `from` in the days kept, and whether
// it did `to` then or after; and the days since forgotten, counted.
let mut stmt = db.prepare(
"SELECT ?1, ?2,
count(*) + (SELECT coalesce(sum(started), 0) FROM cohort
WHERE start = ?1 AND goal = ?2 AND day >= ?3),
coalesce(sum(EXISTS (SELECT 1 FROM account b
WHERE b.name = ?2 AND b.did = a.did AND b.day >= a.day)), 0)
+ (SELECT coalesce(sum(reached), 0) FROM cohort
WHERE start = ?1 AND goal = ?2 AND day >= ?3)
FROM account a WHERE name = ?1 AND first AND day >= ?3",
)?;
for (from, to) in FUNNELS {
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)?))
}
fn count(row: &rusqlite::Row) -> rusqlite::Result<Count> {
Ok(Count {
name: row.get(0)?,
n: row.get(1)?,
})
}
fn event(row: &rusqlite::Row) -> rusqlite::Result<Event> {
Ok(Event {
name: row.get(0)?,
subject: row.get(1)?,
calls: row.get(2)?,
failed: row.get(3)?,
mean_ms: row.get(4)?,
accounts: row.get(5)?,
})
}
fn funnel(row: &rusqlite::Row) -> rusqlite::Result<Funnel> {
Ok(Funnel {
from: row.get(0)?,
to: row.get(1)?,
started: row.get(2)?,
reached: row.get(3)?,
})
}
fn route(row: &rusqlite::Row) -> rusqlite::Result<Route> {
Ok(Route {
surface: row.get(0)?,
method: row.get(1)?,
route: row.get(2)?,
requests: row.get(3)?,
errors: row.get(4)?,
mean_ms: row.get(5)?,
over_half: row.get(6)?,
over_two: row.get(7)?,
})
}
fn hour_row(row: &rusqlite::Row) -> rusqlite::Result<Hour> {
Ok(Hour {
hour: row.get(0)?,
requests: row.get(1)?,
errors: row.get(2)?,
over_two: row.get(3)?,
})
}
/// `YYYY-MM-DD` of `t`, in UTC.
fn day(t: SystemTime) -> String {
humantime::format_rfc3339_seconds(t).to_string()[..10].to_string()
}
/// `YYYY-MM-DDTHH` of `t`, in UTC.
fn hour(t: SystemTime) -> String {
humantime::format_rfc3339_seconds(t).to_string()[..13].to_string()
}
/// The surface a request was to, and the route it is counted under: the
/// pattern it matched. Matching none are the releases, a nested service,
/// and the pages, their assets apart from the rest. None for the health
/// checks and the counting itself.
fn surface(route: Option<&str>, path: &str) -> Option<(&'static str, String)> {
let Some(route) = route else {
let (surface, route) = match path {
p if p.starts_with("/releases/") => ("install", "/releases/*"),
p if p.starts_with("/_app/") => ("pages", "/_app/*"),
_ => ("pages", "*"),
};
return Some((surface, route.into()));
};
let surface = match route {
"/healthz" | "/_status" | "/_hit" | "/xrpc/dev.knotless.getUsage" => return None,
r if r.starts_with("/api/") => "api",
r if r.ends_with("/info/refs") || r.ends_with("/git-upload-pack") => "git",
r if r.starts_with("/xrpc/") => "xrpc",
r if r.starts_with("/mcp") || r.starts_with("/.well-known/oauth") => "mcp",
"/install.sh" => "install",
r if r.starts_with("/v2/") => "oci",
// A repository's releases, counted as a directory's are.
r if r.starts_with("/releases/") => return Some(("install", "/releases/*".into())),
_ => "other",
};
Some((surface, route.to_string()))
}
/// The kind of client `agent` names. The pages' own queries, as they draw
/// a page someone asked for, are `pages`, counted apart from every client.
fn client(agent: &str) -> &'static str {
let lower = agent.to_ascii_lowercase();
if lower.starts_with("git/") {
"git"
} else if lower.starts_with("knotless/") {
"cli"
} else if lower == "knotless-pages" {
"pages"
} else if bot(&lower) {
"bot"
} else if lower.starts_with("mozilla/") {
"browser"
} else {
"other"
}
}
/// Whether `agent`, in lower case, says it is a crawler or a script, as
/// most do, or is a browser no one is at.
fn bot(agent: &str) -> bool {
const WORDS: [&str; 17] = [
"bot",
// Google's crawler for all but search, which names itself so.
"googleother",
"crawl",
"spider",
"slurp",
"fetch",
"scrape",
"preview",
"headless",
"lighthouse",
"python",
"curl",
"wget",
"go-http",
"okhttp",
"axios",
"node",
];
agent.is_empty() || WORDS.iter().any(|w| agent.contains(w))
}
/// The visitor's address, as the proxy in front saw it: the last address
/// in `X-Forwarded-For`, which it wrote, rather than any the client did.
fn address(headers: &HeaderMap) -> Option<IpAddr> {
let forwarded = headers.get("x-forwarded-for")?.to_str().ok()?;
forwarded.rsplit(',').next()?.trim().parse().ok()
}
/// A visitor's hash for the day: of their address without its last byte
/// (its last 80 bits, for IPv6), and their `User-Agent`, salted.
fn visitor(salt: &[u8], ip: Option<IpAddr>, agent: &str) -> Vec<u8> {
let ip = match ip {
Some(IpAddr::V4(v4)) => {
let [a, b, c, _] = v4.octets();
vec![a, b, c]
}
Some(IpAddr::V6(v6)) => v6.octets()[..6].to_vec(),
None => vec![],
};
let mut hash = Sha256::new();
hash.update(salt);
hash.update(&ip);
hash.update([0xff]);
hash.update(agent.as_bytes());
hash.finalize()[..16].to_vec()
}
fn agent(headers: &HeaderMap) -> &str {
let agent = headers
.get(header::USER_AGENT)
.and_then(|v| v.to_str().ok());
agent.unwrap_or_default()
}
/// The middleware counting each request.
pub async fn count_requests(usage: Arc<Usage>, req: Request, next: Next) -> Response {
let route = req
.extensions()
.get::<MatchedPath>()
.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()).to_string();
// The CLI's version, which says how far each release has reached.
let version = agent
.strip_prefix("knotless/")
.map(|v| v.split(' ').next().unwrap_or(v));
usage.cli(version);
let started = Instant::now();
let res = next.run(req).await;
let status = res.status().as_u16();
usage.tally(
route.as_deref(),
&path,
&method,
status,
&agent,
started.elapsed(),
);
res
}
/// `routes`, each request to them counted.
pub fn counted(routes: axum::Router, usage: Arc<Usage>) -> axum::Router {
routes.layer(axum::middleware::from_fn(move |req, next| {
count_requests(usage.clone(), req, next)
}))
}
/// Writes the counted requests every `every`, and the first time at once,
/// which folds the days a stopped server left.
pub async fn flushing(usage: Arc<Usage>, every: Duration) {
loop {
wrote(usage.flush());
tokio::time::sleep(every).await;
}
}
/// Logs, at warn, why the counts were not written, if they were not:
/// they are written with the next.
fn wrote(done: rusqlite::Result<()>) {
if let Err(err) = done {
let fields = Fields {
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "usage not written", &fields);
}
}
impl App {
/// Counts an event, if this server counts usage: see [`Usage::event`].
pub fn event(
&self,
name: &'static str,
subject: &str,
ok: bool,
took: Duration,
did: Option<&str>,
) {
if let Some(usage) = &self.usage {
usage.event(name, subject, ok, took, did);
}
}
}
/// 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);
if let Some(error) = e.error {
usage.failed(e.name, e.subject, error);
}
})
}
/// A page drawn, as the page tells of it.
#[derive(Deserialize)]
struct Hit {
/// The route that drew it, as SvelteKit names it: `/[authority]/[repo]`.
route: String,
/// The page the visitor came from, if on another site.
#[serde(default)]
referrer: String,
}
fn usage(app: &App) -> Result<&Usage, AppError> {
let usage = app.usage.as_deref();
usage.ok_or_else(|| AppError(StatusCode::NOT_FOUND, "this server counts no usage".into()))
}
/// `POST /_hit`: a page drawn in a browser.
pub async fn hit(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<StatusCode, AppError> {
let usage = usage(&app)?;
let agent = agent(&headers);
let private = headers.get("sec-gpc").is_some_and(|v| v == "1");
if private || client(agent) != "browser" {
return Ok(StatusCode::NO_CONTENT);
}
let bad = |why: &str| AppError(StatusCode::BAD_REQUEST, why.into());
let hit: Hit = serde_json::from_slice(&body).map_err(|e| bad(&e.to_string()))?;
if !hit.route.starts_with('/') || hit.route.len() > NAME_MAX || hit.referrer.len() > NAME_MAX {
return Err(bad("a hit is a route, as /[authority], and a referrer"));
}
let referrer = host(&hit.referrer);
let viewed = usage.view(address(&headers), agent, &hit.route, referrer);
viewed
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
Ok(StatusCode::NO_CONTENT)
}
/// The host of `url`, or nothing.
fn host(url: &str) -> &str {
let rest = url.split_once("://").map_or("", |(_, rest)| rest);
rest.split(['/', '?', '#', ':']).next().unwrap_or_default()
}
#[derive(Deserialize)]
pub struct UsageParams {
days: Option<u32>,
}
/// `dev.knotless.getUsage?days=<n>`, for an operator: the last
/// `n` days, 14 if not given.
pub async fn report(
State(app): State<Arc<App>>,
headers: HeaderMap,
Query(p): Query<UsageParams>,
) -> Result<Json<Report>, AppError> {
crate::moderation::admin(&app, &headers, knotless::moderation::GET_USAGE).await?;
let n = p.days.unwrap_or(14).clamp(1, DAYS_MAX);
let mut report = Report::default();
let (days, network) = counted_days(&app, n, Some(&mut report))?;
report.days = days;
report.network = network;
Ok(Json(report))
}
/// The last `n` days, today (UTC) last, each with what usage and the
/// index counted in it, and the network as the index has it now. Given
/// `report`, the rest of what usage keeps goes in it.
fn counted_days(
app: &App,
n: u32,
report: Option<&mut Report>,
) -> Result<(Vec<Day>, Option<Network>), AppError> {
let now = SystemTime::now();
let ago = |k: u32| day(now - Duration::from_secs(86_400 * u64::from(k)));
let mut days: BTreeMap<String, Day> = (0..n)
.rev()
.map(|k| {
(
ago(k),
Day {
day: ago(k),
..Day::default()
},
)
})
.collect();
if let Some(usage) = app.usage.as_deref() {
let read = match report {
Some(report) => usage.read(&mut days, report),
None => usage.days(&mut days),
};
read.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
}
let mut network = None;
if let Some(index) = app.index.as_deref() {
let found = index.activity(&mut days, &ago(6), &ago(29));
let found = found
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
network = Some(found);
}
Ok((days.into_values().collect(), network))
}
/// How many days `dev.knotless.getStats` shows.
const STATS_DAYS: u32 = 30;
/// `dev.knotless.getStats`, for anyone: the network, and the last 30 days
/// of what it wrote and of the pages' visitors, as the public `/stats`
/// page shows them. Kept for five minutes.
pub async fn stats(State(app): State<Arc<App>>) -> Result<Response, AppError> {
let (days, network) = counted_days(&app, STATS_DAYS, None)?;
let stats = Stats {
days: days.into_iter().map(Into::into).collect(),
network,
};
let cache = [(header::CACHE_CONTROL, "public, max-age=300")];
Ok((cache, Json(stats)).into_response())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn surfaces() {
let s = |route: Option<&str>, path: &str| surface(route, path);
assert_eq!(s(None, "/a.test/r"), Some(("pages", "*".into())));
assert_eq!(
s(None, "/_app/immutable/x.js"),
Some(("pages", "/_app/*".into()))
);
assert_eq!(s(Some("/healthz"), "/healthz"), None);
for (route, surface) in [
("/api/{authority}/{repo}/tree", "api"),
("/{authority}/{repo}/info/refs", "git"),
("/{authority}/{repo}/git-upload-pack", "git"),
("/xrpc/dev.knotless.getFeed", "xrpc"),
("/mcp", "mcp"),
("/.well-known/oauth-authorization-server", "mcp"),
("/install.sh", "install"),
("/v2/{*rest}", "oci"),
("/card.png", "other"),
] {
assert_eq!(s(Some(route), ""), Some((surface, route.into())), "{route}");
}
assert_eq!(
s(Some("/releases/{version}/{file}"), "/releases/0.1.0/a"),
Some(("install", "/releases/*".into()))
);
assert_eq!(
s(None, "/releases/latest"),
Some(("install", "/releases/*".into()))
);
}
#[test]
fn clients() {
for (agent, kind) in [
("git/2.47.0", "git"),
("knotless/0.1.0", "cli"),
("knotless-pages", "pages"),
("node", "bot"),
("Mozilla/5.0 (compatible; Googlebot/2.1)", "bot"),
("Mozilla/5.0 (compatible; GoogleOther)", "bot"),
("Mozilla/5.0 (X11; Linux x86_64) HeadlessChrome/120", "bot"),
("curl/8.0", "bot"),
("", "bot"),
("Mozilla/5.0 (Macintosh) Safari/605", "browser"),
("Something/1", "other"),
] {
assert_eq!(client(agent), kind, "{agent}");
}
}
#[test]
fn addresses_and_hashes() {
let mut headers = HeaderMap::new();
assert_eq!(address(&headers), None);
headers.insert("x-forwarded-for", "6.6.6.6, 1.2.3.4".parse().unwrap());
let ip = address(&headers);
assert_eq!(ip, Some("1.2.3.4".parse().unwrap()));
// The last byte, or 80 bits, does not tell visitors apart.
let near = Some("1.2.3.200".parse().unwrap());
assert_eq!(visitor(b"s", ip, "a"), visitor(b"s", near, "a"));
assert_ne!(visitor(b"s", ip, "a"), visitor(b"t", ip, "a"));
assert_ne!(visitor(b"s", ip, "a"), visitor(b"s", ip, "b"));
let v6 = |s: &str| Some(s.parse().unwrap());
assert_eq!(
visitor(b"s", v6("2001:db8:1::1"), "a"),
visitor(b"s", v6("2001:db8:1:ff::9"), "a")
);
assert_eq!(visitor(b"s", None, "a").len(), 16);
assert_eq!(host("https://bsky.app/profile/x?y"), "bsky.app");
assert_eq!(host("http://localhost:5173/"), "localhost");
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(()));
wrote(Err(rusqlite::Error::InvalidQuery));
}
#[test]
fn forgets_accounts_after_a_month() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("usage.db");
let now = SystemTime::now();
let ago = |k: u64| day(now - Duration::from_secs(86_400 * k));
// As an older serve left it: who, for good.
{
let db = Connection::open(&path).unwrap();
db.execute_batch(
"CREATE TABLE actor (day TEXT NOT NULL, name TEXT NOT NULL, did TEXT NOT NULL,
PRIMARY KEY (day, name, did)) WITHOUT ROWID;",
)
.unwrap();
for (d, name, did) in [
(ago(40), "gateway.delegated", "did:a"),
(ago(40), "gateway.delegated", "did:b"),
(ago(39), "gateway.served", "did:a"),
(ago(35), "gateway.delegated", "did:a"),
(ago(29), "gateway.delegated", "did:c"),
] {
db.execute(
"INSERT INTO actor VALUES (?1, ?2, ?3)",
params![d, name, did],
)
.unwrap();
}
}
let usage = Usage::open(&path).unwrap();
// 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"));
usage.flush().unwrap();
usage.flush().unwrap();
{
let db = usage.db.lock().unwrap();
let gone: bool = db
.query_row(
"SELECT NOT EXISTS (SELECT 1 FROM sqlite_master WHERE name = 'actor')",
[],
|r| r.get(0),
)
.unwrap();
assert!(gone);
// No one past the month; the day at its edge still kept.
let oldest: String = db
.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
.prepare("SELECT day, start, goal, started, reached FROM cohort ORDER BY day")
.unwrap();
let cohorts: Vec<(String, String, String, u64, u64)> = stmt
.query_map([], |r| {
Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?))
})
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap();
assert_eq!(
cohorts,
[(
ago(40),
"gateway.delegated".into(),
"gateway.served".into(),
2,
1
)]
);
}
let mut days: BTreeMap<String, Day> = [ago(40), day(now)]
.into_iter()
.map(|d| {
(
d.clone(),
Day {
day: d,
..Day::default()
},
)
})
.collect();
let mut report = Report::default();
usage.read(&mut days, &mut report).unwrap();
// a and b, counted, c, who pushed since, and d, who has not; c
// again today was not a start.
let f = &report.funnels[0];
assert_eq!(
(f.from.as_str(), f.started, f.reached),
("gateway.delegated", 4, 2)
);
// The accounts are of the month only.
let delegated = report.events.iter().find(|e| e.name == "gateway.delegated");
assert_eq!(delegated.unwrap().accounts, 2);
}
/// Past the busiest hundred, a write and a slow route are kept, however
/// few their requests; a route that is neither is not.
#[test]
fn writes_and_slow_routes_are_kept() {
let dir = tempfile::tempdir().unwrap();
let usage = Usage::open(&dir.path().join("usage.db")).unwrap();
let today = day(SystemTime::now());
let hour = format!("{today}T00");
{
let db = usage.db.lock().unwrap();
let put = |route: &str, method: &str, n: i64, over_two: i64| {
db.execute(
"INSERT INTO request VALUES (?1, 'xrpc', ?2, ?3, 2, 'browser', ?4, ?4, 0, ?5)",
rusqlite::params![hour, route, method, n, over_two],
)
.unwrap();
};
for i in 0..100 {
put(&format!("/busy/{i}"), "GET", 1000 + i, 0);
}
put("/xrpc/dev.knotless.inviteToOrg", "POST", 1, 1);
put("/slow", "GET", 1, 1);
put("/quiet", "GET", 1, 0);
}
let mut days = BTreeMap::from([(
today.clone(),
Day {
day: today,
..Day::default()
},
)]);
let mut report = Report::default();
usage.read(&mut days, &mut report).unwrap();
let routes: Vec<_> = report.routes.iter().map(|r| r.route.as_str()).collect();
assert_eq!(routes.len(), 102);
assert_eq!(routes[0], "/busy/99");
assert_eq!(&routes[100..], ["/slow", "/xrpc/dev.knotless.inviteToOrg"]);
}
/// A long poll is counted, but never as slow: it waits on purpose.
#[test]
fn long_polls_are_not_slow() {
let dir = tempfile::tempdir().unwrap();
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, "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));
}
#[test]
fn counts_folds_and_reads() {
let dir = tempfile::tempdir().unwrap();
let usage = Usage::open(&dir.path().join("usage.db")).unwrap();
let fast = Duration::from_millis(3);
usage.tally(
Some("/api/{authority}/{repo}/tree"),
"/api/a/r/tree",
"GET",
200,
"knotless/0.4.0",
fast,
);
usage.tally(
Some("/api/{authority}/{repo}/tree"),
"/api/b/r/tree",
"GET",
502,
"knotless/0.4.0",
Duration::from_secs(3),
);
usage.tally(Some("/healthz"), "/healthz", "GET", 200, "curl/8.0", fast);
usage.tally(
None,
"/a.test",
"GET",
200,
"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, and by client.
assert_eq!(pending.len(), 4);
let failed = pending.iter().find(|(k, _)| k.class == 5).unwrap().1;
assert_eq!(
*failed,
Tally {
n: 1,
ms: 3000,
over_half: 1,
over_two: 1
}
);
}
let ip = Some("1.2.3.4".parse().unwrap());
usage
.view(ip, "Mozilla/5.0", "/[authority]", "bsky.app")
.unwrap();
usage
.view(ip, "Mozilla/5.0", "/[authority]/[repo]", "")
.unwrap();
usage
.view(
Some("5.6.7.8".parse().unwrap()),
"Mozilla/5.0",
"/[authority]",
"",
)
.unwrap();
usage.event("mcp.tool", "read_file", true, fast, Some("did:x"));
usage.event("mcp.tool", "read_file", false, fast, None);
// Yesterday's hashes, and its salt, as a serve left them.
{
let db = usage.db.lock().unwrap();
db.execute_batch(
"INSERT INTO visitor VALUES ('2000-01-01', x'01'), ('2000-01-01', x'02');
INSERT INTO salt VALUES ('2000-01-01', x'00');",
)
.unwrap();
}
usage.flush().unwrap();
// Folded twice, as two serves would: counted once.
usage.flush().unwrap();
{
let db = usage.db.lock().unwrap();
let one = |sql: &str| db.query_row(sql, [], |r| r.get::<_, i64>(0)).unwrap();
assert_eq!(one("SELECT n FROM visitors WHERE day = '2000-01-01'"), 2);
assert_eq!(
one("SELECT count(*) FROM visitor WHERE day = '2000-01-01'"),
0
);
assert_eq!(one("SELECT count(*) FROM salt"), 1);
}
let now = SystemTime::now();
let today = day(now);
let mut days = BTreeMap::from([
(
"2000-01-01".to_string(),
Day {
day: "2000-01-01".into(),
..Day::default()
},
),
(
today.clone(),
Day {
day: today.clone(),
..Day::default()
},
),
]);
let mut report = Report::default();
usage.read(&mut days, &mut report).unwrap();
let old = &days["2000-01-01"];
assert_eq!((old.visitors, old.views, old.requests), (2, 0, 0));
let now = &days[&today];
assert_eq!(
(now.visitors, now.views, now.requests, now.errors),
(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, [(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!(
names(&report.pages),
[
("/[authority]".into(), 2),
("/[authority]/[repo]".into(), 1)
]
);
assert_eq!(names(&report.referrers), [("bsky.app".into(), 1)]);
assert_eq!(
names(&report.clients),
[("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");
}
}