johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! The lines serve logs, kept to read back, for its operator:
//! `dev.knotless.getLogs`, `knotless admin logs` and MCP's `admin_logs`.
//!
//! Each JSON line serve writes to stderr ([`telemetry::keep_lines`]) is
//! kept here too, in a SQLite file beside the usage counts, written a
//! second at a time off the requests' way. Every serve running writes to
//! it, as two do for a moment in each deploy, so a deploy loses none:
//! the lines of a request a serve now gone answered are still here.
//!
//! What is kept is what the journal has (no query string, header, body
//! or address), and for no longer: lines at info and below, the most
//! numerous, only the last [`KEEP`] (`--logs-keep`); every line only
//! [`DAYS`] days. A read is itself logged, `logs read`, with who asked
//! and for what.
use crate::moderation::admin;
use crate::{App, AppError};
use axum::Json;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode};
use knotless::telemetry::{self, Fields, Level};
use rusqlite::{Connection, OptionalExtension, params};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::VecDeque;
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};
/// How often the lines logged are written.
pub const FLUSH: Duration = Duration::from_secs(1);
/// How often the oldest are let go.
pub const PRUNE: Duration = Duration::from_secs(60);
/// How many lines at info and below are kept, unless `--logs-keep` says.
pub const KEEP: u64 = 500_000;
/// How many days any line is kept: as long as the journal keeps them.
pub const DAYS: u64 = 30;
/// The most lines held between writes: a write that cannot keep up lets
/// the oldest go rather than grow without end.
const PENDING_MAX: usize = 50_000;
/// The most lines a read answers.
const LIMIT_MAX: u32 = 500;
pub struct Logs {
db: Mutex<Connection>,
pending: Mutex<VecDeque<Vec<u8>>>,
keep: u64,
}
impl Logs {
/// The lines kept at `path`, keeping `keep` at info and below.
pub fn open(path: &Path, keep: u64) -> rusqlite::Result<Logs> {
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 line (
id INTEGER PRIMARY KEY, at TEXT NOT NULL, level INTEGER NOT NULL,
message TEXT NOT NULL, trace TEXT, ms INTEGER, line TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS line_trace ON line (trace) WHERE trace IS NOT NULL;
CREATE INDEX IF NOT EXISTS line_level ON line (level, id);
CREATE INDEX IF NOT EXISTS line_message ON line (message, id);
CREATE INDEX IF NOT EXISTS line_at ON line (at);",
);
made?;
Ok(Logs {
db: Mutex::new(db),
pending: Mutex::default(),
keep,
})
}
/// Holds `line`, one JSON line as logged, for the next write.
pub fn keep(&self, line: &[u8]) {
let mut pending = self.pending.lock().unwrap();
if pending.len() >= PENDING_MAX {
pending.pop_front();
}
pending.push_back(line.to_vec());
}
/// Writes the lines held: each with its time, level, message, trace
/// and duration beside it, to find it by. A line that is not JSON is
/// kept as a message of its own. Lines a write fails to keep, as when
/// another serve holds the file past the busy timeout, are held again,
/// ahead of those logged since, for the next.
pub fn flush(&self) -> rusqlite::Result<()> {
let batch = std::mem::take(&mut *self.pending.lock().unwrap());
let written = self.write(&batch);
if written.is_err() {
let mut pending = self.pending.lock().unwrap();
let since = std::mem::replace(&mut *pending, batch);
pending.extend(since);
let over = pending.len().saturating_sub(PENDING_MAX);
pending.drain(..over);
}
written
}
/// Writes `pending` in one transaction: all of it or none.
fn write(&self, pending: &VecDeque<Vec<u8>>) -> rusqlite::Result<()> {
let mut db = self.db.lock().unwrap();
let tx = db.transaction()?;
let mut stmt =
tx.prepare_cached("INSERT INTO line VALUES (NULL, ?1, ?2, ?3, ?4, ?5, ?6)")?;
for line in pending {
let text = String::from_utf8_lossy(line);
let v: Value = serde_json::from_str(text.trim())
.unwrap_or_else(|_| json!({"message": text.trim()}));
let text = v.to_string();
let at = v["timestamp"].as_str().map_or_else(now, str::to_string);
let level = rank(v["level"].as_str().unwrap_or("INFO"));
let message = v["message"].as_str().unwrap_or_default();
let ms = v["duration_ms"].as_u64();
stmt.execute(params![
at,
level,
message,
v["trace_id"].as_str(),
ms,
text
])?;
}
drop(stmt);
tx.commit()
}
/// Lets go of what is past keeping: lines at info and below past the
/// last [`Logs::keep`] of them, and any line older than [`DAYS`].
pub fn prune(&self) -> rusqlite::Result<()> {
let db = self.db.lock().unwrap();
let since = SystemTime::now() - Duration::from_secs(86_400 * DAYS);
let since = humantime::format_rfc3339_seconds(since).to_string();
db.execute("DELETE FROM line WHERE at < ?1", [since])?;
let last: Option<i64> = db
.query_row(
"SELECT id FROM line WHERE level <= ?1 ORDER BY id DESC LIMIT 1 OFFSET ?2",
params![rank("INFO"), self.keep],
|r| r.get(0),
)
.optional()?;
if let Some(last) = last {
db.execute(
"DELETE FROM line WHERE level <= ?1 AND id <= ?2",
params![rank("INFO"), last],
)?;
}
Ok(())
}
/// The lines `asked` names, oldest first: all of one request's, by
/// its trace, up to [`LIMIT_MAX`] unless asked for fewer, or the last
/// of those of a message, a level or a duration.
pub fn read(&self, asked: &Asked) -> rusqlite::Result<Vec<Value>> {
let unasked = if asked.trace.is_some() {
LIMIT_MAX
} else {
100
};
let limit = asked.limit.unwrap_or(unasked).clamp(1, LIMIT_MAX);
let db = self.db.lock().unwrap();
let level = asked.level.as_deref().map_or(0, rank);
// A statement for each way of asking, each by its own index: one
// asking `?2 IS NULL OR message = ?2` could use none, and read
// every line kept, newest first, for a message seldom logged.
let stmt = match (&asked.trace, &asked.message) {
(Some(_), _) => db.prepare_cached(
"SELECT line FROM line WHERE trace = ?1 AND level >= ?3 ORDER BY id DESC LIMIT ?5",
),
(None, Some(_)) => db.prepare_cached(
"SELECT line FROM line INDEXED BY line_message
WHERE message = ?2 AND level >= ?3 AND (?4 IS NULL OR ms >= ?4)
ORDER BY id DESC LIMIT ?5",
),
(None, None) => db.prepare_cached(
"SELECT line FROM line WHERE level >= ?3 AND (?4 IS NULL OR ms >= ?4)
ORDER BY id DESC LIMIT ?5",
),
};
let mut stmt = stmt?;
let p = params![asked.trace, asked.message, level, asked.min_ms, limit];
let rows = stmt.query_map(p, |r| r.get::<_, String>(0))?;
let mut lines: Vec<Value> = rows
.map(|r| r.map(|text| serde_json::from_str(&text).unwrap_or_default()))
.collect::<rusqlite::Result<_>>()?;
lines.reverse();
Ok(lines)
}
}
/// A level, as a number to compare: `ERROR` the highest.
fn rank(level: &str) -> i64 {
match level.to_ascii_uppercase().as_str() {
"ERROR" => 5,
"WARN" => 4,
"INFO" => 3,
"DEBUG" => 2,
_ => 1,
}
}
fn now() -> String {
humantime::format_rfc3339_micros(SystemTime::now()).to_string()
}
/// Writes the lines held every `every`, and lets go of the oldest every
/// [`PRUNE`], at once the first time.
pub async fn flushing(logs: Arc<Logs>, every: Duration) {
let mut pruned: Option<std::time::Instant> = None;
loop {
wrote(logs.flush());
if pruned.is_none_or(|t| t.elapsed() >= PRUNE) {
wrote(logs.prune());
pruned = Some(std::time::Instant::now());
}
tokio::time::sleep(every).await;
}
}
/// Says, to stderr alone, why the lines were not written: they are
/// written with the next, unless too many wait.
fn wrote(done: rusqlite::Result<()>) {
if let Err(err) = done {
eprintln!("logs not kept: {err:#}");
}
}
/// What a read asks for: at least one of these.
#[derive(Deserialize, Default, Debug)]
#[serde(rename_all = "camelCase")]
pub struct Asked {
/// All of one request's lines.
pub trace: Option<String>,
/// Only lines with this message: `sql`, `outgoing`, `request`.
pub message: Option<String>,
/// Only lines at this level or above.
pub level: Option<String>,
/// Only lines that took this many milliseconds or more.
pub min_ms: Option<u64>,
/// How many, the last of them: at most 500, and unless asked, 100,
/// or for a trace, 500.
pub limit: Option<u32>,
}
impl Asked {
/// What it asks, as the line telling of the read says it.
fn said(&self) -> String {
let parts = [
self.trace.as_ref().map(|t| format!("trace={t}")),
self.message.as_ref().map(|m| format!("message={m}")),
self.level.as_ref().map(|l| format!("level={l}")),
self.min_ms.map(|ms| format!("minMs={ms}")),
];
parts.into_iter().flatten().collect::<Vec<_>>().join(" ")
}
}
/// `dev.knotless.getLogs`, for an operator: the lines `asked` names.
pub async fn get_logs(
State(app): State<Arc<App>>,
headers: HeaderMap,
Query(asked): Query<Asked>,
) -> Result<Json<Value>, AppError> {
let did = admin(&app, &headers, knotless::moderation::GET_LOGS).await?;
let Some(logs) = app.logs.as_deref() else {
let why = "this server keeps no logs";
return Err(AppError(StatusCode::NOT_FOUND, why.into()));
};
let said = asked.said();
if said.is_empty() {
let why = "ask for a trace, a message, a level or minMs";
return Err(AppError(StatusCode::BAD_REQUEST, why.into()));
}
let f = Fields {
did: Some(&did),
reason: Some(&said),
..Fields::default()
};
telemetry::note(Level::INFO, "logs read", &f);
// What was logged up to now, this read's own line with it.
let read = logs.flush().and_then(|()| logs.read(&asked));
let lines = read
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
let lines: Vec<Value> = lines
.into_iter()
.map(|line| json!({"line": line}))
.collect();
Ok(Json(json!({ "lines": lines })))
}
#[cfg(test)]
mod tests {
use super::*;
fn line(at: &str, level: &str, message: &str, trace: Option<&str>, ms: Option<u64>) -> Vec<u8> {
let v = json!({"timestamp": at, "level": level, "message": message,
"trace_id": trace, "duration_ms": ms});
format!("{v}\n").into_bytes()
}
#[test]
fn keeps_reads_and_lets_go() {
let dir = tempfile::tempdir().unwrap();
let logs = Logs::open(&dir.path().join("logs.db"), 3).unwrap();
let today = now();
let long_ago = "2000-01-01T00:00:00.000000Z";
logs.keep(&line(long_ago, "WARN", "outgoing", Some("t0"), Some(5000)));
for i in 0..5u64 {
logs.keep(&line(&today, "INFO", "request", Some("t1"), Some(i * 100)));
}
logs.keep(&line(&today, "WARN", "sql", Some("t1"), Some(400)));
logs.keep(&line(&today, "DEBUG", "outgoing", Some("t1"), Some(20)));
logs.keep(b"not json\n");
logs.flush().unwrap();
// Read twice: a write with nothing held writes nothing.
logs.flush().unwrap();
let read = |asked: Asked| logs.read(&asked).unwrap();
let messages = |lines: &[Value]| -> Vec<String> {
lines
.iter()
.map(|l| l["message"].as_str().unwrap_or("?").to_string())
.collect()
};
let trace = read(Asked {
trace: Some("t1".into()),
..Asked::default()
});
assert_eq!(trace.len(), 7);
assert_eq!(messages(&trace)[5..], ["sql", "outgoing"]);
let slow = read(Asked {
min_ms: Some(400),
..Asked::default()
});
assert_eq!(messages(&slow), ["outgoing", "request", "sql"]);
let warned = read(Asked {
level: Some("warn".into()),
limit: Some(1),
..Asked::default()
});
assert_eq!(messages(&warned), ["sql"]);
let plain = read(Asked {
message: Some("not json".into()),
..Asked::default()
});
assert_eq!(plain[0]["message"], "not json");
// Past keeping: the line of long ago, and all but the last three at
// info and below; what warned stays.
logs.prune().unwrap();
let left = read(Asked {
level: Some("trace".into()),
..Asked::default()
});
assert_eq!(messages(&left), ["request", "sql", "outgoing", "not json"]);
// With nothing past keeping, nothing goes.
logs.prune().unwrap();
assert_eq!(
read(Asked {
level: Some("trace".into()),
..Asked::default()
})
.len(),
4
);
}
/// A write that fails keeps what it held, ahead of what was logged
/// since, for the next; and still holds only so many.
#[test]
fn a_failed_write_keeps_its_lines() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("logs.db");
let logs = Logs::open(&path, KEEP).unwrap();
let other = Connection::open(&path).unwrap();
let today = now();
logs.keep(&line(&today, "INFO", "first", Some("t"), None));
other
.execute_batch("ALTER TABLE line RENAME TO away")
.unwrap();
assert!(logs.flush().is_err());
logs.keep(&line(&today, "INFO", "second", Some("t"), None));
other
.execute_batch("ALTER TABLE away RENAME TO line")
.unwrap();
logs.flush().unwrap();
let trace = logs
.read(&Asked {
trace: Some("t".into()),
..Asked::default()
})
.unwrap();
let messages: Vec<&str> = trace
.iter()
.map(|l| l["message"].as_str().unwrap())
.collect();
assert_eq!(messages, ["first", "second"]);
other
.execute_batch("ALTER TABLE line RENAME TO away")
.unwrap();
for _ in 0..PENDING_MAX {
logs.keep(b"{}");
}
assert!(logs.flush().is_err());
logs.keep(b"{}");
assert_eq!(logs.pending.lock().unwrap().len(), PENDING_MAX);
}
/// A trace reads up to the most a read answers unless asked for
/// fewer; anything else, the last 100.
#[test]
fn a_trace_reads_whole() {
let dir = tempfile::tempdir().unwrap();
let logs = Logs::open(&dir.path().join("logs.db"), KEEP).unwrap();
let today = now();
for _ in 0..150 {
logs.keep(&line(&today, "INFO", "request", Some("t"), None));
}
logs.flush().unwrap();
let read = |trace: Option<&str>, limit| {
let asked = Asked {
trace: trace.map(str::to_string),
message: Some("request".into()),
limit,
..Asked::default()
};
logs.read(&asked).unwrap().len()
};
assert_eq!(read(Some("t"), None), 150);
assert_eq!(read(Some("t"), Some(10)), 10);
assert_eq!(read(None, None), 100);
}
/// A write that cannot keep up lets the oldest go.
#[test]
fn holds_only_so_many() {
let dir = tempfile::tempdir().unwrap();
let logs = Logs::open(&dir.path().join("logs.db"), KEEP).unwrap();
for _ in 0..PENDING_MAX + 1 {
logs.keep(b"{}");
}
assert_eq!(logs.pending.lock().unwrap().len(), PENDING_MAX);
}
#[test]
fn ranks_and_says() {
let ranks: Vec<i64> = ["ERROR", "warn", "INFO", "DEBUG", "TRACE"]
.into_iter()
.map(rank)
.collect();
assert_eq!(ranks, [5, 4, 3, 2, 1]);
let asked = Asked {
trace: Some("t".into()),
message: Some("sql".into()),
level: Some("warn".into()),
min_ms: Some(100),
limit: None,
};
assert_eq!(asked.said(), "trace=t message=sql level=warn minMs=100");
assert_eq!(Asked::default().said(), "");
wrote(Ok(()));
wrote(Err(rusqlite::Error::InvalidQuery));
}
#[tokio::test]
async fn flushes_and_prunes_as_it_goes() {
let dir = tempfile::tempdir().unwrap();
let logs = Arc::new(Logs::open(&dir.path().join("logs.db"), KEEP).unwrap());
logs.keep(&line(&now(), "INFO", "flushed", None, None));
let going = tokio::spawn(flushing(logs.clone(), Duration::from_millis(10)));
tokio::time::sleep(Duration::from_millis(100)).await;
going.abort();
let asked = Asked {
message: Some("flushed".into()),
..Asked::default()
};
assert_eq!(logs.read(&asked).unwrap().len(), 1);
}
}