From d5186da27129f77f302d3fed54ecdc333cdceeab Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 11 Apr 2025 23:19:29 +0200 Subject: [PATCH] allow multiple workers on agent mode (#5607) --- ...05d394a7cbcf0038c72a78add5c7b02ef5927.json | 2 +- ...1fe2dce615ff76a53f72cb5386dc17e4e07aa.json | 21 +++++ backend/Cargo.lock | 11 +++ backend/Cargo.toml | 2 + backend/ee-repo-ref.txt | 2 +- backend/src/main.rs | 84 +++++++++-------- backend/windmill-api/src/jobs.rs | 15 +++ backend/windmill-api/src/lib.rs | 6 +- backend/windmill-worker/Cargo.toml | 1 + .../windmill-worker/src/result_processor.rs | 31 ++++-- backend/windmill-worker/src/worker.rs | 94 +++++++++++-------- 11 files changed, 176 insertions(+), 93 deletions(-) create mode 100644 backend/.sqlx/query-fe7221651a982861dede4116bc71fe2dce615ff76a53f72cb5386dc17e4e07aa.json diff --git a/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json b/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json index 5bfff47576..c2dfed73a2 100644 --- a/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json +++ b/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - true + null ] }, "hash": "ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927" diff --git a/backend/.sqlx/query-fe7221651a982861dede4116bc71fe2dce615ff76a53f72cb5386dc17e4e07aa.json b/backend/.sqlx/query-fe7221651a982861dede4116bc71fe2dce615ff76a53f72cb5386dc17e4e07aa.json new file mode 100644 index 0000000000..1ca60067d0 --- /dev/null +++ b/backend/.sqlx/query-fe7221651a982861dede4116bc71fe2dce615ff76a53f72cb5386dc17e4e07aa.json @@ -0,0 +1,21 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) \n SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "UuidArray", + "Varchar", + "Varchar", + "Bool", + "Bool", + "JsonbArray", + "TextArray", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "fe7221651a982861dede4116bc71fe2dce615ff76a53f72cb5386dc17e4e07aa" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 46ef0b5b2d..4737d318f6 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -4804,6 +4804,7 @@ checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" dependencies = [ "futures-core", "futures-sink", + "nanorand", "spin 0.9.8", ] @@ -7519,6 +7520,15 @@ dependencies = [ "unicode-xid", ] +[[package]] +name = "nanorand" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a51313c5820b0b02bd422f4b44776fbf47961755c74ce64afc73bfad10226c3" +dependencies = [ + "getrandom 0.2.15", +] + [[package]] name = "napi_sym" version = "0.120.0" @@ -14480,6 +14490,7 @@ dependencies = [ "deno_webidl", "dotenv", "dyn-iter", + "flume", "futures", "gcp_auth", "git-version", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 4ca7a5fa49..ef4d7c5712 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -365,6 +365,8 @@ tantivy = "0.22.0" backon = "1.3.0" +flume = { version = "0.11.1", features = ["async"] } + # Macro-related proc-macro2 = "1.0" pulldown-cmark = "0.9" diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index b9a3d2881d..9c55ceeb81 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -85c37983ffb8f622458425c182613206625c6cee +44c3e23922097d4386183e56b6b3dc540c70d71b \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index a076b4c63f..7f4bcda628 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -329,26 +329,16 @@ async fn windmill_main() -> anyhow::Result<()> { IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)) }; - let mut first_worker_suffix = None; - let mut worker_names = vec![]; - - for _ in 0..num_workers { + let (conn, first_suffix) = if mode == Mode::Agent { + tracing::info!( + "Creating http client for cluster using base internal url {}", + std::env::var("BASE_INTERNAL_URL").unwrap_or_default() + ); let suffix = windmill_common::utils::worker_suffix(&hostname, &rd_string(5)); - worker_names.push(windmill_common::utils::worker_name_with_suffix( - mode == Mode::Agent, - WORKER_GROUP.as_str(), - &suffix, - )); - if first_worker_suffix.is_none() { - first_worker_suffix = Some(suffix); - } - } - - let conn = if mode == Mode::Agent { - let worker_suffix = first_worker_suffix.unwrap_or_else(|| { - panic!("there must be at least one worker in agent mode"); - }); - Connection::Http(build_agent_http_client(&worker_suffix)) + ( + Connection::Http(build_agent_http_client(&suffix)), + Some(suffix), + ) } else { println!("Connecting to database..."); @@ -366,7 +356,7 @@ async fn windmill_main() -> anyhow::Result<()> { load_otel(&db).await; tracing::info!("Database connected"); - Connection::Sql(db) + (Connection::Sql(db), None) }; let environment = load_base_url(&conn) @@ -409,6 +399,7 @@ async fn windmill_main() -> anyhow::Result<()> { let conn = if mode == Mode::Agent { conn } else { + // This time we use a pool of connections let db = windmill_common::connect_db(server_mode, indexer_mode, worker_mode).await?; Connection::Sql(db) }; @@ -439,16 +430,6 @@ Windmill Community Edition {GIT_VERSION} display_config(&ENV_SETTINGS); - if let Err(e) = reload_base_url_setting(&conn).await { - tracing::error!("Error loading base url: {:?}", e) - } - - if let Some(db) = conn.as_sql() { - if let Err(e) = reload_critical_error_channels_setting(&db).await { - tracing::error!("Could loading critical error emails setting: {:?}", e); - } - } - #[cfg(feature = "enterprise")] { // load the license key and check if it's valid @@ -686,14 +667,34 @@ Windmill Community Edition {GIT_VERSION} if !killpill_rx.try_recv().is_ok() { let base_internal_url = base_internal_rx.await?; if worker_mode { + let mut workers = vec![]; + for i in 0..num_workers { + let suffix: String = if i == 0 && first_suffix.as_ref().is_some() { + first_suffix.as_ref().unwrap().clone() + } else { + windmill_common::utils::worker_suffix(&hostname, &rd_string(5)) + }; + let worker_conn = WorkerConn { + conn: if i == 0 || mode != Mode::Agent { + conn.clone() + } else { + Connection::Http(build_agent_http_client(&suffix)) + }, + worker_name: windmill_common::utils::worker_name_with_suffix( + mode == Mode::Agent, + WORKER_GROUP.as_str(), + &suffix, + ), + }; + workers.push(worker_conn); + } + run_workers( - conn.clone(), rx, killpill_tx.clone(), - num_workers, base_internal_url.clone(), hostname.clone(), - &worker_names, + &workers, ) .await?; tracing::info!("All workers exited."); @@ -1134,16 +1135,20 @@ fn display_config(envs: &[&str]) { ) } +pub struct WorkerConn { + conn: Connection, + worker_name: String, +} + pub async fn run_workers( - db: Connection, mut rx: tokio::sync::broadcast::Receiver<()>, tx: KillpillSender, - num_workers: i32, base_internal_url: String, hostname: String, - worker_names: &[String], + workers: &[WorkerConn], ) -> anyhow::Result<()> { let mut killpill_rxs = vec![]; + let num_workers = workers.len(); for _ in 0..num_workers { killpill_rxs.push(rx.resubscribe()); } @@ -1202,8 +1207,9 @@ pub async fn run_workers( *windmill_worker::SLEEP_QUEUE ); for i in 1..(num_workers + 1) { - let db1 = db.clone(); - let worker_name = worker_names[i as usize - 1].clone(); + let wk_conf = &workers[i as usize - 1]; + let conn1 = wk_conf.conn.clone(); + let worker_name = wk_conf.worker_name.clone(); let ip = ip.clone(); let rx = killpill_rxs.pop().unwrap(); let tx = tx.clone(); @@ -1216,7 +1222,7 @@ pub async fn run_workers( } let f = windmill_worker::run_worker( - &db1, + &conn1, &hostname, worker_name, i as u64, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 71ec9c6c8f..4111f28b6b 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -5095,6 +5095,21 @@ async fn add_batch_jobs( .execute(&mut *tx) .await?; + sqlx::query!( + "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id) + SELECT unnest($1::uuid[]), $2, $3, $4, $5, $6, $7, $8", + &uuids, + authed.email, + authed.username, + authed.is_admin, + authed.is_operator, + &[], + &[], + w_id, + ) + .execute(&mut *tx) + .await?; + if let Some(flow_status) = flow_status { sqlx::query!( "INSERT INTO v2_job_status (id, flow_status) diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 922cb57ceb..ece80c64b1 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -728,10 +728,10 @@ pub async fn run_server( server.await?; #[cfg(feature = "agent_worker_server")] - if let Some(bg_processor) = agent_workers_bg_processor { - tracing::info!("server off. shutting down agent workers bg processor"); + for (i, bg_processor) in agent_workers_bg_processor.into_iter().enumerate() { + tracing::info!("server off. shutting down agent worker bg processor {i}"); bg_processor.await?; - tracing::info!("agent workers bg processor shut down"); + tracing::info!("agent worker bg processor {i} shut down"); } Ok(()) } diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index cad3ccc84a..e4de400d84 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -52,6 +52,7 @@ windmill-parser-sql.workspace = true windmill-parser-graphql.workspace = true windmill-parser-php = { workspace = true, optional = true } windmill-git-sync.workspace = true +flume.workspace = true sqlx.workspace = true uuid.workspace = true tracing.workspace = true diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 141c886fe0..389bd067a8 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -34,7 +34,7 @@ use windmill_queue::{ use serde_json::{json, value::RawValue}; -use tokio::{sync::mpsc::Receiver, task::JoinHandle}; +use tokio::{sync::broadcast, task::JoinHandle}; use windmill_queue::{add_completed_job, add_completed_job_error}; @@ -118,7 +118,7 @@ async fn process_jc( } pub fn start_background_processor( - mut job_completed_rx: Receiver, + job_completed_rx: flume::Receiver, job_completed_sender: JobCompletedSender, same_worker_queue_size: Arc, job_completed_processor_is_done: Arc, @@ -127,6 +127,7 @@ pub fn start_background_processor( worker_dir: String, same_worker_tx: SameWorkerSender, worker_name: String, + mut killpill_rx: broadcast::Receiver<()>, killpill_tx: KillpillSender, is_dedicated_worker: bool, ) -> JoinHandle<()> { @@ -136,19 +137,33 @@ pub fn start_background_processor( #[cfg(feature = "benchmark")] let mut infos = BenchmarkInfo::new(); + enum JobCompletedRx { + JobCompleted(SendResult), + Killpill, + } //if we have been killed, we want to drain the queue of jobs while let Some(sr) = { if has_been_killed && same_worker_queue_size.load(Ordering::SeqCst) == 0 { - job_completed_rx.try_recv().ok() + job_completed_rx + .try_recv() + .ok() + .map(JobCompletedRx::JobCompleted) } else { - job_completed_rx.recv().await + tokio::select! { + result = job_completed_rx.recv_async() => { + result.ok().map(JobCompletedRx::JobCompleted) + } + _ = killpill_rx.recv() => { + Some(JobCompletedRx::Killpill) + } + } } } { #[cfg(feature = "benchmark")] let mut bench = BenchmarkIter::new(); match sr { - SendResult::JobCompleted(jc) => { + JobCompletedRx::JobCompleted(SendResult::JobCompleted(jc)) => { let is_init_script_and_failure = !jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG; let is_dependency_job = matches!( @@ -192,7 +207,7 @@ pub fn start_background_processor( infos.add_iter(bench, true); } } - SendResult::UpdateFlow { + JobCompletedRx::JobCompleted(SendResult::UpdateFlow { flow, w_id, success, @@ -200,7 +215,7 @@ pub fn start_background_processor( worker_dir, stop_early_override, token, - } => { + }) => { // let r; tracing::info!(parent_flow = %flow, "updating flow status"); if let Err(e) = update_flow_status_after_job_completion( @@ -230,7 +245,7 @@ pub fn start_background_processor( tracing::error!("Error updating flow status after job completion for {flow} on {worker_name}: {e:#}"); } } - SendResult::Kill => { + JobCompletedRx::Killpill => { has_been_killed = true; } } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 46420fb73e..2081b90f5f 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -85,6 +85,7 @@ use tokio::fs::symlink_file as symlink; use tokio::{ sync::{ + broadcast, mpsc::{self, Receiver, Sender}, RwLock, }, @@ -210,8 +211,8 @@ pub const DEFAULT_NATIVE_JOBS: usize = 1; const VACUUM_PERIOD: u32 = 50000; -#[cfg(any(target_os = "linux"))] -const DROP_CACHE_PERIOD: u32 = 1000; +// #[cfg(any(target_os = "linux"))] +// const DROP_CACHE_PERIOD: u32 = 1000; pub const MAX_BUFFERED_DEDICATED_JOBS: usize = 3; @@ -518,20 +519,33 @@ impl AuthedClient { } } +#[derive(Clone)] +pub struct SameWorkerSender(pub Sender, pub Arc); + #[allow(dead_code)] #[derive(Clone)] pub enum JobCompletedSender { - Sql(Sender), + Sql(flume::Sender, broadcast::Sender<()>), Http(HttpClient), NeverUsed, } impl JobCompletedSender { - pub fn new(conn: &Connection, buffer_size: usize) -> (Self, Option>) { + pub fn new( + conn: &Connection, + buffer_size: usize, + ) -> ( + Self, + Option<(flume::Receiver, broadcast::Receiver<()>)>, + ) { match conn { Connection::Sql(_) => { - let (sender, receiver) = mpsc::channel::(buffer_size); - (Self::Sql(sender), Some(receiver)) + let (sender, receiver) = flume::bounded::(buffer_size); + let (killpill_tx, killpill_rx) = broadcast::channel::<()>(buffer_size); + ( + Self::Sql(sender, killpill_tx), + Some((receiver, killpill_rx)), + ) } Connection::Http(client) => (Self::Http(client.clone()), None), } @@ -539,16 +553,11 @@ impl JobCompletedSender { pub fn new_never_used() -> (Self, Option>) { (Self::NeverUsed, None) } -} -#[derive(Clone)] -pub struct SameWorkerSender(pub Sender, pub Arc); - -impl JobCompletedSender { pub async fn send_job(&self, jc: JobCompleted) -> anyhow::Result<()> { match self { - Self::Sql(sender) => sender - .send(SendResult::JobCompleted(jc)) + Self::Sql(sender, _) => sender + .send_async(SendResult::JobCompleted(jc)) .await .map_err(|_e| { anyhow::anyhow!("Failed to send job completed to background processor") @@ -566,12 +575,9 @@ impl JobCompletedSender { } } - pub async fn send( - &self, - send_result: SendResult, - ) -> Result<(), tokio::sync::mpsc::error::SendError> { + pub async fn send(&self, send_result: SendResult) -> Result<(), flume::SendError> { match self { - Self::Sql(sender) => sender.send(send_result).await, + Self::Sql(sender, _) => sender.send_async(send_result).await, Self::Http(_) => { tracing::error!("Sending job completed to http client, this should not happen"); Ok(()) @@ -585,9 +591,13 @@ impl JobCompletedSender { } } - pub async fn kill(&self) -> Result<(), tokio::sync::mpsc::error::SendError> { + pub async fn kill(&self) -> Result<(), broadcast::error::SendError<()>> { match self { - Self::Sql(sender) => sender.send(SendResult::Kill).await, + Self::Sql(_, killpill_tx) => { + tracing::info!("Sending killpill to bg processors"); + killpill_tx.send(())?; + Ok(()) + } Self::Http(_) => { tracing::error!("Sending kill to http client, this should not happen"); Ok(()) @@ -627,11 +637,11 @@ pub async fn drop_cache() { Ok(mut file) => { // Write '3' to the file to drop caches if let Err(e) = tokio::io::AsyncWriteExt::write_all(&mut file, b"3").await { - tracing::warn!("Failed to write to /proc/sys/vm/drop_caches (expected to not work in not in privileged mode, only required to forcefully drop the cache to avoid spurrious oom killer): {}", e); + tracing::warn!("Failed to write to /proc/sys/vm/drop_caches (expected to work only in privileged mode, only required to forcefully drop the cache to avoid spurrious oom killer): {}", e); } } Err(e) => { - tracing::warn!("Failed to open /proc/sys/vm/drop_caches (expected to not work in not in privileged mode, only required to forcefully drop the cache to avoid spurrious oom killer):: {}", e); + tracing::warn!("Failed to open /proc/sys/vm/drop_caches (expected to work only in privileged mode, only required to forcefully drop the cache to avoid spurrious oom killer):: {}", e); } } } @@ -1012,19 +1022,22 @@ pub async fn run_worker( Arc::new(AtomicBool::new(matches!(conn, Connection::Http(_)))); let send_result = match (conn, job_completed_rx) { - (Connection::Sql(db), Some(job_completed_rx)) => Some(start_background_processor( - job_completed_rx, - job_completed_tx.clone(), - same_worker_queue_size.clone(), - job_completed_processor_is_done.clone(), - base_internal_url.to_string(), - db.clone(), - worker_dir.clone(), - same_worker_tx.clone(), - worker_name.clone(), - killpill_tx.clone(), - is_dedicated_worker, - )), + (Connection::Sql(db), Some((job_completed_rx, bg_killpill_rx))) => { + Some(start_background_processor( + job_completed_rx, + job_completed_tx.clone(), + same_worker_queue_size.clone(), + job_completed_processor_is_done.clone(), + base_internal_url.to_string(), + db.clone(), + worker_dir.clone(), + same_worker_tx.clone(), + worker_name.clone(), + bg_killpill_rx, + killpill_tx.clone(), + is_dedicated_worker, + )) + } _ => None, }; @@ -1190,11 +1203,11 @@ pub async fn run_worker( jobs_executed += 1; } - #[cfg(any(target_os = "linux"))] - if (jobs_executed as u32 + 1) % DROP_CACHE_PERIOD == 0 { - drop_cache().await; - jobs_executed += 1; - } + // #[cfg(any(target_os = "linux"))] + // if (jobs_executed as u32 + 1) % DROP_CACHE_PERIOD == 0 { + // drop_cache().await; + // jobs_executed += 1; + // } #[cfg(feature = "benchmark")] if benchmark_jobs > 0 && infos.iters == benchmark_jobs as u64 { @@ -1843,7 +1856,6 @@ pub enum SendResult { stop_early_override: Option, token: String, }, - Kill, } async fn do_nativets(