chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
knotless-serve: /_status tells of a disk running out of room
.changeset/status-disk.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: patch | ||
--- | ||
| ||
# `knotless status` shows the server's disk | ||
| ||
`knotless status` adds a `disk` line: how much room is left on the disk a knotless-serve keeps its mirrors on. Under 15% free the disk shows as `low` among the problems, and under 5% the server counts as unhealthy. | ||
crates/knotless/src/status.rs+69−3
| @@ -1,6 +1,7 @@ | ||
//! What knotless-serve's `/_status` answers: whether its feeds still bring | ||
//! events, its index answers, and a mirror can be read. 200 when all of it | ||
//! holds, 503 when not, for a monitor to poll and `knotless status` to show. | ||
//! events, its index answers, a mirror can be read, and its disk has room. | ||
//! 200 when all of it holds, 503 when not, for a monitor to poll and | ||
//! `knotless status` to show. | ||
| ||
use crate::env::Ctx; | ||
use anyhow::{Context, Result}; | ||
| @@ -23,6 +24,27 @@ pub struct Status { | ||
pub index: Option<Check>, | ||
/// The mirror asked about with `?repo=`, when one was. | ||
pub mirror: Option<MirrorCheck>, | ||
/// The disk its mirrors and caches are on, when it watches one. Absent | ||
/// from servers older than the field. | ||
#[serde(default)] | ||
pub disk: Option<Disk>, | ||
} | ||
| ||
/// The disk a server keeps its mirrors and caches on. | ||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] | ||
#[serde(rename_all = "camelCase")] | ||
pub struct Disk { | ||
/// The directory asked about. | ||
pub path: String, | ||
/// `ok`; `low`, under 15% free, which is told of among the problems | ||
/// but still answers 200; `full`, under 5% free, which fails; or | ||
/// `unknown`, when the disk could not be asked, which fails too. | ||
pub state: String, | ||
pub ok: bool, | ||
/// Bytes free to the server, and the disk's size. | ||
pub free_bytes: u64, | ||
pub total_bytes: u64, | ||
pub error: Option<String>, | ||
} | ||
| ||
/// A firehose or Jetstream followed. | ||
| @@ -114,6 +136,18 @@ pub fn render(base: &str, s: &Status, out: &mut dyn Write) -> std::io::Result<() | ||
let (ok, error) = (word(m.ok), detail(&m.error)); | ||
writeln!(out, "mirror\t{}\t{ok}{at}{packs}{error}", m.repo)?; | ||
} | ||
if let Some(d) = &s.disk { | ||
let room = match d.total_bytes { | ||
0 => String::new(), | ||
total => format!(", {} free of {}", gb(d.free_bytes), gb(total)), | ||
}; | ||
let state = match d.state.as_str() { | ||
"ok" => String::new(), | ||
state => format!(" ({state})"), | ||
}; | ||
let (ok, error) = (word(d.ok), detail(&d.error)); | ||
writeln!(out, "disk\t{}\t{ok}{state}{room}{error}", d.path)?; | ||
} | ||
Ok(()) | ||
} | ||
| ||
| @@ -128,6 +162,11 @@ fn detail(error: &Option<String>) -> String { | ||
.unwrap_or_default() | ||
} | ||
| ||
/// Bytes, in whole gigabytes. | ||
fn gb(bytes: u64) -> String { | ||
format!("{}G", bytes / 1_000_000_000) | ||
} | ||
| ||
/// A span of seconds, short. | ||
fn ago(secs: u64) -> String { | ||
match secs { | ||
| @@ -172,6 +211,14 @@ mod tests { | ||
commit: Some("0123456789abcdef".into()), | ||
packs: Some(3), | ||
}), | ||
disk: Some(Disk { | ||
path: "/data/cache".into(), | ||
state: "low".into(), | ||
ok: true, | ||
free_bytes: 18_500_000_000, | ||
total_bytes: 150_000_000_000, | ||
error: None, | ||
}), | ||
}; | ||
let text = |s: &Status| { | ||
let mut out = Vec::new(); | ||
| @@ -186,6 +233,7 @@ feed\tjetstream\tok\tconnected, last event 3s ago (stale after 10m), 7 events\tw | ||
feed\tfirehose\tFAILING\tdisconnected, last event 2h ago (stale after 10m), 7 events\twss://firehose | ||
index\tFAILING: file is not a database | ||
mirror\ta.test/r\tok at 0123456789ab, 3 packs | ||
disk\t/data/cache\tok (low), 18G free of 150G | ||
" | ||
); | ||
s = Status { | ||
| @@ -201,11 +249,29 @@ mirror\ta.test/r\tok at 0123456789ab, 3 packs | ||
commit: None, | ||
packs: None, | ||
}), | ||
disk: Some(Disk { | ||
path: "/gone".into(), | ||
state: "unknown".into(), | ||
ok: false, | ||
free_bytes: 0, | ||
total_bytes: 0, | ||
error: Some("no such directory".into()), | ||
}), | ||
}; | ||
assert_eq!( | ||
text(&s), | ||
"ok\thttps://x\tup 5s\nfeeds\tnone followed\nmirror\ta.test/r\tFAILING: no such repository\n" | ||
"ok\thttps://x\tup 5s\nfeeds\tnone followed\nmirror\ta.test/r\tFAILING: no such repository\ndisk\t/gone\tFAILING (unknown): no such directory\n" | ||
); | ||
s.disk = Some(Disk { | ||
path: "/data".into(), | ||
state: "ok".into(), | ||
ok: true, | ||
free_bytes: 90_000_000_000, | ||
total_bytes: 150_000_000_000, | ||
error: None, | ||
}); | ||
assert!(text(&s).ends_with("disk\t/data\tok, 90G free of 150G\n")); | ||
s.disk = None; | ||
s.mirror = None; | ||
assert_eq!(text(&s), "ok\thttps://x\tup 5s\nfeeds\tnone followed\n"); | ||
} | ||
crates/knotless-serve/src/lib.rs+5−0
| @@ -87,6 +87,9 @@ pub struct App { | ||
/// How long a feed may stay silent before `/_status` calls it dead, | ||
/// rather than each feed's own default. | ||
pub stale: Option<Duration>, | ||
/// The directory whose disk `/_status` watches for room: the cache's, | ||
/// where mirrors go. None in tests, which share whatever disk they get. | ||
pub disk: Option<std::path::PathBuf>, | ||
started: Instant, | ||
/// Set when this server also takes writes for owners who delegate them. | ||
pub gateway: Option<gateway::Gateway>, | ||
| @@ -220,6 +223,7 @@ impl App { | ||
following: AtomicBool::new(false), | ||
feeds: Mutex::default(), | ||
stale: None, | ||
disk: None, | ||
started: Instant::now(), | ||
gateway, | ||
index: index.map(Arc::new), | ||
| @@ -1132,6 +1136,7 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> { | ||
let client_minimum = opts.client_minimum.clone().map(|m| (m, reason)); | ||
if let Some(app) = Arc::get_mut(&mut app) { | ||
app.stale = opts.stale; | ||
app.disk = Some(app.ctx.env.cache_dir.clone()); | ||
app.usage = usage.clone(); | ||
app.default_runner = opts.default_runner.clone(); | ||
app.client_minimum = client_minimum; | ||
crates/knotless-serve/src/status.rs+113−3
| @@ -7,7 +7,10 @@ | ||
//! events is enough, since each brings all the index needs; | ||
//! - the index answers a read; | ||
//! - with `?repo=<handle or did>/<name>`, that repository's mirror reads as | ||
//! its code page does: its default branch's commit and tree. | ||
//! its code page does: its default branch's commit and tree; | ||
//! - the disk its mirrors and caches are on has room: under 15% free is told | ||
//! of, under 5% fails. A full disk fails every push a mirror syncs, and | ||
//! answers 502 for every page that needs one. | ||
//! | ||
//! 200 when all of that holds, else 503, and JSON saying which part failed. | ||
//! `/healthz` stays the liveness check: a dead feed upstream is no reason | ||
| @@ -18,7 +21,8 @@ use axum::Json; | ||
use axum::extract::{Query, State}; | ||
use axum::http::{StatusCode, header}; | ||
use axum::response::{IntoResponse, Response}; | ||
use knotless::status::{Check, Feed, MirrorCheck, Status}; | ||
use knotless::status::{Check, Disk, Feed, MirrorCheck, Status}; | ||
use std::path::Path; | ||
use std::sync::Arc; | ||
use std::sync::atomic::Ordering; | ||
use std::time::{Duration, Instant}; | ||
| @@ -44,6 +48,58 @@ pub fn stale_after(name: &str) -> Duration { | ||
} | ||
} | ||
| ||
/// The share of the disk free below which it is low: told of among the | ||
/// problems, still 200. Prod's 150 GB disk then has 22 GB left, a day or | ||
/// more of deploys and runs, time to clear it before it fills. | ||
pub const DISK_LOW: f64 = 0.15; | ||
| ||
/// The share free below which the disk counts as full and `/_status` | ||
/// fails: 7.5 GB of prod's, a single cargo build's worth. | ||
pub const DISK_FULL: f64 = 0.05; | ||
| ||
/// How a disk with `free` of its `total` bytes is: `ok`, `low` or `full`. | ||
pub fn disk_state(free: u64, total: u64) -> &'static str { | ||
let share = free as f64 / total.max(1) as f64; | ||
match share { | ||
s if s < DISK_FULL => "full", | ||
s if s < DISK_LOW => "low", | ||
_ => "ok", | ||
} | ||
} | ||
| ||
/// The disk `path` is on, asked of its nearest directory that exists. | ||
pub(crate) fn disk(path: &Path) -> Disk { | ||
let at = path.ancestors().find(|p| p.exists()).unwrap_or(path); | ||
let mut disk = Disk { | ||
path: path.display().to_string(), | ||
state: "unknown".into(), | ||
ok: false, | ||
free_bytes: 0, | ||
total_bytes: 0, | ||
error: None, | ||
}; | ||
match rustix::fs::statvfs(at) { | ||
Ok(space) => { | ||
disk.free_bytes = space.f_bavail * space.f_frsize; | ||
disk.total_bytes = space.f_blocks * space.f_frsize; | ||
let state = disk_state(disk.free_bytes, disk.total_bytes); | ||
(disk.state, disk.ok) = (state.into(), state != "full"); | ||
} | ||
Err(err) => disk.error = Some(format!("{}: {err}", at.display())), | ||
} | ||
disk | ||
} | ||
| ||
/// What `/_status` says of a disk that is not `ok`, among its problems. | ||
fn disk_problem(disk: &Disk) -> Option<String> { | ||
let free = 100 * disk.free_bytes / disk.total_bytes.max(1); | ||
match (disk.state.as_str(), &disk.error) { | ||
("ok", _) => None, | ||
(_, Some(err)) => Some(format!("disk: {err}")), | ||
(state, None) => Some(format!("disk {state}: {free}% free")), | ||
} | ||
} | ||
| ||
/// A feed as its watcher last told of it. | ||
pub(crate) struct Followed { | ||
url: String, | ||
| @@ -155,13 +211,17 @@ impl App { | ||
.and_then(|m| Some(format!("mirror {}: {}", m.repo, m.error.as_ref()?))), | ||
]; | ||
problems.extend(failed.into_iter().flatten()); | ||
let disk = self.disk.as_deref().map(disk); | ||
problems.extend(disk.as_ref().and_then(disk_problem)); | ||
let parts = index.as_ref().is_none_or(|i| i.ok) && mirror.as_ref().is_none_or(|m| m.ok); | ||
Status { | ||
ok: fed && index.as_ref().is_none_or(|i| i.ok) && mirror.as_ref().is_none_or(|m| m.ok), | ||
ok: fed && parts && disk.as_ref().is_none_or(|d| d.ok), | ||
problems, | ||
uptime_seconds: self.started.elapsed().as_secs(), | ||
feeds, | ||
index, | ||
mirror, | ||
disk, | ||
} | ||
} | ||
} | ||
| @@ -306,6 +366,56 @@ mod tests { | ||
assert_eq!(app.feeds()[0].stale_after_seconds, FIREHOSE_STALE.as_secs()); | ||
} | ||
| ||
#[test] | ||
fn disk_thresholds() { | ||
assert_eq!(disk_state(100, 100), "ok"); | ||
assert_eq!(disk_state(15, 100), "ok"); | ||
assert_eq!(disk_state(14, 100), "low"); | ||
assert_eq!(disk_state(5, 100), "low"); | ||
assert_eq!(disk_state(4, 100), "full"); | ||
assert_eq!(disk_state(0, 0), "full"); | ||
let at = |state: &str, free: u64, error: Option<&str>| Disk { | ||
path: "/data".into(), | ||
state: state.into(), | ||
ok: state != "full", | ||
free_bytes: free, | ||
total_bytes: 100, | ||
error: error.map(String::from), | ||
}; | ||
assert_eq!(disk_problem(&at("ok", 50, None)), None); | ||
assert_eq!( | ||
disk_problem(&at("low", 12, None)).unwrap(), | ||
"disk low: 12% free" | ||
); | ||
assert_eq!( | ||
disk_problem(&at("full", 3, None)).unwrap(), | ||
"disk full: 3% free" | ||
); | ||
assert_eq!( | ||
disk_problem(&at("unknown", 0, Some("gone"))).unwrap(), | ||
"disk: gone" | ||
); | ||
} | ||
| ||
#[tokio::test] | ||
async fn disk_asked() { | ||
// A directory not made yet is asked of the nearest that is. | ||
let dir = tempfile::tempdir().unwrap(); | ||
let mut app = app(None); | ||
let path = dir.path().join("cache/serve"); | ||
Arc::get_mut(&mut app).unwrap().disk = Some(path.clone()); | ||
let d = app.status(None).await.disk.unwrap(); | ||
assert_eq!(d.path, path.display().to_string()); | ||
assert!(d.total_bytes > 0 && d.free_bytes <= d.total_bytes, "{d:?}"); | ||
assert_eq!(d.state, disk_state(d.free_bytes, d.total_bytes)); | ||
assert_eq!(d.error, None); | ||
| ||
// Nowhere to ask: unknown, and failing. | ||
let d = disk(Path::new("nowhere/at/all")); | ||
assert_eq!((d.state.as_str(), d.ok), ("unknown", false)); | ||
assert!(d.error.unwrap().contains(": ")); | ||
} | ||
| ||
#[tokio::test] | ||
async fn index() { | ||
let dir = tempfile::tempdir().unwrap(); | ||
crates/knotless-serve/Cargo.toml+1−0
| @@ -18,6 +18,7 @@ jiff = { version = "0.2.38", features = ["tzdb-bundle-always"] } | ||
flate2.workspace = true | ||
futures.workspace = true | ||
humantime.workspace = true | ||
rustix.workspace = true | ||
serde.workspace = true | ||
serde_json.workspace = true | ||
tokio.workspace = true | ||
deploy/README.md+10−1
| @@ -581,6 +581,13 @@ the host to poll: 200 when it is, 503 and what failed when not. | ||
and tree are read as its code page reads them, so a mirror whose packs | ||
can no longer be read (as when one outgrew its open handle's pack slots) | ||
fails here, and its pack count shows. | ||
- **Disk**: the one serve's cache, and with it the mirrors, is on | ||
(`KNOTLESS_CACHE_DIR`, `/data/cache`). Under 15% free it is `low`, a | ||
problem told of while the answer stays 200; under 5% it is `full` and | ||
the answer is 503. On 2026-10-31 that disk filled and every page that | ||
needed a mirror answered 502 (`No space left on device`); 15% of its | ||
150 GB is a day or more of deploys and runs to clear it in. A keyword | ||
monitor (step 4 below) shows a `low` disk before it fails. | ||
| ||
It takes milliseconds, so polling it every minute costs nothing. A dead | ||
upstream feed is not a reason to restart serve, so the compose healthcheck | ||
| @@ -604,7 +611,9 @@ curl -s 'https://knotless.dev/_status?repo=chadtmiller.com/knotless' | jq | ||
], | ||
"index": {"ok": true, "error": null}, | ||
"mirror": {"repo": "chadtmiller.com/knotless", "ok": true, "error": null, | ||
"commit": "b1fbf45…", "packs": 3} | ||
"commit": "b1fbf45…", "packs": 3}, | ||
"disk": {"path": "/data/cache", "state": "ok", "ok": true, "freeBytes": 41203998720, | ||
"totalBytes": 160982630400, "error": null} | ||
} | ||
``` | ||
| ||
docs/design/serve.md+3−2
| @@ -1173,8 +1173,9 @@ no reason to restart. | ||
| ||
`/_status` says whether the server is doing its job: each feed is connected and | ||
has brought an event recently, the index answers a read, and with `?repo=` that | ||
repository's mirror reads as its code page would. It answers 200 when all hold | ||
and 503 otherwise. One feed bringing events is enough, though dead feeds are | ||
repository's mirror reads as its code page would, and the disk its cache is on | ||
has room: under 15% free it is told of, under 5% it fails. It answers 200 when | ||
all hold and 503 otherwise. One feed bringing events is enough, though dead feeds are | ||
still listed. | ||
[../../deploy/README.md#monitoring](../../deploy/README.md#monitoring) says | ||
what each part means. | ||
docs/reference/cli.md+1−1
| @@ -314,7 +314,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) | | ||
| `status [<server>] [--repo <handle>/<name>] [--json]` | a `knotless-serve`'s health, from `/_status`: its feeds, index and a mirror; the server defaults to `KNOTLESS_APPVIEW` | | ||
| `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>`, | ||
as `reasonSpam`. Its subject is a `com.atproto.admin.defs#repoRef` for a | ||
Cargo.lock+1−0
| @@ -2424,6 +2424,7 @@ dependencies = [ | ||
"reqwest", | ||
"resvg", | ||
"rusqlite", | ||
"rustix", | ||
"serde", | ||
"serde_json", | ||
"sha2", | ||