From f4fd7afbb3e09da1de19c56c5a5f3aa542d45117 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 6 Mar 2024 15:52:52 +0100 Subject: [PATCH] fix: improve custom concurrency key handling --- ...294317fcd039870268eaa930f1ba8b8250559.json | 29 +++++++++ ...dc615a0ee10ce6e6a98bd4906e86225b387b3.json | 14 +++++ ...2b75a95b1d042b9e0995186aa5a2d0b7ee552.json | 30 ---------- ...8b29a3308f973cc04fd10979c44ae19169abf.json | 14 +++++ ...986071ccc94df0de137b56d63d673db11d8aa.json | 14 +++++ ...636a9442edd5b7939f61894288103be0c03ef.json | 14 ----- ...4abc53ada670155921a4a4c2f05f229ae560a.json | 14 +++++ ...ed33e3d15b16bce2510160166a66fb7598e61.json | 24 ++++++++ ...0306121903_custom_concurrency_key.down.sql | 2 + ...240306121903_custom_concurrency_key.up.sql | 6 ++ backend/src/monitor.rs | 51 ++++++++-------- .../windmill-api/src/concurrency_groups.rs | 8 +++ backend/windmill-queue/src/jobs.rs | 59 ++++++++++++------- .../src/lib/components/ScriptBuilder.svelte | 7 ++- .../flows/content/FlowModuleComponent.svelte | 6 ++ .../flows/content/FlowSettings.svelte | 5 ++ 16 files changed, 207 insertions(+), 90 deletions(-) create mode 100644 backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json create mode 100644 backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json delete mode 100644 backend/.sqlx/query-330e85c6fe52355971262d3a44c2b75a95b1d042b9e0995186aa5a2d0b7ee552.json create mode 100644 backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json create mode 100644 backend/.sqlx/query-6c0f74c56789ac51ccb06cd8a14986071ccc94df0de137b56d63d673db11d8aa.json delete mode 100644 backend/.sqlx/query-74949628ea78a5fecfd415e6406636a9442edd5b7939f61894288103be0c03ef.json create mode 100644 backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json create mode 100644 backend/.sqlx/query-abc7c72dfe9b01cde6f5b206300ed33e3d15b16bce2510160166a66fb7598e61.json create mode 100644 backend/migrations/20240306121903_custom_concurrency_key.down.sql create mode 100644 backend/migrations/20240306121903_custom_concurrency_key.up.sql diff --git a/backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json b/backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json new file mode 100644 index 0000000000..f02f523b96 --- /dev/null +++ b/backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COUNT(*) as count, COALESCE(MAX(ended_at), now() - INTERVAL '1 second' * $2) as max_ended_at FROM custom_concurrency_key_ended WHERE key = $1 AND ended_at >= (now() - INTERVAL '1 second' * $2)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + }, + { + "ordinal": 1, + "name": "max_ended_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "Float8" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559" +} diff --git a/backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json b/backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json new file mode 100644 index 0000000000..18a59b9349 --- /dev/null +++ b/backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM custom_concurrency_key_ended WHERE key = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3" +} diff --git a/backend/.sqlx/query-330e85c6fe52355971262d3a44c2b75a95b1d042b9e0995186aa5a2d0b7ee552.json b/backend/.sqlx/query-330e85c6fe52355971262d3a44c2b75a95b1d042b9e0995186aa5a2d0b7ee552.json deleted file mode 100644 index 95c46a8841..0000000000 --- a/backend/.sqlx/query-330e85c6fe52355971262d3a44c2b75a95b1d042b9e0995186aa5a2d0b7ee552.json +++ /dev/null @@ -1,30 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT COALESCE(j.min_started_at, q.min_started_at) AS min_started_at, COALESCE(completed_count, 0) AS completed_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 job_kind != 'dependencies' AND started_at + INTERVAL '1 MILLISECOND' * duration_ms > (now() - INTERVAL '1 second' * $2) AND workspace_id = $3 AND canceled = false\n GROUP BY script_path) as j\n FULL OUTER JOIN\n (SELECT script_path, MIN(started_at) as min_started_at\n FROM queue\n WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $3 AND canceled = false\n GROUP BY script_path) as q\n ON q.script_path = j.script_path", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "min_started_at", - "type_info": "Timestamptz" - }, - { - "ordinal": 1, - "name": "completed_count", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Float8", - "Text" - ] - }, - "nullable": [ - true, - true - ] - }, - "hash": "330e85c6fe52355971262d3a44c2b75a95b1d042b9e0995186aa5a2d0b7ee552" -} diff --git a/backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json b/backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json new file mode 100644 index 0000000000..f069e7b851 --- /dev/null +++ b/backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM custom_concurrency_key_ended WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [] + }, + "hash": "636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf" +} diff --git a/backend/.sqlx/query-6c0f74c56789ac51ccb06cd8a14986071ccc94df0de137b56d63d673db11d8aa.json b/backend/.sqlx/query-6c0f74c56789ac51ccb06cd8a14986071ccc94df0de137b56d63d673db11d8aa.json new file mode 100644 index 0000000000..d50a50ebd3 --- /dev/null +++ b/backend/.sqlx/query-6c0f74c56789ac51ccb06cd8a14986071ccc94df0de137b56d63d673db11d8aa.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM job_stats WHERE job_id = ANY($1)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [] + }, + "hash": "6c0f74c56789ac51ccb06cd8a14986071ccc94df0de137b56d63d673db11d8aa" +} diff --git a/backend/.sqlx/query-74949628ea78a5fecfd415e6406636a9442edd5b7939f61894288103be0c03ef.json b/backend/.sqlx/query-74949628ea78a5fecfd415e6406636a9442edd5b7939f61894288103be0c03ef.json deleted file mode 100644 index e9f542da19..0000000000 --- a/backend/.sqlx/query-74949628ea78a5fecfd415e6406636a9442edd5b7939f61894288103be0c03ef.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM job_stats WHERE job_id IN (SELECT id FROM completed_job WHERE created_at <= now() - ($1::bigint::text || ' s')::interval AND started_at + ((duration_ms/1000 + $1::bigint) || ' s')::interval <= now())", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [] - }, - "hash": "74949628ea78a5fecfd415e6406636a9442edd5b7939f61894288103be0c03ef" -} diff --git a/backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json b/backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json new file mode 100644 index 0000000000..2f67c8e0ed --- /dev/null +++ b/backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO custom_concurrency_key_ended VALUES ($1)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a" +} diff --git a/backend/.sqlx/query-abc7c72dfe9b01cde6f5b206300ed33e3d15b16bce2510160166a66fb7598e61.json b/backend/.sqlx/query-abc7c72dfe9b01cde6f5b206300ed33e3d15b16bce2510160166a66fb7598e61.json new file mode 100644 index 0000000000..f422bc1758 --- /dev/null +++ b/backend/.sqlx/query-abc7c72dfe9b01cde6f5b206300ed33e3d15b16bce2510160166a66fb7598e61.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COALESCE((SELECT MIN(started_at) as min_started_at\n FROM queue\n WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $2 AND canceled = false\n GROUP BY script_path), $3)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "coalesce", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Timestamptz" + ] + }, + "nullable": [ + null + ] + }, + "hash": "abc7c72dfe9b01cde6f5b206300ed33e3d15b16bce2510160166a66fb7598e61" +} diff --git a/backend/migrations/20240306121903_custom_concurrency_key.down.sql b/backend/migrations/20240306121903_custom_concurrency_key.down.sql new file mode 100644 index 0000000000..6201a1de29 --- /dev/null +++ b/backend/migrations/20240306121903_custom_concurrency_key.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +DROP TABLE custom_concurrency_key_ended; diff --git a/backend/migrations/20240306121903_custom_concurrency_key.up.sql b/backend/migrations/20240306121903_custom_concurrency_key.up.sql new file mode 100644 index 0000000000..eb99135d3d --- /dev/null +++ b/backend/migrations/20240306121903_custom_concurrency_key.up.sql @@ -0,0 +1,6 @@ +-- Add up migration script here +CREATE TABLE custom_concurrency_key_ended ( + key VARCHAR(255) NOT NULL, + ended_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(), + PRIMARY KEY (key, ended_at) +); diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index f193b19089..08ebc9edd3 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -276,39 +276,44 @@ pub async fn delete_expired_items(db: &DB) -> () { if job_retention_secs > 0 { match db.begin().await { Ok(mut tx) => { - let r = sqlx::query!( - "DELETE FROM job_stats WHERE job_id IN (SELECT id FROM completed_job WHERE created_at <= now() - ($1::bigint::text || ' s')::interval AND started_at + ((duration_ms/1000 + $1::bigint) || ' s')::interval <= now())", - job_retention_secs - ) - .fetch_all(&mut *tx) - .await; - match r { - Ok(_) => { - let deleted_jobs = sqlx::query_scalar!( + let deleted_jobs = sqlx::query_scalar!( "DELETE FROM completed_job WHERE created_at <= now() - ($1::bigint::text || ' s')::interval AND started_at + ((duration_ms/1000 + $1::bigint) || ' s')::interval <= now() RETURNING id", job_retention_secs ) .fetch_all(&mut *tx) .await; - match deleted_jobs { - Ok(deleted_jobs) => { - if deleted_jobs.len() > 0 { - tracing::info!( - "deleted {} jobs completed JOB_RETENTION_SECS {} ago: {:?}", - deleted_jobs.len(), - job_retention_secs, - deleted_jobs, - ) - } + match deleted_jobs { + Ok(deleted_jobs) => { + if deleted_jobs.len() > 0 { + tracing::info!( + "deleted {} jobs completed JOB_RETENTION_SECS {} ago: {:?}", + deleted_jobs.len(), + job_retention_secs, + deleted_jobs, + ); + if let Err(e) = sqlx::query!( + "DELETE FROM job_stats WHERE job_id = ANY($1)", + &deleted_jobs + ) + .execute(&mut *tx) + .await + { + tracing::error!("Error deleting job stats: {:?}", e); } - Err(e) => { - tracing::error!("Error deleting expired jobs: {:?}", e) + if let Err(e) = sqlx::query!( + "DELETE FROM custom_concurrency_key_ended WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ", + job_retention_secs + ) + .execute(&mut *tx) + .await + { + tracing::error!("Error deleting custom concurrency key: {:?}", e); } } } - Err(err) => { - tracing::error!("Error deleting expired job stats: {:?}", err) + Err(e) => { + tracing::error!("Error deleting expired jobs: {:?}", e) } } diff --git a/backend/windmill-api/src/concurrency_groups.rs b/backend/windmill-api/src/concurrency_groups.rs index 83918cb04e..e17547b861 100644 --- a/backend/windmill-api/src/concurrency_groups.rs +++ b/backend/windmill-api/src/concurrency_groups.rs @@ -112,6 +112,14 @@ async fn delete_concurrency_group( ) .execute(&mut *tx) .await?; + + sqlx::query!( + "DELETE FROM custom_concurrency_key_ended WHERE key = $1", + concurrency_id.clone(), + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; Ok(Json(())) } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index de238e1d71..30b8151994 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -666,6 +666,20 @@ pub async fn add_completed_job< { tracing::error!("Could not decrement concurrency counter: {}", e); } + + if let Err(e) = sqlx::query_scalar!( + "INSERT INTO custom_concurrency_key_ended VALUES ($1)", + concurrency_key, + ) + .execute(&mut tx) + .await + .map_err(|e| { + Error::InternalErr(format!( + "Error inserting into custom_concurrency_key_ended for key {concurrency_key}: {e}" + )) + }) { + tracing::error!("Could not insert into custom_concurrency_key_ended: {}", e); + } tracing::debug!("decremented concurrency counter"); } @@ -1517,7 +1531,7 @@ pub async fn pull( let job_script_path = pulled_job.script_path.clone().unwrap(); let job_concurrency_key = concurrency_key(db, &pulled_job).await; - tracing::warn!("Concurrency key is '{}'", job_concurrency_key); + tracing::debug!("Concurrency key is '{}'", job_concurrency_key); 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 = @@ -1558,22 +1572,24 @@ pub async fn pull( })?; tracing::debug!("running_job: {}", running_job.unwrap_or(0)); - 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) AS completed_count - FROM - (SELECT script_path, MIN(started_at) as min_started_at, COUNT(*) as completed_count - FROM completed_job - WHERE script_path = $1 AND job_kind != 'dependencies' AND started_at + INTERVAL '1 MILLISECOND' * duration_ms > (now() - INTERVAL '1 second' * $2) AND workspace_id = $3 AND canceled = false - GROUP BY script_path) as j - FULL OUTER JOIN - (SELECT script_path, MIN(started_at) as min_started_at - FROM queue - WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $3 AND canceled = false - GROUP BY script_path) as q - ON q.script_path = j.script_path", - job_script_path, + let completed_count = sqlx::query!( + "SELECT COUNT(*) as count, COALESCE(MAX(ended_at), now() - INTERVAL '1 second' * $2) as max_ended_at FROM custom_concurrency_key_ended WHERE key = $1 AND ended_at >= (now() - INTERVAL '1 second' * $2)", + job_concurrency_key, f64::from(job_custom_concurrency_time_window_s), - &pulled_job.workspace_id + ).fetch_one(&mut tx).await.map_err(|e| { + Error::InternalErr(format!( + "Error getting completed count for key {job_concurrency_key}: {e}" + )) + })?; + + let min_started_at = sqlx::query_scalar!( + "SELECT COALESCE((SELECT MIN(started_at) as min_started_at + FROM queue + WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $2 AND canceled = false + GROUP BY script_path), $3)", + job_script_path, + &pulled_job.workspace_id, + completed_count.max_ended_at ) .fetch_one(&mut tx) .await @@ -1584,8 +1600,7 @@ pub async fn pull( })?; let concurrent_jobs_for_this_script = - script_path_live_stats.completed_count.unwrap_or_default() as i32 - + running_job.unwrap_or(0) as i32; + completed_count.count.unwrap_or_default() as i32 + running_job.unwrap_or(0) as i32; tracing::debug!( "Current concurrent jobs for this script: {}", concurrent_jobs_for_this_script @@ -1611,10 +1626,10 @@ pub async fn pull( "Error decreasing concurrency count for script path {job_script_path}: {e}" )) })?; + tracing::debug!("running_job after decrease: {}", x.unwrap_or(0)); 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 BIGINT) AS avg_duration_s FROM (SELECT duration_ms FROM completed_job WHERE script_path = $1 @@ -1629,12 +1644,12 @@ pub async fn pull( 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); + tracing::info!("Job '{}' from path '{}' with concurrency key '{}' 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_concurrency_key, 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" + "Re-scheduled job to {estimated_next_schedule_timestamp} due to concurrency limits with key {job_concurrency_key} and limit {job_custom_concurrent_limit} in the last {job_custom_concurrency_time_window_s} seconds", ); if rsmq.is_some() { // if let Some(ref mut rsmq) = tx.rsmq { diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index 829c35622c..08f7b6f8af 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -686,8 +686,13 @@ type="text" autofocus bind:value={script.concurrency_key} - placeholder={`$workspace/script/${script.path}`} + placeholder={`$workspace/script/${script.path}-$args[foo]`} /> + Concurrency keys are global, you can have them be workspace specific using + the variable `$workspace`. You can also use an argument's value using + `$args[name_of_arg]` diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index bb2240a77f..d526b4a4a4 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -420,6 +420,12 @@ bind:seconds={flowModule.value.concurrency_time_window_s} /> + {:else} The concurrency limit of a workspace script is only settable in the script diff --git a/frontend/src/lib/components/flows/content/FlowSettings.svelte b/frontend/src/lib/components/flows/content/FlowSettings.svelte index 2246f3f080..d6e98a78af 100644 --- a/frontend/src/lib/components/flows/content/FlowSettings.svelte +++ b/frontend/src/lib/components/flows/content/FlowSettings.svelte @@ -493,6 +493,11 @@ bind:seconds={$flowStore.value.concurrency_time_window_s} /> +