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

fix: Failing jobs in dedicated worker mode are now marked as failing (#2894)

Browse Source 
* fix: Failing jobs in dedicated worker mode are now marked as failing

* remove log line

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
Guillaume Bouvignies
2023-12-21 10:29:26 +01:00
committed by GitHub
parent 4c3c988f7b
commit 5f85b67dfc
4 changed files with 17 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/bun_executor.rs
+2 -2
View File
@@ -680,9 +680,9 @@ for await (const chunk of Bun.stdin.stream()) {{
try {{
let {{ {spread} }} = JSON.parse(line)
let res: any = await main(...[ {spread} ]);
stdout.write("wm_res:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n');
stdout.write("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n');
}} catch (e) {{
stdout.write("wm_res:" + JSON.stringify({{ error: {{ message: e.message, name: e.name, stack: e.stack, line: line }}}}) + '\n');
stdout.write("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: line }}) + '\n');
}}
stdout.flush();
}}
backend/windmill-worker/src/dedicated_worker.rs
+10 -5
View File
@@ -104,8 +104,7 @@ pub async fn handle_dedicated_process(
.wait()
.await
.expect("child process encountered an error");
println!("child status was: {}", status);
tracing::info!("child status was: {}", status);
});
let mut jobs = VecDeque::with_capacity(MAX_BUFFERED_DEDICATED_JOBS);
@@ -144,10 +143,16 @@ pub async fn handle_dedicated_process(
continue;
}
tracing::debug!("processed job: {line}");
if line.starts_with("wm_res:") {
if line.starts_with("wm_res[") {
let job: Arc<QueuedJob> = jobs.pop_front().expect("pop");
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&line.replace("wm_res:", "")) {
Ok(result) => job_completed_tx.send(JobCompleted { job , result, logs: logs, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap(),
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&line.replace("wm_res[success]:", "").replace("wm_res[error]:", "")) {
Ok(result) => {
if line.starts_with("wm_res[success]:") {
job_completed_tx.send(JobCompleted { job , result, logs: logs, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap()
} else {
job_completed_tx.send(JobCompleted { job , result, logs: logs, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap()
}
},
Err(e) => {
tracing::error!("Could not deserialize job result `{line}`: {e:?}");
job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), logs: "".to_string(), mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap();
backend/windmill-worker/src/deno_executor.rs
+2 -2
View File
@@ -463,9 +463,9 @@ for await (const chunk of Deno.stdin.readable) {{
try {{
let {{ {spread} }} = JSON.parse(line)
let res: any = await main(...[ {spread} ]);
console.log("wm_res:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n');
console.log("wm_res[success]:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n');
}} catch (e) {{
console.log("wm_res:" + JSON.stringify({{ error: {{ message: e.message, name: e.name, stack: e.stack, line: line }}}}) + '\n');
console.log("wm_res[error]:" + JSON.stringify({{ message: e.message, name: e.name, stack: e.stack, line: line }}) + '\n');
}}
}}
if (exit) {{
backend/windmill-worker/src/python_executor.rs
+3 -3
View File
@@ -925,12 +925,12 @@ for line in sys.stdin:
if type(v).__name__ == 'bytes':
res[k] = to_b_64(v)
res_json = re.sub(replace_nan, ' null ', json.dumps(res, separators=(',', ':'), default=str).replace('\n', ''))
sys.stdout.write("wm_res:" + res_json + "\n")
sys.stdout.write("wm_res[success]:" + res_json + "\n")
except BaseException as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
tb = traceback.format_tb(exc_traceback)
err_json = json.dumps({{ "error": {{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }} }}, separators=(',', ':'), default=str).replace('\n', '')
sys.stdout.write("wm_res:" + err_json + "\n")
err_json = json.dumps({{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }}, separators=(',', ':'), default=str).replace('\n', '')
sys.stdout.write("wm_res[error]:" + err_json + "\n")
sys.stdout.flush()
"#,
);