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

allow multiple workers on agent mode (#5607)

Browse Source
This commit is contained in:
Ruben Fiszel
2025-04-11 23:19:29 +02:00
committed by GitHub
parent cdb0e42979
commit d5186da271
11 changed files with 176 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json generated
View File

@@ -15,7 +15,7 @@
]
},
"nullable": [
true
null
]
},
"hash": "ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927"

21
backend/.sqlx/query-fe7221651a982861dede4116bc71fe2dce615ff76a53f72cb5386dc17e4e07aa.json generated Normal file
View File

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

11
backend/Cargo.lock generated
View File

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

2
backend/Cargo.toml
View File

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

2
backend/ee-repo-ref.txt
View File

@@ -1 +1 @@
85c37983ffb8f622458425c182613206625c6cee
44c3e23922097d4386183e56b6b3dc540c70d71b

84
backend/src/main.rs
View File

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

15
backend/windmill-api/src/jobs.rs
View File

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

6
backend/windmill-api/src/lib.rs
View File

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

1
backend/windmill-worker/Cargo.toml
View File

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

31
backend/windmill-worker/src/result_processor.rs
View File

@@ -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<SendResult>,
job_completed_rx: flume::Receiver<SendResult>,
job_completed_sender: JobCompletedSender,
same_worker_queue_size: Arc<AtomicU16>,
job_completed_processor_is_done: Arc<AtomicBool>,
@@ -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;
}
}

94
backend/windmill-worker/src/worker.rs
View File

@@ -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<SameWorkerPayload>, pub Arc<AtomicU16>);
#[allow(dead_code)]
#[derive(Clone)]
pub enum JobCompletedSender {
Sql(Sender<SendResult>),
Sql(flume::Sender<SendResult>, broadcast::Sender<()>),
Http(HttpClient),
NeverUsed,
}
impl JobCompletedSender {
pub fn new(conn: &Connection, buffer_size: usize) -> (Self, Option<Receiver<SendResult>>) {
pub fn new(
conn: &Connection,
buffer_size: usize,
) -> (
Self,
Option<(flume::Receiver<SendResult>, broadcast::Receiver<()>)>,
) {
match conn {
Connection::Sql(_) => {
let (sender, receiver) = mpsc::channel::<SendResult>(buffer_size);
(Self::Sql(sender), Some(receiver))
let (sender, receiver) = flume::bounded::<SendResult>(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<Receiver<SendResult>>) {
(Self::NeverUsed, None)
}
}
#[derive(Clone)]
pub struct SameWorkerSender(pub Sender<SameWorkerPayload>, pub Arc<AtomicU16>);
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<SendResult>> {
pub async fn send(&self, send_result: SendResult) -> Result<(), flume::SendError<SendResult>> {
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<SendResult>> {
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<bool>,
token: String,
},
Kill,
}
async fn do_nativets(