From fa3efd3f608a754f0c271da557a244ff6b359cfd Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 11 Nov 2023 13:39:06 +0100 Subject: [PATCH] fix: handle console logs in dedicated workers --- ...6944d4cd5057d27d09de9c5dc494e3123ab4.json} | 10 +++++-- backend/windmill-worker/src/bun_executor.rs | 5 ++-- .../windmill-worker/src/dedicated_worker.rs | 30 ++++++++++++------- .../windmill-worker/src/python_executor.rs | 5 ++-- 4 files changed, 34 insertions(+), 16 deletions(-) rename backend/.sqlx/{query-e25829ed40f35a0f59a9838d030a7f888ca88441e069e73a52d75450aecae70e.json => query-ce4733a0460cbef39fb50f4ca9746944d4cd5057d27d09de9c5dc494e3123ab4.json} (64%) diff --git a/backend/.sqlx/query-e25829ed40f35a0f59a9838d030a7f888ca88441e069e73a52d75450aecae70e.json b/backend/.sqlx/query-ce4733a0460cbef39fb50f4ca9746944d4cd5057d27d09de9c5dc494e3123ab4.json similarity index 64% rename from backend/.sqlx/query-e25829ed40f35a0f59a9838d030a7f888ca88441e069e73a52d75450aecae70e.json rename to backend/.sqlx/query-ce4733a0460cbef39fb50f4ca9746944d4cd5057d27d09de9c5dc494e3123ab4.json index d3e3837086..8112e9e17f 100644 --- a/backend/.sqlx/query-e25829ed40f35a0f59a9838d030a7f888ca88441e069e73a52d75450aecae70e.json +++ b/backend/.sqlx/query-ce4733a0460cbef39fb50f4ca9746944d4cd5057d27d09de9c5dc494e3123ab4.json @@ -1,12 +1,17 @@ { "db_name": "PostgreSQL", - "query": "UPDATE queue SET canceled = true, canceled_by = $2, scheduled_for = now(), suspend = 0 WHERE workspace_id = $1 AND schedule_path IS NULL RETURNING id", + "query": "UPDATE queue SET canceled = true, canceled_by = $2, scheduled_for = now(), suspend = 0 WHERE workspace_id = $1 AND schedule_path IS NULL RETURNING id, running", "describe": { "columns": [ { "ordinal": 0, "name": "id", "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "running", + "type_info": "Bool" } ], "parameters": { @@ -16,8 +21,9 @@ ] }, "nullable": [ + false, false ] }, - "hash": "e25829ed40f35a0f59a9838d030a7f888ca88441e069e73a52d75450aecae70e" + "hash": "ce4733a0460cbef39fb50f4ca9746944d4cd5057d27d09de9c5dc494e3123ab4" } diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 1da5365f46..015fcd0476 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -686,9 +686,9 @@ for await (const chunk of Bun.stdin.stream()) {{ try {{ let {{ {spread} }} = JSON.parse(line) let res: any = await main(...[ {spread} ]); - stdout.write(JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); + stdout.write("wm_res:" + JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value) + '\n'); }} catch (e) {{ - stdout.write(JSON.stringify({{ error: {{ message: e.message, name: e.name, stack: e.stack, line: line }}}}) + '\n'); + stdout.write("wm_res:" + JSON.stringify({{ error: {{ message: e.message, name: e.name, stack: e.stack, line: line }}}}) + '\n'); }} stdout.flush(); }} @@ -740,6 +740,7 @@ plugin(p) job_completed_tx, token, jobs_rx, + worker_name, ) .await } diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index 32e7562535..738d21a230 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -55,6 +55,7 @@ pub async fn handle_dedicated_process( job_completed_tx: JobCompletedSender, token: &str, mut jobs_rx: Receiver>, + worker_name: &str, ) -> std::result::Result<(), error::Error> { //do not cache local dependencies let mut child = { @@ -112,6 +113,8 @@ pub async fn handle_dedicated_process( // let mut j = 0; let mut alive = true; + let init_log = format!("dedicated worker: {worker_name}\n\n"); + let mut logs = init_log.clone(); loop { tokio::select! { biased; @@ -124,7 +127,9 @@ pub async fn handle_dedicated_process( }, line = err_reader.next_line() => { if let Some(line) = line.expect("line is ok") { - tracing::error!("dedicated worker process stderr: {:?}", line); + logs.push_str("[stderr] "); + logs.push_str(&line); + logs.push_str("\n"); } else { tracing::info!("dedicated worker process exited"); break; @@ -139,15 +144,20 @@ pub async fn handle_dedicated_process( continue; } tracing::debug!("processed job: {line}"); - - let job: Arc = jobs.pop_front().expect("pop"); - match serde_json::from_str::>(&line) { - Ok(result) => job_completed_tx.send(JobCompleted { job , result, logs: "".to_string(), mem_peak: 0, canceled_by: None, success: true, 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(); - }, - }; + if line.starts_with("wm_res:") { + let job: Arc = jobs.pop_front().expect("pop"); + match serde_json::from_str::>(&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(), + 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(); + }, + }; + logs = init_log.clone(); + } else { + logs.push_str(&line); + logs.push_str("\n"); + } } else { tracing::info!("dedicated worker process exited"); break; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 9b303ec07a..e2f4fd6980 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -902,12 +902,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(res_json + "\n") + sys.stdout.write("wm_res:" + res_json + "\n") except Exception 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(err_json + "\n") + sys.stdout.write("wm_res:" + err_json + "\n") sys.stdout.flush() "#, ); @@ -947,6 +947,7 @@ for line in sys.stdin: job_completed_tx, token, jobs_rx, + worker_name, ) .await }