starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-25 00:01:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve suspend_first behavior and frequency

Browse Source
This commit is contained in:
Ruben Fiszel
2024-09-28 15:43:06 +02:00
parent e8e6e233de
commit b5e226b977
15 changed files with 691 additions and 413 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.toml
+1 -1
View File
@@ -44,7 +44,7 @@ default = []
enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise", "windmill-git-sync/enterprise", "windmill-common/prometheus", "windmill-common/enterprise", "windmill-indexer/enterprise"]
enterprise_saml = ["windmill-api/enterprise_saml"]
stripe = ["windmill-api/stripe"]
benchmark = ["windmill-api/benchmark", "windmill-worker/benchmark", "windmill-queue/benchmark"]
benchmark = ["windmill-api/benchmark", "windmill-worker/benchmark", "windmill-queue/benchmark", "windmill-common/benchmark"]
flamegraph = ["windmill-common/flamegraph", "windmill-worker/flamegraph"]
loki = ["windmill-common/loki"]
pg_embed = ["dep:pg-embed"]
backend/plot.py
+42 -12
View File
@@ -11,14 +11,19 @@ def load_json_data(filepath):
def plot_two_arrays_of_subarrays(arrays1, arrays2):
# Function to calculate sum of durations for each step
def calculate_sums(arrays):
steps = [step for step, _ in arrays[0]] # Extract steps from the first iteration
sums = {step: 0 for step in steps} # Initialize sums dictionary with step names
steps = [step for step, _ in arrays[0]['timings']] # Extract steps from the first iteration
sums = {step: 0.0 for step in steps} # Initialize sums dictionary with step names
# Sum up the durations for each step across all subarrays
for subarray in arrays:
for step_name, duration in subarray:
for step_name, duration in subarray['timings']:
if step_name not in sums:
sums[step_name] = 0
sums[step_name] += duration
for step_name, duration in sums.items():
sums[step_name] = duration / 1000000000
# Convert the sums dictionary to two lists (for plotting)
step_names = list(sums.keys())
durations = list(sums.values())
@@ -29,21 +34,23 @@ def plot_two_arrays_of_subarrays(arrays1, arrays2):
step_names2, sums2 = calculate_sums(arrays2)
# Create two subplots, one on top of the other
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 12))
# First plot (top) for the first array of subarrays
ax1.plot(step_names1, sums1, marker='o', linestyle='-', color='b')
ax1.bar(step_names1, sums1, color='b')
ax1.set_title('Total Duration per Step - Main Loop')
ax1.set_xlabel('Step Name')
ax1.set_ylabel('Total Duration')
ax1.grid(True)
ax1.set_ylabel('Total Duration (s)')
ax1.grid(True, axis='y')
ax1.tick_params(axis='x', rotation=45)
# Second plot (bottom) for the second array of subarrays
ax2.plot(step_names2, sums2, marker='o', linestyle='-', color='r')
ax2.bar(step_names2, sums2, color='r')
ax2.set_title('Total Duration per Step - Result Processor')
ax2.set_xlabel('Step Name')
ax2.set_ylabel('Total Duration')
ax2.grid(True)
ax2.set_ylabel('Total Duration (s)')
ax2.grid(True, axis='y')
ax2.tick_params(axis='x', rotation=45)
# Adjust layout so the plots don't overlap
plt.tight_layout()
@@ -52,8 +59,31 @@ def plot_two_arrays_of_subarrays(arrays1, arrays2):
plt.show()
# Load arrays from the JSON files
arrays1 = load_json_data('/tmp/windmill/profiling_main.json')
arrays2 = load_json_data('/tmp/windmill/profiling_result_processor.json')
main = load_json_data('/tmp/windmill/profiling_main.json')
result_processor = load_json_data('/tmp/windmill/profiling_result_processor.json')
arrays1 = main['timings']
arrays2 = result_processor['timings']
total_duration1 = main['total_duration']/1000
total_duration2 = result_processor['total_duration']/1000
print(f"Total duration for main: {total_duration1}s")
print(f"Total duration for result processor: {total_duration2}s")
iterations_total = sum(main['iter_durations']) / 1000000000
iterations_total2 = sum(result_processor['iter_durations']) / 1000000000
print(f"Number of iterations: {len(main['iter_durations'])}")
print(f"Total iterations for main: {iterations_total}s")
print(f"Total iterations for result processor: {iterations_total2}s")
# Calculate RPS
rps1 = len(main['iter_durations']) / total_duration1
rps2 = len(result_processor['iter_durations']) / total_duration2
print(f"RPS for main: {rps1}")
print(f"RPS for result processor: {rps2}")
# Plot the data
plot_two_arrays_of_subarrays(arrays1, arrays2)
backend/src/monitor.rs
+2
View File
@@ -1368,6 +1368,8 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
rsmq.clone(),
worker_name,
send_result_never_used,
#[cfg(feature = "benchmark")]
&mut windmill_common::bench::BenchmarkIter::new(),
)
.await;
}
backend/windmill-common/Cargo.toml
+1
View File
@@ -11,6 +11,7 @@ jemalloc = ["dep:tikv-jemalloc-ctl"]
prometheus = ["dep:prometheus"]
flamegraph = ["dep:tracing-flame"]
loki = ["dep:tracing-loki"]
benchmark = []
parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"]
[lib]
backend/windmill-common/src/bench.rs
+213
View File
@@ -0,0 +1,213 @@
use crate::{
worker::{write_file, TMP_DIR},
DB,
};
use serde::Serialize;
use tokio::time::Instant;
#[derive(Serialize)]
pub struct BenchmarkInfo {
#[serde(skip)]
pub start: Instant,
#[serde(skip)]
pub iters: u64,
timings: Vec<BenchmarkIter>,
pub iter_durations: Vec<u64>,
pub total_duration: Option<u64>,
}
impl BenchmarkInfo {
pub fn new() -> Self {
BenchmarkInfo {
iters: 0,
timings: vec![],
start: Instant::now(),
iter_durations: vec![],
total_duration: None,
}
}
pub fn add_iter(&mut self, bench: BenchmarkIter, inc_iters: bool) {
if inc_iters {
self.iters += 1;
}
let elapsed_total = bench.start.elapsed().as_nanos() as u64;
self.timings.push(bench);
self.iter_durations.push(elapsed_total);
}
pub fn write_to_file(&mut self, path: &str) -> anyhow::Result<()> {
let total_duration = self.start.elapsed().as_millis() as u64;
self.total_duration = Some(total_duration as u64);
println!(
"Writing benchmark {path}, duration of benchmark: {total_duration}s and RPS: {}",
self.iters as f64 / total_duration as f64
);
write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling");
Ok(())
}
}
#[derive(Serialize)]
pub struct BenchmarkIter {
#[serde(skip)]
pub start: Instant,
#[serde(skip)]
last_instant: Instant,
last_step: String,
timings: Vec<(String, u32)>,
}
impl BenchmarkIter {
pub fn new() -> Self {
BenchmarkIter {
last_instant: Instant::now(),
timings: vec![],
start: Instant::now(),
last_step: String::new(),
}
}
pub fn add_timing(&mut self, name: &str) {
let elapsed = self.last_instant.elapsed().as_nanos() as u32;
self.timings
.push((format!("{}->{}", self.last_step, name), elapsed));
self.last_instant = Instant::now();
self.last_step = name.to_string();
}
}
pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
use crate::{jobs::JobKind, scripts::ScriptLang};
let benchmark_kind = std::env::var("BENCHMARK_KIND").unwrap_or("noop".to_string());
if benchmark_jobs > 0 {
match benchmark_kind.as_str() {
"dedicated" => {
// you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47
let hash = sqlx::query_scalar!(
"SELECT hash FROM script WHERE path = $1 AND workspace_id = $2",
"f/benchmarks/dedicated",
"admins"
)
.fetch_one(db)
.await
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))",
hash,
"f/benchmarks/dedicated",
JobKind::Script as JobKind,
ScriptLang::Bun as ScriptLang,
"admins:f/benchmarks/dedicated",
"admin",
"u/admin",
"admin@windmill.dev",
chrono::Utc::now(),
"admins",
benchmark_jobs
)
.execute(db)
.await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
}
"parallelflow" => {
//create dedicated script
sqlx::query!("INSERT INTO script (summary, description, dedicated_worker, content, workspace_id, path, hash, language, tag, created_by, lock) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '') ON CONFLICT (workspace_id, hash) DO NOTHING",
"export async function main() {
console.log('hello world');
}",
"admins",
"u/admin/parallelflow",
1234567890,
ScriptLang::Deno as ScriptLang,
"flow",
"admin",
)
.execute(db)
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs {_e:#}"));
sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, raw_flow, flow_status) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12 FROM generate_series(1, 1))",
None::<i64>,
None::<String>,
JobKind::FlowPreview as JobKind,
ScriptLang::Deno as ScriptLang,
"flow",
"admin",
"u/admin",
"admin@windmill.dev",
chrono::Utc::now(),
"admins",
serde_json::from_str::<serde_json::Value>(r#"
{
"modules": [
{
"id": "a",
"value": {
"type": "forloopflow",
"modules": [
{
"id": "b",
"value": {
"path": "u/admin/parallelflow",
"type": "script",
"tag_override": "",
"input_transforms": {}
},
"summary": "calctest"
}
],
"iterator": {
"expr": "[...new Array(300)]",
"type": "javascript"
},
"parallel": true,
"parallelism": 10,
"skip_failures": true
}
}
],
"preprocessor_module": null
}
"#).unwrap(),
serde_json::from_str::<serde_json::Value>(r#"
{
"step": 0,
"modules": [
{
"id": "a",
"type": "WaitingForPriorSteps"
}
],
"cleanup_module": {},
"failure_module": {
"id": "failure",
"type": "WaitingForPriorSteps"
},
"preprocessor_module": null
}
"#).unwrap()
)
.execute(db)
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs"));
}
_ => {
sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))",
None::<i64>,
None::<String>,
JobKind::Noop as JobKind,
ScriptLang::Deno as ScriptLang,
"deno",
"admin",
"u/admin",
"admin@windmill.dev",
chrono::Utc::now(),
"admins",
benchmark_jobs
)
.execute(db)
.await.unwrap_or_else(|_e| panic!("failed to insert noop jobs"));
}
}
}
}
backend/windmill-common/src/lib.rs
+13
View File
@@ -17,6 +17,8 @@ use scripts::ScriptLang;
use sqlx::{Pool, Postgres};
pub mod apps;
#[cfg(feature = "benchmark")]
pub mod bench;
pub mod db;
pub mod ee;
pub mod error;
@@ -51,6 +53,17 @@ pub const DEFAULT_MAX_CONNECTIONS_INDEXER: u32 = 5;
pub const DEFAULT_HUB_BASE_URL: &str = "https://hub.windmill.dev";
#[macro_export]
macro_rules! add_time {
($bench:expr, $name:expr) => {
#[cfg(feature = "benchmark")]
{
$bench.add_timing($name);
// println!("{}: {:?}", $z, $y.elapsed());
}
};
}
lazy_static::lazy_static! {
pub static ref METRICS_PORT: u16 = std::env::var("METRICS_PORT")
.ok()
backend/windmill-queue/Cargo.toml
+1 -1
View File
@@ -12,7 +12,7 @@ path = "src/lib.rs"
default = []
enterprise = ["windmill-common/enterprise"]
cloud = []
benchmark = []
benchmark = ["windmill-common/benchmark"]
prometheus = ["dep:prometheus"]
[dependencies]
backend/windmill-queue/src/jobs.rs
+26 -14
View File
@@ -39,6 +39,7 @@ use windmill_audit::audit_ee::{audit_log, AuditAuthor};
use windmill_audit::ActionKind;
use windmill_common::{
add_time,
auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username},
db::{Authed, UserDB},
error::{self, to_anyhow, Error},
@@ -178,6 +179,8 @@ pub async fn cancel_single_job<'c>(
rsmq.clone(),
"server",
false,
#[cfg(feature = "benchmark")]
&mut windmill_common::bench::BenchmarkIter::new(),
)
.await;
@@ -495,6 +498,7 @@ pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Sen
rsmq: Option<R>,
_worker_name: &str,
flow_is_done: bool,
#[cfg(feature = "benchmark")] bench: &mut windmill_common::bench::BenchmarkIter,
) -> Result<WrappedError, Error> {
#[cfg(feature = "prometheus")]
register_metric(
@@ -532,6 +536,8 @@ pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Sen
canceled_by,
rsmq,
flow_is_done,
#[cfg(feature = "benchmark")]
bench,
)
.await?;
Ok(result)
@@ -555,10 +561,12 @@ pub async fn add_completed_job<
canceled_by: Option<CanceledBy>,
rsmq: Option<R>,
flow_is_done: bool,
#[cfg(feature = "benchmark")] bench: &mut windmill_common::bench::BenchmarkIter,
) -> Result<Uuid, Error> {
// tracing::error!("Start");
// let start = tokio::time::Instant::now();
add_time!(bench, "add_completed_job start");
if !result.is_valid_json() {
return Err(Error::InternalErr(
"Result of job is invalid json (empty)".to_string(),
@@ -566,6 +574,7 @@ pub async fn add_completed_job<
}
let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
let job_id = queued_job.id;
// tracing::error!("1 {:?}", start.elapsed());
@@ -576,6 +585,7 @@ pub async fn add_completed_job<
);
let mem_peak = mem_peak.max(queued_job.mem_peak.unwrap_or(0));
add_time!(bench, "add_completed_job query START");
let _duration: i64 = sqlx::query_scalar!(
"INSERT INTO completed_job AS cj
( workspace_id
@@ -647,6 +657,8 @@ pub async fn add_completed_job<
.map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?;
// tracing::error!("2 {:?}", start.elapsed());
add_time!(bench, "add_completed_job query END");
if !queued_job.is_flow_step {
if _duration > 500
&& (queued_job.job_kind == JobKind::Script || queued_job.job_kind == JobKind::Preview)
@@ -1766,9 +1778,9 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
db: &Pool<Postgres>,
rsmq: Option<R>,
suspend_first: bool,
) -> windmill_common::error::Result<Option<QueuedJob>> {
) -> windmill_common::error::Result<(Option<QueuedJob>, bool)> {
loop {
let job = pull_single_job_and_mark_as_running_no_concurrency_limit(
let (job, suspended) = pull_single_job_and_mark_as_running_no_concurrency_limit(
db,
rsmq.clone(),
suspend_first,
@@ -1776,7 +1788,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
.await?;
if job.is_none() {
return Ok(None);
return Ok((None, suspended));
}
let has_concurent_limit = job.as_ref().unwrap().concurrent_limit.is_some();
@@ -1796,7 +1808,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_PULL_COUNT.inc();
}
return Ok(Option::Some(pulled_job));
return Ok((Option::Some(pulled_job), suspended));
}
let itx = db.begin().await?;
@@ -1897,7 +1909,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
QUEUE_PULL_COUNT.inc();
}
tx.commit().await?;
return Ok(Option::Some(pulled_job));
return Ok((Option::Some(pulled_job), suspended));
}
let x = sqlx::query_scalar!(
"UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1 RETURNING (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids))",
@@ -2024,8 +2036,8 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
db: &Pool<Postgres>,
rsmq: Option<R>,
suspend_first: bool,
) -> windmill_common::error::Result<Option<QueuedJob>> {
let job: Option<QueuedJob> = if let Some(mut rsmq) = rsmq {
) -> windmill_common::error::Result<(Option<QueuedJob>, bool)> {
let job_and_suspended: (Option<QueuedJob>, bool) = if let Some(mut rsmq) = rsmq {
#[cfg(feature = "benchmark")]
let instant = Instant::now();
@@ -2083,9 +2095,9 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
#[cfg(feature = "benchmark")]
println!("rsmq 2: {:?}", instant.elapsed());
m2r
(m2r, false)
} else {
None
(None, false)
}
} else {
/* Jobs can be started if they:
@@ -2099,7 +2111,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
if query.is_empty() {
tracing::warn!("No suspended pull queries available");
return Ok(None);
return Ok((None, false));
}
let r = if suspend_first {
@@ -2119,7 +2131,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
if queries.is_empty() {
tracing::warn!("No pull queries available");
return Ok(None);
return Ok((None, false));
}
for query in queries.iter() {
@@ -2137,12 +2149,12 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
// #[cfg(feature = "benchmark")]
// println!("pull query: {:?}", instant.elapsed());
highest_priority_job
(highest_priority_job, false)
} else {
r
(r, true)
}
};
Ok(job)
Ok(job_and_suspended)
}
pub async fn custom_concurrency_key(
backend/windmill-worker/Cargo.toml
+1 -1
View File
@@ -12,7 +12,7 @@ path = "src/lib.rs"
default = []
prometheus = ["dep:prometheus", "windmill-common/prometheus"]
enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:pem", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"]
benchmark = ["windmill-queue/benchmark"]
benchmark = ["windmill-queue/benchmark", "windmill-common/benchmark"]
flamegraph = []
parquet = ["windmill-common/parquet", "dep:object_store"]
flow_testing = []
backend/windmill-worker/src/bench.rs
-111
View File
@@ -1,111 +0,0 @@
use serde::Serialize;
use tokio::time::Instant;
use windmill_common::{
worker::{write_file, TMP_DIR},
DB,
};
pub struct BenchmarkInfo {
iters: u64,
timings: Vec<BenchmarkIter>,
}
impl Serialize for BenchmarkInfo {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
let timings: Vec<Vec<(String, u32)>> = self
.timings
.iter()
.map(|x| x.timings.clone())
.collect::<Vec<Vec<(String, u32)>>>();
//serialize timings as vec of vec of tuples
timings.serialize(serializer)
}
}
impl BenchmarkInfo {
pub fn new() -> Self {
BenchmarkInfo { iters: 0, timings: vec![] }
}
pub fn add_iter(&mut self, bench: BenchmarkIter) {
self.iters += 1;
self.timings.push(bench);
}
pub fn write_to_file(&self, path: &str) -> anyhow::Result<()> {
println!("Writing benchmark {path}");
write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling");
Ok(())
}
}
pub struct BenchmarkIter {
last_instant: Instant,
timings: Vec<(String, u32)>,
}
impl BenchmarkIter {
pub fn new() -> Self {
BenchmarkIter { last_instant: Instant::now(), timings: vec![] }
}
pub fn add_timing(&mut self, name: &str) {
let elapsed = self.last_instant.elapsed().as_nanos() as u32;
self.timings.push((name.to_string(), elapsed));
self.last_instant = Instant::now();
}
}
pub async fn benchmark_init(is_dedicated_worker: bool, db: &DB) {
use windmill_common::{jobs::JobKind, scripts::ScriptLang};
let benchmark_jobs: i32 = std::env::var("BENCHMARK_JOBS_AT_INIT")
.unwrap_or("5000".to_string())
.parse::<i32>()
.unwrap();
if is_dedicated_worker {
// you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47
let hash = sqlx::query_scalar!(
"SELECT hash FROM script WHERE path = $1 AND workspace_id = $2",
"f/benchmarks/dedicated",
"admins"
)
.fetch_one(db)
.await
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))",
hash,
"f/benchmarks/dedicated",
JobKind::Script as JobKind,
ScriptLang::Bun as ScriptLang,
"admins:f/benchmarks/dedicated",
"admin",
"u/admin",
"admin@windmill.dev",
chrono::Utc::now(),
"admins",
benchmark_jobs
)
.execute(db)
.await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
} else {
sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))",
None::<i64>,
None::<String>,
JobKind::Noop as JobKind,
ScriptLang::Deno as ScriptLang,
"deno",
"admin",
"u/admin",
"admin@windmill.dev",
chrono::Utc::now(),
"admins",
benchmark_jobs
)
.execute(db)
.await.unwrap_or_else(|_e| panic!("failed to insert noop jobs"));
}
}
backend/windmill-worker/src/dedicated_worker.rs
+1 -1
View File
@@ -128,7 +128,7 @@ pub async fn handle_dedicated_process(
if let Err(e) = process_status(status) {
tracing::error!("child exit status was not success: {e:#}");
} else {
tracing::info!("child exist status was success");
tracing::info!("child exit status was success");
}
});
backend/windmill-worker/src/lib.rs
+1 -2
View File
@@ -7,8 +7,7 @@ mod snowflake_executor;
mod ansible_executor;
mod bash_executor;
#[cfg(feature = "benchmark")]
mod bench;
mod bun_executor;
pub mod common;
mod config;
backend/windmill-worker/src/result_processor.rs
+179 -12
View File
@@ -1,13 +1,21 @@
use serde::Serialize;
use sqlx::{types::Json, Pool, Postgres};
use std::{collections::HashMap, sync::Arc};
use std::{
collections::HashMap,
sync::{
atomic::{AtomicBool, AtomicU16, Ordering},
Arc,
},
};
use uuid::Uuid;
use windmill_common::{
add_time,
bench::BenchmarkInfo,
error::{self, Error},
jobs::QueuedJob,
worker::to_raw_value,
jobs::{JobKind, QueuedJob},
worker::{to_raw_value, WORKER_GROUP},
DB,
};
@@ -18,7 +26,13 @@ use windmill_queue::register_metric;
use serde_json::{json, value::RawValue};
use tokio::sync::mpsc::Sender;
use tokio::{
sync::{
self,
mpsc::{Receiver, Sender},
},
task::JoinHandle,
};
use windmill_queue::{add_completed_job, add_completed_job_error};
@@ -26,13 +40,151 @@ use crate::{
bash_executor::ANSI_ESCAPE_RE,
common::{read_result, save_in_cache},
worker_flow::update_flow_status_after_job_completion,
AuthedClient, JobCompleted, JobCompletedSender, SameWorkerSender, SendResult,
AuthedClient, JobCompleted, JobCompletedSender, SameWorkerSender, SendResult, INIT_SCRIPT_TAG,
};
use crate::add_time;
#[cfg(feature = "benchmark")]
use crate::bench::BenchmarkIter;
use windmill_common::bench::BenchmarkIter;
pub fn start_background_processor<R>(
mut job_completed_rx: Receiver<SendResult>,
job_completed_sender: Sender<SendResult>,
same_worker_queue_size: Arc<AtomicU16>,
job_completed_processor_is_done: Arc<AtomicBool>,
base_internal_url: String,
db: DB,
worker_dir: String,
same_worker_tx: SameWorkerSender,
rsmq: Option<R>,
worker_name: String,
killpill_tx: sync::broadcast::Sender<()>,
is_dedicated_worker: bool,
) -> JoinHandle<()>
where
R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static,
{
tokio::spawn(async move {
let mut has_been_killed = false;
#[cfg(feature = "benchmark")]
let mut infos = BenchmarkInfo::new();
//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()
} else {
job_completed_rx.recv().await
}
} {
#[cfg(feature = "benchmark")]
let mut bench = BenchmarkIter::new();
match sr {
SendResult::JobCompleted(jc) => {
let rsmq = rsmq.clone();
let is_init_script_and_failure =
!jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG;
let is_dependency_job = matches!(
jc.job.job_kind,
JobKind::Dependencies | JobKind::FlowDependencies
);
handle_receive_completed_job(
jc,
&base_internal_url,
&db,
&worker_dir,
&same_worker_tx,
rsmq,
&worker_name,
job_completed_sender.clone(),
#[cfg(feature = "benchmark")]
&mut bench,
)
.await;
if is_init_script_and_failure {
tracing::error!("init script errored, exiting");
killpill_tx.send(()).unwrap_or_default();
break;
}
if is_dependency_job && is_dedicated_worker {
tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted.");
sqlx::query!(
"UPDATE config SET config = config WHERE name = $1",
format!("worker__{}", *WORKER_GROUP)
)
.execute(&db)
.await
.expect("update config to trigger restart of all dedicated workers at that config");
killpill_tx.send(()).unwrap_or_default();
}
add_time!(bench, "job completed processed");
#[cfg(feature = "benchmark")]
{
infos.add_iter(bench, true);
}
}
SendResult::UpdateFlow {
flow,
w_id,
success,
result,
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(
&db,
&AuthedClient {
base_internal_url: base_internal_url.to_string(),
workspace: w_id.clone(),
token: token.clone(),
force_client: None,
},
flow,
&Uuid::nil(),
&w_id,
success,
Arc::new(result),
true,
same_worker_tx.clone(),
&worker_dir,
stop_early_override,
rsmq.clone(),
&worker_name,
job_completed_sender.clone(),
#[cfg(feature = "benchmark")]
&mut bench,
)
.await
{
tracing::error!("Error updating flow status after job completion for {flow} on {worker_name}: {e:#}");
}
}
SendResult::Kill => {
has_been_killed = true;
}
}
}
job_completed_processor_is_done.store(true, Ordering::SeqCst);
tracing::info!("finished processing all completed jobs");
#[cfg(feature = "benchmark")]
{
infos
.write_to_file("profiling_result_processor.json")
.expect("write to file profiling");
}
})
}
async fn send_job_completed(
job_completed_tx: JobCompletedSender,
@@ -214,6 +366,8 @@ pub async fn handle_receive_completed_job<
rsmq.clone(),
worker_name,
job_completed_tx,
#[cfg(feature = "benchmark")]
bench,
)
.await;
}
@@ -242,8 +396,6 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
let job_id = job.id.clone();
let workspace_id = job.workspace_id.clone();
add_time!(bench, "pre add_completed_job");
add_completed_job(
db,
&job,
@@ -254,11 +406,13 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
canceled_by,
rsmq.clone(),
false,
#[cfg(feature = "benchmark")]
bench,
)
.await?;
drop(job);
add_time!(bench, "post add_completed_job");
add_time!(bench, "add_completed_job END");
if is_flow_step {
if let Some(parent_job) = parent_job {
@@ -278,11 +432,13 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
rsmq.clone(),
worker_name,
job_completed_tx,
#[cfg(feature = "benchmark")]
bench,
)
.await?;
}
}
add_time!(bench, "post update_flow_status");
add_time!(bench, "updated flow status END");
} else {
let result = add_completed_job_error(
db,
@@ -295,6 +451,8 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
rsmq.clone(),
worker_name,
false,
#[cfg(feature = "benchmark")]
bench,
)
.await?;
if job.is_flow_step {
@@ -315,6 +473,8 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
rsmq,
worker_name,
job_completed_tx,
#[cfg(feature = "benchmark")]
bench,
)
.await?;
}
@@ -337,6 +497,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
rsmq: Option<R>,
worker_name: &str,
job_completed_tx: Sender<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) {
let err = match err {
Error::JsonErr(err) => err,
@@ -361,6 +522,8 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
rsmq_2,
worker_name,
false,
#[cfg(feature = "benchmark")]
bench,
)
.await
};
@@ -395,6 +558,8 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
rsmq.clone(),
worker_name,
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
bench,
)
.await;
@@ -420,6 +585,8 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
rsmq,
worker_name,
false,
#[cfg(feature = "benchmark")]
bench,
)
.await;
}
backend/windmill-worker/src/worker.rs
+184 -256
View File
@@ -18,9 +18,8 @@ use windmill_common::{
use anyhow::{Context, Result};
use const_format::concatcp;
#[cfg(feature = "prometheus")]
use prometheus::{
IntCounter,
};
use prometheus::IntCounter;
use tracing::Instrument;
#[cfg(feature = "prometheus")]
use windmill_common::METRICS_DEBUG_ENABLED;
@@ -31,13 +30,16 @@ use reqwest::Response;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use sqlx::{types::Json, Pool, Postgres};
use std::{
collections::{hash_map::DefaultHasher, HashMap}, fs::DirBuilder, hash::Hash, sync::{
collections::{hash_map::DefaultHasher, HashMap},
fs::DirBuilder,
hash::Hash,
sync::{
atomic::{AtomicBool, AtomicU16, Ordering},
Arc,
}, time::Duration
},
time::Duration,
};
use uuid::Uuid;
use windmill_common::{
@@ -52,8 +54,8 @@ use windmill_common::{
};
use windmill_queue::{
append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy,
PushArgs, PushIsolationLevel, HTTP_CLIENT,
append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy, PushArgs,
PushIsolationLevel, HTTP_CLIENT,
};
#[cfg(feature = "prometheus")]
@@ -78,15 +80,31 @@ use tokio::{
use rand::Rng;
use crate::{
ansible_executor::handle_ansible_job, bash_executor::{handle_bash_job, handle_powershell_job}, bun_executor::handle_bun_job, common::{
build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args, update_worker_ping_for_failed_init_script, OccupancyMetrics
}, deno_executor::handle_deno_job, go_executor::handle_go_job, graphql_executor::do_graphql, handle_child::SLOW_LOGS, handle_job_error, job_logger::NO_LOGS_AT_ALL, js_eval::{eval_fetch_timeout, transpile_ts}, mysql_executor::do_mysql, pg_executor::do_postgresql, php_executor::handle_php_job, python_executor::handle_python_job, result_processor::{handle_receive_completed_job, process_result}, rust_executor::handle_rust_job, worker_flow::{
handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress, Step,
}, worker_lockfiles::{
ansible_executor::handle_ansible_job,
bash_executor::{handle_bash_job, handle_powershell_job},
bun_executor::handle_bun_job,
common::{
build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args,
update_worker_ping_for_failed_init_script, OccupancyMetrics,
},
deno_executor::handle_deno_job,
go_executor::handle_go_job,
graphql_executor::do_graphql,
handle_child::SLOW_LOGS,
handle_job_error,
job_logger::NO_LOGS_AT_ALL,
js_eval::{eval_fetch_timeout, transpile_ts},
mysql_executor::do_mysql,
pg_executor::do_postgresql,
php_executor::handle_php_job,
python_executor::handle_python_job,
result_processor::{process_result, start_background_processor},
rust_executor::handle_rust_job,
worker_flow::{handle_flow, update_flow_status_in_progress, Step},
worker_lockfiles::{
handle_app_dependency_job, handle_dependency_job, handle_flow_dependency_job,
}
},
};
#[cfg(feature = "enterprise")]
@@ -98,18 +116,9 @@ use crate::{
};
#[cfg(feature = "benchmark")]
use crate::bench::{BenchmarkInfo, BenchmarkIter, benchmark_init};
use windmill_common::bench::{benchmark_init, BenchmarkInfo, BenchmarkIter};
#[macro_export]
macro_rules! add_time {
($bench:expr, $name:expr) => {
#[cfg(feature = "benchmark")]
{
$bench.add_timing($name);
// println!("{}: {:?}", $z, $y.elapsed());
}
};
}
use windmill_common::add_time;
pub async fn create_token_for_owner_in_bg(
db: &Pool<Postgres>,
@@ -552,17 +561,13 @@ impl AuthedClient {
}
}
#[allow(dead_code)]
#[derive(Clone)]
pub struct JobCompletedSender(Sender<SendResult>);
#[derive(Clone)]
pub struct SameWorkerSender(pub Sender<SameWorkerPayload>, pub Arc<AtomicU16>);
pub struct SameWorkerPayload {
pub job_id: Uuid,
pub recoverable: bool,
@@ -587,7 +592,6 @@ impl SameWorkerSender {
}
}
// on linux, we drop caches every DROP_CACHE_PERIOD to avoid OOM killer believing we are using too much memory just because we create lots of files when executing jobs
#[cfg(any(target_os = "linux"))]
pub async fn drop_cache() {
@@ -674,8 +678,6 @@ fn add_outstanding_wait_time(
}.in_current_span());
}
#[tracing::instrument(name = "worker", level = "info", skip_all, fields(worker = %worker_name, hostname = %hostname))]
pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
db: &Pool<Postgres>,
@@ -780,8 +782,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
None
};
#[cfg(feature = "prometheus")]
let worker_save_completed_job_duration = if METRICS_DEBUG_ENABLED.load(Ordering::Relaxed)
&& METRICS_ENABLED.load(Ordering::Relaxed)
@@ -798,8 +798,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
None
};
#[cfg(feature = "prometheus")]
let worker_pull_duration_counter_empty =
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
@@ -921,7 +919,13 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let is_dedicated_worker: bool = WORKER_CONFIG.read().await.dedicated_worker.is_some();
#[cfg(feature = "benchmark")]
benchmark_init(is_dedicated_worker, &db).await;
let benchmark_jobs: i32 = std::env::var("BENCHMARK_JOBS")
.unwrap_or("5000".to_string())
.parse::<i32>()
.unwrap();
#[cfg(feature = "benchmark")]
benchmark_init(benchmark_jobs, &db).await;
#[cfg(feature = "prometheus")]
if let Some(ws) = WORKER_STARTED.as_ref() {
@@ -930,158 +934,30 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let (same_worker_tx, mut same_worker_rx) = mpsc::channel::<SameWorkerPayload>(5);
let (job_completed_tx, mut job_completed_rx) = mpsc::channel::<SendResult>(3);
let (job_completed_tx, job_completed_rx) = mpsc::channel::<SendResult>(3);
let job_completed_tx = JobCompletedSender(
job_completed_tx,
);
let job_completed_tx = JobCompletedSender(job_completed_tx);
let same_worker_queue_size = Arc::new(AtomicU16::new(0));
let same_worker_tx = SameWorkerSender(same_worker_tx, same_worker_queue_size.clone());
let db2 = db.clone();
let base_internal_url2 = base_internal_url.to_string();
let same_worker_tx2 = same_worker_tx.clone();
let rsmq2 = rsmq.clone();
let worker_dir2 = worker_dir.clone();
let worker_name2 = worker_name.clone();
let killpill_tx2 = killpill_tx.clone();
let job_completed_sender = job_completed_tx.0.clone();
let job_completed_processor_is_done = Arc::new(AtomicBool::new(false));
let job_completed_processor_is_done2 = job_completed_processor_is_done.clone();
let same_worker_queue_size2 = same_worker_queue_size.clone();
let send_result = tokio::spawn(
(async move {
let mut has_been_killed = false;
#[cfg(feature = "benchmark")]
let mut infos = BenchmarkInfo::new();
//if we have been killed, we want to drain the queue of jobs
while let Some(sr) = {
if has_been_killed && same_worker_queue_size2.load(Ordering::SeqCst) == 0 {
job_completed_rx.try_recv().ok()
} else {
job_completed_rx.recv().await
}
} {
#[cfg(feature = "benchmark")]
let mut bench = BenchmarkIter::new();
match sr {
SendResult::JobCompleted(jc) => {
let rsmq2 = rsmq2.clone();
let is_init_script_and_failure = !jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG;
let is_dependency_job = matches!(
jc.job.job_kind,
JobKind::Dependencies | JobKind::FlowDependencies
);
add_time!(bench, "pre handle_receive_completed_job");
handle_receive_completed_job(
jc,
&base_internal_url2,
&db2,
&worker_dir2,
&same_worker_tx2,
rsmq2,
&worker_name2,
job_completed_sender.clone(),
#[cfg(feature = "benchmark")]
&mut bench
)
.await;
if is_init_script_and_failure {
tracing::error!("init script errored, exiting");
killpill_tx2.send(()).unwrap_or_default();
break;
}
if is_dependency_job && is_dedicated_worker {
tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted.");
sqlx::query!(
"UPDATE config SET config = config WHERE name = $1",
format!("worker__{}", *WORKER_GROUP)
)
.execute(&db2)
.await
.expect("update config to trigger restart of all dedicated workers at that config");
killpill_tx2.send(()).unwrap_or_default();
}
add_time!(bench, "post handle_receive_completed_job");
#[cfg(feature = "benchmark")] {
infos.add_iter(bench);
}
}
SendResult::UpdateFlow {
flow,
w_id,
success,
result,
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(
&db2,
&AuthedClient {
base_internal_url: base_internal_url2.to_string(),
workspace: w_id.clone(),
token: token.clone(),
force_client: None,
},
flow,
&Uuid::nil(),
&w_id,
success,
Arc::new(result),
true,
same_worker_tx2.clone(),
&worker_dir,
stop_early_override,
rsmq2.clone(),
&worker_name2,
job_completed_sender.clone(),
)
.await
{
tracing::error!("Error updating flow status after job completion for {flow} on {worker_name2}: {e:#}");
}
}
SendResult::Kill => {
has_been_killed = true;
}
}
}
job_completed_processor_is_done2.store(true, Ordering::SeqCst);
tracing::info!("finished processing all completed jobs");
#[cfg(feature = "benchmark")]
{
infos.write_to_file("profiling_result_processor.json").expect("write to file profiling");
}
})
.instrument(tracing::Span::current()),
let send_result = start_background_processor(
job_completed_rx,
job_completed_tx.0.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(),
rsmq.clone(),
worker_name.clone(),
killpill_tx.clone(),
is_dedicated_worker,
);
let mut last_executed_job: Option<Instant> = None;
let mut last_checked_suspended = Instant::now();
#[cfg(feature = "benchmark")]
let mut started = false;
@@ -1155,8 +1031,10 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
};
let mut suspend_first_success = false;
let mut last_reading = Instant::now() - Duration::from_secs(NUM_SECS_READINGS + 1);
let mut last_30jobs_suspended: Vec<bool> = vec![false; 30];
let mut last_suspend_first = Instant::now();
let mut killed_but_draining_same_worker_jobs = false;
loop {
#[cfg(feature = "benchmark")]
let mut bench = BenchmarkIter::new();
@@ -1185,7 +1063,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let memory_usage = get_worker_memory_usage();
let wm_memory_usage = get_windmill_memory_usage();
let (vcpus, memory) = if *REFRESH_CGROUP_READINGS
&& last_reading.elapsed().as_secs() > NUM_SECS_READINGS
{
@@ -1195,9 +1072,9 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
(None, None)
};
let (occupancy_rate, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m) =
occupancy_metrics.update_occupancy_metrics();
let (occupancy_rate, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m) = occupancy_metrics.update_occupancy_metrics();
if let Err(e) = 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),
@@ -1251,6 +1128,19 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
jobs_executed += 1;
}
#[cfg(feature = "benchmark")]
if benchmark_jobs > 0 && infos.iters == benchmark_jobs as u64 {
tracing::info!("benchmark finished, exiting");
job_completed_tx
.0
.send(SendResult::Kill)
.await
.expect("send kill to job completed tx");
break;
} else {
tracing::info!("benchmark not finished, still pulling jobs {}", infos.iters);
}
let next_job = {
// println!("2: {:?}", instant.elapsed());
#[cfg(feature = "benchmark")]
@@ -1275,7 +1165,11 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
tracing::error!(
"failed to fetch same_worker job on a non recoverable job, exiting"
);
job_completed_tx.0.send(SendResult::Kill).await.expect("send kill to job completed tx");
job_completed_tx
.0
.send(SendResult::Kill)
.await
.expect("send kill to job completed tx");
break;
} else {
r
@@ -1284,7 +1178,11 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
if !killed_but_draining_same_worker_jobs {
tracing::info!("received killpill for worker {}, jobs are not pulled anymore except same_worker jobs", i_worker);
killed_but_draining_same_worker_jobs = true;
job_completed_tx.0.send(SendResult::Kill).await.expect("send kill to job completed tx");
job_completed_tx
.0
.send(SendResult::Kill)
.await
.expect("send kill to job completed tx");
}
continue;
} else if killed_but_draining_same_worker_jobs {
@@ -1298,21 +1196,24 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
} else {
let pull_time = Instant::now();
let suspend_first =
if suspend_first_success || last_checked_suspended.elapsed().as_secs() > 3 {
last_checked_suspended = Instant::now();
true
} else {
false
};
add_time!(bench, "pre pull");
let likelihood_of_suspend =
(1.0 + last_30jobs_suspended.iter().filter(|&&x| x).count() as f64) / 31.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 job = pull(&db, rsmq.clone(), suspend_first).await;
add_time!(bench, "post pull");
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());
suspend_first_success = suspend_first && !empty;
// let empty = job.as_ref().is_ok_and(|x| x.is_none());
if !agent_mode && duration_pull_s > 0.5 {
let empty = job.as_ref().is_ok_and(|x| x.0.is_none());
tracing::warn!("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 {
@@ -1323,6 +1224,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
wp.inc();
}
} else if !agent_mode && duration_pull_s > 0.1 {
let empty = job.as_ref().is_ok_and(|x| x.0.is_none());
tracing::warn!("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 {
@@ -1334,8 +1236,16 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
}
#[cfg(feature = "prometheus")]
if let Ok(j) = job.as_ref() {
let suspend_success = j.1;
if suspend_first {
last_30jobs_suspended.push(suspend_success);
if last_30jobs_suspended.len() > 30 {
last_30jobs_suspended.remove(0);
}
}
suspend_first_success = suspend_first && suspend_success;
#[cfg(feature = "prometheus")]
if j.is_some() {
if let Some(wp) = worker_pull_duration_counter.as_ref() {
wp.inc_by(duration_pull_s);
@@ -1352,7 +1262,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
}
}
job
job.map(|x| x.0)
}
};
@@ -1364,21 +1274,14 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
occupancy_metrics.running_job_started_at = Some(Instant::now());
add_time!(bench, "pre next job");
match next_job {
Ok(Some(job)) => {
last_executed_job = None;
jobs_executed += 1;
tracing::debug!("started handling of job {}", job.id);
if matches!(job.job_kind, JobKind::Script | JobKind::Preview) {
if !dedicated_workers.is_empty() {
let key_o = if is_flow_worker {
job.flow_step_id.as_ref().map(|x| x.to_string())
@@ -1386,11 +1289,16 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
job.script_path.as_ref().map(|x| x.to_string())
};
if let Some(key) = key_o {
if let Some(dedicated_worker_tx) = dedicated_workers.get(&key) {
if let Some(dedicated_worker_tx) = dedicated_workers.get(&key) {
if let Err(e) = dedicated_worker_tx.send(Arc::new(job)).await {
tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}");
}
#[cfg(feature = "benchmark")]
{
add_time!(bench, "sent to dedicated worker");
infos.add_iter(bench, true);
}
continue;
}
@@ -1398,7 +1306,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
}
if matches!(job.job_kind, JobKind::Noop) {
add_time!(bench, "send job completed START");
job_completed_tx
.send(JobCompleted {
job: Arc::new(job),
@@ -1410,9 +1318,8 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
canceled_by: None,
})
.await
.expect("send job completed");
.expect("send job completed END");
add_time!(bench, "sent job completed");
} else {
let token = create_token_for_owner_in_bg(&db, &job).await;
add_outstanding_wait_time(&job, db, OUTSTANDING_WAIT_TIME_THRESHOLD_MS);
@@ -1477,7 +1384,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.expect("could not create job dir");
let same_worker = job.same_worker;
let folder = if job.language == Some(ScriptLang::Go) {
DirBuilder::new()
.recursive(true)
@@ -1487,7 +1394,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
} else {
""
};
let target = &format!("{job_dir}{folder}/shared");
if same_worker && job.parent_job.is_some() {
@@ -1521,7 +1428,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let is_init_script: bool = job.tag.as_str() == INIT_SCRIPT_TAG;
let arc_job = Arc::new(job);
add_time!(bench, "handle_queued_job START");
match handle_queued_job(
arc_job.clone(),
db,
@@ -1535,38 +1442,48 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
rsmq.clone(),
job_completed_tx.clone(),
&mut occupancy_metrics,
#[cfg(feature = "benchmark")]
&mut bench,
)
.await {
.await
{
Err(err) => {
handle_job_error(
db,
&authed_client.get_authed().await,
arc_job.as_ref(),
0,
None,
err,
false,
same_worker_tx.clone(),
&worker_dir,
rsmq.clone(),
&worker_name,
(&job_completed_tx.0).clone(),
)
.await;
if is_init_script {
tracing::error!("init script job failed (in handler), exiting");
update_worker_ping_for_failed_init_script(db, &worker_name, arc_job.id).await;
db,
&authed_client.get_authed().await,
arc_job.as_ref(),
0,
None,
err,
false,
same_worker_tx.clone(),
&worker_dir,
rsmq.clone(),
&worker_name,
(&job_completed_tx.0).clone(),
#[cfg(feature = "benchmark")]
&mut bench,
)
.await;
if is_init_script {
tracing::error!("init script job failed (in handler), exiting");
update_worker_ping_for_failed_init_script(
db,
&worker_name,
arc_job.id,
)
.await;
break;
}
}
Ok(false) if is_init_script => {
tracing::error!("init script job failed, exiting");
update_worker_ping_for_failed_init_script(db, &worker_name, arc_job.id)
.await;
break;
}
},
Ok(false) if is_init_script => {
tracing::error!("init script job failed, exiting");
update_worker_ping_for_failed_init_script(db, &worker_name, arc_job.id).await;
break;
_ => {}
}
_ => {}
}
#[cfg(feature = "prometheus")]
if let Some(duration) = _timer.map(|x| x.stop_and_record()) {
@@ -1594,13 +1511,12 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let _ = tokio::fs::remove_dir_all(job_dir).await;
}
}
add_time!(bench, "post next job");
#[cfg(feature = "benchmark")]
{
if started {
add_time!(bench, "post iter");
infos.add_iter(bench);
add_time!(bench, "job processed");
infos.add_iter(bench, true);
}
}
}
@@ -1623,9 +1539,13 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
None
};
tokio::time::sleep(Duration::from_millis(*SLEEP_QUEUE)).await;
#[cfg(feature = "benchmark")]
{
add_time!(bench, "sleep because empty job queue");
infos.add_iter(bench, false);
}
#[cfg(feature = "prometheus")]
_timer.map(|timer| {
let duration = timer.elapsed().as_secs_f64();
@@ -1638,18 +1558,17 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
tracing::error!("Failed to pull jobs: {}", err);
}
};
}
tracing::info!("worker {} exiting", worker_name);
#[cfg(feature = "benchmark")]
{
infos.write_to_file("profiling_main.json").expect("write to file profiling");
infos
.write_to_file("profiling_main.json")
.expect("write to file profiling");
}
drop(dedicated_workers);
let has_dedicated_workers = !dedicated_handles.is_empty();
@@ -1668,6 +1587,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
tracing::error!("error in awaiting send_result process: {e:?}")
}
tracing::info!("worker {} exited", worker_name);
tracing::info!("number of jobs executed: {}", jobs_executed);
}
async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
@@ -1741,7 +1661,6 @@ pub enum SendResult {
Kill,
}
#[derive(Debug, Clone)]
pub struct JobCompleted {
pub job: Arc<QueuedJob>,
@@ -1810,6 +1729,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
rsmq: Option<R>,
job_completed_tx: JobCompletedSender,
occupancy_metrics: &mut OccupancyMetrics,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> windmill_common::error::Result<bool> {
if job.canceled {
return Err(Error::JsonErr(canceled_job_to_result(&job)));
@@ -1906,7 +1826,11 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
)
.fetch_one(db)
.await
.map_err(|e| Error::InternalErr(format!("Fetching script path from queue for caching purposes: {e:#}")))?
.map_err(|e| {
Error::InternalErr(format!(
"Fetching script path from queue for caching purposes: {e:#}"
))
})?
.ok_or_else(|| Error::InternalErr(format!("Expected script_path")))?;
let step = match step.unwrap() {
Step::Step(i) => i.to_string(),
@@ -2082,7 +2006,8 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
occupancy_metrics,
)
.await;
occupancy_metrics.total_duration_of_running_jobs += metric_timer.elapsed().as_secs_f32();
occupancy_metrics.total_duration_of_running_jobs +=
metric_timer.elapsed().as_secs_f32();
r
}
};
@@ -2116,7 +2041,6 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
}
}
pub fn build_envs(
envs: Option<Vec<String>>,
) -> windmill_common::error::Result<HashMap<String, String>> {
@@ -2446,7 +2370,11 @@ async fn handle_code_execution_job(
);
let shared_mount = if job.same_worker && job.language != Some(ScriptLang::Deno) {
let folder = if job.language == Some(ScriptLang::Go) { "/go" } else { "" };
let folder = if job.language == Some(ScriptLang::Go) {
"/go"
} else {
""
};
format!(
r#"
mount {{
@@ -2618,7 +2546,6 @@ mount {{
.await
}
Some(ScriptLang::Ansible) => {
handle_ansible_job(
requirements_o,
job_dir,
@@ -2634,7 +2561,8 @@ mount {{
base_internal_url,
envs,
occupancy_metrics,
).await
)
.await
}
_ => panic!("unreachable, language is not supported: {language:#?}"),
};
backend/windmill-worker/src/worker_flow.rs
+26 -2
View File
@@ -29,7 +29,10 @@ use sqlx::FromRow;
use tokio::sync::mpsc::Sender;
use tracing::instrument;
use uuid::Uuid;
use windmill_common::add_time;
use windmill_common::auth::JobPerms;
#[cfg(feature = "benchmark")]
use windmill_common::bench::BenchmarkIter;
use windmill_common::db::Authed;
use windmill_common::flow_status::{
ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult,
@@ -76,6 +79,7 @@ pub async fn update_flow_status_after_job_completion<
rsmq: Option<R>,
worker_name: &str,
job_completed_tx: Sender<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> error::Result<()> {
// this is manual tailrecursion because async_recursion blows up the stack
// todo!();
@@ -97,6 +101,8 @@ pub async fn update_flow_status_after_job_completion<
rsmq.clone(),
worker_name,
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
bench,
)
.await?;
while let Some(nrec) = rec {
@@ -117,6 +123,8 @@ pub async fn update_flow_status_after_job_completion<
rsmq.clone(),
worker_name,
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
bench,
)
.await
{
@@ -141,6 +149,8 @@ pub async fn update_flow_status_after_job_completion<
rsmq.clone(),
worker_name,
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
bench,
)
.await?
}
@@ -193,7 +203,9 @@ pub async fn update_flow_status_after_job_completion_internal<
rsmq: Option<R>,
worker_name: &str,
job_completed_tx: Sender<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> error::Result<Option<RecUpdateFlowStatusAfterJobCompletion>> {
add_time!(bench, "update flow status internal START");
let (
should_continue_flow,
flow_job,
@@ -376,6 +388,8 @@ pub async fn update_flow_status_after_job_completion_internal<
})?;
}
add_time!(bench, "process module status START");
let (inc_step_counter, new_status) = match module_status {
FlowStatusModule::InProgress {
iterator,
@@ -517,9 +531,9 @@ pub async fn update_flow_status_after_job_completion_internal<
tracing::info!(
"parallel iteration {job_id_for_status} of flow {flow} has finished",
);
(true, Some(new_status))
} else {
add_time!(bench, "handle parallel flow start");
tx.commit().await?;
if parallelism.is_some() {
@@ -562,7 +576,7 @@ pub async fn update_flow_status_after_job_completion_internal<
r.unwrap()
);
}
add_time!(bench, "non final parallel flow finished");
return Ok(None);
}
}
@@ -964,6 +978,8 @@ pub async fn update_flow_status_after_job_completion_internal<
rsmq.clone(),
worker_name,
true,
#[cfg(feature = "benchmark")]
bench,
)
.await?;
} else {
@@ -994,6 +1010,8 @@ pub async fn update_flow_status_after_job_completion_internal<
let success = success
&& (!is_failure_step || result_has_recover_true(nresult.clone()))
&& !skip_error_handler;
add_time!(bench, "flow status update 1");
if success {
add_completed_job(
db,
@@ -1005,6 +1023,8 @@ pub async fn update_flow_status_after_job_completion_internal<
None,
rsmq.clone(),
true,
#[cfg(feature = "benchmark")]
bench,
)
.await?;
} else {
@@ -1022,6 +1042,8 @@ pub async fn update_flow_status_after_job_completion_internal<
None,
rsmq.clone(),
true,
#[cfg(feature = "benchmark")]
bench,
)
.await?;
}
@@ -1059,6 +1081,8 @@ pub async fn update_flow_status_after_job_completion_internal<
rsmq.clone(),
worker_name,
true,
#[cfg(feature = "benchmark")]
bench,
)
.await;
true