kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

warn for slow queries in push scheduled job

Browse Source
This commit is contained in:
Ruben Fiszel
2025-10-03 17:30:40 +00:00
parent f9a035831f
commit 993b8d5694
5 changed files with 46 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json → backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json generated
View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
"query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner,\n NULL as permissioned_as_end_user_email\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
"describe": {
"columns": [
{
@@ -241,6 +241,11 @@
"ordinal": 31,
"name": "visible_to_owner",
"type_info": "Bool"
},
{
"ordinal": 32,
"name": "permissioned_as_end_user_email",
"type_info": "Text"
}
],
"parameters": {
@@ -280,8 +285,9 @@
true,
true,
true,
false
false,
null
]
},
"hash": "f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22"
"hash": "3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5"
}

7
backend/.sqlx/query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json → backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json generated
View File

@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"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, 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, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,\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 OR $40 THEN now() END, $30, $31)",
"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, 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, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,\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, end_user_email) \n values ($1, $32, $33, $34, $35, $36, $37, $2, $41) \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 OR $40 THEN now() END, $30, $31)",
"describe": {
"columns": [],
"parameters": {
@@ -123,10 +123,11 @@
}
}
},
"Bool"
"Bool",
"Varchar"
]
},
"nullable": []
},
"hash": "193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b"
"hash": "8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce"
}

38
backend/windmill-api/src/apps.rs
View File

