feat: worker metrics (#3697)

* feat: worker metrics * fix: limit worker metrics to superadmin * fix: limit worker metrics to superadmin + feat: queue metrics charts * fix: improve graphs
2026-09-07 00:01:49 +00:00 · 2024-05-10 20:50:09 +02:00
parent 056baa7a7d
commit cb4b2f60ff
31 changed files with 894 additions and 83 deletions
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT created_at FROM metrics WHERE id = 'author_count' ORDER BY created_at DESC LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "021be0f26ea87e587e656b24a9a94538efbf54a1447a3898e19773789cfc9063"
+}
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "SELECT tag, count(*) as count FROM queue WHERE\n        scheduled_for <= now() - ('3 seconds')::interval AND running = false\n        GROUP BY tag",
+  "query": "SELECT tag, count(*) as count FROM queue WHERE\n                scheduled_for <= now() - ('3 seconds')::interval AND running = false\n                GROUP BY tag",
  "describe": {
    "columns": [
      {
@@ -22,5 +22,5 @@
      null
    ]
  },
-  "hash": "3e0e5e0076ae0f7771abd64f8c5bb7e003e3f85351c946a012037e0c874acf1a"
+  "hash": "02b516dac764662194db1bc33e365c01f40bae70af3683f1f09748f6020f0d49"
 }
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM metrics \n                WHERE (id = 'author_count' OR id = 'operator_count' OR id = 'worker_usage') AND created_at < NOW() - INTERVAL '6 month'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "3667d72d23e8c35ab96d5e5d1fbfd94a8a9f74b6398e21f13c206c3f9427c6a9"
+}
@@ -0,0 +1,38 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT worker, worker_instance, vcpus, memory FROM worker_ping WHERE ping_at > NOW() - INTERVAL '2 minutes'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "worker",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "worker_instance",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "vcpus",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 3,
+        "name": "memory",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      true,
+      true
+    ]
+  },
+  "hash": "4887db074e9058bc0c1887428d6b2a897fe4b30c83f2aa5e9464c156f08f1c3a"
+}
@@ -1,16 +1,17 @@
 {
  "db_name": "PostgreSQL",
-  "query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2 WHERE worker = $3",
+  "query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $4",
  "describe": {
    "columns": [],
    "parameters": {
      "Left": [
        "Int4",
        "TextArray",
+        "Float4",
        "Text"
      ]
    },
    "nullable": []
  },
-  "hash": "47beea5cd6324b53bfb349665fb215280f32b70a617fde87f70ea53ca9ade39f"
+  "hash": "54fef88cc6b9e8db7c07fccbaa845edfefa339153a32e468faad9f063008863d"
 }
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO metrics (id, value) VALUES ('worker_usage', $1)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5ce9a9e0669b299b998bc9080d24f8d6f56eb049bec2e0745d27ca3a4fc0c296"
+}
@@ -5,7 +5,7 @@
    "columns": [
      {
        "ordinal": 0,
-        "name": "?column?",
+        "name": "bool",
        "type_info": "Bool"
      }
    ],
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version, vcpus, memory) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "TextArray",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Int8",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6afc5c7cbb3abe11ade0cedf1f7328005ce4de3165cdd998e5a0d27e044c7153"
+}
@@ -1,20 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Varchar",
-        "Varchar",
-        "Varchar",
-        "TextArray",
-        "Varchar",
-        "Varchar",
-        "Varchar"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "7b609ed87f974dfe887778746230b40b485658a75a1186c6ebb4c2de1b52a2fa"
-}
@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH queue_metrics as (\n            SELECT id, value, created_at\n            FROM metrics\n            WHERE id LIKE 'queue_%'\n                AND created_at > now() - interval '14 day'\n            ORDER BY created_at ASC\n        )\n        SELECT id, array_agg(json_build_object('value', value, 'created_at', created_at)) as \"values!\"\n        FROM queue_metrics\n        GROUP BY id\n        ORDER BY id ASC",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "values!",
+        "type_info": "JsonArray"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "7c95f3652de4561b18c1a340db3e0113a4777d38de4468fc41b22ce5b6fd1db7"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM metrics WHERE id LIKE 'queue_%' AND created_at < NOW() - INTERVAL '14 day'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "7c9a464ac807051b99fe37f2078f1b17f824e6d9b1124db618855a15a98e31f6"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT created_at FROM metrics WHERE id LIKE 'queue_count_%' ORDER BY created_at DESC LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "82f6674f19e8ad51a992505a46f46fc4a48172f104e9e849f755ac041c3eef92"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO metrics (id, value) VALUES ($1, $2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "8824b382c4e98dfa17b4aa656af3a6c1ff99973e778d71bd598a50d022da8f15"
+}
@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2 WHERE worker = $3",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "8c62e3bb264c7336b3b3a68677993ba3b729c7358cc3da14513d8b447ce65aaf"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO metrics (id, value)\n                            VALUES ($1, to_jsonb((SELECT EXTRACT(EPOCH FROM now() - scheduled_for)\n                            FROM queue WHERE tag = $2 AND running = false AND scheduled_for <= now() - ('3 seconds')::interval\n                            ORDER BY priority DESC NULLS LAST, scheduled_for, created_at LIMIT 1)))",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "9bf41c3161a02b7d0731c4e1d79519cef5255f5df1b759af3aa4985bb64313e5"
+}
@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH all_users as (SELECT count(*)::INT as count FROM usr WHERE disabled IS false),\n                authors as (SELECT count(distinct email)::INT as count FROM usr WHERE usr.operator IS false AND disabled IS false)\n                SELECT authors.count as author_count, all_users.count - authors.count as operator_count FROM all_users, authors",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "author_count",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 1,
+        "name": "operator_count",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null,
+      null
+    ]
+  },
+  "hash": "c584aeec21716f7405c355d5684ff3ac1be3f70849d67b8c2d4787e809b7ea2b"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO metrics (id, value) VALUES ('author_count', $1), ('operator_count', $2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Jsonb",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c6b2791dd109c7bf40a40b557d9d5a140e70c49718fcb3a776b77f8f2ab901d4"
+}
@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE worker_ping SET ping_at = now() WHERE worker = $1",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "c9d97800eb0ec87df8e8959b283dacb2c6cce422365ed394375641488ceb6b65"
-}
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "SELECT worker, worker_instance,  EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags, worker_group, wm_version FROM worker_ping\n         WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)\n         ORDER BY ping_at desc LIMIT $2 OFFSET $3",
+  "query": "SELECT worker, worker_instance,  EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as current_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as current_job_workspace_id, custom_tags, worker_group, wm_version, occupancy_rate\n        FROM worker_ping\n        WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)\n        ORDER BY ping_at desc LIMIT $2 OFFSET $3",
  "describe": {
    "columns": [
      {
@@ -35,25 +35,41 @@
      },
      {
        "ordinal": 6,
-        "name": "custom_tags",
-        "type_info": "TextArray"
+        "name": "current_job_id",
+        "type_info": "Uuid"
      },
      {
        "ordinal": 7,
-        "name": "worker_group",
+        "name": "current_job_workspace_id",
        "type_info": "Varchar"
      },
      {
        "ordinal": 8,
+        "name": "custom_tags",
+        "type_info": "TextArray"
+      },
+      {
+        "ordinal": 9,
+        "name": "worker_group",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 10,
        "name": "wm_version",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 11,
+        "name": "occupancy_rate",
+        "type_info": "Float4"
      }
    ],
    "parameters": {
      "Left": [
        "Int4",
        "Int8",
-        "Int8"
+        "Int8",
+        "Bool"
      ]
    },
    "nullable": [
@@ -63,10 +79,13 @@
      false,
      false,
      false,
+      null,
+      null,
      true,
      false,
-      false
+      false,
+      true
    ]
  },
