chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
Kept logs survive a failed write, and a trace reads whole
.changeset/logs-trace-whole.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: patch | ||
--- | ||
| ||
# `knotless admin logs --trace` reads up to 500 lines of the request | ||
| ||
A read by trace now answers up to 500 lines unless `--limit` says fewer, instead of the last 100, so a long request's first lines are not cut off. Lines serve fails to write to its log store, as when another serve holds the file during a deploy, are kept and written with the next batch instead of lost. | ||
crates/knotless-serve/src/logs.rs+95−7
| @@ -87,14 +87,29 @@ impl Logs { | ||
| ||
/// 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. | ||
/// 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 pending = std::mem::take(&mut *self.pending.lock().unwrap()); | ||
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 { | ||
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()})); | ||
| @@ -140,10 +155,15 @@ impl Logs { | ||
} | ||
| ||
/// The lines `asked` names, oldest first: all of one request's, by | ||
/// its trace, or the last of those of a message, a level or a | ||
/// duration. | ||
/// 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 limit = asked.limit.unwrap_or(100).clamp(1, LIMIT_MAX); | ||
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); | ||
let stmt = match &asked.trace { | ||
| @@ -216,7 +236,8 @@ pub struct Asked { | ||
pub level: Option<String>, | ||
/// Only lines that took this many milliseconds or more. | ||
pub min_ms: Option<u64>, | ||
/// How many, the last of them: 100 unless asked, at most 500. | ||
/// How many, the last of them: at most 500, and unless asked, 100, | ||
/// or for a trace, 500. | ||
pub limit: Option<u32>, | ||
} | ||
| ||
| @@ -343,6 +364,73 @@ mod tests { | ||
); | ||
} | ||
| ||
/// 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() { | ||
docs/reference/api.md+1−1
| @@ -319,7 +319,7 @@ token for another method or server, they answer 401; from anyone else, | ||
| Method | Auth | Parameters | Answer | | ||
| --- | --- | --- | --- | | ||
| `POST com.atproto.moderation.createReport` | service token, `lxm` this method | `subject`, `reasonType`, `reason?` | the stored report: `id`, `reasonType`, `subject`, `reportedBy`, `createdAt`, `reason?` | | ||
| `GET dev.knotless.getLogs` | admin | `trace`, or any of `message`, `level` (that or above), `minMs`; `limit` (default 100, clamped to 1 to 500); needs `--logs` | `lines`, oldest first, each `{line}`, the JSON line as logged; 400 asking for none of them, 404 keeping no logs | | ||
| `GET dev.knotless.getLogs` | admin | `trace`, or any of `message`, `level` (that or above), `minMs`; `limit` (default 500 for a `trace`, else 100; clamped to 1 to 500); needs `--logs` | `lines`, oldest first, each `{line}`, the JSON line as logged; 400 asking for none of them, 404 keeping no logs | | ||
| `GET dev.knotless.listReports` | admin | `limit` (default 50, clamped to 1 to 500), `cursor` (an `id`: reports before it) | `reports`, newest first, and `cursor` when the page is full | | ||
| `POST com.atproto.admin.updateSubjectStatus` | admin | `subject`, `takedown.applied` | `subject`, `takedown` | | ||
| `GET com.atproto.admin.getSubjectStatus` | admin | `did` or `uri` | `subject`, `takedown.applied` | | ||
docs/reference/cli.md+1−1
| @@ -340,7 +340,7 @@ MCP, `merge` takes `wait` (at most 300 s) and `force`. | ||
| `admin status <handle, url or at-uri>` | as the operator, a subject's moderation status: `taken down` or `up` | | ||
| `admin reports [<limit>]` | as the operator, the reports received, newest first (default 50): a line per report of `<id>`, `<reporter did>`, `<reason>`, `<subject>`, `<text>`, tab-separated | | ||
| `admin stats [--days <n>] [--json]` | as the operator, how the appview is used, day by day, from `dev.knotless.getUsage` (14 days by default) | | ||
| `admin logs [--trace <id>] [--message <m>] [--level <l>] [--min-ms <n>] [--limit <n>]` | as the operator, the lines the appview logged, from `dev.knotless.getLogs`, one JSON line each, oldest first: a request's by its trace, or the last (100 by default, at most 500) of a message, a level and above, or a time; one of them is needed | | ||
| `admin logs [--trace <id>] [--message <m>] [--level <l>] [--min-ms <n>] [--limit <n>]` | as the operator, the lines the appview logged, from `dev.knotless.getLogs`, one JSON line each, oldest first: a request's by its trace (up to 500, unless `--limit` says fewer), or the last (100 by default, at most 500) of a message, a level and above, or a time; one of them is needed | | ||
| `status [<server>] [--repo <handle>/<name>] [--json]` | a `knotless-serve`'s health, from `/_status`: its feeds, index, a mirror and its disk's room; the server defaults to `KNOTLESS_APPVIEW` | | ||
| ||
A report's reason is sent as `com.atproto.moderation.defs#reason<Name>`, | ||
lexicons/dev/knotless/getLogs.json+1−2
| @@ -44,8 +44,7 @@ | ||
"type": "integer", | ||
"minimum": 1, | ||
"maximum": 500, | ||
"default": 100, | ||
"description": "How many, the last of those asked for." | ||
"description": "How many, the last of those asked for. Unless given, 500 for a trace, so a request's lines come whole, and 100 otherwise." | ||
} | ||
} | ||
}, | ||