diff --git a/backend/.sqlx/query-1ae4aab40783b4cd588e5a4da57eabca765789cd706ad682e8094c8960bc46a8.json b/backend/.sqlx/query-1ae4aab40783b4cd588e5a4da57eabca765789cd706ad682e8094c8960bc46a8.json new file mode 100644 index 0000000000..6ae002d4f7 --- /dev/null +++ b/backend/.sqlx/query-1ae4aab40783b4cd588e5a4da57eabca765789cd706ad682e8094c8960bc46a8.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM runnable_job_stats s WHERE s.hour >= $1 AND NOT EXISTS (SELECT 1 FROM workspace w WHERE w.id = s.workspace_id)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [] + }, + "hash": "1ae4aab40783b4cd588e5a4da57eabca765789cd706ad682e8094c8960bc46a8" +} diff --git a/backend/.sqlx/query-24cfd719a45af5beb4eb264939290c01e1b684525bae4cb329515989cd341875.json b/backend/.sqlx/query-24cfd719a45af5beb4eb264939290c01e1b684525bae4cb329515989cd341875.json new file mode 100644 index 0000000000..eaa7d60dea --- /dev/null +++ b/backend/.sqlx/query-24cfd719a45af5beb4eb264939290c01e1b684525bae4cb329515989cd341875.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO runnable_job_stats\n (hour, workspace_id, runnable_path, worker_group, job_count, total_duration_ms,\n max_memory_peak, sum_memory_peak, memory_sample_count, total_cpu_ms)\n SELECT s.hour, s.workspace_id, s.runnable_path, s.worker_group, s.job_count,\n s.total_duration_ms, s.max_memory_peak, s.sum_memory_peak,\n s.memory_sample_count, s.total_cpu_ms\n FROM UNNEST($1::bigint[], $2::text[], $3::text[], $4::text[], $5::int[], $6::bigint[],\n $7::int[], $8::bigint[], $9::int[], $10::bigint[])\n AS s(hour, workspace_id, runnable_path, worker_group, job_count, total_duration_ms,\n max_memory_peak, sum_memory_peak, memory_sample_count, total_cpu_ms)\n JOIN workspace w ON w.id = s.workspace_id\n ORDER BY s.hour, s.workspace_id, s.runnable_path, s.worker_group\n ON CONFLICT (hour, workspace_id, runnable_path, worker_group) DO UPDATE SET\n job_count = runnable_job_stats.job_count + EXCLUDED.job_count,\n total_duration_ms = runnable_job_stats.total_duration_ms + EXCLUDED.total_duration_ms,\n max_memory_peak = GREATEST(runnable_job_stats.max_memory_peak, EXCLUDED.max_memory_peak),\n sum_memory_peak = runnable_job_stats.sum_memory_peak + EXCLUDED.sum_memory_peak,\n memory_sample_count = runnable_job_stats.memory_sample_count + EXCLUDED.memory_sample_count,\n total_cpu_ms = runnable_job_stats.total_cpu_ms + EXCLUDED.total_cpu_ms\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8Array", + "TextArray", + "TextArray", + "TextArray", + "Int4Array", + "Int8Array", + "Int4Array", + "Int8Array", + "Int4Array", + "Int8Array" + ] + }, + "nullable": [] + }, + "hash": "24cfd719a45af5beb4eb264939290c01e1b684525bae4cb329515989cd341875" +} diff --git a/backend/.sqlx/query-97d8a9835344b8c72d59afc99d76b681c5120ef29eb09b8626a3fbc43a32cc0e.json b/backend/.sqlx/query-97d8a9835344b8c72d59afc99d76b681c5120ef29eb09b8626a3fbc43a32cc0e.json new file mode 100644 index 0000000000..abf8b56d79 --- /dev/null +++ b/backend/.sqlx/query-97d8a9835344b8c72d59afc99d76b681c5120ef29eb09b8626a3fbc43a32cc0e.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM runnable_job_stats WHERE hour < $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [] + }, + "hash": "97d8a9835344b8c72d59afc99d76b681c5120ef29eb09b8626a3fbc43a32cc0e" +} diff --git a/backend/.sqlx/query-c5971326bdbfaa7dff74689d1365db4f26c41fa56b326dbd02e604c793a13477.json b/backend/.sqlx/query-c5971326bdbfaa7dff74689d1365db4f26c41fa56b326dbd02e604c793a13477.json new file mode 100644 index 0000000000..d84578d37c --- /dev/null +++ b/backend/.sqlx/query-c5971326bdbfaa7dff74689d1365db4f26c41fa56b326dbd02e604c793a13477.json @@ -0,0 +1,74 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT workspace_id, runnable_path, worker_group,\n SUM(job_count)::bigint AS \"job_count!\",\n SUM(total_duration_ms)::bigint AS \"total_duration_ms!\",\n MAX(max_memory_peak) AS \"max_memory_peak!\",\n SUM(sum_memory_peak)::bigint AS \"sum_memory_peak!\",\n SUM(memory_sample_count)::bigint AS \"memory_sample_count!\",\n SUM(total_cpu_ms)::bigint AS \"total_cpu_ms!\"\n FROM runnable_job_stats\n WHERE hour >= $1\n AND ($2::text IS NULL OR worker_group = $2)\n AND ($3::text IS NULL OR workspace_id = $3)\n GROUP BY workspace_id, runnable_path, worker_group\n ORDER BY CASE $4::text\n WHEN 'max_memory' THEN MAX(max_memory_peak)::numeric\n WHEN 'cpu' THEN SUM(total_cpu_ms)\n ELSE SUM(total_duration_ms)\n END DESC, workspace_id, runnable_path, worker_group\n LIMIT $5\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "runnable_path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "worker_group", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "job_count!", + "type_info": "Int8" + }, + { + "ordinal": 4, + "name": "total_duration_ms!", + "type_info": "Int8" + }, + { + "ordinal": 5, + "name": "max_memory_peak!", + "type_info": "Int4" + }, + { + "ordinal": 6, + "name": "sum_memory_peak!", + "type_info": "Int8" + }, + { + "ordinal": 7, + "name": "memory_sample_count!", + "type_info": "Int8" + }, + { + "ordinal": 8, + "name": "total_cpu_ms!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text", + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + false, + false, + false, + null, + null, + null, + null, + null, + null + ] + }, + "hash": "c5971326bdbfaa7dff74689d1365db4f26c41fa56b326dbd02e604c793a13477" +} diff --git a/backend/.sqlx/query-d2fbaaaea6c542342862d8b535673d19dc4ca721ab656a39330bad5ea48999d8.json b/backend/.sqlx/query-d2fbaaaea6c542342862d8b535673d19dc4ca721ab656a39330bad5ea48999d8.json new file mode 100644 index 0000000000..55b335fba0 --- /dev/null +++ b/backend/.sqlx/query-d2fbaaaea6c542342862d8b535673d19dc4ca721ab656a39330bad5ea48999d8.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM runnable_job_stats WHERE workspace_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "d2fbaaaea6c542342862d8b535673d19dc4ca721ab656a39330bad5ea48999d8" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index cc3c1a826c..22d527d5f5 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -017710bdd0a10de857f0f40800dde238cdb1ad24 +a1c00d73a05f227a61d2bf37cc86cd9f5bb1777a diff --git a/backend/migrations/20261008165042_runnable_job_stats.down.sql b/backend/migrations/20261008165042_runnable_job_stats.down.sql new file mode 100644 index 0000000000..21208736b4 --- /dev/null +++ b/backend/migrations/20261008165042_runnable_job_stats.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS runnable_job_stats; diff --git a/backend/migrations/20261008165042_runnable_job_stats.up.sql b/backend/migrations/20261008165042_runnable_job_stats.up.sql new file mode 100644 index 0000000000..cf676a037e --- /dev/null +++ b/backend/migrations/20261008165042_runnable_job_stats.up.sql @@ -0,0 +1,17 @@ +-- Hourly resource rollup per runnable and worker group, fed by the workers. +-- Memory figures are in kB, like v2_job_completed.memory_peak. +-- No foreign key to workspace: every worker upserts here, and the key would cost each row +-- a lookup and a lock on its workspace. Workspace deletion removes the rows itself. +CREATE TABLE IF NOT EXISTS runnable_job_stats ( + hour BIGINT NOT NULL, + workspace_id VARCHAR(50) NOT NULL, + runnable_path VARCHAR(255) NOT NULL, + worker_group TEXT NOT NULL, + job_count INTEGER NOT NULL DEFAULT 0, + total_duration_ms BIGINT NOT NULL DEFAULT 0, + max_memory_peak INTEGER NOT NULL DEFAULT 0, + sum_memory_peak BIGINT NOT NULL DEFAULT 0, + memory_sample_count INTEGER NOT NULL DEFAULT 0, + total_cpu_ms BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (hour, workspace_id, runnable_path, worker_group) +); diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 8d0af4915d..8de85baa24 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -4136,6 +4136,20 @@ pub async fn monitor_db( } _ => {} } + match windmill_common::runnable_job_stats::cleanup_old_runnable_job_stats( + db, + windmill_common::runnable_job_stats::RETENTION_DAYS, + ) + .await + { + Ok(count) if count > 0 => { + tracing::info!("Deleted {} old runnable job stats rows", count); + } + Err(e) => { + tracing::error!("Error cleaning up runnable job stats: {:?}", e); + } + _ => {} + } } } }; diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 284a822b47..ac07dc06d6 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -180,6 +180,7 @@ resource_type: workspace_id(char), name(char), schema(jsonb), description(text), FK: (workspace_id) -> workspace(id) resume_job: id(uuid), job(uuid), flow(uuid), created_at(ts), value(jsonb), approver(char), resume_id(int), approved(bool) FK: (flow) -> v2_job_queue(id) +runnable_job_stats: hour(bigint), workspace_id(char), runnable_path(char), worker_group(text), job_count(int), total_duration_ms(bigint), max_memory_peak(int), sum_memory_peak(bigint), memory_sample_count(int), total_cpu_ms(bigint) runnable_settings: hash(bigint), debouncing_settings(bigint), concurrency_settings(bigint) schedule: workspace_id(char), path(char), edited_by(char), edited_at(ts), schedule(char), enabled(bool), script_path(char), args(jsonb), extra_perms(jsonb), is_flow(bool), email(char), error(text), timezone(char), on_failure(char), on_recovery(char), on_failure_times(int), on_failure_exact(bool), on_failure_extra_args(jsonb), on_recovery_times(int), on_recovery_extra_args(jsonb), ws_error_handler_muted(bool), retry(jsonb), summary(char), no_flow_overlap(bool), tag(char), paused_until(ts), on_success(char), on_success_extra_args(jsonb), cron_version(text), description(text), dynamic_skip(char), labels(text[]) FK: (workspace_id) -> workspace(id) diff --git a/backend/windmill-api-workers/src/lib.rs b/backend/windmill-api-workers/src/lib.rs index cf0b85609a..8ebee2a967 100644 --- a/backend/windmill-api-workers/src/lib.rs +++ b/backend/windmill-api-workers/src/lib.rs @@ -41,6 +41,7 @@ pub fn global_service() -> Router { .route("/queue_metrics", get(get_queue_metrics)) .route("/queue_metrics_series", get(get_queue_metrics_series)) .route("/queue_status", get(get_queue_status)) + .route("/runnable_stats", get(get_runnable_stats)) .route("/queue_counts", get(get_queue_counts)) .route("/queue_running_counts", get(get_queue_running_counts)) .route( @@ -299,6 +300,81 @@ async fn get_queue_metrics( Ok(Json(queue_metrics)) } +#[derive(Deserialize)] +struct RunnableStatsQuery { + window_secs: Option, + worker_group: Option, + workspace: Option, + order_by: Option, + limit: Option, +} + +#[derive(Serialize)] +struct RunnableStats { + workspace_id: String, + runnable_path: String, + worker_group: String, + job_count: i64, + total_duration_ms: i64, + max_memory_peak: i32, + sum_memory_peak: i64, + memory_sample_count: i64, + total_cpu_ms: i64, +} + +async fn get_runnable_stats( + authed: ApiAuthed, + Extension(db): Extension, + Query(query): Query, +) -> JsonResult> { + require_devops_role(&db, &authed).await?; + + let window = query.window_secs.unwrap_or(24 * 3600).clamp( + 3600, + windmill_common::runnable_job_stats::RETENTION_DAYS * 24 * 3600, + ); + // Rows are hourly buckets keyed by their start, so the bucket the window begins in + // is included whole. + let from_hour = (chrono::Utc::now().timestamp() - window) / 3600 * 3600; + let order_by = match query.order_by.as_deref() { + Some("max_memory") => "max_memory", + Some("cpu") => "cpu", + _ => "total_duration", + }; + + let rows = sqlx::query_as!( + RunnableStats, + r#" + SELECT workspace_id, runnable_path, worker_group, + SUM(job_count)::bigint AS "job_count!", + SUM(total_duration_ms)::bigint AS "total_duration_ms!", + MAX(max_memory_peak) AS "max_memory_peak!", + SUM(sum_memory_peak)::bigint AS "sum_memory_peak!", + SUM(memory_sample_count)::bigint AS "memory_sample_count!", + SUM(total_cpu_ms)::bigint AS "total_cpu_ms!" + FROM runnable_job_stats + WHERE hour >= $1 + AND ($2::text IS NULL OR worker_group = $2) + AND ($3::text IS NULL OR workspace_id = $3) + GROUP BY workspace_id, runnable_path, worker_group + ORDER BY CASE $4::text + WHEN 'max_memory' THEN MAX(max_memory_peak)::numeric + WHEN 'cpu' THEN SUM(total_cpu_ms) + ELSE SUM(total_duration_ms) + END DESC, workspace_id, runnable_path, worker_group + LIMIT $5 + "#, + from_hour, + query.worker_group, + query.workspace, + order_by, + query.limit.unwrap_or(100).clamp(1, 1000), + ) + .fetch_all(&db) + .await?; + Ok(Json(rows)) +} + #[derive(Deserialize)] struct QueueMetricsSeriesQuery { window_secs: Option, diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index 50833bfe06..2f0d4deb13 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -1299,6 +1299,13 @@ pub(crate) async fn delete_workspace( .execute(&mut *tx) .await?; + sqlx::query!( + "DELETE FROM runnable_job_stats WHERE workspace_id = $1", + &w_id + ) + .execute(&mut *tx) + .await?; + // If this workspace is itself a dev workspace, deleting it dissolves the pairing, so also drop // the parent prod's reserved dev_workspace_lock (mirrors detach_dev_workspace) — otherwise prod // stays locked against direct deploy/forking with no dev workspace left to make changes in. diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 0d21402a78..6eaf9f06ea 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -23524,6 +23524,54 @@ paths: - id - values + /workers/runnable_stats: + get: + summary: get resource usage per runnable and worker group over a time window (require devops role) + operationId: getRunnableStats + tags: + - worker + parameters: + - name: window_secs + in: query + required: false + description: how far back to read, in seconds (defaults to one day, capped at the 30-day retention) + schema: + type: integer + - name: worker_group + in: query + required: false + description: only the jobs run by this worker group + schema: + type: string + - name: workspace + in: query + required: false + description: only the jobs of this workspace + schema: + type: string + - name: order_by + in: query + required: false + description: "what the rows are ranked by, descending (default: total_duration)" + schema: + type: string + enum: [total_duration, max_memory, cpu] + - name: limit + in: query + required: false + description: number of rows to return (default 100, at most 1000) + schema: + type: integer + responses: + "200": + description: resource usage per runnable and worker group, heaviest first + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/RunnableStats" + /workers/queue_metrics_series: get: summary: get the queue metrics of a time window, as a bounded line per tag @@ -35875,6 +35923,45 @@ components: `u/`, `g/`, or an email that exists in this workspace. + RunnableStats: + type: object + description: > + Resource usage of the jobs of one runnable on one worker group. Memory and CPU are + only known for jobs run as a subprocess; memory is the peak of the job's main process. + properties: + workspace_id: + type: string + runnable_path: + type: string + description: path of the script or flow step; `` gathers previews and other runs without a stable path + worker_group: + type: string + job_count: + type: integer + total_duration_ms: + type: integer + max_memory_peak: + type: integer + description: highest peak memory of a job, in kB + sum_memory_peak: + type: integer + description: sum of the peak memory of the jobs that have one, in kB + memory_sample_count: + type: integer + description: number of jobs that have a peak memory reading + total_cpu_ms: + type: integer + required: + - workspace_id + - runnable_path + - worker_group + - job_count + - total_duration_ms + - max_memory_peak + - sum_memory_peak + - memory_sample_count + - total_cpu_ms + WorkerPing: type: object properties: diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 5754b8c8c1..2c94d95618 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -123,6 +123,7 @@ pub mod query_builders; pub mod queue; pub mod queue_metrics; pub mod result_stream; +pub mod runnable_job_stats; pub mod runnable_settings; pub mod runnables; pub mod schedule; diff --git a/backend/windmill-common/src/runnable_job_stats.rs b/backend/windmill-common/src/runnable_job_stats.rs new file mode 100644 index 0000000000..fa1b99e215 --- /dev/null +++ b/backend/windmill-common/src/runnable_job_stats.rs @@ -0,0 +1,288 @@ +//! Hourly resource rollup per runnable and worker group (`runnable_job_stats`). +//! +//! Workers accumulate completed jobs in memory and upsert periodically, so that the +//! "heaviest scripts" view never has to aggregate `v2_job_completed`. +//! +//! What a row leaves out: executors that run in the worker process (SQL, native +//! TypeScript, ...) have no CPU reading, and for memory report nothing or, for some SQL +//! ones, the size of their result, as the job's `memory_peak` does. Memory is otherwise +//! the peak of the job's main process only, and CPU counts that process and the children +//! it waited for, and nothing for a job that was killed. +//! +//! Nothing here checks authorization: these functions write and prune every workspace's +//! rows, and are for the worker's completion path and the server's monitor only. Reads +//! go through the devops-gated API handler. + +use std::{collections::HashMap, sync::Mutex}; + +use sqlx::{Pool, Postgres}; + +use crate::{jobs::JobKind, worker_group_job_stats::get_current_hour}; + +/// Every run without a stable path is counted under this one, per workspace and worker +/// group: previews and other ad-hoc runs carry caller-chosen paths that would otherwise +/// grow the table without bound. +pub const ADHOC_RUNNABLE_PATH: &str = ""; + +/// Also the furthest back the API reads. +pub const RETENTION_DAYS: i64 = 30; + +#[derive(Debug, Default, Clone, Copy, PartialEq)] +struct RunnableStats { + job_count: i32, + total_duration_ms: i64, + max_memory_peak: i32, + sum_memory_peak: i64, + memory_sample_count: i32, + total_cpu_ms: i64, +} + +impl RunnableStats { + /// `mem_peak` is in kB and non-positive when the job had no reading. + fn add_job(&mut self, duration_ms: i64, mem_peak: i32, cpu_time_ms: Option) { + self.job_count += 1; + self.total_duration_ms += duration_ms; + if mem_peak > 0 { + self.max_memory_peak = self.max_memory_peak.max(mem_peak); + self.sum_memory_peak += mem_peak as i64; + self.memory_sample_count += 1; + } + self.total_cpu_ms += cpu_time_ms.unwrap_or(0).max(0); + } + + fn merge(&mut self, other: &RunnableStats) { + self.job_count += other.job_count; + self.total_duration_ms += other.total_duration_ms; + self.max_memory_peak = self.max_memory_peak.max(other.max_memory_peak); + self.sum_memory_peak += other.sum_memory_peak; + self.memory_sample_count += other.memory_sample_count; + self.total_cpu_ms += other.total_cpu_ms; + } +} + +/// (hour, workspace_id, runnable_path, worker_group) +type StatsKey = (i64, String, String, String); + +lazy_static::lazy_static! { + static ref STATS: Mutex> = Mutex::new(HashMap::new()); +} + +/// The path a job is rolled up under: its own for the kinds whose path names something +/// deployed (a flow step's is `/`), the ad-hoc bucket otherwise. +pub fn stats_path(kind: JobKind, runnable_path: Option<&str>) -> &str { + match (kind, runnable_path) { + ( + JobKind::Script + | JobKind::Script_Hub + | JobKind::FlowScript + | JobKind::FlowNode + | JobKind::AIAgent, + Some(path), + ) if !path.is_empty() => path, + _ => ADHOC_RUNNABLE_PATH, + } +} + +pub fn accumulate_runnable_job_stats( + workspace_id: &str, + kind: JobKind, + runnable_path: Option<&str>, + worker_group: &str, + duration_ms: i64, + mem_peak: i32, + cpu_time_ms: Option, +) { + let key = ( + get_current_hour(), + workspace_id.to_string(), + stats_path(kind, runnable_path).to_string(), + worker_group.to_string(), + ); + STATS + .lock() + .unwrap() + .entry(key) + .or_default() + .add_job(duration_ms, mem_peak, cpu_time_ms); +} + +/// Counts the time spent on a round of a job that is not completing, without counting a +/// job: a suspended workflow-as-code round replays its finished steps for free when it +/// resumes, so the time they took would otherwise never reach the rollup. A row can +/// therefore hold time and no job until the workflow completes. +pub fn accumulate_runnable_round( + workspace_id: &str, + kind: JobKind, + runnable_path: Option<&str>, + worker_group: &str, + duration_ms: i64, + cpu_time_ms: Option, +) { + let key = ( + get_current_hour(), + workspace_id.to_string(), + stats_path(kind, runnable_path).to_string(), + worker_group.to_string(), + ); + let mut stats = STATS.lock().unwrap(); + let entry = stats.entry(key).or_default(); + entry.total_duration_ms += duration_ms; + entry.total_cpu_ms += cpu_time_ms.unwrap_or(0).max(0); +} + +pub async fn flush_runnable_job_stats(db: &Pool) -> Result<(), sqlx::Error> { + let drained: Vec<(StatsKey, RunnableStats)> = { + let mut stats = STATS.lock().unwrap(); + if stats.is_empty() { + return Ok(()); + } + stats.drain().collect() + }; + + let mut hours = Vec::with_capacity(drained.len()); + let mut workspace_ids = Vec::with_capacity(drained.len()); + let mut runnable_paths = Vec::with_capacity(drained.len()); + let mut worker_groups = Vec::with_capacity(drained.len()); + let mut job_counts = Vec::with_capacity(drained.len()); + let mut durations = Vec::with_capacity(drained.len()); + let mut max_peaks = Vec::with_capacity(drained.len()); + let mut sum_peaks = Vec::with_capacity(drained.len()); + let mut sample_counts = Vec::with_capacity(drained.len()); + let mut cpus = Vec::with_capacity(drained.len()); + for ((hour, workspace_id, runnable_path, worker_group), s) in &drained { + hours.push(*hour); + workspace_ids.push(workspace_id.clone()); + runnable_paths.push(runnable_path.clone()); + worker_groups.push(worker_group.clone()); + job_counts.push(s.job_count); + durations.push(s.total_duration_ms); + max_peaks.push(s.max_memory_peak); + sum_peaks.push(s.sum_memory_peak); + sample_counts.push(s.memory_sample_count); + cpus.push(s.total_cpu_ms); + } + + // The join drops rows of a workspace deleted since its jobs ran. The ORDER BY makes + // every worker lock the rows they share in one order: without it two concurrent + // flushes can deadlock, and the join is free to undo an ordering of the arrays. + let res = sqlx::query!( + r#" + INSERT INTO runnable_job_stats + (hour, workspace_id, runnable_path, worker_group, job_count, total_duration_ms, + max_memory_peak, sum_memory_peak, memory_sample_count, total_cpu_ms) + SELECT s.hour, s.workspace_id, s.runnable_path, s.worker_group, s.job_count, + s.total_duration_ms, s.max_memory_peak, s.sum_memory_peak, + s.memory_sample_count, s.total_cpu_ms + FROM UNNEST($1::bigint[], $2::text[], $3::text[], $4::text[], $5::int[], $6::bigint[], + $7::int[], $8::bigint[], $9::int[], $10::bigint[]) + AS s(hour, workspace_id, runnable_path, worker_group, job_count, total_duration_ms, + max_memory_peak, sum_memory_peak, memory_sample_count, total_cpu_ms) + JOIN workspace w ON w.id = s.workspace_id + ORDER BY s.hour, s.workspace_id, s.runnable_path, s.worker_group + ON CONFLICT (hour, workspace_id, runnable_path, worker_group) DO UPDATE SET + job_count = runnable_job_stats.job_count + EXCLUDED.job_count, + total_duration_ms = runnable_job_stats.total_duration_ms + EXCLUDED.total_duration_ms, + max_memory_peak = GREATEST(runnable_job_stats.max_memory_peak, EXCLUDED.max_memory_peak), + sum_memory_peak = runnable_job_stats.sum_memory_peak + EXCLUDED.sum_memory_peak, + memory_sample_count = runnable_job_stats.memory_sample_count + EXCLUDED.memory_sample_count, + total_cpu_ms = runnable_job_stats.total_cpu_ms + EXCLUDED.total_cpu_ms + "#, + &hours, + &workspace_ids, + &runnable_paths, + &worker_groups, + &job_counts, + &durations, + &max_peaks, + &sum_peaks, + &sample_counts, + &cpus, + ) + .execute(db) + .await; + + if let Err(e) = res { + let mut stats = STATS.lock().unwrap(); + for (key, s) in drained { + stats.entry(key).or_default().merge(&s); + } + return Err(e); + } + Ok(()) +} + +/// Also removes the rows of workspaces that no longer exist: a flush that overlaps a +/// workspace's deletion still sees the workspace and can write after the deletion's own +/// cleanup ran. Such a row is at most a couple of hours old, so only the last day is +/// searched, which keeps both statements on a range of the primary key. +pub async fn cleanup_old_runnable_job_stats( + db: &Pool, + retention_days: i64, +) -> Result { + let current_hour = get_current_hour(); + let expired = sqlx::query!( + "DELETE FROM runnable_job_stats WHERE hour < $1", + current_hour - retention_days * 24 * 3600 + ) + .execute(db) + .await?; + let orphaned = sqlx::query!( + "DELETE FROM runnable_job_stats s WHERE s.hour >= $1 \ + AND NOT EXISTS (SELECT 1 FROM workspace w WHERE w.id = s.workspace_id)", + current_hour - 24 * 3600 + ) + .execute(db) + .await?; + Ok(expired.rows_affected() + orphaned.rows_affected()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_deployed_runnables_keep_their_path() { + assert_eq!( + stats_path(JobKind::Script, Some("f/etl/load")), + "f/etl/load" + ); + assert_eq!( + stats_path(JobKind::FlowScript, Some("f/etl/flow/a")), + "f/etl/flow/a" + ); + // A preview carries the path being edited, which is not a stable identity. + assert_eq!( + stats_path(JobKind::Preview, Some("f/etl/load")), + ADHOC_RUNNABLE_PATH + ); + assert_eq!( + stats_path(JobKind::FlowPreview, Some("f/etl/flow")), + ADHOC_RUNNABLE_PATH + ); + assert_eq!( + stats_path(JobKind::Dependencies, Some("f/etl/load")), + ADHOC_RUNNABLE_PATH + ); + assert_eq!(stats_path(JobKind::Script, None), ADHOC_RUNNABLE_PATH); + } + + #[test] + fn jobs_without_a_memory_reading_do_not_dilute_it() { + let mut s = RunnableStats::default(); + s.add_job(1000, 2048, Some(700)); + s.add_job(500, -3, None); + s.add_job(250, 0, Some(-5)); + s.add_job(250, 1024, Some(50)); + assert_eq!( + s, + RunnableStats { + job_count: 4, + total_duration_ms: 2000, + max_memory_peak: 2048, + sum_memory_peak: 3072, + memory_sample_count: 2, + total_cpu_ms: 750, + } + ); + } +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 389ae1d112..68749d2b90 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -182,6 +182,15 @@ pub struct CanceledBy { pub reason: Option, } +/// What only the worker that ran a job knows about it, for the per-runnable rollup: an +/// agent worker's completion is processed by the server, whose own worker group is not +/// the one the job ran in. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobResourceUsage { + pub worker_group: String, + pub cpu_time_ms: Option, +} + #[derive(Debug, Serialize, Deserialize)] pub struct JobCompleted { pub job: MiniCompletedJob, @@ -196,6 +205,8 @@ pub struct JobCompleted { pub duration: Option, pub has_stream: Option, pub from_cache: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resource_usage: Option, #[serde(skip)] pub flow_runners: Option>, #[serde(skip)] @@ -4320,6 +4331,7 @@ impl PulledJobResult { ))), result_columns: None, mem_peak: 0, + resource_usage: None, cached_res_path: None, token: "".to_string(), canceled_by: None, @@ -4341,6 +4353,7 @@ impl PulledJobResult { }))), result_columns: None, mem_peak: 0, + resource_usage: None, cached_res_path: None, token: "".to_string(), canceled_by: None, diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 1ae588ef11..59fa6456da 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -4086,6 +4086,7 @@ async fn handle_dedicated_bunnative( result, result_columns: None, mem_peak: 0, + resource_usage: None, canceled_by: None, success, cached_res_path: None, diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 84fd547641..31eb8034f9 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -59,6 +59,56 @@ use windmill_common::tracing_init::{OTEL_JOB_LOGS, OTEL_PREFIX, QUIET_MODE, VERB lazy_static::lazy_static! { pub static ref SLOW_LOGS: bool = std::env::var("SLOW_LOGS").ok().is_some_and(|x| x == "1" || x == "true"); + // A job runs several children in turn (install, build, run): their CPU time adds up + // here until the job completes. + static ref JOB_CPU_TIME_MS: std::sync::Mutex> = Default::default(); +} + +/// CPU time of the children run so far for the job, which stops being tracked. +pub(crate) fn take_job_cpu_time_ms(job_id: &Uuid) -> Option { + JOB_CPU_TIME_MS.lock().unwrap().remove(job_id) +} + +/// CPU time in ms from the content of `/proc//stat`: the process's own user and +/// system time plus that of the children it waited for, which is where the time of a +/// job run under a wrapper (nsjail, unshare) ends up. +#[cfg_attr(not(target_os = "linux"), allow(dead_code))] +fn parse_proc_stat_cpu_ms(stat: &str) -> Option { + // The values are in USER_HZ ticks, which Linux fixes at 100 per second. + const MS_PER_TICK: i64 = 10; + // `comm` is free text in parentheses: the numbered fields resume after its last `)`, + // with the state as field 3 and utime, stime, cutime, cstime as fields 14 to 17. + let fields: Vec<&str> = stat.rsplit_once(')')?.1.split_ascii_whitespace().collect(); + let ticks = fields + .get(11..15)? + .iter() + .map(|f| f.parse::().ok()) + .sum::>()?; + Some(ticks * MS_PER_TICK) +} + +/// Resolves once the child has exited, to its CPU time. The child is left un-reaped: +/// `/proc//stat` is only readable until then, so the caller reaps afterwards. +#[cfg(target_os = "linux")] +async fn wait_for_exit_cpu_time_ms(pid: u32) -> Option { + use nix::sys::wait::{waitid, Id, WaitPidFlag}; + let (tx, rx) = tokio::sync::oneshot::channel(); + // A plain thread rather than `spawn_blocking`: the wait cannot be cancelled, and a + // blocking task still parked on a live child would hold up the runtime's shutdown. + std::thread::Builder::new() + .name("job-cpu-time".to_string()) + .spawn(move || { + let cpu = waitid( + Id::Pid(Pid::from_raw(pid as i32)), + WaitPidFlag::WEXITED | WaitPidFlag::WNOWAIT, + ) + .ok() + .and_then(|_| std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()) + .and_then(|stat| parse_proc_stat_cpu_ms(&stat)); + let _ = tx.send(cpu); + }) + .ok()?; + rx.await.ok().flatten() } // - kill windows process along with all child processes @@ -267,7 +317,15 @@ pub async fn handle_child( let wait_on_child = async { let kill_reason = tokio::select! { biased; - result = Box::into_pin(child.wait()) => return result.map(Ok), + result = async { + #[cfg(target_os = "linux")] + if let Some(pid) = pid.filter(|_| job_id != Uuid::nil()) { + if let Some(cpu) = wait_for_exit_cpu_time_ms(pid).await { + *JOB_CPU_TIME_MS.lock().unwrap().entry(job_id).or_insert(0) += cpu; + } + } + Box::into_pin(child.wait()).await + } => return result.map(Ok), Ok(()) = too_many_logs.changed() => KillReason::TooManyLogs, _ = sleep(timeout_duration) => KillReason::Timeout { phase: child_name.to_string(), @@ -1214,6 +1272,21 @@ pub fn process_status( mod tests { use super::*; + #[test] + fn proc_stat_cpu_time_survives_a_hostile_process_name() { + // utime 150, stime 50, cutime 700, cstime 100 ticks: 1000 ticks, 10 s. + let tail = "Z 1 42 42 0 -1 4194560 100 200 0 0 150 50 700 100 20 0 1 0 12345 0 0"; + assert_eq!( + parse_proc_stat_cpu_ms(&format!("42 (bun) {tail}")), + Some(10_000) + ); + assert_eq!( + parse_proc_stat_cpu_ms(&format!("42 (a) R 9 (9 9 9 9 9 9 9 9 9) {tail}")), + Some(10_000) + ); + assert_eq!(parse_proc_stat_cpu_ms("42 (bun) Z 1 42"), None); + } + // A non-positive timeout means "unset" to `resolve_job_timeout`, so an // exhausted budget that reported 0 would hand the next phase no deadline at // all — the opposite of what running out of time should do. diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 12edd7823e..e5b84f39b9 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -21,6 +21,7 @@ use windmill_common::{ error::{self, Error}, flow_status::FlowJobDuration, jobs::JobKind, + runnable_job_stats::{accumulate_runnable_job_stats, flush_runnable_job_stats}, utils::WarnAfterExt, worker::{error_to_value, to_raw_value, Connection, WORKER_GROUP}, worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, @@ -32,8 +33,8 @@ use windmill_common::bench::{BenchmarkInfo, BenchmarkIter}; use windmill_queue::{ append_logs, asset_dispatch, get_mini_completed_job, is_pre_shaped_wm_failure_result, - parse_result_object, CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob, MiniPulledJob, - ValidableJson, WrappedError, INIT_SCRIPT_TAG, MANUAL_FAILURE_ERROR_NAME, + parse_result_object, CanceledBy, FlowRunners, JobCompleted, JobResourceUsage, MiniCompletedJob, + MiniPulledJob, ValidableJson, WrappedError, INIT_SCRIPT_TAG, MANUAL_FAILURE_ERROR_NAME, }; use serde_json::{json, value::RawValue, Value}; @@ -231,6 +232,10 @@ async fn process_jc( let duration_ms = jc.duration.clone(); let script_lang = jc.job.script_lang.clone(); let workspace_id = jc.job.workspace_id.clone(); + let job_kind = jc.job.kind; + let runnable_path = jc.job.runnable_path.clone(); + let mem_peak = jc.mem_peak; + let resource_usage = jc.resource_usage.take(); let root_job = handle_receive_completed_job( jc, @@ -269,6 +274,17 @@ async fn process_jc( duration_ms, ) .await; + accumulate_runnable_job_stats( + &workspace_id, + job_kind, + runnable_path.as_deref(), + resource_usage + .as_ref() + .map_or(WORKER_GROUP.as_str(), |u| u.worker_group.as_str()), + duration_ms, + mem_peak, + resource_usage.as_ref().and_then(|u| u.cpu_time_ms), + ); } success @@ -317,6 +333,14 @@ pub fn start_background_processor( let flush_handle = tokio::spawn(async move { let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(900)); // Flush every 15 min + // Once per hourly bucket, at an offset of its own per worker so that a fleet + // started together does not write the same rows in the same second. + let runnable_period = tokio::time::Duration::from_secs(3600); + let mut runnable_interval = tokio::time::interval_at( + tokio::time::Instant::now() + runnable_period.mul_f64(rand::random::()), + runnable_period, + ); + loop { tokio::select! { _ = interval.tick() => { @@ -324,6 +348,11 @@ pub fn start_background_processor( tracing::error!("Failed to flush worker group job stats: {}", e); } } + _ = runnable_interval.tick() => { + if let Err(e) = flush_runnable_job_stats(&db_clone).await { + tracing::error!("Failed to flush runnable job stats: {}", e); + } + } _ = killpill_rx_clone.recv() => { tracing::info!("bg processor received killpill signal, flushing remaining stats"); break; @@ -524,6 +553,18 @@ pub fn start_background_processor( Ok(Err(join_err)) => tracing::error!("Stats flush task failed: {}", join_err), Err(_) => tracing::error!("Stats flush timed out after 10 seconds"), } + // The periodic task stops at the killpill, before the jobs drained above were + // accumulated: without this a short-lived worker would lose its whole rollup. + match tokio::time::timeout( + std::time::Duration::from_secs(10), + flush_runnable_job_stats(&db), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) => tracing::error!("Failed to flush runnable job stats: {}", e), + Err(_) => tracing::error!("Runnable job stats flush timed out after 10 seconds"), + } job_completed_processor_is_done.store(true, Ordering::SeqCst); @@ -555,6 +596,7 @@ pub async fn process_result( job_dir: &str, job_completed_tx: JobCompletedSender, mem_peak: i32, + cpu_time_ms: Option, canceled_by: Option, cached_res_path: Option, token: &str, @@ -565,6 +607,7 @@ pub async fn process_result( has_stream: bool, flow_runners: Option>, ) -> error::Result { + let resource_usage = Some(JobResourceUsage { worker_group: WORKER_GROUP.clone(), cpu_time_ms }); match result { Ok(result) => { send_job_completed( @@ -575,6 +618,7 @@ pub async fn process_result( result, result_columns, mem_peak, + resource_usage, canceled_by, success: true, cached_res_path, @@ -658,6 +702,7 @@ pub async fn process_result( result_columns: None, preprocessed_args: None, mem_peak, + resource_usage, canceled_by, success: false, cached_res_path, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 5b5df51b30..5ae626f27e 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2247,6 +2247,7 @@ pub async fn handle_all_job_kind_error( ))), result_columns: None, mem_peak: 0, + resource_usage: None, canceled_by: None, success: false, cached_res_path: None, @@ -3772,6 +3773,7 @@ pub async fn run_worker( result: Arc::new(empty_result()), result_columns: None, mem_peak: 0, + resource_usage: None, cached_res_path: None, token: "".to_string(), canceled_by: None, @@ -4763,6 +4765,7 @@ pub async fn handle_queued_job( result, result_columns: None, mem_peak: 0, + resource_usage: None, canceled_by: None, success: true, cached_res_path: None, @@ -5112,6 +5115,8 @@ pub async fn handle_queued_job( let cjob = MiniCompletedJob::from(job.to_owned()); drop(job); + // Taken ahead of the returns below so that none of them leaves the entry behind. + let cpu_time_ms = crate::handle_child::take_job_cpu_time_ms(&cjob.id); //it's a test job, no need to update the db if cjob.workspace_id == "" { return Ok(JobOutcome::Completed); @@ -5128,6 +5133,17 @@ pub async fn handle_queued_job( .is_err_and(|err| matches!(err, &Error::WacSuspended(_))) { // WAC v2 job suspended while waiting for child jobs — don't complete it + // Only a worker with a database connection flushes the rollup. + if let Connection::Sql(_) = conn { + windmill_common::runnable_job_stats::accumulate_runnable_round( + &cjob.workspace_id, + cjob.kind, + cjob.runnable_path.as_deref(), + &WORKER_GROUP, + started.elapsed().as_millis() as i64, + cpu_time_ms, + ); + } return Ok(JobOutcome::Completed); } crate::resource_metrics::record_job_memory_peak(&cjob.tag, mem_peak); @@ -5137,6 +5153,7 @@ pub async fn handle_queued_job( job_dir, job_completed_tx, mem_peak, + cpu_time_ms, canceled_by, cached_res_path, &client.token, diff --git a/frontend/src/lib/components/HeaviestScriptsDrawer.svelte b/frontend/src/lib/components/HeaviestScriptsDrawer.svelte new file mode 100644 index 0000000000..a4591924c9 --- /dev/null +++ b/frontend/src/lib/components/HeaviestScriptsDrawer.svelte @@ -0,0 +1,201 @@ + + + (opened = false)}> + +
+ {#snippet action()} + + {#snippet children({ item })} + {#each Object.keys(WINDOWS) as key (key)} + + {/each} + {/snippet} + + {/snippet} + +
+
+ Sort by + orderBy, + (v) => { + orderBy = v + logFeatureUsage('worker_resource_stats', 'sort', { key: v }) + } + } + noWFull + > + {#snippet children({ item })} + {#each Object.entries(ORDERS) as [value, label] (value)} + + {/each} + {/snippet} + +
+
+