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

feat: batch job pulling for native workers

Browse Source 
Reduce DB polling overhead for native workers by batch-fetching jobs
server-side and serving them from an in-memory buffer via HTTP.

- Add batch_pull() in windmill-queue: single SELECT...FOR UPDATE SKIP LOCKED LIMIT N
- Add batch pull SQL helpers (make_batch_pull_query, format_batch_pull_query)
- OSS stubs for agent-workers accept batch_buffer parameter (4-tuple return)
- Native workers self-sign JWT and pull jobs via HTTP when co-located with server
- Add uses_batch_http_pull column to worker_ping for server-side tracking
- Worker pull loop: HTTP batch pull when client available, SQL otherwise

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
HugoCasa
2026-03-05 16:08:28 +01:00
parent 63ebae8829
commit 876a9cfc8e
15 changed files with 383 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
backend/.sqlx/query-3e8afd021088a99a24f27fa6f0a1b7f3edba3e9b834c814b464305bc2eb6ba80.json generated Normal file
View File

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

26
backend/.sqlx/query-6cd099d458ac380d5da27b9e69da035755496ea50f2b78fb9b1cd3a2eb7e7625.json generated Normal file
View File

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

1
backend/migrations/20260305000000_worker_ping_batch_pull.down.sql Normal file
View File

@@ -0,0 +1 @@
ALTER TABLE worker_ping DROP COLUMN IF EXISTS uses_batch_http_pull;

1
backend/migrations/20260305000000_worker_ping_batch_pull.up.sql Normal file
View File

@@ -0,0 +1 @@
ALTER TABLE worker_ping ADD COLUMN IF NOT EXISTS uses_batch_http_pull BOOLEAN NOT NULL DEFAULT false;

65
backend/src/main.rs
View File

@@ -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<HttpClient>,
}
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<HttpClient> {
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<String>,
native_mode: Option<bool>,
}
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;

2
backend/summarized_schema.txt
View File

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

1
backend/tests/nativets_stress.rs
View File

@@ -241,6 +241,7 @@ fn spawn_workers(
rx,
tx2,
&base_internal_url,
None,
)
.await;
};

8
backend/windmill-api-agent-workers/src/lib.rs
View File

@@ -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<tokio::task::JoinHandle<()>>,
Option<windmill_worker::JobCompletedSender>,
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"))]

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

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

74
backend/windmill-common/src/worker.rs
View File

@@ -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<f32>,
pub job_isolation: Option<String>,
pub native_mode: Option<bool>,
#[serde(default)]
pub uses_batch_http_pull: Option<bool>,
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<i64>,
job_isolation: Option<String>,
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<f32>,
occupancy_rate_30m: Option<f32>,
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??;

32
backend/windmill-queue/src/jobs.rs
View File

@@ -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<Postgres>,
worker_name: &str,
tags: &[String],
limit: u32,
) -> windmill_common::error::Result<Vec<PulledJob>> {
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<PulledJob> = 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<Postgres>,
job_id: &Uuid,

1
backend/windmill-test-utils/src/lib.rs
View File

@@ -412,6 +412,7 @@ pub fn spawn_test_worker(
rx,
tx2,
&base_internal_url,
None,
)
.await
};

1
backend/windmill-worker/src/common.rs
View File

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

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

@@ -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::<f64>() < 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::<f64>() < 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))),
}
}
};

13
backend/windmill-worker/src/worker_utils.rs
View File

@@ -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<f32>,
occupancy_rate_30m: Option<f32>,
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?;