Metrics and traces for what had none, and alerts on them (#365-#377)
Reviewing what production's traces and metrics leave out turned up twelve gaps; each is an issue. Metrics. A Prometheus histogram for ipx and the directory both (ipx_directory::metrics::Histograms), which ipx's request histogram (#364) now uses too. - ipx: each statement's time once it has a connection, by its first word, from SeaORM's metric callback; and the wait for one, for hand-written SQL in db.rs, measured as the call less the statement (the callback runs in the poll that returns the result, so the time read after the await is that statement's) (#367). SeaORM's own queries are timed but their wait is not seen: a shortcut, noted. - ipx: counters of what the events say, counted in emit, which every event passes once: feed failures by error.type, new items, downloads saved and failed, DeadAir's seconds found (#368). - ipx: the page's list_shown, play_start and stall timings (#369). - directory: requests by method, route (a route layer puts the matched route on the response, or each feed's page was a row) and status; feeds checked and failed; snapshots taken (#370). - ipx-web: Caddy's own metrics on METRICS_BIND (:9180), served apart from the port the tunnel reaches. `servers { metrics }`, which Caddy 2.11 calls deprecated, because the browser tests' Caddy is 2.6 (#371). Traces. - A list's sync is list_sync with steps, an OPML upload opml_import (#372). Doing so showed sync_group reading every account and its subscription to the list once per listed feed; read once now (#377). - Each command the worker runs, from the socket, the CLI or the page, is a command span with its job inside (#373). - The daemon's start is a trace, from opening the database through the catalogue's repairs to "daemon started" (#374). - The directory's lines carry trace_id and span_id: WithTrace moves to ipx_directory::otel and both use it (#375). - A tap or key is its trace's root span (#376). The page names a root span id in each request's traceparent, where the parent was random and pointed at nothing, and reports the tap (ids, start, its last answer) when the next begins or the page is hidden. opentelemetry 0.33 gives a span builder no way to take ids, so the shared tracer provider has an id generator that takes given ids inside with_ids, on one thread. Alerts. grafana/alerts.py, Grafana alerting provisioning beside the dashboard (#365): ipx, the directory or ipx-web down, scans stopped, a job over 30 minutes, the pool full, 5xx, 502s, Jev refused, Signals stopped, the directory's refresher stopped, Tower's load and memory, the monitoring down, Tempo not reading its blocks, iPX's lines not reaching Loki. No contact point yet: firing shows in Grafana. Prometheus now scrapes Tempo, Loki, Alloy and itself (#366), in the monitoring project's config, not this repository's. The dashboard has database and page-timing panels; the prod check knows the new metrics, spans and alerts. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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<anyhow::Error> 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::<Who>().map_or("", |w| w.0.as_str()).to_owned();
|
||||
let status = resp.status().as_u16();
|
||||
if path != "/healthz" {
|
||||
let route = resp.extensions().get::<axum::extract::MatchedPath>().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"
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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()
|
||||
.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)
|
||||
// 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)
|
||||
.with_span_list(false),
|
||||
))
|
||||
.with_writer(std::io::stderr)
|
||||
.with_filter(filter())
|
||||
}))
|
||||
|
||||
@@ -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<f64>, Option<f64>) {
|
||||
(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<BTreeMap<String, Row>>,
|
||||
}
|
||||
|
||||
/// Under each bound, all told, and the sum.
|
||||
type Row = (Vec<u64>, 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());
|
||||
|
||||
@@ -17,6 +17,7 @@ pub fn provider(service: &'static str) -> anyhow::Result<Option<opentelemetry_sd
|
||||
opentelemetry_sdk::trace::SdkTracerProvider::builder()
|
||||
.with_batch_exporter(exporter)
|
||||
.with_resource(resource)
|
||||
.with_id_generator(Ids::default())
|
||||
.build(),
|
||||
))
|
||||
}
|
||||
@@ -58,6 +59,72 @@ pub fn traceparent() -> Option<String> {
|
||||
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<Option<(opentelemetry::trace::TraceId, opentelemetry::trace::SpanId)>> =
|
||||
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<R>(
|
||||
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<F>(pub F);
|
||||
|
||||
impl<S, N, F> tracing_subscriber::fmt::FormatEvent<S, N> for WithTrace<F>
|
||||
where
|
||||
F: tracing_subscriber::fmt::FormatEvent<S, N>,
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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).
|
||||
|
||||
|
||||
@@ -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,14 +50,18 @@ pub(crate) async fn daemon(
|
||||
config_path: PathBuf,
|
||||
web_addr: Option<String>,
|
||||
events: broadcast::Sender<Event>,
|
||||
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());
|
||||
}
|
||||
|
||||
// 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().await?.is_empty() {
|
||||
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?;
|
||||
@@ -66,44 +71,48 @@ pub(crate) async fn daemon(
|
||||
}
|
||||
|
||||
// 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 {
|
||||
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<String> = ctx.cfg().feeds.keys().cloned().collect();
|
||||
match ctx.db.adopt_catalogue(admin.id, &catalogue).await {
|
||||
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().await {
|
||||
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).await {
|
||||
match catalogue::retire_stranded(&ctx).instrument(db::step("retire_stranded")).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 {
|
||||
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::<Cmd>(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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -275,7 +275,34 @@ async fn connect(location: &str, keep_open: bool) -> Result<DatabaseConnection>
|
||||
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<Option<std::time::Duration>> = 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<sea_orm::Value>) -> Result<u64> {
|
||||
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<sea_orm::Value>) -> Result<Vec<sea_orm::QueryResult>> {
|
||||
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
|
||||
|
||||
@@ -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"));
|
||||
|
||||
@@ -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<Vec<(String, String)>> {
|
||||
///
|
||||
/// 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<Ctx>, parent_id: &str, listed: &[(String, String)]) -> Result<Outcome> {
|
||||
// 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,13 +123,14 @@ pub(crate) async fn sync_group(ctx: &Arc<Ctx>, 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();
|
||||
|
||||
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) {
|
||||
@@ -145,9 +150,21 @@ pub(crate) async fn sync_group(ctx: &Arc<Ctx>, parent_id: &str, listed: &[(Strin
|
||||
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.
|
||||
// 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() {
|
||||
readers.push(user.id);
|
||||
}
|
||||
}
|
||||
let managed = ctx.db.managed_feeds().await?;
|
||||
let list_and_feeds = managed
|
||||
.iter()
|
||||
@@ -155,17 +172,20 @@ pub(crate) async fn sync_group(ctx: &Arc<Ctx>, parent_id: &str, listed: &[(Strin
|
||||
.map(|m| m.id.clone())
|
||||
.chain(std::iter::once(parent_id.to_string()));
|
||||
for id in list_and_feeds {
|
||||
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?;
|
||||
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();
|
||||
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) {
|
||||
@@ -187,6 +207,10 @@ pub(crate) async fn sync_group(ctx: &Arc<Ctx>, parent_id: &str, listed: &[(Strin
|
||||
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() })
|
||||
}
|
||||
|
||||
@@ -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>(F);
|
||||
|
||||
impl<S, N, F> FormatEvent<S, N> for WithTrace<F>
|
||||
where
|
||||
F: FormatEvent<S, N>,
|
||||
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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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<Option<(&'static str, Instant)>>,
|
||||
/// Jobs finished, by kind: how many, and their seconds all told.
|
||||
jobs: std::sync::Mutex<BTreeMap<&'static str, (u64, f64)>>,
|
||||
/// 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<BTreeMap<(&'static str, String, u16), Requests>>,
|
||||
/// Feed failures by kind, `error.type` (#368).
|
||||
feed_failures: std::sync::Mutex<BTreeMap<String, u64>>,
|
||||
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<String> = 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::<Vec<_>>(),
|
||||
);
|
||||
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}");
|
||||
}
|
||||
|
||||
@@ -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<u64>,
|
||||
/// The list's filter, for `list_shown`: all, unread, flagged.
|
||||
filter: Option<String>,
|
||||
/// For `tap`: the trace the page began and its root span, hex, and when it began, Unix ms.
|
||||
trace: Option<String>,
|
||||
span: Option<String>,
|
||||
start: Option<u64>,
|
||||
}
|
||||
|
||||
async fn client_event(user: crate::db::User, Json(e): Json<ClientEvent>) -> StatusCode {
|
||||
@@ -2763,6 +2792,16 @@ async fn client_event(user: crate::db::User, Json(e): Json<ClientEvent>) -> 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<ClientEvent>) -> 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
|
||||
|
||||
88
grafana/alerts.py
Normal file
88
grafana/alerts.py
Normal file
@@ -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))
|
||||
@@ -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")
|
||||
|
||||
@@ -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]);
|
||||
});
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<any>{
|
||||
// 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
|
||||
|
||||
Reference in New Issue
Block a user