-  "hash": "b38044d94e2ab03167c2f6fbb553ab3c19930ed11abf51763cd3ee378229443d"
+  "hash": "e00171cc3fc8f32922562d4fc4d3db56955fbc6ee9872d4b2e99157ebed131e0"
 }
@@ -0,0 +1 @@
+-- Add down migration script here
@@ -0,0 +1,7 @@
+-- Add up migration script here
+ALTER TABLE worker_ping 
+ADD COLUMN current_job_id UUID,
+ADD COLUMN current_job_workspace_id VARCHAR(50),
+ADD COLUMN vcpus BIGINT,
+ADD COLUMN memory BIGINT,
+ADD COLUMN occupancy_rate REAL;
@@ -0,0 +1 @@
+-- Add down migration script here
@@ -0,0 +1,8 @@
+-- Add up migration script here
+CREATE TABLE metrics (
+  id VARCHAR(255) NOT NULL,
+  value JSONB NOT NULL,
+  created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL
+);
+CREATE INDEX metrics_key_idx ON metrics(id);
+CREATE INDEX metrics_sort_idx ON metrics(created_at DESC);
@@ -8,7 +8,7 @@ use std::{
 };

 use rsmq_async::MultiplexedRsmq;
-use serde::de::DeserializeOwned;
+use serde::{de::DeserializeOwned, Serialize};
 use sqlx::{Pool, Postgres};
 use tokio::{
    join,
@@ -678,39 +678,182 @@ pub async fn monitor_db(
    };

    let expose_queue_metrics_f = async {
-        if !initial_load
-            && METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed)
-            && server_mode
-        {
+        if !initial_load && server_mode {
            expose_queue_metrics(&db).await;
        }
    };
+
+    let save_usage_metrics_f = async {
+        if !initial_load && server_mode {
+            save_usage_metrics(&db).await;
+        }
+    };
+
    join!(
        expired_items_f,
        zombie_jobs_f,
        expose_queue_metrics_f,
+        save_usage_metrics_f,
        verify_license_key_f
    );
 }

 pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
