janitor emits DLQ depth

rust/cyclotron-core/src/janitor.rs

@@ -1,3 +1,4 @@
+use crate::DEAD_LETTER_QUEUE;
 use chrono::Duration;
 use sqlx::PgPool;
 
@@ -64,4 +65,16 @@ impl Janitor {
     pub async fn waiting_jobs(&self) -> Result<u64, QueueError> {
         count_total_waiting_jobs(&self.pool).await
     }
+
+    pub async fn count_dlq_depth(&self) -> Result<u64, QueueError> {
+        let result = sqlx::query_scalar!(
+            "SELECT COUNT(*) FROM cyclotron_jobs WHERE queue_name = $1",
+            DEAD_LETTER_QUEUE
+        )
+        .fetch_one(&self.pool)
+        .await
+        .map_err(QueueError::from)?;
+
+        Ok(result.unwrap_or(0) as u64)
+    }
 }

rust/cyclotron-janitor/src/janitor.rs

@@ -84,10 +84,16 @@ impl Janitor {
         }
 
         let available = {
-            let _time = common_metrics::timing_guard(QUEUE_DEPTH, &self.metrics_labels);
+            let _time = common_metrics::timing_guard(AVAILABLE_DEPTH_TIME, &self.metrics_labels);
             self.inner.waiting_jobs().await?
         };
-        common_metrics::gauge(QUEUE_DEPTH, &self.metrics_labels, available as f64);
+        common_metrics::gauge(AVAILABLE_DEPTH, &self.metrics_labels, available as f64);
+
+        let dlq_depth = {
+            let _time = common_metrics::timing_guard(DLQ_DEPTH_TIME, &self.metrics_labels);
+            self.inner.count_dlq_depth().await?
+        };
+        common_metrics::gauge(DLQ_DEPTH, &self.metrics_labels, dlq_depth as f64);
 
         common_metrics::inc(RUN_ENDS, &self.metrics_labels, 1);
         info!("Janitor loop complete");

rust/cyclotron-janitor/src/metrics_constants.rs

@@ -15,4 +15,7 @@ pub const STALLED_COUNT: &str = "cyclotron_janitor_stalled_jobs_reset";
 pub const STALLED_TIME: &str = "cyclotron_janitor_stalled_jobs_reset_ms";
 
 // The janitor should report some basic shard-level metrics
-pub const QUEUE_DEPTH: &str = "cyclotron_available_jobs";
+pub const AVAILABLE_DEPTH: &str = "cyclotron_available_jobs";
+pub const AVAILABLE_DEPTH_TIME: &str = "cyclotron_available_jobs_ms";
+pub const DLQ_DEPTH: &str = "cyclotron_dead_letter_queue_depth";
+pub const DLQ_DEPTH_TIME: &str = "cyclotron_dead_letter_queue_depth_ms";