Prometheus metrics for the daemon's state between requests (#194)

Logs and traces show requests, scans and downloads; nothing showed what
the daemon holds meanwhile. /metrics, behind sign-in, gives database
connections in use and idle against the pool's maximum, commands waiting
for the worker, live-update listeners, free torrent slots, episodes
playing through DeadAir, the library's counts, the tokio runtime's
workers, tasks and queue, and the process's memory and open files, in
Prometheus's text format, written by hand rather than with a client
library. Not logged or traced: Prometheus asks every 15 s. The iPX
dashboard gains a Daemon row of them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -9,6 +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).
 - 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).

grafana/dashboard.py

@@ -2,15 +2,17 @@
 
     python3 grafana/dashboard.py > /mnt/fast/arcane/projects/monitoring/grafana-provisioning/dashboards/ipx.json
 
-Grafana reads that file on its own within a minute; edits made in Grafana are refused. The panels
-read the fields of ipx's JSON log (IPX_LOG_FORMAT=json): renaming a field in web.rs access_log or
-ipc.rs log_event has to be matched here. `dashboard.py queries` prints each
+Grafana reads that file on its own within a minute; edits made in Grafana are refused. The
+Daemon row reads ipx's /metrics from Prometheus (job ipx, #194); the rest read the fields of
+ipx's JSON log (IPX_LOG_FORMAT=json): renaming a field in web.rs access_log or ipc.rs log_event
+has to be matched here. `dashboard.py queries` prints each
 query, to try against Loki.
 """
 import json, sys
 
 LOKI = {"type": "loki", "uid": "${loki}"}
 TEMPO = {"type": "tempo", "uid": "${tempo}"}
+PROM = {"type": "prometheus", "uid": "${prometheus}"}
 SEL = '{container="iPX"}'
 # ipx logs one JSON object a line (IPX_LOG_FORMAT=json). Each scan and download event carries its
 # fields (ev, feed, new, bytes, msg, error.type, ...); each request its method, path, route, status
@@ -50,6 +52,14 @@ def loki(expr, ref="A", legend=None, instant=False, kind=None):
     return t
 
 
+def prom(expr, ref="A", legend=None):
+    # ipx's /metrics (#194), scraped by Prometheus as job ipx.
+    t = {"refId": ref, "datasource": PROM, "expr": expr}
+    if legend:
+        t["legendFormat"] = legend
+    return t
+
+
 def row(title):
     global y, pid
     pid += 1
@@ -135,6 +145,25 @@ table("Busiest routes", [loki(f'topk(15, sum by (method, route) (count_over_time
       12, rename={"method": "Method", "route": "Route", "Value": "Requests"}, sort=[{"displayName": "Requests", "desc": True}])
 y += 8
 
+# ---- Daemon, from Prometheus: what the daemon holds between requests (#194)
+row("Daemon")
+ts("Database connections", [prom('ipx_db_connections{job="ipx", state="in_use"}', legend="in use"),
+                            prom('ipx_db_connections{job="ipx", state="idle"}', ref="B", legend="idle"),
+                            prom('ipx_db_connections_max{job="ipx"}', ref="C", legend="most the pool opens")],
+   0, w=8, desc="In use reaching the most the pool opens means requests and scans are waiting for a connection.")
+ts("Waiting", [prom('ipx_worker_queue{job="ipx"}', legend="commands for the worker"),
+               prom('ipx_runtime_queue{job="ipx"}', ref="B", legend="tasks for a thread"),
+               prom('ipx_downloads_pending{job="ipx"}', ref="C", legend="downloads")],
+   8, w=8, desc="Scans and downloads run one at a time; a queue that only grows means the worker is stuck.")
+ts("Listeners and tasks", [prom('ipx_event_listeners{job="ipx"}', legend="live-update listeners"),
+                           prom('ipx_runtime_tasks{job="ipx"}', ref="B", legend="runtime tasks"),
+                           prom('ipx_deadair_playing{job="ipx"}', ref="C", legend="playing through DeadAir")],
+   16, w=8)
+y += 8
+ts("Memory", [prom('ipx_process_resident_bytes{job="ipx"}', legend="resident")], 0, w=12, unit="bytes")
+ts("Open files and sockets", [prom('ipx_process_open_fds{job="ipx"}', legend="open")], 12, w=12)
+y += 8
+
 # ---- Traces
 row("Traces")
 for x, title, q in [(0, "Recent traces", TRACES + "}"),
@@ -162,6 +191,7 @@ dash = {
     "templating": {"list": [
         {"name": "loki", "label": "Logs", "type": "datasource", "query": "loki", "current": {}, "hide": 0},
         {"name": "tempo", "label": "Traces", "type": "datasource", "query": "tempo", "current": {}, "hide": 0},
+        {"name": "prometheus", "label": "Metrics", "type": "datasource", "query": "prometheus", "current": {}, "hide": 0},
     ]},
     "links": [{"title": "iPX", "type": "link", "url": "https://ipodderx.sdf1.net", "targetBlank": True}],
     "panels": panels,

src/db.rs

@@ -130,6 +130,24 @@ pub fn location() -> String {
 /// The Postgres connection option that keeps notices from being sent at all; see `connect`.
 const QUIET: &str = "options=-c%20client_min_messages%3Dwarning";
 
+impl Db {
+    /// The connection pool as it stands, for /metrics (#194): connections open, those of them
+    /// idle, and the most it will open.
+    pub fn pool(&self) -> (u32, usize, u32) {
+        use sea_orm::DbBackend;
+        match self.orm.get_database_backend() {
+            DbBackend::Postgres => {
+                let p = self.orm.get_postgres_connection_pool();
+                (p.size(), p.num_idle(), p.options().get_max_connections())
+            }
+            _ => {
+                let p = self.orm.get_sqlite_connection_pool();
+                (p.size(), p.num_idle(), p.options().get_max_connections())
+            }
+        }
+    }
+}
+
 /// A URL fit for a log or an error: the password taken out.
 fn redact(url: &str) -> String {
     match (url.find("://"), url.rfind('@')) {

src/main.rs

@@ -9,6 +9,7 @@ mod download;
 mod feed;
 mod ipc;
 mod logbuf;
+mod metrics;
 mod retention;
 mod torrent;
 mod web;
@@ -773,6 +774,7 @@ async fn start_web(
         tokio::spawn(async move { access.prefetch(&ctx.client, &team).await });
     }
     let state = web::WebState {
+        started: std::time::Instant::now(),
         ctx: ctx.clone(),
         cmds: cmds.clone(),
         events: events.clone(),

src/metrics.rs

@@ -0,0 +1,49 @@
+//! 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.
+
+use std::fmt::Write as _;
+
+/// One gauge in Prometheus's text format: its help, its type, and a line for each set of labels
+/// (`state="idle"`, or "" for none).
+pub fn gauge(out: &mut String, name: &str, help: &str, samples: &[(&str, f64)]) {
+    let _ = writeln!(out, "# HELP {name} {help}\n# TYPE {name} gauge");
+    for (labels, value) in samples {
+        let labels = if labels.is_empty() { String::new() } else { format!("{{{labels}}}") };
+        let _ = writeln!(out, "{name}{labels} {value}");
+    }
+}
+
+/// The process's resident memory in bytes and its open files, from /proc; None elsewhere.
+pub fn process() -> (Option<f64>, Option<f64>) {
+    let rss = std::fs::read_to_string("/proc/self/status").ok().and_then(|s| {
+        let kb = s.lines().find_map(|l| l.strip_prefix("VmRSS:"))?.trim().strip_suffix("kB")?.trim().parse::<f64>().ok()?;
+        Some(kb * 1024.0)
+    });
+    let fds = std::fs::read_dir("/proc/self/fd").ok().map(|d| d.count() as f64);
+    (rss, fds)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn gauges_are_written_as_prometheus_reads_them() {
+        let mut out = String::new();
+        gauge(&mut out, "ipx_db_connections", "Connections.", &[("state=\"idle\"", 3.0), ("state=\"in_use\"", 1.0)]);
+        gauge(&mut out, "ipx_feeds", "Feeds.", &[("", 1535.0)]);
+        assert_eq!(
+            out,
+            "# HELP ipx_db_connections Connections.\n# TYPE ipx_db_connections gauge\n\
+             ipx_db_connections{state=\"idle\"} 3\nipx_db_connections{state=\"in_use\"} 1\n\
+             # HELP ipx_feeds Feeds.\n# TYPE ipx_feeds gauge\nipx_feeds 1535\n"
+        );
+    }
+
+    #[test]
+    fn the_process_reads_its_memory_and_files_on_linux() {
+        let (rss, fds) = process();
+        assert!(rss.unwrap() > 0.0 && fds.unwrap() > 0.0);
+    }
+}

src/web.rs

@@ -29,6 +29,8 @@ 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>,
@@ -60,6 +62,7 @@ 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))
@@ -2320,6 +2323,46 @@ 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,
@@ -2507,7 +2550,8 @@ 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();
-    let quiet = path.starts_with("/api/logs");
+    // 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";
     // 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" {

tests/ui/app.spec.js

@@ -1390,6 +1390,18 @@ 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}$/);