From 93fa6a0e9758bf242a6625378c0176fb8b2687f6 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 24 Sep 2024 19:24:58 +0200 Subject: [PATCH] improve init script error handling --- ...b8590e51171e2811c62737bdb81fb0ac8f182.json | 15 ++++ backend/windmill-common/src/ee.rs | 5 +- backend/windmill-worker/src/common.rs | 22 ++++++ .../windmill-worker/src/result_processor.rs | 7 +- backend/windmill-worker/src/worker.rs | 70 +++++++++++-------- 5 files changed, 87 insertions(+), 32 deletions(-) create mode 100644 backend/.sqlx/query-0e52a588f3edeb8fb58d6d62247b8590e51171e2811c62737bdb81fb0ac8f182.json diff --git a/backend/.sqlx/query-0e52a588f3edeb8fb58d6d62247b8590e51171e2811c62737bdb81fb0ac8f182.json b/backend/.sqlx/query-0e52a588f3edeb8fb58d6d62247b8590e51171e2811c62737bdb81fb0ac8f182.json new file mode 100644 index 0000000000..f23cf3f710 --- /dev/null +++ b/backend/.sqlx/query-0e52a588f3edeb8fb58d6d62247b8590e51171e2811c62737bdb81fb0ac8f182.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE worker_ping SET \n ping_at = now(), \n jobs_executed = 1, \n current_job_id = $1, \n current_job_workspace_id = 'admins' \n WHERE worker = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "0e52a588f3edeb8fb58d6d62247b8590e51171e2811c62737bdb81fb0ac8f182" +} diff --git a/backend/windmill-common/src/ee.rs b/backend/windmill-common/src/ee.rs index 482b61a0fe..2f32756e45 100644 --- a/backend/windmill-common/src/ee.rs +++ b/backend/windmill-common/src/ee.rs @@ -26,7 +26,10 @@ pub async fn get_license_plan() -> LicensePlan { #[derive(Deserialize)] #[serde(untagged)] -pub enum CriticalErrorChannel {} +pub enum CriticalErrorChannel { + Email { email: String }, + Slack { slack_channel: String }, +} pub enum CriticalAlertKind { #[cfg(feature = "enterprise")] diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 4df77dc21f..1cda8fb03b 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -573,6 +573,28 @@ pub enum UpdateJobPollingExit { AlreadyCompleted, } +pub async fn update_worker_ping_for_failed_init_script( + db: &DB, + worker_name: &str, + last_job_id: Uuid, +) { + if let Err(e) = sqlx::query!( + "UPDATE worker_ping SET + ping_at = now(), + jobs_executed = 1, + current_job_id = $1, + current_job_workspace_id = 'admins' + WHERE worker = $2", + last_job_id, + worker_name + ) + .execute(db) + .await + { + tracing::error!("Error updating worker ping for failed init script: {e:?}"); + } +} + pub async fn update_job_poller( job_id: Uuid, db: &DB, diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 9aaa4f2799..5203a65432 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -55,7 +55,7 @@ pub async fn process_result( column_order: Option>, new_args: Option>>, db: &DB, -) -> error::Result<()> { +) -> error::Result { match result { Ok(r) => { let job = if column_order.is_some() || new_args.is_some() { @@ -111,6 +111,7 @@ pub async fn process_result( token, ) .await; + Ok(true) } Err(e) => { let error_value = match e { @@ -152,9 +153,9 @@ pub async fn process_result( token, ) .await; + Ok(false) } - }; - Ok(()) + } } pub async fn handle_receive_completed_job< diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 812fd78cd1..097c177cd3 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -84,8 +84,7 @@ use rand::Rng; use crate::{ ansible_executor::handle_ansible_job, bash_executor::{handle_bash_job, handle_powershell_job}, bun_executor::handle_bun_job, common::{ - build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args, - NO_LOGS_AT_ALL, SLOW_LOGS, + build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args, update_worker_ping_for_failed_init_script, NO_LOGS_AT_ALL, SLOW_LOGS }, deno_executor::handle_deno_job, go_executor::handle_go_job, graphql_executor::do_graphql, handle_job_error, js_eval::{eval_fetch_timeout, transpile_ts}, mysql_executor::do_mysql, pg_executor::do_postgresql, php_executor::handle_php_job, python_executor::handle_python_job, result_processor::{handle_receive_completed_job, process_result}, rust_executor::handle_rust_job, worker_flow::{ handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress, Step, }, worker_lockfiles::{ @@ -1143,6 +1142,8 @@ pub async fn run_worker { + handle_job_error( db, &authed_client.get_authed().await, arc_job.as_ref(), @@ -1730,12 +1733,20 @@ pub async fn run_worker { + tracing::error!("init script job failed, exiting"); + update_worker_ping_for_failed_init_script(db, &worker_name, arc_job.id).await; + break; + + } + _ => {} + } #[cfg(feature = "prometheus")] if let Some(duration) = _timer.map(|x| x.stop_and_record()) { @@ -1810,6 +1821,7 @@ pub async fn run_worker( same_worker_tx: SameWorkerSender, worker_name: &str, rsmq: Option, -) -> error::Result<()> { +) -> error::Result { if let Some(content) = WORKER_CONFIG.read().await.init_bash.clone() { let tx = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq); let ehm = HashMap::new(); @@ -1889,8 +1905,10 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>( .await .map_err(to_anyhow)?; tracing::info!("Creating initial job {uuid} from initial script script: {content}"); + Ok(true) + } else { + Ok(false) } - Ok(()) } // async fn process_result( @@ -2041,7 +2059,7 @@ async fn handle_queued_job( worker_code_execution_metric: &mut f32, _worker_flow_initial_transition_duration: Option, _worker_code_execution_duration: Option, -) -> windmill_common::error::Result<()> { +) -> windmill_common::error::Result { if job.canceled { return Err(Error::JsonErr(canceled_job_to_result(&job))); } @@ -2137,7 +2155,7 @@ async fn handle_queued_job( ) .fetch_one(db) .await - .map_err(|e| Error::InternalErr(format!("fetching step flow status: {e:#}")))? + .map_err(|e| Error::InternalErr(format!("Fetching script path from queue for caching purposes: {e:#}")))? .ok_or_else(|| Error::InternalErr(format!("Expected script_path")))?; let step = match step.unwrap() { Step::Step(i) => i.to_string(), @@ -2186,12 +2204,10 @@ async fn handle_queued_job( .await .expect("send job completed"); - return Ok(()); + return Ok(true); } }; if job.is_flow() { - #[cfg(feature = "prometheus")] - let timer = _worker_flow_initial_transition_duration.map(|x| x.start_timer()); handle_flow( job, db, @@ -2203,8 +2219,7 @@ async fn handle_queued_job( job_completed_tx.0.clone(), ) .await?; - #[cfg(feature = "prometheus")] - timer.map(|x| x.stop_and_record()); + Ok(true) } else { let mut logs = "".to_string(); let mut mem_peak: i32 = 0; @@ -2323,14 +2338,14 @@ async fn handle_queued_job( //it's a test job, no need to update the db if job.as_ref().workspace_id == "" { - return Ok(()); + return Ok(true); } if result .as_ref() .is_err_and(|err| matches!(err, &Error::AlreadyCompleted(_))) { - return Ok(()); + return Ok(false); } process_result( @@ -2346,9 +2361,8 @@ async fn handle_queued_job( new_args, db, ) - .await?; - }; - Ok(()) + .await + } }