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

feat: refactor logging to improve performance by order of magnitude for longer jobs

Browse Source
This commit is contained in:
Ruben Fiszel
2024-03-16 17:55:34 +01:00
parent bab67fe3ae
commit e5366b84bb
27 changed files with 530 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/bash_executor.rs
+55 -38
View File
@@ -5,7 +5,7 @@ use serde_json::{json, value::RawValue};
use sqlx::types::Json;
use tokio::process::Command;
use windmill_common::{error::Error, jobs::QueuedJob, worker::to_raw_value};
use windmill_queue::CanceledBy;
use windmill_queue::{append_logs, CanceledBy};
const BIN_BASH: &str = "/bin/bash";
const NSJAIL_CONFIG_RUN_BASH_CONTENT: &str = include_str!("../nsjail/run.bash.config.proto");
@@ -19,7 +19,7 @@ lazy_static::lazy_static! {
use crate::{
common::{
build_args_map, get_reserved_variables, handle_child, read_file, read_file_content,
set_logs, start_child_process, write_file,
start_child_process, write_file,
},
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV,
POWERSHELL_CACHE_DIR, POWERSHELL_PATH, TZ_ENV,
@@ -32,7 +32,6 @@ lazy_static::lazy_static! {
#[tracing::instrument(level = "trace", skip_all)]
pub async fn handle_bash_job(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
@@ -45,14 +44,17 @@ pub async fn handle_bash_job(
worker_name: &str,
envs: HashMap<String, String>,
) -> Result<Box<RawValue>, Error> {
logs.push_str("\n\n--- BASH CODE EXECUTION ---\n");
set_logs(logs, &job.id, db).await;
let logs1 = "\n\n--- BASH CODE EXECUTION ---\n".to_string();
append_logs(job.id, job.workspace_id.clone(), logs1, db).await;
write_file(job_dir, "main.sh", &format!("set -e\n{content}")).await?;
write_file(
job_dir,
"main.sh",
&format!("set -e\n{content}\necho \"\"\nsleep 0.02"),
"wrapper.sh",
&format!("set -o pipefail\nset -e\nmkfifo bp\ncat bp | tail -1 > ./result2.out &\n /bin/bash ./main.sh \"$@\" 2>&1 | tee bp\nwait $!"),
)
.await?;
let token = client.get_token().await;
let mut reserved_variables = get_reserved_variables(job, &token, db).await?;
reserved_variables.insert("RUST_LOG".to_string(), "info".to_string());
@@ -76,6 +78,7 @@ pub async fn handle_bash_job(
let args = args_owned.iter().map(|s| &s[..]).collect::<Vec<&str>>();
let _ = write_file(job_dir, "result.json", "").await?;
let _ = write_file(job_dir, "result.out", "").await?;
let _ = write_file(job_dir, "result2.out", "").await?;
let child = if !*DISABLE_NSJAIL {
let _ = write_file(
@@ -87,7 +90,13 @@ pub async fn handle_bash_job(
.replace("{SHARED_MOUNT}", shared_mount),
)
.await?;
let mut cmd_args = vec!["--config", "run.config.proto", "--", "/bin/bash", "main.sh"];
let mut cmd_args = vec![
"--config",
"run.config.proto",
"--",
"/bin/bash",
"wrapper.sh",
];
cmd_args.extend(args);
let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str());
nsjail_cmd
@@ -101,7 +110,7 @@ pub async fn handle_bash_job(
.stderr(Stdio::piped());
start_child_process(nsjail_cmd, NSJAIL_PATH.as_str()).await?
} else {
let mut cmd_args = vec!["main.sh"];
let mut cmd_args = vec!["wrapper.sh"];
cmd_args.extend(&args);
let mut bash_cmd = Command::new(BIN_BASH);
bash_cmd
@@ -120,7 +129,6 @@ pub async fn handle_bash_job(
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
child,
@@ -148,13 +156,15 @@ pub async fn handle_bash_job(
}
}
//for now bash jobs have an empty result object
let last_line = serde_json::json!(logs
.lines()
.last()
.map(|x| ANSI_ESCAPE_RE.replace_all(x, "").to_string())
.unwrap_or_else(String::new));
Ok(to_raw_value(&last_line))
let result_out_path2 = format!("{job_dir}/result2.out");
if tokio::fs::metadata(&result_out_path2).await.is_ok() {
let result = read_file_content(&result_out_path2).await?;
return Ok(to_raw_value(&json!(result)));
}
Ok(to_raw_value(&json!(
"No result.out, result2.out or result.json found"
)))
}
fn raw_to_string(x: &str) -> String {
@@ -166,7 +176,6 @@ fn raw_to_string(x: &str) -> String {
}
#[tracing::instrument(level = "trace", skip_all)]
pub async fn handle_powershell_job(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
@@ -221,26 +230,26 @@ pub async fn handle_powershell_job(
.collect::<Vec<String>>();
let mut install_string: String = String::new();
let mut logs1 = String::new();
for line in content.lines() {
for cap in RE_POWERSHELL_IMPORTS.captures_iter(line) {
let module = cap.get(1).unwrap().as_str();
if !installed_modules.contains(&module.to_lowercase()) {
logs.push_str(&format!("\n{} not found in cache", module.to_string()));
logs1.push_str(&format!("\n{} not found in cache", module.to_string()));
// instead of using Install-Module, we use Save-Module so that we can specify the installation path
install_string.push_str(&format!(
"Save-Module -Path {} -Force {};",
POWERSHELL_CACHE_DIR, module
));
} else {
logs.push_str(&format!("\n{} found in cache", module.to_string()));
logs1.push_str(&format!("\n{} found in cache", module.to_string()));
}
}
}
set_logs(logs, &job.id, db).await;
if !install_string.is_empty() {
logs.push_str("\n\nInstalling modules...");
set_logs(logs, &job.id, db).await;
logs1.push_str("\n\nInstalling modules...");
append_logs(job.id, job.workspace_id.clone(), logs1, db).await;
let child = Command::new("pwsh")
.args(&["-Command", &install_string])
.stdout(Stdio::piped())
@@ -250,7 +259,6 @@ pub async fn handle_powershell_job(
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
child,
@@ -264,8 +272,9 @@ pub async fn handle_powershell_job(
.await?;
}
logs.push_str("\n\n--- POWERSHELL CODE EXECUTION ---\n");
set_logs(logs, &job.id, db).await;
let mut logs2 = "".to_string();
logs2.push_str("\n\n--- POWERSHELL CODE EXECUTION ---\n");
append_logs(job.id, job.workspace_id.clone(), logs2, db).await;
// make sure default (only allhostsallusers) modules are loaded, disable autoload (cache can be large to explore especially on cloud) and add /tmp/windmill/cache to PSModulePath
let profile = format!(
@@ -291,12 +300,19 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
};
write_file(job_dir, "main.ps1", content.as_str()).await?;
write_file(
job_dir,
"wrapper.sh",
&format!("set -o pipefail\nset -e\nmkfifo bp\ncat bp | tail -1 > ./result2.out &\n{} -F ./main.ps1 \"$@\" 2>&1 | tee bp\nwait $!", POWERSHELL_PATH.as_str()),
)
.await?;
let token = client.get_token().await;
let mut reserved_variables = get_reserved_variables(job, &token, db).await?;
reserved_variables.insert("RUST_LOG".to_string(), "info".to_string());
let _ = write_file(job_dir, "result.json", "").await?;
let _ = write_file(job_dir, "result.out", "").await?;
let _ = write_file(job_dir, "result2.out", "").await?;
let child = if !*DISABLE_NSJAIL {
let _ = write_file(
@@ -313,9 +329,8 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
"--config",
"run.config.proto",
"--",
POWERSHELL_PATH.as_str(),
"-F",
"main.ps1",
"/bin/bash",
"wrapper.sh",
];
cmd_args.extend(pwsh_args.iter().map(|x| x.as_str()));
Command::new(NSJAIL_PATH.as_str())
@@ -330,9 +345,9 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
.stderr(Stdio::piped())
.spawn()?
} else {
let mut cmd_args = vec!["-F", "main.ps1"];
let mut cmd_args = vec!["wrapper.sh"];
cmd_args.extend(pwsh_args.iter().map(|x| x.as_str()));
Command::new(POWERSHELL_PATH.as_str())
Command::new("/bin/bash")
.current_dir(job_dir)
.env_clear()
.envs(envs)
@@ -349,7 +364,6 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
child,
@@ -362,10 +376,13 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
)
.await?;
let last_line = serde_json::json!(logs
.lines()
.last()
.map(|x| ANSI_ESCAPE_RE.replace_all(x, "").to_string())
.unwrap_or_else(String::new));
Ok(to_raw_value(&last_line))
let result_out_path2 = format!("{job_dir}/result2.out");
if tokio::fs::metadata(&result_out_path2).await.is_ok() {
let result = read_file_content(&result_out_path2).await?;
return Ok(to_raw_value(&json!(result)));
}
Ok(to_raw_value(&json!(
"No result.out, result2.out or result.json found"
)))
}
backend/windmill-worker/src/bun_executor.rs
+28 -33
View File
@@ -6,7 +6,7 @@ use regex::Regex;
use serde_json::value::RawValue;
use uuid::Uuid;
use windmill_parser_ts::remove_pinned_imports;
use windmill_queue::CanceledBy;
use windmill_queue::{append_logs, CanceledBy};
#[cfg(feature = "enterprise")]
use crate::common::build_envs_map;
@@ -14,8 +14,7 @@ use crate::common::build_envs_map;
use crate::{
common::{
create_args_and_out_file, get_main_override, get_reserved_variables, handle_child,
parse_npm_config, read_result, set_logs, start_child_process, write_file,
write_file_binary,
parse_npm_config, read_result, start_child_process, write_file, write_file_binary,
},
AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL,
DISABLE_NUSER, HOME_ENV, NODE_PATH, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_PATH, PATH_ENV,
@@ -53,7 +52,6 @@ lazy_static::lazy_static! {
}
pub async fn gen_lockfile(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_id: &Uuid,
@@ -110,7 +108,6 @@ pub async fn gen_lockfile(
handle_child(
job_id,
db,
logs,
mem_peak,
canceled_by,
child_process,
@@ -124,10 +121,12 @@ pub async fn gen_lockfile(
.await?;
if trusted_deps.len() > 0 {
logs.push_str(&format!(
let logs1 = format!(
"\ndetected trustedDependencies: {}\n",
trusted_deps.join(", ")
));
);
append_logs(job_id.clone(), w_id.to_string(), logs1, db).await;
let mut content = "".to_string();
{
let mut file = File::open(format!("{job_dir}/package.json")).await?;
@@ -150,7 +149,6 @@ pub async fn gen_lockfile(
}
install_lockfile(
logs,
mem_peak,
canceled_by,
job_id,
@@ -224,7 +222,6 @@ registry = {}
}
pub async fn install_lockfile(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_id: &Uuid,
@@ -243,9 +240,13 @@ pub async fn install_lockfile(
.args(vec!["install"])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
if npm_mode {
logs.push_str("NPM mode\n")
}
let mut npm_logs = if npm_mode {
"NPM mode\n".to_string()
} else {
"".to_string()
};
let has_file = if npm_mode {
let registry = NPM_CONFIG_REGISTRY.read().await.clone();
if let Some(registry) = registry {
@@ -255,7 +256,7 @@ pub async fn install_lockfile(
let mut splitted = registry.split(":_authToken=");
let custom_registry = splitted.next().unwrap_or_default();
logs.push_str(&format!(
npm_logs.push_str(&format!(
"Using custom npm registry: {custom_registry} {}\n",
if splitted.next().is_some() {
"with authToken"
@@ -274,13 +275,16 @@ pub async fn install_lockfile(
false
};
if npm_mode {
append_logs(job_id.clone(), w_id.to_string(), npm_logs, db).await;
}
let child_process = start_child_process(child_cmd, &*BUN_PATH).await?;
gen_bunfig(job_dir).await?;
handle_child(
job_id,
db,
logs,
mem_peak,
canceled_by,
child_process,
@@ -337,7 +341,6 @@ fn get_annotation(inner_content: &str) -> Annotations {
#[tracing::instrument(level = "trace", skip_all)]
pub async fn handle_bun_job(
requirements_o: Option<String>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
@@ -392,7 +395,6 @@ pub async fn handle_bun_job(
}
install_lockfile(
logs,
mem_peak,
canceled_by,
&job.id,
@@ -410,10 +412,10 @@ pub async fn handle_bun_job(
let trusted_deps = get_trusted_deps(inner_content);
// if !*DISABLE_NSJAIL || !empty_trusted_deps || has_custom_config_registry {
logs.push_str("\n\n--- BUN INSTALL ---\n");
set_logs(&logs, &job.id, &db).await;
let logs1 = "\n\n--- BUN INSTALL ---\n".to_string();
append_logs(job.id, job.workspace_id.clone(), logs1, db).await;
let _ = gen_lockfile(
logs,
mem_peak,
canceled_by,
&job.id,
@@ -436,17 +438,14 @@ pub async fn handle_bun_job(
let main_code = remove_pinned_imports(inner_content)?;
let _ = write_file(job_dir, "main.ts", &main_code).await?;
if annotation.nodejs_mode {
logs.push_str("\n\n--- NODE CODE EXECUTION ---\n");
let init_logs = if annotation.nodejs_mode {
"\n\n--- NODE CODE EXECUTION ---\n".to_string()
} else {
logs.push_str("\n\n--- BUN CODE EXECUTION ---\n");
}
let logs_f = async {
set_logs(&logs, &job.id, &db).await;
Ok(()) as error::Result<()>
"\n\n--- BUN CODE EXECUTION ---\n".to_string()
};
append_logs(job.id.clone(), job.workspace_id.to_string(), init_logs, db).await;
let write_wrapper_f = async {
// let mut start = Instant::now();
let args =
@@ -583,10 +582,9 @@ plugin(p)
}
};
let (reserved_variables, _, _, _) = tokio::try_join!(
let (reserved_variables, _, _) = tokio::try_join!(
reserved_variables_args_out_f,
write_wrapper_f,
logs_f,
write_loader_f
)?;
@@ -604,7 +602,6 @@ plugin(p)
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
child_process,
@@ -730,7 +727,6 @@ plugin(p)
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
child,
@@ -838,7 +834,6 @@ pub async fn start_worker(
}
install_lockfile(
&mut logs,
&mut mem_peak,
&mut canceled_by,
&Uuid::nil(),
@@ -856,7 +851,6 @@ pub async fn start_worker(
let trusted_deps = get_trusted_deps(inner_content);
logs.push_str("\n\n--- BUN INSTALL ---\n");
let _ = gen_lockfile(
&mut logs,
&mut mem_peak,
&mut canceled_by,
&Uuid::nil(),
@@ -977,6 +971,7 @@ plugin(p)
token,
jobs_rx,
worker_name,
db,
)
.await
}
backend/windmill-worker/src/common.rs
+21 -53
View File
@@ -29,13 +29,12 @@ use windmill_common::{
};
use anyhow::Result;
use windmill_queue::CanceledBy;
use windmill_queue::{append_logs, CanceledBy};
#[cfg(any(target_os = "linux", target_os = "macos"))]
use std::os::unix::process::ExitStatusExt;
use std::{
borrow::Borrow,
collections::{hash_map::DefaultHasher, HashMap},
hash::{Hash, Hasher},
io, panic,
@@ -350,21 +349,6 @@ pub async fn read_result(job_dir: &str) -> error::Result<Box<RawValue>> {
return read_file(&format!("{job_dir}/result.json")).await;
}
#[tracing::instrument(level = "trace", skip_all)]
pub async fn set_logs(logs: &str, id: &uuid::Uuid, db: &Pool<Postgres>) {
if sqlx::query!(
"UPDATE queue SET logs = $1 WHERE id = $2",
logs.to_owned(),
id
)
.execute(db)
.await
.is_err()
{
tracing::error!(%id, "error updating logs for id {id}")
};
}
pub fn capitalize(s: &str) -> String {
let mut c = s.chars();
match c.next() {
@@ -614,7 +598,6 @@ pub async fn update_job_poller<F, Fut>(
pub async fn handle_child(
job_id: &Uuid,
db: &Pool<Postgres>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by_ref: &mut Option<CanceledBy>,
mut child: Child,
@@ -631,8 +614,17 @@ pub async fn handle_child(
#[cfg(target_os = "linux")]
if let Some(pid) = pid {
//set the highest oom priority
let mut file = File::create(format!("/proc/{pid}/oom_score_adj")).await?;
let _ = file.write_all(b"1000").await;
if let Some(mut file) = File::create(format!("/proc/{pid}/oom_score_adj"))
.await
.map_err(|e| {
tracing::error!("Could not create oom_score_file to pid {pid}: {e}");
e
})
.ok()
{
let _ = file.write_all(b"1000").await;
let _ = file.sync_all().await;
}
} else {
tracing::info!("could not get child pid");
}
@@ -667,8 +659,7 @@ pub async fn handle_child(
let (timeout_duration, timeout_warn_msg) =
resolve_job_timeout(&db, w_id, job_id, custom_timeout).await;
if let Some(msg) = timeout_warn_msg {
logs.push_str(msg.as_str());
append_logs(job_id, msg.as_str(), db).await;
append_logs(job_id, w_id.to_string(), msg.as_str(), db).await;
}
/* a future that completes when the child process exits */
@@ -745,6 +736,7 @@ pub async fn handle_child(
/* a future that reads output from the child and appends to the database */
let lines = async move {
let max_log_size = if *CLOUD_HOSTED {
MAX_RESULT_SIZE
} else {
@@ -753,12 +745,16 @@ pub async fn handle_child(
/* log_remaining is zero when output limit was reached */
let mut log_remaining = if *CLOUD_HOSTED {
max_log_size.saturating_sub(logs.chars().count())
max_log_size
} else {
usize::MAX
};
let mut result = io::Result::Ok(());
let mut output = output.take_until(rx2.recv()).boxed();
let mut output = output.take_until(async {
let _ = rx2.recv().await;
//wait at most 50ms after end of a script for output stream to end
tokio::time::sleep(Duration::from_millis(50)).await;
}).boxed();
/* `do_write` resolves the task, but does not contain the Result.
* It's useful to know if the task completed. */
let (mut do_write, mut write_result) = tokio::spawn(ready(())).remote_handle();
@@ -795,7 +791,6 @@ pub async fn handle_child(
while let Some(line) = read_lines.next().await {
match line {
Ok(_) if log_remaining == 0 => (),
Ok(line) => {
if line.is_empty() {
continue;
@@ -818,8 +813,6 @@ pub async fn handle_child(
}
}
logs.push_str(&joined);
/* Ensure the last flush completed before starting a new one.
*
@@ -837,7 +830,7 @@ pub async fn handle_child(
panic::resume_unwind(p);
}
(do_write, write_result) = tokio::spawn(append_logs(job_id, joined, db.clone())).remote_handle();
(do_write, write_result) = tokio::spawn(append_logs(job_id, w_id.to_string(), joined, db.clone())).remote_handle();
if let Err(err) = result {
tracing::error!(%job_id, %err, "error reading output for job {job_id}: {err}");
@@ -993,10 +986,8 @@ pub fn lines_to_stream<R: tokio::io::AsyncBufRead + Unpin>(
lazy_static::lazy_static! {
static ref RE_00: Regex = Regex::new('\u{00}'.to_string().as_str()).unwrap();
pub static ref NO_LOGS: bool = std::env::var("NO_LOGS").ok().is_some_and(|x| x == "1" || x == "true");
pub static ref NO_LOGS_AT_ALL: bool = std::env::var("NO_LOGS_AT_ALL").ok().is_some_and(|x| x == "1" || x == "true");
pub static ref SLOW_LOGS: bool = std::env::var("SLOW_LOGS").ok().is_some_and(|x| x == "1" || x == "true");
}
// as a detail, `BufReader::lines()` removes \n and \r\n from the strings it yields,
// so this pushes \n to thd destination string in each call
@@ -1277,29 +1268,6 @@ pub async fn save_in_cache(
}
}
/* TODO retry this? */
#[tracing::instrument(level = "trace", skip_all)]
async fn append_logs(job_id: uuid::Uuid, logs: impl AsRef<str>, db: impl Borrow<Pool<Postgres>>) {
if logs.as_ref().is_empty() {
return;
}
if *NO_LOGS {
tracing::info!("NO LOGS [{job_id}]: {}", logs.as_ref());
return;
}
if let Err(err) = sqlx::query!(
"UPDATE queue SET logs = concat(logs, $1::text) WHERE id = $2",
logs.as_ref(),
job_id,
)
.execute(db.borrow())
.await
{
tracing::error!(%job_id, %err, "error updating logs for job {job_id}: {err}");
}
}
fn tentatively_improve_error(err: Error, executable: &str) -> Error {
if err
.to_string()
backend/windmill-worker/src/dedicated_worker.rs
+7 -4
View File
@@ -8,7 +8,8 @@ use tokio::{
io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
process::Command,
};
use windmill_common::{error, jobs::QueuedJob, variables, worker::to_raw_value};
use windmill_common::{error, jobs::QueuedJob, variables, worker::to_raw_value, DB};
use windmill_queue::append_logs;
use std::{collections::VecDeque, process::Stdio, sync::Arc};
@@ -56,6 +57,7 @@ pub async fn handle_dedicated_process(
token: &str,
mut jobs_rx: Receiver<Arc<QueuedJob>>,
worker_name: &str,
db: &DB,
) -> std::result::Result<(), error::Error> {
//do not cache local dependencies
let mut child = {
@@ -147,15 +149,16 @@ pub async fn handle_dedicated_process(
let job: Arc<QueuedJob> = jobs.pop_front().expect("pop");
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&line.replace("wm_res[success]:", "").replace("wm_res[error]:", "")) {
Ok(result) => {
append_logs(job.id, job.workspace_id.clone(), logs.clone(), db).await;
if line.starts_with("wm_res[success]:") {
job_completed_tx.send(JobCompleted { job , result, logs: logs, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap()
job_completed_tx.send(JobCompleted { job , result, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap()
} else {
job_completed_tx.send(JobCompleted { job , result, logs: logs, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap()
job_completed_tx.send(JobCompleted { job , result, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap()
}
},
Err(e) => {
tracing::error!("Could not deserialize job result `{line}`: {e:?}");
job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), logs: "".to_string(), mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap();
job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap();
},
};
logs = init_log.clone();
backend/windmill-worker/src/deno_executor.rs
+7 -18
View File
@@ -3,12 +3,12 @@ use std::{collections::HashMap, process::Stdio};
use itertools::Itertools;
use serde_json::value::RawValue;
use uuid::Uuid;
use windmill_queue::CanceledBy;
use windmill_queue::{append_logs, CanceledBy};
use crate::{
common::{
create_args_and_out_file, get_main_override, get_reserved_variables, handle_child,
parse_npm_config, read_result, set_logs, start_child_process, write_file,
parse_npm_config, read_result, start_child_process, write_file,
},
AuthedClientBackgroundTask, DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV,
NPM_CONFIG_REGISTRY, PATH_ENV, TZ_ENV,
@@ -85,7 +85,6 @@ async fn get_common_deno_proc_envs(
pub async fn generate_deno_lock(
job_id: &Uuid,
code: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -138,7 +137,6 @@ pub async fn generate_deno_lock(
handle_child(
job_id,
db,
logs,
mem_peak,
canceled_by,
child_process,
@@ -161,7 +159,6 @@ pub async fn generate_deno_lock(
#[tracing::instrument(level = "trace", skip_all)]
pub async fn handle_deno_job(
requirements_o: Option<String>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
@@ -174,16 +171,8 @@ pub async fn handle_deno_job(
envs: HashMap<String, String>,
) -> error::Result<Box<RawValue>> {
// let mut start = Instant::now();
logs.push_str("\n\n--- DENO CODE EXECUTION ---\n");
let logs_to_set = logs.clone();
let id = job.id.clone();
let db2 = db.clone();
let set_logs_f = async {
set_logs(&logs_to_set, &id, &db2).await;
Ok(()) as error::Result<()>
};
let logs1 = "\n\n--- DENO CODE EXECUTION ---\n".to_string();
append_logs(job.id.clone(), job.workspace_id.to_string(), logs1, db).await;
let main_override = get_main_override(job.args.as_ref());
@@ -267,8 +256,7 @@ try {{
Ok(reserved_variables) as error::Result<(HashMap<String, String>, String)>
};
let (_, (reserved_variables, token), _, _, _) = tokio::try_join!(
set_logs_f,
let ((reserved_variables, token), _, _, _) = tokio::try_join!(
reserved_variables_args_out_f,
write_main_f,
write_wrapper_f,
@@ -342,7 +330,6 @@ try {{
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
child,
@@ -424,6 +411,7 @@ pub async fn start_worker(
job_completed_tx: JobCompletedSender,
jobs_rx: Receiver<Arc<QueuedJob>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
db: &sqlx::Pool<sqlx::Postgres>,
) -> Result<()> {
use windmill_common::variables;
@@ -537,6 +525,7 @@ for await (const chunk of Deno.stdin.readable) {{
token,
jobs_rx,
worker_name,
db,
)
.await
}
backend/windmill-worker/src/go_executor.rs
+29 -26
View File
@@ -14,12 +14,12 @@ use windmill_common::{
utils::calculate_hash,
};
use windmill_parser_go::{parse_go_imports, REQUIRE_PARSE};
use windmill_queue::CanceledBy;
use windmill_queue::{append_logs, CanceledBy};
use crate::{
common::{
capitalize, create_args_and_out_file, get_reserved_variables, handle_child, read_result,
set_logs, start_child_process, write_file,
start_child_process, write_file,
},
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, GOPRIVATE, GOPROXY,
GO_BIN_CACHE_DIR, GO_CACHE_DIR, HOME_ENV, NSJAIL_PATH, PATH_ENV, TZ_ENV,
@@ -34,7 +34,6 @@ lazy_static::lazy_static! {
#[tracing::instrument(level = "trace", skip_all)]
pub async fn handle_go_job(
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
@@ -74,13 +73,12 @@ pub async fn handle_go_job(
};
if !bin_exists {
logs.push_str("\n\n--- GO DEPENDENCIES SETUP ---\n");
set_logs(logs, &job.id, db).await;
let logs1 = "\n\n--- GO DEPENDENCIES SETUP ---\n".to_string();
append_logs(job.id.clone(), job.workspace_id.to_string(), logs1, db).await;
install_go_dependencies(
&job.id,
inner_content,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -93,8 +91,9 @@ pub async fn handle_go_job(
)
.await?;
logs.push_str("\n\n--- GO CODE EXECUTION ---\n");
set_logs(logs, &job.id, db).await;
let logs2 = "\n\n--- GO CODE EXECUTION ---\n".to_string();
append_logs(job.id.clone(), job.workspace_id.to_string(), logs2, db).await;
create_args_and_out_file(client, job, job_dir, db).await?;
{
let sig = windmill_parser_go::parse_go_sig(&inner_content)?;
@@ -197,7 +196,6 @@ func Run(req Req) (interface{{}}, error){{
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
build_go_process,
@@ -211,20 +209,28 @@ func Run(req Req) (interface{{}}, error){{
.await?;
create_dir(&bin_path).await?;
tokio::fs::copy(format!("{job_dir}/main"), format!("{bin_path}/main")).await?;
logs.push_str(&format!("write cached binary: {}\n", bin_path));
let target = format!("{bin_path}/main");
tokio::fs::copy(format!("{job_dir}/main"), &target).await?;
append_logs(
job.id.clone(),
job.workspace_id.to_string(),
format!("write cached binary: {}\n", bin_path),
db,
)
.await;
} else {
let path = format!("{bin_path}/main");
logs.push_str(&format!("found cached binary: {path}\n"));
tokio::fs::copy(&path, format!("{job_dir}/main"))
.await
.map_err(|e| {
Error::ExecutionErr(format!(
"could not copy cached binary from {path} to {job_dir}/main: {e:?}"
))
})?;
logs.push_str("\n\n--- GO CODE EXECUTION ---\n");
set_logs(logs, &job.id, db).await;
let mut logs2 = "".to_string();
logs2.push_str(&format!("found cached binary: {path}\n"));
let target = format!("{job_dir}/main");
tokio::fs::symlink(&path, &target).await.map_err(|e| {
Error::ExecutionErr(format!(
"could not copy cached binary from {path} to {job_dir}/main: {e:?}"
))
})?;
logs2.push_str("\n\n--- GO CODE EXECUTION ---\n");
append_logs(job.id.clone(), job.workspace_id.to_string(), logs2, db).await;
create_args_and_out_file(client, job, job_dir, db).await?;
}
@@ -283,7 +289,6 @@ func Run(req Req) (interface{{}}, error){{
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
child,
@@ -322,7 +327,6 @@ use std::io::prelude::*;
pub async fn install_go_dependencies(
job_id: &Uuid,
code: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -346,7 +350,6 @@ pub async fn install_go_dependencies(
handle_child(
job_id,
db,
logs,
mem_peak,
canceled_by,
child_process,
@@ -389,7 +392,8 @@ pub async fn install_go_dependencies(
.fetch_optional(db)
.await?
{
logs.push_str(&format!("\nfound cached resolution: {}", hash));
let logs1 = format!("\nfound cached resolution: {}", hash);
append_logs(job_id.clone(), w_id.to_string(), logs1, db).await;
gen_go_mod(code, job_dir, &cached).await?;
skip_tidy = true;
new_lockfile = false;
@@ -411,7 +415,6 @@ pub async fn install_go_dependencies(
handle_child(
job_id,
db,
logs,
mem_peak,
canceled_by,
child_process,
backend/windmill-worker/src/python_executor.rs
+16 -22
View File
@@ -23,7 +23,7 @@ use windmill_common::{
#[cfg(feature = "enterprise")]
use windmill_common::variables::get_secret_value_as_admin;
use windmill_queue::CanceledBy;
use windmill_queue::{append_logs, CanceledBy};
lazy_static::lazy_static! {
static ref PYTHON_PATH: String =
@@ -56,7 +56,7 @@ use windmill_common::s3_helpers::S3_CACHE_BUCKET;
use crate::{
common::{
create_args_and_out_file, get_main_override, get_reserved_variables, handle_child,
read_result, set_logs, start_child_process, write_file,
read_result, start_child_process, write_file,
},
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, HTTPS_PROXY, HTTP_PROXY,
LOCK_CACHE_DIR, NO_PROXY, NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR, PIP_EXTRA_INDEX_URL,
@@ -95,7 +95,6 @@ pub fn handle_ephemeral_token(x: String) -> String {
pub async fn pip_compile(
job_id: &Uuid,
requirements: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -103,8 +102,8 @@ pub async fn pip_compile(
worker_name: &str,
w_id: &str,
) -> error::Result<String> {
let mut logs = String::new();
logs.push_str(&format!("\nresolving dependencies..."));
set_logs(logs, job_id, db).await;
logs.push_str(&format!("\ncontent of requirements:\n{}\n", requirements));
let requirements = if let Some(pip_local_dependencies) =
WORKER_CONFIG.read().await.pip_local_dependencies.as_ref()
@@ -192,10 +191,10 @@ pub async fn pip_compile(
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let child_process = start_child_process(child_cmd, "pip-compile").await?;
append_logs(job_id.clone(), w_id.to_string(), logs, db).await;
handle_child(
job_id,
db,
logs,
mem_peak,
canceled_by,
child_process,
@@ -233,7 +232,6 @@ pub async fn handle_python_job(
worker_dir: &str,
worker_name: &str,
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -254,14 +252,18 @@ pub async fn handle_python_job(
db,
worker_name,
worker_dir,
logs,
mem_peak,
canceled_by,
)
.await?;
logs.push_str("\n\n--- PYTHON CODE EXECUTION ---\n");
set_logs(logs, &job.id, db).await;
append_logs(
job.id.clone(),
job.workspace_id.to_string(),
"\n\n--- PYTHON CODE EXECUTION ---\n".to_string(),
db,
)
.await;
let (
import_loader,
@@ -432,7 +434,6 @@ mount {{
handle_child(
&job.id,
db,
logs,
mem_peak,
canceled_by,
child,
@@ -627,7 +628,6 @@ async fn handle_python_deps(
db: &DB,
worker_name: &str,
worker_dir: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
) -> error::Result<Vec<String>> {
@@ -661,7 +661,6 @@ async fn handle_python_deps(
pip_compile(
job_id,
&requirements,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -685,7 +684,6 @@ async fn handle_python_deps(
.collect(),
job_id,
w_id,
logs,
mem_peak,
canceled_by,
db,
@@ -707,7 +705,6 @@ pub async fn handle_python_reqs(
requirements: Vec<&str>,
job_id: &Uuid,
w_id: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -793,8 +790,10 @@ pub async fn handle_python_reqs(
}
}
logs.push_str("\n--- PIP INSTALL ---\n");
logs.push_str(&format!("\n{req} is being installed for the first time.\n It will be cached for all ulterior uses."));
let mut logs1 = String::new();
logs1.push_str("\n\n--- PIP INSTALL ---\n");
logs1.push_str(&format!("\n{req} is being installed for the first time.\n It will be cached for all ulterior uses."));
append_logs(job_id.clone(), w_id.to_string(), logs1, db).await;
tracing::info!(
worker_name = %worker_name,
@@ -899,7 +898,6 @@ pub async fn handle_python_reqs(
let child = handle_child(
&job_id,
db,
logs,
mem_peak,
canceled_by,
child,
@@ -963,7 +961,6 @@ pub async fn start_worker(
jobs_rx: Receiver<Arc<QueuedJob>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> error::Result<()> {
let mut logs = "".to_string();
let mut mem_peak: i32 = 0;
let mut canceled_by: Option<CanceledBy> = None;
let context = variables::get_reserved_variables(
@@ -995,15 +992,11 @@ pub async fn start_worker(
db,
worker_name,
job_dir,
&mut logs,
&mut mem_peak,
&mut canceled_by,
)
.await?;
logs.push_str("\n\n--- PYTHON CODE EXECUTION ---\n");
set_logs(&mut logs, &Uuid::nil(), db).await;
let _args = None;
let (
import_loader,
@@ -1123,6 +1116,7 @@ for line in sys.stdin:
token,
jobs_rx,
worker_name,
db,
)
.await
}
backend/windmill-worker/src/worker.rs
+66 -92
View File
@@ -8,7 +8,6 @@
use anyhow::Result;
use const_format::concatcp;
use itertools::Itertools;
#[cfg(feature = "prometheus")]
use prometheus::{
core::{AtomicI64, GenericGauge},
@@ -46,14 +45,15 @@ use windmill_common::{
users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SYNC_EMAIL},
utils::{rd_string, StripPath},
worker::{
to_raw_value, to_raw_value_owned, update_ping, CLOUD_HOSTED, WORKER_CONFIG, WORKER_GROUP,
to_raw_value, to_raw_value_owned, update_ping, CLOUD_HOSTED, NO_LOGS, WORKER_CONFIG,
WORKER_GROUP,
},
DB, IS_READY,
};
use windmill_queue::{
canceled_job_to_result, empty_result, get_queued_job, pull, push, CanceledBy, PushArgs,
PushIsolationLevel, WrappedError, HTTP_CLIENT,
append_logs, canceled_job_to_result, empty_result, get_queued_job, pull, push, CanceledBy,
PushArgs, PushIsolationLevel, WrappedError, HTTP_CLIENT,
};
#[cfg(feature = "prometheus")]
@@ -96,7 +96,7 @@ use crate::{
bun_executor::{gen_lockfile, get_trusted_deps, handle_bun_job},
common::{
build_args_map, get_cached_resource_value_if_valid, hash_args, read_result, save_in_cache,
write_file, NO_LOGS, NO_LOGS_AT_ALL, SLOW_LOGS,
write_file, NO_LOGS_AT_ALL, SLOW_LOGS,
},
deno_executor::{generate_deno_lock, handle_deno_job},
go_executor::{handle_go_job, install_go_dependencies},
@@ -1586,7 +1586,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
job: Arc::new(job),
success: true,
result: empty_result(),
logs: String::new(),
mem_peak: 0,
cached_res_path: None,
token: "".to_string(),
@@ -2156,6 +2155,7 @@ async fn spawn_dedicated_worker(
job_completed_tx,
dedicated_worker_rx,
killpill_rx,
&db,
)
.await
}
@@ -2237,16 +2237,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
// ) -> error::Result<()> {
pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
JobCompleted {
job,
result,
logs,
mem_peak,
success,
cached_res_path,
canceled_by,
..
}: JobCompleted,
JobCompleted { job, result, mem_peak, success, cached_res_path, canceled_by, .. }: JobCompleted,
client: &AuthedClient,
db: &DB,
worker_dir: &str,
@@ -2273,7 +2264,6 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
true,
false,
Json(&result),
logs,
mem_peak.to_owned(),
canceled_by,
rsmq.clone(),
@@ -2315,7 +2305,6 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
let result = add_completed_job_error(
db,
&job,
logs.to_string(),
mem_peak.to_owned(),
canceled_by,
serde_json::from_str(result.get()).unwrap_or_else(
@@ -2406,11 +2395,17 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
};
let rsmq_2 = rsmq.clone();
let update_job_future = || {
let update_job_future = || async {
append_logs(
job.id,
job.workspace_id.clone(),
format!("Unexpected error during job execution:\n{err:#?}"),
db,
)
.await;
add_completed_job_error(
db,
job,
format!("Unexpected error during job execution:\n{err:#?}"),
mem_peak,
canceled_by.clone(),
err.clone(),
@@ -2418,6 +2413,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
worker_name,
false,
)
.await
};
let update_job_future = if job.is_flow_step || job.is_flow() {
@@ -2458,10 +2454,16 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
get_queued_job(&parent_job_id, &job.workspace_id, &db).await
{
let e = json!({"message": err.to_string(), "name": "InternalErr"});
append_logs(
parent_job.id,
job.workspace_id.clone(),
format!("Unexpected error during flow job error handling:\n{err}"),
db,
)
.await;
let _ = add_completed_job_error(
db,
&parent_job,
format!("Unexpected error during flow job error handling:\n{err}"),
mem_peak,
canceled_by.clone(),
e,
@@ -2522,7 +2524,6 @@ pub enum SendResult {
pub struct JobCompleted {
pub job: Arc<QueuedJob>,
pub result: Box<RawValue>,
pub logs: String,
pub mem_peak: i32,
pub success: bool,
pub cached_res_path: Option<String>,
@@ -2532,7 +2533,6 @@ pub struct JobCompleted {
async fn do_nativets(
job: &QueuedJob,
logs: String,
client: &AuthedClientBackgroundTask,
code: String,
db: &Pool<Postgres>,
@@ -2560,7 +2560,7 @@ async fn do_nativets(
&job.workspace_id,
)
.await?;
Ok((result.0, [logs, result.1].join("\n\n")))
Ok((result.0, result.1))
}
#[derive(Deserialize, Serialize, Default)]
@@ -2678,12 +2678,15 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
)
.await;
if let Some(cached_resource_value) = cached_resource_value_maybe {
let logs = "Job skipped because args & path found in cache and not expired".to_string();
{
let logs =
"Job skipped because args & path found in cache and not expired".to_string();
append_logs(job.id, job.workspace_id.clone(), logs, db).await;
}
job_completed_tx
.send(JobCompleted {
job: job,
result: cached_resource_value,
logs,
mem_peak: 0,
canceled_by: None,
success: true,
@@ -2717,10 +2720,6 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
let mut mem_peak: i32 = 0;
let mut canceled_by: Option<CanceledBy> = None;
// println!("handle queue {:?}", SystemTime::now());
if let Some(log_str) = &job.logs {
logs.push_str(&log_str);
logs.push_str("\n");
}
logs.push_str(&format!(
"job {} on worker {} (tag: {})\n",
@@ -2753,12 +2752,12 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
"handling job {}",
job.id
);
append_logs(job.id, job.workspace_id.clone(), logs, db).await;
let result = match job.job_kind {
JobKind::Dependencies => {
handle_dependency_job(
&job,
&mut logs,
&mut mem_peak,
&mut canceled_by,
job_dir,
@@ -2773,7 +2772,6 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
}
JobKind::FlowDependencies => handle_flow_dependency_job(
&job,
&mut logs,
&mut mem_peak,
&mut canceled_by,
job_dir,
@@ -2788,7 +2786,6 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
.map(|()| serde_json::from_str("{}").unwrap()),
JobKind::AppDependencies => handle_app_dependency_job(
&job,
&mut logs,
&mut mem_peak,
&mut canceled_by,
job_dir,
@@ -2817,7 +2814,6 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
client,
job_dir,
worker_dir,
&mut logs,
&mut mem_peak,
&mut canceled_by,
base_internal_url,
@@ -2839,11 +2835,11 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
result,
job_dir,
job_completed_tx,
logs,
mem_peak,
canceled_by,
cached_res_path,
client.get_token().await,
db,
)
.await?;
};
@@ -2855,11 +2851,11 @@ async fn process_result(
result: error::Result<Box<RawValue>>,
job_dir: &str,
job_completed_tx: JobCompletedSender,
logs: String,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
cached_res_path: Option<String>,
token: String,
db: &DB,
) -> error::Result<()> {
match result {
Ok(r) => {
@@ -2867,7 +2863,6 @@ async fn process_result(
.send(JobCompleted {
job: job,
result: r,
logs,
mem_peak,
canceled_by,
success: true,
@@ -2885,17 +2880,16 @@ async fn process_result(
if res.as_ref().is_some_and(|x| !x.get().is_empty()) {
res.unwrap()
} else {
let last_10_log_lines = logs
.lines()
.skip(logs.lines().count().max(13) - 13)
.join("\n")
.to_string()
.replace("\n\n", "\n");
let last_10_log_lines = sqlx::query_scalar!(
"SELECT right(logs, 300) FROM job_logs WHERE job_id = $1 AND workspace_id = $2 ORDER BY created_at DESC LIMIT 1",
&job.id,
&job.workspace_id
).fetch_one(db).await.ok().flatten().unwrap_or("".to_string());
let log_lines = last_10_log_lines
.split("CODE EXECUTION ---")
.last()
.unwrap_or(&logs);
.unwrap_or(&last_10_log_lines);
extract_error_value(log_lines, i)
}
@@ -2910,7 +2904,6 @@ async fn process_result(
.send(JobCompleted {
job: job,
result: to_raw_value(&error_value),
logs: logs,
mem_peak,
canceled_by,
success: false,
@@ -3047,7 +3040,6 @@ async fn handle_code_execution_job(
client: &AuthedClientBackgroundTask,
job_dir: &str,
worker_dir: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
base_internal_url: &str,
@@ -3180,24 +3172,21 @@ async fn handle_code_execution_job(
)
.await;
} else if language == Some(ScriptLang::Nativets) {
logs.push_str("\n--- FETCH TS EXECUTION ---\n");
append_logs(
job.id,
job.workspace_id.clone(),
"\n--- FETCH TS EXECUTION ---\n",
db,
)
.await;
let code = format!(
"const BASE_URL = '{base_internal_url}';\nconst WM_TOKEN = '{}';\n{}",
&client.get_token().await,
inner_content
);
let (result, ts_logs) = do_nativets(
job,
logs.clone(),
&client,
code,
db,
mem_peak,
canceled_by,
worker_name,
)
.await?;
*logs = ts_logs;
let (result, ts_logs) =
do_nativets(job, &client, code, db, mem_peak, canceled_by, worker_name).await?;
append_logs(job.id, job.workspace_id.clone(), ts_logs, db).await;
return Ok(result);
}
@@ -3248,7 +3237,6 @@ mount {{
worker_dir,
worker_name,
job,
logs,
mem_peak,
canceled_by,
db,
@@ -3263,7 +3251,6 @@ mount {{
Some(ScriptLang::Deno) => {
handle_deno_job(
requirements_o,
logs,
mem_peak,
canceled_by,
job,
@@ -3280,7 +3267,6 @@ mount {{
Some(ScriptLang::Bun) => {
handle_bun_job(
requirements_o,
logs,
mem_peak,
canceled_by,
job,
@@ -3297,7 +3283,6 @@ mount {{
}
Some(ScriptLang::Go) => {
handle_go_job(
logs,
mem_peak,
canceled_by,
job,
@@ -3315,7 +3300,6 @@ mount {{
}
Some(ScriptLang::Bash) => {
handle_bash_job(
logs,
mem_peak,
canceled_by,
job,
@@ -3332,7 +3316,6 @@ mount {{
}
Some(ScriptLang::Powershell) => {
handle_powershell_job(
logs,
mem_peak,
canceled_by,
job,
@@ -3366,7 +3349,6 @@ mount {{
#[tracing::instrument(level = "trace", skip_all)]
async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -3398,6 +3380,7 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
.is_some_and(|y| y.to_string().as_str() == "true")
})
.unwrap_or(false);
let content = capture_dependency_job(
&job.id,
job.language.as_ref().map(|v| Ok(v)).unwrap_or_else(|| {
@@ -3406,7 +3389,6 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
))
})?,
&raw_code,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3420,6 +3402,7 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
raw_deps,
)
.await;
match content {
Ok(content) => {
if job.script_hash.is_none() {
@@ -3460,6 +3443,7 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
if &job.language == &Some(ScriptLang::Python3) {
if let Ok(relative_imports) = parse_relative_imports(&raw_code, script_path) {
let mut logs = "".to_string();
logs.push_str("\n--- RELATIVE IMPORTS ---\n\n");
logs.push_str(&relative_imports.join("\n"));
if !relative_imports.is_empty() {
@@ -3487,6 +3471,7 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
}
tx.commit().await?;
}
append_logs(job.id, job.workspace_id.clone(), logs, db).await;
if let Err(e) = trigger_python_dependents_to_recompute_dependencies(
w_id,
script_path,
@@ -3509,9 +3494,18 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
))
}
Err(error) => {
let logs2 = sqlx::query_scalar!(
"SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2",
&job.id,
&job.workspace_id
)
.fetch_optional(db)
.await?
.flatten()
.unwrap_or_else(|| "no logs".to_string());
sqlx::query!(
"UPDATE script SET lock_error_logs = $1 WHERE hash = $2 AND workspace_id = $3",
&format!("{logs}\n{error}"),
&format!("{logs2}\n{error}"),
&job.script_hash.unwrap_or(ScriptHash(0)).0,
&job.workspace_id
)
@@ -3605,7 +3599,6 @@ async fn trigger_python_dependents_to_recompute_dependencies<
async fn handle_flow_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -3631,7 +3624,6 @@ async fn handle_flow_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync
flow.modules = lock_modules(
flow.modules,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3717,7 +3709,6 @@ fn get_deployment_msg_and_parent_path_from_args(
async fn lock_modules(
modules: Vec<FlowModule>,
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -3754,7 +3745,6 @@ async fn lock_modules(
modules: lock_modules(
modules,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3777,7 +3767,6 @@ async fn lock_modules(
b.modules = lock_modules(
b.modules,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3799,7 +3788,6 @@ async fn lock_modules(
b.modules = lock_modules(
b.modules,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3816,7 +3804,6 @@ async fn lock_modules(
let default = lock_modules(
default,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3843,7 +3830,6 @@ async fn lock_modules(
&job.id,
&language,
&content,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3878,7 +3864,6 @@ async fn lock_modules(
path = path,
language = ?language,
error = ?error,
logs = ?logs,
"Failed to generate flow lock for raw script"
);
e.value = FlowModuleValue::RawScript {
@@ -3903,7 +3888,6 @@ async fn lock_modules(
async fn lock_modules_app(
value: Value,
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -3929,12 +3913,12 @@ async fn lock_modules_app(
.as_str()
.unwrap_or_default()
.to_string();
let mut logs = "".to_string();
logs.push_str("Found lockable inline script. Generating lock...\n");
let new_lock = capture_dependency_job(
&job.id,
&language,
&content,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3950,6 +3934,7 @@ async fn lock_modules_app(
.await;
match new_lock {
Ok(new_lock) => {
append_logs(job.id, job.workspace_id.clone(), logs, db).await;
v.insert(
"lock".to_string(),
serde_json::Value::String(new_lock),
@@ -3976,7 +3961,6 @@ async fn lock_modules_app(
lock_modules_app(
b,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3999,7 +3983,6 @@ async fn lock_modules_app(
lock_modules_app(
b,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -4021,7 +4004,6 @@ async fn lock_modules_app(
async fn handle_app_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -4051,7 +4033,6 @@ async fn handle_app_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync +
let value = lock_modules_app(
value,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -4129,7 +4110,6 @@ async fn capture_dependency_job(
job_id: &Uuid,
job_language: &ScriptLang,
job_raw_code: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -4163,7 +4143,6 @@ async fn capture_dependency_job(
let req: std::result::Result<String, Error> = pip_compile(
job_id,
&reqs,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -4178,7 +4157,6 @@ async fn capture_dependency_job(
req.split("\n").filter(|x| !x.starts_with("--")).collect(),
job_id,
w_id,
logs,
mem_peak,
canceled_by,
db,
@@ -4190,9 +4168,8 @@ async fn capture_dependency_job(
if let Err(e) = r {
tracing::error!(
"Failed to install python dependencies to prefill the cache: {:?} \n{}",
e,
logs
"Failed to install python dependencies to prefill the cache: {:?} \n",
e
);
}
}
@@ -4207,7 +4184,6 @@ async fn capture_dependency_job(
install_go_dependencies(
job_id,
job_raw_code,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -4229,7 +4205,6 @@ async fn capture_dependency_job(
generate_deno_lock(
job_id,
job_raw_code,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -4249,7 +4224,6 @@ async fn capture_dependency_job(
vec![]
};
let req = gen_lockfile(
logs,
mem_peak,
canceled_by,
job_id,
backend/windmill-worker/src/worker_flow.rs
+24 -17
View File
@@ -43,8 +43,8 @@ use windmill_common::{
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
};
use windmill_queue::{
add_completed_job, add_completed_job_error, get_queued_job, handle_maybe_scheduled_job,
CanceledBy, PushIsolationLevel, WrappedError,
add_completed_job, add_completed_job_error, append_logs, get_queued_job,
handle_maybe_scheduled_job, CanceledBy, PushIsolationLevel, WrappedError,
};
type DB = sqlx::Pool<sqlx::Postgres>;
@@ -614,16 +614,18 @@ pub async fn update_flow_status_after_job_completion_internal<
};
let done = if !should_continue_flow {
let logs = if flow_job.canceled {
"Flow job canceled\n".to_string()
} else if stop_early {
format!("Flow job stopped early because of a stop early predicate returning true\n")
} else if success {
"Flow job completed with success\n".to_string()
} else {
"Flow job completed with error\n".to_string()
};
{
let logs = if flow_job.canceled {
"Flow job canceled\n".to_string()
} else if stop_early {
format!("Flow job stopped early because of a stop early predicate returning true\n")
} else if success {
"Flow job completed with success\n".to_string()
} else {
"Flow job completed with error\n".to_string()
};
append_logs(flow_job.id, w_id.to_string(), logs, db).await;
}
#[cfg(feature = "enterprise")]
if flow_job.parent_job.is_none() {
// run the cleanup step only when the root job is complete
@@ -646,7 +648,6 @@ pub async fn update_flow_status_after_job_completion_internal<
add_completed_job_error(
db,
&flow_job,
logs,
0,
Some(CanceledBy {
username: flow_job.canceled_by.clone(),
@@ -687,7 +688,6 @@ pub async fn update_flow_status_after_job_completion_internal<
success,
stop_early && skip_if_stop_early,
Json(&nresult),
logs,
0,
None,
rsmq.clone(),
@@ -705,7 +705,6 @@ pub async fn update_flow_status_after_job_completion_internal<
|e| json!({"error": format!("Impossible to serialize error: {e}")}),
),
),
logs,
0,
None,
rsmq.clone(),
@@ -731,10 +730,16 @@ pub async fn update_flow_status_after_job_completion_internal<
{
Err(err) => {
let e = json!({"message": err.to_string(), "name": "InternalError"});
append_logs(
flow_job.id,
w_id.to_string(),
format!("Unexpected error during flow chaining:\n{:#?}", e),
db,
)
.await;
let _ = add_completed_job_error(
db,
&flow_job,
"Unexpected error during flow chaining:\n".to_string(),
0,
None,
e,
@@ -1511,7 +1516,10 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
let success = false;
let skipped = false;
let logs = "Timed out waiting to be resumed".to_string();
append_logs(flow_job.id, flow_job.workspace_id.clone(), logs.clone(), db).await;
let result = json!({ "error": {"message": logs, "name": "SuspendedTimeout"}});
let canceled_by = if flow_job.canceled {
Some(CanceledBy {
@@ -1527,7 +1535,6 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
success,
skipped,
Json(&result),
logs,
0,
canceled_by,
rsmq.clone(),