kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: handle console logs in dedicated workers

Browse Source
This commit is contained in:
Ruben Fiszel
2023-11-11 13:39:06 +01:00
parent fdfb12fd5b
commit fa3efd3f60
4 changed files with 34 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
backend/.sqlx/query-e25829ed40f35a0f59a9838d030a7f888ca88441e069e73a52d75450aecae70e.json → backend/.sqlx/query-ce4733a0460cbef39fb50f4ca9746944d4cd5057d27d09de9c5dc494e3123ab4.json generated
View File

@@ -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"
}

5
backend/windmill-worker/src/bun_executor.rs
View File

@@ -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
}

30
backend/windmill-worker/src/dedicated_worker.rs
View File

@@ -55,6 +55,7 @@ pub async fn handle_dedicated_process(
job_completed_tx: JobCompletedSender,
token: &str,
mut jobs_rx: Receiver<Arc<QueuedJob>>,
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<QueuedJob> = jobs.pop_front().expect("pop");
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&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<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(),
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;

5
backend/windmill-worker/src/python_executor.rs
View File

@@ -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
}