feat: add prometheus metric queue_running_count (#6413)

* feat: add prometheus metric queue_running_count

Adds a new Prometheus metric queue_running_count that tracks the number
of currently running jobs per tag, similar to the existing queue_count
metric but filtered for running=true instead of running=false.

Changes:
- Added get_queue_running_counts() function in windmill-common/src/queue.rs
- Added QUEUE_RUNNING_COUNT Prometheus metric in monitor.rs
- Added /workers/queue_running_counts API endpoint
- Updated OpenAPI specification
- Added SQLx query cache entry

Requested by @rubenfiszel

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>

* Remove database insertion for queue_running_count metrics

Keep Prometheus metrics and API endpoint functionality while removing
the database INSERT statements as requested.

Co-authored-by: Ruben Fiszel <rubenfiszel@users.noreply.github.com>

* sqlx

* improve logic

---------

Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
Co-authored-by: Ruben Fiszel <rubenfiszel@users.noreply.github.com>
Co-authored-by: Ruben Fiszel <ruben@windmill.dev>

backend/.sqlx/query-1b56a720d99a689e80d12ee1efbfeb71d7cd7bb17e936746749c958062cdff9e.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT tag AS \"tag!\", count(*) AS \"count!\" FROM v2_job_queue WHERE\n            running = true\n            GROUP BY tag",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "tag!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "count!",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "1b56a720d99a689e80d12ee1efbfeb71d7cd7bb17e936746749c958062cdff9e"
+}

backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json

@@ -148,7 +148,8 @@
                 "oracledb",
                 "nu",
                 "java",
-                "duckdb"
+                "duckdb",
+                "ruby"
               ]
             }
           }

backend/.sqlx/query-f22964772dc2d67aee437bbbd08b64792c00da1d713d7ca8f9904ccce7bfdae7.json

