starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-11 08:07:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve custom concurrency key handling

Browse Source
This commit is contained in:
Ruben Fiszel
2024-03-06 15:53:01 +01:00
parent c03c73797b
commit f4fd7afbb3
16 changed files with 207 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0b74ab3a237b2b7f54c05c7ea74294317fcd039870268eaa930f1ba8b8250559.json
Generated
+29
View File
@@ -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"
}
backend/.sqlx/query-2946aa519633291de986978d245dc615a0ee10ce6e6a98bd4906e86225b387b3.json
Generated
+14
View File
@@ -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"
}
backend/.sqlx/query-330e85c6fe52355971262d3a44c2b75a95b1d042b9e0995186aa5a2d0b7ee552.json
Generated
-30
View File
@@ -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"
}
backend/.sqlx/query-636db0b9d5963ed540f18ab732a8b29a3308f973cc04fd10979c44ae19169abf.json
Generated
+14
View File
@@ -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"
}
backend/.sqlx/query-6c0f74c56789ac51ccb06cd8a14986071ccc94df0de137b56d63d673db11d8aa.json
Generated
+14
View File
@@ -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"
}
backend/.sqlx/query-74949628ea78a5fecfd415e6406636a9442edd5b7939f61894288103be0c03ef.json
Generated
-14
View File
@@ -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"
}
backend/.sqlx/query-973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO custom_concurrency_key_ended VALUES ($1)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar"
]
},
"nullable": []
},
"hash": "973933b021d2167edff3a48ec4d4abc53ada670155921a4a4c2f05f229ae560a"
}
backend/.sqlx/query-abc7c72dfe9b01cde6f5b206300ed33e3d15b16bce2510160166a66fb7598e61.json
Generated
+24
View File
@@ -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"
}
backend/migrations/20240306121903_custom_concurrency_key.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add down migration script here
DROP TABLE custom_concurrency_key_ended;
backend/migrations/20240306121903_custom_concurrency_key.up.sql
+6
View File
@@ -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)
);
backend/src/monitor.rs
+28 -23
View File
@@ -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)
}
}
backend/windmill-api/src/concurrency_groups.rs
+8
View File
@@ -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(()))
}
backend/windmill-queue/src/jobs.rs
+37 -22
View File
@@ -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<R: rsmq_async::RsmqConnection + Send + Clone>(
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<R: rsmq_async::RsmqConnection + Send + Clone>(
})?;
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<R: rsmq_async::RsmqConnection + Send + Clone>(
})?;
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<R: rsmq_async::RsmqConnection + Send + Clone>(
"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<DateTime<Utc>> = script_path_live_stats.min_started_at;
let avg_script_duration: Option<i64> = 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<R: rsmq_async::RsmqConnection + Send + Clone>(
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 {
frontend/src/lib/components/ScriptBuilder.svelte
+6 -1
View File
@@ -686,8 +686,13 @@
type="text"
autofocus
bind:value={script.concurrency_key}
placeholder={`$workspace/script/${script.path}`}
placeholder={`$workspace/script/${script.path}-$args[foo]`}
/>
<Tooltip
>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]`</Tooltip
>
</Label>
</div>
</Section>
frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
+6
View File
@@ -420,6 +420,12 @@
bind:seconds={flowModule.value.concurrency_time_window_s}
/>
</Label>
<Label label="Custom concurrency key">
<div class="text-tertiary text-xs"
>Custom concurrency keys can only be set as the setting of a workspace
script</div
>
</Label>
{:else}
<Alert type="warning" title="Limitation" size="xs">
The concurrency limit of a workspace script is only settable in the script
frontend/src/lib/components/flows/content/FlowSettings.svelte
+5
View File
@@ -493,6 +493,11 @@
bind:seconds={$flowStore.value.concurrency_time_window_s}
/>
</Label>
<Label label="Custom concurrency key">
<div class="text-tertiary text-xs"
>Custom concurrency keys can only be set as the setting of a workspace script</div
>
</Label>
</div>
</Section>
</TabContent>