starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-29 16:15:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

improve init script error handling

Browse Source
This commit is contained in:
Ruben Fiszel
2024-09-24 19:27:00 +02:00
parent 74c2cabd9f
commit 93fa6a0e97
5 changed files with 87 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0e52a588f3edeb8fb58d6d62247b8590e51171e2811c62737bdb81fb0ac8f182.json
Generated
+15
View File
@@ -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"
}
backend/windmill-common/src/ee.rs
+4 -1
View File
@@ -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")]
backend/windmill-worker/src/common.rs
+22
View File
@@ -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<F, Fut>(
job_id: Uuid,
db: &DB,
backend/windmill-worker/src/result_processor.rs
+4 -3
View File
@@ -55,7 +55,7 @@ pub async fn process_result(
column_order: Option<Vec<String>>,
new_args: Option<HashMap<String, Box<RawValue>>>,
db: &DB,
) -> error::Result<()> {
) -> error::Result<bool> {
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<
backend/windmill-worker/src/worker.rs
+42 -28
View File
@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let job_completed_processor_is_done2 = job_completed_processor_is_done.clone();
let same_worker_queue_size2 = same_worker_queue_size.clone();
let send_result = tokio::spawn(
(async move {
let mut has_been_killed = false;
@@ -1178,6 +1179,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
if is_init_script_and_failure {
tracing::error!("init script errored, exiting");
killpill_tx2.send(()).unwrap_or_default();
break;
}
if is_dependency_job && is_dedicated_worker {
tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted.");
@@ -1695,8 +1697,10 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
#[cfg(feature = "prometheus")]
let tag = job.tag.clone();
let is_init_script: bool = job.tag.as_str() == INIT_SCRIPT_TAG;
let arc_job = Arc::new(job);
if let Err(err) = handle_queued_job(
match handle_queued_job(
arc_job.clone(),
db,
&authed_client,
@@ -1712,10 +1716,9 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
worker_flow_initial_transition_duration.clone(),
worker_code_execution_duration.clone(),
)
.await
{
let is_init_script = arc_job.tag.as_str() == INIT_SCRIPT_TAG;
handle_job_error(
.await {
Err(err) => {
handle_job_error(
db,
&authed_client.get_authed().await,
arc_job.as_ref(),
@@ -1730,12 +1733,20 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
(&job_completed_tx.0).clone(),
)
.await;
if is_init_script {
tracing::error!("failed to execute init_script, exiting");
killpill_tx.send(()).unwrap();
if is_init_script {
tracing::error!("init script job failed (in handler), exiting");
update_worker_ping_for_failed_init_script(db, &worker_name, arc_job.id).await;
break;
}
};
},
Ok(false) if is_init_script => {
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
};
}
tracing::info!("worker {} exiting", worker_name);
// #[cfg(feature = "benchmark")]
// {
// println!("Writing benchmark file");
@@ -1824,10 +1836,14 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
drop(dedicated_workers);
for handle in dedicated_handles {
if let Err(e) = handle.await {
tracing::error!("error in dedicated worker waiting for it to end: {:?}", e)
let has_dedicated_workers = !dedicated_handles.is_empty();
if has_dedicated_workers {
for handle in dedicated_handles {
if let Err(e) = handle.await {
tracing::error!("error in dedicated worker waiting for it to end: {:?}", e)
}
}
tracing::info!("all dedicated workers have exited");
}
drop(job_completed_tx);
@@ -1843,7 +1859,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
same_worker_tx: SameWorkerSender,
worker_name: &str,
rsmq: Option<R>,
) -> error::Result<()> {
) -> error::Result<bool> {
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
@@ -2041,7 +2059,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
worker_code_execution_metric: &mut f32,
_worker_flow_initial_transition_duration: Option<Histo>,
_worker_code_execution_duration: Option<Histo>,
) -> windmill_common::error::Result<()> {
) -> windmill_common::error::Result<bool> {
if job.canceled {
return Err(Error::JsonErr(canceled_job_to_result(&job)));
}
@@ -2137,7 +2155,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
)
.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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
.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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
//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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
new_args,
db,
)
.await?;
};
Ok(())
.await
}
}