rays/ipx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 20 Wiki Activity

Metrics and traces for what had none, and alerts on them (#365-#377)

Browse Source 
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:
Ray Slakinski
2026-10-25 16:18:06 +00:00
parent c788112190
commit dd91602a49
23 changed files with 768 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/directory/CHANGELOG.md
View File

@@ -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).

19
crates/directory/src/api.rs
View File

@@ -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"

1
crates/directory/src/apple.rs
View File

@@ -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,

2
crates/directory/src/directory.rs
View File

@@ -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,
)

15
crates/directory/src/main.rs
View File

@@ -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())
}))

114
crates/directory/src/metrics.rs
View File

@@ -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());

84
crates/directory/src/otel.rs
View File

@@ -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);
}
}

6
crates/ipx/CHANGELOG.md
View File

@@ -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).

92
crates/ipx/src/daemon.rs
View File

@@ -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<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());
}
// 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 <password> | 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 <password> | 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<String> = 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<String> = 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::<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;
}
}

40
crates/ipx/src/db.rs
View File

@@ -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

1
crates/ipx/src/ipc.rs
View File

@@ -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"));

120
crates/ipx/src/lists.rs
View File

@@ -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,74 +123,94 @@ 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();
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() })
}

37
crates/ipx/src/logging.rs
View File

@@ -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.

19
crates/ipx/src/main.rs
View File

@@ -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
}

126
crates/ipx/src/metrics.rs
View File

@@ -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}");
}

71
crates/ipx/src/web.rs
View File

@@ -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