diff --git a/.claude/skills/ipx-prod-check/SKILL.md b/.claude/skills/ipx-prod-check/SKILL.md index 3753eb6..6b69c08 100644 --- a/.claude/skills/ipx-prod-check/SKILL.md +++ b/.claude/skills/ipx-prod-check/SKILL.md @@ -10,11 +10,12 @@ Production is three containers in the Arcane project `ipx` on Tower | container | what | logs, in Loki | traces, in Tempo | metrics, in Prometheus | |---|---|---|---|---| -| `ipx-web` | Caddy: serves the page, passes `/api` and `/media` to ipx (#203). Port 8099, the LAN and the tunnel | `{container="ipx-web"}`, JSON: a line a request (`http.log.access`), its own errors | service `ipx-web`: a span a request, passing `traceparent` on to ipx (#213) | none | +| `ipx-web` | Caddy: serves the page, passes `/api` and `/media` to ipx (#203). Port 8099, the LAN and the tunnel | `{container="ipx-web"}`, JSON: a line a request (`http.log.access`), its own errors | service `ipx-web`: a span a request, passing `traceparent` on to ipx (#213) | `job="ipx-web"`: Caddy's own `caddy_http_*`, from `METRICS_BIND`, published as `192.168.1.130:9190` (#371) | | `iPX` | the daemon: API, scans, downloads, DeadAir | `{container="iPX"}`, JSON (`IPX_LOG_FORMAT=json`) | service `ipx`; passes `traceparent` to the directory | `job="ipx"`, its `/metrics` | | `ipx-directory` | the directory of public podcasts, `directory.ipxreader.app` (#204, #214) | `{container="ipx-directory"}`, JSON (`DIRECTORY_LOG_FORMAT=json`) | service `ipx-directory`: a span a request, joining ipx's trace when ipx called; a trace a refresher round (#213) | `job="ipx-directory"`, its `/metrics` | -Under them, Tower itself: `job="node"` (the node exporter: load, CPU, memory, disks). The Grafana +Grafana's alert rules (`grafana/alerts.py`, #365) watch all of it between checks: start by asking +which are firing (the monitoring section). Under them, Tower itself: `job="node"` (the node exporter: load, CPU, memory, disks). The Grafana dashboard `iPX` draws all of it in that order (`grafana/dashboard.py`). A trace without the production tag (`resource.deployment.environment.name="production"`, on all three) is a daemon run by hand, not production. Since #213 a request is one trace across the services it touched: a @@ -111,6 +112,9 @@ from 2026-10-11 15:15 UTC. Gauges, but for the counters (`_total`) and the one h | `ipx_jobs_total{kind}`, `ipx_job_seconds_total{kind}` | counters: jobs finished, and their seconds all told | | `ipx_last_scan_age_seconds` | since a scan last finished; absent until one has since starting | | `ipx_http_requests_in_flight`, `ipx_scan_feeds_fetching`, `ipx_downloads_active` | work in progress | +| `ipx_db_query_seconds{kind}`, `ipx_db_wait_seconds` | histograms (#367): each statement's time once it has a connection, by its first word (`select`, `insert`, ...); and the wait for a connection, measured for hand-written SQL only, about half the queries. A wait growing while queries stay flat is the pool full | +| `ipx_page_seconds{ev}` | histogram: what the page timed, `list_shown`, `play_start`, `stall`, as the `ipx::client` lines say one by one (#369) | +| `ipx_feed_failures_total{error_type}`, `ipx_new_items_total`, `ipx_downloads_total{result="saved"\|"failed"}`, `ipx_deadair_saved_seconds_total` | counters of what the events say (#368): feeds failed by kind, as the `feed_error` lines' `error.type`; new items; downloads; seconds of silence DeadAir found | | `ipx_http_request_duration_seconds{method,route,status}` | histogram: each request answered and its seconds, buckets 5 ms to 10 s; `_count` is requests, `_sum` their time. `route` is the pattern, `""` for an unrouted path; the event stream is left out (#364). From 2026-10-25 | | `ipx_download_bytes_total` | counter; `rate()` is the download speed | | `ipx_event_queue`, `ipx_events_missed_total` | live updates unread by some listener; counter of those dropped for one that fell behind | @@ -122,7 +126,7 @@ from 2026-10-11 15:15 UTC. Gauges, but for the counters (`_total`) and the one h | `ipx_jev_requests_total{result="ok"\|"key_refused"\|"malformed"\|"rate_limited"\|"error"}`, `ipx_jev_seconds_total`, `ipx_jev_paced_seconds_total` | counters: calls to TypeSafe's Jev, Signals' and icons', by how they ended; their seconds, the wait for a turn included; and that wait alone. `malformed` is ipx's request at fault (#257); `rate_limited` is a 429 from TypeSafe, whose limits are 80 calls and 100,000 input tokens a second, which ipx paces itself under at 60 and 80,000 (#277) | | `ipx_jev_tokens_total{kind="input"\|"output",user,key}` | counter: tokens TypeSafe counted, as its answers say, by person and whose key paid: what Signals, icon picks and key checks cost (#275, #278). A call shared on the server's key is split by each person's share of its questions | | `ipx_signals_last_pass_age_seconds`, `ipx_signals_keys_refused` | since the Signals judge last finished a pass (at most 5 minutes when well; absent until one has); people whose key TypeSafe refused | -| `directory_process_resident_bytes`, `directory_process_open_fds`, `directory_feed_fetches_total`, `directory_feed_bytes_total`, `directory_uptime_seconds`, `directory_build_info` | the directory's, `job="ipx-directory"`, from its `metrics_bind` (production `192.168.1.130:9188`, LAN only) | +| `directory_process_resident_bytes`, `directory_process_open_fds`, `directory_feed_fetches_total`, `directory_feed_bytes_total`, `directory_uptime_seconds`, `directory_build_info`, `directory_http_request_duration_seconds{method,route,status}`, `directory_feed_checks_total`, `directory_feed_checks_failed_total`, `directory_snapshots_total` | the directory's (the last four #370), `job="ipx-directory"`, from its `metrics_bind` (production `192.168.1.130:9188`, LAN only) | | `ipx_feeds`, `ipx_downloads_pending`, `ipx_files_downloaded` | the library, as the daemon counts it every 30 s | ### The checks @@ -208,9 +212,18 @@ a count says something happened, the lines and traces say why. since #193, so a reap is about a second), `trim_artwork`, with what it removed on the `reap` span. DeadAir's work on a file is a `deadair` trace (feed, enclosure, cuts) with `probe`, `silencedetect`, `frame_bytes`, `store_plans`: seconds each, more is a file ffmpeg struggles - with. A trace with no name in `$Q traces` started in the page (#190, #192): its root is the tap - in the browser, which sends no span, so it holds the requests that tap made (API calls, the - playlist and audio of a play, position saves, the page's own report) side by side. + with. A trace begun in the page (#190, #192) has the tap or key as its root span, `tap` or + `key` (#376), from the moment it began to its last API answer, with the requests it made under + it (API calls, the playlist and audio of a play, position saves, the page's own report): a tap + long with its requests short is the page itself, drawing. The page reports the tap when the next + one begins or it is hidden, so the root arrives last; a trace with no name in `$Q traces` is + requests no tap began (a live update's, a timer's), or a tap not reported yet. + + Every command the worker runs from the socket, the CLI or the page is a `command` span (`cmd`, + #373) with its job inside; a scan on the clock is a `scan` of its own. A list's sync is + `list_sync` (`list`, `listed`) with steps `upsert_listed`, `subscribe_readers`, `drop_unlisted`, + and an OPML upload `opml_import` (#372). The daemon's start is a `start` trace (#374), from + `open_database` and `assemble_config` to `daemon started`: how long a deploy took to answer. 6. **The daemon's state, from Prometheus.** Pressure and leaks that requests and traces only hint at. @@ -228,6 +241,8 @@ a count says something happened, the lines and traces say why. $Q prom 'max(max_over_time(ipx_last_scan_age_seconds{job="ipx"}[$range]))' 24h $Q prom 'sum by (kind) (increase(ipx_job_seconds_total{job="ipx"}[$range]))' 24h $Q prom 'max(max_over_time(ipx_http_requests_in_flight{job="ipx"}[$range]))' 24h + $Q prom 'histogram_quantile(0.99, sum by (le) (rate(ipx_db_wait_seconds_bucket{job="ipx"}[$range])))' 24h + $Q prom 'histogram_quantile(0.95, sum by (le, kind) (rate(ipx_db_query_seconds_bucket{job="ipx"}[$range])))' 24h $Q prom 'sum(increase(ipx_events_missed_total{job="ipx"}[$range]))' 24h $Q prom 'max(max_over_time(ipx_deadair_busy_seconds{job="ipx"}[$range]))' 24h $Q prom 'sum by (result) (increase(ipx_jev_requests_total{job="ipx"}[$range]))' 24h @@ -246,7 +261,7 @@ a count says something happened, the lines and traces say why. 15 minutes ago, is the worker stuck, which `ipx status` does not show; its trace and the log lines of that time say on what. A kind whose share of `ipx_job_seconds_total` jumps is where the worker's time went (reap was most of it until #193). Web requests in flight near - the pool's max mean the pool is what they wait on. Live updates missed mean a page showed stale + the pool's max mean the pool is what they wait on, and `ipx_db_wait_seconds` for how long. Live updates missed mean a page showed stale counts until reloaded. DeadAir over a minute on one file is a file ffmpeg chokes on. Any `rate_limited` Jev call means the pacing is not holding ipx under TypeSafe's limits (#277); each `jev` span has `paced_ms`, its wait for a turn. Any `malformed` Jev call is ipx asking wrongly (#257); `key_refused` is someone's key; the Signals @@ -273,7 +288,9 @@ Caddy logs one JSON line a request (#210), logger `http.log.access.log0`: Loki's The request's headers and a `?token=` are filtered out before it is written. Requests from `127.0.0.1` are local checks; leave them out. `/api` and `/media` pass on to ipx, which counts them in `ipx_http_request_duration_seconds` with its own time; Caddy's `duration` for them adds -the hop. This is the only log with a line for every request. +the hop. This is the only log with a line for every request. Caddy's own metrics, `job="ipx-web"` +(`caddy_http_request_duration_seconds{code,handler,method}` and the like), count every request too, +502s included, which ipx never sees. ```sh W='{container="ipx-web"} |= "http.log.access" | json | __error__="" | request_remote_ip != "127.0.0.1"' @@ -319,8 +336,8 @@ token file changes. Its lines carry their own fields and no span (#363). Its met `directory_process_open_fds`, `directory_feed_fetches_total`, `directory_feed_bytes_total`, `directory_uptime_seconds`, `directory_build_info`. Its traces (#213): a span a request (`http`), a child of ipx's when ipx called with a `traceparent`, and a trace a refresher round (`refresh`, `feeds`) -with a `feed` span for each check. Its log lines carry no trace id; find the trace by time or by -the feed. +with a `feed` span for each check. Its lines inside a span carry `trace_id` and `span_id`, as +ipx's do (#375): a `feed check failed` leads to its check's span. ```sh $Q traces '{resource.service.name="ipx-directory" && duration > 2s}' @@ -392,7 +409,22 @@ docker ps -a --filter label=com.docker.compose.project=monitoring --format '{{.N docker logs --since 1h tempo 2>&1 | grep -E 'failed to poll|blocklist poll complete' | tail -2 ``` -`ipx`, `ipx-directory` and `node` should be `up`. +`ipx`, `ipx-directory`, `ipx-web`, `node` and the monitoring's own, `prometheus`, `tempo`, `loki` +and `alloy` (#366), should be `up`. Alloy is scraped at `alloy:12345` because its command sets +`--server.http.listen-addr`; it listens on 127.0.0.1 without it. + +The alert rules (#365) are `grafana/alerts.py`, written to +`/mnt/fast/arcane/projects/monitoring/grafana-provisioning/alerting/ipx.json` and loaded by +`POST /api/admin/provisioning/alerting/reload`. There is no contact point yet, so one firing is +only in Grafana: ask it which are, at the start of every check. The password is in the +monitoring project's compose file; never print it. + +```sh +PW=$(grep -oP 'GF_SECURITY_ADMIN_PASSWORD=\K\S+' /mnt/fast/arcane/projects/monitoring/compose.yaml) +curl -s -u "admin:$PW" http://192.168.1.130:3444/api/prometheus/grafana/api/v1/rules | python3 -c 'import json,sys +for g in json.load(sys.stdin)["data"]["groups"]: + for r in g["rules"]: print(r["state"], r["name"])' | sort +``` Tempo looks over its stored blocks every 5 minutes. `failed to poll or create index for tenant` means it could not, and then searches and trace lookups see only what it still holds in memory, diff --git a/crates/directory/CHANGELOG.md b/crates/directory/CHANGELOG.md index 3825bec..5e19d5d 100644 --- a/crates/directory/CHANGELOG.md +++ b/crates/directory/CHANGELOG.md @@ -10,6 +10,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `/metrics` counts its requests and their times by route and status, the feeds it checks and those that failed, and the snapshots it takes (#370). +- Its log lines inside a trace carry the trace's and span's ids, as ipx's do, so a failed feed check leads to its trace (#375). + ### Changed - Its JSON log lines no longer carry their span, once on its own and again in a list, which said nothing the line's own fields did not (#363). diff --git a/crates/directory/src/api.rs b/crates/directory/src/api.rs index ff6dcd5..627e98d 100644 --- a/crates/directory/src/api.rs +++ b/crates/directory/src/api.rs @@ -36,6 +36,7 @@ pub fn router(dir: Shared) -> Router { .merge(v1) // Anyone may browse, no token, read only (#215). .merge(site::router()) + .route_layer(axum::middleware::from_fn(remember_route)) // Brotli, zstd or gzip as the caller takes them: the feed list went to ipx as 603 KB of // JSON, most of a Directory page's time (#253). Inside the access log, which records what // was sent. @@ -65,6 +66,14 @@ impl From for ApiError { #[derive(Clone)] pub(crate) struct Who(String); +/// Puts the route matched on the response for `access_log`, which runs outside routing: by path, +/// each feed's page was a row of its own in the request metric. +async fn remember_route(route: axum::extract::MatchedPath, req: Request, next: Next) -> Response { + let mut resp = next.run(req).await; + resp.extensions_mut().insert(route); + resp +} + /// One line a request, with the token's name and never the token: what the Grafana panels read. pub(crate) async fn access_log(req: Request, next: Next) -> Response { let (method, path) = (req.method().clone(), req.uri().path().to_owned()); @@ -92,6 +101,15 @@ pub(crate) async fn access_log(req: Request, next: Next) -> Response { let who = resp.extensions().get::().map_or("", |w| w.0.as_str()).to_owned(); let status = resp.status().as_u16(); if path != "/healthz" { + let route = resp.extensions().get::().map_or("", |r| r.as_str()); + crate::metrics::REQUESTS.observe( + &format!( + "method=\"{}\",route=\"{}\",status=\"{status}\"", + crate::metrics::method(method.as_str()), + crate::metrics::label(route) + ), + start.elapsed().as_secs_f64(), + ); tracing::info!( target: "ipx_directory::http", method = %method, path = %path, status, duration_ms = start.elapsed().as_millis() as u64, token = %who, @@ -174,6 +192,7 @@ pub(crate) async fn snapshot( } let r = dir.store.snapshot(&me.name, &entries).await?; dir.wake.notify_one(); + crate::metrics::SNAPSHOTS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); tracing::info!( instance = %me.name, accepted = r.accepted, refused_private = r.refused_private, refused_blocked = r.refused_blocked, refused_invalid = r.refused_invalid, "snapshot" diff --git a/crates/directory/src/apple.rs b/crates/directory/src/apple.rs index bbe5660..29b9a56 100644 --- a/crates/directory/src/apple.rs +++ b/crates/directory/src/apple.rs @@ -96,6 +96,7 @@ async fn sync(dir: &Shared, client: &reqwest::Client, country: &str) -> Result<( tokio::time::sleep(PAUSE).await; } let result = dir.store.snapshot(INSTANCE, &entries).await?; + crate::metrics::SNAPSHOTS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); tracing::info!( shows = ids.len(), accepted = result.accepted, diff --git a/crates/directory/src/directory.rs b/crates/directory/src/directory.rs index 4104e5f..2a9297f 100644 --- a/crates/directory/src/directory.rs +++ b/crates/directory/src/directory.rs @@ -112,6 +112,7 @@ impl Directory { Err(e) => Refreshed::Failed(format!("{e:#}")), }; if let Refreshed::Failed(e) = &got { + crate::metrics::CHECKS_FAILED.fetch_add(1, std::sync::atomic::Ordering::Relaxed); tracing::info!(feed = %due.id, error = %ipx_feed::redact_text(e), "feed check failed"); } // ponytail: a feed that moved for good is still asked for at its old address, which @@ -143,6 +144,7 @@ pub async fn refresher(dir: Shared) { .await; // One line a round, for the dashboard's refresher panel (#210); failures have their own. tracing::info!(checked = due.len(), "feeds checked"); + crate::metrics::CHECKED.fetch_add(due.len() as u64, std::sync::atomic::Ordering::Relaxed); }, round, ) diff --git a/crates/directory/src/main.rs b/crates/directory/src/main.rs index 93bdc41..8229853 100644 --- a/crates/directory/src/main.rs +++ b/crates/directory/src/main.rs @@ -87,11 +87,16 @@ async fn serve(path: &Path) -> Result<()> { .with((!json).then(|| tracing_subscriber::fmt::layer().with_writer(std::io::stderr).with_filter(filter()))) .with(json.then(|| { tracing_subscriber::fmt::layer() - .json() - .flatten_event(true) - // Each line's own fields say all its span did, and the list said it again (#363). - .with_current_span(false) - .with_span_list(false) + .fmt_fields(tracing_subscriber::fmt::format::JsonFields::new()) + // Each line's own fields say all its span did, and the list said it again (#363); + // the trace and span ids lead to the trace instead, as ipx's do (#375). + .event_format(ipx_directory::otel::WithTrace( + tracing_subscriber::fmt::format() + .json() + .flatten_event(true) + .with_current_span(false) + .with_span_list(false), + )) .with_writer(std::io::stderr) .with_filter(filter()) })) diff --git a/crates/directory/src/metrics.rs b/crates/directory/src/metrics.rs index c063808..653096d 100644 --- a/crates/directory/src/metrics.rs +++ b/crates/directory/src/metrics.rs @@ -7,6 +7,10 @@ //! does not come down again is a leak. Not jemalloc, whose upstream was archived in 2025. use axum::Router; +use std::collections::BTreeMap; +use std::fmt::Write as _; +use std::sync::Mutex; +use std::sync::atomic::AtomicU64; use std::sync::atomic::Ordering::Relaxed; use std::time::Instant; use tokio::net::TcpListener; @@ -27,8 +31,75 @@ pub fn process() -> (Option, Option) { (rss, fds) } -/// The directory's /metrics, in Prometheus's text format: memory, open files, and the feeds it -/// has fetched. Nothing from the database, so a scrape costs nothing and answers while it is down. +/// Upper bounds for a request's or a query's time, in seconds: most answer in milliseconds, and a +/// host stalling takes seconds. +pub const SECONDS: &[f64] = &[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]; + +/// Prometheus's histogram, for ipx and the directory both: for each set of labels, how many +/// observations fell under each bound, how many all told, and their sum. +pub struct Histograms { + bounds: &'static [f64], + rows: Mutex>, +} + +/// Under each bound, all told, and the sum. +type Row = (Vec, u64, f64); + +impl Histograms { + #[must_use] + pub const fn new(bounds: &'static [f64]) -> Self { + Self { bounds, rows: Mutex::new(BTreeMap::new()) } + } + + /// Counts `v` under `labels`, written as Prometheus reads them (`a="x",b="y"`, or "" for + /// none); `label` escapes a value that came from outside. + pub fn observe(&self, labels: &str, v: f64) { + let mut rows = self.rows.lock().expect("a panic elsewhere poisoned this lock"); + let (buckets, n, sum) = rows.entry(labels.to_owned()).or_insert_with(|| (vec![0; self.bounds.len()], 0, 0.0)); + for (b, le) in buckets.iter_mut().zip(self.bounds) { + *b += u64::from(v <= *le); + } + *n += 1; + *sum += v; + } + + pub fn render(&self, out: &mut String, name: &str, help: &str) { + let rows = self.rows.lock().expect("a panic elsewhere poisoned this lock").clone(); + let _ = writeln!(out, "# HELP {name} {help}\n# TYPE {name} histogram"); + for (labels, (buckets, n, sum)) in &rows { + let sep = if labels.is_empty() { "" } else { "," }; + for (le, b) in self.bounds.iter().zip(buckets) { + let _ = writeln!(out, "{name}_bucket{{{labels}{sep}le=\"{le}\"}} {b}"); + } + let _ = writeln!(out, "{name}_bucket{{{labels}{sep}le=\"+Inf\"}} {n}"); + let braces = if labels.is_empty() { String::new() } else { format!("{{{labels}}}") }; + let _ = writeln!(out, "{name}_sum{braces} {sum}\n{name}_count{braces} {n}"); + } + } +} + +/// A request's method as a label: one outside the usual is "other", so a caller cannot make rows. +#[must_use] +pub fn method(m: &str) -> &'static str { + ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"].into_iter().find(|x| *x == m).unwrap_or("other") +} + +/// A label's value as Prometheus's text format reads it. +#[must_use] +pub fn label(v: &str) -> String { + v.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n") +} + +/// Requests answered, by method, route and status (#370). +pub static REQUESTS: Histograms = Histograms::new(SECONDS); +/// Feeds the refresher has checked, and of them those that failed (#370). +pub static CHECKED: AtomicU64 = AtomicU64::new(0); +pub static CHECKS_FAILED: AtomicU64 = AtomicU64::new(0); +/// Snapshots taken from ipx servers and the Apple sync (#370). +pub static SNAPSHOTS: AtomicU64 = AtomicU64::new(0); + +/// The directory's /metrics, in Prometheus's text format: memory, open files, the feeds it has +/// fetched and checked, snapshots and requests. Nothing from the database, so a scrape costs nothing and answers while it is down. fn render(started: Instant) -> String { let mut out = String::new(); let mut gauge = |name: &str, kind: &str, help: &str, labels: &str, v: f64| { @@ -69,6 +140,20 @@ fn render(started: Instant) -> String { &format!("{{version=\"{}\"}}", env!("CARGO_PKG_VERSION")), 1.0, ); + gauge("directory_feed_checks_total", "counter", "Feeds the refresher checked.", "", CHECKED.load(Relaxed) as f64); + gauge( + "directory_feed_checks_failed_total", + "counter", + "Feed checks that failed: the publisher's error, or a feed that would not parse.", + "", + CHECKS_FAILED.load(Relaxed) as f64, + ); + gauge("directory_snapshots_total", "counter", "Snapshots taken.", "", SNAPSHOTS.load(Relaxed) as f64); + REQUESTS.render( + &mut out, + "directory_http_request_duration_seconds", + "Requests answered and the seconds each took, by method, route and status.", + ); out } @@ -92,6 +177,31 @@ pub async fn serve(listener: TcpListener) { #[cfg(test)] mod tests { + use super::*; + + #[test] + fn a_histogram_is_written_as_prometheus_reads_it() { + static H: Histograms = Histograms::new(&[0.1, 1.0]); + for v in [0.05, 0.5, 2.0] { + H.observe("route=\"/a\"", v); + } + H.observe("", 0.01); + let mut out = String::new(); + H.render(&mut out, "t_seconds", "Help."); + for line in [ + "# TYPE t_seconds histogram\n", + "t_seconds_bucket{route=\"/a\",le=\"0.1\"} 1\n", + "t_seconds_bucket{route=\"/a\",le=\"1\"} 2\n", + "t_seconds_bucket{route=\"/a\",le=\"+Inf\"} 3\n", + "t_seconds_count{route=\"/a\"} 3\n", + "t_seconds_bucket{le=\"0.1\"} 1\n", + "t_seconds_sum 0.01\n", + ] { + assert!(out.contains(line), "{line} in {out}"); + } + assert_eq!(label("a \"b\"\\"), "a \\\"b\\\"\\\\"); + } + #[test] fn metrics_say_memory_and_feeds() { let text = super::render(std::time::Instant::now()); diff --git a/crates/directory/src/otel.rs b/crates/directory/src/otel.rs index 207b4d8..fb47679 100644 --- a/crates/directory/src/otel.rs +++ b/crates/directory/src/otel.rs @@ -17,6 +17,7 @@ pub fn provider(service: &'static str) -> anyhow::Result Option { sc.is_valid().then(|| format!("00-{}-{}-{:02x}", sc.trace_id(), sc.span_id(), sc.trace_flags().to_u8())) } +/// Random ids, as the SDK's own, but for a span built inside `with_ids`, which takes the ids it is +/// given: the page names its tap's span before ipx makes it (#376), and the SDK gives a span +/// builder no other way to be told them. +#[derive(Debug, Default)] +struct Ids(opentelemetry_sdk::trace::RandomIdGenerator); + +thread_local! { + static GIVEN: std::cell::Cell> = + const { std::cell::Cell::new(None) }; +} + +impl opentelemetry_sdk::trace::IdGenerator for Ids { + fn new_trace_id(&self) -> opentelemetry::trace::TraceId { + GIVEN.get().map_or_else(|| self.0.new_trace_id(), |(t, _)| t) + } + fn new_span_id(&self) -> opentelemetry::trace::SpanId { + GIVEN.get().map_or_else(|| self.0.new_span_id(), |(_, s)| s) + } +} + +/// Runs `f`, which builds one root span, with these ids for it. Synchronous, so nothing else on +/// this thread makes a span in between. +pub fn with_ids( + trace: opentelemetry::trace::TraceId, + span: opentelemetry::trace::SpanId, + f: impl FnOnce() -> R, +) -> R { + GIVEN.set(Some((trace, span))); + let r = f(); + GIVEN.set(None); + r +} + +/// The JSON log line with the trace and span it belongs to (#91), so a line in Loki leads to its +/// trace in Tempo. The JSON formatter cannot take a field of its own, so the ids go on the end of +/// the object it writes; the directory's lines carry them too (#375). A line outside any traced span, or with no trace exporter, is unchanged. +pub struct WithTrace(pub F); + +impl tracing_subscriber::fmt::FormatEvent for WithTrace +where + F: tracing_subscriber::fmt::FormatEvent, + S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, + N: for<'w> tracing_subscriber::fmt::FormatFields<'w> + 'static, +{ + fn format_event( + &self, + ctx: &tracing_subscriber::fmt::FmtContext<'_, S, N>, + mut w: tracing_subscriber::fmt::format::Writer<'_>, + ev: &tracing::Event<'_>, + ) -> std::fmt::Result { + use opentelemetry::trace::TraceContextExt; + use tracing_opentelemetry::OpenTelemetrySpanExt; + let current = tracing::Span::current(); + let sc = if current.is_none() { None } else { Some(current.context().span().span_context().clone()) }; + let Some(sc) = sc.filter(|c| c.is_valid()) else { + return self.0.format_event(ctx, w, ev); + }; + let mut line = String::new(); + self.0.format_event(ctx, tracing_subscriber::fmt::format::Writer::new(&mut line), ev)?; + match line.trim_end().strip_suffix('}') { + Some(body) => writeln!(w, r#"{body},"trace_id":"{}","span_id":"{}"}}"#, sc.trace_id(), sc.span_id()), + None => w.write_str(&line), + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -92,4 +159,21 @@ mod tests { assert_eq!(q.unwrap().span().span_context().trace_id().to_string(), "4bf92f3577b34da6a3ce929d0e0e4736"); assert!(remote_parent(&axum::http::HeaderMap::new(), Some("traceparent=nonsense")).is_none()); } + + #[test] + fn a_span_built_with_given_ids_takes_them_and_the_next_does_not() { + use opentelemetry::trace::{Span as _, SpanId, TraceId, Tracer as _, TracerProvider as _}; + let tracer = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_id_generator(super::Ids::default()) + .build() + .tracer("t"); + let (t, s) = ( + TraceId::from_hex("4bf92f3577b34da6a3ce929d0e0e4736").unwrap(), + SpanId::from_hex("00f067aa0ba902b7").unwrap(), + ); + let given = super::with_ids(t, s, || tracer.start("tap")); + assert_eq!((given.span_context().trace_id(), given.span_context().span_id()), (t, s)); + let next = tracer.start("other"); + assert_ne!(next.span_context().trace_id(), t); + } } diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 5f9a3e8..2b52c94 100644 --- a/crates/ipx/CHANGELOG.md +++ b/crates/ipx/CHANGELOG.md @@ -12,6 +12,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `/metrics` says how long statements take and, for hand-written SQL, how long they waited for a connection (#367); counts feed failures by kind, new items, downloads saved and failed, and the silence DeadAir found (#368); and holds the page's own timings, a click to its list, play to sound, a stall (#369). +- Traces for what had none: each command the worker runs, from the CLI, the socket or the page, with its job inside (#373); the daemon's start, from opening the database to serving (#374); a list's sync and an OPML upload, step by step (#372); and the tap or key in the page that set a trace's requests off, as its root span, from the tap to its last answer (#376). + ### Changed - The log no longer has a line for every request that went well, over half its size: `/metrics` counts each request and its time by route and status as `ipx_http_request_duration_seconds`, a request that failed keeps its warning, and the admin page's log view still shows them all (#364). @@ -19,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Syncing an OPML or Patreon list reads who subscribes to it once, not once for every feed it lists: a list of 922 was thousands of the same queries a sync (#377). - A feed you already have, added again by the same address over http rather than https or the other way round, is the one you have, not a second subscription with every item twice. Changing a feed's address to another feed's over the other scheme is refused the same way (#362). - A podcast that was once in an OPML and is now a feed of its own is no longer hidden from All Subscriptions when that OPML is, so its unread items are listed there as the feed list counts them (#358). diff --git a/crates/ipx/src/daemon.rs b/crates/ipx/src/daemon.rs index e0a592c..94e5974 100644 --- a/crates/ipx/src/daemon.rs +++ b/crates/ipx/src/daemon.rs @@ -12,6 +12,7 @@ use std::time::{Duration, Instant}; use tokio::runtime::Handle; use tokio::sync::{broadcast, mpsc, watch}; use tokio::task::JoinHandle; +use tracing::Instrument; /// The catalogue and server settings as the database now has them, in place of the copy in /// memory, for a daemon told another ipx changed them (#120); config.toml, read again, for where @@ -49,61 +50,69 @@ pub(crate) async fn daemon( config_path: PathBuf, web_addr: Option, events: broadcast::Sender, + start: tracing::Span, ) -> Result<()> { let socket = ctx.cfg().general.socket.clone(); if ipc::daemon_is_live(&socket).await { anyhow::bail!("a daemon is already listening on {}", socket.display()); } - // A database with nobody in it cannot be signed into. - if ctx.db.users().await?.is_empty() { - let dir = ctx.config_path.parent().unwrap_or(Path::new(".")); - let (password, told) = auth::first_password(dir, std::env::var("IPX_ADMIN_PASSWORD").ok())?; - ctx.db.create_user("admin", Some(&crate::auth::hash_password(&password)?), /* admin */ true).await?; - tracing::warn!( - "no accounts yet: created 'admin' {told}. Change it with `echo -n | ipx user passwd admin`" - ); - } + // The rest of start-up, in the trace main began (#374): how long a deploy took to answer, and + // what held it. + async { + // A database with nobody in it cannot be signed into. + if ctx.db.users().instrument(db::step("users")).await?.is_empty() { + let dir = ctx.config_path.parent().unwrap_or(Path::new(".")); + let (password, told) = auth::first_password(dir, std::env::var("IPX_ADMIN_PASSWORD").ok())?; + ctx.db.create_user("admin", Some(&crate::auth::hash_password(&password)?), /* admin */ true).await?; + tracing::warn!( + "no accounts yet: created 'admin' {told}. Change it with `echo -n | ipx user passwd admin`" + ); + } - // Unread lists start from here, and a reap may be a day off (#347). Slower without, never wrong. - if let Err(e) = ctx.db.advance_unread_from().await { - tracing::warn!(error = %format!("{e:#}"), "could not work out where unread items start"); - } + // Unread lists start from here, and a reap may be a day off (#347). Slower without, never wrong. + if let Err(e) = ctx.db.advance_unread_from().instrument(db::step("advance_unread_from")).await { + tracing::warn!(error = %format!("{e:#}"), "could not work out where unread items start"); + } - if let Some(admin) = ctx.db.users().await?.into_iter().find(|u| u.is_admin) { - let catalogue: Vec = ctx.cfg().feeds.keys().cloned().collect(); - match ctx.db.adopt_catalogue(admin.id, &catalogue).await { + if let Some(admin) = ctx.db.users().await?.into_iter().find(|u| u.is_admin) { + let catalogue: Vec = ctx.cfg().feeds.keys().cloned().collect(); + match ctx.db.adopt_catalogue(admin.id, &catalogue).instrument(db::step("adopt_catalogue")).await { + Ok(0) => {} + Ok(n) => tracing::info!(user = %admin.name, feeds = n, "subscribed the first admin to the catalogue"), + Err(e) => tracing::error!(error = %e, "could not subscribe the first admin to the catalogue"), + } + } + + match ctx.db.requeue_interrupted().instrument(db::step("requeue_interrupted")).await { + Ok(n) if n > 0 => tracing::info!(count = n, "requeued downloads interrupted by a restart"), + Ok(_) => {} + Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not requeue interrupted downloads"), + } + + match catalogue::retire_stranded(&ctx).instrument(db::step("retire_stranded")).await { Ok(0) => {} - Ok(n) => tracing::info!(user = %admin.name, feeds = n, "subscribed the first admin to the catalogue"), - Err(e) => tracing::error!(error = %e, "could not subscribe the first admin to the catalogue"), + Ok(n) => tracing::info!(feeds = n, "retired feeds whose OPML is no longer in config"), + Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not retire feeds whose OPML is no longer in config"), } - } - - match ctx.db.requeue_interrupted().await { - Ok(n) if n > 0 => tracing::info!(count = n, "requeued downloads interrupted by a restart"), - Ok(_) => {} - Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not requeue interrupted downloads"), - } - - match catalogue::retire_stranded(&ctx).await { - Ok(0) => {} - Ok(n) => tracing::info!(feeds = n, "retired feeds whose OPML is no longer in config"), - Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not retire feeds whose OPML is no longer in config"), - } - match catalogue::regroup_unmanaged(&ctx).await { - Ok(0) => {} - Ok(n) => tracing::info!(feeds = n, "took feeds of their own out of their old list's group"), - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), "could not take feeds of their own out of their old list's group") + match catalogue::regroup_unmanaged(&ctx).instrument(db::step("regroup_unmanaged")).await { + Ok(0) => {} + Ok(n) => tracing::info!(feeds = n, "took feeds of their own out of their old list's group"), + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "could not take feeds of their own out of their old list's group") + } } + anyhow::Ok(()) } + .instrument(start.clone()) + .await?; // Room for any one request's commands: a web request waits inside itself for a free place, // and Download latest sends up to 100. With 64 the rest waited on downloads an hour long, and // every other request that queues work waited behind them (#154). let (tx_cmd, mut rx_cmd) = mpsc::channel::(1024); - let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).await?; + let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).instrument(start.clone()).await?; // status and reload are answered by the socket itself; everything else waits its turn in the // queue. let answer: ipc::StatusFn = { @@ -123,7 +132,9 @@ pub(crate) async fn daemon( let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd, answer)); // One command at a time: the queue is what keeps two scans from overlapping. - tracing::info!(feeds = catalogue::subscriptions(&ctx).await.map(|s| s.len()).unwrap_or(0), "daemon started"); + let feeds = catalogue::subscriptions(&ctx).await.map(|s| s.len()).unwrap_or(0); + start.in_scope(|| tracing::info!(feeds, "daemon started")); + drop(start); // A signal has to be able to interrupt work in progress, not just the wait between // jobs. Racing `shutdown()` in the outer select only cancels branch selection: once @@ -191,7 +202,10 @@ pub(crate) async fn daemon( "-> {}", serde_json::to_string(&cmd).unwrap_or_else(|_| format!("{cmd:?}")) ); - if !until_stopped(&ctx, &rx_stop, cli::run(&ctx, cmd)).await { + // A trace a command, its job inside it (#373): from the CLI or the page alike. + let kind = serde_json::to_value(&cmd).ok().and_then(|v| v["cmd"].as_str().map(str::to_owned)); + let span = tracing::info_span!("command", cmd = kind.as_deref().unwrap_or("?")); + if !until_stopped(&ctx, &rx_stop, cli::run(&ctx, cmd).instrument(span)).await { break; } } diff --git a/crates/ipx/src/db.rs b/crates/ipx/src/db.rs index 087629d..9d88376 100644 --- a/crates/ipx/src/db.rs +++ b/crates/ipx/src/db.rs @@ -275,7 +275,34 @@ async fn connect(location: &str, keep_open: bool) -> Result if keep_open && is_postgres(location) { opts.min_connections(20); } - Database::connect(opts).await.with_context(|| format!("opening {}", redact(location))) + let mut db = Database::connect(opts).await.with_context(|| format!("opening {}", redact(location)))?; + // Each statement's time once it has a connection (#367), by its first word: the SQL itself + // would be a row a query. + db.set_metric_callback(|info| { + let first = info.statement.sql.split_whitespace().next().unwrap_or("").to_ascii_lowercase(); + let kind = + if ["select", "insert", "update", "delete", "with"].contains(&first.as_str()) { &first } else { "other" }; + crate::metrics::QUERIES.observe(&format!("kind=\"{kind}\""), info.elapsed.as_secs_f64()); + LAST_QUERY.set(Some(info.elapsed)); + }); + Ok(db) +} + +thread_local! { + /// The time of the statement this thread last ran, for `waited`. SeaORM calls the metric + /// callback in the same poll that hands the result back to `exec` and `rows`, with no await + /// between, so the value read there is that statement's own. + static LAST_QUERY: std::cell::Cell> = const { std::cell::Cell::new(None) }; +} + +/// Counts how long a statement waited for a connection: its whole call less its time on one +/// (#367). Only hand-written SQL, about half of ipx's queries, comes through here; SeaORM's own +/// queries are timed but their wait is not seen. +/// shortcut: half the queries, measured by difference; a wrapper on sqlx's pool would see all. +fn waited(started: std::time::Instant) { + if let Some(q) = LAST_QUERY.take() { + crate::metrics::DB_WAIT.observe("", started.elapsed().saturating_sub(q).as_secs_f64()); + } } /// The URL `connect` hands sqlx for a location: a SQLite file, created if missing, or a Postgres @@ -1705,11 +1732,18 @@ impl Db { /// Runs a statement; how many rows it changed. async fn exec(&self, sql: &str, values: Vec) -> Result { - Ok(self.orm.execute_raw(self.stmt(sql, values)).await?.rows_affected()) + let (stmt, started) = (self.stmt(sql, values), std::time::Instant::now()); + // Only on success: a connection never had means no callback, and a stale time. + let n = self.orm.execute_raw(stmt).await?.rows_affected(); + waited(started); + Ok(n) } async fn rows(&self, sql: &str, values: Vec) -> Result> { - Ok(self.orm.query_all_raw(self.stmt(sql, values)).await?) + let (stmt, started) = (self.stmt(sql, values), std::time::Instant::now()); + let rows = self.orm.query_all_raw(stmt).await?; + waited(started); + Ok(rows) } /// The first admin starts subscribed to the whole catalogue: whoever wrote config.toml meant diff --git a/crates/ipx/src/ipc.rs b/crates/ipx/src/ipc.rs index ec62a96..9140221 100644 --- a/crates/ipx/src/ipc.rs +++ b/crates/ipx/src/ipc.rs @@ -245,6 +245,7 @@ impl Emitter { // line per feed per tick for "not due yet" would push everything worth reading // out of the buffer within a few minutes. log_event(&e, self.tx.is_some()); + crate::metrics::STATS.event(&e); // A feed not due changes nothing a listener shows, and a scan has one for nearly every // feed: 1,500 at once overflowed the channel's 1,024 for a page slow to read (#197). let routine = matches!(&e, Event::FeedSkip { reason, .. } if reason.starts_with("not due")); diff --git a/crates/ipx/src/lists.rs b/crates/ipx/src/lists.rs index 662c815..13c6ead 100644 --- a/crates/ipx/src/lists.rs +++ b/crates/ipx/src/lists.rs @@ -2,10 +2,11 @@ use crate::ctx::Ctx; use crate::scan::Outcome; -use crate::{catalogue, config, feed}; +use crate::{catalogue, config, db, feed}; use anyhow::Result; use std::collections::{HashMap, HashSet}; use std::sync::Arc; +use tracing::Instrument; /// The feeds an OPML document lists, up to as many as a subscribed list may bring in: each becomes /// a feed in the catalogue, scanned for everyone, and one upload of tens of thousands held the @@ -31,6 +32,7 @@ pub fn opml_feeds(doc: &opml::OPML) -> Result> { /// /// The caller parses the document and lists its feeds (`opml_feeds`), so each refuses a file in /// its own terms, before anything is touched: a 400 from the web, a message from the CLI. +#[tracing::instrument(name = "opml_import", skip_all, fields(feeds = found.len(), user = user_id))] pub async fn subscribe_opml(ctx: &Ctx, found: Vec<(String, String)>, user_id: i64) -> Result<(usize, usize)> { let mut cfg = ctx.edit_cfg().await; // By address, once: each feed looked through every subscription and the whole catalogue, and @@ -110,6 +112,8 @@ pub(crate) const MAX_LISTED: usize = 2000; /// the list is unsubscribed *only if nothing was ever downloaded for it* -- otherwise it /// is kept and flagged, because dropping it would orphan files on disk with nothing in /// the UI to explain them. +/// A trace of its own steps (#372): one long `feed` span said nothing of where a slow sync went. +#[tracing::instrument(name = "list_sync", skip_all, fields(list = %parent_id, listed = listed.len()))] pub(crate) async fn sync_group(ctx: &Arc, parent_id: &str, listed: &[(String, String)]) -> Result { // Each is a feed scanned for everyone subscribed to the list, and the list is its author's to // write: one of tens of thousands held the scan for hours (#337). davewiner's had 922. @@ -119,74 +123,94 @@ pub(crate) async fn sync_group(ctx: &Arc, parent_id: &str, listed: &[(Strin listed.len() ); let cfg = ctx.cfg(); - let existing = ctx.db.managed_feeds().await?; + let existing = ctx.db.managed_feeds().instrument(db::step("managed_feeds")).await?; let mut added = vec![]; // Read once, not once a new feed: each read is the whole feeds table. - let known = ctx.db.feed_urls().await?; + let known = ctx.db.feed_urls().instrument(db::step("feed_urls")).await?; let fixed: HashSet<&str> = cfg.feeds.keys().chain(existing.iter().map(|m| &m.id)).map(String::as_str).collect(); let known_ids: HashSet<&str> = known.iter().map(|(id, _)| id.as_str()).collect(); - for (title, url) in listed { - // Already known, whether derived or promoted into the config. - if let Some(m) = existing.iter().find(|m| &m.url == url) { - ctx.db.upsert_managed(&m.id, url, title, parent_id).await?; - continue; + async { + for (title, url) in listed { + // Already known, whether derived or promoted into the config. + if let Some(m) = existing.iter().find(|m| &m.url == url) { + ctx.db.upsert_managed(&m.id, url, title, parent_id).await?; + continue; + } + // A Patreon show you added by hand may be spelled differently from the one listed. + if cfg.feeds.values().any(|f| feed::same_feed(&f.url, url)) { + continue; + } + // A removed feed keeps its rows, so its id is only free again for the same feed. + let its_own: Vec<&str> = + known.iter().filter(|(_, u)| feed::same_feed(u, url)).map(|(id, _)| id.as_str()).collect(); + let id = config::unique_slug(title, |s| { + fixed.contains(s) || added.iter().any(|a| a == s) || (known_ids.contains(s) && !its_own.contains(&s)) + }); + ctx.db.upsert_managed(&id, url, title, parent_id).await?; + added.push(id); } - // A Patreon show you added by hand may be spelled differently from the one listed. - if cfg.feeds.values().any(|f| feed::same_feed(&f.url, url)) { - continue; - } - // A removed feed keeps its rows, so its id is only free again for the same feed. - let its_own: Vec<&str> = - known.iter().filter(|(_, u)| feed::same_feed(u, url)).map(|(id, _)| id.as_str()).collect(); - let id = config::unique_slug(title, |s| { - fixed.contains(s) || added.iter().any(|a| a == s) || (known_ids.contains(s) && !its_own.contains(&s)) - }); - ctx.db.upsert_managed(&id, url, title, parent_id).await?; - added.push(id); + anyhow::Ok(()) } + .instrument(db::step("upsert_listed")) + .await?; // Whoever subscribes to the OPML subscribes to what it lists: that is what taking a - // subscription means. Their own feeds are untouched. - let managed = ctx.db.managed_feeds().await?; - let list_and_feeds = managed - .iter() - .filter(|m| m.group_id == parent_id) - .map(|m| m.id.clone()) - .chain(std::iter::once(parent_id.to_string())); - for id in list_and_feeds { + // subscription means. Their own feeds are untouched. Who reads the list is asked once: asked + // for each listed feed, a list of 922 was thousands of the same queries a sync (#377). + async { + let mut readers = vec![]; for user in ctx.db.users().await? { if ctx.db.subscription(user.id, parent_id).await?.is_some() { - ctx.db.subscribe(user.id, &id).await?; + readers.push(user.id); } } + let managed = ctx.db.managed_feeds().await?; + let list_and_feeds = managed + .iter() + .filter(|m| m.group_id == parent_id) + .map(|m| m.id.clone()) + .chain(std::iter::once(parent_id.to_string())); + for id in list_and_feeds { + for &user in &readers { + ctx.db.subscribe(user, &id).await?; + } + } + anyhow::Ok(()) } + .instrument(db::step("subscribe_readers")) + .await?; // Anything in this group the OPML no longer lists. let mut removed = 0; let mut kept = 0; let parent = ctx.cfg().feeds.get(parent_id).cloned(); - let in_group = existing.iter().filter(|m| m.group_id == parent_id); - for m in in_group { - if listed.iter().any(|(_, u)| u == &m.url) { - continue; - } - if catalogue::promote_kept(ctx, m, parent.as_ref()).await? { - kept += 1; - tracing::info!(feed = %m.id, "dropped from the OPML but someone kept it; a feed of its own now"); - continue; - } - if ctx.db.downloaded_count(&m.id).await.unwrap_or(1) > 0 { - // Never orphan a downloaded file: keep the feed and say why in the UI. - ctx.db.set_orphaned(&m.id, true).await?; - kept += 1; - tracing::info!(feed = %m.id, "dropped from the OPML but has downloads; keeping it"); - } else { - ctx.db.drop_managed(&m.id).await?; - removed += 1; - tracing::info!(feed = %m.id, "dropped from the OPML with nothing downloaded; removed"); + async { + let in_group = existing.iter().filter(|m| m.group_id == parent_id); + for m in in_group { + if listed.iter().any(|(_, u)| u == &m.url) { + continue; + } + if catalogue::promote_kept(ctx, m, parent.as_ref()).await? { + kept += 1; + tracing::info!(feed = %m.id, "dropped from the OPML but someone kept it; a feed of its own now"); + continue; + } + if ctx.db.downloaded_count(&m.id).await.unwrap_or(1) > 0 { + // Never orphan a downloaded file: keep the feed and say why in the UI. + ctx.db.set_orphaned(&m.id, true).await?; + kept += 1; + tracing::info!(feed = %m.id, "dropped from the OPML but has downloads; keeping it"); + } else { + ctx.db.drop_managed(&m.id).await?; + removed += 1; + tracing::info!(feed = %m.id, "dropped from the OPML with nothing downloaded; removed"); + } } + anyhow::Ok(()) } + .instrument(db::step("drop_unlisted")) + .await?; Ok(Outcome::Opml { added, removed, kept, total: listed.len() }) } diff --git a/crates/ipx/src/logging.rs b/crates/ipx/src/logging.rs index 727d504..190c209 100644 --- a/crates/ipx/src/logging.rs +++ b/crates/ipx/src/logging.rs @@ -1,45 +1,10 @@ //! Log lines as JSON for Loki, carrying the trace and span they belong to. use crate::logbuf; +use ipx_directory::otel::WithTrace; use std::io::IsTerminal; -use tracing::Subscriber; use tracing_subscriber::EnvFilter; use tracing_subscriber::fmt::format::JsonFields; -use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields}; -use tracing_subscriber::registry::LookupSpan; - -/// The JSON log line with the trace and span it belongs to (#91), so a line in Loki leads to its -/// trace in Tempo. The JSON formatter cannot take a field of its own, so the ids go on the end of -/// the object it writes. A line outside any traced span, or with no trace exporter, is unchanged. -pub(crate) struct WithTrace(F); - -impl FormatEvent for WithTrace -where - F: FormatEvent, - S: Subscriber + for<'a> LookupSpan<'a>, - N: for<'w> FormatFields<'w> + 'static, -{ - fn format_event( - &self, - ctx: &FmtContext<'_, S, N>, - mut w: tracing_subscriber::fmt::format::Writer<'_>, - ev: &tracing::Event<'_>, - ) -> std::fmt::Result { - use opentelemetry::trace::TraceContextExt; - use tracing_opentelemetry::OpenTelemetrySpanExt; - let current = tracing::Span::current(); - let sc = if current.is_none() { None } else { Some(current.context().span().span_context().clone()) }; - let Some(sc) = sc.filter(|c| c.is_valid()) else { - return self.0.format_event(ctx, w, ev); - }; - let mut line = String::new(); - self.0.format_event(ctx, tracing_subscriber::fmt::format::Writer::new(&mut line), ev)?; - match line.trim_end().strip_suffix('}') { - Some(body) => writeln!(w, r#"{body},"trace_id":"{}","span_id":"{}"}}"#, sc.trace_id(), sc.span_id()), - None => w.write_str(&line), - } - } -} /// Sets up logging: to stderr, as text or as JSON (`IPX_LOG_FORMAT`), into the ring the page's log /// reads, and to the trace exporter when there is one. diff --git a/crates/ipx/src/main.rs b/crates/ipx/src/main.rs index 3dcb903..0e7e0c4 100644 --- a/crates/ipx/src/main.rs +++ b/crates/ipx/src/main.rs @@ -35,6 +35,7 @@ use anyhow::Result; use clap::Parser; use std::sync::Arc; use tokio::sync::broadcast; +use tracing::Instrument; #[tokio::main] async fn main() -> Result<()> { @@ -46,12 +47,21 @@ async fn main() -> Result<()> { _ => None, }; logging::init(otel.as_ref()); + // For the spans the page reports of itself (#376), which ipx builds with the page's own ids. + if let Some(p) = &otel { + opentelemetry::global::set_tracer_provider(p.clone()); + } let config_path = cli.config.clone().unwrap_or_else(config::config_path); let cfg = config::Config::load(&config_path)?; // The daemon holds its connections open; a one-off command opens what it uses. let keep_open = matches!(cli.command, Command::Daemon { .. }); - let db = db::Db::open(&db::location(), keep_open).await?; + // The daemon's start-up is a trace (#374), from opening the database to serving; daemon() + // carries it on. + let start = if keep_open { tracing::info_span!("start") } else { tracing::Span::none() }; + let db = async { db::Db::open(&db::location(), keep_open).instrument(db::step("open_database")).await } + .instrument(start.clone()) + .await?; // A daemon owns the state; don't have two processes downloading the same thing. let wire_cmd = match &cli.command { @@ -74,7 +84,10 @@ async fn main() -> Result<()> { } let socket = cfg.general.socket.clone(); - let cfg = catalogue::assemble_config(&db, cfg, &config_path).await?; + let cfg = + async { catalogue::assemble_config(&db, cfg, &config_path).instrument(db::step("assemble_config")).await } + .instrument(start.clone()) + .await?; let is_daemon = matches!(cli.command, Command::Daemon { .. }); // These write the catalogue from a process of their own, which a running daemon, holding its @@ -88,7 +101,7 @@ async fn main() -> Result<()> { let result = match cli.command { Command::List => catalogue::list(&ctx).await, - Command::Daemon { web } => daemon::daemon(ctx, config_path, web, events).await, + Command::Daemon { web } => daemon::daemon(ctx, config_path, web, events, start).await, Command::Add { url, folder, keywords, category, list } => { catalogue::add(&ctx, &url, folder, keywords, category, list).await } diff --git a/crates/ipx/src/metrics.rs b/crates/ipx/src/metrics.rs index a725f2a..f0e7951 100644 --- a/crates/ipx/src/metrics.rs +++ b/crates/ipx/src/metrics.rs @@ -7,6 +7,7 @@ //! the text format is a few lines, and a client library would be a dependency for them. use axum::Router; +use ipx_directory::metrics::{Histograms, SECONDS, label, method}; use std::collections::BTreeMap; use std::fmt::Write as _; use std::net::SocketAddr; @@ -62,16 +63,22 @@ pub struct Stats { job: std::sync::Mutex>, /// Jobs finished, by kind: how many, and their seconds all told. jobs: std::sync::Mutex>, - /// Requests answered, by method, route and status: how many within each of `HTTP_BUCKETS`, - /// how many all told, and their seconds. What the access log's lines were read for (#364). - http: std::sync::Mutex>, + /// Feed failures by kind, `error.type` (#368). + feed_failures: std::sync::Mutex>, + pub new_items: AtomicU64, + pub downloads_done: AtomicU64, + pub downloads_failed: AtomicU64, + pub deadair_saved_seconds: AtomicU64, } -/// The upper bounds, in seconds, a request's time is counted under: most of ipx's answer in tens -/// of milliseconds, and a slow one, a host stalling, in seconds. -const HTTP_BUCKETS: [f64; 11] = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]; - -type Requests = ([u64; HTTP_BUCKETS.len()], u64, f64); +/// Requests answered, by method, route and status (#364). +pub static REQUESTS: Histograms = Histograms::new(SECONDS); +/// Each statement's time once it has a connection, by its first word (#367). +pub static QUERIES: Histograms = Histograms::new(SECONDS); +/// The wait for a connection, as hand-written SQL in db.rs sees it (#367). +pub static DB_WAIT: Histograms = Histograms::new(SECONDS); +/// The page's timings, by `ev`: a click to its list drawn, play to sound, a stall (#369). +pub static PAGE: Histograms = Histograms::new(&[0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]); impl Stats { pub const fn new() -> Self { @@ -98,24 +105,47 @@ impl Stats { people: std::sync::Mutex::new(BTreeMap::new()), job: std::sync::Mutex::new(None), jobs: std::sync::Mutex::new(BTreeMap::new()), - http: std::sync::Mutex::new(BTreeMap::new()), + feed_failures: std::sync::Mutex::new(BTreeMap::new()), + new_items: AtomicU64::new(0), + downloads_done: AtomicU64::new(0), + downloads_failed: AtomicU64::new(0), + deadair_saved_seconds: AtomicU64::new(0), } } /// Counts a request answered. `route` is the pattern matched, "" for none, so a scanner's - /// made-up paths are one row; a method outside the usual is "other", for the same reason. - pub fn request(&self, method: &str, route: &str, status: u16, secs: f64) { - let method = ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"] - .into_iter() - .find(|m| *m == method) - .unwrap_or("other"); - let mut http = self.http.lock().expect("a panic elsewhere poisoned this lock"); - let (buckets, n, sum) = http.entry((method, route.to_owned(), status)).or_default(); - for (b, le) in buckets.iter_mut().zip(HTTP_BUCKETS) { - *b += u64::from(secs <= le); + /// made-up paths are one row. + pub fn request(&self, m: &str, route: &str, status: u16, secs: f64) { + REQUESTS.observe(&format!("method=\"{}\",route=\"{}\",status=\"{status}\"", method(m), label(route)), secs); + } + + /// Counts what an event says happened (#368): one place, as every event passes `emit`. + pub fn event(&self, e: &crate::ipc::Event) { + use crate::ipc::Event; + match e { + Event::FeedDone { new, .. } => { + self.new_items.fetch_add(*new as u64, Relaxed); + } + Event::FeedError { msg, .. } => { + let kind = crate::feed::failure_kind(msg).0; + *self + .feed_failures + .lock() + .expect("a panic elsewhere poisoned this lock") + .entry(kind.to_string()) + .or_default() += 1; + } + Event::DownloadDone { .. } => { + self.downloads_done.fetch_add(1, Relaxed); + } + Event::DownloadError { .. } => { + self.downloads_failed.fetch_add(1, Relaxed); + } + Event::DeadairDone { saved_s, .. } => { + self.deadair_saved_seconds.fetch_add((*saved_s).max(0) as u64, Relaxed); + } + _ => {} } - *n += 1; - *sum += secs; } /// Adds to what `who` used on `key` ("server" or "own"): items judged, fits, input and output @@ -334,19 +364,42 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li "Web requests being answered.", &[("", st.http_in_flight.load(Relaxed) as f64)], ); - let http = st.http.lock().expect("a panic elsewhere poisoned this lock").clone(); - let name = "ipx_http_request_duration_seconds"; - let _ = writeln!( + REQUESTS.render( out, - "# HELP {name} Web requests answered and the seconds each took, by method, route and status.\n# TYPE {name} histogram" + "ipx_http_request_duration_seconds", + "Web requests answered and the seconds each took, by method, route and status.", + ); + QUERIES.render(out, "ipx_db_query_seconds", "Statements' time once they had a connection, by their first word."); + DB_WAIT.render(out, "ipx_db_wait_seconds", "The wait for a database connection, as hand-written SQL saw it."); + PAGE.render( + out, + "ipx_page_seconds", + "What the page timed: a click to its list drawn, play to sound, a stall's wait.", + ); + let failures = st.feed_failures.lock().expect("a panic elsewhere poisoned this lock").clone(); + let labels: Vec = failures.keys().map(|k| format!("error_type=\"{}\"", label(k))).collect(); + counter( + out, + "ipx_feed_failures_total", + "Feeds that failed to fetch or parse, by kind: an HTTP status, dns, timeout, tls, parse...", + &labels.iter().zip(failures.values()).map(|(l, v)| (l.as_str(), *v as f64)).collect::>(), + ); + counter(out, "ipx_new_items_total", "New items scans found.", &[("", st.new_items.load(Relaxed) as f64)]); + counter( + out, + "ipx_downloads_total", + "Downloads finished, saved or failed.", + &[ + ("result=\"saved\"", st.downloads_done.load(Relaxed) as f64), + ("result=\"failed\"", st.downloads_failed.load(Relaxed) as f64), + ], + ); + counter( + out, + "ipx_deadair_saved_seconds_total", + "Seconds of silence DeadAir found in the files it looked at.", + &[("", st.deadair_saved_seconds.load(Relaxed) as f64)], ); - for ((method, route, status), (buckets, n, sum)) in &http { - let l = format!("method=\"{method}\",route=\"{}\",status=\"{status}\"", label(route)); - for (le, b) in HTTP_BUCKETS.iter().zip(buckets) { - let _ = writeln!(out, "{name}_bucket{{{l},le=\"{le}\"}} {b}"); - } - let _ = writeln!(out, "{name}_bucket{{{l},le=\"+Inf\"}} {n}\n{name}_sum{{{l}}} {sum}\n{name}_count{{{l}}} {n}"); - } gauge( out, "ipx_scan_feeds_fetching", @@ -468,11 +521,6 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li } } -/// A label's value as Prometheus's text format reads it. -fn label(v: &str) -> String { - v.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n") -} - /// A counter, which only goes up while the daemon runs, in Prometheus's text format. pub fn counter(out: &mut String, name: &str, help: &str, samples: &[(&str, f64)]) { write(out, "counter", name, help, samples); @@ -556,6 +604,8 @@ mod tests { T.request("GET", "/api/feeds", 200, secs); } T.request("PROPFIND", "", 404, 0.001); + T.event(&crate::ipc::Event::FeedError { feed: "f".into(), msg: "connecting: timed out".into() }); + T.event(&crate::ipc::Event::FeedDone { feed: "f".into(), new: 3, downloaded: 0, failed: 0, torrents: 0 }); let mut out = String::new(); work(&mut out, &T, 0, 0, 0); let l = "method=\"GET\",route=\"/api/feeds\",status=\"200\""; @@ -566,6 +616,8 @@ mod tests { format!("ipx_http_request_duration_seconds_bucket{{{l},le=\"+Inf\"}} 3\n"), format!("ipx_http_request_duration_seconds_count{{{l}}} 3\n"), "ipx_http_request_duration_seconds_count{method=\"other\",route=\"\",status=\"404\"} 1\n".into(), + "ipx_new_items_total 3\n".into(), + "# TYPE ipx_feed_failures_total counter\n".into(), ] { assert!(out.contains(&line), "{line} in {out}"); } diff --git a/crates/ipx/src/web.rs b/crates/ipx/src/web.rs index bd79b9a..0fdaac5 100644 --- a/crates/ipx/src/web.rs +++ b/crates/ipx/src/web.rs @@ -1410,10 +1410,35 @@ mod tests { deadair: None, ms: Some(5), filter: Some("unread".into()), + trace: None, + span: None, + start: None, }; for ev in ["page_error", "rejection", "player_error", "play_start", "stall", "list_shown"] { assert_eq!(client_event(u.clone(), Json(e(ev))).await, StatusCode::NO_CONTENT, "{ev}"); } + // A tap's span takes a trace's ids and a time the server can believe (#376). + let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as u64; + let tap = |trace: &str, start: u64| ClientEvent { + trace: Some(trace.into()), + span: Some("00f067aa0ba902b7".into()), + start: Some(start), + msg: Some("pointerdown".into()), + ..e("tap") + }; + let good = "4bf92f3577b34da6a3ce929d0e0e4736"; + assert_eq!(client_event(u.clone(), Json(tap(good, now - 2000))).await, StatusCode::NO_CONTENT); + assert_eq!(client_event(u.clone(), Json(tap("not hex", now))).await, StatusCode::BAD_REQUEST); + assert_eq!(client_event(u.clone(), Json(tap(&"0".repeat(32), now))).await, StatusCode::BAD_REQUEST); + assert_eq!( + client_event(u.clone(), Json(tap(good, now - 7_200_000))).await, + StatusCode::BAD_REQUEST, + "hours ago" + ); + assert_eq!( + client_event(u.clone(), Json(ClientEvent { trace: None, ..tap(good, now) })).await, + StatusCode::BAD_REQUEST + ); assert_eq!(client_event(u, Json(e("anything"))).await, StatusCode::BAD_REQUEST); } @@ -2754,6 +2779,10 @@ struct ClientEvent { ms: Option, /// The list's filter, for `list_shown`: all, unread, flagged. filter: Option, + /// For `tap`: the trace the page began and its root span, hex, and when it began, Unix ms. + trace: Option, + span: Option, + start: Option, } async fn client_event(user: crate::db::User, Json(e): Json) -> StatusCode { @@ -2763,6 +2792,16 @@ async fn client_event(user: crate::db::User, Json(e): Json) -> Stat let (ev, line, enclosure, deadair, ms, user) = (e.ev.as_str(), e.line, e.enclosure, e.deadair, e.ms, user.name.as_str()); match ev { + // Sent to Tempo, not the log: a line a tap would be most of it. + "tap" => { + let ok = match (&e.trace, &e.span, e.start, ms) { + (Some(t), Some(s), Some(start), Some(ms)) => { + page_span(t, s, start, ms, msg.as_deref().unwrap_or(""), user) + } + _ => false, + }; + return if ok { StatusCode::NO_CONTENT } else { StatusCode::BAD_REQUEST }; + } "page_error" | "rejection" | "player_error" => tracing::warn!( target: "ipx::client", ev, msg, src, line, feed, enclosure, deadair, user, "{user}'s page: {ev}: {}", msg.as_deref().unwrap_or("") @@ -2778,9 +2817,41 @@ async fn client_event(user: crate::db::User, Json(e): Json) -> Stat ), _ => return StatusCode::BAD_REQUEST, } + // The timings in /metrics as well, so weeks of them are percentiles without reading each line + // (#369). Capped at a minute: the clock is the browser's. + if let (Some(ms), "play_start" | "stall" | "list_shown") = (ms, ev) { + crate::metrics::PAGE.observe(&format!("ev=\"{ev}\""), ms.min(60_000) as f64 / 1000.0); + } StatusCode::NO_CONTENT } +/// The root span of a trace the page began with a tap or a key (#376), with the page's own ids, so +/// the requests it sent, which name it as their parent, hang from it in Tempo: how long from the +/// tap to its last answer. The browser's clock, so only a span begun in the last hour and at most +/// a minute long. False for ids that are not a trace's. +fn page_span(trace: &str, span: &str, start_ms: u64, ms: u64, what: &str, user: &str) -> bool { + use opentelemetry::trace::{Span as _, SpanId, SpanKind, TraceId, Tracer as _}; + let (Ok(t), Ok(s)) = (TraceId::from_hex(trace), SpanId::from_hex(span)) else { return false }; + if trace.len() != 32 || span.len() != 16 || t == TraceId::INVALID || s == SpanId::INVALID || ms > 60_000 { + return false; + } + let start = std::time::UNIX_EPOCH + std::time::Duration::from_millis(start_ms); + let now = std::time::SystemTime::now(); + if now.duration_since(start).map_or(true, |ago| ago > std::time::Duration::from_secs(3600)) { + return false; + } + let tracer = opentelemetry::global::tracer("ipx"); + let b = tracer + .span_builder(if what == "keydown" { "key" } else { "tap" }) + .with_kind(SpanKind::Client) + .with_start_time(start) + .with_attributes([opentelemetry::KeyValue::new("user", user.to_owned())]); + // Built with the page's ids, which its requests already name as their parent. + let mut span = ipx_directory::otel::with_ids(t, s, || tracer.build_with_context(b, &opentelemetry::Context::new())); + span.end_with_timestamp(start + std::time::Duration::from_millis(ms)); + true +} + /// Names a request's trace by its route, `POST /api/entries/{feed_id}/{guid}/flags`, once routing /// has found it. Named by the path `access_log` sees, every item's GUID was a trace name of its /// own, and nothing grouped. A route layer, because only one runs after routing. Renamed on the diff --git a/grafana/alerts.py b/grafana/alerts.py new file mode 100644 index 0000000..5ae55e4 --- /dev/null +++ b/grafana/alerts.py @@ -0,0 +1,88 @@ +"""The iPX alert rules in Grafana (#365): what should be looked at now rather than at the next prod +check, written out as Grafana's alerting provisioning, beside the dashboard. + + python3 grafana/alerts.py > /mnt/fast/arcane/projects/monitoring/grafana-provisioning/alerting/ipx.json + +Grafana reads it at start, or at once from POST /api/admin/provisioning/alerting/reload. There is +no contact point yet: a rule that fires shows in Grafana's Alerting page, and adding one (email, +ntfy, a webhook) is a file in the same folder. Each rule is a query and a threshold on it; the +query's value is in the alert, so it says how bad as well as what. +""" +import json + +# Prometheus was added in Grafana's settings, not provisioned, so its uid is Grafana's own. +PROM = "eenybu3ao2t4wb" +LOKI = "P8E80F9AEF21F6940" + + +def rule(uid, title, expr, op, value, for_, summary, *, ds=PROM, window=600, no_data="OK"): + """Fires when `expr` is `op` (gt, lt) `value` for `for_`. No data is fine by default: most of + these metrics are absent until there is something to say (no job running, no scan yet).""" + return { + "uid": f"ipx-{uid}", "title": title, "condition": "C", "for": for_, + "data": [ + {"refId": "A", "relativeTimeRange": {"from": window, "to": 0}, "datasourceUid": ds, + # Loki's instant query is a queryType; Prometheus's a flag. + "model": {"refId": "A", "expr": expr, + **({"queryType": "instant"} if ds == LOKI else {"instant": True})}}, + {"refId": "C", "datasourceUid": "__expr__", + "model": {"refId": "C", "type": "threshold", "expression": "A", + "conditions": [{"evaluator": {"type": op, "params": [value]}}]}}, + ], + "noDataState": no_data, "execErrState": "Error", + "annotations": {"summary": summary}, + "labels": {"service": "ipx"}, + } + + +rules = [ + # Down: Prometheus cannot reach it. A container gone is no data at all, so that fires too. + rule("down", "ipx is down", 'up{job="ipx"}', "lt", 1, "2m", + "Prometheus cannot reach ipx's /metrics: the iPX container is down or restarting.", no_data="Alerting"), + rule("directory-down", "ipx-directory is down", 'up{job="ipx-directory"}', "lt", 1, "5m", + "Prometheus cannot reach the directory's /metrics.", no_data="Alerting"), + rule("web-down", "ipx-web is down", 'up{job="ipx-web"}', "lt", 1, "5m", + "Prometheus cannot reach Caddy's /metrics: the page is likely down too.", no_data="Alerting"), + # The worker: the daemon answers `ipx status` while it is stuck, so only these say so. + rule("scan-stale", "ipx's scans have stopped", 'max(ipx_last_scan_age_seconds{job="ipx"})', "gt", 900, "5m", + "No scan has finished for over 15 minutes; scans run at least every 10. The worker is stuck: its job and trace say on what."), + rule("job-long", "ipx's worker is stuck on one job", 'max(ipx_worker_job_seconds{job="ipx"})', "gt", 1800, "1m", + "One job has run for over 30 minutes (ipx_worker_job_seconds says which kind)."), + rule("pool-full", "ipx's database connections are all in use", + 'max_over_time(ipx_db_connections{job="ipx",state="idle"}[5m])', "lt", 1, "5m", + "No idle connection for five minutes: requests and scans are waiting for one (ipx_db_wait_seconds)."), + rule("5xx", "ipx is answering with server errors", + 'sum(increase(ipx_http_request_duration_seconds_count{job="ipx",status=~"5.."}[10m]))', "gt", 5, "0s", + "More than five 5xx answers in ten minutes; ipx's log has a warning for each, with its trace."), + rule("502", "ipx-web cannot reach ipx", + 'sum(increase(caddy_http_request_duration_seconds_count{job="ipx-web",code=~"502|504"}[10m]))', "gt", 5, "0s", + "Caddy answered 502 or 504 for ipx more than five times in ten minutes: ipx is not answering it."), + rule("jev", "ipx's calls to Jev are refused", + 'sum(increase(ipx_jev_requests_total{job="ipx",result=~"rate_limited|malformed"}[15m]))', "gt", 0, "0s", + "TypeSafe refused calls as over its limits (#277) or malformed (#257): ipx is asking wrongly."), + rule("signals-stale", "Signals have stopped", 'max(ipx_signals_last_pass_age_seconds{job="ipx"})', "gt", 900, "5m", + "The Signals judge has not finished a pass for 15 minutes; its signals trace shows which Jev call holds it."), + rule("directory-refresh", "The directory has stopped checking feeds", + 'sum(increase(directory_feed_checks_total{job="ipx-directory"}[13h]))', "lt", 1, "30m", + "No feed checked in 13 hours; each is due every 12.", window=46800), + # Tower, which everything shares: a stall here is every request slow (2026-10-18, load 160). + rule("load", "Tower is overloaded", 'max(node_load5{job="node"})', "gt", 48, "10m", + "Load over 48 on twelve threads for ten minutes: builds, or processes blocked on a disk."), + rule("memory", "Tower is out of memory", 'min(node_memory_MemAvailable_bytes{job="node"})', "lt", 2e9, "5m", + "Under 2 GB available, with no swap: next comes stalling on page faults, then the OOM killer."), + # The monitoring itself (#366): nothing else watches it. + rule("monitoring-down", "Part of the monitoring is down", 'min by (job) (up{job=~"prometheus|tempo|loki|alloy"})', + "lt", 1, "5m", "Prometheus cannot reach Tempo, Loki or Alloy."), + rule("tempo-blocks", "Tempo cannot read its stored traces", + '(max(tempodb_blocklist_tenant_index_age_seconds) > bool 900) + (max(tempodb_blocklist_length) < bool 1)', + "gt", 0, "10m", + "Tempo's look over its stored blocks is failing (2026-10-25: one empty block); searches see only the last 15 " + "minutes. Its log says which block: move it out of /mnt/user/logs/tempo/blocks/single-tenant."), + rule("logs-stopped", "No lines from iPX reach Loki", 'sum(count_over_time({container="iPX"}[3h]))', "lt", 1, "0s", + "Loki has nothing from iPX in three hours: Alloy is not shipping it, or the container is down.", + ds=LOKI, window=10800, no_data="Alerting"), +] + +print(json.dumps({"apiVersion": 1, "groups": [ + {"orgId": 1, "name": "ipx", "folder": "iPX", "interval": "1m", "rules": rules}, +]}, indent=2)) diff --git a/grafana/dashboard.py b/grafana/dashboard.py index 9f60621..f5595e5 100644 --- a/grafana/dashboard.py +++ b/grafana/dashboard.py @@ -218,6 +218,13 @@ ts("Open connections", [prom('max(ipx_http_requests_in_flight{job="ipx"})', lege prom('sum(increase(ipx_events_missed_total{job="ipx"}[$__rate_interval]))', ref="D", legend="live updates missed")], 16, w=8, desc="Each open page holds one live-update listener. Missed updates mean a page fell behind and reloads its lists.") y += 8 +q95 = lambda m, by="": f'histogram_quantile(0.95, sum by (le{by}) (rate({m}_bucket{{job="ipx"}}[$__rate_interval])))' +ts("Database", [prom(q95("ipx_db_wait_seconds"), legend="wait for a connection, 95th"), + prom(q95("ipx_db_query_seconds", ", kind"), ref="B", legend="{{kind}}, 95th")], + 0, unit="s", desc="A statement's time once it has a connection, and the wait for one (hand-written SQL only, #367). A wait that grows with the queries flat is the pool full.") +ts("The page's timings", [prom(f'histogram_quantile(0.95, sum by (le, ev) (rate(ipx_page_seconds_bucket{{job="ipx"}}[1h])))', legend="{{ev}}, 95th")], + 12, unit="s", desc="What the browser timed (#369): list_shown a click to its list drawn, play_start play to sound, stall a wait mid-play. Over the hour, as there are few.") +y += 8 # ---- ipx: the worker, one job at a time: scans, downloads, DeadAir row("ipx: worker") diff --git a/tests/page-unit.js b/tests/page-unit.js index 479b7dd..84c3481 100644 --- a/tests/page-unit.js +++ b/tests/page-unit.js @@ -228,3 +228,17 @@ test('Signals are everyone\'s on the server\'s key, and yours on your own once a assert.equal(on(null), false); run('S.me = null'); }); + +test('a tap is its trace\'s root span, reported with its ids once it has an answer (#376)', () => { + run(`globalThis.sent = []; fetch = (url, o) => { sent.push([url, o]); return Promise.resolve({ ok: true, status: 204 }); }`); + run(`newTrace('pointerdown'); traceAt = Date.now()`); + const [, trace, parent] = run('traceparent()').split('-'); + assert.equal(parent, run('rootId'), 'requests hang from the tap'); + assert.equal(trace, run('traceId')); + run('endTap()'); + assert.equal(run('sent.length'), 0, 'nothing answered yet, nothing to report'); + run(`newTrace('keydown'); answeredAt = rootAt + 250; newTrace('pointerdown')`); + const body = JSON.parse(run(`sent.at(-1)[1].body`)); + assert.equal(run(`sent.at(-1)[0]`), '/api/client'); + assert.deepEqual([body.ev, body.ms, body.msg, body.span.length, body.trace.length], ['tap', 250, 'keydown', 16, 32]); +}); diff --git a/web/CHANGELOG.md b/web/CHANGELOG.md index 9356965..07e3871 100644 --- a/web/CHANGELOG.md +++ b/web/CHANGELOG.md @@ -21,6 +21,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Caddy's own metrics, its requests and their times and the 502s ipx never sees, for Prometheus on a port of their own, `METRICS_BIND` (#371). +- A tap or a key is its trace's root span: the page sends ipx when it began and when its last answer came, so Tempo shows how long it took beside the requests it made (#376). - Fonts of your choosing in Settings, beside the theme, kept by each browser as the theme is: one for the interface and one for reading an item's title and text, which is the interface's until chosen, with a little more room between the lines in a serif. Inter, your system's, the reading fonts Atkinson Hyperlegible, Lexend, Literata, Source Serif and OpenDyslexic, your device's serif, or monospace. The default keeps each theme's face as it was (#359). ## [1.3.0] - 2026-10-23 diff --git a/web/Caddyfile b/web/Caddyfile index d68cb40..011fafc 100644 --- a/web/Caddyfile +++ b/web/Caddyfile @@ -6,12 +6,19 @@ # OTEL_EXPORTER_OTLP_PROTOCOL, OTEL_SERVICE_NAME), continuing the page's traceparent # and passing it to ipx, so one trace runs browser, Caddy, ipx (#213). Off: none, # and no export failing every few seconds where there is nothing to send to. +# METRICS_BIND where to serve Caddy's metrics for Prometheus (:9180); keep it off the tunnel # TUNNEL_PEER the one address whose Cf-Access-* headers are passed on: where cloudflared's # requests come from. Nobody else's are, so nobody else can claim to be signed in # by Cloudflare. Unset passes none. { admin off auto_https off + # Requests, their times and upstream errors as Prometheus metrics, served below (#371). + # Nested in servers, which Caddy 2.11 calls deprecated, for the browser tests' Caddy, 2.6, + # which knows no other way; the global `metrics` option once both are 2.8 or later. + servers { + metrics + } # Everything else Caddy logs, its errors among them, which carry the request too: the same # filter as the access log's below. log default { @@ -28,6 +35,11 @@ } } +# Caddy's own metrics (#371), on a port of their own: the tunnel reaches only WEB_BIND's. +http://{$METRICS_BIND::9180} { + metrics +} + (tracing-on) { tracing { # Named for the request, as ipx names its own (#286): one name for all left every span in diff --git a/web/src/util.ts b/web/src/util.ts index 53a7323..2ff4cc6 100644 --- a/web/src/util.ts +++ b/web/src/util.ts @@ -103,14 +103,29 @@ function spanText(secs: number){ /// LAN address over plain HTTP does not have. A trace ends after a minute with no request in it: /// the requests a live update or a woken laptop sets off hours later used to join the last tap's /// trace, which then lasted nine hours (#219). A play's position saves, every ten seconds, keep it. -let traceId='', traceAt=0; +/// A tap or a key is the trace's root span (#376): its requests name it as their parent, and once +/// the next one begins, or the page is hidden, the page reports it, from the tap to its last +/// answer, for ipx to send on to Tempo with these ids. +let traceId='', traceAt=0, rootId='', rootAt=0, rootKind='', answeredAt=0; const hex=(n: number)=>Array.from(crypto.getRandomValues(new Uint8Array(n)),b=>b.toString(16).padStart(2,'0')).join(''); -for(const t of ['pointerdown','keydown']) window.addEventListener(t,()=>{ traceId=hex(16); traceAt=Date.now(); },{capture:true,passive:true}); +function endTap(){ + if(rootKind && answeredAt>rootAt) + report('tap',{trace:traceId,span:rootId,start:rootAt,ms:Math.min(answeredAt-rootAt,60_000),msg:rootKind}); + rootKind=''; +} +function newTrace(kind: string){ + endTap(); + traceId=hex(16); rootId=hex(8); rootAt=Date.now(); rootKind=kind; answeredAt=0; +} +for(const t of ['pointerdown','keydown']) window.addEventListener(t,()=>{ newTrace(t); traceAt=Date.now(); },{capture:true,passive:true}); +window.addEventListener('pagehide',endTap); +document.addEventListener('visibilitychange',()=>{ if(document.visibilityState==='hidden') endTap(); }); function traceparent(){ const now=Date.now(); - if(!traceId||now-traceAt>60_000) traceId=hex(16); + // No tap to hang from: a live update's or a timer's requests, a trace with no root of its own. + if(!traceId||now-traceAt>60_000) newTrace(''); traceAt=now; - return `00-${traceId}-${hex(8)}-01`; + return `00-${traceId}-${rootId}-01`; } /// Something that went wrong, or how long something took, in the browser, for the server's log @@ -130,8 +145,11 @@ window.addEventListener('unhandledrejection',e=>report('rejection',{msg:String(e async function api(url: string, opts?: RequestInit): Promise{ // The browser's "Failed to fetch" said nothing anyone could act on (#304). - const r = await fetch(url,{headers:{'Content-Type':'application/json',traceparent:traceparent()},...opts}) + const tp=traceparent(), mine=traceId; + const r = await fetch(url,{headers:{'Content-Type':'application/json',traceparent:tp},...opts}) .catch(()=>{ throw new Error('Unable to reach iPX. Check your connection and try again.'); }); + // The tap lasts until its last answer. + if(mine===traceId) answeredAt=Date.now(); if(r.status===401){ location.href='/login'; throw new Error('signed out'); } if(!r.ok){ // A page in place of ipx's words is Cloudflare or Caddy answering for it: the message became a