diff --git a/backend/src/main.rs b/backend/src/main.rs index 4f014294b4..c93b48af4b 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -244,7 +244,7 @@ Windmill Community Edition {GIT_VERSION} let metrics_f = async { match metrics_addr { - Some(addr) => { + Some(_addr) => { #[cfg(not(feature = "enterprise"))] panic!("Metrics are only available in the Enterprise Edition"); diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs new file mode 100644 index 0000000000..59aab1334d --- /dev/null +++ b/backend/windmill-worker/src/bash_executor.rs @@ -0,0 +1,244 @@ +use std::{collections::HashMap, process::Stdio}; + +use itertools::Itertools; +use regex::Regex; +use serde_json::{json, Value}; +use tokio::process::Command; +use windmill_common::{error::Error, jobs::QueuedJob}; + +const NSJAIL_CONFIG_RUN_BASH_CONTENT: &str = include_str!("../nsjail/run.bash.config.proto"); + +use crate::{ + common::{get_reserved_variables, handle_child, set_logs}, + transform_json_value, write_file, AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, + HOME_ENV, NSJAIL_PATH, PATH_ENV, +}; + +lazy_static::lazy_static! { + + pub static ref ANSI_ESCAPE_RE: Regex = Regex::new(r"\x1b\[[0-9;]*m").unwrap(); +} + +#[tracing::instrument(level = "trace", skip_all)] +pub async fn handle_bash_job( + logs: &mut String, + job: &QueuedJob, + db: &sqlx::Pool, + client: &AuthedClientBackgroundTask, + content: &str, + job_dir: &str, + shared_mount: &str, + base_internal_url: &str, + worker_name: &str, + envs: HashMap, +) -> Result { + logs.push_str("\n\n--- BASH CODE EXECUTION ---\n"); + set_logs(logs, &job.id, db).await; + write_file( + job_dir, + "main.sh", + &format!("set -e\n{content}\necho \"\"\nsleep 0.02"), + ) + .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 client = client.get_authed().await; + let hm = match transform_json_value( + "args", + &client, + &job.workspace_id, + job.args.clone().unwrap_or_else(|| json!({})), + ) + .await? + { + Value::Object(ref hm) => hm.clone(), + _ => serde_json::Map::new(), + }; + let args_owned = windmill_parser_bash::parse_bash_sig(&content)? + .args + .iter() + .map(|arg| { + hm.get(&arg.name) + .and_then(|v| match v { + Value::String(s) => Some(s.clone()), + _ => serde_json::to_string(v).ok(), + }) + .unwrap_or_else(String::new) + }) + .collect::>(); + let args = args_owned.iter().map(|s| &s[..]).collect::>(); + + let child = if !*DISABLE_NSJAIL { + let _ = write_file( + job_dir, + "run.config.proto", + &NSJAIL_CONFIG_RUN_BASH_CONTENT + .replace("{JOB_DIR}", job_dir) + .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) + .replace("{SHARED_MOUNT}", shared_mount), + ) + .await?; + let mut cmd_args = vec!["--config", "run.config.proto", "--", "/bin/bash", "main.sh"]; + cmd_args.extend(args); + Command::new(NSJAIL_PATH.as_str()) + .current_dir(job_dir) + .env_clear() + .envs(reserved_variables) + .env("PATH", PATH_ENV.as_str()) + .env("BASE_INTERNAL_URL", base_internal_url) + .args(cmd_args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()? + } else { + let mut cmd_args = vec!["main.sh"]; + cmd_args.extend(&args); + Command::new("/bin/bash") + .current_dir(job_dir) + .env_clear() + .envs(envs) + .envs(reserved_variables) + .env("PATH", PATH_ENV.as_str()) + .env("BASE_INTERNAL_URL", base_internal_url) + .env("HOME", HOME_ENV.as_str()) + .args(cmd_args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()? + }; + handle_child( + &job.id, + db, + logs, + child, + !*DISABLE_NSJAIL, + worker_name, + &job.workspace_id, + "bash run", + job.timeout, + ) + .await?; + //for now bash jobs have an empty result object + Ok(serde_json::json!(logs + .lines() + .last() + .map(|x| ANSI_ESCAPE_RE.replace_all(x, "").to_string()) + .unwrap_or_else(String::new))) +} + +#[tracing::instrument(level = "trace", skip_all)] +pub async fn handle_powershell_job( + logs: &mut String, + job: &QueuedJob, + db: &sqlx::Pool, + client: &AuthedClientBackgroundTask, + content: &str, + job_dir: &str, + shared_mount: &str, + base_internal_url: &str, + worker_name: &str, + envs: HashMap, +) -> Result { + logs.push_str("\n\n--- POWERSHELL CODE EXECUTION ---\n"); + set_logs(logs, &job.id, db).await; + let pwsh_args = { + let client = client.get_authed().await; + let hm = match transform_json_value( + "args", + &client, + &job.workspace_id, + job.args.clone().unwrap_or_else(|| json!({})), + ) + .await? + { + Value::Object(ref hm) => hm.clone(), + _ => serde_json::Map::new(), + }; + + let args_owned = windmill_parser_bash::parse_powershell_sig(&content)? + .args + .iter() + .map(|arg| { + ( + arg.name.clone(), + hm.get(&arg.name) + .and_then(|v| match v { + Value::String(s) => Some(s.clone()), + _ => serde_json::to_string(v).ok(), + }) + .unwrap_or_else(String::new), + ) + }) + .collect::>(); + args_owned + .iter() + .map(|(n, v)| format!("--{n} {v}")) + .join(" ") + }; + + let content = content + .replace('$', r"\$") // escape powershell variables + .replace("`", r"\`"); // escape powershell backticks + + write_file(job_dir, "main.sh", &format!("set -e\ncat > script.ps1 << EOF\n{content}\nEOF\npwsh -File script.ps1 {pwsh_args}\necho \"\"\nsleep 0.02")).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 child = if !*DISABLE_NSJAIL { + let _ = write_file( + job_dir, + "run.config.proto", + &NSJAIL_CONFIG_RUN_BASH_CONTENT + .replace("{JOB_DIR}", job_dir) + .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) + .replace("{SHARED_MOUNT}", shared_mount), + ) + .await?; + let cmd_args = vec!["--config", "run.config.proto", "--", "/bin/bash", "main.sh"]; + Command::new(NSJAIL_PATH.as_str()) + .current_dir(job_dir) + .env_clear() + .envs(reserved_variables) + .env("PATH", PATH_ENV.as_str()) + .env("BASE_INTERNAL_URL", base_internal_url) + .args(cmd_args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()? + } else { + let cmd_args = vec!["main.sh"]; + Command::new("/bin/bash") + .current_dir(job_dir) + .env_clear() + .envs(envs) + .envs(reserved_variables) + .env("PATH", PATH_ENV.as_str()) + .env("BASE_INTERNAL_URL", base_internal_url) + .env("HOME", HOME_ENV.as_str()) + .args(cmd_args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()? + }; + handle_child( + &job.id, + db, + logs, + child, + !*DISABLE_NSJAIL, + worker_name, + &job.workspace_id, + "bash/powershell run", + job.timeout, + ) + .await?; + //for now bash jobs have an empty result object + Ok(serde_json::json!(logs + .lines() + .last() + .map(|x| ANSI_ESCAPE_RE.replace_all(x, "").to_string()) + .unwrap_or_else(String::new))) +} diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 76c379e2e6..b99f578621 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -5,10 +5,10 @@ use itertools::Itertools; use uuid::Uuid; use crate::{ - common::{read_result, set_logs}, - create_args_and_out_file, get_reserved_variables, handle_child, write_file, write_file_binary, - AuthedClientBackgroundTask, BUN_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL, DISABLE_NUSER, - NPM_CONFIG_REGISTRY, NSJAIL_CONFIG_RUN_BUN_CONTENT, NSJAIL_PATH, PATH_ENV, + common::{get_reserved_variables, handle_child, read_result, set_logs}, + create_args_and_out_file, write_file, write_file_binary, AuthedClientBackgroundTask, + BUN_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL, DISABLE_NUSER, NPM_CONFIG_REGISTRY, NSJAIL_PATH, + PATH_ENV, }; use tokio::{fs::File, io::AsyncReadExt, process::Command}; use windmill_common::error::Result; @@ -22,6 +22,8 @@ const RELATIVE_BUN_LOADER: &str = include_str!("../loader.bun.ts"); const RELATIVE_BUN_BUILDER: &str = include_str!("../loader_builder.bun.ts"); +const NSJAIL_CONFIG_RUN_BUN_CONTENT: &str = include_str!("../nsjail/run.bun.config.proto"); + const BUN_LOCKB_SPLIT: &str = "\n//bun.lockb\n"; const EMPTY_FILE: &str = ""; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index ae7ea903c0..4befd36950 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1,9 +1,34 @@ use sqlx::{Pool, Postgres}; use tokio::{fs::File, io::AsyncReadExt}; -use windmill_common::error::{self}; +use windmill_common::{ + error::{self, Error}, + jobs::QueuedJob, +}; use windmill_queue::CLOUD_HOSTED; -use crate::MAX_RESULT_SIZE; +use anyhow::Result; +use std::{ + borrow::Borrow, collections::HashMap, io, os::unix::process::ExitStatusExt, panic, + time::Duration, +}; + +use tracing::{trace_span, Instrument}; +use uuid::Uuid; +use windmill_common::variables; + +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + process::Child, + sync::{broadcast, watch}, + time::{interval, sleep, Instant, MissedTickBehavior}, +}; + +use futures::{ + future::{self, ready, FutureExt}, + stream, StreamExt, +}; + +use crate::{MAX_RESULT_SIZE, TIMEOUT_DURATION, WHITELIST_ENVS}; pub async fn read_result(job_dir: &str) -> error::Result { let mut file = File::open(format!("{job_dir}/result.json")).await?; @@ -39,3 +64,435 @@ pub fn capitalize(s: &str) -> String { Some(f) => f.to_uppercase().collect::() + c.as_str(), } } + +#[tracing::instrument(level = "trace", skip_all)] +pub async fn get_reserved_variables( + job: &QueuedJob, + token: &str, + db: &sqlx::Pool, +) -> Result, Error> { + let flow_path = if let Some(uuid) = job.parent_job { + sqlx::query_scalar!("SELECT script_path FROM queue WHERE id = $1", uuid) + .fetch_optional(db) + .await? + .flatten() + } else { + None + }; + + let variables = variables::get_reserved_variables( + &job.workspace_id, + token, + &job.email, + &job.created_by, + &job.id.to_string(), + &job.permissioned_as, + job.script_path.clone(), + job.parent_job.map(|x| x.to_string()), + flow_path, + job.schedule_path.clone(), + job.flow_step_id.clone(), + ) + .to_vec(); + + let mut r: HashMap = variables + .into_iter() + .map(|rv| (rv.name, rv.value)) + .collect(); + + if let Some(ref envs) = *WHITELIST_ENVS { + for e in envs { + r.insert(e.0.clone(), e.1.clone()); + } + } + + Ok(r) +} + +async fn get_mem_peak(pid: Option, nsjail: bool) -> i32 { + if pid.is_none() { + return -1; + } + let pid = if nsjail { + // This is a bit hacky, but the process id of the nsjail process is the pid of nsjail + 1. + // Ideally, we would get the number from fork() itself. This works in MOST cases. + pid.unwrap() + 1 + } else { + pid.unwrap() + }; + + if let Ok(file) = File::open(format!("/proc/{}/status", pid)).await { + let mut lines = BufReader::new(file).lines(); + while let Some(line) = lines.next_line().await.unwrap_or(None) { + if line.starts_with("VmHWM:") { + return line + .split_whitespace() + .nth(1) + .and_then(|s| s.parse::().ok()) + .unwrap_or(-1); + }; + } + -2 + } else { + -3 + } +} +/// - wait until child exits and return with exit status +/// - read lines from stdout and stderr and append them to the "queue"."logs" +/// quitting early if output exceedes MAX_LOG_SIZE characters (not bytes) +/// - update the `last_line` and `logs` strings with the program output +/// - update "queue"."last_ping" every five seconds +/// - kill process if we exceed timeout or "queue"."canceled" is set +#[tracing::instrument(level = "trace", skip_all)] +pub async fn handle_child( + job_id: &Uuid, + db: &Pool, + logs: &mut String, + mut child: Child, + nsjail: bool, + worker_name: &str, + _w_id: &str, + child_name: &str, + custom_timeout: Option, +) -> error::Result<()> { + let start = Instant::now(); + let update_job_interval = Duration::from_millis(500); + let write_logs_delay = Duration::from_millis(500); + + let pid = child.id(); + #[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; + } else { + tracing::info!("could not get child pid"); + } + let (set_too_many_logs, mut too_many_logs) = watch::channel::(false); + let (tx, mut rx) = broadcast::channel::<()>(3); + let mut rx2 = tx.subscribe(); + + let output = child_joined_output_stream(&mut child); + + let job_id = job_id.clone(); + + /* the cancellation future is polled on by `wait_on_child` while + * waiting for the child to exit normally */ + let update_job = async { + let db = db.clone(); + + let mut interval = interval(update_job_interval); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + + let mut i = 1; + loop { + tokio::select!( + _ = rx.recv() => break, + _ = interval.tick() => { + // update the last_ping column every 5 seconds + i+=1; + if i % 10 == 0 { + sqlx::query!( + "UPDATE worker_ping SET ping_at = now() WHERE worker = $1", + &worker_name + ) + .execute(&db) + .await + .expect("update worker ping"); + } + let mem_peak = get_mem_peak(pid, nsjail).await; + tracing::info!("{job_id} still running. mem peak: {}kB", mem_peak); + let mem_peak = if mem_peak > 0 { Some(mem_peak) } else { None }; + if sqlx::query_scalar!("UPDATE queue SET mem_peak = GREATEST($1, mem_peak), last_ping = now() WHERE id = $2 RETURNING canceled", mem_peak, job_id) + .fetch_optional(&db) + .await + .map(|v| Some(true) == v) + .unwrap_or_else(|err| { + tracing::error!(%job_id, %err, "error checking cancelation for job {job_id}: {err}"); + false + }) + { + break; + } + }, + ); + } + }; + + #[derive(PartialEq, Debug)] + enum KillReason { + TooManyLogs, + Timeout, + Cancelled, + } + /* a future that completes when the child process exits */ + let wait_on_child = async { + let db = db.clone(); + + #[cfg(not(feature = "enterprise"))] + let instance_timeout_duration = *TIMEOUT_DURATION; + + #[cfg(feature = "enterprise")] + let premium_workspace = *CLOUD_HOSTED + && sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", _w_id) + .fetch_one(&db) + .await + .map_err(|e| { + tracing::error!(%e, "error getting premium workspace for job {job_id}: {e}"); + }) + .unwrap_or(false); + + #[cfg(feature = "enterprise")] + let instance_timeout_duration = if premium_workspace { + *TIMEOUT_DURATION * 6 //30mins + } else { + *TIMEOUT_DURATION + }; + + let timeout_duration = if let Some(custom_timeout) = custom_timeout { + Duration::min( + instance_timeout_duration, + Duration::from_secs(custom_timeout as u64), + ) + } else { + instance_timeout_duration + }; + + let kill_reason = tokio::select! { + biased; + result = child.wait() => return result.map(Ok), + Ok(()) = too_many_logs.changed() => KillReason::TooManyLogs, + _ = sleep(timeout_duration) => KillReason::Timeout, + _ = update_job => KillReason::Cancelled, + }; + tx.send(()).expect("rx should never be dropped"); + drop(tx); + + let set_reason = async { + if kill_reason == KillReason::Timeout { + if let Err(err) = sqlx::query( + r#" + UPDATE queue + SET canceled = true + , canceled_by = 'timeout' + , canceled_reason = $1 + WHERE id = $2 + "#, + ) + .bind(format!("duration > {}", TIMEOUT_DURATION.as_secs())) + .bind(job_id) + .execute(&db) + .await + { + tracing::error!(%job_id, %err, "error setting cancelation reason for job {job_id}: {err}"); + } + } + }; + + /* send SIGKILL and reap child process */ + let (_, kill) = future::join(set_reason, child.kill()).await; + kill.map(|()| Err(kill_reason)) + }; + + /* 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 { + usize::MAX + }; + /* log_remaining is zero when output limit was reached */ + let mut log_remaining = max_log_size.saturating_sub(logs.chars().count()); + let mut result = io::Result::Ok(()); + let mut output = output.take_until(rx2.recv()).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(); + + while let Some(line) = output.by_ref().next().await { + + let do_write_ = do_write.shared(); + + let mut read_lines = stream::once(async { line }) + .chain(output.by_ref()) + /* after receiving a line, continue until some delay has passed + * _and_ the previous database write is complete */ + .take_until(future::join(sleep(write_logs_delay), do_write_.clone())) + .boxed(); + + /* Read up until an error is encountered, + * handle log lines first and then the error... */ + let mut joined = String::new(); + + while let Some(line) = read_lines.next().await { + + match line { + Ok(_) if log_remaining == 0 => (), + Ok(line) => { + if line.is_empty() { + continue; + } + append_with_limit(&mut joined, &line, &mut log_remaining); + if log_remaining == 0 { + tracing::info!(%job_id, "Too many logs lines for job {job_id}"); + let _ = set_too_many_logs.send(true); + joined.push_str(&format!( + "Job logs or result reached character limit of {MAX_RESULT_SIZE}; killing job." + )); + /* stop reading and drop our streams fairly quickly */ + break; + } + } + Err(err) => { + result = Err(err); + break; + } + } + } + + logs.push_str(&joined); + + + /* Ensure the last flush completed before starting a new one. + * + * This shouldn't pause since `take_until()` reads lines until `do_write` + * resolves. We only stop reading lines before `take_until()` resolves if we reach + * EOF or a read error. In those cases, waiting on a database query to complete is + * fine because we're done. */ + + if let Some(Ok(p)) = do_write_ + .then(|()| write_result) + .await + .err() + .map(|err| err.try_into_panic()) + { + panic::resume_unwind(p); + } + + (do_write, write_result) = tokio::spawn(append_logs(job_id, joined, db.clone())).remote_handle(); + + if let Err(err) = result { + tracing::error!(%job_id, %err, "error reading output for job {job_id}: {err}"); + break; + } + + if *set_too_many_logs.borrow() { + break; + } + + } + + /* drop our end of the pipe */ + drop(output); + + if let Some(Ok(p)) = do_write + .then(|()| write_result) + .await + .err() + .map(|err| err.try_into_panic()) + { + panic::resume_unwind(p); + } + }.instrument(trace_span!("child_lines")); + + let (wait_result, _) = tokio::join!(wait_on_child, lines); + + tracing::info!(%job_id, "child process '{child_name}' for {job_id} took {}ms", start.elapsed().as_millis()); + match wait_result { + _ if *too_many_logs.borrow() => Err(Error::ExecutionErr(format!( + "logs or result reached limit. (current max size: {MAX_RESULT_SIZE} characters)" + ))), + Ok(Ok(status)) => { + if status.success() { + Ok(()) + } else if let Some(code) = status.code() { + Err(error::Error::ExitStatus(code)) + } else { + Err(error::Error::ExecutionErr(format!( + "process terminated by signal: {:#?}, stopped_signal: {:#?}, core_dumped: {}", + status.signal(), + status.stopped_signal(), + status.core_dumped() + ))) + } + } + Ok(Err(kill_reason)) => Err(Error::ExecutionErr(format!( + "job process killed because {kill_reason:#?}" + ))), + Err(err) => Err(Error::ExecutionErr(format!("job process io error: {err}"))), + } +} + +/// takes stdout and stderr from Child, panics if either are not present +/// +/// builds a stream joining both stdout and stderr each read line by line +fn child_joined_output_stream( + child: &mut Child, +) -> impl stream::FusedStream> { + let stderr = child + .stderr + .take() + .expect("child did not have a handle to stdout"); + + let stdout = child + .stdout + .take() + .expect("child did not have a handle to stdout"); + + let stdout = BufReader::new(stdout).lines(); + let stderr = BufReader::new(stderr).lines(); + stream::select(lines_to_stream(stderr), lines_to_stream(stdout)) +} + +fn lines_to_stream( + mut lines: tokio::io::Lines, +) -> impl futures::Stream> { + stream::poll_fn(move |cx| { + std::pin::Pin::new(&mut lines) + .poll_next_line(cx) + .map(|result| result.transpose()) + }) +} + +// 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 +fn append_with_limit(dst: &mut String, src: &str, limit: &mut usize) { + if *limit > 0 { + dst.push('\n'); + } + *limit -= 1; + + let src_len = src.chars().count(); + if src_len <= *limit { + dst.push_str(&src); + *limit -= src_len; + } else { + let byte_pos = src + .char_indices() + .skip(*limit) + .next() + .map(|(byte_pos, _)| byte_pos) + .unwrap_or(0); + dst.push_str(&src[0..byte_pos]); + *limit = 0; + } +} + +/* TODO retry this? */ +#[tracing::instrument(level = "trace", skip_all)] +async fn append_logs(job_id: uuid::Uuid, logs: impl AsRef, db: impl Borrow>) { + if logs.as_ref().is_empty() { + 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}"); + } +} diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs new file mode 100644 index 0000000000..598bdf0d4e --- /dev/null +++ b/backend/windmill-worker/src/deno_executor.rs @@ -0,0 +1,255 @@ +use std::{collections::HashMap, process::Stdio}; + +use itertools::Itertools; + +use crate::{ + common::{get_reserved_variables, handle_child, read_result, set_logs}, + create_args_and_out_file, write_file, AuthedClientBackgroundTask, DENO_CACHE_DIR, DENO_PATH, + DISABLE_NSJAIL, NPM_CONFIG_REGISTRY, PATH_ENV, +}; +use tokio::process::Command; +use windmill_common::{error::Result, BASE_URL}; +use windmill_common::{ + error::{self}, + jobs::QueuedJob, +}; +use windmill_parser::Typ; + +lazy_static::lazy_static! { + + static ref DENO_FLAGS: Option> = std::env::var("DENO_FLAGS") + .ok() + .map(|x| x.split(' ').map(|x| x.to_string()).collect()); + + static ref DENO_EXTRA_IMPORT_MAP: String = std::env::var("DENO_EXTRA_IMPORT_MAP") + .ok() + .map(|x| x.split(',').map(|x| { + let mut splitted = x.split("="); + let key = splitted.next().unwrap(); + let value = splitted.next().unwrap(); + format!(",\n \"{key}\": \"{value}\"") + }).join("\n")).unwrap_or_else(|| String::new()); + + static ref DENO_AUTH_TOKENS: String = std::env::var("DENO_AUTH_TOKENS") + .ok() + .map(|x| format!(";{x}")) + .unwrap_or_else(|| String::new()); + + +} +fn get_common_deno_proc_envs(token: &str, base_internal_url: &str) -> HashMap { + let hostname_base = BASE_URL.split("://").last().unwrap_or("localhost"); + let hostname_internal = base_internal_url.split("://").last().unwrap_or("localhost"); + let deno_auth_tokens_base = DENO_AUTH_TOKENS.as_str(); + let deno_auth_tokens = + format!("{token}@{hostname_base};{token}@{hostname_internal}{deno_auth_tokens_base}",); + + let mut deno_envs: HashMap = HashMap::from([ + (String::from("PATH"), PATH_ENV.clone()), + (String::from("DENO_AUTH_TOKENS"), deno_auth_tokens), + ( + String::from("BASE_INTERNAL_URL"), + base_internal_url.to_string(), + ), + ]); + + if let Some(ref s) = *NPM_CONFIG_REGISTRY { + deno_envs.insert(String::from("NPM_CONFIG_REGISTRY"), s.clone()); + } + return deno_envs; +} + +#[tracing::instrument(level = "trace", skip_all)] +pub async fn handle_deno_job( + logs: &mut String, + job: &QueuedJob, + db: &sqlx::Pool, + client: &AuthedClientBackgroundTask, + job_dir: &str, + inner_content: &String, + base_internal_url: &str, + worker_name: &str, + envs: HashMap, +) -> error::Result { + // 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 write_main_f = write_file(job_dir, "main.ts", inner_content); + + let write_wrapper_f = async { + // let mut start = Instant::now(); + let args = windmill_parser_ts::parse_deno_signature(inner_content, true)?.args; + let dates = args + .iter() + .enumerate() + .filter_map(|(i, x)| { + if matches!(x.typ, Typ::Datetime) { + Some(i) + } else { + None + } + }) + .map(|x| return format!("args[{x}] = args[{x}] ? new Date(args[{x}]) : undefined")) + .join("\n"); + + let spread = args.into_iter().map(|x| x.name).join(","); + // logs.push_str(format!("infer args: {:?}\n", start.elapsed().as_micros()).as_str()); + let wrapper_content: String = format!( + r#" +import {{ main }} from "./main.ts"; + +const args = await Deno.readTextFile("args.json") + .then(JSON.parse) + .then(({{ {spread} }}) => [ {spread} ]) + +BigInt.prototype.toJSON = function () {{ + return this.toString(); +}}; + +{dates} +async function run() {{ + let res: any = await main(...args); + const res_json = JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value); + await Deno.writeTextFile("result.json", res_json); + Deno.exit(0); +}} +run().catch(async (e) => {{ + await Deno.writeTextFile("result.json", JSON.stringify({{ message: e.message, name: e.name, stack: e.stack }})); + Deno.exit(1); +}}); + "#, + ); + write_file(job_dir, "wrapper.ts", &wrapper_content).await?; + Ok(()) as error::Result<()> + }; + + let write_import_map_f = async { + let w_id = job.workspace_id.clone(); + let script_path_split = job.script_path().split("/"); + let script_path_parts_len = script_path_split.clone().count(); + let mut relative_mounts = "".to_string(); + for c in 0..script_path_parts_len { + relative_mounts += ",\n "; + relative_mounts += &format!( + "\"./{}\": \"{base_internal_url}/api/w/{w_id}/scripts/raw/p/{}{}\"", + (0..c).map(|_| "../").join(""), + &script_path_split + .clone() + .take(script_path_parts_len - c - 1) + .join("/"), + if c == script_path_parts_len - 1 { + "" + } else { + "/" + }, + ); + } + let extra_import_map = DENO_EXTRA_IMPORT_MAP.as_str(); + let import_map = format!( + r#"{{ + "imports": {{ + "{base_internal_url}/api/w/{w_id}/scripts/raw/p/": "{base_internal_url}/api/w/{w_id}/scripts/raw/p/", + "{base_internal_url}": "{base_internal_url}/api/w/{w_id}/scripts/raw/p/", + "/": "{base_internal_url}/api/w/{w_id}/scripts/raw/p/", + "./wrapper.ts": "./wrapper.ts", + "./main.ts": "./main.ts"{relative_mounts} + {extra_import_map} + }} + }}"#, + ); + write_file(job_dir, "import_map.json", &import_map).await?; + Ok(()) as error::Result<()> + }; + + let reserved_variables_args_out_f = async { + let client = client.get_authed().await; + let args_and_out_f = async { + create_args_and_out_file(&client, job, job_dir).await?; + Ok(()) as Result<()> + }; + let reserved_variables_f = async { + let mut vars = get_reserved_variables(job, &client.token, db).await?; + vars.insert("RUST_LOG".to_string(), "info".to_string()); + Ok(vars) as Result> + }; + let (_, reserved_variables) = tokio::try_join!(args_and_out_f, reserved_variables_f)?; + Ok((reserved_variables, client.token)) as error::Result<(HashMap, String)> + }; + + let (_, (reserved_variables, token), _, _, _) = tokio::try_join!( + set_logs_f, + reserved_variables_args_out_f, + write_main_f, + write_wrapper_f, + write_import_map_f + )?; + + let common_deno_proc_envs = get_common_deno_proc_envs(&token, base_internal_url); + + //do not cache local dependencies + let reload = format!("--reload={base_internal_url}"); + let child = { + let script_path = format!("{job_dir}/wrapper.ts"); + let import_map_path = format!("{job_dir}/import_map.json"); + let mut args = Vec::with_capacity(12); + args.push("run"); + args.push("--no-check"); + args.push("--import-map"); + args.push(&import_map_path); + args.push(&reload); + args.push("--unstable"); + if let Some(deno_flags) = DENO_FLAGS.as_ref() { + for flag in deno_flags { + args.push(flag); + } + } else if !*DISABLE_NSJAIL { + args.push("--allow-net"); + args.push("--allow-read=./,/tmp/windmill/cache/deno/"); + args.push("--allow-write=./"); + args.push("--allow-env"); + } else { + args.push("-A"); + } + args.push(&script_path); + Command::new(DENO_PATH.as_str()) + .current_dir(job_dir) + .env_clear() + .envs(envs) + .envs(reserved_variables) + .envs(common_deno_proc_envs) + .env("DENO_DIR", DENO_CACHE_DIR) + .args(args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()? + }; + // logs.push_str(format!("prepare: {:?}\n", start.elapsed().as_micros()).as_str()); + // start = Instant::now(); + handle_child( + &job.id, + db, + logs, + child, + false, + worker_name, + &job.workspace_id, + "deno run", + job.timeout, + ) + .await?; + // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str()); + if let Err(e) = tokio::fs::remove_dir_all(format!("{DENO_CACHE_DIR}/gen/file/{job_dir}")).await + { + tracing::error!("failed to remove deno gen tmp cache dir: {}", e); + } + read_result(job_dir).await +} diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 7854643a8d..3f970f835c 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -15,10 +15,10 @@ use windmill_common::{ use windmill_parser_go::parse_go_imports; use crate::{ - common::{capitalize, read_result, set_logs}, - create_args_and_out_file, get_reserved_variables, handle_child, write_file, - AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, GOPRIVATE, GOPROXY, - GO_BIN_CACHE_DIR, GO_CACHE_DIR, HOME_ENV, NSJAIL_PATH, PATH_ENV, + common::{capitalize, get_reserved_variables, handle_child, read_result, set_logs}, + create_args_and_out_file, write_file, AuthedClientBackgroundTask, DISABLE_NSJAIL, + DISABLE_NUSER, GOPRIVATE, GOPROXY, GO_BIN_CACHE_DIR, GO_CACHE_DIR, HOME_ENV, NSJAIL_PATH, + PATH_ENV, }; const GO_REQ_SPLITTER: &str = "//go.sum\n"; diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index f417e04a5b..5d5dbb2bfa 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -3,8 +3,10 @@ mod bigquery_executor; #[cfg(feature = "enterprise")] mod snowflake_executor; +mod bash_executor; mod bun_executor; mod common; +mod deno_executor; mod global_cache; mod go_executor; mod graphql_executor; diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 39beb099dd..eda6cb048a 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -57,10 +57,10 @@ use crate::global_cache::{build_tar_and_push, pull_from_tar}; use crate::S3_CACHE_BUCKET; use crate::{ - common::{read_result, set_logs}, - create_args_and_out_file, get_reserved_variables, handle_child, write_file, - AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HTTPS_PROXY, HTTP_PROXY, - LOCK_CACHE_DIR, NO_PROXY, NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR, + common::{get_reserved_variables, handle_child, read_result, set_logs}, + create_args_and_out_file, write_file, AuthedClientBackgroundTask, DISABLE_NSJAIL, + DISABLE_NUSER, HTTPS_PROXY, HTTP_PROXY, LOCK_CACHE_DIR, NO_PROXY, NSJAIL_PATH, PATH_ENV, + PIP_CACHE_DIR, }; pub async fn create_dependencies_dir(job_dir: &str) { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 9ff6393a8b..022c8a4e30 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -12,15 +12,12 @@ use itertools::Itertools; use once_cell::sync::OnceCell; use sqlx::{Pool, Postgres}; use windmill_api_client::Client; -use windmill_parser::Typ; use std::{ - borrow::Borrow, collections::HashMap, io, os::unix::process::ExitStatusExt, panic, - process::Stdio, time::Duration, + collections::HashMap, time::Duration, sync::{Arc, atomic::Ordering}, collections::hash_map::DefaultHasher, hash::{Hasher, Hash}, }; -use regex::Regex; use tracing::{trace_span, Instrument}; use uuid::Uuid; @@ -29,7 +26,7 @@ use windmill_common::{ flows::{FlowModuleValue, FlowValue, FlowModule}, scripts::{ScriptHash, ScriptLang, get_full_hub_script_by_path}, utils::{rd_string, StripPath}, - variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, METRICS_ENABLED, jobs::{JobKind, QueuedJob, Metrics}, IS_READY, + users::SUPERADMIN_SECRET_EMAIL, jobs::{JobKind, QueuedJob, Metrics}, METRICS_ENABLED, IS_READY, }; use windmill_queue::{canceled_job_to_result, get_queued_job, pull, CLOUD_HOSTED, HTTP_CLIENT, ACCEPTED_TAGS, IS_WORKER_TAGS_DEFINED}; @@ -37,18 +34,14 @@ use serde_json::{json, Value}; use tokio::{ fs::{symlink, DirBuilder, File}, - io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, - process::{Child, Command}, + io::AsyncWriteExt, sync::{ - mpsc::{self, Sender}, watch, broadcast, RwLock, Barrier + mpsc::{self, Sender}, RwLock, Barrier }, - time::{interval, sleep, Instant, MissedTickBehavior} + time::Instant }; -use futures::{ - future::{self, ready, FutureExt}, - stream, StreamExt, -}; +use futures::future::FutureExt; use async_recursion::async_recursion; use windmill_api_client::types::CreateResource; @@ -64,7 +57,7 @@ use windmill_queue::{add_completed_job, add_completed_job_error,IDLE_WORKERS}; use crate::{ worker_flow::{ handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress, - }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, mysql_executor::do_mysql, graphql_executor::do_graphql, bun_executor::{handle_bun_job, gen_lockfile}, + }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, mysql_executor::do_mysql, graphql_executor::do_graphql, bun_executor::{handle_bun_job, gen_lockfile}, bash_executor::{ANSI_ESCAPE_RE, handle_powershell_job, handle_bash_job}, deno_executor::handle_deno_job, }; #[cfg(feature = "enterprise")] @@ -153,8 +146,6 @@ const NUM_SECS_PING: u64 = 5; const INCLUDE_DEPS_PY_SH_CONTENT: &str = include_str!("../nsjail/download_deps.py.sh"); -const NSJAIL_CONFIG_RUN_BASH_CONTENT: &str = include_str!("../nsjail/run.bash.config.proto"); -pub const NSJAIL_CONFIG_RUN_BUN_CONTENT: &str = include_str!("../nsjail/run.bun.config.proto"); pub const DEFAULT_TIMEOUT: u64 = 900; @@ -198,25 +189,9 @@ lazy_static::lazy_static! { pub static ref GOPROXY: Option = std::env::var("GOPROXY").ok(); pub static ref NETRC: Option = std::env::var("NETRC").ok(); - static ref DENO_AUTH_TOKENS: String = std::env::var("DENO_AUTH_TOKENS") - .ok() - .map(|x| format!(";{x}")) - .unwrap_or_else(|| String::new()); pub static ref NPM_CONFIG_REGISTRY: Option = std::env::var("NPM_CONFIG_REGISTRY").ok(); - static ref DENO_FLAGS: Option> = std::env::var("DENO_FLAGS") - .ok() - .map(|x| x.split(' ').map(|x| x.to_string()).collect()); - - static ref DENO_EXTRA_IMPORT_MAP: String = std::env::var("DENO_EXTRA_IMPORT_MAP") - .ok() - .map(|x| x.split(',').map(|x| { - let mut splitted = x.split("="); - let key = splitted.next().unwrap(); - let value = splitted.next().unwrap(); - format!(",\n \"{key}\": \"{value}\"") - }).join("\n")).unwrap_or_else(|| String::new()); pub static ref WHITELIST_ENVS: Option> = std::env::var("WHITELIST_ENVS") @@ -244,7 +219,7 @@ lazy_static::lazy_static! { .and_then(|x| x.parse::().ok()) .unwrap_or(DEFAULT_TIMEOUT); - static ref TIMEOUT_DURATION: Duration = Duration::from_secs(*TIMEOUT); + pub static ref TIMEOUT_DURATION: Duration = Duration::from_secs(*TIMEOUT); pub static ref SESSION_TOKEN_EXPIRY: i32 = (*TIMEOUT as i32) * 2; @@ -265,9 +240,7 @@ lazy_static::lazy_static! { pub static ref CAN_PULL: Arc> = Arc::new(RwLock::new(())); - pub static ref ANSI_ESCAPE_RE: Regex = Regex::new(r"\x1b\[[0-9;]*m").unwrap(); } - //only matter if CLOUD_HOSTED pub const MAX_RESULT_SIZE: usize = 1024 * 1024 * 2; // 2MB @@ -1564,379 +1537,6 @@ mount {{ } -#[tracing::instrument(level = "trace", skip_all)] -async fn handle_bash_job( - logs: &mut String, - job: &QueuedJob, - db: &sqlx::Pool, - client: &AuthedClientBackgroundTask, - content: &str, - job_dir: &str, - shared_mount: &str, - base_internal_url: &str, - worker_name: &str, - envs: HashMap, -) -> Result { - logs.push_str("\n\n--- BASH CODE EXECUTION ---\n"); - set_logs(logs, &job.id, db).await; - write_file(job_dir, "main.sh", &format!("set -e\n{content}\necho \"\"\nsleep 0.02")).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 client = client.get_authed().await; - let hm = match transform_json_value("args", &client, &job.workspace_id, job.args.clone().unwrap_or_else(|| json!({}))).await? - { - Value::Object(ref hm) => hm.clone(), - _ => serde_json::Map::new(), - }; - let args_owned = windmill_parser_bash::parse_bash_sig(&content)? - .args - .iter() - .map(|arg| { - hm.get(&arg.name) - .and_then(|v| match v { - Value::String(s) => Some(s.clone()), - _ => serde_json::to_string(v).ok(), - }) - .unwrap_or_else(String::new) - }) - .collect::>(); - let args = args_owned.iter().map(|s| &s[..]).collect::>(); - - let child = if !*DISABLE_NSJAIL { - let _ = write_file( - job_dir, - "run.config.proto", - &NSJAIL_CONFIG_RUN_BASH_CONTENT - .replace("{JOB_DIR}", job_dir) - .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) - .replace("{SHARED_MOUNT}", shared_mount), - ) - .await?; - let mut cmd_args = vec!["--config", "run.config.proto", "--", "/bin/bash", "main.sh"]; - cmd_args.extend(args); - Command::new(NSJAIL_PATH.as_str()) - .current_dir(job_dir) - .env_clear() - .envs(reserved_variables) - .env("PATH", PATH_ENV.as_str()) - .env("BASE_INTERNAL_URL", base_internal_url) - .args(cmd_args) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()? - } else { - let mut cmd_args = vec!["main.sh"]; - cmd_args.extend(&args); - Command::new("/bin/bash") - .current_dir(job_dir) - .env_clear() - .envs(envs) - .envs(reserved_variables) - .env("PATH", PATH_ENV.as_str()) - .env("BASE_INTERNAL_URL", base_internal_url) - .env("HOME", HOME_ENV.as_str()) - .args(cmd_args) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()? - }; - handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL, worker_name, &job.workspace_id, "bash run", job.timeout).await?; - //for now bash jobs have an empty result object - Ok(serde_json::json!(logs - .lines() - .last() - .map(|x| ANSI_ESCAPE_RE.replace_all(x, "").to_string()) - .unwrap_or_else(String::new))) -} - - -#[tracing::instrument(level = "trace", skip_all)] -async fn handle_powershell_job( - logs: &mut String, - job: &QueuedJob, - db: &sqlx::Pool, - client: &AuthedClientBackgroundTask, - content: &str, - job_dir: &str, - shared_mount: &str, - base_internal_url: &str, - worker_name: &str, - envs: HashMap, -) -> Result { - logs.push_str("\n\n--- POWERSHELL CODE EXECUTION ---\n"); - set_logs(logs, &job.id, db).await; - let pwsh_args = { - let client = client.get_authed().await; - let hm = match transform_json_value("args", &client, &job.workspace_id, job.args.clone().unwrap_or_else(|| json!({}))).await? - { - Value::Object(ref hm) => hm.clone(), - _ => serde_json::Map::new(), - }; - - let args_owned = windmill_parser_bash::parse_powershell_sig(&content)? - .args - .iter() - .map(|arg| { - (arg.name.clone(), hm.get(&arg.name) - .and_then(|v| match v { - Value::String(s) => Some(s.clone()), - _ => serde_json::to_string(v).ok(), - }) - .unwrap_or_else(String::new)) - }) - .collect::>(); - args_owned.iter().map(|(n, v)| format!("--{n} {v}")).join(" ") - }; - - let content = content - .replace('$', r"\$") // escape powershell variables - .replace("`", r"\`"); // escape powershell backticks - - write_file(job_dir, "main.sh", &format!("set -e\ncat > script.ps1 << EOF\n{content}\nEOF\npwsh -File script.ps1 {pwsh_args}\necho \"\"\nsleep 0.02")).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 child = if !*DISABLE_NSJAIL { - let _ = write_file( - job_dir, - "run.config.proto", - &NSJAIL_CONFIG_RUN_BASH_CONTENT - .replace("{JOB_DIR}", job_dir) - .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) - .replace("{SHARED_MOUNT}", shared_mount), - ) - .await?; - let cmd_args = vec!["--config", "run.config.proto", "--", "/bin/bash", "main.sh"]; - Command::new(NSJAIL_PATH.as_str()) - .current_dir(job_dir) - .env_clear() - .envs(reserved_variables) - .env("PATH", PATH_ENV.as_str()) - .env("BASE_INTERNAL_URL", base_internal_url) - .args(cmd_args) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()? - } else { - let cmd_args = vec!["main.sh"]; - Command::new("/bin/bash") - .current_dir(job_dir) - .env_clear() - .envs(envs) - .envs(reserved_variables) - .env("PATH", PATH_ENV.as_str()) - .env("BASE_INTERNAL_URL", base_internal_url) - .env("HOME", HOME_ENV.as_str()) - .args(cmd_args) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()? - }; - handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL, worker_name, &job.workspace_id, "bash/powershell run", job.timeout).await?; - //for now bash jobs have an empty result object - Ok(serde_json::json!(logs - .lines() - .last() - .map(|x| ANSI_ESCAPE_RE.replace_all(x, "").to_string()) - .unwrap_or_else(String::new))) -} - -fn get_common_deno_proc_envs(token: &str, base_internal_url: &str) -> HashMap { - let hostname_base = BASE_URL.split("://").last().unwrap_or("localhost"); - let hostname_internal = base_internal_url.split("://").last().unwrap_or("localhost"); - let deno_auth_tokens_base = DENO_AUTH_TOKENS.as_str(); - let deno_auth_tokens = - format!("{token}@{hostname_base};{token}@{hostname_internal}{deno_auth_tokens_base}",); - - let mut deno_envs: HashMap = HashMap::from([ - (String::from("PATH"), PATH_ENV.clone()), - (String::from("DENO_AUTH_TOKENS"), deno_auth_tokens), - (String::from("BASE_INTERNAL_URL"), base_internal_url.to_string()), - ]); - - if let Some(ref s) = *NPM_CONFIG_REGISTRY { - deno_envs.insert(String::from("NPM_CONFIG_REGISTRY"), s.clone()); - } - return deno_envs; -} - - - -#[tracing::instrument(level = "trace", skip_all)] -async fn handle_deno_job( - logs: &mut String, - job: &QueuedJob, - db: &sqlx::Pool, - client: &AuthedClientBackgroundTask, - job_dir: &str, - inner_content: &String, - base_internal_url: &str, - worker_name: &str, - envs: HashMap, -) -> error::Result { - - // 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 write_main_f = write_file(job_dir, "main.ts", inner_content); - - let write_wrapper_f = async { - // let mut start = Instant::now(); - let args = windmill_parser_ts::parse_deno_signature(inner_content, true)?.args; - let dates = args.iter().enumerate().filter_map(|(i, x)| if matches!(x.typ, Typ::Datetime) { - Some(i) - } else { - None - }).map(|x| { - return format!("args[{x}] = args[{x}] ? new Date(args[{x}]) : undefined") - }).join("\n"); - - let spread = args.into_iter().map(|x| x.name).join(","); - // logs.push_str(format!("infer args: {:?}\n", start.elapsed().as_micros()).as_str()); - let wrapper_content: String = format!( - r#" -import {{ main }} from "./main.ts"; - -const args = await Deno.readTextFile("args.json") - .then(JSON.parse) - .then(({{ {spread} }}) => [ {spread} ]) - -BigInt.prototype.toJSON = function () {{ - return this.toString(); -}}; - -{dates} -async function run() {{ - let res: any = await main(...args); - const res_json = JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value); - await Deno.writeTextFile("result.json", res_json); - Deno.exit(0); -}} -run().catch(async (e) => {{ - await Deno.writeTextFile("result.json", JSON.stringify({{ message: e.message, name: e.name, stack: e.stack }})); - Deno.exit(1); -}}); - "#, - ); - write_file(job_dir, "wrapper.ts", &wrapper_content).await?; - Ok(()) as error::Result<()> - }; - - let write_import_map_f = async { - let w_id = job.workspace_id.clone(); - let script_path_split = job.script_path().split("/"); - let script_path_parts_len = script_path_split.clone().count(); - let mut relative_mounts = "".to_string(); - for c in 0..script_path_parts_len { - relative_mounts += ",\n "; - relative_mounts += &format!("\"./{}\": \"{base_internal_url}/api/w/{w_id}/scripts/raw/p/{}{}\"", - (0..c).map(|_| "../").join(""), - &script_path_split.clone().take(script_path_parts_len - c - 1).join("/"), - if c == script_path_parts_len - 1 { "" } else { "/" }, - ); - } - let extra_import_map = DENO_EXTRA_IMPORT_MAP.as_str(); - let import_map = format!( - r#"{{ - "imports": {{ - "{base_internal_url}/api/w/{w_id}/scripts/raw/p/": "{base_internal_url}/api/w/{w_id}/scripts/raw/p/", - "{base_internal_url}": "{base_internal_url}/api/w/{w_id}/scripts/raw/p/", - "/": "{base_internal_url}/api/w/{w_id}/scripts/raw/p/", - "./wrapper.ts": "./wrapper.ts", - "./main.ts": "./main.ts"{relative_mounts} - {extra_import_map} - }} - }}"#, - ); - write_file(job_dir, "import_map.json", &import_map).await?; - Ok(()) as error::Result<()> - }; - - let reserved_variables_args_out_f = async { - let client = client.get_authed().await; - let args_and_out_f = async { - create_args_and_out_file(&client, job, job_dir).await?; - Ok(()) as Result<()> - }; - let reserved_variables_f = async { - let mut vars = get_reserved_variables(job, &client.token, db).await?; - vars.insert("RUST_LOG".to_string(), "info".to_string()); - Ok(vars) as Result> - }; - let (_, reserved_variables) = tokio::try_join!(args_and_out_f, reserved_variables_f)?; - Ok((reserved_variables, client.token)) as error::Result<(HashMap, String)> - }; - - let (_, (reserved_variables, token), _, _, _) = tokio::try_join!( - set_logs_f, - reserved_variables_args_out_f, - write_main_f, - write_wrapper_f, - write_import_map_f)?; - - let common_deno_proc_envs = get_common_deno_proc_envs(&token, base_internal_url); - - //do not cache local dependencies - let reload = format!("--reload={base_internal_url}"); - let child = { - let script_path = format!("{job_dir}/wrapper.ts"); - let import_map_path = format!("{job_dir}/import_map.json"); - let mut args = Vec::with_capacity(12); - args.push("run"); - args.push("--no-check"); - args.push("--import-map"); - args.push(&import_map_path); - args.push(&reload); - args.push("--unstable"); - if let Some(deno_flags) = DENO_FLAGS.as_ref() { - for flag in deno_flags { - args.push(flag); - } - } else if !*DISABLE_NSJAIL { - args.push("--allow-net"); - args.push("--allow-read=./,/tmp/windmill/cache/deno/"); - args.push("--allow-write=./"); - args.push("--allow-env"); - } else { - args.push("-A"); - } - args.push(&script_path); - Command::new(DENO_PATH.as_str()) - .current_dir(job_dir) - .env_clear() - .envs(envs) - .envs(reserved_variables) - .envs(common_deno_proc_envs) - .env("DENO_DIR", DENO_CACHE_DIR) - .args(args) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()? - }; - // logs.push_str(format!("prepare: {:?}\n", start.elapsed().as_micros()).as_str()); - // start = Instant::now(); - handle_child(&job.id, db, logs, child, false, worker_name, &job.workspace_id, "deno run", job.timeout).await?; - // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str()); - if let Err(e) = tokio::fs::remove_dir_all(format!("{DENO_CACHE_DIR}/gen/file/{job_dir}")).await { - tracing::error!("failed to remove deno gen tmp cache dir: {}", e); - } - read_result(job_dir).await -} - - - #[tracing::instrument(level = "trace", skip_all)] pub async fn create_args_and_out_file( client: &AuthedClient, @@ -2259,431 +1859,3 @@ async fn capture_dependency_job( } } - -#[tracing::instrument(level = "trace", skip_all)] -pub async fn get_reserved_variables( - job: &QueuedJob, - token: &str, - db: &sqlx::Pool, -) -> Result, Error> { - let flow_path = if let Some(uuid) = job.parent_job { - sqlx::query_scalar!("SELECT script_path FROM queue WHERE id = $1", uuid) - .fetch_optional(db) - .await? - .flatten() - } else { - None - }; - - let variables = variables::get_reserved_variables( - &job.workspace_id, - token, - &job.email, - &job.created_by, - &job.id.to_string(), - &job.permissioned_as, - - job.script_path.clone(), - job.parent_job.map(|x| x.to_string()), - flow_path, - job.schedule_path.clone(), - job.flow_step_id.clone() - ).to_vec(); - - let mut r: HashMap = variables - .into_iter() - .map(|rv| (rv.name, rv.value)) - .collect(); - - if let Some(ref envs) = *WHITELIST_ENVS { - for e in envs { - r.insert(e.0.clone(), e.1.clone()); - } - } - - Ok(r) -} - -async fn get_mem_peak(pid: Option, nsjail: bool) -> i32 { - if pid.is_none() { - return -1 - } - let pid = if nsjail { - // This is a bit hacky, but the process id of the nsjail process is the pid of nsjail + 1. - // Ideally, we would get the number from fork() itself. This works in MOST cases. - pid.unwrap() + 1 - } else { - pid.unwrap() - }; - - if let Ok(file) = File::open(format!("/proc/{}/status", pid)).await { - let mut lines = BufReader::new(file).lines(); - while let Some(line) = lines.next_line().await.unwrap_or(None) { - if line.starts_with("VmHWM:") { - return line.split_whitespace().nth(1).and_then(|s| s.parse::().ok()).unwrap_or(-1); - }; - } - -2 - } else { - -3 - } -} -/// - wait until child exits and return with exit status -/// - read lines from stdout and stderr and append them to the "queue"."logs" -/// quitting early if output exceedes MAX_LOG_SIZE characters (not bytes) -/// - update the `last_line` and `logs` strings with the program output -/// - update "queue"."last_ping" every five seconds -/// - kill process if we exceed timeout or "queue"."canceled" is set -#[tracing::instrument(level = "trace", skip_all)] -pub async fn handle_child( - job_id: &Uuid, - db: &Pool, - logs: &mut String, - mut child: Child, - nsjail: bool, - worker_name: &str, - _w_id: &str, - child_name: &str, - custom_timeout: Option, -) -> error::Result<()> { - let start = Instant::now(); - let update_job_interval = Duration::from_millis(500); - let write_logs_delay = Duration::from_millis(500); - - let pid = child.id(); - #[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; - } else { - tracing::info!("could not get child pid"); - } - let (set_too_many_logs, mut too_many_logs) = watch::channel::(false); - let (tx, mut rx) = broadcast::channel::<()>(3); - let mut rx2 = tx.subscribe(); - - - let output = child_joined_output_stream(&mut child); - - let job_id = job_id.clone(); - - - /* the cancellation future is polled on by `wait_on_child` while - * waiting for the child to exit normally */ - let update_job = async { - let db = db.clone(); - - let mut interval = interval(update_job_interval); - interval.set_missed_tick_behavior(MissedTickBehavior::Skip); - - let mut i = 1; - loop { - tokio::select!( - _ = rx.recv() => break, - _ = interval.tick() => { - // update the last_ping column every 5 seconds - i+=1; - if i % 10 == 0 { - sqlx::query!( - "UPDATE worker_ping SET ping_at = now() WHERE worker = $1", - &worker_name - ) - .execute(&db) - .await - .expect("update worker ping"); - } - let mem_peak = get_mem_peak(pid, nsjail).await; - tracing::info!("{job_id} still running. mem peak: {}kB", mem_peak); - let mem_peak = if mem_peak > 0 { Some(mem_peak) } else { None }; - if sqlx::query_scalar!("UPDATE queue SET mem_peak = GREATEST($1, mem_peak), last_ping = now() WHERE id = $2 RETURNING canceled", mem_peak, job_id) - .fetch_optional(&db) - .await - .map(|v| Some(true) == v) - .unwrap_or_else(|err| { - tracing::error!(%job_id, %err, "error checking cancelation for job {job_id}: {err}"); - false - }) - { - break; - } - }, - ); - } - }; - - #[derive(PartialEq, Debug)] - enum KillReason { - TooManyLogs, - Timeout, - Cancelled, - } - /* a future that completes when the child process exits */ - let wait_on_child = async { - let db = db.clone(); - - #[cfg(not(feature = "enterprise"))] - let instance_timeout_duration = *TIMEOUT_DURATION; - - #[cfg(feature = "enterprise")] - let premium_workspace = *CLOUD_HOSTED && sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", _w_id) - .fetch_one(&db) - .await - .map_err(|e| { - tracing::error!(%e, "error getting premium workspace for job {job_id}: {e}"); - }).unwrap_or(false); - - #[cfg(feature = "enterprise")] - let instance_timeout_duration = if premium_workspace { - *TIMEOUT_DURATION*6 //30mins - } else { - *TIMEOUT_DURATION - }; - - let timeout_duration = if let Some(custom_timeout) = custom_timeout { - Duration::min(instance_timeout_duration, Duration::from_secs(custom_timeout as u64)) - } else { - instance_timeout_duration - }; - - let kill_reason = tokio::select! { - biased; - result = child.wait() => return result.map(Ok), - Ok(()) = too_many_logs.changed() => KillReason::TooManyLogs, - _ = sleep(timeout_duration) => KillReason::Timeout, - _ = update_job => KillReason::Cancelled, - }; - tx.send(()).expect("rx should never be dropped"); - drop(tx); - - let set_reason = async { - if kill_reason == KillReason::Timeout { - if let Err(err) = sqlx::query( - r#" - UPDATE queue - SET canceled = true - , canceled_by = 'timeout' - , canceled_reason = $1 - WHERE id = $2 - "#, - ) - .bind(format!("duration > {}", TIMEOUT_DURATION.as_secs())) - .bind(job_id) - .execute(&db) - .await - { - tracing::error!(%job_id, %err, "error setting cancelation reason for job {job_id}: {err}"); - } - } - }; - - /* send SIGKILL and reap child process */ - let (_, kill) = future::join(set_reason, child.kill()).await; - kill.map(|()| Err(kill_reason)) - }; - - /* 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 { - usize::MAX - }; - /* log_remaining is zero when output limit was reached */ - let mut log_remaining = max_log_size.saturating_sub(logs.chars().count()); - let mut result = io::Result::Ok(()); - let mut output = output.take_until(rx2.recv()).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(); - - while let Some(line) = output.by_ref().next().await { - - let do_write_ = do_write.shared(); - - let mut read_lines = stream::once(async { line }) - .chain(output.by_ref()) - /* after receiving a line, continue until some delay has passed - * _and_ the previous database write is complete */ - .take_until(future::join(sleep(write_logs_delay), do_write_.clone())) - .boxed(); - - /* Read up until an error is encountered, - * handle log lines first and then the error... */ - let mut joined = String::new(); - - while let Some(line) = read_lines.next().await { - - match line { - Ok(_) if log_remaining == 0 => (), - Ok(line) => { - if line.is_empty() { - continue; - } - append_with_limit(&mut joined, &line, &mut log_remaining); - if log_remaining == 0 { - tracing::info!(%job_id, "Too many logs lines for job {job_id}"); - let _ = set_too_many_logs.send(true); - joined.push_str(&format!( - "Job logs or result reached character limit of {MAX_RESULT_SIZE}; killing job." - )); - /* stop reading and drop our streams fairly quickly */ - break; - } - } - Err(err) => { - result = Err(err); - break; - } - } - } - - logs.push_str(&joined); - - - /* Ensure the last flush completed before starting a new one. - * - * This shouldn't pause since `take_until()` reads lines until `do_write` - * resolves. We only stop reading lines before `take_until()` resolves if we reach - * EOF or a read error. In those cases, waiting on a database query to complete is - * fine because we're done. */ - - if let Some(Ok(p)) = do_write_ - .then(|()| write_result) - .await - .err() - .map(|err| err.try_into_panic()) - { - panic::resume_unwind(p); - } - - (do_write, write_result) = tokio::spawn(append_logs(job_id, joined, db.clone())).remote_handle(); - - if let Err(err) = result { - tracing::error!(%job_id, %err, "error reading output for job {job_id}: {err}"); - break; - } - - if *set_too_many_logs.borrow() { - break; - } - - } - - /* drop our end of the pipe */ - drop(output); - - if let Some(Ok(p)) = do_write - .then(|()| write_result) - .await - .err() - .map(|err| err.try_into_panic()) - { - panic::resume_unwind(p); - } - }.instrument(trace_span!("child_lines")); - - - let (wait_result, _) = tokio::join!(wait_on_child, lines); - - tracing::info!(%job_id, "child process '{child_name}' for {job_id} took {}ms", start.elapsed().as_millis()); - match wait_result { - _ if *too_many_logs.borrow() => Err(Error::ExecutionErr(format!( - "logs or result reached limit. (current max size: {MAX_RESULT_SIZE} characters)" - ))), - Ok(Ok(status)) => { - if status.success() { - Ok(()) - } else if let Some(code) = status.code() { - Err(error::Error::ExitStatus(code)) - } else { - Err(error::Error::ExecutionErr(format!( - "process terminated by signal: {:#?}, stopped_signal: {:#?}, core_dumped: {}", - status.signal(), - status.stopped_signal(), - status.core_dumped() - ))) - } - } - Ok(Err(kill_reason)) => Err(Error::ExecutionErr(format!( - "job process killed because {kill_reason:#?}" - ))), - Err(err) => Err(Error::ExecutionErr(format!("job process io error: {err}"))), - } -} - -/// takes stdout and stderr from Child, panics if either are not present -/// -/// builds a stream joining both stdout and stderr each read line by line -fn child_joined_output_stream( - child: &mut Child, -) -> impl stream::FusedStream> { - let stderr = child - .stderr - .take() - .expect("child did not have a handle to stdout"); - - let stdout = child - .stdout - .take() - .expect("child did not have a handle to stdout"); - - let stdout = BufReader::new(stdout).lines(); - let stderr = BufReader::new(stderr).lines(); - stream::select(lines_to_stream(stderr), lines_to_stream(stdout)) -} - -fn lines_to_stream( - mut lines: tokio::io::Lines, -) -> impl futures::Stream> { - stream::poll_fn(move |cx| { - std::pin::Pin::new(&mut lines) - .poll_next_line(cx) - .map(|result| result.transpose()) - }) -} - -// 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 -fn append_with_limit(dst: &mut String, src: &str, limit: &mut usize) { - if *limit > 0 { - dst.push('\n'); - } - *limit -= 1; - - let src_len = src.chars().count(); - if src_len <= *limit { - dst.push_str(&src); - *limit -= src_len; - } else { - let byte_pos = src - .char_indices() - .skip(*limit) - .next() - .map(|(byte_pos, _)| byte_pos) - .unwrap_or(0); - dst.push_str(&src[0..byte_pos]); - *limit = 0; - } -} - - - -/* TODO retry this? */ -#[tracing::instrument(level = "trace", skip_all)] -async fn append_logs(job_id: uuid::Uuid, logs: impl AsRef, db: impl Borrow>) { - if logs.as_ref().is_empty() { - 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}"); - } -}