@@ -1,248 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT\n                            cj.id AS \"id!\",\n                            cj.workspace_id AS \"workspace_id!\",\n                            cj.parent_job,\n                            cj.created_by AS \"created_by!\",\n                            cj.duration_ms AS \"duration_ms!\",\n                            cj.success AS \"success!\",\n                            cj.script_hash AS \"script_hash!: Option<ScriptHash>\",\n                            cj.script_path,\n                            cj.args AS \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n                            cj.result AS \"result: sqlx::types::Json<Box<RawValue>>\",\n                            cj.deleted AS \"deleted!\",\n                            cj.canceled AS \"canceled!\",\n                            cj.canceled_by,\n                            cj.canceled_reason,\n                            cj.job_kind AS \"job_kind!: JobKind\",\n                            cj.schedule_path,\n                            cj.permissioned_as AS \"permissioned_as!\",\n                            cj.is_flow_step AS \"is_flow_step!\",\n                            cj.language AS \"language: ScriptLang\",\n                            cj.is_skipped AS \"is_skipped!\",\n                            cj.email AS \"email!\",\n                            cj.visible_to_owner AS \"visible_to_owner!\",\n                            cj.mem_peak,\n                            cj.tag AS \"tag!\",\n                            cj.created_at AS \"created_at!\",\n                            cj.started_at,\n                            job_logs.logs,\n                            job_logs.log_offset AS \"log_offset?\",\n                            job_logs.log_file_index\n\n                        FROM v2_as_completed_job AS cj\n                        LEFT JOIN job_logs ON cj.id = job_logs.job_id\n                        WHERE cj.created_at < $1\n                        ORDER BY cj.created_at ASC LIMIT $2",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id!",
-        "type_info": "Uuid"
-      },
-      {
-        "ordinal": 1,
-        "name": "workspace_id!",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 2,
-        "name": "parent_job",
-        "type_info": "Uuid"
-      },
-      {
-        "ordinal": 3,
-        "name": "created_by!",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 4,
-        "name": "duration_ms!",
-        "type_info": "Int8"
-      },
-      {
-        "ordinal": 5,
-        "name": "success!",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 6,
-        "name": "script_hash!: Option<ScriptHash>",
-        "type_info": "Int8"
-      },
-      {
-        "ordinal": 7,
-        "name": "script_path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 8,
-        "name": "args: sqlx::types::Json<HashMap<String, Box<RawValue>>>",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 9,
-        "name": "result: sqlx::types::Json<Box<RawValue>>",
-        "type_info": "Jsonb"
-      },
-      {
-        "ordinal": 10,
-        "name": "deleted!",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 11,
-        "name": "canceled!",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 12,
-        "name": "canceled_by",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 13,
-        "name": "canceled_reason",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 14,
-        "name": "job_kind!: JobKind",
-        "type_info": {
-          "Custom": {
-            "name": "job_kind",
-            "kind": {
-              "Enum": [
-                "script",
-                "preview",
-                "flow",
-                "dependencies",
-                "flowpreview",
-                "script_hub",
-                "identity",
-                "flowdependencies",
-                "http",
-                "graphql",
-                "postgresql",
-                "noop",
-                "appdependencies",
-                "deploymentcallback",
-                "singlescriptflow",
-                "flowscript",
-                "flownode",
-                "appscript"
-              ]
-            }
-          }
-        }
-      },
-      {
-        "ordinal": 15,
-        "name": "schedule_path",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 16,
-        "name": "permissioned_as!",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 17,
-        "name": "is_flow_step!",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 18,
-        "name": "language: ScriptLang",
-        "type_info": {
-          "Custom": {
-            "name": "script_lang",
-            "kind": {
-              "Enum": [
-                "python3",
-                "deno",
-                "go",
-                "bash",
-                "postgresql",
-                "nativets",
-                "bun",
-                "mysql",
-                "bigquery",
-                "snowflake",
-                "graphql",
-                "powershell",
-                "mssql",
-                "php",
-                "bunnative",
-                "rust",
-                "ansible",
-                "csharp",
-                "oracledb",
-                "nu",
-                "java",
-                "duckdb",
-                "ruby"
-              ]
-            }
-          }
-        }
-      },
-      {
-        "ordinal": 19,
-        "name": "is_skipped!",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 20,
-        "name": "email!",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 21,
-        "name": "visible_to_owner!",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 22,
-        "name": "mem_peak",
-        "type_info": "Int4"
-      },
-      {
-        "ordinal": 23,
-        "name": "tag!",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 24,
-        "name": "created_at!",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 25,
-        "name": "started_at",
-        "type_info": "Timestamptz"
-      },
-      {
-        "ordinal": 26,
-        "name": "logs",
-        "type_info": "Text"
-      },
-      {
-        "ordinal": 27,
-        "name": "log_offset?",
-        "type_info": "Int4"
-      },
-      {
-        "ordinal": 28,
-        "name": "log_file_index",
-        "type_info": "TextArray"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Timestamptz",
-        "Int8"
-      ]
-    },
-    "nullable": [
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      true,
-      false,
-      true
-    ]
-  },
-  "hash": "f22964772dc2d67aee437bbbd08b64792c00da1d713d7ca8f9904ccce7bfdae7"
-}

backend/src/monitor.rs

@@ -114,6 +114,12 @@ lazy_static::lazy_static! {
         &["tag"]
     ).unwrap();
 
+    static ref QUEUE_RUNNING_COUNT: prometheus::IntGaugeVec = prometheus::register_int_gauge_vec!(
+        "queue_running_count",
+        "Number of running jobs in the queue",
+        &["tag"]
+    ).unwrap();
+
 }
 lazy_static::lazy_static! {
     static ref ZOMBIE_JOB_TIMEOUT: String = std::env::var("ZOMBIE_JOB_TIMEOUT")
@@ -140,6 +146,7 @@ lazy_static::lazy_static! {
     pub static ref WORKERS_NAMES: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(Vec::new()));
 
     static ref QUEUE_COUNT_TAGS: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(Vec::new()));
+    static ref QUEUE_RUNNING_COUNT_TAGS: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(Vec::new()));
     static ref DISABLE_CONCURRENCY_LIMIT: bool = std::env::var("DISABLE_CONCURRENCY_LIMIT").is_ok_and(|s| s == "true");
 
     static ref STALE_JOB_TRESHOLD_MINUTES: Option<u64> = std::env::var("STALE_JOB_TRESHOLD_MINUTES")
@@ -1324,7 +1331,10 @@ pub async fn reload_url_list_setting(
             match url::Url::parse(url_str) {
                 Ok(url) => urls.push(url),
                 Err(e) => {
-                    return Err(error::Error::BadRequest(format!("Invalid URL in FORCE_{}: '{}': {}", std_env_var, url_str, e)));
+                    return Err(error::Error::BadRequest(format!(
+                        "Invalid URL in FORCE_{}: '{}': {}",
+                        std_env_var, url_str, e
+                    )));
                 }
             }
         }
