feat: add worker_group_job_stats table for efficient job metrics aggregation (#6527)

* feat: add worker_group_job_stats table for job metrics aggregation

- Add new table with hour timestamp, worker group, script lang, workspace_id, job count and total duration
- Workers accumulate stats in memory and update hourly via sum aggregation
- Monitor.rs cleans up rows older than 60 days periodically
- Stats are flushed on worker shutdown to prevent data loss

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

* all

* all

* ee-repo-ref

* nits

* nits

---------

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 <ruben@windmill.dev>

backend/ee-repo-ref.txt

@@ -1 +1 @@
-b72d44c2a8a2ad3a1e36c938863a9b02891ac4a2
+7b62b1663930bc0c0cb751804cfa38883b2ed6e8

backend/migrations/20250903124849_worker_group_job_stats.down.sql

@@ -0,0 +1,2 @@
+-- Drop worker_group_job_stats table and its indices
+DROP TABLE IF EXISTS worker_group_job_stats;

backend/migrations/20250903124849_worker_group_job_stats.up.sql

@@ -0,0 +1,15 @@
+-- Add worker_group_job_stats table
+CREATE TABLE IF NOT EXISTS worker_group_job_stats (
+    hour BIGINT NOT NULL,
+    worker_group TEXT NOT NULL,
+    script_lang VARCHAR(50),
+    workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE,
+    job_count INTEGER NOT NULL DEFAULT 0,
+    total_duration_ms BIGINT NOT NULL DEFAULT 0,
+    PRIMARY KEY (hour, worker_group, script_lang, workspace_id)
+);
+
+-- Create indices for efficient querying
+CREATE INDEX IF NOT EXISTS worker_group_job_stats_hour_idx ON worker_group_job_stats(hour DESC);
+CREATE INDEX IF NOT EXISTS worker_group_job_stats_workspace_idx ON worker_group_job_stats(workspace_id, hour DESC);
+CREATE INDEX IF NOT EXISTS worker_group_job_stats_worker_group_idx ON worker_group_job_stats(worker_group, hour DESC);

backend/src/monitor.rs

@@ -1531,6 +1531,22 @@ pub async fn monitor_db(
             }
         }
     };
+    // run every hour (60 minutes / 30 seconds = 120)
+    let cleanup_worker_group_stats_f = async {
+        if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(120) {
+            if let Some(db) = conn.as_sql() {
+                match windmill_common::worker_group_job_stats::cleanup_old_stats(db, 60).await {
+                    Ok(count) if count > 0 => {
+                        tracing::info!("Deleted {} old worker group job stats rows", count);
+                    }
+                    Err(e) => {
+                        tracing::error!("Error cleaning up worker group job stats: {:?}", e);
+                    }
+                    _ => {}
+                }
+            }
+        }
+    };
 
     // run every hour
     let vacuum_queue_f = async {
@@ -1633,6 +1649,7 @@ pub async fn monitor_db(
         update_min_worker_version_f,
         cleanup_concurrency_counters_f,
         cleanup_concurrency_counters_empty_keys_f,
+        cleanup_worker_group_stats_f,
     );
 }
 

backend/windmill-common/src/lib.rs

@@ -84,6 +84,7 @@ pub mod users;
 pub mod utils;
 pub mod variables;
 pub mod worker;
+pub mod worker_group_job_stats;
 pub mod workspaces;
 
 pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;

backend/windmill-common/src/worker_group_job_stats.rs

