diff --git a/backend/migrations/20230709151109_add_script_concurrency_limit.down.sql b/backend/migrations/20230709151109_add_script_concurrency_limit.down.sql new file mode 100644 index 0000000000..7cc2e00932 --- /dev/null +++ b/backend/migrations/20230709151109_add_script_concurrency_limit.down.sql @@ -0,0 +1,3 @@ +-- Add down migration script here +ALTER TABLE script DROP COLUMN concurrent_limit; +ALTER TABLE script DROP COLUMN concurrency_time_window_s; diff --git a/backend/migrations/20230709151109_add_script_concurrency_limit.up.sql b/backend/migrations/20230709151109_add_script_concurrency_limit.up.sql new file mode 100644 index 0000000000..20b87fc388 --- /dev/null +++ b/backend/migrations/20230709151109_add_script_concurrency_limit.up.sql @@ -0,0 +1,5 @@ +-- Add up migration script here +ALTER TABLE script ADD COLUMN concurrent_limit INTEGER; +ALTER TABLE script ADD COLUMN concurrency_time_window_s INTEGER; +ALTER TABLE queue ADD COLUMN concurrent_limit INTEGER; +ALTER TABLE queue ADD COLUMN concurrency_time_window_s INTEGER; diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 6d04ebcb37..f4aad1877d 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -276,6 +276,33 @@ }, "query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1 WHERE worker = $2" }, + "077b925c0a9f19a3859ca7eb0bc9049fc7c303daf469c049cf03570d22310d74": { + "describe": { + "columns": [ + { + "name": "min_started_at", + "ordinal": 0, + "type_info": "Timestamptz" + }, + { + "name": "total_count", + "ordinal": 1, + "type_info": "Int8" + } + ], + "nullable": [ + true, + true + ], + "parameters": { + "Left": [ + "Text", + "Float8" + ] + } + }, + "query": "SELECT COALESCE(j.min_started_at, q.min_started_at) AS min_started_at, COALESCE(completed_count, 0) + COALESCE(running_count, 0) AS total_count\n FROM\n (SELECT script_path, MIN(started_at) as min_started_at, COUNT(*) as completed_count\n FROM completed_job\n WHERE script_path = $1 AND started_at + INTERVAL '1 MILLISECOND' * duration_ms > (now() - INTERVAL '1 second' * $2)\n GROUP BY script_path) as j\n FULL OUTER JOIN\n (SELECT script_path, MIN(started_at) as min_started_at, COUNT(*) as running_count\n FROM queue\n WHERE script_path = $1 AND running = true\n GROUP BY script_path) as q\n ON q.script_path = j.script_path" + }, "0784bb86a503f02b9ef247a2b83a82ddfa49632552b223a9f4536a449b0a1eb8": { "describe": { "columns": [ @@ -1230,7 +1257,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun" ] }, @@ -1392,33 +1418,6 @@ }, "query": "UPDATE script SET archived = true WHERE hash = $1 AND workspace_id = $2" }, - "2b273ecd84e8fa58825659060f09c362ab61e944a2f37fa02396dbbd61b0c04e": { - "describe": { - "columns": [ - { - "name": "hash", - "ordinal": 0, - "type_info": "Int8" - }, - { - "name": "tag", - "ordinal": 1, - "type_info": "Varchar" - } - ], - "nullable": [ - false, - true - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - } - }, - "query": "select hash, tag from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false AND lock IS not NULL AND lock_error_logs IS NULL)" - }, "2e4115bb2e6c8c85ad1492ad135d6b0454b342126cb5fa17e58caf71b32ee755": { "describe": { "columns": [], @@ -2056,85 +2055,6 @@ }, "query": "UPDATE queue SET canceled = true, canceled_reason = 'http connection broke', canceled_by = $1 WHERE id = $2 AND workspace_id = $3" }, - "468466f5a2f2c0cafc5b62b6550d51951bc67c43b4eaea34b2f88680f20be25a": { - "describe": { - "columns": [ - { - "name": "id", - "ordinal": 0, - "type_info": "Uuid" - } - ], - "nullable": [ - false - ], - "parameters": { - "Left": [ - "Varchar", - "Uuid", - "Bool", - "Uuid", - "Varchar", - "Varchar", - "Timestamptz", - "Int8", - "Varchar", - "Text", - "Text", - "Jsonb", - { - "Custom": { - "kind": { - "Enum": [ - "script", - "preview", - "flow", - "dependencies", - "flowpreview", - "script_hub", - "identity", - "flowdependencies", - "http", - "graphql", - "postgresql", - "noop" - ] - }, - "name": "job_kind" - } - }, - "Varchar", - "Jsonb", - "Jsonb", - "Bool", - { - "Custom": { - "kind": { - "Enum": [ - "python3", - "deno", - "go", - "bash", - "postgresql", - "nativets", - "Nativets", - "bun" - ] - }, - "name": "script_lang" - } - }, - "Bool", - "Text", - "Varchar", - "Bool", - "Uuid", - "Varchar" - ] - } - }, - "query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24) RETURNING id" - }, "46b44de4ea6eb15258c1cb9bbd21cab7fc350b46e8899d59759945eae02a6d40": { "describe": { "columns": [ @@ -2569,6 +2489,86 @@ }, "query": "INSERT INTO token\n (token, email, label, expiration, super_admin)\n VALUES ($1, $2, $3, $4, $5)" }, + "550d1918df914b1b8f61437cab8fc1b5e5278e1499546f1c2d5040aea030b075": { + "describe": { + "columns": [ + { + "name": "id", + "ordinal": 0, + "type_info": "Uuid" + } + ], + "nullable": [ + false + ], + "parameters": { + "Left": [ + "Varchar", + "Uuid", + "Bool", + "Uuid", + "Varchar", + "Varchar", + "Timestamptz", + "Int8", + "Varchar", + "Text", + "Text", + "Jsonb", + { + "Custom": { + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop" + ] + }, + "name": "job_kind" + } + }, + "Varchar", + "Jsonb", + "Jsonb", + "Bool", + { + "Custom": { + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun" + ] + }, + "name": "script_lang" + } + }, + "Bool", + "Text", + "Varchar", + "Bool", + "Uuid", + "Varchar", + "Int4", + "Int4" + ] + } + }, + "query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26) RETURNING id" + }, "567ded2a717af9370a80c00bdb50d965fba9a3422c58e67ed2ed06dd107ae139": { "describe": { "columns": [], @@ -2794,7 +2794,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun" ] }, @@ -2909,33 +2908,6 @@ }, "query": "SELECT value FROM flow WHERE path = $1 AND workspace_id = $2" }, - "6b1f09324392c495d5d3ca518cfcb4a88d5dea993fd19ffbc7dc2c1e82fc4d12": { - "describe": { - "columns": [ - { - "name": "path", - "ordinal": 0, - "type_info": "Varchar" - }, - { - "name": "tag", - "ordinal": 1, - "type_info": "Varchar" - } - ], - "nullable": [ - false, - true - ], - "parameters": { - "Left": [ - "Int8", - "Text" - ] - } - }, - "query": "select path, tag from script where hash = $1 AND workspace_id = $2" - }, "6c63bbcb45d3f51eccaea52ec862700e1f1c2426d823abd951e1eea4fd9b85aa": { "describe": { "columns": [], @@ -3137,26 +3109,44 @@ }, "query": "DELETE FROM group_ WHERE workspace_id = $1" }, - "74c11c1e64cbfbcf316f8152576c34303b411fff547e31b286eddea743a60031": { + "7435be9a71a6b30980d180047486a47d69e36132d40ef0f53448e96a9866842b": { "describe": { "columns": [ { - "name": "tag", + "name": "hash", "ordinal": 0, + "type_info": "Int8" + }, + { + "name": "tag", + "ordinal": 1, "type_info": "Varchar" + }, + { + "name": "concurrent_limit", + "ordinal": 2, + "type_info": "Int4" + }, + { + "name": "concurrency_time_window_s", + "ordinal": 3, + "type_info": "Int4" } ], "nullable": [ + false, + true, + true, true ], "parameters": { "Left": [ - "Int8", + "Text", "Text" ] } }, - "query": "select tag from script where hash = $1 AND workspace_id = $2" + "query": "select hash, tag, concurrent_limit, concurrency_time_window_s from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false AND lock IS not NULL AND lock_error_logs IS NULL)" }, "75e880f9d9fbda36c2314706923cef36e4667d930fb8ee1876dd9ce1c92396b2": { "describe": { @@ -3905,7 +3895,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun" ] }, @@ -4111,6 +4100,39 @@ }, "query": "DELETE FROM usr WHERE workspace_id = $1 AND username = $2" }, + "8f0d98661f2a32659f37a123320d14288ce88c9d6e6279bdc36334410023f5e1": { + "describe": { + "columns": [ + { + "name": "tag", + "ordinal": 0, + "type_info": "Varchar" + }, + { + "name": "concurrent_limit", + "ordinal": 1, + "type_info": "Int4" + }, + { + "name": "concurrency_time_window_s", + "ordinal": 2, + "type_info": "Int4" + } + ], + "nullable": [ + true, + true, + true + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + } + }, + "query": "select tag, concurrent_limit, concurrency_time_window_s from script where hash = $1 AND workspace_id = $2" + }, "902961f15b8c7603dddf2933b5fc7cdd6e5af3545835763ccb29cdf3ac273ef0": { "describe": { "columns": [], @@ -4330,6 +4352,45 @@ }, "query": "SELECT is_admin FROM usr where username = $1 AND workspace_id = $2 AND disabled = false" }, + "9289709b2727425be0e7f1e341d6acd46e1e79d5018f6a18cb41d50ca3f4ba11": { + "describe": { + "columns": [ + { + "name": "hash", + "ordinal": 0, + "type_info": "Int8" + }, + { + "name": "tag", + "ordinal": 1, + "type_info": "Varchar" + }, + { + "name": "concurrent_limit", + "ordinal": 2, + "type_info": "Int4" + }, + { + "name": "concurrency_time_window_s", + "ordinal": 3, + "type_info": "Int4" + } + ], + "nullable": [ + false, + true, + true, + true + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + } + }, + "query": "select hash, tag, concurrent_limit, concurrency_time_window_s from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)" + }, "930ad84a4db26fa6d2c8447d447099e944a0ea7b1266b4d02cee620fe3d761a5": { "describe": { "columns": [], @@ -4360,6 +4421,26 @@ }, "query": "INSERT INTO folder\n (workspace_id, name, display_name, owners, extra_perms)\n VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING" }, + "967ac1b17291ca80d40acc77e3ba01210bfbd0da18ea7daa61d7f3d6aa6bbcc1": { + "describe": { + "columns": [ + { + "name": "avg_duration_s", + "ordinal": 0, + "type_info": "Int4" + } + ], + "nullable": [ + null + ], + "parameters": { + "Left": [ + "Text" + ] + } + }, + "query": "SELECT CAST(ROUND(AVG(duration_ms) / 1000, 0) AS INT) AS avg_duration_s FROM\n (SELECT duration_ms FROM completed_job WHERE script_path = $1\n ORDER BY started_at\n DESC LIMIT 10) AS t" + }, "9681c348429f5e647ab55a4a54933faa64018f8539a143641f10d14ddd3a09c4": { "describe": { "columns": [ @@ -5897,33 +5978,6 @@ }, "query": "INSERT INTO group_\n VALUES ($1, 'all', 'The group that always contains all users of this workspace')" }, - "cd9c254f43fa689245a7e42fe9127d4a53e12fa85e14ae4ab9c1b23691cfdcae": { - "describe": { - "columns": [ - { - "name": "hash", - "ordinal": 0, - "type_info": "Int8" - }, - { - "name": "tag", - "ordinal": 1, - "type_info": "Varchar" - } - ], - "nullable": [ - false, - true - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - } - }, - "query": "select hash, tag from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)" - }, "d0308abac80575038203b60bb66d3b39b586939da0421a595e47c7a759616431": { "describe": { "columns": [], @@ -6040,6 +6094,45 @@ }, "query": "SELECT count(path) FROM app WHERE path LIKE 'f/' || $1 || '%' AND workspace_id = $2" }, + "d164caec6637ab53d8c2e0faad653a458f31a976bb23c3a43d35b8476b066954": { + "describe": { + "columns": [ + { + "name": "path", + "ordinal": 0, + "type_info": "Varchar" + }, + { + "name": "tag", + "ordinal": 1, + "type_info": "Varchar" + }, + { + "name": "concurrent_limit", + "ordinal": 2, + "type_info": "Int4" + }, + { + "name": "concurrency_time_window_s", + "ordinal": 3, + "type_info": "Int4" + } + ], + "nullable": [ + false, + true, + true, + true + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + } + }, + "query": "select path, tag, concurrent_limit, concurrency_time_window_s from script where hash = $1 AND workspace_id = $2" + }, "d243064655e45a3898e39ed151416e4e63ec19292cedddfb408d026cf5218f09": { "describe": { "columns": [ diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 49920821dc..12610e3c00 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1009,6 +1009,8 @@ async fn test_deno_flow(db: Pool) { path: None, lock: None, tag: None, + concurrent_limit: None, + concurrency_time_window_s: None, }, stop_after_if: Default::default(), summary: Default::default(), @@ -1039,6 +1041,8 @@ async fn test_deno_flow(db: Pool) { path: None, lock: None, tag: None, + concurrent_limit: None, + concurrency_time_window_s: None, }, stop_after_if: Default::default(), summary: Default::default(), @@ -1144,6 +1148,8 @@ async fn test_deno_flow_same_worker(db: Pool) { path: None, lock: None, tag: None, + concurrent_limit: None, + concurrency_time_window_s: None, }, stop_after_if: Default::default(), summary: Default::default(), @@ -1185,6 +1191,8 @@ async fn test_deno_flow_same_worker(db: Pool) { path: None, lock: None, tag: None, + concurrent_limit: None, + concurrency_time_window_s: None, }, stop_after_if: Default::default(), summary: Default::default(), @@ -1213,6 +1221,8 @@ async fn test_deno_flow_same_worker(db: Pool) { path: None, lock: None, tag: None, + concurrent_limit: None, + concurrency_time_window_s: None, }, stop_after_if: Default::default(), summary: Default::default(), @@ -1259,6 +1269,8 @@ async fn test_deno_flow_same_worker(db: Pool) { path: None, lock: None, tag: None, + concurrent_limit: None, + concurrency_time_window_s: None, }, stop_after_if: Default::default(), summary: Default::default(), @@ -1564,6 +1576,8 @@ func main(derp string) (string, error) { path: None, lock: None, language: ScriptLang::Go, + concurrent_limit: None, + concurrency_time_window_s: None, })) .arg("derp", json!("world")) .run_until_complete(&db, port) @@ -1591,6 +1605,8 @@ echo "hello $msg" path: None, lock: None, language: ScriptLang::Bash, + concurrent_limit: None, + concurrency_time_window_s: None, })) .arg("msg", json!("world")) .run_until_complete(&db, port) @@ -1616,6 +1632,8 @@ def main(): path: None, language: ScriptLang::Python3, lock: None, + concurrent_limit: None, + concurrency_time_window_s: None, }); let result = run_job_in_new_worker_until_complete(&db, job, port) @@ -1646,6 +1664,8 @@ def main(): path: None, language: ScriptLang::Python3, lock: None, + concurrent_limit: None, + concurrency_time_window_s: None, }); let result = run_job_in_new_worker_until_complete(&db, job, port) @@ -1675,6 +1695,8 @@ def main(): path: None, language: ScriptLang::Python3, lock: None, + concurrent_limit: None, + concurrency_time_window_s: None, }); let result = run_job_in_new_worker_until_complete(&db, job, port) diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 0089ef2a65..00fcfb30d6 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -664,6 +664,8 @@ mod tests { path: None, lock: None, tag: None, + concurrent_limit: None, + concurrency_time_window_s: None, }, stop_after_if: Some(StopAfterIf { expr: "foo = 'bar'".to_string(), diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 8a645a4a2a..0da647d0bd 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -248,13 +248,13 @@ pub async fn get_path_for_hash<'c>( Ok(path) } -pub async fn get_path_and_tag_for_hash<'c>( +pub async fn get_path_tag_and_limits_for_hash<'c>( db: &mut Transaction<'c, Postgres>, w_id: &str, hash: i64, -) -> error::Result<(String, Option)> { +) -> error::Result<(String, Option, Option, Option)> { let script = sqlx::query!( - "select path, tag from script where hash = $1 AND workspace_id = $2", + "select path, tag, concurrent_limit, concurrency_time_window_s from script where hash = $1 AND workspace_id = $2", hash, w_id ) @@ -265,7 +265,7 @@ pub async fn get_path_and_tag_for_hash<'c>( "querying getting path for hash {hash} in {w_id}: {e}" )) })?; - Ok((script.path, script.tag)) + Ok((script.path, script.tag, script.concurrent_limit, script.concurrency_time_window_s)) } async fn get_job( @@ -651,6 +651,8 @@ async fn list_jobs( "suspend", "mem_peak", "tag", + "concurrent_limit", + "concurrency_time_window_s", ], ); let sqlc = list_completed_jobs_query( @@ -687,6 +689,8 @@ async fn list_jobs( "null as suspend", "mem_peak", "tag", + "null as concurrent_limit", + "null as concurrency_time_window_s", ], ); let sql = format!( @@ -1154,6 +1158,8 @@ struct UnifiedJob { suspend: Option, mem_peak: Option, tag: String, + concurrent_limit: Option, + concurrency_time_window_s: Option, } impl From for Job { @@ -1226,6 +1232,8 @@ impl From for Job { root_job: None, leaf_jobs: None, tag: uj.tag, + concurrent_limit: uj.concurrent_limit, + concurrency_time_window_s: uj.concurrency_time_window_s, }), t => panic!("job type {} not valid", t), } @@ -1824,7 +1832,7 @@ pub async fn run_wait_result_script_by_hash( let hash = script_hash.0; let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into(); - let (path, tag) = get_path_and_tag_for_hash(tx.transaction_mut(), &w_id, hash).await?; + let (path, tag, concurrent_limit, concurrency_time_window_s) = get_path_tag_and_limits_for_hash(tx.transaction_mut(), &w_id, hash).await?; check_scopes(&authed, || format!("run:script/{path}"))?; let args = run_query.add_include_headers(headers, args.unwrap_or_default()); @@ -1833,7 +1841,7 @@ pub async fn run_wait_result_script_by_hash( let (uuid, tx) = push( tx, &w_id, - JobPayload::ScriptHash { hash: ScriptHash(hash), path }, + JobPayload::ScriptHash { hash: ScriptHash(hash), path: path, concurrent_limit: concurrent_limit, concurrency_time_window_s: concurrency_time_window_s}, args, &authed.username, &authed.email, @@ -1983,6 +1991,8 @@ async fn run_preview_job( path: preview.path, language: preview.language.unwrap_or(ScriptLang::Deno), lock: None, + concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here + concurrency_time_window_s: None, // TODO(gbouv): same as above }), }, args, @@ -2094,7 +2104,7 @@ pub async fn run_job_by_hash( ) -> error::Result<(StatusCode, String)> { let hash = script_hash.0; let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); - let (path, tag) = get_path_and_tag_for_hash(tx.transaction_mut(), &w_id, hash).await?; + let (path, tag, concurrent_limit, concurrency_time_window_s) = get_path_tag_and_limits_for_hash(tx.transaction_mut(), &w_id, hash).await?; check_scopes(&authed, || format!("run:script/{path}"))?; let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; @@ -2104,7 +2114,7 @@ pub async fn run_job_by_hash( let (uuid, tx) = push( tx, &w_id, - JobPayload::ScriptHash { hash: ScriptHash(hash), path }, + JobPayload::ScriptHash { hash: ScriptHash(hash), path: path, concurrent_limit: concurrent_limit, concurrency_time_window_s: concurrency_time_window_s }, args, &authed.username, &authed.email, diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index fb99f44162..a01864c2eb 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -784,14 +784,14 @@ async fn slack_command( (JobPayload::Flow(path.to_string()), None) } else { let path = path.strip_prefix("script/").unwrap_or_else(|| path); - let (script_hash, tag) = windmill_common::get_latest_deployed_hash_for_path( + let (script_hash, tag, concurrent_limit, concurrency_time_window_s) = windmill_common::get_latest_deployed_hash_for_path( tx.transaction_mut(), &settings.workspace_id, path, ) .await?; ( - JobPayload::ScriptHash { hash: script_hash, path: path.to_owned() }, + JobPayload::ScriptHash { hash: script_hash, path: path.to_owned(), concurrent_limit, concurrency_time_window_s }, tag, ) }; diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 5c95cf1ca7..9c456d9637 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -274,6 +274,10 @@ pub enum FlowModuleValue { #[serde(skip_serializing_if = "Option::is_none")] tag: Option, language: ScriptLang, + #[serde(skip_serializing_if = "Option::is_none")] + concurrent_limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + concurrency_time_window_s: Option, }, Identity, Http, diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index ccebd4d6e6..de76c86c05 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -82,6 +82,10 @@ pub struct QueuedJob { #[serde(skip_serializing_if = "Option::is_none")] pub leaf_jobs: Option, pub tag: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrent_limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_time_window_s: Option, } impl QueuedJob { @@ -144,6 +148,8 @@ impl Default for QueuedJob { root_job: None, leaf_jobs: None, tag: "deno".to_string(), + concurrent_limit: None, + concurrency_time_window_s: None, } } } @@ -151,7 +157,7 @@ impl Default for QueuedJob { #[derive(Debug, Clone)] pub enum JobPayload { ScriptHub { path: String }, - ScriptHash { hash: ScriptHash, path: String }, + ScriptHash { hash: ScriptHash, path: String, concurrent_limit: Option, concurrency_time_window_s: Option }, Code(RawCode), Dependencies { hash: ScriptHash, dependencies: String, language: ScriptLang }, FlowDependencies { path: String }, @@ -168,6 +174,8 @@ pub struct RawCode { pub path: Option, pub language: ScriptLang, pub lock: Option, + pub concurrent_limit: Option, + pub concurrency_time_window_s: Option, } type Tag = String; @@ -180,28 +188,33 @@ pub async fn script_path_to_payload<'c>( let (job_payload, tag) = if script_path.starts_with("hub/") { (JobPayload::ScriptHub { path: script_path.to_owned() }, None) } else { - let (script_hash, tag) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?; + let (script_hash, tag, concurrent_limit, concurrency_time_window_s) = get_latest_deployed_hash_for_path(db, w_id, script_path).await?; ( - JobPayload::ScriptHash { hash: script_hash, path: script_path.to_owned() }, + JobPayload::ScriptHash { hash: script_hash, path: script_path.to_owned(), concurrent_limit, concurrency_time_window_s}, tag, ) }; Ok((job_payload, tag)) } -pub async fn script_hash_to_tag<'c>( +pub async fn script_hash_to_tag_and_limits<'c>( script_hash: &ScriptHash, db: &mut Transaction<'c, Postgres>, w_id: &String, -) -> error::Result> { - Ok(sqlx::query_scalar!( - "select tag from script where hash = $1 AND workspace_id = $2", +) -> error::Result<(Option, Option, Option)> { + let script = sqlx::query!( + "select tag, concurrent_limit, concurrency_time_window_s from script where hash = $1 AND workspace_id = $2", script_hash.0, w_id ) - .fetch_optional(db) - .await? - .flatten()) + .fetch_one(db) + .await + .map_err(|e| { + Error::InternalErr(format!( + "querying getting tag for hash {script_hash}: {e}" + )) + })?; + Ok((script.tag, script.concurrent_limit, script.concurrency_time_window_s)) } pub async fn get_payload_tag_from_prefixed_path<'c>( diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index aafc2b8d7e..995495b36a 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -159,9 +159,9 @@ pub async fn get_latest_deployed_hash_for_path<'c>( db: &mut sqlx::Transaction<'c, sqlx::Postgres>, w_id: &str, script_path: &str, -) -> error::Result<(scripts::ScriptHash, Option)> { +) -> error::Result<(scripts::ScriptHash, Option, Option, Option)> { let r_o = sqlx::query!( - "select hash, tag from script where path = $1 AND workspace_id = $2 AND + "select hash, tag, concurrent_limit, concurrency_time_window_s from script where path = $1 AND workspace_id = $2 AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND deleted = false AND archived = false AND lock IS not NULL AND lock_error_logs IS NULL)", script_path, @@ -172,16 +172,16 @@ pub async fn get_latest_deployed_hash_for_path<'c>( let script = utils::not_found_if_none(r_o, "script", script_path)?; - Ok((scripts::ScriptHash(script.hash), script.tag)) + Ok((scripts::ScriptHash(script.hash), script.tag, script.concurrent_limit, script.concurrency_time_window_s)) } pub async fn get_latest_hash_for_path<'c>( db: &mut sqlx::Transaction<'c, sqlx::Postgres>, w_id: &str, script_path: &str, -) -> error::Result<(scripts::ScriptHash, Option)> { +) -> error::Result<(scripts::ScriptHash, Option, Option, Option)> { let r_o = sqlx::query!( - "select hash, tag from script where path = $1 AND workspace_id = $2 AND + "select hash, tag, concurrent_limit, concurrency_time_window_s from script where path = $1 AND workspace_id = $2 AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND deleted = false AND archived = false)", script_path, @@ -192,5 +192,5 @@ pub async fn get_latest_hash_for_path<'c>( let script = utils::not_found_if_none(r_o, "script", script_path)?; - Ok((scripts::ScriptHash(script.hash), script.tag)) + Ok((scripts::ScriptHash(script.hash), script.tag, script.concurrent_limit, script.concurrency_time_window_s)) } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 8ee0545978..06c02b7fe0 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -10,6 +10,7 @@ use std::{collections::HashMap, vec, sync::atomic::AtomicBool}; use anyhow::Context; use async_recursion::async_recursion; +use chrono::{Duration, DateTime, Utc}; use itertools::Itertools; use reqwest::Client; use rsmq_async::RsmqConnection; @@ -657,7 +658,7 @@ async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c return Ok(tx); } -pub async fn pull( +pub async fn pull( db: &Pool, whitelist_workspaces: Option>, blacklist_workspaces: Option>, @@ -694,6 +695,130 @@ pub async fn pull( // let rs = rd_string(2); // let instant = Instant::now(); + loop { + let tx: QueueTransaction<'_, _> = (rsmq.clone(), db.clone().begin().await?).into(); + let (job, mut tx) = pull_single_job_and_mark_as_running_no_concurrency_limit(tx, workspaces_filter.as_str(), rsmq.clone()).await?; + + if job.is_none() { + return Ok(None); + } + + // concurrency check. If more than X jobs for this path are already running, we re-queue and pull another job from the queue + let pulled_job = job.unwrap(); + if pulled_job.script_path.is_none() || pulled_job.concurrent_limit.is_none() { + if *METRICS_ENABLED { + QUEUE_PULL_COUNT.inc(); + } + tx.commit().await?; + return Ok(Option::Some(pulled_job)) + } + + // Else the job is subject to concurrency limits + let job_script_path = pulled_job.script_path.clone().unwrap(); + + let job_custom_concurrent_limit = pulled_job.concurrent_limit.unwrap(); + // setting concurrency_time_window to 0 will count only the currently running jobs + let job_custom_concurrency_time_window_s = pulled_job.concurrency_time_window_s.unwrap_or(0); + tracing::debug!("Job concurrency limit is {} per {}s", job_custom_concurrent_limit, job_custom_concurrency_time_window_s); + + let script_path_live_stats = sqlx::query!( + "SELECT COALESCE(j.min_started_at, q.min_started_at) AS min_started_at, COALESCE(completed_count, 0) + COALESCE(running_count, 0) AS total_count + FROM + (SELECT script_path, MIN(started_at) as min_started_at, COUNT(*) as completed_count + FROM completed_job + WHERE script_path = $1 AND started_at + INTERVAL '1 MILLISECOND' * duration_ms > (now() - INTERVAL '1 second' * $2) + GROUP BY script_path) as j + FULL OUTER JOIN + (SELECT script_path, MIN(started_at) as min_started_at, COUNT(*) as running_count + FROM queue + WHERE script_path = $1 AND running = true + GROUP BY script_path) as q + ON q.script_path = j.script_path", + job_script_path, + f64::from(job_custom_concurrency_time_window_s), + ) + .fetch_one(&mut tx) + .await + .map_err(|e| { + Error::InternalErr(format!( + "Error getting concurrency count for script path {job_script_path}: {e}" + )) + })?; + + let concurrent_jobs_for_this_script: Option = script_path_live_stats.total_count; + tracing::debug!("Current concurrent jobs for this script: {}", concurrent_jobs_for_this_script.unwrap_or(-1)); + if concurrent_jobs_for_this_script.is_none() || concurrent_jobs_for_this_script.unwrap() < i64::from(job_custom_concurrent_limit) { + if *METRICS_ENABLED { + QUEUE_PULL_COUNT.inc(); + } + tx.commit().await?; + return Ok(Option::Some(pulled_job)); + } + + let job_uuid: Uuid = pulled_job.id; + let min_started_at: Option> = script_path_live_stats.min_started_at; + let avg_script_duration: Option = sqlx::query_scalar!( + "SELECT CAST(ROUND(AVG(duration_ms) / 1000, 0) AS INT) AS avg_duration_s FROM + (SELECT duration_ms FROM completed_job WHERE script_path = $1 + ORDER BY started_at + DESC LIMIT 10) AS t", + job_script_path) + .fetch_one(&mut tx) + .await?; + + // optimal scheduling is: 'older_job_in_concurrency_time_window_started_timestamp + script_avg_duration + concurrency_time_window_s' + let estimated_next_schedule_timestamp = min_started_at.unwrap_or(pulled_job.scheduled_for) + Duration::seconds(avg_script_duration.map(i64::from).unwrap_or(0)) + Duration::seconds(i64::from(job_custom_concurrency_time_window_s)); + tracing::info!("Job '{}' from path '{}' has reached its concurrency limit of {} jobs run in the last {} seconds. This job will be re-queued for next execution at {}", + job_uuid, job_script_path, job_custom_concurrent_limit, job_custom_concurrency_time_window_s, estimated_next_schedule_timestamp); + + let job_log_line_break = '\n'; + let job_log_event = format!("Re-scheduled job to {estimated_next_schedule_timestamp} due to concurrency limits"); + if rsmq.is_some() { + // if let Some(ref mut rsmq) = tx.rsmq { + // if using redis, only one message at a time can be poped from the queue. Process only this message and move to the next elligible job + // In this case, the job might be a job from the same script path, but we can't optimise this further + // if using posgtres, then we're able to re-queue the entire batch of scheduled job for this script_path, so we do it + let _requeued_job = sqlx::query_as::<_, QueuedJob>(&format!( + "UPDATE queue + SET running = false + , started_at = null + , scheduled_for = '{estimated_next_schedule_timestamp}' + , logs = CASE WHEN logs IS NULL OR logs = '' THEN '{job_log_event}'::text WHEN logs LIKE '%{job_log_event}' THEN logs ELSE concat(logs, '{job_log_line_break}{job_log_event}'::text) END + WHERE id = '{job_uuid}' + RETURNING *" + )) + .fetch_one(&mut tx) + .await + .map_err(|e| Error::InternalErr(format!("Could not update and re-queue job {job_uuid}. The job will be marked as running but it is not running: {e}")))?; + + if let Some(ref mut rsmq) = tx.rsmq { + rsmq.send_message(job_uuid.to_bytes_le().to_vec(), Option::Some(estimated_next_schedule_timestamp)); + } + tx.commit().await?; + } else { + // if using posgtres, then we're able to re-queue the entire batch of scheduled job for this script_path, so we do it + let _requeued_jobs = sqlx::query_as::<_, QueuedJob>(&format!( + "UPDATE queue + SET running = false + , started_at = null + , scheduled_for = '{estimated_next_schedule_timestamp}' + , logs = CASE WHEN logs IS NULL OR logs = '' THEN '{job_log_event}'::text WHEN logs LIKE '%{job_log_event}' THEN logs ELSE concat(logs, '{job_log_line_break}{job_log_event}'::text) END + WHERE (id = '{job_uuid}') OR (script_path = '{job_script_path}' AND running = false) + RETURNING *" + )) + .fetch_all(&mut tx) + .await + .map_err(|e| Error::InternalErr(format!("Could not update and re-queue job {job_uuid}. The job will be marked as running but it is not running: {e}")))?; + tx.commit().await? + } + } +} + +async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c, R: rsmq_async::RsmqConnection + Send + Clone>( + mut tx: QueueTransaction<'c, R>, + workspaces_filter: &str, + rsmq: Option, +) -> windmill_common::error::Result<(Option, QueueTransaction<'c, R>)> { let job: Option = if let Some(mut rsmq) = rsmq { // TODO: REDIS: Race conditions / replace last_ping let msg = rsmq @@ -719,7 +844,7 @@ pub async fn pull( RETURNING *", ) .bind(uuid) - .fetch_optional(db) + .fetch_optional(&mut tx) .await? } else { None @@ -755,16 +880,10 @@ pub async fn pull( ) RETURNING *" )) - .fetch_optional(db) + .fetch_optional(&mut tx) .await? }; - // println!("3.2: {:?} {rs}", instant.elapsed()); - - if job.is_some() && *METRICS_ENABLED { - QUEUE_PULL_COUNT.inc(); - } - - Ok(job) + Ok((job, tx)) } #[async_recursion] @@ -1005,9 +1124,9 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( } } - let (script_hash, script_path, raw_code_tuple, job_kind, mut raw_flow, language) = + let (script_hash, script_path, raw_code_tuple, job_kind, mut raw_flow, language, concurrent_limit, concurrency_time_window_s) = match job_payload { - JobPayload::ScriptHash { hash, path } => { + JobPayload::ScriptHash { hash, path , concurrent_limit, concurrency_time_window_s} => { let language = sqlx::query_scalar!( "SELECT language as \"language: ScriptLang\" FROM script WHERE hash = $1 AND workspace_id = $2", hash.0, @@ -1027,6 +1146,8 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( JobKind::Script, None, Some(language), + concurrent_limit, + concurrency_time_window_s, ) } JobPayload::ScriptHub { path } => { @@ -1038,15 +1159,19 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( JobKind::Script_Hub, None, None, + None, + None, ) } - JobPayload::Code(RawCode { content, path, language, lock }) => ( + JobPayload::Code(RawCode { content, path, language, lock , concurrent_limit, concurrency_time_window_s}) => ( None, path, Some((content, lock)), JobKind::Preview, None, Some(language), + concurrent_limit, + concurrency_time_window_s, ), JobPayload::Dependencies { hash, dependencies, language } => ( Some(hash.0), @@ -1055,6 +1180,8 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( JobKind::Dependencies, None, Some(language), + None, + None, ), JobPayload::FlowDependencies { path } => { let value_json = sqlx::query_scalar!( @@ -1077,10 +1204,12 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( JobKind::FlowDependencies, Some(value), None, + None, + None, ) } JobPayload::RawFlow { value, path } => { - (None, path, None, JobKind::FlowPreview, Some(value), None) + (None, path, None, JobKind::FlowPreview, Some(value), None, None, None) } JobPayload::Flow(flow) => { let value_json = sqlx::query_scalar!( @@ -1096,11 +1225,11 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( "could not convert json to flow for {flow}: {err:?}" )) })?; - (None, Some(flow), None, JobKind::Flow, Some(value), None) + (None, Some(flow), None, JobKind::Flow, Some(value), None, None, None) } - JobPayload::Identity => (None, None, None, JobKind::Identity, None, None), - JobPayload::Http => (None, None, None, JobKind::Http, None, None), - JobPayload::Noop => (None, None, None, JobKind::Noop, None, None), + JobPayload::Identity => (None, None, None, JobKind::Identity, None, None, None, None), + JobPayload::Http => (None, None, None, JobKind::Http, None, None, None, None), + JobPayload::Noop => (None, None, None, JobKind::Noop, None, None, None, None), }; let is_running = same_worker; @@ -1179,8 +1308,8 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( "INSERT INTO queue (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, \ - flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag) - VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24) \ + flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s) + VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26) \ RETURNING id", workspace_id, job_id, @@ -1205,7 +1334,9 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( email, visible_to_owner, root_job, - tag + tag, + concurrent_limit, + concurrency_time_window_s, ) .fetch_one(&mut tx) .await diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index d8b44a61da..4e5fe0ea13 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -71,14 +71,14 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>( let (payload, tag) = if schedule.is_flow { (JobPayload::Flow(schedule.script_path), None) } else { - let (hash, tag) = windmill_common::get_latest_hash_for_path( + let (hash, tag, concurrent_limit, concurrency_time_window_s) = windmill_common::get_latest_hash_for_path( tx.transaction_mut(), &schedule.workspace_id, &schedule.script_path, ) .await?; ( - JobPayload::ScriptHash { hash, path: schedule.script_path }, + JobPayload::ScriptHash { hash, path: schedule.script_path, concurrent_limit: concurrent_limit, concurrency_time_window_s: concurrency_time_window_s}, tag, ) }; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index e53acf3376..59fbce19e7 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1033,6 +1033,7 @@ async fn handle_queued_job( // println!("handle queue {:?}", SystemTime::now()); if let Some(log_str) = &job.logs { logs.push_str(&log_str); + logs.push_str("\n"); } logs.push_str(&format!("job {} on worker {}\n", &job.id, &worker_name)); @@ -2153,7 +2154,7 @@ async fn handle_flow_dependency_job( let mut flow = serde_json::from_value::(raw_flow).map_err(to_anyhow)?; let mut new_flow_modules = Vec::new(); for mut e in flow.modules.into_iter() { - let FlowModuleValue::RawScript { lock: _, path, content, language, input_transforms, tag} = e.value else { + let FlowModuleValue::RawScript { lock: _, path, content, language, input_transforms, tag, concurrent_limit, concurrency_time_window_s} = e.value else { new_flow_modules.push(e); continue; }; @@ -2178,11 +2179,13 @@ async fn handle_flow_dependency_job( Ok(new_lock) => { e.value = FlowModuleValue::RawScript { lock: Some(new_lock), - path: path, + path, input_transforms, content, language, - tag + tag, + concurrent_limit, + concurrency_time_window_s, }; new_flow_modules.push(e); continue; @@ -2198,11 +2201,13 @@ async fn handle_flow_dependency_job( ); e.value = FlowModuleValue::RawScript { lock: None, - path: path, + path, input_transforms, content, language, - tag + tag, + concurrent_limit, + concurrency_time_window_s, }; new_flow_modules.push(e); continue; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index ee2050dda0..430bfd9385 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -19,7 +19,7 @@ use tracing::instrument; use uuid::Uuid; use windmill_common::flow_status::{FlowStatusModuleWParent, Iterator, JobResult}; use windmill_common::jobs::{ - script_hash_to_tag, script_path_to_payload, JobPayload, Metrics, QueuedJob, RawCode, + script_hash_to_tag_and_limits, script_path_to_payload, JobPayload, Metrics, QueuedJob, RawCode, }; use windmill_common::{ error::{self, to_anyhow, Error}, @@ -1708,9 +1708,9 @@ async fn compute_next_flow_transform( } else { let hash = script_hash.clone().unwrap(); let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?; - let tag = script_hash_to_tag(&hash, &mut tx, &flow_job.workspace_id).await?; + let (tag, concurrent_limit, concurrency_time_window_s) = script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?; ( - JobPayload::ScriptHash { hash, path: script_path.to_owned() }, + JobPayload::ScriptHash { hash, path: script_path.to_owned(), concurrent_limit, concurrency_time_window_s }, tag, ) }; @@ -1719,7 +1719,7 @@ async fn compute_next_flow_transform( NextStatus::NextStep, )) } - FlowModuleValue::RawScript { path, content, language, lock, tag, .. } => { + FlowModuleValue::RawScript { path, content, language, lock, tag, concurrent_limit, concurrency_time_window_s, ..} => { let path = path .clone() .or_else(|| Some(format!("{}/step-{}", flow_job.script_path(), status.step))); @@ -1730,6 +1730,8 @@ async fn compute_next_flow_transform( content: content.clone(), language: language.clone(), lock: lock.clone(), + concurrent_limit: *concurrent_limit, + concurrency_time_window_s: *concurrency_time_window_s, }), tag: tag.clone(), }),