-    let queue_counts = sqlx::query!(
-        "SELECT tag, count(*) as count FROM queue WHERE
-        scheduled_for <= now() - ('3 seconds')::interval AND running = false
-        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);
+    let tx = db.begin().await;
+    if let Ok(mut tx) = tx {
+        let last_check = sqlx::query_scalar!(
+            "SELECT created_at FROM metrics WHERE id LIKE 'queue_count_%' ORDER BY created_at DESC LIMIT 1"
+        )
+        .fetch_optional(db)
+        .await
+        .unwrap_or(Some(chrono::Utc::now()));
+
+        let metrics_enabled = METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed);
+        let save_metrics = last_check
+            .map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::seconds(25))
+            .unwrap_or(true);
+
+        if metrics_enabled || save_metrics {
+
+            let queue_counts = sqlx::query!(
+                "SELECT tag, count(*) as count FROM queue WHERE
+                scheduled_for <= now() - ('3 seconds')::interval AND running = false
+                GROUP BY tag"
+            )
+            .fetch_all(&mut *tx)
+            .await
+            .ok()
+            .unwrap_or_else(|| vec![]);
+
+            for q in queue_counts {
+                let count = q.count.unwrap_or(0);
+                let tag = q.tag;
+                if metrics_enabled {
+                    let metric = (*QUEUE_COUNT).with_label_values(&[&tag]);
+                    metric.set(count as i64);
+                }
+
+                // save queue_count and delay metrics per tag
+                if save_metrics {
+                    sqlx::query!(
+                        "INSERT INTO metrics (id, value) VALUES ($1, $2)",
+                        format!("queue_count_{}", tag),
+                        serde_json::json!(count)
+                    )
+                    .execute(&mut *tx)
+                    .await
+                    .ok();
+                    if count > 0 {
+                        sqlx::query!(
+                            "INSERT INTO metrics (id, value)
+                            VALUES ($1, to_jsonb((SELECT EXTRACT(EPOCH FROM now() - scheduled_for)
+                            FROM queue WHERE tag = $2 AND running = false AND scheduled_for <= now() - ('3 seconds')::interval
+                            ORDER BY priority DESC NULLS LAST, scheduled_for, created_at LIMIT 1)))",
+                            format!("queue_delay_{}", tag),
+                            tag
+                        ).execute(&mut *tx).await.ok();
+                    }
+                }
+            }
+        }
+
+        // clean queue metrics older than 14 days
+        sqlx::query!(
+            "DELETE FROM metrics WHERE id LIKE 'queue_%' AND created_at < NOW() - INTERVAL '14 day'"
+        ).execute(&mut *tx).await.ok();
+
+        tx.commit().await.ok();
    }
 }

