GITGUD

neiam neiam /dms

public · Issues · Pulls · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

Gauges for the archive and the job pipeline

7ddc104 · Gabriel Morell · 2026-08-24 22:09

6 files +227 -1
Message 
{commit_body(@commit)}

Files changed

modified Cargo.toml
+5 −1
@@ -26,6 +26,9 @@ anyhow = "1.0.100"
26 26 tracing = "0.1.41"
27 27 axum-prometheus = "0.9.0"
28 28 metrics-process = "2.4.0"
29 +# Already in the tree under axum-prometheus; declared so the app can describe
30 +# and set gauges of its own alongside the HTTP ones.
31 +metrics = "0.24"
29 32 axum-htmx = { version = "0.7.0", features = ["tokio"] }
30 33 minijinja = { version = "2.10.2", features = ["loader", "builtins", "json", "urlencode"] }
31 34 minijinja-embed = "2.12"
@@ -35,13 +38,14 @@ sea-orm = { workspace = true }
35 38 chrono = { workspace = true }
36 39 uuid = { workspace = true }
37 40 serde = { workspace = true }
38 −argon2 = "0.6.0-rc.1"
41 +argon2 = { version = "0.6.0-rc.1", features = ["std"] }
39 42 sha2 = "0.11.0-rc.2"
40 43 hex = "0.4.3"
41 44 migration = { path = "./migration" }
42 45
43 46 # CLI dependencies
44 47 clap = { version = "4.5", features = ["derive"] }
48 +rpassword = "7.4"
45 49 serde_json = "1.0.145"
46 50
47 51 # Queue + storage + media
modified src/bin/webservice.rs
+10 −0
@@ -50,6 +50,16 @@ async fn main() -> Result<(), anyhow::Error> {
50 50 collector.describe();
51 51 let (prometheus_layer, metric_handle) = PrometheusMetricLayer::pair();
52 52
53 + // Domain gauges — the archive, and whether the job pipeline behind an
54 + // upload is finishing. Described before the sampler starts so the names
55 + // are in /metrics from the first scrape rather than appearing a minute in,
56 + // which a panel reads as an outage.
57 + //
58 + // Every replica reports the same database-wide numbers, so anything
59 + // graphing these has to aggregate with `max`, never `sum`.
60 + dms::telemetry::describe();
61 + tokio::spawn(dms::telemetry::run(db.clone(), config.worker_max_attempts));
62 +
53 63 let mut jinja = Environment::new();
54 64 minijinja_embed::load_templates!(&mut jinja);
55 65 jinja.set_loader(minijinja::path_loader("templates"));
modified src/lib.rs
+1 −0
@@ -6,6 +6,7 @@ pub mod config;
6 6 pub mod jobs;
7 7 pub mod queue;
8 8 pub mod storage;
9 +pub mod telemetry;
9 10 pub mod tags;
10 11 pub mod workers;
11 12
added src/telemetry.rs
+211 −0
@@ -0,0 +1,211 @@
1 +//! Gauges describing the archive itself, alongside the HTTP and process
2 +//! metrics.
3 +//!
4 +//! These answer questions the request metrics cannot: how much has been
5 +//! stored, and — the reason this module exists — whether the processing
6 +//! pipeline behind an upload is actually finishing. They come from counting
7 +//! rows, so they are sampled on a timer rather than updated on every write; a
8 +//! gauge that every code path touching a table has to keep correct is a gauge
9 +//! that drifts.
10 +//!
11 +//! **Every replica reports the same numbers.** They describe the database, not
12 +//! the pod, so a dashboard must aggregate them with `max` and never `sum`;
13 +//! summing counts the same uploads once per web pod.
14 +//!
15 +//! The job gauges are the point. An upload here fans out into up to five
16 +//! independent jobs — thumbnail, ocr, describe, extract_tags, similarity — run
17 +//! by four separate worker Deployments. When one of those stalls, the site
18 +//! stays up, uploads keep succeeding, and the only visible symptom is that
19 +//! thumbnails quietly stop appearing. [`sample`] breaks the job table down by
20 +//! `kind` and `status` so the stalled one names itself.
21 +use std::sync::Arc;
22 +use std::time::Duration;
23 +
24 +use chrono::Utc;
25 +use sea_orm::{
26 + ColumnTrait, ConnectionTrait, DatabaseBackend, DatabaseConnection, EntityTrait, PaginatorTrait,
27 + QueryFilter, Statement,
28 +};
29 +
30 +use crate::entities::{
31 + ai_descriptions, api_tokens, ocr_results, tags, upload_jobs, uploads, user_sessions, users,
32 +};
33 +use crate::jobs::{
34 + STATUS_COMPLETE, STATUS_FAILED, STATUS_PENDING, STATUS_PROCESSING, UPLOAD_STATUS_COMPLETE,
35 + UPLOAD_STATUS_PARTIAL, UPLOAD_STATUS_PROCESSING,
36 +};
37 +
38 +/// How often the counts are refreshed.
39 +const SAMPLE_EVERY: Duration = Duration::from_secs(60);
40 +
41 +/// How many ticks between refreshes of the one query that scans a big table.
42 +///
43 +/// Summing `file_size_bytes` across every upload cannot use an index, so
44 +/// Postgres reads the heap. Once a minute forever is not a reasonable price
45 +/// for a number that is read as a slow-moving line; ten minutes is well below
46 +/// the resolution anyone looks at it.
47 +const SLOW_EVERY: u64 = 10;
48 +
49 +/// How long a job sits in `processing` before it is called stuck.
50 +///
51 +/// Thumbnailing is seconds and a `describe` against the VLM is at worst a
52 +/// couple of minutes, so an hour is an order of magnitude past anything
53 +/// healthy. This is the signal for a worker killed mid-job: the row is claimed
54 +/// and no longer queued, so nothing will ever pick it up again.
55 +const STUCK_AFTER: chrono::TimeDelta = chrono::TimeDelta::hours(1);
56 +
57 +/// The job kinds, in the order `queue::JobKind` declares them.
58 +///
59 +/// Listed rather than discovered so a kind that has never run still reports
60 +/// zero instead of dropping its series — a missing line and a line at zero
61 +/// mean very different things on the pipeline panel.
62 +const KINDS: [&str; 5] = ["thumbnail", "ocr", "describe", "extract_tags", "similarity"];
63 +
64 +const JOB_STATUSES: [&str; 4] = [STATUS_PENDING, STATUS_PROCESSING, STATUS_COMPLETE, STATUS_FAILED];
65 +
66 +const UPLOAD_STATUSES: [&str; 3] =
67 + [UPLOAD_STATUS_PROCESSING, UPLOAD_STATUS_COMPLETE, UPLOAD_STATUS_PARTIAL];
68 +
69 +/// Register the metric names and what they mean.
70 +///
71 +/// Called at boot so the names exist in `/metrics` from the first scrape.
72 +/// Without this a fresh pod exposes nothing until the first sample lands, and
73 +/// a panel over a just-restarted deployment reads as an outage.
74 +pub fn describe() {
75 + metrics::describe_gauge!(
76 + "dms_uploads",
77 + "Uploads held, by processing state. `partial` means every job finished but at least \
78 + one failed — the file is stored and usable, it just has no thumbnail or no OCR text"
79 + );
80 + metrics::describe_gauge!(
81 + "dms_uploads_bytes",
82 + "Total stored bytes across every upload. Refreshed every ten minutes rather than every \
83 + minute; see SLOW_EVERY"
84 + );
85 + metrics::describe_gauge!(
86 + "dms_jobs",
87 + "Processing jobs by kind and status. This is the pipeline: five kinds fan out from one \
88 + upload, and a worker Deployment stalling shows up as its kind stuck in `pending`"
89 + );
90 + metrics::describe_gauge!(
91 + "dms_jobs_stuck",
92 + "Jobs claimed as `processing` more than an hour ago. The worker that took them is gone \
93 + and the row is no longer queued, so nothing will finish them without intervention"
94 + );
95 + metrics::describe_gauge!(
96 + "dms_jobs_exhausted",
97 + "Failed jobs that have used every retry (attempts >= WORKER_MAX_ATTEMPTS). These will \
98 + not be retried again and are what the /failed page exists to show"
99 + );
100 + metrics::describe_gauge!("dms_ocr_results", "Uploads with extracted text");
101 + metrics::describe_gauge!("dms_ai_descriptions", "Uploads the VLM has described");
102 + metrics::describe_gauge!("dms_tags", "Distinct tags across the archive");
103 + metrics::describe_gauge!("dms_users", "Accounts that exist");
104 + metrics::describe_gauge!("dms_sessions_active", "Sessions not yet expired");
105 + metrics::describe_gauge!("dms_api_tokens", "API tokens issued");
106 +}
107 +
108 +/// Sample the counts forever, on a timer.
109 +///
110 +/// `max_attempts` is the app's own `WORKER_MAX_ATTEMPTS`, so the exhausted
111 +/// count agrees with the retry limit the workers actually enforce rather than
112 +/// with a second threshold invented here.
113 +///
114 +/// Failures are logged and skipped rather than retried or fatal: a metrics
115 +/// sampler that can take the process down, or that hammers a struggling
116 +/// database, is worse than one that misses a minute.
117 +pub async fn run(db: Arc<DatabaseConnection>, max_attempts: i32) {
118 + let mut ticker = tokio::time::interval(SAMPLE_EVERY);
119 + // The first tick fires immediately, so the numbers are right from boot
120 + // rather than a minute after it.
121 + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
122 + let mut tick: u64 = 0;
123 + loop {
124 + ticker.tick().await;
125 + // Tick zero includes the slow gauge, so the archive size is present
126 + // from the first scrape rather than ten minutes into the pod's life.
127 + let include_slow = tick.is_multiple_of(SLOW_EVERY);
128 + tick = tick.wrapping_add(1);
129 + if let Err(e) = sample(db.as_ref(), max_attempts, include_slow).await {
130 + tracing::warn!(error = %e, "could not sample metrics");
131 + }
132 + }
133 +}
134 +
135 +async fn sample(
136 + db: &DatabaseConnection,
137 + max_attempts: i32,
138 + include_slow: bool,
139 +) -> Result<(), sea_orm::DbErr> {
140 + let now = Utc::now();
141 +
142 + for status in UPLOAD_STATUSES {
143 + let n = uploads::Entity::find()
144 + .filter(uploads::Column::ProcessingStatus.eq(status))
145 + .count(db)
146 + .await?;
147 + metrics::gauge!("dms_uploads", "status" => status).set(n as f64);
148 + }
149 +
150 + // The pipeline, broken down the way it actually fails. Twenty cheap
151 + // counted queries rather than one grouped one, because a kind or status
152 + // with no rows has to report zero and a GROUP BY simply omits it.
153 + for kind in KINDS {
154 + for status in JOB_STATUSES {
155 + let n = upload_jobs::Entity::find()
156 + .filter(upload_jobs::Column::Kind.eq(kind))
157 + .filter(upload_jobs::Column::Status.eq(status))
158 + .count(db)
159 + .await?;
160 + metrics::gauge!("dms_jobs", "kind" => kind, "status" => status).set(n as f64);
161 + }
162 + }
163 +
164 + // Deliberately unwindowed: a job wedged since last week is exactly the one
165 + // worth knowing about, and any recent-only window would hide it.
166 + metrics::gauge!("dms_jobs_stuck").set(
167 + upload_jobs::Entity::find()
168 + .filter(upload_jobs::Column::Status.eq(STATUS_PROCESSING))
169 + .filter(upload_jobs::Column::StartedAt.lt(now - STUCK_AFTER))
170 + .count(db)
171 + .await? as f64,
172 + );
173 + metrics::gauge!("dms_jobs_exhausted").set(
174 + upload_jobs::Entity::find()
175 + .filter(upload_jobs::Column::Status.eq(STATUS_FAILED))
176 + .filter(upload_jobs::Column::Attempts.gte(max_attempts))
177 + .count(db)
178 + .await? as f64,
179 + );
180 +
181 + metrics::gauge!("dms_ocr_results").set(ocr_results::Entity::find().count(db).await? as f64);
182 + metrics::gauge!("dms_ai_descriptions")
183 + .set(ai_descriptions::Entity::find().count(db).await? as f64);
184 + metrics::gauge!("dms_tags").set(tags::Entity::find().count(db).await? as f64);
185 +
186 + metrics::gauge!("dms_users").set(users::Entity::find().count(db).await? as f64);
187 + metrics::gauge!("dms_sessions_active").set(
188 + user_sessions::Entity::find()
189 + .filter(user_sessions::Column::ExpiresAt.gt(now))
190 + .count(db)
191 + .await? as f64,
192 + );
193 + metrics::gauge!("dms_api_tokens").set(api_tokens::Entity::find().count(db).await? as f64);
194 +
195 + if include_slow {
196 + // Raw SQL rather than sea-orm's `sum()`: Postgres returns SUM(bigint)
197 + // as numeric, which would need a Decimal round-trip to read back. The
198 + // cast makes the type unambiguous and the gauge is an f64 anyway.
199 + let row = db
200 + .query_one(Statement::from_string(
201 + DatabaseBackend::Postgres,
202 + "SELECT COALESCE(SUM(file_size_bytes), 0)::float8 AS total FROM uploads",
203 + ))
204 + .await?;
205 + if let Some(row) = row {
206 + metrics::gauge!("dms_uploads_bytes").set(row.try_get::<f64>("", "total")?);
207 + }
208 + }
209 +
210 + Ok(())
211 +}

Parents: 2e5cc52