From e54a8013a033fe3021df58a285d09ebc7461cfec Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Sat, 16 Sep 2023 02:31:58 +0200
Subject: [PATCH] fix: add queue_count to metrics

---
 backend/src/main.rs    |  4 +++-
 backend/src/monitor.rs | 28 +++++++++++++++++++++++++++-
 2 files changed, 30 insertions(+), 2 deletions(-)

diff --git a/backend/src/main.rs b/backend/src/main.rs
index f6cbd90bb6..b6a9c12503 100644
--- a/backend/src/main.rs
+++ b/backend/src/main.rs
@@ -8,6 +8,7 @@
 
 use gethostname::gethostname;
 use git_version::git_version;
+use rand::Rng;
 use sqlx::{Pool, Postgres};
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
@@ -201,9 +202,10 @@ Windmill Community Edition {GIT_VERSION}
 
             let mut rx = rx.resubscribe();
             let base_internal_url = base_internal_url.to_string();
+            let rd_delay = rand::thread_rng().gen_range(0..30);
             tokio::spawn(async move {
                 //monitor_db is applied at start, no need to apply it twice
-                tokio::time::sleep(Duration::from_secs(30)).await;
+                tokio::time::sleep(Duration::from_secs(rd_delay)).await;
                 loop {
                     monitor_db(
                         &db,
diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs
index 21366c4809..ca6d67f64b 100644
--- a/backend/src/monitor.rs
+++ b/backend/src/monitor.rs
@@ -34,6 +34,12 @@ lazy_static::lazy_static! {
         "Total number of jobs deleted due to their ping timing out in an unrecoverable state."
     )
     .unwrap();
+
+    static ref QUEUE_COUNT: prometheus::IntGaugeVec = prometheus::register_int_gauge_vec!(
+        "queue_count",
+        "Number of jobs in the queue",
+        &["tag"]
+    ).unwrap();
 }
 
 pub async fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
@@ -66,14 +72,34 @@ pub async fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
             }
         }
     };
+    let expose_queue_metrics_f = async {
+        if *METRICS_ENABLED && server_mode {
+            expose_queue_metrics(&db).await;
+        }
+    };
     join!(
         expired_items_f,
         zombie_jobs_f,
         reload_worker_config_f,
-        reload_custom_tags_f
+        reload_custom_tags_f,
+        expose_queue_metrics_f
     );
 }
 
+pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
+    let queue_counts = sqlx::query!("SELECT tag, count(*) as count FROM queue GROUP BY tag")
+        .fetch_all(db)
+        .await
+        .ok()
+        .unwrap_or_else(|| vec![]);
+    for q in queue_counts {
+        let count = q.count.unwrap_or(0);
+        let tag = q.tag;
+        let metric = (*QUEUE_COUNT).with_label_values(&[&tag]);
+        metric.set(count as i64);
+    }
+}
+
 pub async fn reload_worker_config(db: &Pool<Postgres>, tx: tokio::sync::broadcast::Sender<()>) {
     let config = load_worker_config(&db).await;
     if let Err(e) = config {