diff --git a/backend/.sqlx/query-3e8afd021088a99a24f27fa6f0a1b7f3edba3e9b834c814b464305bc2eb6ba80.json b/backend/.sqlx/query-3e8afd021088a99a24f27fa6f0a1b7f3edba3e9b834c814b464305bc2eb6ba80.json new file mode 100644 index 0000000000..1e8e44d120 --- /dev/null +++ b/backend/.sqlx/query-3e8afd021088a99a24f27fa6f0a1b7f3edba3e9b834c814b464305bc2eb6ba80.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation, native_mode, uses_batch_http_pull) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT (worker)\n DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_workers = EXCLUDED.dedicated_workers, native_mode = EXCLUDED.native_mode, uses_batch_http_pull = EXCLUDED.uses_batch_http_pull", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar", + "TextArray", + "Varchar", + "Varchar", + "TextArray", + "Varchar", + "Int8", + "Int8", + "Text", + "Bool", + "Bool" + ] + }, + "nullable": [] + }, + "hash": "3e8afd021088a99a24f27fa6f0a1b7f3edba3e9b834c814b464305bc2eb6ba80" +} diff --git a/backend/.sqlx/query-6cd099d458ac380d5da27b9e69da035755496ea50f2b78fb9b1cd3a2eb7e7625.json b/backend/.sqlx/query-6cd099d458ac380d5da27b9e69da035755496ea50f2b78fb9b1cd3a2eb7e7625.json new file mode 100644 index 0000000000..5d42486afe --- /dev/null +++ b/backend/.sqlx/query-6cd099d458ac380d5da27b9e69da035755496ea50f2b78fb9b1cd3a2eb7e7625.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2,\n occupancy_rate = $3, memory_usage = $4, wm_memory_usage = $5, vcpus = COALESCE($7, vcpus),\n memory = COALESCE($8, memory), occupancy_rate_15s = $9, occupancy_rate_5m = $10, occupancy_rate_30m = $11, native_mode = $12, uses_batch_http_pull = $13 WHERE worker = $6", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int4", + "TextArray", + "Float4", + "Int8", + "Int8", + "Text", + "Int8", + "Int8", + "Float4", + "Float4", + "Float4", + "Bool", + "Bool" + ] + }, + "nullable": [] + }, + "hash": "6cd099d458ac380d5da27b9e69da035755496ea50f2b78fb9b1cd3a2eb7e7625" +} diff --git a/backend/migrations/20260305000000_worker_ping_batch_pull.down.sql b/backend/migrations/20260305000000_worker_ping_batch_pull.down.sql new file mode 100644 index 0000000000..bccfcd2f54 --- /dev/null +++ b/backend/migrations/20260305000000_worker_ping_batch_pull.down.sql @@ -0,0 +1 @@ +ALTER TABLE worker_ping DROP COLUMN IF EXISTS uses_batch_http_pull; diff --git a/backend/migrations/20260305000000_worker_ping_batch_pull.up.sql b/backend/migrations/20260305000000_worker_ping_batch_pull.up.sql new file mode 100644 index 0000000000..be3343f230 --- /dev/null +++ b/backend/migrations/20260305000000_worker_ping_batch_pull.up.sql @@ -0,0 +1 @@ +ALTER TABLE worker_ping ADD COLUMN IF NOT EXISTS uses_batch_http_pull BOOLEAN NOT NULL DEFAULT false; diff --git a/backend/src/main.rs b/backend/src/main.rs index 97980484a5..061c637943 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -61,8 +61,9 @@ use windmill_common::{ MODE_AND_ADDONS, }, worker::{ - is_native_mode_from_env, reload_custom_tags_setting, Connection, HUB_CACHE_DIR, - HUB_RT_CACHE_DIR, NATIVE_MODE_RESOLVED, TMP_LOGS_DIR, WINDMILL_DIR, WORKER_GROUP, + is_native_mode_from_env, reload_custom_tags_setting, Connection, HttpClient, HUB_CACHE_DIR, + HUB_RT_CACHE_DIR, NATIVE_MODE_RESOLVED, TMP_LOGS_DIR, USES_BATCH_HTTP_PULL, WINDMILL_DIR, + WORKER_GROUP, }, KillpillSender, DEFAULT_HUB_BASE_URL, METRICS_ENABLED, }; @@ -1130,6 +1131,35 @@ Windmill Community Edition {GIT_VERSION} )?; let mut workers = vec![]; + // For native workers, create a self-signed JWT for batch pulling via HTTP. + // Only when server_mode is true (co-located server), since the HTTP + // endpoint lives on the server process. + let batch_pull_client = if NATIVE_MODE_RESOLVED + .load(std::sync::atomic::Ordering::Relaxed) + && server_mode + && mode != Mode::Agent + { + match create_native_batch_pull_client(&base_internal_url).await { + Ok(client) => { + tracing::info!( + "Native batch pull client created for HTTP pull at {}", + base_internal_url + ); + USES_BATCH_HTTP_PULL + .store(true, std::sync::atomic::Ordering::Relaxed); + Some(client) + } + Err(e) => { + tracing::warn!( + "Failed to create native batch pull client, falling back to SQL pull: {e:#}" + ); + None + } + } + } else { + None + }; + for i in 0..num_workers { let suffix = if i == 0 && first_suffix.is_some() { first_suffix.as_ref().unwrap().clone() @@ -1153,6 +1183,7 @@ Windmill Community Edition {GIT_VERSION} WORKER_GROUP.as_str(), &suffix, ), + batch_pull_client: batch_pull_client.clone(), }; workers.push(worker_conn); } @@ -1761,6 +1792,7 @@ fn display_config(envs: &[&str]) { pub struct WorkerConn { conn: Connection, worker_name: String, + batch_pull_client: Option, } pub async fn run_workers( @@ -1831,6 +1863,7 @@ pub async fn run_workers( let wk_conf = &workers[i as usize - 1]; let conn1 = wk_conf.conn.clone(); let worker_name = wk_conf.worker_name.clone(); + let batch_pull_client = wk_conf.batch_pull_client.clone(); WORKERS_NAMES.write().await.push(worker_name.clone()); let ip = ip.clone(); let rx = killpill_rxs.pop().unwrap(); @@ -1853,6 +1886,7 @@ pub async fn run_workers( rx, tx, &base_internal_url, + batch_pull_client.as_ref(), ); // #[cfg(tokio_unstable)] @@ -1871,6 +1905,33 @@ pub async fn run_workers( Ok(()) } +/// Create an HTTP client for native workers to pull jobs from the local server's batch buffer. +/// Self-signs a JWT with native_mode=true using the same JWT secret the server uses. +async fn create_native_batch_pull_client(base_internal_url: &str) -> anyhow::Result { + use windmill_common::agent_workers::{build_agent_http_client, AGENT_JWT_PREFIX}; + use windmill_common::jwt::encode_with_internal_secret; + + #[derive(serde::Serialize)] + struct NativeAgentAuth { + worker_group: String, + tags: Vec, + native_mode: Option, + } + + let worker_config = windmill_common::worker::WORKER_CONFIG.read().await; + let tags = worker_config.worker_tags.clone(); + drop(worker_config); + + let claims = + NativeAgentAuth { worker_group: WORKER_GROUP.to_string(), tags, native_mode: Some(true) }; + + let jwt = encode_with_internal_secret(claims).await?; + let token = format!("{}{}", AGENT_JWT_PREFIX, jwt); + + let suffix = create_default_worker_suffix(&HOSTNAME); + Ok(build_agent_http_client(&suffix, &token, base_internal_url)) +} + async fn send_delayed_killpill(tx: &KillpillSender, mut max_delay_secs: u64, context: &str) { if max_delay_secs == 0 { max_delay_secs = 1; diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 01db7590f5..f196804768 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -172,7 +172,7 @@ websocket_trigger: path(char), url(char), script_path(char), is_flow(bool), work windmill_migrations: name(text), created_at(ts) worker_group_job_stats: hour(bigint), worker_group(text), script_lang(char), workspace_id(char), job_count(int), total_duration_ms(bigint) FK: (workspace_id) -> workspace(id) -worker_ping: worker(char), worker_instance(char), ping_at(ts), started_at(ts), ip(char), jobs_executed(int), custom_tags(text[]), worker_group(char), dedicated_worker(char), wm_version(char), current_job_id(uuid), current_job_workspace_id(char), vcpus(bigint), memory(bigint), occupancy_rate(float), memory_usage(bigint), wm_memory_usage(bigint), occupancy_rate_15s(float), occupancy_rate_5m(float), occupancy_rate_30m(float), job_isolation(text), dedicated_workers(text[]) +worker_ping: worker(char), worker_instance(char), ping_at(ts), started_at(ts), ip(char), jobs_executed(int), custom_tags(text[]), worker_group(char), dedicated_worker(char), wm_version(char), current_job_id(uuid), current_job_workspace_id(char), vcpus(bigint), memory(bigint), occupancy_rate(float), memory_usage(bigint), wm_memory_usage(bigint), occupancy_rate_15s(float), occupancy_rate_5m(float), occupancy_rate_30m(float), job_isolation(text), dedicated_workers(text[]), native_mode(bool), uses_batch_http_pull(bool) workspace: id(char), name(char), owner(char), deleted(bool), premium(bool), parent_workspace_id(char) FK: (parent_workspace_id) -> workspace(id) workspace_dependencies: id(bigint), name(char), content(text), language(script_lang), description(text), archived(bool), workspace_id(char), created_at(ts) diff --git a/backend/tests/nativets_stress.rs b/backend/tests/nativets_stress.rs index 082a43be2c..3d3eb4da6f 100644 --- a/backend/tests/nativets_stress.rs +++ b/backend/tests/nativets_stress.rs @@ -241,6 +241,7 @@ fn spawn_workers( rx, tx2, &base_internal_url, + None, ) .await; }; diff --git a/backend/windmill-api-agent-workers/src/lib.rs b/backend/windmill-api-agent-workers/src/lib.rs index d2913a217c..795084052d 100644 --- a/backend/windmill-api-agent-workers/src/lib.rs +++ b/backend/windmill-api-agent-workers/src/lib.rs @@ -19,7 +19,10 @@ use windmill_common::DB; use axum::Router; #[cfg(not(feature = "private"))] -pub fn global_service(_job_completed_tx: windmill_worker::JobCompletedSender) -> Router { +pub fn global_service( + _job_completed_tx: windmill_worker::JobCompletedSender, + _batch_buffer: Option<()>, +) -> Router { Router::new() } @@ -31,6 +34,7 @@ pub fn workspaced_service( Router, Vec>, Option, + Option<()>, ) { use windmill_common::worker::Connection; use windmill_worker::JobCompletedSender; @@ -40,7 +44,7 @@ pub fn workspaced_service( let router = Router::new(); - (router, vec![], Some(job_completed_tx)) + (router, vec![], Some(job_completed_tx), None) } #[cfg(not(feature = "private"))] diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 528e60b02f..9c66526d4e 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -422,12 +422,16 @@ pub async fn run_server( }; #[cfg(feature = "agent_worker_server")] - let (agent_workers_router, agent_workers_bg_processor, agent_workers_job_completed_tx) = - if server_mode { - windmill_api_agent_workers::workspaced_service(db.clone(), _base_internal_url.clone()) - } else { - (Router::new(), vec![], None) - }; + let ( + agent_workers_router, + agent_workers_bg_processor, + agent_workers_job_completed_tx, + batch_buffer, + ) = if server_mode { + windmill_api_agent_workers::workspaced_service(db.clone(), _base_internal_url.clone()) + } else { + (Router::new(), vec![], None, None) + }; #[cfg(feature = "agent_worker_server")] let agent_cache = Arc::new(AgentCache::new()); @@ -612,6 +616,7 @@ pub async fn run_server( { windmill_api_agent_workers::global_service( agent_workers_job_completed_tx, + batch_buffer.clone(), ) .layer(Extension(agent_cache.clone())) } else { diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 8d95541f61..6b24ec4ef5 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -288,6 +288,10 @@ pub fn is_native_mode_from_env() -> bool { /// Use this for hot-path checks (e.g. per-job dispatch) to avoid read-locking WORKER_CONFIG. pub static NATIVE_MODE_RESOLVED: AtomicBool = AtomicBool::new(false); +/// Whether this worker uses HTTP batch pull (set at startup in main.rs). +/// Reported in worker_ping so the server knows which native workers to batch-pull for. +pub static USES_BATCH_HTTP_PULL: AtomicBool = AtomicBool::new(false); + pub static MIN_VERSION_IS_LATEST: AtomicBool = AtomicBool::new(false); #[derive(Clone)] pub struct HttpClient { @@ -477,6 +481,62 @@ pub fn make_pull_query(tags: &[String]) -> String { query } +pub fn make_batch_pull_query(tags: &[String], limit: u32) -> String { + format_batch_pull_query(format!( + "SELECT id + FROM v2_job_queue + WHERE running = false AND tag IN ({}) AND scheduled_for <= now() + ORDER BY priority DESC NULLS LAST, scheduled_for + FOR UPDATE SKIP LOCKED + LIMIT {limit}", + tags.iter().map(|x| format!("'{x}'")).join(", ") + )) +} + +fn format_batch_pull_query(peek: String) -> String { + // Optimizations vs single-row format_pull_query: + // 1. ANY(ARRAY(SELECT ...)) instead of IN (SELECT ...) — forces PG to materialize IDs + // into an array, enabling Bitmap Index Scan instead of Hash Semi Join / Nested Loop + // 2. r CTE chains off q (not peek) — only updates runtime for actually-locked rows, + // avoids re-scanning peek + // 3. No separate j CTE — join v2_job directly in final SELECT off q's IDs + format!( + "WITH peek AS ( + {} + ), q AS NOT MATERIALIZED ( + UPDATE v2_job_queue SET + running = true, + started_at = coalesce(started_at, now()), + suspend_until = null, + worker = $1 + WHERE id = ANY(ARRAY(SELECT id FROM peek)) + RETURNING + id, started_at, scheduled_for, + canceled_by, canceled_reason, worker, cache_ignore_s3_path, runnable_settings_handle + ), r AS NOT MATERIALIZED ( + UPDATE v2_job_runtime SET + ping = now() + WHERE id = ANY(ARRAY(SELECT id FROM q)) + ) SELECT j.id, j.workspace_id, j.parent_job, j.created_by, q.started_at, q.scheduled_for, + j.runnable_id, j.runnable_path, j.args, q.canceled_by, + q.canceled_reason, j.kind, j.trigger, j.trigger_kind, j.permissioned_as, + f.flow_status, j.script_lang, + j.same_worker, j.pre_run_error, j.visible_to_owner, + j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job, j.root_job, + j.timeout, j.flow_step_id, j.cache_ttl, q.cache_ignore_s3_path, q.runnable_settings_handle, j.priority, j.raw_code, j.raw_lock, j.raw_flow, + j.script_entrypoint_override, j.preprocessed, COALESCE(pj.runnable_path, j.args->>'_FLOW_PATH') as parent_runnable_path, + COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin, + p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email + FROM q + JOIN v2_job j ON q.id = j.id + LEFT JOIN v2_job_status f ON f.id = q.id + LEFT JOIN job_perms p ON p.job_id = q.id + LEFT JOIN v2_job pj ON j.parent_job = pj.id + ", + peek + ) +} + pub async fn store_pull_query(wc: &WorkerConfig) { let mut queries = vec![]; for tags in wc.priority_tags_sorted.iter() { @@ -1143,6 +1203,8 @@ pub struct Ping { pub occupancy_rate_30m: Option, pub job_isolation: Option, pub native_mode: Option, + #[serde(default)] + pub uses_batch_http_pull: Option, pub ping_type: PingType, } pub async fn update_ping_http( @@ -1167,6 +1229,7 @@ pub async fn update_ping_http( insert_ping.occupancy_rate_5m, insert_ping.occupancy_rate_30m, insert_ping.native_mode.unwrap_or(false), + insert_ping.uses_batch_http_pull.unwrap_or(false), db, ) .await? @@ -1194,6 +1257,7 @@ pub async fn update_ping_http( insert_ping.memory, insert_ping.job_isolation, insert_ping.native_mode.unwrap_or(false), + insert_ping.uses_batch_http_pull.unwrap_or(false), db, ) .await?; @@ -1326,11 +1390,12 @@ pub async fn insert_ping_query( memory: Option, job_isolation: Option, native_mode: bool, + uses_batch_http_pull: bool, db: &DB, ) -> anyhow::Result<()> { sqlx::query!( - "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation, native_mode) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT (worker) - DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_workers = EXCLUDED.dedicated_workers, native_mode = EXCLUDED.native_mode", + "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation, native_mode, uses_batch_http_pull) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT (worker) + DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_workers = EXCLUDED.dedicated_workers, native_mode = EXCLUDED.native_mode, uses_batch_http_pull = EXCLUDED.uses_batch_http_pull", worker_instance, worker_name, ip, @@ -1343,6 +1408,7 @@ pub async fn insert_ping_query( memory, job_isolation.as_deref(), native_mode, + uses_batch_http_pull, ) .execute(db) .await?; @@ -1434,12 +1500,13 @@ pub async fn update_worker_ping_main_loop_query( occupancy_rate_5m: Option, occupancy_rate_30m: Option, native_mode: bool, + uses_batch_http_pull: bool, db: &DB, ) -> anyhow::Result<()> { timeout(Duration::from_secs(10), sqlx::query!( "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, memory_usage = $4, wm_memory_usage = $5, vcpus = COALESCE($7, vcpus), - memory = COALESCE($8, memory), occupancy_rate_15s = $9, occupancy_rate_5m = $10, occupancy_rate_30m = $11, native_mode = $12 WHERE worker = $6", + memory = COALESCE($8, memory), occupancy_rate_15s = $9, occupancy_rate_5m = $10, occupancy_rate_30m = $11, native_mode = $12, uses_batch_http_pull = $13 WHERE worker = $6", jobs_executed, tags, occupancy_rate, @@ -1452,6 +1519,7 @@ pub async fn update_worker_ping_main_loop_query( occupancy_rate_5m, occupancy_rate_30m, native_mode, + uses_batch_http_pull, ) .execute(db)) .await??; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 806e3e99a2..b0889751e2 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3434,6 +3434,38 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>( Ok(job_and_suspended) } +/// Batch-pull up to `limit` jobs in a single query, marking them all as running. +/// The caller controls which tags are queried, so flow/dependency jobs are never +/// pulled (they use distinct tags like "flow" / "dependency"). +pub async fn batch_pull( + db: &Pool, + worker_name: &str, + tags: &[String], + limit: u32, +) -> windmill_common::error::Result> { + use windmill_common::worker::make_batch_pull_query; + + if limit == 0 || tags.is_empty() { + return Ok(vec![]); + } + + let query = make_batch_pull_query(tags, limit); + let jobs: Vec = timeout( + Duration::from_secs(15), + sqlx::query_as::<_, PulledJob>(&query) + .bind(worker_name) + .fetch_all(db), + ) + .await + .map_err(|_| { + windmill_common::error::Error::internal_err( + "batch_pull query timed out after 15s".to_string(), + ) + })??; + + Ok(jobs) +} + pub async fn custom_concurrency_key( db: &Pool, job_id: &Uuid, diff --git a/backend/windmill-test-utils/src/lib.rs b/backend/windmill-test-utils/src/lib.rs index 69e7fc0558..874ce621ed 100644 --- a/backend/windmill-test-utils/src/lib.rs +++ b/backend/windmill-test-utils/src/lib.rs @@ -412,6 +412,7 @@ pub fn spawn_test_worker( rx, tx2, &base_internal_url, + None, ) .await }; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 1239aa8d7d..7bf588264e 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -563,6 +563,7 @@ pub async fn update_worker_ping_for_failed_init_script( wm_memory_usage: None, job_isolation: None, native_mode: None, + uses_batch_http_pull: None, ping_type: PingType::InitScript, }, ) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 49a1048afd..fd08c904b1 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1359,6 +1359,7 @@ pub async fn run_worker( mut killpill_rx: tokio::sync::broadcast::Receiver<()>, killpill_tx: KillpillSender, base_internal_url: &str, + batch_pull_client: Option<&HttpClient>, ) { #[cfg(not(feature = "enterprise"))] if is_sandboxing_enabled() { @@ -2059,135 +2060,149 @@ pub async fn run_worker( continue; } } else { - match &conn { - Connection::Sql(db) => { - let pull_time = Instant::now(); - let likelihood_of_suspend = last_30jobs_suspended as f64 / 30.0; - - let suspend_first = suspend_first_success - || rand::random::() < likelihood_of_suspend - || last_suspend_first.elapsed().as_secs_f64() > 5.0; - - if suspend_first { - last_suspend_first = Instant::now(); - } - let mut job = match timeout( - Duration::from_secs(30), - pull( - &db, - suspend_first, - &worker_name, - None, - #[cfg(feature = "benchmark")] - &mut bench, - ) - .warn_after_seconds(2), - ) + // If batch_pull_client is set (native worker with co-located server), + // use HTTP pull from batch buffer. Otherwise use direct SQL pull. + if let Some(bpc) = batch_pull_client { + crate::agent_workers::pull_job(bpc, None, None) .await - { - Ok(job) => job, - Err(e) => { - tracing::error!(worker = %worker_name, hostname = %hostname, "pull timed out after 20s, sleeping for 30s: {e:?}"); - tokio::time::sleep(Duration::from_secs(30)).await; - continue; - } - }; + .map_err(|e| error::Error::InternalErr(e.to_string())) + .map(|x| x.map(|y| NextJob::Http(y))) + } else { + match &conn { + Connection::Sql(db) => { + let pull_time = Instant::now(); + let likelihood_of_suspend = last_30jobs_suspended as f64 / 30.0; - // Preprocess pulled job result - if let Ok(ref mut pulled_job_res) = job { - if let Err(e) = timeout( - // Will fail if longer than 10 seconds - core::time::Duration::from_secs(10), - pulled_job_res.maybe_apply_debouncing(db), - ) - .warn_after_seconds(2) - .await - // Flatten result - .map_err(error::Error::from) - .and_then(|r| r) - { - pulled_job_res.error_while_preprocessing = Some(e.to_string()); - } - } + let suspend_first = suspend_first_success + || rand::random::() < likelihood_of_suspend + || last_suspend_first.elapsed().as_secs_f64() > 5.0; - add_time!(bench, "job pulled from DB"); - let duration_pull_s = pull_time.elapsed().as_secs_f64(); - let err_pull = job.is_ok(); - // let empty = job.as_ref().is_ok_and(|x| x.is_none()); - - if duration_pull_s > 0.5 { - let empty = job.as_ref().is_ok_and(|x| x.job.is_none()); - tracing::warn!(worker = %worker_name, hostname = %hostname, "pull took more than 0.5s ({duration_pull_s}), this is a sign that the database is VERY undersized for this load. empty: {empty}, err: {err_pull}"); - #[cfg(feature = "prometheus")] - if empty { - if let Some(wp) = worker_pull_over_500_counter_empty.as_ref() { - wp.inc(); - } - } else if let Some(wp) = worker_pull_over_500_counter.as_ref() { - wp.inc(); - } - } else if duration_pull_s > 0.1 { - let empty = job.as_ref().is_ok_and(|x| x.job.is_none()); - tracing::warn!(worker = %worker_name, hostname = %hostname, "pull took more than 0.1s ({duration_pull_s}) this is a sign that the database is undersized for this load. empty: {empty}, err: {err_pull}"); - #[cfg(feature = "prometheus")] - if empty { - if let Some(wp) = worker_pull_over_100_counter_empty.as_ref() { - wp.inc(); - } - } else if let Some(wp) = worker_pull_over_100_counter.as_ref() { - wp.inc(); - } - } - - if let Ok(j) = job.as_ref() { - let suspend_success = j.suspended; if suspend_first { - if last_30jobs_suspended < 30 { - last_30jobs_suspended += 1; - } - } else { - last_30jobs_suspended -= 1; + last_suspend_first = Instant::now(); } - suspend_first_success = suspend_first && suspend_success; - #[cfg(feature = "prometheus")] - if j.job.is_some() { - if let Some(wp) = worker_pull_duration_counter.as_ref() { - wp.inc_by(duration_pull_s); + let mut job = match timeout( + Duration::from_secs(30), + pull( + &db, + suspend_first, + &worker_name, + None, + #[cfg(feature = "benchmark")] + &mut bench, + ) + .warn_after_seconds(2), + ) + .await + { + Ok(job) => job, + Err(e) => { + tracing::error!(worker = %worker_name, hostname = %hostname, "pull timed out after 20s, sleeping for 30s: {e:?}"); + tokio::time::sleep(Duration::from_secs(30)).await; + continue; } - if let Some(wp) = worker_pull_duration.as_ref() { - wp.observe(duration_pull_s); - } - } else { - if let Some(wp) = worker_pull_duration_counter_empty.as_ref() { - wp.inc_by(duration_pull_s); - } - if let Some(wp) = worker_pull_duration_empty.as_ref() { - wp.observe(duration_pull_s); + }; + + // Preprocess pulled job result + if let Ok(ref mut pulled_job_res) = job { + if let Err(e) = timeout( + // Will fail if longer than 10 seconds + core::time::Duration::from_secs(10), + pulled_job_res.maybe_apply_debouncing(db), + ) + .warn_after_seconds(2) + .await + // Flatten result + .map_err(error::Error::from) + .and_then(|r| r) + { + pulled_job_res.error_while_preprocessing = Some(e.to_string()); } } - } - match job { - Ok(pulled_job_result) => match pulled_job_result.to_pulled_job() { - Ok(j) => Ok(j.map(|job| NextJob::Sql { flow_runners: None, job })), - Err(PulledJobResultToJobErr::MissingConcurrencyKey(jc)) - | Err(PulledJobResultToJobErr::ErrorWhilePreprocessing(jc)) => { - if let Err(err) = job_completed_tx.send_job(jc, true).await { - tracing::error!( + + add_time!(bench, "job pulled from DB"); + let duration_pull_s = pull_time.elapsed().as_secs_f64(); + let err_pull = job.is_ok(); + // let empty = job.as_ref().is_ok_and(|x| x.is_none()); + + if duration_pull_s > 0.5 { + let empty = job.as_ref().is_ok_and(|x| x.job.is_none()); + tracing::warn!(worker = %worker_name, hostname = %hostname, "pull took more than 0.5s ({duration_pull_s}), this is a sign that the database is VERY undersized for this load. empty: {empty}, err: {err_pull}"); + #[cfg(feature = "prometheus")] + if empty { + if let Some(wp) = worker_pull_over_500_counter_empty.as_ref() { + wp.inc(); + } + } else if let Some(wp) = worker_pull_over_500_counter.as_ref() { + wp.inc(); + } + } else if duration_pull_s > 0.1 { + let empty = job.as_ref().is_ok_and(|x| x.job.is_none()); + tracing::warn!(worker = %worker_name, hostname = %hostname, "pull took more than 0.1s ({duration_pull_s}) this is a sign that the database is undersized for this load. empty: {empty}, err: {err_pull}"); + #[cfg(feature = "prometheus")] + if empty { + if let Some(wp) = worker_pull_over_100_counter_empty.as_ref() { + wp.inc(); + } + } else if let Some(wp) = worker_pull_over_100_counter.as_ref() { + wp.inc(); + } + } + + if let Ok(j) = job.as_ref() { + let suspend_success = j.suspended; + if suspend_first { + if last_30jobs_suspended < 30 { + last_30jobs_suspended += 1; + } + } else { + last_30jobs_suspended -= 1; + } + suspend_first_success = suspend_first && suspend_success; + #[cfg(feature = "prometheus")] + if j.job.is_some() { + if let Some(wp) = worker_pull_duration_counter.as_ref() { + wp.inc_by(duration_pull_s); + } + if let Some(wp) = worker_pull_duration.as_ref() { + wp.observe(duration_pull_s); + } + } else { + if let Some(wp) = worker_pull_duration_counter_empty.as_ref() { + wp.inc_by(duration_pull_s); + } + if let Some(wp) = worker_pull_duration_empty.as_ref() { + wp.observe(duration_pull_s); + } + } + } + match job { + Ok(pulled_job_result) => match pulled_job_result.to_pulled_job() { + Ok(j) => { + Ok(j.map(|job| NextJob::Sql { flow_runners: None, job })) + } + Err(PulledJobResultToJobErr::MissingConcurrencyKey(jc)) + | Err(PulledJobResultToJobErr::ErrorWhilePreprocessing(jc)) => { + if let Err(err) = job_completed_tx.send_job(jc, true).await + { + tracing::error!( "An error occurred while sending job completed: {:#?}", err ) + } + Ok(None) } - Ok(None) - } - }, - Err(err) => Err(err), + }, + Err(err) => Err(err), + } + } + + Connection::Http(client) => { + crate::agent_workers::pull_job(&client, None, None) + .await + .map_err(|e| error::Error::InternalErr(e.to_string())) + .map(|x| x.map(|y| NextJob::Http(y))) } } - - Connection::Http(client) => crate::agent_workers::pull_job(&client, None, None) - .await - .map_err(|e| error::Error::InternalErr(e.to_string())) - .map(|x| x.map(|y| NextJob::Http(y))), } } }; diff --git a/backend/windmill-worker/src/worker_utils.rs b/backend/windmill-worker/src/worker_utils.rs index 7a8c1e7ebb..8606b2c272 100644 --- a/backend/windmill-worker/src/worker_utils.rs +++ b/backend/windmill-worker/src/worker_utils.rs @@ -8,7 +8,7 @@ use windmill_common::{ get_memory, get_vcpus, get_windmill_memory_usage, get_worker_memory_usage, insert_ping_query, update_job_ping_query, update_worker_ping_from_job_query, update_worker_ping_main_loop_query, Connection, Ping, PingType, NATIVE_MODE_RESOLVED, - WORKER_CONFIG, WORKER_GROUP, + USES_BATCH_HTTP_PULL, WORKER_CONFIG, WORKER_GROUP, }, KillpillSender, DB, }; @@ -31,6 +31,7 @@ pub(crate) async fn update_worker_ping_full( let tags = wc.worker_tags.clone(); let native_mode = wc.native_mode; drop(wc); + let uses_batch_http_pull = USES_BATCH_HTTP_PULL.load(std::sync::atomic::Ordering::Relaxed); let memory_usage = get_worker_memory_usage(); let wm_memory_usage = get_windmill_memory_usage(); @@ -64,6 +65,7 @@ pub(crate) async fn update_worker_ping_full( occupancy_rate_5m, occupancy_rate_30m, native_mode, + uses_batch_http_pull, ) }) .retry( @@ -110,6 +112,7 @@ async fn update_worker_ping_full_inner( occupancy_rate_5m: Option, occupancy_rate_30m: Option, native_mode: bool, + uses_batch_http_pull: bool, ) -> anyhow::Result<()> { match conn { Connection::Sql(db) => { @@ -126,6 +129,7 @@ async fn update_worker_ping_full_inner( occupancy_rate_5m, occupancy_rate_30m, native_mode, + uses_batch_http_pull, db, ) .await?; @@ -155,6 +159,7 @@ async fn update_worker_ping_full_inner( wm_memory_usage: get_windmill_memory_usage(), job_isolation: None, native_mode: Some(native_mode), + uses_batch_http_pull: Some(uses_batch_http_pull), ping_type: PingType::MainLoop, }, ) @@ -186,6 +191,7 @@ pub async fn insert_ping( wc.native_mode, ) }; + let uses_batch_http_pull = USES_BATCH_HTTP_PULL.load(std::sync::atomic::Ordering::Relaxed); let vcpus = get_vcpus(); let memory = get_memory(); @@ -213,6 +219,7 @@ pub async fn insert_ping( memory, job_isolation, native_mode, + uses_batch_http_pull, db, ) .await?; @@ -242,6 +249,7 @@ pub async fn insert_ping( wm_memory_usage: get_windmill_memory_usage(), job_isolation, native_mode: Some(native_mode), + uses_batch_http_pull: Some(uses_batch_http_pull), ping_type: PingType::Initial, }, ) @@ -318,6 +326,9 @@ pub async fn update_worker_ping_from_job( native_mode: Some( NATIVE_MODE_RESOLVED.load(std::sync::atomic::Ordering::Relaxed), ), + uses_batch_http_pull: Some( + USES_BATCH_HTTP_PULL.load(std::sync::atomic::Ordering::Relaxed), + ), }, ) .await?;