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

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 {