diff --git a/.claude/skills/ipx-prod-check/SKILL.md b/.claude/skills/ipx-prod-check/SKILL.md index 5395811..e6d7bc0 100644 --- a/.claude/skills/ipx-prod-check/SKILL.md +++ b/.claude/skills/ipx-prod-check/SKILL.md @@ -109,9 +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_signals_judged_total{user,key}`, `ipx_signals_hits_total{user,key}` | counters: items Jev judged for Signals, and Signals they fit, by person and whose key paid (`server` or `own`, #278) (#254) | | `ipx_jev_requests_total{result="ok"\|"key_refused"\|"malformed"\|"rate_limited"\|"error"}`, `ipx_jev_seconds_total`, `ipx_jev_paced_seconds_total` | counters: calls to TypeSafe's Jev, Signals' and icons', by how they ended; their seconds, the wait for a turn included; and that wait alone. `malformed` is ipx's request at fault (#257); `rate_limited` is a 429 from TypeSafe, whose limits are 80 calls and 100,000 input tokens a second, which ipx paces itself under at 60 and 80,000 (#277) | -| `ipx_jev_tokens_total{kind="input"\|"output"}` | counter: tokens TypeSafe counted, as its answers say: what Signals cost (#275) | +| `ipx_jev_tokens_total{kind="input"\|"output",user,key}` | counter: tokens TypeSafe counted, as its answers say, by person and whose key paid: what Signals, icon picks and key checks cost (#275, #278). A call shared on the server's key is split by each person's share of its questions | | `ipx_signals_last_pass_age_seconds`, `ipx_signals_keys_refused` | since the Signals judge last finished a pass (at most 5 minutes when well; absent until one has); people whose key TypeSafe refused | | `directory_process_resident_bytes`, `directory_process_open_fds`, `directory_feed_fetches_total`, `directory_feed_bytes_total`, `directory_uptime_seconds`, `directory_build_info` | the directory's, `job="ipx-directory"`, from its `metrics_bind` (production `192.168.1.130:9188`, LAN only) | | `ipx_feeds`, `ipx_downloads_pending`, `ipx_files_downloaded` | the library, as the healthcheck last counted it (every 30 s) | diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 20cf009..589eb90 100644 --- a/crates/ipx/CHANGELOG.md +++ b/crates/ipx/CHANGELOG.md @@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `/metrics` counts Signals: items judged and fitting, calls to Jev by how they ended, the time they took and the tokens TypeSafe counted, time since the judge's last pass, and keys refused; a failed call to Jev shows as a failed trace (#270, #275). - An admin setting for how many days back a new Signal reads (#276). +- `/metrics` breaks Signals' items, fits and TypeSafe tokens out by person and by whose key paid, so the server's bill can be split (#278). - An admin setting, Signals for everyone on the server's TypeSafe key, so the server pays for Jev instead of each person (#271). ### Changed diff --git a/crates/ipx/src/db.rs b/crates/ipx/src/db.rs index 955b03e..03cb569 100644 --- a/crates/ipx/src/db.rs +++ b/crates/ipx/src/db.rs @@ -2596,6 +2596,8 @@ pub struct Ask { #[derive(Debug)] pub struct Unjudged { pub user_id: i64, + /// Their account's name, what /metrics bills by (#278). + pub user_name: String, /// Their own key; None when the server's is used for everyone. pub key: Option, pub feed_id: String, @@ -2743,7 +2745,7 @@ impl Db { // rows of signal_verdicts. The same placeholders serve both. let half = |items: &str| { format!( - "SELECT g.id AS signal_id, g.name, g.keywords, g.description AS signal_description, g.user_id, u.typesafe_key, + "SELECT g.id AS signal_id, g.name, g.keywords, g.description AS signal_description, g.user_id, u.name AS user_name, u.typesafe_key, e.feed_id, e.guid, e.title, e.description, f.title AS feed_title, coalesce(e.published, e.first_seen) AS at FROM signals g JOIN users u ON u.id = g.user_id @@ -2781,6 +2783,7 @@ impl Db { Some(u) if (u.user_id, &u.feed_id, &u.guid) == (user_id, &feed_id, &guid) => u.signals.push(signal), _ => out.push(Unjudged { user_id, + user_name: r.try_get("", "user_name")?, key: r.try_get("", "typesafe_key")?, feed_id, guid, diff --git a/crates/ipx/src/metrics.rs b/crates/ipx/src/metrics.rs index c4567bb..ba5e5bc 100644 --- a/crates/ipx/src/metrics.rs +++ b/crates/ipx/src/metrics.rs @@ -40,9 +40,6 @@ 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. @@ -50,8 +47,10 @@ pub struct Stats { pub jev_micros: AtomicU64, /// Microseconds calls waited for their turn under TypeSafe's limits (#277), part of `jev_micros`. pub jev_paced_micros: AtomicU64, - /// Tokens TypeSafe counted, input and output, as its answers say. - pub jev_tokens: [AtomicU64; 2], + /// What each person's Signals, icons and key checks used, and whose key paid (#278): items + /// judged, Signals they fit, and TypeSafe's input and output tokens. Floats: a call shared by + /// several people on the server's key is split between them. + people: std::sync::Mutex>, /// The worker's job now, and since when. job: std::sync::Mutex>, /// Jobs finished, by kind: how many, and their seconds all told. @@ -76,18 +75,26 @@ 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) }; 5], jev_micros: AtomicU64::new(0), jev_paced_micros: AtomicU64::new(0), - jev_tokens: [const { AtomicU64::new(0) }; 2], + people: std::sync::Mutex::new(std::collections::BTreeMap::new()), job: std::sync::Mutex::new(None), jobs: std::sync::Mutex::new(std::collections::BTreeMap::new()), } } + /// Adds to what `who` used on `key` ("server" or "own"): items judged, fits, input and output + /// tokens. + pub fn bill(&self, who: &str, key: &'static str, used: [f64; 4]) { + let mut people = self.people.lock().unwrap(); + let row = people.entry((who.to_owned(), key)).or_default(); + for (r, u) in row.iter_mut().zip(used) { + *r += u; + } + } + /// Marks the worker as on a job of `kind` until what it returns is dropped, however the job /// ends: finished, failed or stopped. pub fn job(&'static self, kind: &'static str) -> Job { @@ -247,8 +254,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)]); + // By person and whose key paid (#278). An account's name is often an email address: /metrics + // is on the LAN only, as Prometheus is. + let people = st.people.lock().unwrap().clone(); + let labels: Vec = people.keys().map(|(who, key)| format!("user=\"{}\",key=\"{key}\"", label(who))).collect(); + let by = |n: usize| labels.iter().zip(people.values()).map(|(l, v)| (l.as_str(), v[n])).collect::>(); + counter(out, "ipx_signals_judged_total", "Items Jev has judged for someone's Signals, by person and whose key paid.", &by(0)); + counter(out, "ipx_signals_hits_total", "Signals an item was found to fit, by person and whose key paid.", &by(1)); + let tokens: Vec = labels.iter().flat_map(|l| [format!("kind=\"input\",{l}"), format!("kind=\"output\",{l}")]).collect(); + let used: Vec<(&str, f64)> = tokens.iter().zip(people.values().flat_map(|v| [v[2], v[3]])).map(|(l, v)| (l.as_str(), v)).collect(); + counter(out, "ipx_jev_tokens_total", "Tokens TypeSafe counted for calls to Jev, what they cost, by person and whose key paid.", &used); 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)]); @@ -259,8 +274,6 @@ fn work(out: &mut String, st: &Stats, event_queue: usize, retries: usize, art_li counter(out, "ipx_jev_seconds_total", "Seconds spent on calls to Jev, their wait for a turn included.", &[("", st.jev_micros.load(Relaxed) as f64 / 1e6)]); counter(out, "ipx_jev_paced_seconds_total", "Seconds calls to Jev waited for their turn under TypeSafe's rate limits.", &[("", st.jev_paced_micros.load(Relaxed) as f64 / 1e6)]); - counter(out, "ipx_jev_tokens_total", "Tokens TypeSafe counted for calls to Jev, what they cost.", - &[("kind=\"input\"", st.jev_tokens[0].load(Relaxed) as f64), ("kind=\"output\"", st.jev_tokens[1].load(Relaxed) as f64)]); 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), @@ -273,6 +286,11 @@ 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); @@ -328,6 +346,14 @@ mod tests { 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}"); + T.bill("sam@example.net", "server", [1.0, 1.0, 600.0, 50.0]); + T.bill("sam@example.net", "server", [1.0, 0.0, 300.5, 25.0]); + T.bill("a \"b\"", "own", [0.0, 0.0, 10.0, 1.0]); + let mut out = String::new(); + work(&mut out, &T, 0, 0, 0); + assert!(out.contains("ipx_signals_judged_total{user=\"sam@example.net\",key=\"server\"} 2\n"), "{out}"); + assert!(out.contains("ipx_jev_tokens_total{kind=\"input\",user=\"sam@example.net\",key=\"server\"} 900.5\n"), "{out}"); + assert!(out.contains("ipx_jev_tokens_total{kind=\"output\",user=\"a \\\"b\\\"\",key=\"own\"} 1\n"), "{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 49aeb63..32a8e1f 100644 --- a/crates/ipx/src/signals.rs +++ b/crates/ipx/src/signals.rs @@ -88,6 +88,7 @@ async fn pass(ctx: &Ctx) -> Result { let span = tracing::info_span!("signals", entries = items.len(), calls = tracing::field::Empty, judged = tracing::field::Empty, hits = tracing::field::Empty); async { let groups = calls(&items, server.is_some()); + let paid = if server.is_some() { "server" } else { "own" }; tracing::Span::current().record("calls", groups.len()); let refused = std::sync::Mutex::new(HashSet::new()); let stopped = AtomicBool::new(false); @@ -106,7 +107,7 @@ async fn pass(ctx: &Ctx) -> Result { return None; } let key = server.as_deref().or(first.key.as_deref())?; - Some((group, judge(ctx, group, key).await)) + Some((group, judge(ctx, group, key, paid).await)) } }) .buffer_unordered(IN_FLIGHT); @@ -118,8 +119,7 @@ async fn pass(ctx: &Ctx) -> Result { for (item, verdicts) in group.iter().zip(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); + STATS.bill(&item.user_name, paid, [1.0, hits as f64, 0.0, 0.0]); 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"); @@ -263,7 +263,7 @@ fn malformed(e: &anyhow::Error) -> bool { } /// One call about one item, for everyone in `group`: their verdicts, in the group's order. -async fn judge(ctx: &Ctx, group: &[&Unjudged], key: &str) -> Result>> { +async fn judge(ctx: &Ctx, group: &[&Unjudged], key: &str, paid: &'static str) -> Result>> { let asks: Vec<&Ask> = group.iter().flat_map(|i| &i.signals).collect(); let span = tracing::info_span!( "jev", @@ -278,6 +278,12 @@ async fn judge(ctx: &Ctx, group: &[&Unjudged], key: &str) -> Result Option { /// chooses among options and cannot name one, so the page searches Font Awesome, whose index is /// ipx-web's, and sends the likeliest. On the server's key, not the Signal owner's: it is the /// 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 { +pub async fn pick_icon(ctx: &Ctx, who: &str, key: &str, name: &str, words: &[String], description: Option<&str>, candidates: &[(String, String)]) -> Result { let span = tracing::info_span!( "jev_icon", candidates = candidates.len(), @@ -342,6 +348,8 @@ pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], descr ); called(span, async { let answer = ask(ctx, key, &icon_request(name, words, description, candidates)).await?; + let [input, output] = usage(&answer); + STATS.bill(who, "server", [0.0, 0.0, input, output]); chosen(&answer, candidates).ok_or_else(|| anyhow::anyhow!("Jev chose no icon offered: {answer}")) }) .await @@ -349,7 +357,7 @@ pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], descr /// Whether TypeSafe takes `key`, by one short question, before it is saved (#271): a key it /// refuses would sit saved with nothing ever picked out, and nobody told why. -pub async fn check_key(ctx: &Ctx, key: &str) -> Result<()> { +pub async fn check_key(ctx: &Ctx, who: &str, key: &str) -> Result<()> { let span = tracing::info_span!( "jev_key_check", "http.response.status_code" = tracing::field::Empty, @@ -365,7 +373,10 @@ pub async fn check_key(ctx: &Ctx, key: &str) -> Result<()> { "questions": { "ok": { "type": "noul", "instructions": "Is this a key check?", "criteria": { "true": "It is.", "false": "It is not." } } }, }); - called(span, ask(ctx, key, &body)).await.map(|_| ()) + let answer = called(span, ask(ctx, key, &body)).await?; + let [input, output] = usage(&answer); + STATS.bill(who, "own", [0.0, 0.0, input, output]); + Ok(()) } /// Puts `body` to Jev on `key` and gives its answer, or what TypeSafe refused it with. Carries @@ -388,16 +399,21 @@ async fn ask(ctx: &Ctx, key: &str, body: &serde_json::Value) -> Result [f64; 2] { + ["input_tokens", "output_tokens"].map(|k| answer["usage"][k].as_f64().unwrap_or(0.0)) +} + /// 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 { @@ -541,7 +557,7 @@ mod tests { #[test] fn on_the_servers_key_one_item_is_one_call_for_everyone() { let row = |user_id, guid: &str| Unjudged { - user_id, key: None, feed_id: "f".into(), guid: guid.into(), feed_title: None, title: None, description: None, + user_id, user_name: format!("u{user_id}"), key: None, feed_id: "f".into(), guid: guid.into(), feed_title: None, title: None, description: None, signals: vec![Ask { id: user_id * 10, name: "s".into(), keywords: vec![], description: None }], }; let items = [row(1, "x"), row(2, "x"), row(1, "y")]; @@ -571,7 +587,7 @@ mod tests { #[test] fn a_batch_is_full_by_its_rows_not_its_items() { let item = |n: usize| Unjudged { - user_id: 1, key: Some("k".into()), feed_id: "f".into(), guid: "g".into(), feed_title: None, title: None, description: None, + user_id: 1, user_name: "a".into(), key: Some("k".into()), feed_id: "f".into(), guid: "g".into(), feed_title: None, title: None, description: None, signals: (0..n as i64).map(|id| Ask { id, name: "s".into(), keywords: vec![], description: None }).collect(), }; assert!(full(&(0..25).map(|_| item(2)).collect::>()), "25 items of two Signals each is the 50 asked for"); @@ -582,6 +598,7 @@ mod tests { fn each_signal_is_its_own_question_on_the_items_words() { let item = Unjudged { user_id: 1, + user_name: "a".into(), key: Some("k".into()), feed_id: "f".into(), guid: "g".into(), diff --git a/crates/ipx/src/web.rs b/crates/ipx/src/web.rs index ec291d2..df394b7 100644 --- a/crates/ipx/src/web.rs +++ b/crates/ipx/src/web.rs @@ -418,7 +418,7 @@ async fn patch_me( return Err(ApiError::bad_request("that does not look like a TypeSafe key")); } if let Some(k) = &key { - crate::signals::check_key(&state.ctx, k).await.map_err(|e| { + crate::signals::check_key(&state.ctx, &user.name, k).await.map_err(|e| { if crate::signals::refuses_key(&e) { ApiError::bad_request("TypeSafe did not accept that key") } else { @@ -527,7 +527,7 @@ struct IconAsk { /// or 404 when the server has no key. async fn pick_signal_icon( State(state): State, - _user: crate::db::User, + user: crate::db::User, Json(body): Json, ) -> Result, ApiError> { let Some(key) = crate::signals::typesafe_key() else { @@ -547,7 +547,7 @@ async fn pick_signal_icon( if candidates.is_empty() || candidates.len() > 60 { return Err(ApiError::bad_request("between 1 and 60 icons to choose from")); } - let icon = crate::signals::pick_icon(&state.ctx, &key, &name, &words, description.as_deref(), &candidates) + let icon = crate::signals::pick_icon(&state.ctx, &user.name, &key, &name, &words, description.as_deref(), &candidates) .await .map_err(|e| { tracing::warn!(error = %format!("{e:#}"), "Jev could not pick an icon"); diff --git a/docs/signals.md b/docs/signals.md index eb0be1d..e5db252 100644 --- a/docs/signals.md +++ b/docs/signals.md @@ -392,7 +392,12 @@ erDiagram - **Keywords changed while an item is being judged** can leave that one item's verdict from the old words. - **The meter is the server's**, `ipx_jev_tokens_total` and `ipx_jev_requests_total` on - `/metrics`, and the Signals row of the dashboard: tokens as TypeSafe's answers count them. It is + `/metrics`, and the Signals row of the dashboard: tokens as TypeSafe's answers count them, by + person (`user`, the account's name) and whose key paid (`key`, `server` or `own`; #278), with + `ipx_signals_judged_total` and `ipx_signals_hits_total` the same way. A call shared by several + people on the server's key is split by each one's share of its questions. Icon picks bill the + person on the server's key, key checks the person on their own. The dashboard's **Who used what** + sums them over its time range. It is one call per item in the person's feeds as they come, and per item in the window for a new Signal. - **Only the window** is judged by publish date, apart from what an edit asks again. diff --git a/grafana/dashboard.py b/grafana/dashboard.py index 05d324b..5b9ae88 100644 --- a/grafana/dashboard.py +++ b/grafana/dashboard.py @@ -291,6 +291,29 @@ p = ts("Against TypeSafe's limits", [prom('sum(rate(ipx_jev_requests_total{job=" p["fieldConfig"]["defaults"]["custom"]["thresholdsStyle"] = {"mode": "line"} p["fieldConfig"]["defaults"]["thresholds"] = {"mode": "absolute", "steps": LIMIT(80)} y += 8 +# By person and whose key paid (#278): who the server's TypeSafe bill is for. An item shared on the +# server's key is split by each person's share of its questions. +ts("Input tokens by person", [prom('sum by (user, key) (increase(ipx_jev_tokens_total{job="ipx", kind="input"}[$__rate_interval]))', + legend="{{user}} ({{key}} key)")], + 0, w=8, bars=True, stack=True, desc="server is the server's TypeSafe key, what the server pays for; own is the person's own key.") +ts("Items judged by person", [prom('sum by (user, key) (increase(ipx_signals_judged_total{job="ipx"}[$__rate_interval]))', legend="{{user}} ({{key}} key)"), + prom('sum by (user) (increase(ipx_signals_hits_total{job="ipx"}[$__rate_interval]))', ref="B", legend="{{user}}: fit a Signal")], + 8, w=8, bars=True) +def ptab(expr, ref): + return {"refId": ref, "datasource": PROM, "expr": expr, "instant": True, "format": "table"} +panel("table", "Who used what", 8, 8, 16, + [ptab('sum by (user, key) (increase(ipx_jev_tokens_total{job="ipx", kind="input"}[$__range]))', "A"), + ptab('sum by (user, key) (increase(ipx_jev_tokens_total{job="ipx", kind="output"}[$__range]))', "B"), + ptab('sum by (user, key) (increase(ipx_signals_judged_total{job="ipx"}[$__range]))', "C")], + description="Over the dashboard's time range: each person's input and output tokens and items judged, and whose key paid. " + "Icon picks are on the server's key, key checks on the person's own.", + transformations=[{"id": "merge", "options": {}}, + {"id": "organize", "options": {"excludeByName": {"Time": True}, + "renameByName": {"user": "Person", "key": "Key", "Value #A": "Input tokens", + "Value #B": "Output tokens", "Value #C": "Items judged"}}}], + options={"showHeader": True, "sortBy": [{"displayName": "Input tokens", "desc": True}]}, + fieldConfig={"defaults": {"decimals": 0}, "overrides": []}) +y += 8 # ---- ipx: the process, what it holds between requests and jobs (#194) row("ipx: process")