starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Per script concurrency limit with time window (#1816)

Browse Source 
* Add concurrency a per-script concurrency limit

* Add a per-script per time window concurrency limit

* Improve feature with a better estimation of the next schedule timestamp

* Add concurrency limits for inline scripts

* Fix tests

* run sqlx prepare

* Add requeue event to job logs
This commit is contained in:
Guillaume Bouvignies
2023-07-11 23:35:26 +02:00
committed by GitHub
parent 8f6b00803a
commit 0327b85f89
14 changed files with 515 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20230709151109_add_script_concurrency_limit.down.sql
+3
View File
@@ -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;
backend/migrations/20230709151109_add_script_concurrency_limit.up.sql
+5
View File
@@ -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;
backend/sqlx-data.json
+260 -167
View File
@@ -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": [
backend/tests/worker.rs
+22
View File
@@ -1009,6 +1009,8 @@ async fn test_deno_flow(db: Pool<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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)
backend/windmill-api/src/flows.rs
+2
View File
@@ -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(),
backend/windmill-api/src/jobs.rs
+18 -8
View File
@@ -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<String>)> {
) -> error::Result<(String, Option<String>, Option<i32>, Option<i32>)> {
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<i32>,
mem_peak: Option<i32>,
tag: String,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
}
impl From<UnifiedJob> for Job {
@@ -1226,6 +1232,8 @@ impl From<UnifiedJob> 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,
backend/windmill-api/src/oauth2.rs
+2 -2
View File
@@ -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,
)
};
backend/windmill-common/src/flows.rs
+4
View File
@@ -274,6 +274,10 @@ pub enum FlowModuleValue {
#[serde(skip_serializing_if = "Option::is_none")]
tag: Option<String>,
language: ScriptLang,
#[serde(skip_serializing_if = "Option::is_none")]
concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrency_time_window_s: Option<i32>,
},
Identity,
Http,
backend/windmill-common/src/jobs.rs
+23 -10
View File
@@ -82,6 +82,10 @@ pub struct QueuedJob {
#[serde(skip_serializing_if = "Option::is_none")]
pub leaf_jobs: Option<serde_json::Value>,
pub tag: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
}
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<i32>, concurrency_time_window_s: Option<i32> },
Code(RawCode),
Dependencies { hash: ScriptHash, dependencies: String, language: ScriptLang },
FlowDependencies { path: String },
@@ -168,6 +174,8 @@ pub struct RawCode {
pub path: Option<String>,
pub language: ScriptLang,
pub lock: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
}
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<Option<Tag>> {
Ok(sqlx::query_scalar!(
"select tag from script where hash = $1 AND workspace_id = $2",
) -> error::Result<(Option<Tag>, Option<i32>, Option<i32>)> {
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>(
backend/windmill-common/src/lib.rs
+6 -6
View File
@@ -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<Tag>)> {
) -> error::Result<(scripts::ScriptHash, Option<Tag>, Option<i32>, Option<i32>)> {
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<Tag>)> {
) -> error::Result<(scripts::ScriptHash, Option<Tag>, Option<i32>, Option<i32>)> {
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))
}
backend/windmill-queue/src/jobs.rs
+152 -21
View File
@@ -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<R: rsmq_async::RsmqConnection + Clone>(
pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
db: &Pool<Postgres>,
whitelist_workspaces: Option<Vec<String>>,
blacklist_workspaces: Option<Vec<String>>,
@@ -694,6 +695,130 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Clone>(
// 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<i64> = 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<DateTime<Utc>> = script_path_live_stats.min_started_at;
let avg_script_duration: Option<i32> = 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<R>,
) -> windmill_common::error::Result<(Option<QueuedJob>, QueueTransaction<'c, R>)> {
let job: Option<QueuedJob> = if let Some(mut rsmq) = rsmq {
// TODO: REDIS: Race conditions / replace last_ping
let msg = rsmq
@@ -719,7 +844,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Clone>(
RETURNING *",
)
.bind(uuid)
.fetch_optional(db)
.fetch_optional(&mut tx)
.await?
} else {
None
@@ -755,16 +880,10 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Clone>(
)
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
backend/windmill-queue/src/schedule.rs
+2 -2
View File
@@ -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,
)
};
backend/windmill-worker/src/worker.rs
+10 -5
View File
@@ -1033,6 +1033,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
// 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::<FlowValue>(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;
backend/windmill-worker/src/worker_flow.rs
+6 -4
View File
@@ -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(),
}),