From 8870ddca9a2af659189941bbd68494d70311d626 Mon Sep 17 00:00:00 2001 From: Abel Lucas Date: Fri, 24 Jan 2025 10:16:12 +0100 Subject: [PATCH] backend: rework `queue::push` after v2 --- ...204f9df256e03a588c23b33b4edab0d9c3e1a.json | 21 ++ ...e86cde5ffefb4578d6e091855bc7a4e94d1b0.json | 47 +++ ...6e4490d6004ba58ae26c52a4f21f867e1df70.json | 112 +++++++ backend/Cargo.lock | 1 + backend/src/monitor.rs | 2 +- backend/windmill-api/src/jobs.rs | 55 ++-- backend/windmill-common/src/bench.rs | 246 -------------- backend/windmill-common/src/jobs.rs | 16 + backend/windmill-common/src/lib.rs | 2 - backend/windmill-queue/src/jobs.rs | 303 +++++++++++++----- backend/windmill-worker/Cargo.toml | 1 + backend/windmill-worker/src/bench.rs | 226 +++++++++++++ backend/windmill-worker/src/lib.rs | 2 + .../windmill-worker/src/result_processor.rs | 2 +- backend/windmill-worker/src/worker.rs | 6 +- backend/windmill-worker/src/worker_flow.rs | 4 +- 16 files changed, 682 insertions(+), 364 deletions(-) create mode 100644 backend/.sqlx/query-5d0284028d0dc16fbbf353496a2204f9df256e03a588c23b33b4edab0d9c3e1a.json create mode 100644 backend/.sqlx/query-90d1c1ed47f1a9186cdde2f720fe86cde5ffefb4578d6e091855bc7a4e94d1b0.json create mode 100644 backend/.sqlx/query-f8e0103dae8d80682a6b694762b6e4490d6004ba58ae26c52a4f21f867e1df70.json delete mode 100644 backend/windmill-common/src/bench.rs create mode 100644 backend/windmill-worker/src/bench.rs diff --git a/backend/.sqlx/query-5d0284028d0dc16fbbf353496a2204f9df256e03a588c23b33b4edab0d9c3e1a.json b/backend/.sqlx/query-5d0284028d0dc16fbbf353496a2204f9df256e03a588c23b33b4edab0d9c3e1a.json new file mode 100644 index 0000000000..981c595d69 --- /dev/null +++ b/backend/.sqlx/query-5d0284028d0dc16fbbf353496a2204f9df256e03a588c23b33b4edab0d9c3e1a.json @@ -0,0 +1,21 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO v2_job_queue (\n id, workspace_id, started_at, scheduled_for, running, created_at, tag, priority\n ) SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "UuidArray", + "Varchar", + "Timestamptz", + "Timestamptz", + "Bool", + "Timestamptz", + "Varchar", + "Int2" + ] + }, + "nullable": [] + }, + "hash": "5d0284028d0dc16fbbf353496a2204f9df256e03a588c23b33b4edab0d9c3e1a" +} diff --git a/backend/.sqlx/query-90d1c1ed47f1a9186cdde2f720fe86cde5ffefb4578d6e091855bc7a4e94d1b0.json b/backend/.sqlx/query-90d1c1ed47f1a9186cdde2f720fe86cde5ffefb4578d6e091855bc7a4e94d1b0.json new file mode 100644 index 0000000000..259d59e3ea --- /dev/null +++ b/backend/.sqlx/query-90d1c1ed47f1a9186cdde2f720fe86cde5ffefb4578d6e091855bc7a4e94d1b0.json @@ -0,0 +1,47 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO script (\n summary, description, dedicated_worker, content, workspace_id, path, hash,\n language, tag, created_by, lock\n ) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '')\n ON CONFLICT (workspace_id, hash) DO NOTHING", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Varchar", + "Varchar", + "Int8", + { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb" + ] + } + } + }, + "Varchar", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "90d1c1ed47f1a9186cdde2f720fe86cde5ffefb4578d6e091855bc7a4e94d1b0" +} diff --git a/backend/.sqlx/query-f8e0103dae8d80682a6b694762b6e4490d6004ba58ae26c52a4f21f867e1df70.json b/backend/.sqlx/query-f8e0103dae8d80682a6b694762b6e4490d6004ba58ae26c52a4f21f867e1df70.json new file mode 100644 index 0000000000..7fff2a91a1 --- /dev/null +++ b/backend/.sqlx/query-f8e0103dae8d80682a6b694762b6e4490d6004ba58ae26c52a4f21f867e1df70.json @@ -0,0 +1,112 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO v2_job (\n id, workspace_id,\n created_at, created_by, permissioned_as, permissioned_as_email,\n kind, runnable_id, runnable_path, parent_job, script_lang,\n flow_step_id, flow_root_job,\n trigger, trigger_kind,\n tag, same_worker, visible_to_owner, concurrent_limit, concurrency_time_window_s,\n cache_ttl, timeout, priority,\n args, pre_run_error,\n raw_code, raw_lock, raw_flow\n ) SELECT\n unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,\n $16, $17, $18, $19, $20, $21, $22, $23, unnest($24::jsonb[]), $25, $26, $27, $28\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "UuidArray", + "Varchar", + "Timestamptz", + "Varchar", + "Varchar", + "Varchar", + { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlescriptflow", + "flowscript", + "flownode", + "appscript" + ] + } + } + }, + "Int8", + "Varchar", + "Uuid", + { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb" + ] + } + } + }, + "Varchar", + "Uuid", + "Varchar", + { + "Custom": { + "name": "job_trigger_kind", + "kind": { + "Enum": [ + "webhook", + "http", + "websocket", + "kafka", + "email", + "nats", + "schedule", + "app", + "ui", + "postgres" + ] + } + } + }, + "Varchar", + "Bool", + "Bool", + "Int4", + "Int4", + "Int4", + "Int4", + "Int2", + "JsonbArray", + "Text", + "Text", + "Text", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "f8e0103dae8d80682a6b694762b6e4490d6004ba58ae26c52a4f21f867e1df70" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 1829a48658..4e394024ca 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -11423,6 +11423,7 @@ dependencies = [ "tokio-postgres 0.7.12", "tokio-util", "tracing", + "ulid", "urlencoding", "uuid 1.12.1", "windmill-audit", diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 718e741a68..2c72dd7d57 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1641,7 +1641,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker worker_name, send_result_never_used, #[cfg(feature = "benchmark")] - &mut windmill_common::bench::BenchmarkIter::new(), + &mut windmill_worker::bench::BenchmarkIter::new(), ) .await; } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index b2d09e7aea..59b700ef24 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -14,6 +14,7 @@ use quick_cache::sync::Cache; use serde_json::value::RawValue; use sqlx::Pool; use std::collections::HashMap; +use std::iter; use std::ops::{Deref, DerefMut}; use std::str::FromStr; #[cfg(feature = "prometheus")] @@ -87,7 +88,7 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED}; use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL}; use windmill_queue::{ cancel_job, get_result_and_success_by_id_from_flow, job_is_complete, push, PushArgs, - PushArgsOwned, PushIsolationLevel, + PushArgsOwned, PushIsolationLevel, RawJob, }; #[cfg(feature = "prometheus")] @@ -4678,7 +4679,7 @@ async fn add_batch_jobs( dedicated_worker, custom_concurrency_key, concurrent_limit, - concurrent_time_window_s, + concurrency_time_window_s, timeout, raw_code, raw_lock, @@ -4831,36 +4832,30 @@ async fn add_batch_jobs( }; let mut tx = user_db.begin(&authed).await?; - - let uuids = sqlx::query_scalar!( - r#"WITH uuid_table as ( - select gen_random_uuid() as uuid from generate_series(1, $16) - ) - INSERT INTO v2_job - (id, workspace_id, raw_code, raw_lock, raw_flow, tag, runnable_id, runnable_path, kind, - script_lang, created_by, permissioned_as, permissioned_as_email, concurrent_limit, - concurrency_time_window_s, timeout, args) - (SELECT uuid, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, - ('{ "uuid": "' || uuid || '" }')::jsonb FROM uuid_table) - RETURNING id AS "id!""#, - w_id, - raw_code, - raw_lock, - raw_flow.map(sqlx::types::Json) as Option>, - tag, - hash.map(|h| h.0), - path, - job_kind.clone() as JobKind, - language as ScriptLang, - authed.username, - username_to_permissioned_as(&authed.username), - authed.email, + let uuids = Vec::from_iter(iter::repeat_with(|| ulid::Ulid::new().into()).take(n as usize)); + let args = uuids + .iter() + .map(|uuid| sqlx::types::Json(to_raw_value(&serde_json::json!({ "uuid": uuid })))) + .collect::>(); + tx = RawJob { + created_by: &authed.username, + permissioned_as: &username_to_permissioned_as(&authed.username), + permissioned_as_email: &authed.email, + kind: job_kind, + runnable_id: hash.map(|h| h.0), + runnable_path: path.as_deref(), + script_lang: Some(language), + tag: &tag, concurrent_limit, - concurrent_time_window_s, + concurrency_time_window_s, timeout, - n, - ) - .fetch_all(&mut *tx) + raw_code: raw_code.as_deref(), + raw_lock: raw_lock.as_deref(), + raw_flow: raw_flow.as_ref(), + flow_status: flow_status.as_ref(), + ..RawJob::default() + } + .push_many(tx, &uuids, &w_id, &args) .await?; let uuids = sqlx::query_scalar!( diff --git a/backend/windmill-common/src/bench.rs b/backend/windmill-common/src/bench.rs deleted file mode 100644 index 649a202ddb..0000000000 --- a/backend/windmill-common/src/bench.rs +++ /dev/null @@ -1,246 +0,0 @@ -use crate::{ - worker::{write_file, TMP_DIR}, - DB, -}; -use serde::Serialize; -use tokio::time::Instant; - -#[derive(Serialize)] -pub struct BenchmarkInfo { - #[serde(skip)] - pub start: Instant, - #[serde(skip)] - pub iters: u64, - timings: Vec, - pub iter_durations: Vec, - pub total_duration: Option, -} - -impl BenchmarkInfo { - pub fn new() -> Self { - BenchmarkInfo { - iters: 0, - timings: vec![], - start: Instant::now(), - iter_durations: vec![], - total_duration: None, - } - } - - pub fn add_iter(&mut self, bench: BenchmarkIter, inc_iters: bool) { - if inc_iters { - self.iters += 1; - } - let elapsed_total = bench.start.elapsed().as_nanos() as u64; - self.timings.push(bench); - self.iter_durations.push(elapsed_total); - } - - pub fn write_to_file(&mut self, path: &str) -> anyhow::Result<()> { - let total_duration = self.start.elapsed().as_millis() as u64; - self.total_duration = Some(total_duration as u64); - - println!( - "Writing benchmark {path}, duration of benchmark: {total_duration}s and RPS: {}", - self.iters as f64 / total_duration as f64 - ); - write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling"); - Ok(()) - } -} - -#[derive(Serialize)] -pub struct BenchmarkIter { - #[serde(skip)] - pub start: Instant, - #[serde(skip)] - last_instant: Instant, - last_step: String, - timings: Vec<(String, u32)>, -} - -impl BenchmarkIter { - pub fn new() -> Self { - BenchmarkIter { - last_instant: Instant::now(), - timings: vec![], - start: Instant::now(), - last_step: String::new(), - } - } - - pub fn add_timing(&mut self, name: &str) { - let elapsed = self.last_instant.elapsed().as_nanos() as u32; - self.timings - .push((format!("{}->{}", self.last_step, name), elapsed)); - self.last_instant = Instant::now(); - self.last_step = name.to_string(); - } -} - -pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) { - use crate::{jobs::JobKind, scripts::ScriptLang}; - - let benchmark_kind = std::env::var("BENCHMARK_KIND").unwrap_or("noop".to_string()); - - if benchmark_jobs > 0 { - let mut tx = db.begin().await.unwrap(); - match benchmark_kind.as_str() { - "dedicated" => { - // you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47 - let hash = sqlx::query_scalar!( - "SELECT hash FROM script WHERE path = $1 AND workspace_id = $2", - "f/benchmarks/dedicated", - "admins" - ) - .fetch_one(&mut *tx) - .await - .unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")); - let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id", - hash, - "f/benchmarks/dedicated", - JobKind::Script as JobKind, - ScriptLang::Bun as ScriptLang, - "admins:f/benchmarks/dedicated", - "admin", - "u/admin", - "admin@windmill.dev", - "admins", - benchmark_jobs - ) - .fetch_all(&mut *tx) - .await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (1)")); - sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "admins:f/benchmarks/dedicated") - .execute(&mut *tx) - .await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (2)")); - sqlx::query!( - "INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", - &uuids - ) - .execute(&mut *tx) - .await - .unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (3)")); - } - "parallelflow" => { - //create dedicated script - sqlx::query!("INSERT INTO script (summary, description, dedicated_worker, content, workspace_id, path, hash, language, tag, created_by, lock) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '') ON CONFLICT (workspace_id, hash) DO NOTHING", - "export async function main() { - console.log('hello world'); - }", - "admins", - "u/admin/parallelflow", - 1234567890, - ScriptLang::Deno as ScriptLang, - "flow", - "admin", - ) - .execute(&mut *tx) - .await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs {_e:#}")); - let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_flow) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, 1)) RETURNING id", - None::, - None::, - JobKind::FlowPreview as JobKind, - ScriptLang::Deno as ScriptLang, - "flow", - "admin", - "u/admin", - "admin@windmill.dev", - "admins", - serde_json::from_str::(r#" -{ - "modules": [ - { - "id": "a", - "value": { - "type": "forloopflow", - "modules": [ - { - "id": "b", - "value": { - "path": "u/admin/parallelflow", - "type": "script", - "tag_override": "", - "input_transforms": {} - }, - "summary": "calctest" - } - ], - "iterator": { - "expr": "[...new Array(300)]", - "type": "javascript" - }, - "parallel": true, - "parallelism": 10, - "skip_failures": true - } - } - ], - "preprocessor_module": null -} - "#).unwrap(), - ) - .fetch_all(&mut *tx) - .await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (1)")); - sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "flow") - .execute(&mut *tx) - .await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (2)")); - sqlx::query!( - "INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", - &uuids - ) - .execute(&mut *tx) - .await - .unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (3)")); - sqlx::query!("INSERT INTO v2_job_flow_runtime (id, flow_status) SELECT unnest($1::uuid[]), $2", &uuids, - serde_json::from_str::(r#" -{ - "step": 0, - "modules": [ - { - "id": "a", - "type": "WaitingForPriorSteps" - } - ], - "cleanup_module": {}, - "failure_module": { - "id": "failure", - "type": "WaitingForPriorSteps" - }, - "preprocessor_module": null - } - - "#).unwrap() - ) - .execute(&mut *tx) - .await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (4)")); - } - _ => { - let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id", - None::, - None::, - JobKind::Noop as JobKind, - ScriptLang::Deno as ScriptLang, - "deno", - "admin", - "u/admin", - "admin@windmill.dev", - "admins", - benchmark_jobs - ) - .fetch_all(&mut *tx) - .await.unwrap_or_else(|_e| panic!("failed to insert noop jobs (1)")); - sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "deno") - .execute(&mut *tx) - .await.unwrap_or_else(|_e| panic!("failed to insert noop jobs (2)")); - sqlx::query!( - "INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", - &uuids - ) - .execute(&mut *tx) - .await - .unwrap_or_else(|_e| panic!("failed to insert noop jobs (3)")); - } - } - tx.commit().await.unwrap(); - } -} diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 5a78466627..01b135ddfe 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -46,6 +46,22 @@ pub enum JobKind { AppScript, } +#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone)] +#[sqlx(type_name = "JOB_TRIGGER_KIND", rename_all = "lowercase")] +#[serde(rename_all(serialize = "lowercase"))] +pub enum JobTriggerKind { + Webhook, + Http, + Websocket, + Kafka, + Email, + Nats, + Schedule, + App, + Ui, + Postgres, +} + impl JobKind { pub fn is_flow(&self) -> bool { matches!( diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index e47c4fc87f..710bbc0c36 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -18,8 +18,6 @@ use sqlx::{Pool, Postgres}; pub mod apps; pub mod auth; -#[cfg(feature = "benchmark")] -pub mod bench; pub mod cache; pub mod db; pub mod ee; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 5bda00c458..fa25593713 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -41,8 +41,8 @@ use windmill_common::{ add_virtual_items_if_necessary, FlowModule, FlowModuleValue, FlowValue, InputTransform, }, jobs::{ - get_payload_tag_from_prefixed_path, JobKind, JobPayload, QueuedJob, RawCode, - ENTRYPOINT_OVERRIDE, PREPROCESSOR_FAKE_ENTRYPOINT, + get_payload_tag_from_prefixed_path, JobKind, JobPayload, JobTriggerKind, QueuedJob, + RawCode, ENTRYPOINT_OVERRIDE, PREPROCESSOR_FAKE_ENTRYPOINT, }, schedule::Schedule, scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang}, @@ -2740,6 +2740,184 @@ lazy_static::lazy_static! { pub static ref RE_ARG_TAG: Regex = Regex::new(r#"\$args\[(\w+)\]"#).unwrap(); } +pub struct RawJob<'a> { + pub created_by: &'a str, + pub permissioned_as: &'a str, + pub permissioned_as_email: &'a str, + pub kind: JobKind, + pub runnable_id: Option, + pub runnable_path: Option<&'a str>, + pub parent_job: Option, + pub script_lang: Option, + pub flow_step_id: Option<&'a str>, + pub flow_root_job: Option, + pub trigger: Option<&'a str>, + pub trigger_kind: Option, + pub tag: &'a str, + pub same_worker: bool, + pub visible_to_owner: bool, + pub concurrent_limit: Option, + pub concurrency_time_window_s: Option, + pub cache_ttl: Option, + pub timeout: Option, + pub priority: Option, + pub pre_run_error: Option<&'a str>, + pub raw_code: Option<&'a str>, + pub raw_lock: Option<&'a str>, + pub raw_flow: Option<&'a FlowValue>, + pub flow_status: Option<&'a FlowStatus>, + pub scheduled_for: Option>, + pub running: bool, +} + +impl<'a> RawJob<'a> { + pub const fn default() -> Self { + Self { + created_by: "missing", + permissioned_as: "g/all", + permissioned_as_email: "missing@email.xyz", + kind: JobKind::Script, + runnable_id: None, + runnable_path: None, + parent_job: None, + script_lang: None, + flow_step_id: None, + flow_root_job: None, + trigger: None, + trigger_kind: None, + tag: "other", + same_worker: false, + visible_to_owner: false, + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + timeout: None, + priority: None, + pre_run_error: None, + raw_code: None, + raw_lock: None, + raw_flow: None, + flow_status: None, + scheduled_for: None, + running: false, + } + } + + pub async fn push<'c>( + &self, + tx: sqlx::PgTransaction<'c>, + id: Uuid, + workspace_id: &str, + args: Json>, + ) -> error::Result> { + self.push_many(tx, &[id], workspace_id, &[args]).await + } + + pub async fn push_many<'c>( + &self, + mut tx: sqlx::PgTransaction<'c>, + id: &[Uuid], + workspace_id: &str, + args: &[Json>], + ) -> error::Result> { + if id.len() > args.len() { + return Err(Error::InternalErr( + "args must be at least as long as id".to_string(), + )); + } + let created_at = Utc::now(); + sqlx::query!( + "INSERT INTO v2_job ( + id, workspace_id, + created_at, created_by, permissioned_as, permissioned_as_email, + kind, runnable_id, runnable_path, parent_job, script_lang, + flow_step_id, flow_root_job, + trigger, trigger_kind, + tag, same_worker, visible_to_owner, concurrent_limit, concurrency_time_window_s, + cache_ttl, timeout, priority, + args, pre_run_error, + raw_code, raw_lock, raw_flow + ) SELECT + unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, + $16, $17, $18, $19, $20, $21, $22, $23, unnest($24::jsonb[]), $25, $26, $27, $28 + ", + id, + workspace_id, + created_at, + self.created_by, + self.permissioned_as, + self.permissioned_as_email, + self.kind as JobKind, + self.runnable_id, + self.runnable_path, + self.parent_job, + self.script_lang as Option, + self.flow_step_id, + self.flow_root_job, + self.trigger, + self.trigger_kind as Option, + self.tag, + self.same_worker, + self.visible_to_owner, + self.concurrent_limit, + self.concurrency_time_window_s, + self.cache_ttl, + self.timeout, + self.priority, + args as &[Json>], + self.pre_run_error, + self.raw_code, + self.raw_lock, + self.raw_flow.map(Json) as Option>, + ) + .execute(&mut *tx) + .warn_after_seconds(1) + .await?; + sqlx::query!( + "INSERT INTO v2_job_queue ( + id, workspace_id, started_at, scheduled_for, running, created_at, tag, priority + ) SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8", + id, + workspace_id, + self.running.then(|| created_at), + self.scheduled_for.unwrap_or(created_at), + self.running, + created_at, + self.tag, + self.priority, + ) + .execute(&mut *tx) + .warn_after_seconds(1) + .await?; + sqlx::query!( + "INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", + id + ) + .execute(&mut *tx) + .warn_after_seconds(1) + .await?; + if let Some(flow_status) = self.flow_status { + sqlx::query!( + "INSERT INTO v2_job_flow_runtime (id, flow_status) SELECT unnest($1::uuid[]), $2", + id, + Json(flow_status) as Json<&FlowStatus>, + ) + .execute(&mut *tx) + .warn_after_seconds(1) + .await?; + } + + #[cfg(feature = "prometheus")] + if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { + // TODO: technically the job isn't queued yet, as the transaction can be rolled back. + // Should be solved when moving these metrics to the queue abstraction. + QUEUE_PUSH_COUNT.inc_by(id.len() as u64); + } + + Ok(tx) + } +} + // #[instrument(level = "trace", skip_all)] pub async fn push<'c, 'd>( _db: &Pool, @@ -3666,89 +3844,56 @@ pub async fn push<'c, 'd>( None }; - let raw_flow = raw_flow.map(Json); - - sqlx::query!( - "INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job, - created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger, - script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner, - flow_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, - cache_ttl, priority, trigger_kind) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, - $19, $20, $21, $22, $23, $24, $25, $26, - CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END)", - job_id, - workspace_id, - raw_code, - raw_lock, - raw_flow as Option>, - tag, + let pre_run_error = pre_run_error.map(Error::to_string); + let args = serde_json::value::to_raw_value(&args).map_err(|e| { + Error::InternalErr(format!("Could not serialize args for job {job_id}: {e:#}")) + })?; + tracing::debug!( + "Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, \ + script_path: {script_path:?}, email {email}, workspace_id {workspace_id}" + ); + tx = RawJob { + created_by: &user, + permissioned_as: &permissioned_as, + permissioned_as_email: &email, + kind: job_kind, + runnable_id: script_hash, + runnable_path: script_path.as_deref(), parent_job, - user, - permissioned_as, - script_hash, - script_path.clone(), - Json(args) as Json, - job_kind.clone() as JobKind, - schedule_path, - language as Option, + script_lang: language, + flow_step_id: flow_step_id.as_deref(), + flow_root_job: root_job, + trigger: schedule_path.as_deref(), + trigger_kind: schedule_path + .as_ref() + .and_then(|_| Some(JobTriggerKind::Schedule)), + tag: &tag, same_worker, - pre_run_error.map(|e| e.to_string()), - email, visible_to_owner, - root_job, concurrent_limit, - if concurrent_limit.is_some() { - concurrency_time_window_s - } else { - None - }, - custom_timeout, - flow_step_id, + + concurrency_time_window_s, cache_ttl, - final_priority, - ) - .execute(&mut *tx) - .warn_after_seconds(1) - .await?; - - tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}"); - let uuid = sqlx::query_scalar!( - "INSERT INTO v2_job_queue - (workspace_id, id, running, scheduled_for, started_at, tag, priority) - VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) \ - RETURNING id AS \"id!\"", - workspace_id, - job_id, - is_running, - scheduled_for_o, - tag, - final_priority, - ) - .fetch_one(&mut *tx) - .warn_after_seconds(1) + timeout: custom_timeout, + priority: final_priority, + pre_run_error: pre_run_error.as_deref(), + raw_code: raw_code.as_deref(), + raw_lock: raw_lock.as_deref(), + raw_flow: raw_flow.as_ref(), + flow_status: flow_status.as_ref(), + scheduled_for: scheduled_for_o, + running: is_running, + } + .push(tx, job_id, workspace_id, Json(args)) .await - .map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?; - - sqlx::query!("INSERT INTO v2_job_runtime (id) VALUES ($1)", job_id) - .execute(&mut *tx) - .await?; - if let Some(flow_status) = flow_status { - sqlx::query!( - "INSERT INTO v2_job_flow_runtime (id, flow_status) VALUES ($1, $2)", - job_id, - Json(flow_status) as Json, - ) - .execute(&mut *tx) - .await?; - } - + .map_err(|e| { + Error::InternalErr(format!( + "Could not insert into queue {job_id} with tag {tag}, \ + schedule_path {schedule_path:?}, script_path: {script_path:?}, \ + email {email}, workspace_id {workspace_id}: {e:#}" + )) + })?; tracing::debug!("Pushed {job_id}"); - // TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction. - #[cfg(feature = "prometheus")] - if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { - QUEUE_PUSH_COUNT.inc(); - } if JOB_TOKEN.is_none() { let job_authed = match authed { @@ -3858,7 +4003,7 @@ pub async fn push<'c, 'd>( .await?; } - Ok((uuid, tx)) + Ok((job_id, tx)) } pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value { diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 7bec644550..ca00a3185f 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -102,6 +102,7 @@ object_store = { workspace = true, optional = true} convert_case.workspace = true yaml-rust.workspace = true backon.workspace = true +ulid.workspace = true opentelemetry = { workspace = true, optional = true } bollard = { workspace = true, optional = true } diff --git a/backend/windmill-worker/src/bench.rs b/backend/windmill-worker/src/bench.rs new file mode 100644 index 0000000000..a22ff71ea7 --- /dev/null +++ b/backend/windmill-worker/src/bench.rs @@ -0,0 +1,226 @@ +use serde::Serialize; +use serde_json::json; +use tokio::time::Instant; +use windmill_common::{ + worker::{write_file, TMP_DIR}, + DB, +}; + +#[derive(Serialize)] +pub struct BenchmarkInfo { + #[serde(skip)] + pub start: Instant, + #[serde(skip)] + pub iters: u64, + timings: Vec, + pub iter_durations: Vec, + pub total_duration: Option, +} + +impl BenchmarkInfo { + pub fn new() -> Self { + BenchmarkInfo { + iters: 0, + timings: vec![], + start: Instant::now(), + iter_durations: vec![], + total_duration: None, + } + } + + pub fn add_iter(&mut self, bench: BenchmarkIter, inc_iters: bool) { + if inc_iters { + self.iters += 1; + } + let elapsed_total = bench.start.elapsed().as_nanos() as u64; + self.timings.push(bench); + self.iter_durations.push(elapsed_total); + } + + pub fn write_to_file(&mut self, path: &str) -> anyhow::Result<()> { + let total_duration = self.start.elapsed().as_millis() as u64; + self.total_duration = Some(total_duration as u64); + + println!( + "Writing benchmark {path}, duration of benchmark: {total_duration}s and RPS: {}", + self.iters as f64 / total_duration as f64 + ); + write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling"); + Ok(()) + } +} + +#[derive(Serialize)] +pub struct BenchmarkIter { + #[serde(skip)] + pub start: Instant, + #[serde(skip)] + last_instant: Instant, + last_step: String, + timings: Vec<(String, u32)>, +} + +impl BenchmarkIter { + pub fn new() -> Self { + BenchmarkIter { + last_instant: Instant::now(), + timings: vec![], + start: Instant::now(), + last_step: String::new(), + } + } + + pub fn add_timing(&mut self, name: &str) { + let elapsed = self.last_instant.elapsed().as_nanos() as u32; + self.timings + .push((format!("{}->{}", self.last_step, name), elapsed)); + self.last_instant = Instant::now(); + self.last_step = name.to_string(); + } +} + +pub async fn benchmark_init(benchmark_jobs: usize, db: &DB) { + use std::iter; + + use windmill_common::{jobs::JobKind, scripts::ScriptLang}; + use windmill_queue::RawJob; + + let benchmark_kind = std::env::var("BENCHMARK_KIND").unwrap_or("noop".to_string()); + let uuids = Vec::from_iter(iter::repeat_with(|| ulid::Ulid::new().into()).take(benchmark_jobs)); + + if !uuids.is_empty() { + match benchmark_kind.as_str() { + "dedicated" => { + // you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47 + let hash = sqlx::query_scalar!( + "SELECT hash FROM script WHERE path = $1 AND workspace_id = $2", + "f/benchmarks/dedicated", + "admins" + ) + .fetch_one(db) + .await + .unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")); + RawJob { + runnable_id: Some(hash), + runnable_path: Some("f/benchmarks/dedicated"), + kind: JobKind::Script, + script_lang: Some(ScriptLang::Bun), + tag: "admins:f/benchmarks/dedicated", + created_by: "admin", + permissioned_as: "u/admin", + permissioned_as_email: "admin@windmill.dev", + ..RawJob::default() + } + .push_many(db.begin().await.unwrap(), uuids.as_slice(), "admins", &[]) + .await + .unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")) + .commit() + .await + .unwrap_or_else(|_e| panic!("failed to commit insert of dedicated jobs")); + } + "parallelflow" => { + // create dedicated script + sqlx::query!( + "INSERT INTO script ( + summary, description, dedicated_worker, content, workspace_id, path, hash, + language, tag, created_by, lock + ) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '') + ON CONFLICT (workspace_id, hash) DO NOTHING", + "export async function main() { + console.log('hello world'); + }", + "admins", + "u/admin/parallelflow", + 1234567890, + ScriptLang::Deno as ScriptLang, + "flow", + "admin", + ) + .execute(db) + .await + .unwrap_or_else(|e| panic!("failed to insert parallelflow script {e:#}")); + RawJob { + kind: JobKind::FlowPreview, + tag: "flow", + created_by: "admin", + permissioned_as: "u/admin", + permissioned_as_email: "admin@windmill.dev", + raw_flow: Some( + &serde_json::from_value(json!({ + "modules": [ + { + "id": "a", + "value": { + "type": "forloopflow", + "modules": [ + { + "id": "b", + "value": { + "path": "u/admin/parallelflow", + "type": "script", + "tag_override": "", + "input_transforms": {} + }, + "summary": "calctest" + } + ], + "iterator": { + "expr": "[...new Array(300)]", + "type": "javascript" + }, + "parallel": true, + "parallelism": 10, + "skip_failures": true + } + } + ], + "preprocessor_module": null + })) + .unwrap(), + ), + flow_status: Some( + &serde_json::from_value(json!({ + "step": 0, + "modules": [ + { + "id": "a", + "type": "WaitingForPriorSteps" + } + ], + "cleanup_module": {}, + "failure_module": { + "id": "failure", + "type": "WaitingForPriorSteps" + }, + "preprocessor_module": null + })) + .unwrap(), + ), + ..RawJob::default() + } + .push_many(db.begin().await.unwrap(), uuids.as_slice(), "admins", &[]) + .await + .unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs")) + .commit() + .await + .unwrap_or_else(|_e| panic!("failed to commit insert of parallelflow jobs")); + } + _ => { + RawJob { + kind: JobKind::Noop, + tag: "deno", + created_by: "admin", + permissioned_as: "u/admin", + permissioned_as_email: "admin@windmill.dev", + ..RawJob::default() + } + .push_many(db.begin().await.unwrap(), uuids.as_slice(), "admins", &[]) + .await + .unwrap_or_else(|_e| panic!("failed to insert noop jobs")) + .commit() + .await + .unwrap_or_else(|_e| panic!("failed to commit insert of noop jobs")); + } + } + } +} diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index cc34dd72ab..6fbb9538e6 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -9,6 +9,8 @@ mod snowflake_executor; mod ansible_executor; mod bash_executor; +#[cfg(feature = "benchmark")] +pub mod bench; mod bun_executor; pub mod common; mod config; diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 3449c2ab24..3784612cf3 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -26,7 +26,7 @@ use windmill_common::{ }; #[cfg(feature = "benchmark")] -use windmill_common::bench::{BenchmarkInfo, BenchmarkIter}; +use crate::bench::{BenchmarkInfo, BenchmarkIter}; use windmill_queue::{append_logs, get_queued_job, CanceledBy, WrappedError}; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 65d4b5d1d9..262bede599 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -146,7 +146,7 @@ use crate::mssql_executor::do_mssql; use crate::bigquery_executor::do_bigquery; #[cfg(feature = "benchmark")] -use windmill_common::bench::{benchmark_init, BenchmarkInfo, BenchmarkIter}; +use crate::bench::{benchmark_init, BenchmarkInfo, BenchmarkIter}; use windmill_common::add_time; @@ -1050,9 +1050,9 @@ pub async fn run_worker( let is_dedicated_worker: bool = WORKER_CONFIG.read().await.dedicated_worker.is_some(); #[cfg(feature = "benchmark")] - let benchmark_jobs: i32 = std::env::var("BENCHMARK_JOBS") + let benchmark_jobs: usize = std::env::var("BENCHMARK_JOBS") .unwrap_or("5000".to_string()) - .parse::() + .parse::() .unwrap(); #[cfg(feature = "benchmark")] diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 708aed3c17..7d80b2e8cf 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -11,6 +11,8 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::Duration; +#[cfg(feature = "benchmark")] +use crate::bench::BenchmarkIter; use crate::common::{cached_result_path, save_in_cache}; use crate::js_eval::{eval_timeout, IdContext}; use crate::{ @@ -29,8 +31,6 @@ use tracing::instrument; use uuid::Uuid; use windmill_common::add_time; use windmill_common::auth::JobPerms; -#[cfg(feature = "benchmark")] -use windmill_common::bench::BenchmarkIter; use windmill_common::cache::{self, RawData}; use windmill_common::db::Authed; use windmill_common::flow_status::{