+
+#[derive(Serialize)]
+struct WorkerUsage {
+    worker: String,
+    worker_instance: String,
+    vcpus: Option<i64>,
+    memory: Option<i64>,
+
+}
+
+pub async fn save_usage_metrics(db: &Pool<Postgres>) {
+    let tx = db.begin().await;
+
+    if let Ok(mut tx) = tx {
+        let last_check = sqlx::query_scalar!(
+            "SELECT created_at FROM metrics WHERE id = 'author_count' ORDER BY created_at DESC LIMIT 1"
+        )
+        .fetch_optional(db)
+        .await
+        .unwrap_or(Some(chrono::Utc::now()));
+
+        let random_nb = rand::random::<i64>();
+
+        // save author and operator count every ~24 hours
+        if last_check
+            .map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::hours(24) - chrono::Duration::minutes(random_nb % 60))
+            .unwrap_or(true)
+        {
+            let counts = sqlx::query!(
+                "WITH all_users as (SELECT count(*)::INT as count FROM usr WHERE disabled IS false),
+                authors as (SELECT count(distinct email)::INT as count FROM usr WHERE usr.operator IS false AND disabled IS false)
+                SELECT authors.count as author_count, all_users.count - authors.count as operator_count FROM all_users, authors"
+            )
+            .fetch_one(&mut *tx)
+            .await
+            .ok();
+
+            if let Some(counts) = counts {
+                sqlx::query!(
+                    "INSERT INTO metrics (id, value) VALUES ('author_count', $1), ('operator_count', $2)",
+                    serde_json::json!(counts.author_count), 
+                    serde_json::json!(counts.operator_count)
+                )
+                .execute(&mut *tx)
+                .await
+                .ok();
+            }
+
+
+            // clean metrics older than 6 months (including worker usage)
+            sqlx::query!(
+                "DELETE FROM metrics 
+                WHERE (id = 'author_count' OR id = 'operator_count' OR id = 'worker_usage') AND created_at < NOW() - INTERVAL '6 month'"
+            )
+            .execute(&mut *tx)
+            .await
+            .ok();
+        }
+
+        // save worker usage every ~60 minutes
+        if last_check
+            .map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::minutes(60) - chrono::Duration::seconds(random_nb % 300))
+            .unwrap_or(true)
+        {
+            let worker_usage = sqlx::query_as!(
+                WorkerUsage,
+                "SELECT worker, worker_instance, vcpus, memory FROM worker_ping WHERE ping_at > NOW() - INTERVAL '2 minutes'"
+            )
+            .fetch_all(&mut *tx)
+            .await
+            .ok();
+
+            if let Some(worker_usage) = worker_usage {
+                sqlx::query!(
+                    "INSERT INTO metrics (id, value) VALUES ('worker_usage', $1)",
+                    serde_json::json!(worker_usage)
+                )
+                .execute(&mut *tx)
+                .await
+                .ok();
+            }
+        }
+
+        tx.commit().await.ok();
+    }
+
+}
+
 pub async fn reload_server_config(db: &Pool<Postgres>) {
    let config = load_server_config(&db).await;
    if let Err(e) = config {
@@ -6959,6 +6959,40 @@ paths:
              schema:
                type: boolean

+  /workers/queue_metrics:
+    get:
+      summary: get queue metrics
+      operationId: getQueueMetrics
+      tags:
+        - worker
+      responses:
+        "200":
+          description: metrics
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  type: object
+                  properties:
+                    id:
+                      type: string
+                    values: 
+                      type: array
+                      items:
+                        type: object
+                        properties:
+                          created_at:
+                            type: string
+                          value:
+                            type: number
+                        required:
+                          - created_at
+                          - value
+                  required:
+                    - id
+                    - values
+
  /configs/list_worker_groups:
    get:
      summary: list worker groups
@@ -9704,6 +9738,12 @@ components:
          type: string
        wm_version:
          type: string
+        current_job_id:
+          type: string
+        current_job_workspace_id:
+          type: string
+        occupancy_rate:
+          type: number
      required:
        - worker
        - worker_instance
@@ -14,14 +14,16 @@ use axum::{

 use serde::{Deserialize, Serialize};
 use sqlx::FromRow;
+use uuid::Uuid;
 use windmill_common::{
    db::UserDB,
    error::JsonResult,
    utils::{paginate, Pagination},
    worker::{ALL_TAGS, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE},
+    DB,
 };

-use crate::db::ApiAuthed;
+use crate::{db::ApiAuthed, utils::require_super_admin};

 pub fn global_service() -> Router {
    Router::new()
@@ -33,6 +35,7 @@ pub fn global_service() -> Router {
            get(get_default_tags_per_workspace),
        )
        .route("/get_default_tags", get(get_default_tags))
+        .route("/queue_metrics", get(get_queue_metrics))
 }

 #[derive(FromRow, Serialize, Deserialize)]
@@ -43,9 +46,12 @@ struct WorkerPing {
    started_at: chrono::DateTime<chrono::Utc>,
    ip: String,
    jobs_executed: i32,
+    current_job_id: Option<Uuid>,
+    current_job_workspace_id: Option<String>,
    custom_tags: Option<Vec<String>>,
    worker_group: String,
    wm_version: String,
+    occupancy_rate: Option<f32>,
 }

 #[derive(Serialize, Deserialize)]
@@ -62,21 +68,25 @@ pub struct ListWorkerQuery {

 async fn list_worker_pings(
    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
    Extension(user_db): Extension<UserDB>,
    Query(query): Query<ListWorkerQuery>,
 ) -> JsonResult<Vec<WorkerPing>> {
+    let is_super_admin = require_super_admin(&db, &authed.email).await.is_ok();
    let mut tx = user_db.begin(&authed).await?;

    let (per_page, offset) = paginate(Pagination { page: query.page, per_page: query.per_page });

    let rows = sqlx::query_as!(
        WorkerPing,
-        "SELECT worker, worker_instance,  EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags, worker_group, wm_version FROM worker_ping
-         WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)
-         ORDER BY ping_at desc LIMIT $2 OFFSET $3",
+        "SELECT worker, worker_instance,  EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as current_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as current_job_workspace_id, custom_tags, worker_group, wm_version, occupancy_rate
+        FROM worker_ping
+        WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)
+        ORDER BY ping_at desc LIMIT $2 OFFSET $3",
        query.ping_since,
        per_page as i64,
-        offset as i64
+        offset as i64,
+        is_super_admin
    )
    .fetch_all(&mut *tx)
    .await?;
@@ -118,3 +128,35 @@ async fn get_default_tags_per_workspace() -> JsonResult<bool> {
 async fn get_default_tags() -> JsonResult<Vec<String>> {
    Ok(Json(DEFAULT_TAGS.clone()))
 }
+
+#[derive(Serialize)]
+struct QueueMetric {
+    id: String,
+    values: Vec<serde_json::Value>,
+}
+
+async fn get_queue_metrics(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+) -> JsonResult<Vec<QueueMetric>> {
+    require_super_admin(&db, &authed.email).await?;
+
+    let queue_metrics = sqlx::query_as!(
+        QueueMetric,
+        "WITH queue_metrics as (
+            SELECT id, value, created_at
+            FROM metrics
+            WHERE id LIKE 'queue_%'
+                AND created_at > now() - interval '14 day'
+            ORDER BY created_at ASC
+        )
+        SELECT id, array_agg(json_build_object('value', value, 'created_at', created_at)) as \"values!\"
+        FROM queue_metrics
+        GROUP BY id
+        ORDER BY id ASC"
+    )
+    .fetch_all(&db)
+    .await?;
+
+    Ok(Json(queue_metrics))
+}
@@ -140,15 +140,54 @@ pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db:
                .map(|x| format!("{}:{}", x.workspace_id, x.path)),
        )
    };
+    let mut vcpus = std::process::Command::new("cat")
+        .args(["/sys/fs/cgroup/cpu.max"])
+        .output()
+        .ok()
+        .map(|o| {
+            String::from_utf8_lossy(&o.stdout)
+                .to_string()
+                .split(" ")
+                .map(|s| s.to_string())
+                .collect::<Vec<String>>()
+                .get(0)
+                .map(|s| s.to_string())
+        })
+        .flatten();
+
+    if vcpus.is_none() {
+        vcpus = std::process::Command::new("cat")
+            .args(["/sys/fs/cgroup/cpu/cpu.cfs_quota_us"])
+            .output()
+            .ok()
+            .map(|o| String::from_utf8_lossy(&o.stdout).to_string())
+    }
+
+    let mut memory = std::process::Command::new("cat")
+        .args(["/sys/fs/cgroup/memory.max"])
+        .output()
+        .ok()
+        .map(|o| String::from_utf8_lossy(&o.stdout).to_string());
+
+    if memory.is_none() {
+        memory = std::process::Command::new("cat")
+            .args(["/sys/fs/cgroup/memory/memory.limit_in_bytes"])
+            .output()
+            .ok()
+            .map(|o| String::from_utf8_lossy(&o.stdout).to_string())
+    }
+
    sqlx::query!(
-        "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
+        "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version, vcpus, memory) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
        worker_instance,
        worker_name,
        ip,
        tags.as_slice(),
        *WORKER_GROUP,
        dw,
-        crate::utils::GIT_VERSION
+        crate::utils::GIT_VERSION,
+        vcpus.map(|x| x.parse::<i64>().ok()).flatten(),
+        memory.map(|x| x.parse::<i64>().ok()).flatten()
    )
    .execute(db)
    .await
@@ -567,7 +567,9 @@ where
                i+=1;
                if i % 10 == 0 {
                    sqlx::query!(
-                        "UPDATE worker_ping SET ping_at = now() WHERE worker = $1",
+                        "UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2 WHERE worker = $3",
+                        &job_id,
+                        &w_id,
                        &worker_name
                    )
                    .execute(&db)
@@ -607,7 +607,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
        );
    }

