more logs for dedicated workers

This commit is contained in:
Ruben Fiszel
2024-04-30 17:18:54 +02:00
parent b53aaa49e8
commit 1cac7bf239
5 changed files with 10 additions and 1 deletions
@@ -983,6 +983,7 @@ plugin(p)
jobs_rx,
worker_name,
db,
script_path,
)
.await
}
@@ -58,6 +58,7 @@ pub async fn handle_dedicated_process(
mut jobs_rx: Receiver<Arc<QueuedJob>>,
worker_name: &str,
db: &DB,
script_path: &str,
) -> std::result::Result<(), error::Error> {
//do not cache local dependencies
let mut child = {
@@ -147,6 +148,7 @@ pub async fn handle_dedicated_process(
tracing::debug!("processed job: {line}");
if line.starts_with("wm_res[") {
let job: Arc<QueuedJob> = jobs.pop_front().expect("pop");
tracing::info!("job completed on dedicated worker {script_path}: {}", job.id);
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&line.replace("wm_res[success]:", "").replace("wm_res[error]:", "")) {
Ok(result) => {
append_logs(job.id, job.workspace_id.clone(), logs.clone(), db).await;
@@ -174,8 +176,9 @@ pub async fn handle_dedicated_process(
job = conditional_polling(jobs_rx.recv(), alive && jobs.len() < MAX_BUFFERED_DEDICATED_JOBS) => {
// i += 1;
if let Some(job) = job {
tracing::debug!("received job");
jobs.push_back(job.clone());
tracing::info!("received job and adding to queue on dedicated worker for {script_path}: {} (queue_size: {})", job.id, jobs.len());
// write_stdin(&mut stdin, &serde_json::to_string(&job.args.unwrap_or_else(|| serde_json::json!({"x": job.id}))).expect("serialize")).await?;
write_stdin(&mut stdin, &serde_json::to_string(&job.args).expect("serialize")).await?;
stdin.flush().await.context("stdin flush")?;
@@ -527,6 +527,7 @@ for await (const chunk of Deno.stdin.readable) {{
jobs_rx,
worker_name,
db,
script_path,
)
.await
}
@@ -1193,6 +1193,7 @@ for line in sys.stdin:
jobs_rx,
worker_name,
db,
script_path,
)
.await
}
+3
View File
@@ -2155,6 +2155,9 @@ async fn spawn_dedicated_worker(
} {
tracing::error!("error in dedicated worker: {:?}", e);
};
if let Err(e) = killpill_tx.clone().send(()) {
tracing::error!("failed to send final killpill to dedicated worker: {:?}", e);
}
});
return Some((node_id.unwrap_or(path2), dedicated_worker_tx, Some(handle)));
// (Some(dedi_path), Some(dedicated_worker_tx), Some(handle))