From 993b8d5694d7bd65331d35e0066a9bdcf1a650ce Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 3 Oct 2025 17:30:40 +0000 Subject: [PATCH] warn for slow queries in push scheduled job --- ...740b5dea1304ce4dfdb4d3d0efa4266f38c5.json} | 12 ++++-- ...3e372949f5ec7d71b80d1ba3eaddf851adce.json} | 7 ++-- backend/windmill-api/src/apps.rs | 38 +++++++++---------- backend/windmill-common/src/utils.rs | 1 + backend/windmill-queue/src/schedule.rs | 15 ++++++++ 5 files changed, 46 insertions(+), 27 deletions(-) rename backend/.sqlx/{query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json => query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json} (93%) rename backend/.sqlx/{query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json => query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json} (86%) diff --git a/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json b/backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json similarity index 93% rename from backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json rename to backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json index 28c0756b0f..6007094518 100644 --- a/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json +++ b/backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json @@ -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>>\",\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>\",\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>>\",\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>\",\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" } diff --git a/backend/.sqlx/query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json b/backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json similarity index 86% rename from backend/.sqlx/query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json rename to backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json index 2d4df84899..a5a077eb73 100644 --- a/backend/.sqlx/query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json +++ b/backend/.sqlx/query-8d67ed8e1271a27b072a8e8ae9973e372949f5ec7d71b80d1ba3eaddf851adce.json @@ -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" } diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index d3e6d549b3..3043be10ce 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -399,8 +399,6 @@ async fn get_raw_app_data( ) -> 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; // 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 = 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!( diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 3dfacbcd85..58bad93e06 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -246,6 +246,7 @@ pub async fn now_from_db<'c, E: sqlx::PgExecutor<'c>>( ) -> Result> { Ok(sqlx::query_scalar!("SELECT now()") .fetch_one(db) + .warn_after_seconds_with_sql(1, "now_from_db".to_string()) .await? .unwrap()) } diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 0870d2f24a..df57c5e67c 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -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?; }