diff --git a/.claude/skills/ipx-prod-check/SKILL.md b/.claude/skills/ipx-prod-check/SKILL.md index d852fc6..5e063a5 100644 --- a/.claude/skills/ipx-prod-check/SKILL.md +++ b/.claude/skills/ipx-prod-check/SKILL.md @@ -109,6 +109,9 @@ from 2026-10-11 15:15 UTC. All gauges: | `ipx_feed_retries` | feeds waiting to retry a failed name lookup | | `ipx_art_cache_bytes`, `ipx_art_cache_limit_bytes` | artwork kept, as of the last sweep, and the most kept: kept sits at the limit by design | | `ipx_art_evicted_total`, `ipx_art_requests_total{result="hit"\|"miss"}` | counters: images the sweep threw out to stay under the limit, and artwork served from disk or fetched from the publisher; many evicted at every sweep with a falling hit rate means the limit is too small | +| `ipx_signals_judged_total`, `ipx_signals_hits_total` | counters: items Jev judged for Signals, and Signals they fit (#254) | +| `ipx_jev_requests_total{result="ok"\|"key_refused"\|"malformed"\|"error"}`, `ipx_jev_seconds_total` | counters: calls to TypeSafe's Jev, Signals' and icons', by how they ended, and the seconds waited; `malformed` is ipx's request at fault (#257) | +| `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) | | `ipx_feeds`, `ipx_downloads_pending`, `ipx_files_downloaded` | the library, as the healthcheck last counted it (every 30 s) | @@ -194,6 +197,8 @@ a count says something happened, the lines and traces say why. $Q prom 'max(max_over_time(ipx_http_requests_in_flight{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 + $Q prom 'max(max_over_time(ipx_signals_last_pass_age_seconds{job="ipx"}[$range]))' 24h ``` `up` 0 means Prometheus cannot reach the port: the container is down or `metrics_bind` is unset. The daemon keeps all 20 connections open on Postgres (#241), so `idle` is the pool's 20 less @@ -210,6 +215,8 @@ a count says something happened, the lines and traces say why. 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 counts until reloaded. DeadAir over a minute on one file is a file ffmpeg chokes on. + Any `malformed` Jev call is ipx asking wrongly (#257); `key_refused` is someone's key; the Signals + judge over 10 minutes since a pass is stuck on Jev: its `signals` trace shows which `jev` call. 7. **What the page saw.** The browser's own errors and timings, which no server line shows. ``` diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index dc8132b..313bb67 100644 --- a/crates/ipx/CHANGELOG.md +++ b/crates/ipx/CHANGELOG.md @@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `/metrics` counts Signals: items judged and fitting, calls to Jev by how they ended and the time they took, time since the judge's last pass, and keys refused; a failed call to Jev shows as a failed trace (#270). + ## [1.3.0] - 2026-10-16 ### Added diff --git a/crates/ipx/src/metrics.rs b/crates/ipx/src/metrics.rs index 9697f13..4486436 100644 --- a/crates/ipx/src/metrics.rs +++ b/crates/ipx/src/metrics.rs @@ -40,6 +40,14 @@ pub struct Stats { pub feeds: AtomicI64, pub pending: AtomicI64, pub downloaded: AtomicI64, + /// Items Jev has judged for Signals, and Signals they fit (#254). + pub signals_judged: AtomicU64, + pub signals_hits: AtomicU64, + /// When the Signals judge last finished a pass, Unix seconds; 0 for not since starting. + pub signals_last_pass: AtomicI64, + /// Calls to Jev by how they ended, in `JEV_RESULTS`' order, and their microseconds all told. + pub jev_calls: [AtomicU64; 4], + pub jev_micros: AtomicU64, /// The worker's job now, and since when. job: std::sync::Mutex>, /// Jobs finished, by kind: how many, and their seconds all told. @@ -64,6 +72,11 @@ impl Stats { feeds: AtomicI64::new(-1), pending: AtomicI64::new(-1), downloaded: AtomicI64::new(-1), + signals_judged: AtomicU64::new(0), + signals_hits: AtomicU64::new(0), + signals_last_pass: AtomicI64::new(0), + jev_calls: [const { AtomicU64::new(0) }; 4], + jev_micros: AtomicU64::new(0), job: std::sync::Mutex::new(None), jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()), } @@ -90,6 +103,10 @@ impl Drop for Job { } } +/// How a call to Jev ended: answered, its key refused (401-403), the request refused as +/// malformed (400, 422), or anything else, TypeSafe's 5xx and timeouts. +pub const JEV_RESULTS: [&str; 4] = ["ok", "key_refused", "malformed", "error"]; + /// Counts one more of something in progress until dropped, however it ends. pub struct Busy(&'static AtomicI64); @@ -167,6 +184,8 @@ pub fn render(s: &Sources) -> String { gauge(&mut out, "ipx_deadair_playing", "Episodes someone is playing through DeadAir's playlist.", &[("", s.playing.lock().map_or(0, |p| p.len()) as f64)]); work(&mut out, &STATS, s.events.len(), s.ctx.retry.lock().map_or(0, |r| r.len()), s.ctx.cfg().general.art_cache_mb * 1_048_576); + gauge(&mut out, "ipx_signals_keys_refused", "People whose TypeSafe key was refused, whose Signals wait until they save another.", + &[("", s.ctx.signals_refused.lock().map_or(0, |r| r.len()) as f64)]); let rt = s.runtime.metrics(); gauge(&mut out, "ipx_runtime_workers", "The async runtime's worker threads.", &[("", rt.num_workers() as f64)]); gauge(&mut out, "ipx_runtime_tasks", "Tasks alive in the async runtime.", &[("", rt.num_alive_tasks() as f64)]); @@ -221,6 +240,16 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li counter(out, "ipx_art_evicted_total", "Images thrown out to keep the artwork under its limit.", &[("", st.art_evicted.load(Relaxed) as f64)]); counter(out, "ipx_art_requests_total", "Artwork asked for: served from disk, or fetched from the publisher.", &[("result=\"hit\"", st.art_hits.load(Relaxed) as f64), ("result=\"miss\"", st.art_misses.load(Relaxed) as f64)]); + counter(out, "ipx_signals_judged_total", "Items Jev has judged for someone's Signals.", &[("", st.signals_judged.load(Relaxed) as f64)]); + counter(out, "ipx_signals_hits_total", "Signals an item was found to fit.", &[("", st.signals_hits.load(Relaxed) as f64)]); + let last = st.signals_last_pass.load(Relaxed); + if last > 0 { + gauge(out, "ipx_signals_last_pass_age_seconds", "Seconds since the Signals judge last finished a pass.", &[("", (now - last) as f64)]); + } + let labels: Vec = JEV_RESULTS.iter().map(|r| format!("result=\"{r}\"")).collect(); + let calls: Vec<(&str, f64)> = labels.iter().zip(&st.jev_calls).map(|(l, c)| (l.as_str(), c.load(Relaxed) as f64)).collect(); + counter(out, "ipx_jev_requests_total", "Calls to TypeSafe's Jev, Signals' and their icons', by how they ended.", &calls); + counter(out, "ipx_jev_seconds_total", "Seconds spent waiting on Jev.", &[("", st.jev_micros.load(Relaxed) as f64 / 1e6)]); for (name, help, v) in [ ("ipx_feeds", "Feeds in the catalogue, as last counted.", &st.feeds), ("ipx_downloads_pending", "Files a scan will download on its own, as last counted.", &st.pending), @@ -287,6 +316,7 @@ mod tests { assert!(out.contains("ipx_worker_busy 0\n") && !out.contains("ipx_worker_job_seconds"), "{out}"); assert!(out.contains("ipx_jobs_total{kind=\"scan\"} 1\n") && out.contains("# TYPE ipx_jobs_total counter"), "{out}"); assert!(out.contains("ipx_event_queue 3\n") && out.contains("ipx_feed_retries 1\n")); + assert!(out.contains("ipx_jev_requests_total{result=\"key_refused\"} 0\n") && !out.contains("ipx_signals_last_pass_age_seconds"), "{out}"); // The library's counts are left out until the daemon has counted it. assert!(!out.contains("ipx_feeds ")); } diff --git a/crates/ipx/src/signals.rs b/crates/ipx/src/signals.rs index 345dd45..1364f3e 100644 --- a/crates/ipx/src/signals.rs +++ b/crates/ipx/src/signals.rs @@ -9,9 +9,12 @@ use std::sync::Arc; use anyhow::Result; use tracing::Instrument; +use std::sync::atomic::Ordering::Relaxed; + use crate::Ctx; -use crate::db::{Ask, Unjudged}; +use crate::db::{Ask, Unjudged, step}; use crate::feed; +use crate::metrics::STATS; /// Items put to Jev a pass. A pass ends early once there are none left, so this only says how /// long the queue is held before people who got a key later are looked at. @@ -29,6 +32,7 @@ pub async fn run(ctx: Arc, mut stop: tokio::sync::watch::Receiver) { _ = stop.changed() => return, done = pass(&ctx) => done, }; + crate::metrics::STATS.signals_last_pass.store(crate::db::now(), Relaxed); match done { Ok(true) => continue, Ok(false) => {} @@ -49,23 +53,27 @@ async fn pass(ctx: &Ctx) -> Result { if items.is_empty() { return Ok(false); } - let span = tracing::info_span!("signals", entries = items.len()); + let span = tracing::info_span!("signals", entries = items.len(), judged = tracing::field::Empty, hits = tracing::field::Empty); async { let mut refused = HashSet::new(); let mut hits_for = std::collections::BTreeMap::::new(); let mut stopped = false; + let mut judged = 0; for item in &items { if refused.contains(&item.user_id) { continue; } match judge(ctx, item).await { Ok(verdicts) => { + judged += 1; let hits = verdicts.iter().filter(|(_, h)| *h).count(); + STATS.signals_judged.fetch_add(1, Relaxed); + STATS.signals_hits.fetch_add(hits as u64, Relaxed); if hits > 0 { *hits_for.entry(item.user_id).or_default() += hits; tracing::info!(ev = "signal", user = item.user_id, feed = %item.feed_id, new = hits, "an item fits a Signal"); } - ctx.db.record_verdicts(&item.feed_id, &item.guid, &verdicts).await?; + ctx.db.record_verdicts(&item.feed_id, &item.guid, &verdicts).instrument(step("record_verdicts")).await?; } // A key TypeSafe will not take is not sent the rest of the queue. It is tried // again once its owner saves one (`web::patch_me`) or the daemon restarts. @@ -84,6 +92,9 @@ async fn pass(ctx: &Ctx) -> Result { Err(e) => tracing::warn!(user = item.user_id, feed = %item.feed_id, error = %format!("{e:#}"), "Jev could not judge an item"), } } + let span = tracing::Span::current(); + span.record("judged", judged); + span.record("hits", hits_for.values().sum::()); // Once a pass, not once an item: a page fetches its counts again on each. for (user, hits) in hits_for { ctx.out.emit(crate::ipc::Event::SignalHits { user, hits }); @@ -141,19 +152,18 @@ async fn judge(ctx: &Ctx, item: &Unjudged) -> Result> { signal_count = item.signals.len(), "http.response.status_code" = tracing::field::Empty, error = tracing::field::Empty, + "otel.status_code" = tracing::field::Empty, ); - let answer: serde_json::Value = async { - let r = ctx.client.post(url).bearer_auth(&item.key).json(&request(item)).timeout(feed::FEED_TIMEOUT).send().await?; + let answer: serde_json::Value = called(span.clone(), async { + let r = jev_post(ctx, url, &item.key).json(&request(item)).send().await?; let status = r.status().as_u16(); tracing::Span::current().record("http.response.status_code", status); if !r.status().is_success() { let body: String = r.text().await.unwrap_or_default().chars().take(500).collect(); - tracing::Span::current().record("error", body.as_str()); return Err(Refused { status, body }.into()); } anyhow::Ok(r.json().await?) - } - .instrument(span) + }) .await?; // A question Jev left unanswered stays unjudged and is asked again. Ok(item.signals.iter().filter_map(|a| hit(&answer, a.id).map(|h| (a.id, h))).collect()) @@ -208,9 +218,15 @@ fn hit(answer: &serde_json::Value, id: i64) -> Option { /// server's page doing it, once a Signal. pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], description: Option<&str>, candidates: &[(String, String)]) -> Result { let url = std::env::var("IPX_TYPESAFE_URL").unwrap_or_else(|_| feed::JEV_URL.to_owned()); - let span = tracing::info_span!("jev_icon", candidates = candidates.len(), "http.response.status_code" = tracing::field::Empty); - async { - let r = ctx.client.post(url).bearer_auth(key).json(&icon_request(name, words, description, candidates)).timeout(feed::FEED_TIMEOUT).send().await?; + let span = tracing::info_span!( + "jev_icon", + candidates = candidates.len(), + "http.response.status_code" = tracing::field::Empty, + error = tracing::field::Empty, + "otel.status_code" = tracing::field::Empty, + ); + called(span, async { + let r = jev_post(ctx, url, key).json(&icon_request(name, words, description, candidates)).send().await?; let status = r.status().as_u16(); tracing::Span::current().record("http.response.status_code", status); if !r.status().is_success() { @@ -219,11 +235,43 @@ pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], descr } let answer: serde_json::Value = r.json().await?; chosen(&answer, candidates).ok_or_else(|| anyhow::anyhow!("Jev chose no icon offered: {answer}")) - } - .instrument(span) + }) .await } +/// A request to Jev, carrying this trace's `traceparent` as every call out of ipx does. +fn jev_post(ctx: &Ctx, url: String, key: &str) -> reqwest::RequestBuilder { + let req = ctx.client.post(url).bearer_auth(key).timeout(feed::FEED_TIMEOUT); + match ipx_directory::otel::traceparent() { + Some(tp) => req.header("traceparent", tp), + None => req, + } +} + +/// Runs a call to Jev in `span`, counting it in /metrics by how it ended, and marking the span +/// failed, which a field named `error` alone does not: Tempo's failed traces left Jev's out. +async fn called(span: tracing::Span, call: impl std::future::Future>) -> Result { + let started = std::time::Instant::now(); + let r = call.instrument(span.clone()).await; + STATS.jev_micros.fetch_add(started.elapsed().as_micros() as u64, Relaxed); + STATS.jev_calls[outcome(&r)].fetch_add(1, Relaxed); + if let Err(e) = &r { + span.record("error", format!("{e:#}").as_str()); + span.record("otel.status_code", "ERROR"); + } + r +} + +/// Which of `metrics::JEV_RESULTS` a call's result is. +fn outcome(r: &Result) -> usize { + match r { + Ok(_) => 0, + Err(e) if refuses_key(e) => 1, + Err(e) if malformed(e) => 2, + Err(_) => 3, + } +} + fn icon_request(name: &str, words: &[String], description: Option<&str>, candidates: &[(String, String)]) -> serde_json::Value { let mut state = format!("Signal: {name}\nListens for: {}", words.join(", ")); if let Some(d) = description { @@ -327,6 +375,17 @@ mod tests { assert_eq!(text_of(r#""#), ""); } + #[test] + fn a_call_is_counted_by_how_it_ended() { + let refused = |status| -> Result<()> { Err(Refused { status, body: String::new() }.into()) }; + let named = |r: Result<()>| crate::metrics::JEV_RESULTS[outcome(&r)]; + assert_eq!(named(Ok(())), "ok"); + assert_eq!(named(refused(402)), "key_refused"); + assert_eq!(named(refused(422)), "malformed"); + assert_eq!(named(refused(503)), "error"); + assert_eq!(named(Err(anyhow::anyhow!("timed out"))), "error"); + } + #[test] fn a_batch_is_full_by_its_rows_not_its_items() { let item = |n: usize| Unjudged { diff --git a/grafana/dashboard.py b/grafana/dashboard.py index e7ac53c..1d64b9f 100644 --- a/grafana/dashboard.py +++ b/grafana/dashboard.py @@ -254,6 +254,28 @@ table("Failed downloads", [loki(f'sum by (feed, msg) (count_over_time({EV} | ev= 12, rename={"feed": "Feed", "msg": "Error", "Value": "Failures"}, sort=[{"displayName": "Failures", "desc": True}]) y += 8 +# ---- ipx: Signals, items put to TypeSafe's Jev on each person's own key (#254) +row("ipx: Signals") +ts("Items judged", [prom('sum(increase(ipx_signals_judged_total{job="ipx"}[$__rate_interval]))', legend="judged"), + prom('sum(increase(ipx_signals_hits_total{job="ipx"}[$__rate_interval]))', ref="B", legend="fit a Signal")], + 0, w=8, bars=True, desc="Each item is judged once against every Signal of its subscribers that has not had it; a burst follows a scan or a new Signal.") +ts("Calls to Jev", [prom('sum by (result) (increase(ipx_jev_requests_total{job="ipx"}[$__rate_interval]))', legend="{{result}}"), + prom('sum(increase(ipx_jev_seconds_total{job="ipx"}[$__rate_interval])) / ' + 'sum(increase(ipx_jev_requests_total{job="ipx"}[$__rate_interval]))', ref="B", legend="seconds a call")], + 8, w=8, bars=True, stack=True, + desc="key_refused stops that person's Signals until they save another key; malformed stops the pass and is ipx's to fix; " + "error is TypeSafe failing or slow, and the item is asked again next pass.")["fieldConfig"]["overrides"] = [ + {"matcher": {"id": "byName", "options": "seconds a call"}, + "properties": [{"id": "unit", "value": "s"}, {"id": "custom.axisPlacement", "value": "right"}, + {"id": "custom.drawStyle", "value": "line"}, {"id": "custom.stacking", "value": {"mode": "none"}}]}] +ts("Signals judge", [prom('max(ipx_signals_last_pass_age_seconds{job="ipx"})', legend="since the last pass"), + prom('max(ipx_signals_keys_refused{job="ipx"})', ref="B", legend="keys refused")], + 16, w=8, unit="s", desc="It passes at least every five minutes; much longer means it is stuck on Jev. Keys refused, on the right, " + "are people whose Signals wait for a new key.")["fieldConfig"]["overrides"] = [ + {"matcher": {"id": "byName", "options": "keys refused"}, + "properties": [{"id": "unit", "value": "short"}, {"id": "custom.axisPlacement", "value": "right"}]}] +y += 8 + # ---- ipx: the process, what it holds between requests and jobs (#194) row("ipx: process") # Open connections stay open, idle, for ten minutes after a burst, so open at the max is not