~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

An operator reads the appview's logs without SSH

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:serve/recent-logs
.changeset/admin-logs.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: minor
---
# `knotless admin logs` reads the lines the appview logged
An appview's operator can read the lines it logged, without SSH: `knotless admin logs --trace <id>` shows one request with the calls it made and how long each took, and `--message`, `--level` and `--min-ms` show the last of the rest. The MCP `logs` tool does the same. The appview keeps them across deploys, for 30 days at most, and logs each read with who asked.
crates/knotless/src/cli.rs+21−0
View file
@@ -315,6 +315,10 @@ const USAGE: &str = "usage: knotless <command>
admin reports [<limit>] moderate the appview, as an account
its KNOTLESS_ADMIN_DIDS names
admin stats [--days <n>] [--json] how the appview is used, day by day
admin logs [--trace <id>] [--message <m>] [--level <l>] [--min-ms <n>]
[--limit <n>] the lines the appview logged, oldest
first: a request's by its trace, or
the last of a message, level or time
describe <url> [--description <text>] [--link <url> [--title <text>]]...
[--topic <name>]... [--default-branch <name>]
change what a repository says it is,
@@ -2050,6 +2054,23 @@ async fn command(
crate::usage::render(&report, out)?;
}
}
"logs" => {
let mut asked = Vec::new();
while let Some(arg) = args.rest.next() {
let name = match arg.as_str() {
"--trace" => "trace",
"--message" => "message",
"--level" => "level",
"--min-ms" => "minMs",
"--limit" => "limit",
other => bail!("unexpected {other}"),
};
asked.push((name, args.next("<value>")?.to_string()));
}
for line in admin.logs(&asked).await? {
writeln!(out, "{line}")?;
}
}
other => bail!("unknown admin command {other}\n\n{USAGE}"),
}
}
crates/knotless/src/moderation.rs+11−0
View file
@@ -32,6 +32,7 @@ pub const UPDATE_STATUS: &str = "com.atproto.admin.updateSubjectStatus";
pub const GET_STATUS: &str = "com.atproto.admin.getSubjectStatus";
pub const LIST_REPORTS: &str = "dev.knotless.listReports";
pub const GET_USAGE: &str = "dev.knotless.getUsage";
pub const GET_LOGS: &str = "dev.knotless.getLogs";
/// The DID document service a moderation service takes reports at.
pub const LABELER: &str = "#atproto_labeler";
/// The labels that take something out of view.
@@ -302,6 +303,16 @@ impl Admin {
.await
}
/// The lines the appview logged that `asked` names, as
/// `(name, value)` pairs of `dev.knotless.getLogs`' parameters.
pub async fn logs(&self, asked: &[(&str, String)]) -> Result<Vec<Value>> {
let auth = self.auth(GET_LOGS).await?;
let params: Vec<(&str, &str)> = asked.iter().map(|(k, v)| (*k, v.as_str())).collect();
let answer: Value = self.xrpc.query(GET_LOGS, &params, Some(&auth)).await?;
let lines = answer["lines"].as_array().cloned().unwrap_or_default();
Ok(lines.into_iter().map(|l| l["line"].clone()).collect())
}
/// Takes `subject` (see [`subject`]) down, or puts it back.
pub async fn takedown(&self, ctx: &Ctx, target: &str, applied: bool) -> Result<Value> {
let subject = subject(ctx, target).await?;
crates/knotless/src/oauth.rs+1−0
View file
@@ -87,6 +87,7 @@ pub static SCOPE: LazyLock<String> = LazyLock::new(|| {
moderation::GET_STATUS,
moderation::LIST_REPORTS,
moderation::GET_USAGE,
moderation::GET_LOGS,
crate::invites::INVITE,
crate::invites::LIST,
crate::invites::OUTPUT,
crates/knotless/src/telemetry.rs+47−2
View file
@@ -14,7 +14,7 @@ use reqwest::header::{HeaderName, HeaderValue};
use std::collections::BTreeMap;
use std::fmt::{Display, Write as _};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{LazyLock, Mutex, OnceLock};
use std::sync::{Arc, LazyLock, Mutex, OnceLock, RwLock};
use std::time::{Duration, Instant, SystemTime};
pub use tracing::Level;
use tracing_subscriber::layer::SubscriberExt;
@@ -174,7 +174,37 @@ where
.flatten_event(true)
.with_current_span(false)
.with_span_list(false)
.with_writer(std::io::stderr)
.with_writer(|| Tee)
}
/// What is handed each JSON line as it is written, besides stderr.
type Keeper = Arc<dyn Fn(&[u8]) + Send + Sync>;
static KEEPER: RwLock<Option<Keeper>> = RwLock::new(None);
/// Hands each JSON line logged from now on to `keep` as well, as it goes
/// to stderr: for serve to keep them to read back. The last given wins.
pub fn keep_lines(keep: impl Fn(&[u8]) + Send + Sync + 'static) {
*KEEPER.write().unwrap() = Some(Arc::new(keep));
}
/// Writes each line to stderr, and to what [`keep_lines`] named. The
/// JSON layer writes an event whole, in one call.
struct Tee;
impl std::io::Write for Tee {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let keeper = KEEPER.read().unwrap().clone();
if let Some(keep) = keeper {
keep(buf);
}
std::io::stderr().write_all(buf)?;
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
std::io::stderr().flush()
}
}
fn slow_ms(given: Option<&str>) -> u64 {
@@ -679,6 +709,21 @@ mod tests {
}
}
/// Each line goes to stderr and to what keeps them.
#[test]
fn lines_are_kept_as_written() {
use std::io::Write as _;
let kept = Arc::new(Mutex::new(Vec::new()));
let into = kept.clone();
keep_lines(move |line| into.lock().unwrap().push(line.to_vec()));
let line = b"{\"message\":\"tee\"}\n";
let mut tee = Tee;
assert_eq!(tee.write(line).unwrap(), line.len());
tee.flush().unwrap();
let lines = kept.lock().unwrap();
assert!(lines.iter().any(|l| l == line));
}
#[test]
fn notes_at_each_level() {
let lines = logged(|| {
crates/knotless-mcp/src/lib.rs+39−0
View file
@@ -820,6 +820,22 @@ pub struct Stats {
pub days: Option<u32>,
}
#[derive(Deserialize, schemars::JsonSchema)]
pub struct Logs {
/// All of one request's lines, by the trace ID its answer's
/// `x-trace-id` and each of its lines carry.
pub trace: Option<String>,
/// Only lines with this message: `request`, `outgoing`, `sql`, `mcp
/// tool failed`, `git`.
pub message: Option<String>,
/// Only lines at this level or above: `error`, `warn`, `info`, `debug`.
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 by default, at most 500.
pub limit: Option<u32>,
}
#[derive(Deserialize, schemars::JsonSchema)]
pub struct Protect {
/// `handle/name`, `did/name` or `knotless://handle/name`.
@@ -2769,6 +2785,29 @@ impl Server {
)
}
#[tool(
description = "As the appview's operator, the lines it logged, oldest first, as JSON: all of one request's by its trace (the request, the calls it made to PDSes and others, its slow SQL statements, with their times), or the last of a message, a level or a duration. Ask for at least one. Kept across deploys, for 30 days at most, fewer at info. Each read is logged, with who asked."
)]
async fn logs(&self, Parameters(p): Parameters<Logs>) -> Result<CallToolResult, McpError> {
let asked: Vec<(&str, String)> = [
("trace", p.trace),
("message", p.message),
("level", p.level),
("minMs", p.min_ms.map(|n| n.to_string())),
("limit", p.limit.map(|n| n.to_string())),
]
.into_iter()
.filter_map(|(k, v)| v.map(|v| (k, v)))
.collect();
answer(
async {
let lines = Admin::new(&self.ctx).await?.logs(&asked).await?;
Ok(serde_json::json!({ "lines": lines }))
}
.await,
)
}
#[tool(
description = "As the appview's operator, how it is used, day by day: visitors and page views, requests by route and kind of client, MCP tool calls and why those that failed did, the index's slow statements, crawlers by user agent, clones by repository, CLI versions, sign-up funnels, and what the network wrote (repositories, pushes, pulls, issues, comments, checks, active accounts)."
)]
crates/knotless-mcp/tests/mcp.rs+8−1
View file
@@ -975,12 +975,15 @@ fn appview(rt: &tokio::runtime::Runtime, operators: &str) -> (String, Sandbox, T
let proxy = tls_proxy(listener.local_addr().unwrap().port());
let mut gateway = knotless_serve::gateway::Gateway::new(&base).unwrap();
gateway.oauth_url = proxy.url.clone();
let app = knotless_serve::App::build(
let mut app = knotless_serve::App::build(
ctx(&s),
std::time::Duration::ZERO,
Some(gateway),
Some(index),
);
// It keeps its logs, for its operators to read.
let logs = knotless_serve::logs::Logs::open(&s.path("logs.db"), knotless_serve::logs::KEEP);
std::sync::Arc::get_mut(&mut app).unwrap().logs = Some(std::sync::Arc::new(logs.unwrap()));
let router = knotless_serve::router(app.clone());
rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
rt.spawn(knotless_serve::firehose::watch(
@@ -1750,6 +1753,9 @@ fn agents_moderate() {
assert_eq!(reports["reports"].as_array().unwrap().len(), 2);
let stats = c.json("stats", json!({"days": 1}));
assert_eq!(stats["days"].as_array().unwrap().len(), 1, "{stats}");
// Its operator reads its logs.
let logs = c.json("logs", json!({"level": "warn", "min_ms": 1, "limit": 5}));
assert!(logs["lines"].is_array(), "{logs}");
let down = c.json("takedown", json!({"subject": issue}));
assert_eq!(down["takedown"]["applied"], true);
wait_until(|| items() == 0);
@@ -1766,6 +1772,7 @@ fn agents_moderate() {
(&o, "moderation_status", json!({"subject": issue})),
(&o, "list_reports", json!({})),
(&o, "stats", json!({})),
(&o, "logs", json!({"trace": "t"})),
] {
let (text, err) = who.call(tool, args.clone());
assert!(err, "{tool} {args}: {text}");
crates/knotless-serve/src/lib.rs+30−1
View file
@@ -28,6 +28,7 @@ pub mod jetstream;
pub mod languages;
pub mod live;
pub mod login;
pub mod logs;
pub mod moderation;
pub mod objects;
pub mod oci;
@@ -116,6 +117,8 @@ pub struct App {
registered: tokio::sync::Mutex<HashMap<String, Instant>>,
/// Set when this server counts how it is used, for its operator.
pub usage: Option<Arc<usage::Usage>>,
/// Set when this server keeps the lines it logs, for its operator.
pub logs: Option<Arc<logs::Logs>>,
/// The labels this server gives out, to those following them.
pub labels: tokio::sync::broadcast::Sender<(i64, serde_json::Value)>,
/// When each check still running was last read again from its PDS.
@@ -235,6 +238,7 @@ impl App {
told: tokio::sync::Notify::new(),
registered: tokio::sync::Mutex::default(),
usage: None,
logs: None,
labels: tokio::sync::broadcast::channel(256).0,
rechecked: Mutex::default(),
spent: Mutex::default(),
@@ -826,6 +830,7 @@ pub fn router(app: Arc<App>) -> Router {
.route("/_hit", post(usage::hit))
.route("/xrpc/dev.knotless.getUsage", get(usage::report))
.route("/xrpc/dev.knotless.getStats", get(usage::stats))
.route("/xrpc/dev.knotless.getLogs", get(logs::get_logs))
.route("/{authority}/{repo}/info/refs", get(smart::info_refs))
.route(
"/{authority}/{repo}/git-upload-pack",
@@ -979,6 +984,9 @@ pub struct Options {
pub stale: Option<Duration>,
/// Where to keep the counts of how this server is used.
pub usage: Option<std::path::PathBuf>,
/// Where to keep the lines it logs, and how many at info and below.
pub logs: Option<std::path::PathBuf>,
pub logs_keep: u64,
/// Only check that the index at this path is of this build's tables,
/// so that starting would keep it rather than rebuild it.
pub check_index: Option<std::path::PathBuf>,
@@ -1015,6 +1023,8 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> {
releases: None,
stale: None,
usage: None,
logs: None,
logs_keep: logs::KEEP,
check_index: None,
metrics: None,
bots: None,
@@ -1073,6 +1083,11 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> {
opts.stale = Some(Duration::from_secs(secs));
}
"--usage" => opts.usage = Some(value()?.into()),
"--logs" => opts.logs = Some(value()?.into()),
"--logs-keep" => {
let n = value()?.parse();
opts.logs_keep = n.map_err(|_| "--logs-keep takes a count of lines".to_string())?;
}
"--check-index" => opts.check_index = Some(value()?.into()),
"--metrics" => opts.metrics = Some(value()?),
"--bots" => opts.bots = Some(value()?.into()),
@@ -1080,7 +1095,7 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> {
"--oauth-url" => opts.oauth_url = Some(value()?),
other => {
return Err(format!(
"unexpected {other}\nusage: knotless-serve [--listen <addr>] [--ttl <secs>] [--firehose <wss://relay or pds>] [--jetstream <wss://host>] [--public-url <https://host>] [--oauth-url <https://host>] [--index <file>] [--backfill <https://pds or relay>] [--relay <https://relay>] [--labeler <did>]... [--default-runner <did>] [--client-minimum <version> [--client-minimum-reason <text>]] [--web <http://app>] [--releases <dir or knotless://repo>] [--stale <secs>] [--usage <file>] [--check-index <file>] [--metrics <addr>] [--bots <dir>] [--sign-up-pds <https://pds>]"
"unexpected {other}\nusage: knotless-serve [--listen <addr>] [--ttl <secs>] [--firehose <wss://relay or pds>] [--jetstream <wss://host>] [--public-url <https://host>] [--oauth-url <https://host>] [--index <file>] [--backfill <https://pds or relay>] [--relay <https://relay>] [--labeler <did>]... [--default-runner <did>] [--client-minimum <version> [--client-minimum-reason <text>]] [--web <http://app>] [--releases <dir or knotless://repo>] [--stale <secs>] [--usage <file>] [--logs <file> [--logs-keep <lines>]] [--check-index <file>] [--metrics <addr>] [--bots <dir>] [--sign-up-pds <https://pds>]"
));
}
}
@@ -1136,6 +1151,14 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
Some(path) => Some(Arc::new(usage::Usage::open(path)?)),
None => None,
};
let logs = match &opts.logs {
Some(path) => Some(Arc::new(logs::Logs::open(path, opts.logs_keep)?)),
None => None,
};
// Every line logged from here on, kept to read back.
if let Some(logs) = logs.clone() {
knotless::telemetry::keep_lines(move |line| logs.keep(line));
}
// The index's slow statements are counted in it.
usage.iter().for_each(usage::sink);
// MCP over HTTP, for any account not taken down, made once
@@ -1148,6 +1171,7 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
app.stale = opts.stale;
app.disk = Some(app.ctx.env.cache_dir.clone());
app.usage = usage.clone();
app.logs = logs.clone();
app.default_runner = opts.default_runner.clone();
app.client_minimum = client_minimum;
}
@@ -1205,6 +1229,9 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
tokio::spawn(usage::flushing(usage.clone(), usage::FLUSH));
routes = usage::counted(routes, usage);
}
if let Some(logs) = logs.clone() {
tokio::spawn(logs::flushing(logs, logs::FLUSH));
}
axum::serve(listener, telemetry::observed(routes))
.with_graceful_shutdown(async move {
let mut term =
@@ -1222,6 +1249,8 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
// What was counted since the last write.
usage.as_deref().map_or(Ok(()), usage::Usage::flush)?;
telemetry::said("stopped", None);
// The lines logged since the last write, that one with them.
logs.as_deref().map_or(Ok(()), logs::Logs::flush)?;
Ok(())
}
crates/knotless-serve/src/logs.rsnew+391−0
View file
@@ -0,0 +1,391 @@
//! The lines serve logs, kept to read back, for its operator:
//! `dev.knotless.getLogs`, `knotless admin logs` and MCP's `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.
pub fn flush(&self) -> rusqlite::Result<()> {
let pending = std::mem::take(&mut *self.pending.lock().unwrap());
crates/knotless-serve/src/logs.rsline 92
review.knotless.dev

The batch is taken before the transaction. If the transaction, an insert or the commit fails, these lines are gone. Restore them on failure.

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, 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 db = self.db.lock().unwrap();
let level = asked.level.as_deref().map_or(0, rank);
let stmt = match &asked.trace {
Some(_) => db.prepare_cached(
"SELECT line FROM line WHERE trace = ?1 AND level >= ?3 ORDER BY id DESC LIMIT ?5",
),
None => db.prepare_cached(
"SELECT line FROM line WHERE level >= ?3 AND (?2 IS NULL OR message = ?2)
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: 100 unless asked, at most 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 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);
}
}
crates/knotless-serve/tests/serve.rs+24−0
View file
@@ -326,6 +326,10 @@ fn binary_lifecycle() {
"127.0.0.1:0",
"--usage",
&s.path("usage.db").to_string_lossy(),
"--logs",
&s.path("logs.db").to_string_lossy(),
"--logs-keep",
"100000",
"--client-minimum",
"0.1.0",
"--client-minimum-reason",
@@ -409,6 +413,16 @@ fn binary_lifecycle() {
.status()
.unwrap();
assert!(child.wait().unwrap().success());
// Each line it logged was kept as it went to stderr, the last as it
// stopped; the requests with their traces.
let db = rusqlite::Connection::open(s.path("logs.db")).unwrap();
let kept = |sql: &str| db.query_row(sql, [], |r| r.get::<_, i64>(0)).unwrap();
assert_eq!(
kept("SELECT count(*) FROM line WHERE message = 'stopped'"),
1
);
let traced = "SELECT count(*) FROM line WHERE message = 'request' AND length(trace) = 32";
assert!(kept(traced) >= 1);
// Each request was counted, under the route it matched, and written as
// serve stopped; the health check was not.
let db = rusqlite::Connection::open(s.path("usage.db")).unwrap();
@@ -555,6 +569,16 @@ fn arguments() {
let o = args(&["--stale", "300"]).unwrap();
assert_eq!(o.stale, Some(Duration::from_secs(300)));
assert!(args(&["--stale", "x"]).is_err());
assert_eq!(
(o.logs.as_deref(), o.logs_keep),
(None, knotless_serve::logs::KEEP)
);
let o2 = args(&["--logs", "/data/logs.db", "--logs-keep", "9"]).unwrap();
assert_eq!(
(o2.logs.as_deref(), o2.logs_keep),
(Some(std::path::Path::new("/data/logs.db")), 9)
);
assert!(args(&["--logs-keep", "x"]).is_err());
assert_eq!(o.usage, None);
let o = args(&["--usage", "/var/lib/knotless/usage.db"]).unwrap();
assert_eq!(
crates/knotless-serve/tests/usage.rs+56−0
View file
deploy/docker-compose.yml+4−0
View file
deploy/README.md+20−0
View file
docs/reference/api.md+1−0
View file
docs/reference/cli.md+3−0
View file
docs/reference/mcp.md+1−1
View file
docs/observability.md+7−1
View file
lexicons/dev/knotless/getLogs.jsonnew+86−0
View file