@@ -1347,7 +1357,11 @@ pub async fn reload_url_list_setting(
                 }
             }
         }
-        if urls.is_empty() { None } else { Some(urls) }
+        if urls.is_empty() {
+            None
+        } else {
+            Some(urls)
+        }
     } else {
         None
     };
@@ -1365,7 +1379,11 @@ pub async fn reload_url_list_setting(
                     }
                 }
             }
-            tracing::info!("Loaded setting {} from db config: {} URLs", setting_name, urls.len());
+            tracing::info!(
+                "Loaded setting {} from db config: {} URLs",
+                setting_name,
+                urls.len()
+            );
             value = if urls.is_empty() { None } else { Some(urls) };
         } else {
             tracing::error!("Could not parse {} found: {:#?}", setting_name, &q);
@@ -1671,6 +1689,30 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
             let mut w = QUEUE_COUNT_TAGS.write().await;
             *w = tags_to_watch;
         }
+
+        #[cfg(feature = "prometheus")]
+        if metrics_enabled {
+            // Handle queue running count metrics
+            let queue_running_counts = windmill_common::queue::get_queue_running_counts(db).await;
+
+            for q in QUEUE_RUNNING_COUNT_TAGS.read().await.iter() {
+                if queue_running_counts.get(q).is_none() {
+                    (*QUEUE_RUNNING_COUNT).with_label_values(&[q]).set(0);
+                }
+            }
+
+            let mut running_tags_to_watch = vec![];
+            for q in queue_running_counts {
+                let count = q.1;
+                let tag = q.0;
+
+                let metric = (*QUEUE_RUNNING_COUNT).with_label_values(&[&tag]);
+                metric.set(count as i64);
+                running_tags_to_watch.push(tag.to_string());
+            }
+            let mut w = QUEUE_RUNNING_COUNT_TAGS.write().await;
+            *w = running_tags_to_watch;
+        }
     }
 
     // clean queue metrics older than 14 days

backend/windmill-api/openapi.yaml

@@ -11502,6 +11502,22 @@ paths:
                 additionalProperties:
                   type: integer
 
+  /workers/queue_running_counts:
+    get:
+      summary: get counts of currently running jobs per tag
+      operationId: getCountsOfRunningJobsPerTag
+      tags:
+        - worker
+      responses:
+        "200":
+          description: queue running counts
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties:
+                  type: integer
+
   /configs/list_worker_groups:
     get:
       summary: list worker groups

backend/windmill-api/src/workers.rs

@@ -37,6 +37,7 @@ pub fn global_service() -> Router {
         .route("/get_default_tags", get(get_default_tags))
         .route("/queue_metrics", get(get_queue_metrics))
         .route("/queue_counts", get(get_queue_counts))
+        .route("/queue_running_counts", get(get_queue_running_counts))
 }
 
 #[derive(FromRow, Serialize, Deserialize)]
@@ -219,3 +220,12 @@ async fn get_queue_counts(
     let queue_counts = windmill_common::queue::get_queue_counts(&db).await;
     Ok(Json(queue_counts))
 }
+
+async fn get_queue_running_counts(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+) -> JsonResult<std::collections::HashMap<String, u32>> {
+    require_super_admin(&db, &authed.email).await?;
+    let queue_running_counts = windmill_common::queue::get_queue_running_counts(&db).await;
+    Ok(Json(queue_running_counts))
+}

backend/windmill-common/src/queue.rs

@@ -14,3 +14,16 @@ pub async fn get_queue_counts(db: &Pool<Postgres>) -> HashMap<String, u32> {
     .map(|v| v.into_iter().map(|x| (x.tag, x.count as u32)).collect())
     .unwrap_or_else(|| HashMap::new())
 }
+
+pub async fn get_queue_running_counts(db: &Pool<Postgres>) -> HashMap<String, u32> {
+    sqlx::query!(
+        "SELECT tag AS \"tag!\", count(*) AS \"count!\" FROM v2_job_queue WHERE
+            running = true
+            GROUP BY tag",
+    )
+    .fetch_all(db)
+    .await
+    .ok()
+    .map(|v| v.into_iter().map(|x| (x.tag, x.count as u32)).collect())
+    .unwrap_or_else(|| HashMap::new())
+}