mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-12 16:02:56 +00:00
feat: per-script resource rollup and heaviest scripts table (#11631)
* feat: export worker cpu, memory and occupancy metrics Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: refresh the exported memory limit with the cgroup readings Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * feat: per-script resource rollup and heaviest scripts table Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: pin the companion ee commit and drop job cpu entries on every exit Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: count the cpu time of suspended workflow rounds Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: count suspended workflow rounds' runtime and guard empty averages Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * perf: flush the runnable rollup hourly in key order and drop its workspace fk Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: order the rollup upsert in sql and sweep rows of deleted workspaces Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * perf: keep the rollup orphan sweep on a primary key range Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref to a1c00d73a05f227a61d2bf37cc86cd9f5bb1777a This commit updates the EE repository reference after PR #866 was merged in windmill-ee-private. Previous ee-repo-ref: c6bcb1bbb5a7e85e870f9de232a0ad1540acf2bd New ee-repo-ref: a1c00d73a05f227a61d2bf37cc86cd9f5bb1777a Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 5.5
windmill-internal-app[bot]
Claude Opus 5.5
windmill-internal-app[bot]
parent
ab8606f379
commit
442357dbe8
+14
@@ -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"
|
||||
}
|
||||
+23
@@ -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"
|
||||
}
|
||||
+14
@@ -0,0 +1,14 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "DELETE FROM runnable_job_stats WHERE hour < $1",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Int8"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "97d8a9835344b8c72d59afc99d76b681c5120ef29eb09b8626a3fbc43a32cc0e"
|
||||
}
|
||||
+74
@@ -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"
|
||||
}
|
||||
+14
@@ -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"
|
||||
}
|
||||
@@ -1 +1 @@
|
||||
017710bdd0a10de857f0f40800dde238cdb1ad24
|
||||
a1c00d73a05f227a61d2bf37cc86cd9f5bb1777a
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS runnable_job_stats;
|
||||
@@ -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)
|
||||
);
|
||||
@@ -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);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<i64>,
|
||||
worker_group: Option<String>,
|
||||
workspace: Option<String>,
|
||||
order_by: Option<String>,
|
||||
limit: Option<i64>,
|
||||
}
|
||||
|
||||
#[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<DB>,
|
||||
Query(query): Query<RunnableStatsQuery>,
|
||||
) -> JsonResult<Vec<RunnableStats>> {
|
||||
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<i64>,
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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/<username>`, `g/<groupname>`, 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; `<adhoc>` 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:
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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 = "<adhoc>";
|
||||
|
||||
/// 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<i64>) {
|
||||
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<HashMap<StatsKey, RunnableStats>> = 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 `<flow path>/<step id>`), 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<i64>,
|
||||
) {
|
||||
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<i64>,
|
||||
) {
|
||||
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<Postgres>) -> 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<Postgres>,
|
||||
retention_days: i64,
|
||||
) -> Result<u64, sqlx::Error> {
|
||||
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,
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -182,6 +182,15 @@ pub struct CanceledBy {
|
||||
pub reason: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<i64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct JobCompleted {
|
||||
pub job: MiniCompletedJob,
|
||||
@@ -196,6 +205,8 @@ pub struct JobCompleted {
|
||||
pub duration: Option<i64>,
|
||||
pub has_stream: Option<bool>,
|
||||
pub from_cache: Option<bool>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub resource_usage: Option<JobResourceUsage>,
|
||||
#[serde(skip)]
|
||||
pub flow_runners: Option<Arc<FlowRunners>>,
|
||||
#[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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<std::collections::HashMap<Uuid, i64>> = 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<i64> {
|
||||
JOB_CPU_TIME_MS.lock().unwrap().remove(job_id)
|
||||
}
|
||||
|
||||
/// CPU time in ms from the content of `/proc/<pid>/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<i64> {
|
||||
// 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::<i64>().ok())
|
||||
.sum::<Option<i64>>()?;
|
||||
Some(ticks * MS_PER_TICK)
|
||||
}
|
||||
|
||||
/// Resolves once the child has exited, to its CPU time. The child is left un-reaped:
|
||||
/// `/proc/<pid>/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<i64> {
|
||||
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.
|
||||
|
||||
@@ -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::<f64>()),
|
||||
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<i64>,
|
||||
canceled_by: Option<CanceledBy>,
|
||||
cached_res_path: Option<String>,
|
||||
token: &str,
|
||||
@@ -565,6 +607,7 @@ pub async fn process_result(
|
||||
has_stream: bool,
|
||||
flow_runners: Option<Arc<FlowRunners>>,
|
||||
) -> error::Result<crate::worker::JobOutcome> {
|
||||
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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -0,0 +1,201 @@
|
||||
<script lang="ts">
|
||||
import { WorkerService } from '$lib/gen'
|
||||
import { resource } from 'runed'
|
||||
import { Alert, Drawer, DrawerContent, Section, Skeleton } from './common'
|
||||
import TextInput from './text_input/TextInput.svelte'
|
||||
import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte'
|
||||
import ToggleButton from './common/toggleButton-v2/ToggleButton.svelte'
|
||||
import Select from './select/Select.svelte'
|
||||
import DataTable from './table/DataTable.svelte'
|
||||
import Head from './table/Head.svelte'
|
||||
import Cell from './table/Cell.svelte'
|
||||
import { formatMemory, msToReadableTime } from '$lib/utils'
|
||||
import { logFeatureUsage } from '$lib/utils/featureUsage'
|
||||
|
||||
interface Props {
|
||||
workerGroups: string[]
|
||||
}
|
||||
|
||||
let { workerGroups }: Props = $props()
|
||||
|
||||
const WINDOWS = {
|
||||
'24h': 24 * 3600,
|
||||
'7d': 7 * 24 * 3600,
|
||||
'30d': 30 * 24 * 3600
|
||||
}
|
||||
const ORDERS = {
|
||||
total_duration: 'Total runtime',
|
||||
max_memory: 'Max memory',
|
||||
cpu: 'CPU time'
|
||||
} as const
|
||||
type OrderBy = keyof typeof ORDERS
|
||||
// The server gathers every run without a stable path (previews, dependency jobs) under this one.
|
||||
const ADHOC_PATH = '<adhoc>'
|
||||
|
||||
let drawer: Drawer | undefined = $state()
|
||||
let opened = $state(false)
|
||||
let windowKey: keyof typeof WINDOWS = $state('24h')
|
||||
let orderBy: OrderBy = $state('total_duration')
|
||||
let workerGroup: string | undefined = $state()
|
||||
let workspace = $state('')
|
||||
|
||||
export function openDrawer(group?: string) {
|
||||
workerGroup = group
|
||||
opened = true
|
||||
drawer?.openDrawer()
|
||||
logFeatureUsage('worker_resource_stats', 'opened')
|
||||
}
|
||||
|
||||
const stats = resource(
|
||||
() => ({ opened, windowKey, orderBy, workerGroup, workspace: workspace.trim() }),
|
||||
async (q, _, { signal }) => {
|
||||
if (!q.opened) return undefined
|
||||
try {
|
||||
return await WorkerService.getRunnableStats({
|
||||
windowSecs: WINDOWS[q.windowKey],
|
||||
orderBy: q.orderBy,
|
||||
workerGroup: q.workerGroup || undefined,
|
||||
workspace: q.workspace || undefined
|
||||
})
|
||||
} finally {
|
||||
// A slower answer for filters no longer selected must not replace the current one:
|
||||
// `resource` drops the abort error thrown in its place.
|
||||
signal.throwIfAborted()
|
||||
}
|
||||
},
|
||||
{ debounce: 200 }
|
||||
)
|
||||
|
||||
// Many scripts run for well under a second, which `msToReadableTime` rounds to "0.0s".
|
||||
function duration(ms: number) {
|
||||
return ms < 1000 ? `${Math.round(ms)}ms` : msToReadableTime(ms, 1)
|
||||
}
|
||||
|
||||
function cores(cpuMs: number, durationMs: number) {
|
||||
return durationMs > 0 ? (cpuMs / durationMs).toFixed(2) : '-'
|
||||
}
|
||||
</script>
|
||||
|
||||
<Drawer bind:this={drawer} size="1100px" on:close={() => (opened = false)}>
|
||||
<DrawerContent title="Heaviest scripts" on:close={drawer?.closeDrawer}>
|
||||
<Section
|
||||
label="Resource usage per script"
|
||||
tooltip="Aggregated by the workers once an hour, per script and worker group, so the last hour is incomplete. Memory is the peak of the job's main process. CPU time counts that process and the child processes it waited for. Scripts run inside the worker process (SQL, native TypeScript) and jobs of dedicated workers report neither, and a killed job reports no CPU time."
|
||||
>
|
||||
{#snippet action()}
|
||||
<ToggleButtonGroup bind:selected={windowKey} noWFull>
|
||||
{#snippet children({ item })}
|
||||
{#each Object.keys(WINDOWS) as key (key)}
|
||||
<ToggleButton value={key} label={key} size="sm" {item} />
|
||||
{/each}
|
||||
{/snippet}
|
||||
</ToggleButtonGroup>
|
||||
{/snippet}
|
||||
|
||||
<div class="flex flex-wrap items-center gap-4 pb-4">
|
||||
<div class="flex items-center gap-2">
|
||||
<span class="text-xs text-secondary">Sort by</span>
|
||||
<ToggleButtonGroup
|
||||
bind:selected={
|
||||
() => orderBy,
|
||||
(v) => {
|
||||
orderBy = v
|
||||
logFeatureUsage('worker_resource_stats', 'sort', { key: v })
|
||||
}
|
||||
}
|
||||
noWFull
|
||||
>
|
||||
{#snippet children({ item })}
|
||||
{#each Object.entries(ORDERS) as [value, label] (value)}
|
||||
<ToggleButton {value} {label} size="sm" {item} />
|
||||
{/each}
|
||||
{/snippet}
|
||||
</ToggleButtonGroup>
|
||||
</div>
|
||||
<div class="w-48">
|
||||
<Select
|
||||
items={workerGroups.map((g) => ({ label: g, value: g }))}
|
||||
bind:value={workerGroup}
|
||||
placeholder="All worker groups"
|
||||
clearable
|
||||
size="sm"
|
||||
/>
|
||||
</div>
|
||||
<div class="w-48">
|
||||
<TextInput
|
||||
bind:value={workspace}
|
||||
inputProps={{ placeholder: 'Workspace id' }}
|
||||
size="sm"
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{#if stats.error}
|
||||
<Alert type="error" title="Failed to load the script statistics">
|
||||
{stats.error.message}
|
||||
</Alert>
|
||||
{:else if stats.current === undefined}
|
||||
<Skeleton layout={[[8]]} />
|
||||
{:else if stats.current.length === 0}
|
||||
<p class="text-secondary text-xs">No jobs were recorded for this selection.</p>
|
||||
{:else}
|
||||
<DataTable size="sm" noBorder={false} rounded={true}>
|
||||
<Head>
|
||||
<tr>
|
||||
<Cell head first>Script</Cell>
|
||||
<Cell head>Workspace</Cell>
|
||||
<Cell head>Worker group</Cell>
|
||||
<Cell head numeric>Jobs</Cell>
|
||||
<Cell head numeric>Total runtime</Cell>
|
||||
<Cell head numeric>Avg runtime</Cell>
|
||||
<Cell head numeric>Max memory</Cell>
|
||||
<Cell head numeric>Avg memory</Cell>
|
||||
<Cell head numeric>CPU time</Cell>
|
||||
<Cell head numeric last>Avg CPUs</Cell>
|
||||
</tr>
|
||||
</Head>
|
||||
<tbody>
|
||||
{#each stats.current as s (s.workspace_id + '\0' + s.runnable_path + '\0' + s.worker_group)}
|
||||
<tr class="border-b last:border-b-0">
|
||||
<Cell first class="text-xs text-primary">
|
||||
{#if s.runnable_path === ADHOC_PATH}
|
||||
<span class="text-secondary">Previews and other ad-hoc runs</span>
|
||||
{:else}
|
||||
<span class="font-mono">{s.runnable_path}</span>
|
||||
{/if}
|
||||
</Cell>
|
||||
<Cell class="text-xs text-primary">{s.workspace_id}</Cell>
|
||||
<Cell class="text-xs text-primary">{s.worker_group}</Cell>
|
||||
<Cell numeric class="text-xs text-primary">{s.job_count}</Cell>
|
||||
<Cell numeric class="text-xs text-primary">
|
||||
{duration(s.total_duration_ms)}
|
||||
</Cell>
|
||||
<Cell numeric class="text-xs text-primary">
|
||||
<!-- A workflow suspended and not yet completed has time recorded and no job. -->
|
||||
{s.job_count > 0 ? duration(s.total_duration_ms / s.job_count) : '-'}
|
||||
</Cell>
|
||||
{#if s.memory_sample_count > 0}
|
||||
<Cell numeric class="text-xs text-primary">
|
||||
{formatMemory(s.max_memory_peak)}
|
||||
</Cell>
|
||||
<Cell numeric class="text-xs text-primary">
|
||||
{formatMemory(Math.round(s.sum_memory_peak / s.memory_sample_count))}
|
||||
</Cell>
|
||||
{:else}
|
||||
<Cell numeric class="text-xs text-secondary">-</Cell>
|
||||
<Cell numeric class="text-xs text-secondary">-</Cell>
|
||||
{/if}
|
||||
<Cell numeric class="text-xs text-primary">
|
||||
{duration(s.total_cpu_ms)}
|
||||
</Cell>
|
||||
<Cell numeric last class="text-xs text-primary">
|
||||
{cores(s.total_cpu_ms, s.total_duration_ms)}
|
||||
</Cell>
|
||||
</tr>
|
||||
{/each}
|
||||
</tbody>
|
||||
</DataTable>
|
||||
{/if}
|
||||
</Section>
|
||||
</DrawerContent>
|
||||
</Drawer>
|
||||
@@ -8,6 +8,7 @@
|
||||
import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte'
|
||||
import PageHeader from '$lib/components/PageHeader.svelte'
|
||||
import QueueMetricsDrawer from '$lib/components/QueueMetricsDrawer.svelte'
|
||||
import HeaviestScriptsDrawer from '$lib/components/HeaviestScriptsDrawer.svelte'
|
||||
import ManageTagsDrawer from '$lib/components/ManageTagsDrawer.svelte'
|
||||
import SimpleEditor from '$lib/components/SimpleEditor.svelte'
|
||||
import Cell from '$lib/components/table/Cell.svelte'
|
||||
@@ -34,6 +35,7 @@
|
||||
import { displayDate, groupBy, pluralize, retrieveCommonWorkerPrefix, truncate } from '$lib/utils'
|
||||
import {
|
||||
ExternalLink,
|
||||
Gauge,
|
||||
LineChart,
|
||||
Loader2,
|
||||
Plus,
|
||||
@@ -493,6 +495,7 @@
|
||||
}
|
||||
|
||||
let queueMetricsDrawer: QueueMetricsDrawer | undefined = $state(undefined)
|
||||
let heaviestScriptsDrawer: HeaviestScriptsDrawer | undefined = $state(undefined)
|
||||
let manageTagsDrawer: ManageTagsDrawer | undefined = $state(undefined)
|
||||
let selectedTab: string = $state('default')
|
||||
|
||||
@@ -604,6 +607,10 @@
|
||||
|
||||
{#if $superadmin || $devopsRole}
|
||||
<QueueMetricsDrawer bind:this={queueMetricsDrawer} />
|
||||
<HeaviestScriptsDrawer
|
||||
bind:this={heaviestScriptsDrawer}
|
||||
workerGroups={groupedWorkers.map(([name]) => name)}
|
||||
/>
|
||||
<ManageTagsDrawer
|
||||
bind:this={manageTagsDrawer}
|
||||
bind:defaultTagPerWorkspace
|
||||
@@ -775,6 +782,19 @@
|
||||
Queue metrics
|
||||
</Button>
|
||||
|
||||
<Button
|
||||
unifiedSize="md"
|
||||
variant="default"
|
||||
startIcon={{
|
||||
icon: Gauge
|
||||
}}
|
||||
on:click={() => {
|
||||
heaviestScriptsDrawer?.openDrawer(selectedTab)
|
||||
}}
|
||||
>
|
||||
Heaviest scripts
|
||||
</Button>
|
||||
|
||||
<Button
|
||||
unifiedSize="md"
|
||||
variant="default"
|
||||
|
||||
Reference in New Issue
Block a user