From 480fd781b6a7faa3b73f646eeb4089ced76a27d2 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 10 Mar 2023 01:38:12 +0100 Subject: [PATCH] worker ping at least every 5s even when running long jobs --- backend/windmill-worker/src/worker.rs | 80 ++++++++++++++++++--------- 1 file changed, 55 insertions(+), 25 deletions(-) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index a398c44d72..c367571147 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -840,7 +840,7 @@ async fn handle_queued_job( job_dir: &str, metrics: Metrics, same_worker_tx: Sender, - base_internal_url: &str + base_internal_url: &str, ) -> windmill_common::error::Result<()> { if job.canceled { return Err(Error::JsonErr(canceled_job_to_result(&job)))?; @@ -888,10 +888,10 @@ async fn handle_queued_job( logs.push_str(&format!("job {} on worker {}\n", &job.id, &worker_name)); let result = match job.job_kind { JobKind::Dependencies => { - handle_dependency_job(&job, &mut logs, job_dir, db).await + handle_dependency_job(&job, &mut logs, job_dir, db, worker_name).await } JobKind::FlowDependencies => { - handle_flow_dependency_job(&job, &mut logs, job_dir, db) + handle_flow_dependency_job(&job, &mut logs, job_dir, db, worker_name) .await .map(|()| Value::Null) } @@ -912,7 +912,8 @@ async fn handle_queued_job( job_dir, worker_dir, &mut logs, - base_internal_url + base_internal_url, + worker_name ) .await } @@ -1064,7 +1065,9 @@ async fn handle_code_execution_job( job_dir: &str, worker_dir: &str, logs: &mut String, - base_internal_url: &str + base_internal_url: &str, + worker_name: &str + ) -> error::Result { let (inner_content, requirements_o, language) = match job.job_kind { JobKind::Preview | JobKind::Script_Hub => ( @@ -1086,7 +1089,6 @@ async fn handle_code_execution_job( "handle_code_execution_job should never be reachable with a non-code execution job" ), }; - let worker_name = worker_dir.split("/").last().unwrap_or("unknown"); let lang_str = job .language .as_ref() @@ -1136,7 +1138,7 @@ mount {{ token, &inner_content, &shared_mount, - base_internal_url + base_internal_url, ) .await } @@ -1151,7 +1153,8 @@ mount {{ &inner_content, &shared_mount, requirements_o, - base_internal_url + base_internal_url, + worker_name ) .await } @@ -1166,7 +1169,8 @@ mount {{ job_dir, requirements_o, &shared_mount, - base_internal_url + base_internal_url, + worker_name ) .await } @@ -1179,7 +1183,8 @@ mount {{ &inner_content, job_dir, &shared_mount, - base_internal_url + base_internal_url, + worker_name ) .await } @@ -1208,6 +1213,7 @@ async fn handle_go_job( requirements_o: Option, shared_mount: &str, base_internal_url: &str, + worker_name: &str, ) -> Result { //go does not like executing modules at temp root let job_dir = &format!("{job_dir}/go"); @@ -1236,6 +1242,7 @@ async fn handle_go_job( db, true, skip_go_mod, + worker_name ) .await?; @@ -1353,7 +1360,7 @@ func Run(req Req) (interface{{}}, error){{ .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()?; - handle_child(&job.id, db, logs, build_go, false).await?; + handle_child(&job.id, db, logs, build_go, false, worker_name).await?; Command::new(NSJAIL_PATH.as_str()) .current_dir(job_dir) @@ -1379,7 +1386,7 @@ func Run(req Req) (interface{{}}, error){{ .stderr(Stdio::piped()) .spawn()? }; - handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL).await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL, worker_name).await?; read_result(job_dir).await } @@ -1393,6 +1400,7 @@ async fn handle_bash_job( job_dir: &str, shared_mount: &str, base_internal_url: &str, + worker_name: &str, ) -> Result { logs.push_str("\n\n--- BASH CODE EXECUTION ---\n"); set_logs(logs, &job.id, db).await; @@ -1456,7 +1464,7 @@ async fn handle_bash_job( .stderr(Stdio::piped()) .spawn()? }; - handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL).await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL, worker_name).await?; //for now bash jobs have an empty result object Ok(serde_json::json!(logs .lines() @@ -1484,7 +1492,8 @@ async fn handle_deno_job( inner_content: &String, shared_mount: &str, lockfile: Option, - base_internal_url: &str + base_internal_url: &str, + worker_name: &str ) -> error::Result { logs.push_str("\n\n--- DENO CODE EXECUTION ---\n"); set_logs(logs, &job.id, db).await; @@ -1619,7 +1628,7 @@ run().catch(async (e) => {{ } .instrument(trace_span!("create_deno_jail")) .await?; - handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL).await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL, worker_name).await?; read_result(job_dir).await } @@ -1657,7 +1666,7 @@ async fn handle_python_job( token: String, inner_content: &String, shared_mount: &str, - base_internal_url: &str + base_internal_url: &str, ) -> error::Result { create_dependencies_dir(job_dir).await; @@ -1671,7 +1680,7 @@ async fn handle_python_job( if requirements.is_empty() { "".to_string() } else { - pip_compile(&job.id, &requirements, logs, job_dir, db) + pip_compile(&job.id, &requirements, logs, job_dir, db, worker_name) .await .map_err(|e| { Error::ExecutionErr(format!("pip compile failed: {}", e.to_string())) @@ -1888,7 +1897,7 @@ mount {{ .spawn()? }; - handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL).await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL, worker_name).await?; read_result(job_dir).await } @@ -1918,6 +1927,7 @@ async fn handle_dependency_job( logs: &mut String, job_dir: &str, db: &sqlx::Pool, + worker_name: &str, ) -> error::Result { let content = capture_dependency_job( &job.id, @@ -1932,7 +1942,8 @@ async fn handle_dependency_job( .unwrap_or_else(|| "no raw code"), logs, job_dir, - db + db, + worker_name ) .await; match content { @@ -1967,6 +1978,7 @@ async fn handle_flow_dependency_job( logs: &mut String, job_dir: &str, db: &sqlx::Pool, + worker_name: &str, ) -> error::Result<()> { let path = job.script_path.clone().ok_or_else(|| { error::Error::InternalErr( @@ -1997,6 +2009,7 @@ async fn handle_flow_dependency_job( logs, job_dir, db, + worker_name ) .await; match new_lock { @@ -2111,12 +2124,13 @@ async fn capture_dependency_job( job_raw_code: &str, logs: &mut String, job_dir: &str, - db: &sqlx::Pool + db: &sqlx::Pool, + worker_name: &str ) -> error::Result { match job_language { ScriptLang::Python3 => { create_dependencies_dir(job_dir).await; - pip_compile(job_id, job_raw_code, logs, job_dir, db ).await + pip_compile(job_id, job_raw_code, logs, job_dir, db, worker_name).await } ScriptLang::Go => { install_go_dependencies( @@ -2127,6 +2141,7 @@ async fn capture_dependency_job( db, false, false, + worker_name ) .await } @@ -2144,6 +2159,7 @@ async fn pip_compile( logs: &mut String, job_dir: &str, db: &Pool, + worker_name: &str ) -> error::Result { logs.push_str(&format!("\nresolving dependencies...")); set_logs(logs, job_id, db).await; @@ -2176,7 +2192,7 @@ async fn pip_compile( .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()?; - handle_child(job_id, db, logs, child, false) + handle_child(job_id, db, logs, child, false, worker_name) .await .map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?; let path_lock = format!("{job_dir}/requirements.txt"); @@ -2199,6 +2215,7 @@ async fn install_go_dependencies( db: &sqlx::Pool, preview: bool, skip_go_mod: bool, + worker_name: &str ) -> error::Result { if !skip_go_mod { gen_go_mymod(code, job_dir).await?; @@ -2209,7 +2226,7 @@ async fn install_go_dependencies( .stderr(Stdio::piped()) .spawn()?; - handle_child(job_id, db, logs, child, false).await?; + handle_child(job_id, db, logs, child, false, worker_name).await?; } let child = Command::new(GO_PATH.as_str()) .current_dir(job_dir) @@ -2217,7 +2234,7 @@ async fn install_go_dependencies( .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()?; - handle_child(job_id, db, logs, child, false) + handle_child(job_id, db, logs, child, false, worker_name) .await .map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?; @@ -2323,6 +2340,7 @@ async fn handle_child( logs: &mut String, mut child: Child, nsjail: bool, + worker_name: &str, ) -> error::Result<()> { let update_job_interval = Duration::from_millis(500); let write_logs_delay = Duration::from_millis(500); @@ -2350,10 +2368,22 @@ async fn handle_child( let mut interval = interval(update_job_interval); interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + let mut i = 1; loop { tokio::select!( _ = rx.recv() => break, _ = interval.tick() => { + // update the last_ping column every 5 seconds + i+=1; + if i % 10 == 0 { + sqlx::query!( + "UPDATE worker_ping SET ping_at = now() WHERE worker = $1", + &worker_name + ) + .execute(&db) + .await + .expect("update worker ping"); + } let mem_peak = get_mem_peak(pid, nsjail).await; tracing::info!("{job_id} still running. mem peak: {}kB", mem_peak); let mem_peak = if mem_peak > 0 { Some(mem_peak) } else { None }; @@ -2813,7 +2843,7 @@ async fn handle_python_reqs( .spawn()? }; - let child = handle_child(&job.id, db, logs, child, false).await; + let child = handle_child(&job.id, db, logs, child, false, worker_name).await; tracing::info!( worker_name = %worker_name, job_id = %job.id,