@@ -399,8 +399,6 @@ async fn get_raw_app_data(
) -> Result<Response> {
#[cfg(all(feature = "enterprise", feature = "parquet"))]
let object_store = windmill_common::s3_helpers::get_object_store().await;
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
let object_store: Option<()> = None;
// tracing::info!("secret_with_ext: {}", secret_with_ext);
let mut splitted = secret_with_ext.split('.');
@@ -429,12 +427,12 @@ async fn get_raw_app_data(
));
};
// tracing::info!("file_type: {}", file_type);
let path = format!("/app_bundles/{}/{}.{}", w_id, id, file_type);
#[allow(unused_assignments)]
let mut body: Option<Body> = None;
#[cfg(all(feature = "enterprise", feature = "parquet"))]
if let Some(os) = object_store {
let path = format!("/app_bundles/{}/{}.{}", w_id, id, file_type);
let stream = os
.get(&object_store::path::Path::from(path))
.await?
@@ -920,8 +918,6 @@ async fn get_latest_version_secret_id(
Ok(hx)
}
use windmill_common::error;
async fn store_raw_app_file<'a>(
w_id: &str,
id: &i64,
@@ -930,24 +926,24 @@ async fn store_raw_app_file<'a>(
tx: &mut sqlx::Transaction<'a, sqlx::Postgres>,
) -> Result<()> {
#[cfg(all(feature = "enterprise", feature = "parquet"))]
let object_store = windmill_common::s3_helpers::get_object_store().await;
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
let object_store: Option<()> = None;
{
let object_store = windmill_common::s3_helpers::get_object_store().await;
let path = format!("/app_bundles/{}/{}.{}", w_id, id, file_type);
#[cfg(all(feature = "enterprise", feature = "parquet"))]
if let Some(os) = object_store {
if let Err(e) = os
.put(&object_store::path::Path::from(path.clone()), data.into())
.await
{
tracing::error!("Failed to put snapshot to s3 at {path}: {:?}", e);
return Err(error::Error::ExecutionErr(format!(
"Failed to put {path} to s3"
)));
let path: String = format!("/app_bundles/{}/{}.{}", w_id, id, file_type);
if let Some(os) = object_store {
if let Err(e) = os
.put(&object_store::path::Path::from(path.clone()), data.into())
.await
{
tracing::error!("Failed to put snapshot to s3 at {path}: {:?}", e);
return Err(windmill_common::error::Error::ExecutionErr(format!(
"Failed to put {path} to s3"
)));
}
tracing::info!("Successfully put snapshot to s3 at {path}");
return Ok(());
}
tracing::info!("Successfully put snapshot to s3 at {path}");
return Ok(());
}
sqlx::query!(

1
backend/windmill-common/src/utils.rs
View File

@@ -246,6 +246,7 @@ pub async fn now_from_db<'c, E: sqlx::PgExecutor<'c>>(
) -> Result<chrono::DateTime<chrono::Utc>> {
Ok(sqlx::query_scalar!("SELECT now()")
.fetch_one(db)
.warn_after_seconds_with_sql(1, "now_from_db".to_string())
.await?
.unwrap())
}

15
backend/windmill-queue/src/schedule.rs
View File

@@ -22,6 +22,7 @@ use windmill_common::get_latest_flow_version_info_for_path_from_version;
use windmill_common::jobs::check_tag_available_for_workspace_internal;
use windmill_common::jobs::JobPayload;
use windmill_common::schedule::schedule_to_user;
use windmill_common::utils::WarnAfterExt;
use windmill_common::FlowVersionInfo;
use windmill_common::DB;
use windmill_common::{
@@ -76,6 +77,7 @@ pub async fn push_scheduled_job<'c>(
&schedule.path
)
.execute(&mut *tx)
.warn_after_seconds_with_sql(1, "update_schedule_paused_until".to_string())
.await
.context("Failed to clear paused_until for schedule")?;
}
@@ -107,6 +109,7 @@ pub async fn push_scheduled_job<'c>(
&schedule.script_path
)
.fetch_one(&mut *tx)
.warn_after_seconds_with_sql(1, "already_exists_job".to_string())
.await?
.unwrap_or(false);
@@ -141,6 +144,7 @@ pub async fn push_scheduled_job<'c>(
&schedule.script_path,
false,
)
.warn_after_seconds_with_sql(1, "get_latest_flow_version_id_for_path".to_string())
.await?;
let FlowVersionInfo {
@@ -151,6 +155,10 @@ pub async fn push_scheduled_job<'c>(
&schedule.workspace_id,
&schedule.script_path,
)
.warn_after_seconds_with_sql(
1,
"get_latest_flow_version_info_for_path_from_version".to_string(),
)
.await?;
(
JobPayload::Flow {
@@ -184,6 +192,7 @@ pub async fn push_scheduled_job<'c>(
&schedule.script_path,
false,
)
.warn_after_seconds_with_sql(1, "get_latest_hash_for_path".to_string())
.await?;
if schedule.retry.is_some() {
@@ -257,6 +266,7 @@ pub async fn push_scheduled_job<'c>(
&schedule.path
)
.execute(&mut *tx)
.warn_after_seconds_with_sql(1, "clear_schedule_error".to_string())
.await
{
tracing::error!(
@@ -272,10 +282,12 @@ pub async fn push_scheduled_job<'c>(
let is_windmill_user =
sqlx::query_scalar!("SELECT CURRENT_USER = 'windmill_user' as \"is_windmill_user!\"")
.fetch_one(&mut *tx)
.warn_after_seconds_with_sql(1, "is_windmill_user".to_string())
.await?;
if is_windmill_user {
sqlx::query!("SET LOCAL ROLE NONE")
.execute(&mut *tx)
.warn_after_seconds_with_sql(1, "set_local_role_none".to_string())
.await?;
}
(
@@ -301,6 +313,7 @@ pub async fn push_scheduled_job<'c>(
email,
None, // no token for schedules so no scopes so no scope_tags
)
.warn_after_seconds_with_sql(1, "check_tag_available_for_workspace_internal".to_string())
.await?;
}
@@ -333,11 +346,13 @@ pub async fn push_scheduled_job<'c>(
false,
None,
)
.warn_after_seconds_with_sql(1, "push in push_scheduled_job".to_string())
.await?;
if revert_to_windmill_user {
sqlx::query!("SET LOCAL ROLE windmill_user")
.execute(&mut *tx)
.warn_after_seconds_with_sql(1, "set_local_role_windmill_user".to_string())
.await?;
}