Metrics on a thread and port of their own, from memory alone (#194)

/metrics shared the web server's threads, and answered after a token
lookup and three counting queries, so a pool run dry, one of the things it
is there to show, would have stalled it, and a slow scrape held a web
thread. Now [web] metrics_bind starts a listener on its own OS thread with
a current-thread runtime, reading only what is in memory: pool counts,
channel capacity, the daemon runtime's metrics through its handle, /proc.
No sign-in, as exporters have none; publish the port to the LAN only. The
feed, pending and downloaded counts stay with the healthcheck's log lines,
which the dashboard already reads.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-11 15:04:51 +00:00
parent 9fc1e7e248
commit 973eeb67c7
8 changed files with 138 additions and 62 deletions

View File

@@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- `/metrics` gives Prometheus what the daemon holds between requests: database connections in use against the pool's size, commands waiting for the worker, live-update listeners, the runtime's tasks, memory and open files. The iPX dashboard graphs them in a Daemon row (#194).
- With `[web] metrics_bind` set, `/metrics` gives Prometheus, on a thread and port of its own, what the daemon holds between requests: database connections in use against the pool's size, commands waiting for the worker, live-update listeners, the runtime's tasks, memory and open files. The iPX dashboard graphs them in a Daemon row (#194).
- The page reports its errors, the player's, how long play takes to start and how long it stalls to the server's log, and the requests one tap sets off share a trace, the playlist, audio and saved place of what it played included (#190, #192).
- Saving an episode from a show with DeadAir on saves it without its dead air, at your breathing room (#186).
- Choose the pause DeadAir leaves where it cuts a silence, from 0.2 to 0.5 seconds, for each show in its settings, as its breathing room, on a slider from brisk to relaxed. It is 0.3 seconds unless you change it, up from 0.2, which made speech sound rushed (#182, #183).

View File

@@ -19,6 +19,7 @@ services:
- ipx.env # relative: Arcane resolves it inside its own container
ports:
- "8099:8099" # web UI
- "192.168.1.130:9187:9187" # /metrics for Prometheus, LAN only (ipx#194)
- "6881:6881/tcp" # BitTorrent peers
- "6881:6881/udp" # DHT
volumes:

View File

@@ -103,6 +103,7 @@ access_aud = "" # the Access application's AUD tag
auto_create_users = true
sign_out_url = "" # e.g. "/cdn-cgi/access/logout"
session_days = 30
metrics_bind = "" # e.g. "0.0.0.0:9187"; empty serves no metrics
```
* **`token`** — the shared secret, which signs in as the **admin**. `?token=…` sets a cookie, so
@@ -119,6 +120,9 @@ session_days = 30
`/cdn-cgi/access/logout` behind Cloudflare Access. Empty sends them to the sign-in page, where
the proxy signs them straight back in.
* **`session_days`** — sign a session out after this long without a request.
* **`metrics_bind`** — where to serve `/metrics` for Prometheus: database connections in use, the
worker's queue, the runtime's tasks, memory and open files. It has a thread and a port of its own
and no sign-in, so publish it to the LAN only. Empty, the default, serves none.
It is plain HTTP. On a LAN bind everything crosses the network in the clear — and a feed URL can
itself carry a credential. Put TLS in front of it if that matters.

View File

@@ -97,6 +97,9 @@ pub struct Web {
pub sign_out_url: String,
/// Sign a session out after this long without a request.
pub session_days: i64,
/// Where to serve /metrics for Prometheus (#194), on a thread of its own, without sign-in:
/// `0.0.0.0:9187`. Empty serves none.
pub metrics_bind: String,
}
impl Default for Web {
@@ -112,6 +115,7 @@ impl Default for Web {
auto_create_users: true,
sign_out_url: String::new(),
session_days: 30,
metrics_bind: String::new(),
}
}
}

View File

@@ -773,13 +773,25 @@ async fn start_web(
let (access, ctx, team) = (access.clone(), ctx.clone(), team.to_owned());
tokio::spawn(async move { access.prefetch(&ctx.client, &team).await });
}
let state = web::WebState {
let playing = deadair::Playing::default();
if !cfg.web.metrics_bind.is_empty() {
let sources = metrics::Sources {
ctx: ctx.clone(),
cmds: cmds.clone(),
events: events.clone(),
playing: playing.clone(),
runtime: tokio::runtime::Handle::current(),
started: std::time::Instant::now(),
};
let at = metrics::spawn(&cfg.web.metrics_bind, sources)?;
tracing::info!(%at, "serving metrics for Prometheus at /metrics");
}
let state = web::WebState {
ctx: ctx.clone(),
cmds: cmds.clone(),
events: events.clone(),
access,
playing: Default::default(),
playing,
};
Ok(Some(tokio::spawn(async move {
if let Err(e) = web::serve(state, &bind).await {
@@ -2484,6 +2496,35 @@ mod tests {
}
}
#[tokio::test]
async fn metrics_answer_from_a_thread_of_their_own() {
let ctx = Arc::new(test_ctx(config::Config::default()).await);
let (cmds, _rx) = mpsc::channel(8);
let (events, _) = broadcast::channel(8);
let sources = metrics::Sources {
ctx: ctx.clone(),
cmds,
events,
playing: Default::default(),
runtime: tokio::runtime::Handle::current(),
started: std::time::Instant::now(),
};
let at = metrics::spawn("127.0.0.1:0", sources).unwrap();
let text = ctx.client.get(format!("http://{at}/metrics")).send().await.unwrap().text().await.unwrap();
// Every sample a name, optional labels and a number, as Prometheus parses them (#194).
for line in text.lines().filter(|l| !l.starts_with('#')) {
let (name, value) = line.rsplit_once(' ').unwrap();
assert!(name.starts_with("ipx_") && value.parse::<f64>().is_ok(), "{line}");
}
for name in ["ipx_db_connections{state=\"in_use\"}", "ipx_db_connections_max", "ipx_worker_queue", "ipx_runtime_tasks"] {
assert!(text.contains(name), "{name}");
}
assert!(metrics::spawn(&at.to_string(), metrics::Sources {
ctx: ctx.clone(), cmds: mpsc::channel(1).0, events: broadcast::channel(1).0, playing: Default::default(),
runtime: tokio::runtime::Handle::current(), started: std::time::Instant::now(),
}).is_err(), "a port already taken is an error at start");
}
#[tokio::test]
async fn dead_feeds_leave_the_directory_once_a_day_not_before_every_scan() {
let ctx = test_ctx(config::Config::default()).await;

View File

@@ -1,8 +1,90 @@
//! Prometheus metrics for what logs and traces cannot show: the daemon's own state between
//! requests, such as how many database connections are in use (#194). Written out by hand: the
//! text format is a few lines, and a client library would be a dependency for them.
//! requests, such as how many database connections are in use (#194). Served on a port and a
//! thread of their own, with a runtime of their own, and made only of what is already in memory:
//! a scrape neither waits for the web server's threads or the database, whose pool running dry
//! is one of the things it is there to show, nor holds anything up itself. No sign-in, as
//! Prometheus exporters have none: checking a token is a database lookup. Written out by hand:
//! the text format is a few lines, and a client library would be a dependency for them.
use std::fmt::Write as _;
use std::sync::Arc;
/// What the metrics are read from.
pub struct Sources {
pub ctx: Arc<crate::Ctx>,
pub cmds: tokio::sync::mpsc::Sender<crate::ipc::Command>,
pub events: tokio::sync::broadcast::Sender<crate::ipc::Event>,
pub playing: crate::deadair::Playing,
/// The daemon's runtime, not the metrics thread's own.
pub runtime: tokio::runtime::Handle,
pub started: std::time::Instant,
}
/// Starts serving /metrics on `bind`, on a thread of its own, and says where.
pub fn spawn(bind: &str, s: Sources) -> anyhow::Result<std::net::SocketAddr> {
use anyhow::Context as _;
// Bound here, so a port already taken is the daemon's error at start, not a thread's later.
let listener = std::net::TcpListener::bind(bind).with_context(|| format!("binding metrics to {bind}"))?;
listener.set_nonblocking(true)?;
let addr = listener.local_addr()?;
let s = Arc::new(s);
std::thread::Builder::new().name("metrics".into()).spawn(move || {
let rt = match tokio::runtime::Builder::new_current_thread().enable_all().build() {
Ok(rt) => rt,
Err(e) => return tracing::error!(error = %e, "metrics could not start"),
};
rt.block_on(async move {
let app = axum::Router::new().route(
"/metrics",
axum::routing::get(move || {
let s = s.clone();
async move { ([(axum::http::header::CONTENT_TYPE, "text/plain; version=0.0.4")], render(&s)) }
}),
);
match tokio::net::TcpListener::from_std(listener) {
Ok(l) => {
if let Err(e) = axum::serve(l, app).await {
tracing::error!(error = %e, "metrics stopped");
}
}
Err(e) => tracing::error!(error = %e, "metrics could not listen"),
}
});
})?;
Ok(addr)
}
/// Every metric, from memory alone.
pub fn render(s: &Sources) -> String {
let mut out = String::new();
let (open, idle, max) = s.ctx.db.pool();
gauge(&mut out, "ipx_db_connections", "Database connections open, in use or idle.",
&[("state=\"in_use\"", open as f64 - idle as f64), ("state=\"idle\"", idle as f64)]);
gauge(&mut out, "ipx_db_connections_max", "The most database connections the pool opens.", &[("", max as f64)]);
gauge(&mut out, "ipx_worker_queue", "Commands waiting for the worker: scans, downloads and the like.",
&[("", (s.cmds.max_capacity() - s.cmds.capacity()) as f64)]);
gauge(&mut out, "ipx_event_listeners", "Open pages and socket clients listening for live updates.",
&[("", s.events.receiver_count() as f64)]);
gauge(&mut out, "ipx_torrent_slots_free", "Torrents that could start now.",
&[("", s.ctx.torrent_slots.available_permits() as f64)]);
gauge(&mut out, "ipx_deadair_playing", "Episodes someone is playing through DeadAir's playlist.",
&[("", s.playing.lock().map_or(0, |p| p.len()) as f64)]);
let rt = s.runtime.metrics();
gauge(&mut out, "ipx_runtime_workers", "The async runtime's worker threads.", &[("", rt.num_workers() as f64)]);
gauge(&mut out, "ipx_runtime_tasks", "Tasks alive in the async runtime.", &[("", rt.num_alive_tasks() as f64)]);
gauge(&mut out, "ipx_runtime_queue", "Tasks waiting in the runtime's global queue.", &[("", rt.global_queue_depth() as f64)]);
let (rss, fds) = process();
if let Some(v) = rss {
gauge(&mut out, "ipx_process_resident_bytes", "Memory the process holds.", &[("", v)]);
}
if let Some(v) = fds {
gauge(&mut out, "ipx_process_open_fds", "Files and sockets the process has open.", &[("", v)]);
}
gauge(&mut out, "ipx_uptime_seconds", "Seconds since the daemon started.", &[("", s.started.elapsed().as_secs() as f64)]);
let version = format!("version=\"{}\"", env!("CARGO_PKG_VERSION"));
gauge(&mut out, "ipx_build_info", "The version running, as a label.", &[(&version, 1.0)]);
out
}
/// One gauge in Prometheus's text format: its help, its type, and a line for each set of labels
/// (`state="idle"`, or "" for none).

View File

@@ -29,8 +29,6 @@ const COOKIE: &str = "ipx_token";
#[derive(Clone)]
pub struct WebState {
/// When the web server started, for the uptime in /metrics.
pub started: std::time::Instant,
pub ctx: Arc<Ctx>,
pub cmds: mpsc::Sender<Command>,
pub events: broadcast::Sender<Event>,
@@ -62,7 +60,6 @@ pub fn router(state: WebState) -> Router {
.route("/api/opml", get(export_opml).post(import_opml))
.route("/api/settings", get(get_settings).patch(patch_settings))
.route("/api/status", get(status))
.route("/metrics", get(metrics))
.route("/api/popular", get(get_popular))
.route("/api/directory", get(get_directory))
.route("/api/directory/{id}", get(get_listed))
@@ -2323,46 +2320,6 @@ async fn status(State(state): State<WebState>) -> Response {
}
}
/// The daemon's state between requests, for Prometheus (#194): what logs and traces cannot show.
/// Behind sign-in like the rest; Prometheus sends an API token.
async fn metrics(State(state): State<WebState>) -> Response {
use crate::metrics::gauge;
let mut out = String::new();
let (open, idle, max) = state.ctx.db.pool();
let in_use = open as f64 - idle as f64;
gauge(&mut out, "ipx_db_connections", "Database connections open, in use or idle.",
&[("state=\"in_use\"", in_use), ("state=\"idle\"", idle as f64)]);
gauge(&mut out, "ipx_db_connections_max", "The most database connections the pool opens.", &[("", max as f64)]);
gauge(&mut out, "ipx_worker_queue", "Commands waiting for the worker: scans, downloads and the like.",
&[("", (state.cmds.max_capacity() - state.cmds.capacity()) as f64)]);
gauge(&mut out, "ipx_event_listeners", "Open pages and socket clients listening for live updates.",
&[("", state.events.receiver_count() as f64)]);
gauge(&mut out, "ipx_torrent_slots_free", "Torrents that could start now.",
&[("", state.ctx.torrent_slots.available_permits() as f64)]);
gauge(&mut out, "ipx_deadair_playing", "Episodes someone is playing through DeadAir's playlist.",
&[("", state.playing.lock().unwrap().len() as f64)]);
if let Event::Status { feeds, pending, downloaded } = crate::status(&state.ctx).await {
gauge(&mut out, "ipx_feeds", "Feeds in the catalogue.", &[("", feeds as f64)]);
gauge(&mut out, "ipx_downloads_pending", "Files a scan will download on its own.", &[("", pending as f64)]);
gauge(&mut out, "ipx_files_downloaded", "Files on disk.", &[("", downloaded as f64)]);
}
let rt = tokio::runtime::Handle::current().metrics();
gauge(&mut out, "ipx_runtime_workers", "The async runtime's worker threads.", &[("", rt.num_workers() as f64)]);
gauge(&mut out, "ipx_runtime_tasks", "Tasks alive in the async runtime.", &[("", rt.num_alive_tasks() as f64)]);
gauge(&mut out, "ipx_runtime_queue", "Tasks waiting in the runtime's global queue.", &[("", rt.global_queue_depth() as f64)]);
let (rss, fds) = crate::metrics::process();
if let Some(v) = rss {
gauge(&mut out, "ipx_process_resident_bytes", "Memory the process holds.", &[("", v)]);
}
if let Some(v) = fds {
gauge(&mut out, "ipx_process_open_fds", "Files and sockets the process has open.", &[("", v)]);
}
gauge(&mut out, "ipx_uptime_seconds", "Seconds since the daemon started.", &[("", state.started.elapsed().as_secs() as f64)]);
let version = format!("version=\"{}\"", env!("CARGO_PKG_VERSION"));
gauge(&mut out, "ipx_build_info", "The version running, as a label.", &[(&version, 1.0)]);
([(header::CONTENT_TYPE, "text/plain; version=0.0.4")], out).into_response()
}
#[derive(Serialize)]
struct Settings {
schedule: String,
@@ -2550,8 +2507,7 @@ async fn name_span(route: MatchedPath, req: Request, next: Next) -> Response {
async fn access_log(req: Request, next: Next) -> Response {
let path = req.uri().path().to_owned();
let method = req.method().clone();
// Prometheus scrapes /metrics every 15 s: a line and a trace each time would be the same noise.
let quiet = path.starts_with("/api/logs") || path == "/metrics";
let quiet = path.starts_with("/api/logs");
// No trace for the event stream either: an open page asks for it again and again, and Tempo
// would fill with nothing else.
let span = if quiet || path == "/api/events" {

View File

@@ -1390,18 +1390,6 @@ test('/api/status gives a dashboard the counts, with the shared token as a cooki
expect(none.status()).toBe(401);
});
test('/metrics gives Prometheus the daemon\'s state, to someone signed in', async ({ page }) => {
const r = await page.request.get('/metrics', { headers: { cookie: `ipx_token=${TOKEN}` } });
expect(r.status()).toBe(200);
const text = await r.text();
// Every sample a name, optional labels and a number, as Prometheus parses them (#194).
for (const line of text.split('\n').filter(l => l && !l.startsWith('#')))
expect(line).toMatch(/^ipx_[a-z_]+(\{[^}]*\})? -?[0-9.e+]+$/);
for (const name of ['ipx_db_connections{state="in_use"}', 'ipx_db_connections_max', 'ipx_worker_queue', 'ipx_feeds', 'ipx_runtime_tasks', 'ipx_process_resident_bytes'])
expect(text).toContain(name);
expect((await page.request.get('/metrics', { headers: { cookie: '' } })).status()).toBe(401);
});
test('the favicon is the logo, square, from both pages', async ({ page }) => {
await page.evaluate(() => setTheme('modern', 'light'));
await expect(page.locator('#favicon')).toHaveAttribute('href', /^\/favicon\.png\?v=[0-9a-f]{12}$/);