-    #[cfg(feature = "prometheus")]
    let start_time = Instant::now();

    let worker_dir = format!("{TMP_DIR}/{worker_name}");
@@ -919,6 +918,8 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
            None
        };

+    let mut worker_code_execution_metric: f32 = 0.0;
+
    let mut jobs_executed = 0;

    #[cfg(feature = "prometheus")]
@@ -1387,14 +1388,12 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
            let tags = WORKER_CONFIG.read().await.worker_tags.clone();

            if let Err(e) = sqlx::query!(
-                "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2 WHERE worker = $3",
+                "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $4",
                jobs_executed,
                tags.as_slice(),
+                worker_code_execution_metric / start_time.elapsed().as_secs_f32(),
                &worker_name
-            )
-            .execute(db)
-            .await
-            {
+            ).execute(db).await {
                tracing::error!("failed to update worker ping, exiting: {}", e);
                killpill_tx.send(()).unwrap_or_default();
            }
@@ -1696,6 +1695,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
                        base_internal_url,
                        rsmq.clone(),
                        job_completed_tx.clone(),
+                        &mut worker_code_execution_metric,
                        worker_flow_initial_transition_duration.clone(),
                        worker_code_execution_duration.clone(),
                    )
@@ -2603,6 +2603,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
    base_internal_url: &str,
    rsmq: Option<R>,
    job_completed_tx: JobCompletedSender,