@@ -0,0 +1,112 @@
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+use sqlx::{Pool, Postgres};
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+use crate::scripts::ScriptLang;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobStatsAccumulator {
+    pub worker_group: String,
+    pub script_lang: Option<ScriptLang>,
+    pub workspace_id: String,
+    pub job_count: i32,
+    pub total_duration_ms: i64,
+}
+
+pub type JobStatsMap =
+    Arc<RwLock<HashMap<(i64, String, Option<ScriptLang>, String), JobStatsAccumulator>>>;
+
+pub fn get_current_hour() -> i64 {
+    let now = Utc::now();
+    (now.timestamp() / 3600) * 3600
+}
+
+pub async fn accumulate_job_stats(
+    stats_map: &JobStatsMap,
+    worker_group: &str,
+    script_lang: Option<ScriptLang>,
+    workspace_id: &str,
+    duration_ms: i64,
+) {
+    let hour = get_current_hour();
+    let key = (
+        hour,
+        worker_group.to_string(),
+        script_lang.clone(),
+        workspace_id.to_string(),
+    );
+
+    let mut stats = stats_map.write().await;
+    let entry = stats
+        .entry(key.clone())
+        .or_insert_with(|| JobStatsAccumulator {
+            worker_group: worker_group.to_string(),
+            script_lang,
+            workspace_id: workspace_id.to_string(),
+            job_count: 0,
+            total_duration_ms: 0,
+        });
+
+    entry.job_count += 1;
+    entry.total_duration_ms += duration_ms;
+}
+
+pub async fn flush_stats_to_db(
+    db: &Pool<Postgres>,
+    stats_map: &JobStatsMap,
+) -> Result<(), sqlx::Error> {
+    let current_stats: Vec<(
+        (i64, String, Option<ScriptLang>, String),
+        JobStatsAccumulator,
+    )> = {
+        let mut stats = stats_map.write().await;
+        if stats.is_empty() {
+            return Ok(());
+        }
+        stats.drain().collect()
+    };
+
+    for ((hour, _worker_group, script_lang, _workspace_id), accumulator) in current_stats {
+        let script_lang_str = script_lang.as_ref().map(|l| l.as_str());
+
+        // Use ON CONFLICT to sum existing values
+        sqlx::query(
+            r#"
+            INSERT INTO worker_group_job_stats 
+            (hour, worker_group, script_lang, workspace_id, job_count, total_duration_ms)
+            VALUES ($1, $2, $3, $4, $5, $6)
+            ON CONFLICT (hour, worker_group, script_lang, workspace_id)
+            DO UPDATE SET
+                job_count = worker_group_job_stats.job_count + EXCLUDED.job_count,
+                total_duration_ms = worker_group_job_stats.total_duration_ms + EXCLUDED.total_duration_ms
+            "#,
+        )
+        .bind(hour)
+        .bind(&accumulator.worker_group)
+        .bind(script_lang_str)
+        .bind(&accumulator.workspace_id)
+        .bind(accumulator.job_count)
+        .bind(accumulator.total_duration_ms)
+        .execute(db)
+        .await?;
+    }
+
+    Ok(())
+}
+
+pub async fn cleanup_old_stats(
+    db: &Pool<Postgres>,
+    retention_days: i64,
+) -> Result<u64, sqlx::Error> {
+    let cutoff_timestamp = get_current_hour() - (retention_days * 24 * 3600);
+
+    let result = sqlx::query("DELETE FROM worker_group_job_stats WHERE hour < $1")
+        .bind(cutoff_timestamp)
+        .execute(db)
+        .await?;
+
+    Ok(result.rows_affected())
+}

backend/windmill-worker/src/result_processor.rs

@@ -22,6 +22,7 @@ use windmill_common::{
     jobs::JobKind,
     utils::WarnAfterExt,
     worker::{to_raw_value, Connection, WORKER_GROUP},
+    worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap},
     KillpillSender, DB,
 };
 
@@ -63,6 +64,7 @@ async fn process_jc(
     worker_dir: &str,
     same_worker_tx: Option<&SameWorkerSender>,
     job_completed_sender: &JobCompletedSender,
+    stats_map: &JobStatsMap,
     #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
 ) {
     let success: bool = jc.success;
@@ -139,6 +141,11 @@ async fn process_jc(
         }
     }
 
