johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! What this server logs of what it serves, and the numbers it keeps
//! (`knotless::telemetry`): a line for each request, with the trace its
//! calls hand on, and `/metrics` and `/log` on a port of their own.
use crate::App;
use axum::Router;
use axum::extract::{MatchedPath, Request, State};
use axum::http::{HeaderValue, StatusCode, header};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use axum::routing::{get, put};
use knotless::telemetry::{self, Fields, Level};
use rusqlite::Connection;
use rusqlite::trace::{TraceEvent, TraceEventCodes};
use std::collections::HashMap;
use std::sync::{Arc, LazyLock, Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime};
/// Why a response failed, for the line logged for it.
#[derive(Clone)]
pub struct Cause(pub String);
/// `status`, saying `message`, as the line logged for it says why.
pub fn failed(status: StatusCode, message: String) -> Response {
let mut res = (status, message.clone()).into_response();
res.extensions_mut().insert(Cause(message));
res
}
/// The route a request matched, as the routing finds it.
#[derive(Clone, Default)]
struct Route(Arc<OnceLock<String>>);
/// `routes`, each request through them logged and counted, in a trace
/// of its own or the one its `traceparent` names, which the answer gives
/// back as `x-trace-id`.
pub fn observed(routes: Router) -> Router {
routes
.route_layer(axum::middleware::from_fn(matched))
.layer(axum::middleware::from_fn(observe))
}
/// The routes that wait on purpose before they answer, for something to
/// happen or a while to pass: long polls, never slow however long they
/// take.
const WAITS: [&str; 2] = [
"/xrpc/dev.knotless.awaitSpaceWrites",
"/xrpc/dev.knotless.getDelegation",
];
/// Whether `route` is one of [`WAITS`].
pub fn waits(route: Option<&str>) -> bool {
route.is_some_and(|r| WAITS.contains(&r))
}
async fn matched(req: Request, next: Next) -> Response {
let path = req.extensions().get::<MatchedPath>();
if let (Some(route), Some(path)) = (req.extensions().get::<Route>(), path) {
let _ = route.0.set(path.as_str().to_string());
}
next.run(req).await
}
async fn observe(mut req: Request, next: Next) -> Response {
let started = Instant::now();
let route = Route::default();
req.extensions_mut().insert(route.clone());
let method = req.method().to_string();
let path = telemetry::short(req.uri().path());
let query = telemetry::query(&path, req.uri().query());
let agent = req.headers().get(header::USER_AGENT);
let agent = agent.map(HeaderValue::to_str).and_then(Result::ok);
let agent = agent.map(str::to_string);
let parent = req.headers().get("traceparent");
let trace = telemetry::trace_from(parent.and_then(|v| v.to_str().ok()));
req.extensions_mut().insert(telemetry::Trace(trace.clone()));
let mut res = telemetry::traced(trace.clone(), next.run(req)).await;
let id = HeaderValue::try_from(trace.as_str()).expect("hex is a header value");
res.headers_mut().insert("x-trace-id", id);
let cause = res.extensions().get::<Cause>();
let bytes = axum::body::HttpBody::size_hint(res.body()).exact();
let matched = route.0.get().map(String::as_str);
telemetry::served(&telemetry::Served {
route: matched,
method: &method,
path: &path,
query: query.as_deref(),
agent: agent.as_deref(),
status: res.status().as_u16(),
bytes,
took: started.elapsed(),
trace: &trace,
cause: cause.map(|c| c.0.as_str()),
waits: waits(matched),
});
res
}
/// Has `db` log each statement it runs that is slow ([`profiled`]), and
/// hand each to the planner, when they are planned ([`crate::plans`]).
pub fn watch(db: &Connection) {
let mut codes = TraceEventCodes::SQLITE_TRACE_PROFILE;
if crate::plans::planning() {
codes |= TraceEventCodes::SQLITE_TRACE_STMT;
}
db.trace_v2(codes, Some(traced));
}
fn traced(event: TraceEvent<'_>) {
if let TraceEvent::Stmt(stmt, _) = &event {
crate::plans::check(&stmt.sql());
}
if let TraceEvent::Profile(stmt, took) = event {
profiled(took, &stmt.sql());
}
}
/// Logs the statement `sql`, and counts it in usage, which took `took`, if that is a tenth
/// of a slow request or more (`KNOTLESS_SLOW_MS`, a second unless set):
/// a request seldom runs one statement alone, so one that slow is what
/// made it slow, and the line, in its trace, says which.
fn profiled(took: Duration, sql: &str) {
if telemetry::slow(took.saturating_mul(10)) {
telemetry::sql(sql, took);
crate::usage::slow_statement(sql, took);
}
}
/// What is answered on `--metrics`, a port kept off the proxy: the
/// numbers at `/metrics`, and `PUT /log` setting what is logged, as
/// `RUST_LOG` does.
pub fn admin(app: Arc<App>) -> Router {
Router::new()
.route("/metrics", get(metrics))
.route("/log", put(level))
.with_state(app)
}
/// Answers [`admin`] on `addr`, if given, and prints where.
pub async fn serve_admin(addr: Option<&str>, app: Arc<App>) -> anyhow::Result<()> {
let Some(addr) = addr else {
return Ok(());
};
let listener = tokio::net::TcpListener::bind(addr).await?;
let at = format!("http://{}", listener.local_addr()?);
println!("metrics on {at}");
said("metrics listening", Some(&at));
tokio::spawn(axum::serve(listener, admin(app)).into_future());
Ok(())
}
async fn metrics(State(app): State<Arc<App>>) -> Response {
let mut text = telemetry::metrics();
text.push_str(&feed_metrics(&app));
let kind = "text/plain; version=0.0.4";
([(header::CONTENT_TYPE, kind)], text).into_response()
}
async fn level(body: String) -> Response {
match telemetry::set_filter(body.trim()) {
Ok(()) => format!("logging {}\n", body.trim()).into_response(),
Err(err) => (StatusCode::BAD_REQUEST, err).into_response(),
}
}
/// Each feed's state, as `/metrics` gives it.
fn feed_metrics(app: &App) -> String {
let mut out = String::from("# TYPE knotless_feed_connected gauge\n");
out.push_str("# TYPE knotless_feed_events_total counter\n");
out.push_str("# TYPE knotless_feed_lag_seconds gauge\n");
for f in app.feeds() {
let name = &f.name;
let up = u8::from(f.connected);
out.push_str(&format!(
"knotless_feed_connected{{feed=\"{name}\"}} {up}\n"
));
out.push_str(&format!(
"knotless_feed_events_total{{feed=\"{name}\"}} {}\n",
f.events
));
let lag = lag_ms(name).map(|ms| ms as f64 / 1000.0);
let line = lag.map(|s| format!("knotless_feed_lag_seconds{{feed=\"{name}\"}} {s}\n"));
out.push_str(&line.unwrap_or_default());
}
out
}
/// How far behind now each feed was at its last event, in milliseconds.
static LAGS: LazyLock<Mutex<HashMap<String, u64>>> = LazyLock::new(Mutex::default);
/// Keeps how far behind now the event in the Jetstream `frame` is, by
/// the time it was made: its sequence numbers are its own, not times.
pub fn lagged(feed: &str, frame: &str) {
let frame: serde_json::Value = serde_json::from_str(frame).unwrap_or_default();
let time = frame["payload"]["time"].as_str().unwrap_or_default();
if let Ok(at) = humantime::parse_rfc3339_weak(time) {
let lag = SystemTime::now().duration_since(at).unwrap_or_default();
LAGS.lock()
.unwrap()
.insert(feed.to_string(), telemetry::millis(lag));
}
}
/// How far behind now the feed `name` was at its last event.
fn lag_ms(name: &str) -> Option<u64> {
LAGS.lock().unwrap().get(name).copied()
}
/// Logs a summary of what was counted every `every`, the first at once,
/// and how each feed is doing.
pub async fn summaries(app: Arc<App>, every: Duration) {
let mut tick = tokio::time::interval(every);
loop {
tick.tick().await;
telemetry::summarize();
for f in app.feeds() {
let fields = Fields {
feed: Some(&f.name),
connected: Some(f.connected),
count: Some(f.events),
lag_ms: lag_ms(&f.name),
duration_ms: Some(f.silent_seconds * 1000),
..Fields::default()
};
telemetry::note(Level::INFO, "summary: feed", &fields);
}
}
}
/// Logs that the feed `name`, from `url`, connected or dropped.
pub fn feed(name: &str, url: &str, connected: bool) {
const SAID: [(Level, &str); 2] = [
(Level::WARN, "feed disconnected"),
(Level::INFO, "feed connected"),
];
let (level, said) = SAID[usize::from(connected)];
let fields = Fields {
feed: Some(name),
source: Some(url),
..Fields::default()
};
telemetry::note(level, said, &fields);
}
/// Logs `said` at `level` of `did`, an account said to be deleted, as it
/// was asked after ([`crate::accounts::settle`]): the status its host
/// holds it at, the asks so far, and why none settled it.
pub fn account_asked(
level: Level,
said: &str,
did: &str,
status: Option<&str>,
tried: Option<u32>,
why: Option<String>,
) {
let fields = Fields {
did: Some(did),
status,
count: tried.map(u64::from),
error: why,
..Fields::default()
};
telemetry::note(level, said, &fields);
}
/// Logs, at warn, an event of `did`'s the index did not take: one the
/// index will miss until a sweep fills it.
pub fn indexed<T, E: std::fmt::Display>(did: &str, done: Result<T, E>) {
if let Err(err) = done {
let fields = Fields {
did: Some(did),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "event dropped: not indexed", &fields);
}
}
/// Logs, at warn, why a backfill could not read `did`: a later one tries
/// again.
pub fn backfill_failed(did: &str, err: &anyhow::Error) {
let fields = Fields {
did: Some(did),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "account not backfilled", &fields);
}
/// Seconds since the epoch, now.
pub fn now_secs() -> u64 {
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
now.unwrap_or_default().as_secs()
}
/// Logs, at warn, why a sync of the space `uri` failed, if it did: it is
/// tried again at the next.
pub fn space_synced(uri: &str, done: anyhow::Result<()>) {
if let Err(err) = done {
let fields = Fields {
source: Some(uri),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "space not synced", &fields);
}
}
/// Logs, at warn, why the authority of the space `uri` could not be asked
/// to tell this server of writes there, if it could not: it is synced
/// every so often all the same.
pub fn space_unheard<T>(uri: &str, asked: anyhow::Result<T>) {
if let Err(err) = asked {
let fields = Fields {
source: Some(uri),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "space not listened to", &fields);
}
}
/// Logs why `did` was not proven to read the space `uri`, if they were
/// not: at info when its authority refused them, which takes it out of
/// what they are listed as reading, or when their session's grant leaves
/// the space out, which only logging in again changes; at warn when it
/// could not be asked, which leaves that as it was, to ask again.
pub fn reader_checked<T>(uri: &str, did: &str, joined: &anyhow::Result<T>) {
if let Err(err) = joined {
let fields = Fields {
source: Some(uri),
did: Some(did),
error: Some(format!("{err:#}")),
..Fields::default()
};
if knotless::space::denied(err) {
telemetry::note(Level::INFO, "space reader refused", &fields);
} else if knotless::oauth::out_of_scope(err) {
telemetry::note(Level::INFO, "space reader out of scope", &fields);
} else {
telemetry::note(Level::WARN, "space reader not checked", &fields);
}
}
}
/// Logs, at warn, why `did`'s repo in the space `uri` could not be read on
/// through its oplog, if it could not: it is read whole instead.
pub fn read_whole(uri: &str, did: &str, read: &anyhow::Result<()>) {
if let Err(err) = read {
let fields = Fields {
source: Some(uri),
did: Some(did),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "space repo read whole", &fields);
}
}
/// Logs, at warn, why `did`'s repo in the space `uri` could not be read
/// whole, if it could not: it is read from the start of its oplog instead.
pub fn read_from_start(uri: &str, did: &str, read: &anyhow::Result<()>) {
if let Err(err) = read {
let fields = Fields {
source: Some(uri),
did: Some(did),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "space repo read from its oplog", &fields);
}
}
/// Logs, at warn, each record of `did`'s repo in the space `uri` that
/// could not be read, by its path, and why: each is left out of the
/// index, and the rest of the repo read all the same.
pub fn unreadable(uri: &str, did: &str, records: &[knotless::space_sync::Unreadable]) {
for record in records {
let path = format!("{}/{}", record.collection, record.rkey);
let fields = Fields {
source: Some(uri),
did: Some(did),
key: Some(&path),
error: Some(record.why.clone()),
..Fields::default()
};
telemetry::note(Level::WARN, "space record unreadable", &fields);
}
}
/// Logs, at warn, why following the labeler `did` stopped, if it did
/// for a reason: it is followed again at once.
pub fn labeler<T>(did: &str, done: anyhow::Result<T>) {
if let Err(err) = done {
let fields = Fields {
did: Some(did),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "labeler dropped", &fields);
}
}
/// Logs that the index was opened at startup, taking `took`.
pub fn opened(index: Option<&crate::index::Index>, took: Duration) {
if let Some(index) = index {
let fields = Fields {
rebuilt: Some(index.rebuilt),
duration_ms: Some(telemetry::millis(took)),
..Fields::default()
};
telemetry::note(Level::INFO, "index opened", &fields);
}
}
/// Fills the index from `service`, saying when it starts and how it
/// ended.
pub async fn sweep(
ctx: &knotless::env::Ctx,
index: &crate::index::Index,
service: &str,
sources: &crate::Sources,
) {
let started = Instant::now();
let source = Fields {
source: Some(service),
..Fields::default()
};
telemetry::note(Level::INFO, "index sweep started", &source);
let said = crate::backfilled(ctx, index, service, sources).await;
let fields = Fields {
reason: Some(&said),
duration_ms: Some(telemetry::millis(started.elapsed())),
..source
};
telemetry::note(Level::INFO, "index sweep ended", &fields);
}
/// Logs, at info, `message` about this server: its address, or that it
/// is stopping.
pub fn said(message: &str, source: Option<&str>) {
let fields = Fields {
source,
..Fields::default()
};
telemetry::note(Level::INFO, message, &fields);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn slow_statements() {
profiled(Duration::ZERO, "SELECT 1");
profiled(Duration::from_secs(5), "SELECT 1");
let db = Connection::open_in_memory().unwrap();
watch(&db);
db.execute_batch("SELECT 1").unwrap();
}
#[test]
fn reader_checks() {
let uri = "at://did:plc:o/space/dev.knotless.space/r";
reader_checked(uri, "did:plc:m", &Ok(()));
let no = knotless::xrpc::XrpcError {
status: 400,
error: "NotAuthorized".into(),
message: String::new(),
reset: None,
};
reader_checked::<()>(uri, "did:plc:m", &Err(no.into()));
reader_checked::<()>(
uri,
"did:plc:m",
&Err(anyhow::anyhow!("connection refused")),
);
let scope = anyhow::anyhow!("InvalidToken (403): Scope does not permit reading {uri}");
reader_checked::<()>(uri, "did:plc:m", &Err(scope));
}
#[test]
fn lags() {
assert_eq!(lag_ms("firehose"), None);
let ago = humantime::format_rfc3339_micros(SystemTime::now() - Duration::from_secs(3));
let frame = serde_json::json!({"payload": {"seq": 5, "time": ago.to_string()}});
lagged("jetstream-test", &frame.to_string());
assert!(lag_ms("jetstream-test").unwrap() >= 3000);
// A frame of no time leaves the last.
lagged("jetstream-test", "junk");
assert!(lag_ms("jetstream-test").unwrap() >= 3000);
}
#[test]
fn notes() {
// Nothing is set up to log here: each only has to run.
feed("jetstream", "wss://jetstream.test", true);
feed("jetstream", "wss://jetstream.test", false);
indexed::<(), _>("did:plc:x", Err("no PDS"));
indexed::<(), &str>("did:plc:x", Ok(()));
said("stopping", None);
labeler::<()>("did:web:labeler.test", Err(anyhow::anyhow!("gone")));
space_synced("at://did:plc:x/space/t/k", Err(anyhow::anyhow!("refused")));
space_unheard::<()>("at://did:plc:x/space/t/k", Err(anyhow::anyhow!("no")));
read_whole(
"at://did:plc:x/space/t/k",
"did:plc:y",
&Err(anyhow::anyhow!("no match")),
);
read_from_start(
"at://did:plc:x/space/t/k",
"did:plc:y",
&Err(anyhow::anyhow!("no match")),
);
let cut = knotless::space_sync::Unreadable {
collection: "a.b.c".into(),
rkey: "1".into(),
why: "a record that does not come to its CID".into(),
};
unreadable("at://did:plc:x/space/t/k", "did:plc:y", &[cut]);
opened(None, Duration::ZERO);
let res = failed(StatusCode::BAD_GATEWAY, "down".into());
assert_eq!(res.extensions().get::<Cause>().unwrap().0, "down");
}
#[tokio::test]
async fn admin_answers() {
let ctx = knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None));
let app = App::new(ctx, Duration::from_secs(5));
app.follows("jetstream", "wss://jetstream.test");
app.heard("jetstream", Some(1));
let now = humantime::format_rfc3339_micros(SystemTime::now()).to_string();
lagged(
"jetstream",
&serde_json::json!({"payload": {"time": now}}).to_string(),
);
let shown = metrics(State(app.clone())).await;
let body = axum::body::to_bytes(shown.into_body(), usize::MAX)
.await
.unwrap();
let body = String::from_utf8_lossy(&body);
assert!(body.contains("knotless_feed_connected{feed=\"jetstream\"} 0"));
assert!(body.contains("knotless_feed_lag_seconds{feed=\"jetstream\"}"));
// Logging is not set up in tests, so a level is refused.
let refused = level("debug".into()).await;
assert_eq!(refused.status(), StatusCode::BAD_REQUEST);
serve_admin(None, app.clone()).await.unwrap();
serve_admin(Some("127.0.0.1:0"), app.clone()).await.unwrap();
let summarizing = tokio::spawn(summaries(app, Duration::from_secs(60)));
tokio::task::yield_now().await;
summarizing.abort();
}
}