diff --git a/backend/windmill-worker/nsjail/run.bash.config.proto b/backend/windmill-worker/nsjail/run.bash.config.proto index 525f88e481..183f66537f 100644 --- a/backend/windmill-worker/nsjail/run.bash.config.proto +++ b/backend/windmill-worker/nsjail/run.bash.config.proto @@ -86,6 +86,20 @@ mount { is_bind: true } +mount { + src: "{JOB_DIR}/result.json" + dst: "/tmp/result.json" + rw: true + is_bind: true +} + +mount { + src: "{JOB_DIR}/result.out" + dst: "/tmp/result.out" + rw: true + is_bind: true +} + iface_no_lo: true {SHARED_MOUNT} diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 59aab1334d..b680c12d60 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -9,9 +9,8 @@ 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, + 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! { diff --git a/backend/windmill-worker/src/bigquery_executor.rs b/backend/windmill-worker/src/bigquery_executor.rs index 88cb93d86d..08a374202d 100644 --- a/backend/windmill-worker/src/bigquery_executor.rs +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -6,7 +6,7 @@ use windmill_queue::HTTP_CLIENT; use serde::Deserialize; -use crate::{transform_json_value, AuthedClient, JobCompleted}; +use crate::{common::transform_json_value, AuthedClient, JobCompleted}; use gcp_auth::{AuthenticationManager, CustomServiceAccount}; diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index b99f578621..5d458aded1 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -5,10 +5,12 @@ use itertools::Itertools; use uuid::Uuid; use crate::{ - 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, + common::{ + create_args_and_out_file, get_reserved_variables, handle_child, read_result, set_logs, + 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; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 4befd36950..471a5cdf67 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1,3 +1,5 @@ +use async_recursion::async_recursion; +use serde_json::{Value, json}; use sqlx::{Pool, Postgres}; use tokio::{fs::File, io::AsyncReadExt}; use windmill_common::{ @@ -28,7 +30,87 @@ use futures::{ stream, StreamExt, }; -use crate::{MAX_RESULT_SIZE, TIMEOUT_DURATION, WHITELIST_ENVS}; +use crate::{AuthedClient, MAX_RESULT_SIZE, TIMEOUT_DURATION, WHITELIST_ENVS}; + +#[tracing::instrument(level = "trace", skip_all)] +pub async fn create_args_and_out_file( + client: &AuthedClient, + job: &QueuedJob, + job_dir: &str, +) -> Result<(), Error> { + let args = if let Some(args) = &job.args { + Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + } else { + None + }; + let ser_args = serde_json::to_string(&args).map_err(|e| Error::ExecutionErr(e.to_string()))?; + write_file(job_dir, "args.json", &ser_args).await?; + write_file(job_dir, "result.json", "").await?; + Ok(()) +} + +#[tracing::instrument(level = "trace", skip_all)] +pub async fn write_file(dir: &str, path: &str, content: &str) -> error::Result { + let path = format!("{}/{}", dir, path); + let mut file = File::create(&path).await?; + file.write_all(content.as_bytes()).await?; + file.flush().await?; + Ok(file) +} + +#[tracing::instrument(level = "trace", skip_all)] +pub async fn write_file_binary(dir: &str, path: &str, content: &[u8]) -> error::Result { + let path = format!("{}/{}", dir, path); + let mut file = File::create(&path).await?; + file.write_all(content).await?; + file.flush().await?; + Ok(file) +} + +#[async_recursion] +pub async fn transform_json_value( + name: &str, + client: &AuthedClient, + workspace: &str, + v: Value, +) -> error::Result { + tracing::info!("transform_json_value {name}", name = name); + match v { + Value::String(y) if y.starts_with("$var:") => { + let path = y.strip_prefix("$var:").unwrap(); + client + .get_client() + .get_variable_value(workspace, path) + .await + .map_err(|_| Error::NotFound(format!("Variable {path} not found for `{name}`"))) + .map(|v| json!(v.into_inner())) + } + Value::String(y) if y.starts_with("$res:") => { + let path = y.strip_prefix("$res:").unwrap(); + if path.split("/").count() < 2 { + return Err(Error::InternalErr(format!( + "Argument `{name}` is an invalid resource path: {path}", + ))); + } + Ok(client + .get_client() + .get_resource_value_interpolated(workspace, path) + .await + .map_err(|_| Error::NotFound(format!("Resource {path} not found for `{name}`")))? + .into_inner()) + } + Value::Object(mut m) => { + for (a, b) in m.clone().into_iter() { + m.insert( + a.clone(), + transform_json_value(&a, client, workspace, b).await?, + ); + } + Ok(Value::Object(m)) + } + a @ _ => Ok(a), + } +} pub async fn read_result(job_dir: &str) -> error::Result { let mut file = File::open(format!("{job_dir}/result.json")).await?; diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 598bdf0d4e..5883ae4cd8 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -3,9 +3,12 @@ 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, + common::{ + create_args_and_out_file, get_reserved_variables, handle_child, read_result, set_logs, + 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}; diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 3f970f835c..8a5a2ed67c 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -15,10 +15,12 @@ use windmill_common::{ use windmill_parser_go::parse_go_imports; use crate::{ - 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, + common::{ + capitalize, create_args_and_out_file, get_reserved_variables, handle_child, read_result, + set_logs, 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/graphql_executor.rs b/backend/windmill-worker/src/graphql_executor.rs index f2c13be746..7e5cb7f67a 100644 --- a/backend/windmill-worker/src/graphql_executor.rs +++ b/backend/windmill-worker/src/graphql_executor.rs @@ -7,7 +7,7 @@ use windmill_queue::HTTP_CLIENT; use serde::Deserialize; -use crate::{transform_json_value, AuthedClient, JobCompleted}; +use crate::{common::transform_json_value, AuthedClient, JobCompleted}; #[derive(Deserialize)] struct GraphqlApi { diff --git a/backend/windmill-worker/src/mysql_executor.rs b/backend/windmill-worker/src/mysql_executor.rs index a787f07729..19a7901d41 100644 --- a/backend/windmill-worker/src/mysql_executor.rs +++ b/backend/windmill-worker/src/mysql_executor.rs @@ -10,7 +10,7 @@ use windmill_common::{ }; use windmill_parser_sql::parse_mysql_sig; -use crate::{transform_json_value, AuthedClient, JobCompleted}; +use crate::{common::transform_json_value, AuthedClient, JobCompleted}; #[derive(Deserialize)] struct MysqlDatabase { diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 627a869093..206a382d5d 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -17,7 +17,8 @@ use windmill_common::error::Error; use windmill_common::{error::to_anyhow, jobs::QueuedJob}; use windmill_parser_sql::parse_pgsql_sig; -use crate::{transform_json_value, AuthedClient, JobCompleted}; +use crate::common::transform_json_value; +use crate::{AuthedClient, JobCompleted}; #[derive(Deserialize)] struct PgDatabase { diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index eda6cb048a..86b8e2b849 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -57,10 +57,12 @@ use crate::global_cache::{build_tar_and_push, pull_from_tar}; use crate::S3_CACHE_BUCKET; use crate::{ - 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, + common::{ + create_args_and_out_file, get_reserved_variables, handle_child, read_result, set_logs, + 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/snowflake_executor.rs b/backend/windmill-worker/src/snowflake_executor.rs index 7802184e5b..6eb29af4ee 100644 --- a/backend/windmill-worker/src/snowflake_executor.rs +++ b/backend/windmill-worker/src/snowflake_executor.rs @@ -12,7 +12,7 @@ use windmill_queue::HTTP_CLIENT; use serde::{Deserialize, Serialize}; -use crate::{transform_json_value, AuthedClient, JobCompleted}; +use crate::{common::transform_json_value, AuthedClient, JobCompleted}; #[derive(Serialize)] struct Claims { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 2d608dca32..2f36970816 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -33,8 +33,7 @@ use windmill_queue::{canceled_job_to_result, get_queued_job, pull, CLOUD_HOSTED, use serde_json::{json, Value}; use tokio::{ - fs::{symlink, DirBuilder, File}, - io::AsyncWriteExt, + fs::{symlink, DirBuilder}, sync::{ mpsc::{self, Sender}, RwLock, Barrier }, @@ -57,7 +56,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}, bash_executor::{ANSI_ESCAPE_RE, handle_powershell_job, handle_bash_job}, deno_executor::handle_deno_job, + }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs, write_file, transform_json_value}, 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")] @@ -1214,67 +1213,6 @@ async fn handle_queued_job( Ok(()) } -#[tracing::instrument(level = "trace", skip_all)] -pub async fn write_file(dir: &str, path: &str, content: &str) -> error::Result { - let path = format!("{}/{}", dir, path); - let mut file = File::create(&path).await?; - file.write_all(content.as_bytes()).await?; - file.flush().await?; - Ok(file) -} - -#[tracing::instrument(level = "trace", skip_all)] -pub async fn write_file_binary(dir: &str, path: &str, content: &[u8]) -> error::Result { - let path = format!("{}/{}", dir, path); - let mut file = File::create(&path).await?; - file.write_all(content).await?; - file.flush().await?; - Ok(file) -} - -#[async_recursion] -pub async fn transform_json_value( - name: &str, - client: &AuthedClient, - workspace: &str, - v: Value, -) -> error::Result { - tracing::info!("transform_json_value {name}", name = name); - match v { - Value::String(y) if y.starts_with("$var:") => { - let path = y.strip_prefix("$var:").unwrap(); - client.get_client() - .get_variable_value(workspace, path) - .await - .map_err(|_| Error::NotFound(format!("Variable {path} not found for `{name}`"))) - .map(|v| json!(v.into_inner())) - } - Value::String(y) if y.starts_with("$res:") => { - let path = y.strip_prefix("$res:").unwrap(); - if path.split("/").count() < 2 { - return Err(Error::InternalErr(format!( - "Argument `{name}` is an invalid resource path: {path}", - ))); - } - Ok(client.get_client() - .get_resource_value_interpolated(workspace, path) - .await - .map_err(|_| Error::NotFound(format!("Resource {path} not found for `{name}`")))? - .into_inner()) - } - Value::Object(mut m) => { - for (a, b) in m.clone().into_iter() { - m.insert( - a.clone(), - transform_json_value(&a, client, workspace, b).await?, - ); - } - Ok(Value::Object(m)) - } - a @ _ => Ok(a), - } -} - #[tracing::instrument(level = "trace", skip_all)] async fn handle_code_execution_job( job: &QueuedJob, @@ -1541,24 +1479,6 @@ mount {{ } -#[tracing::instrument(level = "trace", skip_all)] -pub async fn create_args_and_out_file( - client: &AuthedClient, - job: &QueuedJob, - job_dir: &str, -) -> Result<(), Error> { - let args = if let Some(args) = &job.args { - Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) - } else { - None - }; - let ser_args = serde_json::to_string(&args).map_err(|e| Error::ExecutionErr(e.to_string()))?; - write_file(job_dir, "args.json", &ser_args).await?; - write_file(job_dir, "result.json", "").await?; - Ok(()) -} - - #[tracing::instrument(level = "trace", skip_all)]