diff --git a/CHANGELOG.md b/CHANGELOG.md index 4065a09..f202077 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/grafana/dashboard.py b/grafana/dashboard.py index 16280ba..89487f5 100644 --- a/grafana/dashboard.py +++ b/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, diff --git a/src/db.rs b/src/db.rs index ce5d480..ac5d569 100644 --- a/src/db.rs +++ b/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('@')) { diff --git a/src/main.rs b/src/main.rs index 49431fd..62998da 100644 --- a/src/main.rs +++ b/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(), diff --git a/src/metrics.rs b/src/metrics.rs new file mode 100644 index 0000000..0005d94 --- /dev/null +++ b/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, Option) { + 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::().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); + } +} diff --git a/src/web.rs b/src/web.rs index 44f7bce..60680ba 100644 --- a/src/web.rs +++ b/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, pub cmds: mpsc::Sender, pub events: broadcast::Sender, @@ -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) -> 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) -> 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" { diff --git a/tests/ui/app.spec.js b/tests/ui/app.spec.js index 7efcd6c..a895f6c 100644 --- a/tests/ui/app.spec.js +++ b/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}$/);