+    worker_code_execution_metric: &mut f32,
    _worker_flow_initial_transition_duration: Option<Histo>,
    _worker_code_execution_duration: Option<Histo>,
 ) -> windmill_common::error::Result<()> {
@@ -2829,6 +2830,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
            _ => {
                #[cfg(feature = "prometheus")]
                let timer = _worker_code_execution_duration.map(|x| x.start_timer());
+                let metric_timer = Instant::now();
                let r = handle_code_execution_job(
                    job.as_ref(),
                    db,
@@ -2842,6 +2844,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
                    &mut column_order,
                )
                .await;
+                *worker_code_execution_metric += metric_timer.elapsed().as_secs_f32();
                #[cfg(feature = "prometheus")]
                timer.map(|x| x.stop_and_record());
                r
@@ -0,0 +1,249 @@
+<script lang="ts">
+	import { Drawer, DrawerContent } from './common'
+	import 'chartjs-adapter-date-fns'
+	import { Line } from 'svelte-chartjs'
+
+	import {
+		Chart as ChartJS,
+		Title,
+		Tooltip,
+		Legend,
+		LineElement,
+		CategoryScale,
+		LinearScale,
+		PointElement,
+		LogarithmicScale,
+		TimeScale,
+		type ChartData,
+		type Point
+	} from 'chart.js'
+	import { WorkerService } from '$lib/gen'
+	import { onDestroy } from 'svelte'
+	import { superadmin } from '$lib/stores'
+	export let drawer: Drawer
+
+	const colorTuples = [
+		['#7EB26D', 'rgba(126, 178, 109, 0.2)'],
+		['#EAB839', 'rgba(234, 184, 57, 0.2)'],
+		['#6ED0E0', 'rgba(110, 208, 224, 0.2)'],
+		['#EF843C', 'rgba(239, 132, 60, 0.2)'],
+		['#E24D42', 'rgba(226, 77, 66, 0.2)'],
+		['#1F78C1', 'rgba(31, 120, 193, 0.2)'],
+		['#BA43A9', 'rgba(186, 67, 169, 0.2)'],
+		['#705DA0', 'rgba(112, 93, 160, 0.2)'],
+		['#508642', 'rgba(80, 134, 66, 0.2)'],
+		['#CCA300', 'rgba(204, 163, 0, 0.2)'],
+		['#447EBC', 'rgba(68, 126, 188, 0.2)'],
+		['#C15C17', 'rgba(193, 92, 23, 0.2)'],
+		['#890F02', 'rgba(137, 15, 2, 0.2)'],
+		['#666666', 'rgba(102, 102, 102, 0.2)'],
+		['#44AA99', 'rgba(68, 170, 153, 0.2)'],
+		['#6D8764', 'rgba(109, 135, 100, 0.2)'],
+		['#555555', 'rgba(85, 85, 85, 0.2)'],
+		['#B3B3B3', 'rgba(179, 179, 179, 0.2)'],
+		['#008C9E', 'rgba(0, 140, 158, 0.2)'],
+		['#6BBA70', 'rgba(107, 186, 112, 0.2)']
+	]
+
+	function getColors(labels: string[]) {
+		const colors = labels.map((_, i) => colorTuples[i % colorTuples.length])
+		return Object.fromEntries(colors.map((c, i) => [labels[i], c]))
+	}
+
+	ChartJS.register(
+		Title,
+		Tooltip,
+		Legend,
+		LineElement,
+		LinearScale,
+		PointElement,
+		CategoryScale,
+		TimeScale,
+		LogarithmicScale
+	)
+
+	let countData: ChartData<'line', Point[], undefined> | undefined = undefined
+	let delayData: ChartData<'line', Point[], undefined> | undefined = undefined
+
+	let minDate = new Date()
+
+	let noMetrics = false
+
+	function fillData(
+		data: {
+			created_at: string
+			value: number
+		}[],
+		zero: number = 0
+	) {
+		let last = -1
+		const newElements: typeof data = []
+		for (const el of [
+			...data,
+			{
+				created_at: new Date().toISOString(),
+				value: zero
+			}
+		]) {
+			const currentTs = new Date(el.created_at).getTime()
+			if (last > -1 && currentTs - last > 1000 * 60 * 2) {
+				const numElements = Math.floor((currentTs - last) / (1000 * 30))
+				for (let i = 1; i < numElements; i++) {
+					newElements.push({
+						created_at: new Date(last + i * (1000 * 30)).toISOString(),
+						value: zero
+					})
+				}
+			}
+			last = currentTs
+		}
+
+		return [...data, ...newElements].sort(
+			(a, b) => new Date(a.created_at).getTime() - new Date(b.created_at).getTime()
+		)
+	}
+
+	async function loadMetrics() {
+		let metrics = await WorkerService.getQueueMetrics()
+
+		if (metrics.length == 0) {
+			noMetrics = true
+			return
+		}
+
+		const labels = metrics
+			.map((m) => m.id.slice(12))
+			.filter((v, i, a) => a.indexOf(v) === i)
+			.sort()
+		const labelColors = getColors(labels)
+
+		countData = {
+			datasets: metrics
+				.filter((m) => m.id.startsWith('queue_count_'))
+				.map((m) => {
+					const [color, bgColor] = labelColors[m.id.slice(12)]
+					return {
+						label: m.id.slice(12),
+						backgroundColor: bgColor,
+						borderColor: color,
+						data: fillData(m.values).map((v) => ({ x: v.created_at as any, y: v.value }))
+					}
+				})
+		}
+
+		delayData = {
+			datasets: metrics
+				.filter((m) => m.id.startsWith('queue_delay_'))
+				.map((m) => {
+					const [color, bgColor] = labelColors[m.id.slice(12)]
+					return {
+						label: m.id.slice(12),
+						borderColor: color,
+						backgroundColor: bgColor,
+						data: fillData(m.values, 1).map((v) => ({ x: v.created_at as any, y: v.value }))
+					}
+				})
+		}
+
+		minDate = new Date(Math.min(...countData.datasets.map((d) => new Date(d.data[0].x).getTime())))
+	}
+
+	let interval: NodeJS.Timeout | undefined = undefined
+
+	$: if ($superadmin) {
+		loadMetrics()
+		if (interval) {
+			clearInterval(interval)
+		}
+		interval = setInterval(loadMetrics, 35000)
+	} else {
+		if (interval) {
+			clearInterval(interval)
+		}
+	}
+
+	onDestroy(() => {
+		clearInterval(interval)
+	})
+</script>
+
+<Drawer bind:this={drawer}>
+	<DrawerContent title="Queue Metrics" on:close={drawer.closeDrawer}>
+		{#if noMetrics}
+			<p class="text-secondary">No jobs delayed by more than 3 seconds in the last 14 days</p>
+		{:else}
+			<div class="flex flex-col gap-4">
+				{#if countData}
+					<Line
+						data={countData}
+						options={{
+							plugins: {
+								title: {
+									display: true,
+									text: 'Number of delayed jobs per tag (> 3s)'
+								}
+							},
+							scales: {
+								x: {
+									type: 'time',
+									min: minDate.toISOString(),
+									max: new Date().toISOString()
+								},
+								y: {
+									title: {
+										display: true,
+										text: 'count'
+									}
+								}
+							}
+						}}
+					/>
+				{/if}
+				{#if delayData}
+					<Line
+						data={delayData}
+						options={{
+							plugins: {
+								title: {
+									display: true,
+									text: 'Queue delay per tag (> 3s)'
+								},
+								tooltip: {
+									callbacks: {
+										label: function (context) {
+											// @ts-ignore
+											if (context.raw.y === 1) {
+												return context.dataset.label + ': 0'
+											} else {
+												// @ts-ignore
+												return context.dataset.label + ': ' + context.raw.y
+											}
+										}
+									}
+								}
+							},
+							scales: {
+								x: {
+									type: 'time',
+									min: minDate.toISOString(),
+									max: new Date().toISOString()
+								},
+
+								y: {
+									type: 'logarithmic',
+									title: {
+										display: true,
+										text: 'delay (s)'
+									},
+									ticks: {
+										callback: (value, _) => (value === 1 ? '0' : value)
+									}
+								}
+							}
+						}}
+					/>
+				{/if}
+			</div>
+		{/if}
+	</DrawerContent>
+</Drawer>
@@ -7,6 +7,7 @@
 	import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
 	import DefaultTags from '$lib/components/DefaultTags.svelte'
 	import PageHeader from '$lib/components/PageHeader.svelte'
+	import QueueMetricsDrawer from '$lib/components/QueueMetricsDrawer.svelte'
 	import SimpleEditor from '$lib/components/SimpleEditor.svelte'
 	import Cell from '$lib/components/table/Cell.svelte'
 	import DataTable from '$lib/components/table/DataTable.svelte'
@@ -17,7 +18,7 @@
 	import { enterpriseLicense, superadmin } from '$lib/stores'
 	import { sendUserToast } from '$lib/toast'
 	import { displayDate, groupBy, truncate } from '$lib/utils'
-	import { AlertTriangle, FileJson, Plus } from 'lucide-svelte'
+	import { AlertTriangle, FileJson, LineChart, Plus } from 'lucide-svelte'
 	import { onDestroy, onMount } from 'svelte'
 	import YAML from 'yaml'

@@ -160,8 +161,12 @@

 		await loadWorkerGroups()
 	}
+
+	let queueMetricsDrawer: Drawer
 </script>

+<QueueMetricsDrawer bind:drawer={queueMetricsDrawer} />
+
 <Drawer bind:this={importConfigDrawer} size="800px">
 	<DrawerContent
 		title="Import groups config from YAML"
@@ -197,6 +202,20 @@
 				<div>
 					<DefaultTags bind:defaultTagPerWorkspace />
 				</div>
+				<div>
+					<Button
+						size="xs"
+						color="dark"
+						startIcon={{
+							icon: LineChart
+						}}
+						on:click={() => {
+							queueMetricsDrawer?.openDrawer()
+						}}
+					>
+						Queue metrics
+					</Button>
+				</div>
 			</div>
 		{/if}
 	</PageHeader>
@@ -289,10 +308,11 @@
 			{/if}</div
 		>
 		{#each groupedWorkers as worker_group (worker_group[0])}
+			{@const config = (workerGroups ?? {})[worker_group[0]]}
 			<WorkspaceGroup
 				{customTags}
 				name={worker_group[0]}
-				config={(workerGroups ?? {})[worker_group[0]]}
+				{config}
 				on:reload={() => {
 					loadWorkerGroups()
 				}}
@@ -319,6 +339,10 @@
 						<Cell head>Last ping</Cell>
 						<Cell head>Worker start</Cell>
 						<Cell head>Nb of jobs executed</Cell>
+						{#if (!config || config?.dedicated_worker == undefined) && $superadmin}
+							<Cell head>Current job</Cell>
+							<Cell head>Occupancy rate</Cell>
+						{/if}
 						<Cell head>Version</Cell>
 						<Cell head last>Liveness</Cell>
 					</tr>
@@ -328,7 +352,7 @@
 						<tr class="border-t">
 							<Cell
 								first
-								colspan="7"
+								colspan={(!config || config?.dedicated_worker == undefined) && $superadmin ? 9 : 7}
 								scope="colgroup"
 								class="bg-surface-secondary/60 py-2 border-b"
 							>
@@ -341,7 +365,7 @@
 						</tr>

 						{#if workers}
-							{#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed, wm_version }}
+							{#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed, current_job_id, current_job_workspace_id, occupancy_rate, wm_version }}
 								<tr>
 									<Cell first>{worker}</Cell>
 									<Cell>
@@ -355,6 +379,21 @@
 									<Cell>{last_ping != undefined ? last_ping + timeSinceLastPing : -1}s ago</Cell>
 									<Cell>{displayDate(started_at)}</Cell>
 									<Cell>{jobs_executed}</Cell>
+									{#if (!config || config?.dedicated_worker == undefined) && $superadmin}
+										<Cell>
+											{#if current_job_id}
+												<a href={`/run/${current_job_id}?workspace=${current_job_workspace_id}`}>
+													View job
+												</a>
+												(workspace {current_job_workspace_id})
+											{:else}
+												None
+											{/if}
+										</Cell>
+										<Cell>
+											{Math.ceil(occupancy_rate ?? 0 * 100)}%
+										</Cell>
+									{/if}
 									<Cell
 										><div class="!text-2xs"
 											>{wm_version.split('-')[0]}<Tooltip>{wm_version}</Tooltip></div