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

improve perf of push

Browse Source
This commit is contained in:
Ruben Fiszel
2025-04-12 15:26:56 +00:00
parent 496960d349
commit 7a4f64b92d
12 changed files with 158 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-122090a0f89e5248a0a0f199ebd24582fdb302883aebd2da187ac0084e767ea3.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.pgroups', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "122090a0f89e5248a0a0f199ebd24582fdb302883aebd2da187ac0084e767ea3"
}
backend/.sqlx/query-18ddbb9eb4c0ec03e6170f8c70634546cddaf1029618ed14015fd7b0a7017441.json
Generated
+27
View File
@@ -0,0 +1,27 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_session_context($1, $2, $3, $4, $5, $6)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_session_context",
"type_info": "Void"
}
],
"parameters": {
"Left": [
"Bool",
"Text",
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "18ddbb9eb4c0ec03e6170f8c70634546cddaf1029618ed14015fd7b0a7017441"
}
backend/.sqlx/query-29fbc3a8c35845a997cd548ed417b9cc3c82d815d99b3d435adcfbb5a9246124.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.folders_read', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "29fbc3a8c35845a997cd548ed417b9cc3c82d815d99b3d435adcfbb5a9246124"
}
backend/.sqlx/query-31df83e9eb6078e93ec5fe4168306caccb849db9e0f71d86da655b01c6a3e8d0.json
Generated
-27
View File
@@ -1,27 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) RETURNING id AS \"id!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Uuid"
}
],
"parameters": {
"Left": [
"Varchar",
"Uuid",
"Bool",
"Timestamptz",
"Varchar",
"Int2"
]
},
"nullable": [
false
]
},
"hash": "31df83e9eb6078e93ec5fe4168306caccb849db9e0f71d86da655b01c6a3e8d0"
}
backend/.sqlx/query-6199e8be5cb13db71108e555ea20f0b76dc38476670f9fc0667b057d2766d42e.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.groups', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "6199e8be5cb13db71108e555ea20f0b76dc38476670f9fc0667b057d2766d42e"
}
backend/.sqlx/query-6c90d5ea2a09b47b81fdab465062c25f8768b220e53bf469550a3a3697ab756a.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.user', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "6c90d5ea2a09b47b81fdab465062c25f8768b220e53bf469550a3a3697ab756a"
}
backend/.sqlx/query-7d78efab0a588f56a13a7b5251f0a72f5d341b053218e7aec83a834cf7ccc98f.json
Generated
-21
View File
@@ -1,21 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $2, $3, $4, $5, $6, $7, $8) \n ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Varchar",
"Varchar",
"Bool",
"Bool",
"JsonbArray",
"TextArray",
"Varchar"
]
},
"nullable": []
},
"hash": "7d78efab0a588f56a13a7b5251f0a72f5d341b053218e7aec83a834cf7ccc98f"
}
backend/.sqlx/query-9897705b7c265333559df757a725711a40924ec551a7589a64ee96f8aa7f6a21.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT set_config('session.folders_write', $1, true)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "set_config",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null
]
},
"hash": "9897705b7c265333559df757a725711a40924ec551a7589a64ee96f8aa7f6a21"
}
backend/.sqlx/query-ca8bf3dda133556ee7051d5436a5d76ae2223d59e9f321a8b6d0c27adc09f741.json
Generated
-14
View File
@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "ca8bf3dda133556ee7051d5436a5d76ae2223d59e9f321a8b6d0c27adc09f741"
}
backend/.sqlx/query-29624682d687790dd199c4af759132d79fdb2982de111cb5fd43e3d9ecd0f15e.json → backend/.sqlx/query-cccdcb7fe7968eadfc04d8957a8e98b2f2d92a6d7f687a9dd5a70edb3d5a63e6.json
Generated
+13 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $21, $22, $23, $24, $25, $26,\n CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)",
"query": "WITH inserted_job AS (\n INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $21, $22, $23, $24, $25, $26,\n CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n values ($1, $32, $33, $34, $35, $36, $37, $2) \n ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 THEN now() END, $30, $31)",
"describe": {
"columns": [],
"parameters": {
@@ -86,10 +86,20 @@
"Varchar",
"Int4",
"Int2",
"Bool"
"Bool",
"Bool",
"Timestamptz",
"Varchar",
"Int2",
"Varchar",
"Varchar",
"Bool",
"Bool",
"JsonbArray",
"TextArray"
]
},
"nullable": []
},
"hash": "29624682d687790dd199c4af759132d79fdb2982de111cb5fd43e3d9ecd0f15e"
"hash": "cccdcb7fe7968eadfc04d8957a8e98b2f2d92a6d7f687a9dd5a70edb3d5a63e6"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
47869a5803b421a754173250fed79b63fe001ebd
1475a504133766e005ec551004c2af2283663352
backend/windmill-queue/src/jobs.rs
+117 -95
View File
@@ -4190,93 +4190,6 @@ pub async fn push<'c, 'd>(
_ => None,
});
sqlx::query!(
"INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,
created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,
script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,
flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,
cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
$19, $20, $21, $22, $23, $24, $25, $26,
CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,
($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)",
job_id,
workspace_id,
raw_code,
raw_lock,
raw_flow as Option<Json<FlowValue>>,
tag,
parent_job,
user,
permissioned_as,
script_hash,
script_path.clone(),
Json(args) as Json<PushArgs>,
job_kind.clone() as JobKind,
schedule_path,
language as Option<ScriptLang>,
same_worker,
pre_run_error.map(|e| e.to_string()),
email,
visible_to_owner,
root_job,
concurrent_limit,
if concurrent_limit.is_some() {
concurrency_time_window_s
} else {
None
},
custom_timeout,
flow_step_id,
cache_ttl,
final_priority,
preprocessed,
)
.execute(&mut *tx)
.warn_after_seconds(1)
.await?;
tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}");
let uuid = sqlx::query_scalar!(
"INSERT INTO v2_job_queue
(workspace_id, id, running, scheduled_for, started_at, tag, priority)
VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) \
RETURNING id AS \"id!\"",
workspace_id,
job_id,
is_running,
scheduled_for_o,
tag,
final_priority,
)
.fetch_one(&mut *tx)
.warn_after_seconds(1)
.await
.map_err(|e| Error::internal_err(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?;
sqlx::query!(
"INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)",
job_id
)
.execute(&mut *tx)
.await?;
if let Some(flow_status) = flow_status {
sqlx::query!(
"INSERT INTO v2_job_status (id, flow_status) VALUES ($1, $2)",
job_id,
Json(flow_status) as Json<FlowStatus>,
)
.execute(&mut *tx)
.await?;
}
tracing::debug!("Pushed {job_id}");
// TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
#[cfg(feature = "prometheus")]
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_PUSH_COUNT.inc();
}
let job_authed = match authed {
Some(authed)
if authed.email == email
@@ -4309,21 +4222,130 @@ pub async fn push<'c, 'd>(
.filter_map(|x| serde_json::to_value(x).ok())
.collect::<Vec<_>>();
if let Err(err) = sqlx::query!("INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id)
values ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8",
// if let Err(err) = sqlx::query!("INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id)
// values ($1, $2, $3, $4, $5, $6, $7, $8)
// ON CONFLICT (job_id) DO UPDATE SET email = $2, username = $3, is_admin = $4, is_operator = $5, folders = $6, groups = $7, workspace_id = $8",
// job_id,
// job_authed.email,
// job_authed.username,
// job_authed.is_admin,
// job_authed.is_operator,
// folders.as_slice(),
// job_authed.groups.as_slice(),
// workspace_id,
// ).execute(&mut *tx).await {
// tracing::error!("Could not insert job_perms for job {job_id}: {err:#}");
// }
sqlx::query!(
"WITH inserted_job AS (
INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,
created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,
script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,
flow_innermost_root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,
cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
$19, $20, $21, $22, $23, $24, $25, $26,
CASE WHEN $14::VARCHAR IS NOT NULL THEN 'schedule'::job_trigger_kind END,
($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)
),
inserted_runtime AS (
INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)
),
inserted_job_perms AS (
INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id)
values ($1, $32, $33, $34, $35, $36, $37, $2)
ON CONFLICT (job_id) DO UPDATE SET email = $32, username = $33, is_admin = $34, is_operator = $35, folders = $36, groups = $37, workspace_id = $2
)
INSERT INTO v2_job_queue
(workspace_id, id, running, scheduled_for, started_at, tag, priority)
VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 THEN now() END, $30, $31)",
job_id,
workspace_id,
raw_code,
raw_lock,
raw_flow as Option<Json<FlowValue>>,
tag,
parent_job,
user,
permissioned_as,
script_hash,
script_path.clone(),
Json(args) as Json<PushArgs>,
job_kind.clone() as JobKind,
schedule_path,
language as Option<ScriptLang>,
same_worker,
pre_run_error.map(|e| e.to_string()),
email,
visible_to_owner,
root_job,
concurrent_limit,
if concurrent_limit.is_some() {
concurrency_time_window_s
} else {
None
},
custom_timeout,
flow_step_id,
cache_ttl,
final_priority,
preprocessed,
is_running,
scheduled_for_o,
tag,
final_priority,
job_authed.email,
job_authed.username,
job_authed.is_admin,
job_authed.is_operator,
folders.as_slice(),
job_authed.groups.as_slice(),
workspace_id,
).execute(&mut *tx).await {
tracing::error!("Could not insert job_perms for job {job_id}: {err:#}");
)
.execute(&mut *tx)
.warn_after_seconds(1)
.await?;
// tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}");
// let uuid = sqlx::query_scalar!(
// "INSERT INTO v2_job_queue
// (workspace_id, id, running, scheduled_for, started_at, tag, priority)
// VALUES ($1, $2, $3, COALESCE($4, now()), CASE WHEN $3 THEN now() END, $5, $6) \
// RETURNING id AS \"id!\"",
// workspace_id,
// job_id,
// ,
// )
// .fetch_one(&mut *tx)
// .warn_after_seconds(1)
// .await
// .map_err(|e| Error::internal_err(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?;
// sqlx::query!(
// "INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)",
// job_id
// )
// .execute(&mut *tx)
// .await?;
if let Some(flow_status) = flow_status {
sqlx::query!(
"INSERT INTO v2_job_status (id, flow_status) VALUES ($1, $2)",
job_id,
Json(flow_status) as Json<FlowStatus>,
)
.execute(&mut *tx)
.await?;
}
tracing::debug!("Pushed {job_id}");
// TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
#[cfg(feature = "prometheus")]
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_PUSH_COUNT.inc();
}
{
let uuid_string = job_id.to_string();
@@ -4384,7 +4406,7 @@ pub async fn push<'c, 'd>(
.await?;
}
Ok((uuid, tx))
Ok((job_id, tx))
}
pub fn canceled_job_to_result(job: &MiniPulledJob) -> serde_json::Value {