+    // Extract stats info before moving jc
+    let duration_ms = jc.duration.clone();
+    let script_lang = jc.job.script_lang.clone();
+    let workspace_id = jc.job.workspace_id.clone();
+
     let root_job = handle_receive_completed_job(
         jc,
         &base_internal_url,
@@ -156,6 +163,18 @@ async fn process_jc(
     if let Some(root_job) = root_job {
         add_root_flow_job_to_otlp(&root_job, success);
     }
+
+    // Accumulate job stats if duration is available
+    if let Some(duration_ms) = duration_ms {
+        accumulate_job_stats(
+            stats_map,
+            &*WORKER_GROUP,
+            script_lang,
+            &workspace_id,
+            duration_ms,
+        )
+        .await;
+    }
 }
 
 enum JobCompletedRx {
@@ -178,6 +197,7 @@ pub fn start_background_processor(
     worker_name: String,
     killpill_tx: KillpillSender,
     is_dedicated_worker: bool,
+    stats_map: JobStatsMap,
 ) -> JoinHandle<()> {
     tokio::spawn(async move {
         let mut has_been_killed = false;
@@ -187,6 +207,28 @@ pub fn start_background_processor(
         #[cfg(feature = "benchmark")]
         let mut infos = BenchmarkInfo::new();
 
+        // Start periodic stats flush task
+        let db_clone = db.clone();
+        let stats_map_clone = stats_map.clone();
+        let mut killpill_rx_clone = killpill_rx.resubscribe();
+        let flush_handle = tokio::spawn(async move {
+            let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(900)); // Flush every 15 min
+
+            loop {
+                tokio::select! {
+                    _ = interval.tick() => {
+                        if let Err(e) = flush_stats_to_db(&db_clone, &stats_map_clone).await {
+                            tracing::error!("Failed to flush worker group job stats: {}", e);
+                        }
+                    }
+                    _ = killpill_rx_clone.recv() => {
+                        tracing::info!("bg processor received killpill signal, flushing remaining stats");
+                        break;
+                    }
+                }
+            }
+        });
+
         //if we have been killed, we want to drain the queue of jobs
         while let Some(sr) = {
             if has_been_killed {
@@ -241,6 +283,7 @@ pub fn start_background_processor(
                         &worker_dir,
                         Some(&same_worker_tx),
                         &job_completed_sender,
+                        &stats_map,
                         #[cfg(feature = "benchmark")]
                         &mut bench,
                     )
@@ -323,6 +366,16 @@ pub fn start_background_processor(
             }
         }
 
+        // Flush any remaining stats before shutting down
+        tracing::info!("flushing remaining stats before shutting down");
+        let flush_result =
+            tokio::time::timeout(std::time::Duration::from_secs(10), flush_handle).await;
+        match flush_result {
+            Ok(Ok(())) => tracing::info!("Stats flushed successfully"),
+            Ok(Err(join_err)) => tracing::error!("Stats flush task failed: {}", join_err),
+            Err(_) => tracing::error!("Stats flush timed out after 10 seconds"),
+        }
+
         job_completed_processor_is_done.store(true, Ordering::SeqCst);
 
         tracing::info!("finished processing all completed jobs");

backend/windmill-worker/src/worker.rs

@@ -26,6 +26,7 @@ use windmill_common::{
         make_pull_query, write_file, Connection, HttpClient, MAX_TIMEOUT,
         MIN_PERIODIC_SCRIPT_INTERVAL_SECONDS, ROOT_CACHE_DIR, ROOT_CACHE_NOMOUNT_DIR, TMP_DIR,
     },
+    worker_group_job_stats::JobStatsMap,
     KillpillSender,
 };
 
@@ -1218,6 +1219,7 @@ pub async fn run_worker(
 
     // This is used to wake up the background processor when main loop is done and just waiting for new same workers jobs, and that bg processor is also not processing any jobs, bg processing can exit if no more same worker jobs
     let wake_up_notify = Arc::new(tokio::sync::Notify::new());
+    let stats_map = JobStatsMap::default();
     let send_result = match (conn, job_completed_rx) {
         (Connection::Sql(db), Some(job_completed_receiver)) => Some(start_background_processor(
             job_completed_receiver,
@@ -1233,6 +1235,7 @@ pub async fn run_worker(
             worker_name.clone(),
             killpill_tx.clone(),
             is_dedicated_worker,
+            stats_map,
         )),
         _ => None,
     };