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

Signals in /metrics and on the dashboard; failed Jev calls are failed spans (#270)

Browse Source 
The judge had spans but nothing in Prometheus: no count of items judged or
of how calls to Jev ended, nor whether the judge was still passing. The jev
span recorded an error field without the span's status, so Tempo's failed
traces left Jev out, and a timeout recorded nothing. Calls now send
traceparent like every other call out of ipx.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ray Slakinski
2026-10-17 02:55:37 +00:00
parent 3b48f51a56
commit 33eb1f4db2
5 changed files with 135 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.claude/skills/ipx-prod-check/SKILL.md
View File

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

4
crates/ipx/CHANGELOG.md
View File

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

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

@@ -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<Option<(&'static str, std::time::Instant)>>,
/// 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<String> = 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 "));
}

85
crates/ipx/src/signals.rs
View File

@@ -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<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
_ = 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<bool> {
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::<i64, usize>::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<bool> {
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::<usize>());
// 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<Vec<(i64, bool)>> {
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<bool> {
/// 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<String> {
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<T>(span: tracing::Span, call: impl std::future::Future<Output = Result<T>>) -> Result<T> {
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<T>(r: &Result<T>) -> 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#"<img src="x.png" alt="">"#), "");
}
#[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 {

22
grafana/dashboard.py
View File

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