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

fix: cache hub scripts in more cases + pre-cache hub scripts deps in deno

Browse Source
This commit is contained in:
Ruben Fiszel
2024-08-27 02:30:54 +02:00
parent 61a7093ac3
commit 16465e47c8
17 changed files with 192 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+1
View File
@@ -10630,6 +10630,7 @@ dependencies = [
"axum",
"bytes",
"chrono",
"const_format",
"cron",
"futures-core",
"git-version",
backend/src/main.rs
+24 -4
View File
@@ -20,6 +20,7 @@ use tokio::{
fs::{create_dir_all, DirBuilder, File},
io::AsyncReadExt,
};
use uuid::Uuid;
use windmill_api::HTTP_CLIENT;
#[cfg(feature = "enterprise")]
@@ -36,9 +37,10 @@ use windmill_common::{
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
},
scripts::ScriptLang,
stats_ee::schedule_stats,
utils::{rd_string, Mode},
worker::{reload_custom_tags_setting, WORKER_GROUP},
worker::{reload_custom_tags_setting, HUB_CACHE_DIR, TMP_DIR, WORKER_GROUP},
DB, METRICS_ENABLED,
};
@@ -61,8 +63,8 @@ use windmill_common::global_settings::OBJECT_STORE_CACHE_CONFIG_SETTING;
use windmill_worker::{
get_hub_script_content_and_requirements, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR,
BUN_DEPSTAR_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM,
GO_BIN_CACHE_DIR, GO_CACHE_DIR, HUB_CACHE_DIR, LOCK_CACHE_DIR, PIP_CACHE_DIR,
POWERSHELL_CACHE_DIR, TAR_PIP_CACHE_DIR, TMP_LOGS_DIR,
GO_BIN_CACHE_DIR, GO_CACHE_DIR, LOCK_CACHE_DIR, PIP_CACHE_DIR, POWERSHELL_CACHE_DIR,
TAR_PIP_CACHE_DIR, TMP_LOGS_DIR,
};
use crate::monitor::{
@@ -134,7 +136,25 @@ async fn cache_hub_scripts(file_path: Option<String>) -> anyhow::Result<()> {
create_dir_all(HUB_CACHE_DIR).await?;
for path in paths.values() {
get_hub_script_content_and_requirements(Some(path.to_string()), None).await?;
tracing::info!("Caching hub script at {path}");
let res = get_hub_script_content_and_requirements(Some(path.to_string()), None).await?;
if res.language.is_some_and(|x| x == ScriptLang::Deno) {
let job_dir = format!("{}/cache_init/{}", TMP_DIR, Uuid::new_v4());
create_dir_all(&job_dir).await?;
let _ = windmill_worker::generate_deno_lock(
&Uuid::nil(),
&res.content,
&mut 0,
&mut None,
"/tmp/windmill/",
None,
"global",
"global",
"",
)
.await?;
tokio::fs::remove_dir_all(job_dir).await?;
}
}
Ok(())
}
backend/windmill-common/Cargo.toml
+1
View File
@@ -52,6 +52,7 @@ bytes.workspace = true
mail-send.workspace = true
futures-core.workspace = true
async-stream.workspace = true
const_format.workspace = true
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemalloc-ctl = { optional = true, workspace = true }
backend/windmill-common/src/scripts.rs
+29
View File
@@ -16,6 +16,8 @@ use crate::{
utils::http_get_from_hub,
DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL,
};
use crate::worker::HUB_CACHE_DIR;
use anyhow::Context;
use serde::de::Error as _;
use serde::{ser::SerializeSeq, Deserialize, Deserializer, Serialize};
@@ -434,6 +436,33 @@ pub async fn get_full_hub_script_by_path(
.strip_prefix("hub/")
.ok_or_else(|| Error::BadRequest("Impossible to remove prefix hex".to_string()))?;
let mut path_iterator = path.split("/");
let version = path_iterator
.next()
.ok_or_else(|| Error::InternalErr(format!("expected hub path to have version number")))?;
let cache_path = format!("{HUB_CACHE_DIR}/{version}");
let script;
if tokio::fs::metadata(&cache_path).await.is_err() {
script = get_full_hub_script_by_path_inner(path, http_client, db).await?;
crate::worker::write_file(
HUB_CACHE_DIR,
&version,
&serde_json::to_string(&script).map_err(to_anyhow)?,
)?;
tracing::info!("wrote hub script {path} to cache");
} else {
let cache_content = tokio::fs::read_to_string(cache_path).await?;
script = serde_json::from_str(&cache_content).unwrap();
tracing::info!("read hub script {path} from cache");
}
Ok(script)
}
async fn get_full_hub_script_by_path_inner(
path: &str,
http_client: &reqwest::Client,
db: Option<&DB>,
) -> crate::error::Result<HubScript> {
let hub_base_url = HUB_BASE_URL.read().await.clone();
let result = http_get_from_hub(
backend/windmill-common/src/worker.rs
+14
View File
@@ -1,3 +1,4 @@
use const_format::concatcp;
use itertools::Itertools;
use regex::Regex;
use serde::{Deserialize, Serialize};
@@ -5,6 +6,8 @@ use serde_json::value::RawValue;
use std::{
cmp::Reverse,
collections::{HashMap, HashSet},
fs::File,
io::Write,
path::Path,
str::FromStr,
sync::{atomic::AtomicBool, Arc},
@@ -151,6 +154,17 @@ pub async fn make_pull_query(wc: &WorkerConfig) {
}
pub const TMP_DIR: &str = "/tmp/windmill";
pub const HUB_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "hub");
pub const ROOT_CACHE_DIR: &str = concatcp!(TMP_DIR, "/cache/");
pub fn write_file(dir: &str, path: &str, content: &str) -> error::Result<File> {
let path = format!("{}/{}", dir, path);
let mut file = File::create(&path)?;
file.write_all(content.as_bytes())?;
file.flush()?;
Ok(file)
}
pub async fn reload_custom_tags_setting(db: &DB) -> error::Result<()> {
let q = sqlx::query!(
backend/windmill-worker/src/bash_executor.rs
+18 -18
View File
@@ -4,7 +4,11 @@ use regex::Regex;
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_common::{
error::Error,
jobs::QueuedJob,
worker::{to_raw_value, write_file},
};
use windmill_queue::{append_logs, CanceledBy};
const BIN_BASH: &str = "/bin/bash";
@@ -19,7 +23,7 @@ lazy_static::lazy_static! {
use crate::{
common::{
build_args_map, get_reserved_variables, handle_child, read_file, read_file_content,
start_child_process, write_file,
start_child_process,
},
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV,
POWERSHELL_CACHE_DIR, POWERSHELL_PATH, TZ_ENV,
@@ -47,13 +51,12 @@ pub async fn handle_bash_job(
let logs1 = "\n\n--- BASH CODE EXECUTION ---\n".to_string();
append_logs(&job.id, &job.workspace_id, 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}"))?;
write_file(
job_dir,
"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?;
@@ -76,9 +79,9 @@ pub async fn handle_bash_job(
})
.collect::<Vec<String>>();
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 _ = write_file(job_dir, "result.json", "")?;
let _ = write_file(job_dir, "result.out", "")?;
let _ = write_file(job_dir, "result2.out", "")?;
let child = if !*DISABLE_NSJAIL {
let _ = write_file(
@@ -88,8 +91,7 @@ pub async fn handle_bash_job(
.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",
@@ -302,20 +304,19 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
format!("{}\n{}", profile, content)
};
write_file(job_dir, "main.ps1", content.as_str()).await?;
write_file(job_dir, "main.ps1", content.as_str())?;
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 _ = write_file(job_dir, "result.json", "")?;
let _ = write_file(job_dir, "result.out", "")?;
let _ = write_file(job_dir, "result2.out", "")?;
let child = if !*DISABLE_NSJAIL {
let _ = write_file(
@@ -326,8 +327,7 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string())
.replace("{SHARED_MOUNT}", shared_mount)
.replace("{CACHE_DIR}", POWERSHELL_CACHE_DIR),
)
.await?;
)?;
let mut cmd_args = vec![
"--config",
"run.config.proto",
backend/windmill-worker/src/bun_executor.rs
+23 -30
View File
@@ -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_file_content, read_result, start_child_process, write_file,
write_file_binary,
parse_npm_config, read_file_content, read_result, start_child_process, write_file_binary,
},
AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR,
BUN_DEPSTAR_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NODE_BIN_PATH,
@@ -37,7 +36,7 @@ use windmill_common::{
get_latest_hash_for_path,
jobs::QueuedJob,
scripts::ScriptLang,
worker::{exists_in_cache, get_annotation, save_cache},
worker::{exists_in_cache, get_annotation, save_cache, write_file},
DB,
};
@@ -77,7 +76,7 @@ pub async fn gen_lockfile(
if let Some(raw_deps) = raw_deps {
gen_bunfig(job_dir).await?;
write_file(job_dir, "package.json", raw_deps.as_str()).await?;
write_file(job_dir, "package.json", raw_deps.as_str())?;
} else {
let _ = write_file(
&job_dir,
@@ -98,8 +97,7 @@ pub async fn gen_lockfile(
)
.replace("RAW_GET_ENDPOINT", "raw")
),
)
.await?;
)?;
gen_bunfig(job_dir).await?;
@@ -210,7 +208,7 @@ registry = {}
.unwrap_or("".to_string())
);
tracing::debug!("Writing following bunfig.toml: {bunfig_toml}");
let _ = write_file(&job_dir, "bunfig.toml", &bunfig_toml).await?;
let _ = write_file(&job_dir, "bunfig.toml", &bunfig_toml)?;
}
Ok(())
}
@@ -260,7 +258,7 @@ pub async fn install_lockfile(
));
child_cmd.env("NPM_CONFIG_REGISTRY", custom_registry);
write_file(job_dir, ".npmrc", content).await?;
write_file(job_dir, ".npmrc", content)?;
true
} else {
false
@@ -356,8 +354,7 @@ if (!bo.success) {{
"#,
loader
),
)
.await?;
)?;
} else if mode == LoaderMode::Bun {
write_file(
&job_dir,
@@ -372,8 +369,7 @@ plugin(p)
"#,
loader
),
)
.await?;
)?;
} else if mode == LoaderMode::BunBundle
|| mode == LoaderMode::NodeBundle
|| mode == LoaderMode::BrowserBundle
@@ -412,8 +408,7 @@ if (!bo.success) {{
"browser"
}
),
)
.await?;
)?;
}
Ok(())
}
@@ -608,7 +603,7 @@ pub async fn prebundle_script(
return Ok(());
}
let origin = format!("{job_dir}/main.js");
write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?).await?;
write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?)?;
build_loader(
job_dir,
base_internal_url,
@@ -730,9 +725,9 @@ pub async fn handle_bun_job(
};
if !codebase.is_some() && !has_bundle_cache {
let _ = write_file(job_dir, "main.ts", inner_content).await?;
let _ = write_file(job_dir, "main.ts", inner_content)?;
} else if !annotation.native_mode && codebase.is_none() {
let _ = write_file(job_dir, "package.json", r#"{ "type": "module" }"#).await?;
let _ = write_file(job_dir, "package.json", r#"{ "type": "module" }"#)?;
};
let common_bun_proc_envs: HashMap<String, String> =
@@ -768,7 +763,7 @@ pub async fn handle_bun_job(
));
}
let _ = write_file(job_dir, "package.json", &splitted[0]).await?;
let _ = write_file(job_dir, "package.json", &splitted[0])?;
let lockb = if annotation.npm_mode { "" } else { splitted[1] };
if lockb != EMPTY_FILE {
let mut skip_install = false;
@@ -879,10 +874,10 @@ pub async fn handle_bun_job(
} else if annotation.native_mode {
"\n\n--- NATIVE CODE EXECUTION ---\n".to_string()
} else if annotation.nodejs_mode {
write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?).await?;
write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?)?;
"\n\n--- NODE CODE EXECUTION ---\n".to_string()
} else {
write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?).await?;
write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?)?;
"\n\n--- BUN CODE EXECUTION ---\n".to_string()
};
@@ -963,7 +958,7 @@ try {{
}}
"#,
);
write_file(job_dir, "wrapper.mjs", &wrapper_content).await?;
write_file(job_dir, "wrapper.mjs", &wrapper_content)?;
Ok(()) as error::Result<()>
};
@@ -1068,9 +1063,8 @@ try {{
"import * as Main from \"./main.ts\"",
"import * as Main from \"./main.js\"",
),
)
.await?;
write_file(job_dir, "package.json", r#"{ "type": "module" }"#).await?;
)?;
write_file(job_dir, "package.json", r#"{ "type": "module" }"#)?;
}
fs::remove_file(format!("{job_dir}/main.ts"))?;
has_bundle_cache = true;
@@ -1157,8 +1151,7 @@ try {{
},
),
),
)
.await?;
)?;
let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str());
let args = if annotation.nodejs_mode {
@@ -1329,7 +1322,7 @@ pub async fn start_worker(
let mut canceled_by: Option<CanceledBy> = None;
tracing::info!("Starting worker {w_id};{script_path} (codebase: {codebase:?}");
if !codebase.is_some() {
let _ = write_file(job_dir, "main.ts", inner_content).await?;
let _ = write_file(job_dir, "main.ts", inner_content)?;
}
let common_bun_proc_envs: HashMap<String, String> =
@@ -1369,7 +1362,7 @@ pub async fn start_worker(
format!("Invalid requirements, expected to find //bun.lockb split pattern in reqs. Found: |{reqs}|")
));
}
let _ = write_file(job_dir, "package.json", &splitted[0]).await?;
let _ = write_file(job_dir, "package.json", &splitted[0])?;
let lockb = splitted[1];
if lockb != EMPTY_FILE {
let _ = write_file_binary(
@@ -1418,7 +1411,7 @@ pub async fn start_worker(
}
let main_code = remove_pinned_imports(inner_content)?;
let _ = write_file(job_dir, "main.ts", &main_code).await?;
let _ = write_file(job_dir, "main.ts", &main_code)?;
{
// let mut start = Instant::now();
@@ -1479,7 +1472,7 @@ for await (const line of Readline.createInterface({{ input: process.stdin }})) {
}}
"#,
);
write_file(job_dir, "wrapper.mjs", &wrapper_content).await?;
write_file(job_dir, "wrapper.mjs", &wrapper_content)?;
}
if !codebase.is_some() {
backend/windmill-worker/src/common.rs
+7 -18
View File
@@ -28,7 +28,8 @@ use windmill_common::s3_helpers::{
};
use windmill_common::variables::{build_crypt_with_key_suffix, decrypt_value_with_mc};
use windmill_common::worker::{
get_windmill_memory_usage, get_worker_memory_usage, CLOUD_HOSTED, TMP_DIR, WORKER_CONFIG,
get_windmill_memory_usage, get_worker_memory_usage, write_file, CLOUD_HOSTED, ROOT_CACHE_DIR,
TMP_DIR, WORKER_CONFIG,
};
use windmill_common::{
error::{self, Error},
@@ -73,7 +74,7 @@ use futures::{
use crate::{
AuthedClient, AuthedClientBackgroundTask, JOB_DEFAULT_TIMEOUT, MAX_RESULT_SIZE,
MAX_TIMEOUT_DURATION, MAX_WAIT_FOR_SIGINT, MAX_WAIT_FOR_SIGTERM, ROOT_CACHE_DIR,
MAX_TIMEOUT_DURATION, MAX_WAIT_FOR_SIGINT, MAX_WAIT_FOR_SIGTERM,
};
pub async fn build_args_map<'a>(
@@ -112,34 +113,22 @@ pub async fn create_args_and_out_file(
job_dir,
"args.json",
&serde_json::to_string(&x).unwrap_or_else(|_| "{}".to_string()),
)
.await?;
)?;
} else {
write_file(
job_dir,
"args.json",
&serde_json::to_string(&args).unwrap_or_else(|_| "{}".to_string()),
)
.await?;
)?;
}
} else {
write_file(job_dir, "args.json", "{}").await?;
write_file(job_dir, "args.json", "{}")?;
};
write_file(job_dir, "result.json", "").await?;
write_file(job_dir, "result.json", "")?;
Ok(())
}
#[tracing::instrument(level = "trace", skip_all)]
pub async fn write_file(dir: &str, path: &str, content: &str) -> error::Result<File> {
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<File> {
let path = format!("{}/{}", dir, path);
let mut file = File::create(&path).await?;
backend/windmill-worker/src/deno_executor.rs
+34 -34
View File
@@ -8,13 +8,13 @@ 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, start_child_process, write_file,
parse_npm_config, read_result, start_child_process,
},
AuthedClientBackgroundTask, DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV,
NPM_CONFIG_REGISTRY, PATH_ENV, TZ_ENV,
};
use tokio::{fs::File, io::AsyncReadExt, process::Command};
use windmill_common::{error::Result, BASE_URL};
use windmill_common::{error::Result, worker::write_file, BASE_URL};
use windmill_common::{
error::{self},
jobs::QueuedJob,
@@ -88,12 +88,12 @@ pub async fn generate_deno_lock(
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
db: Option<&sqlx::Pool<sqlx::Postgres>>,
w_id: &str,
worker_name: &str,
base_internal_url: &str,
) -> error::Result<String> {
let _ = write_file(job_dir, "main.ts", code).await?;
let _ = write_file(job_dir, "main.ts", code)?;
let import_map_path = format!("{job_dir}/import_map.json");
let import_map = format!(
@@ -103,14 +103,11 @@ pub async fn generate_deno_lock(
}}
}}"#,
);
write_file(job_dir, "import_map.json", &import_map).await?;
write_file(job_dir, "empty.ts", "").await?;
write_file(job_dir, "import_map.json", &import_map)?;
write_file(job_dir, "empty.ts", "")?;
let deno_envs = get_common_deno_proc_envs("", base_internal_url).await;
let mut deno_envs = HashMap::new();
if let Some(ref s) = NPM_CONFIG_REGISTRY.read().await.clone() {
let (url, _token_opt) = parse_npm_config(s);
deno_envs.insert(String::from("NPM_CONFIG_REGISTRY"), url);
}
let mut child_cmd = Command::new(DENO_PATH.as_str());
child_cmd
.current_dir(job_dir)
@@ -132,22 +129,26 @@ pub async fn generate_deno_lock(
.envs(deno_envs)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let child_process = start_child_process(child_cmd, DENO_PATH.as_str()).await?;
let mut child_process = start_child_process(child_cmd, DENO_PATH.as_str()).await?;
handle_child(
job_id,
db,
mem_peak,
canceled_by,
child_process,
false,
worker_name,
w_id,
"deno cache",
None,
false,
)
.await?;
if let Some(db) = db {
handle_child(
job_id,
db,
mem_peak,
canceled_by,
child_process,
false,
worker_name,
w_id,
"deno cache",
None,
false,
)
.await?;
} else {
child_process.wait().await?;
}
let path_lock = format!("{job_dir}/lock.json");
let mut file = File::open(path_lock).await?;
@@ -176,7 +177,7 @@ pub async fn handle_deno_job(
let main_override = get_main_override(job.args.as_ref());
let write_main_f = write_file(job_dir, "main.ts", inner_content);
write_file(job_dir, "main.ts", inner_content)?;
let write_wrapper_f = async {
// let mut start = Instant::now();
@@ -231,7 +232,7 @@ try {{
}}
"#,
);
write_file(job_dir, "wrapper.ts", &wrapper_content).await?;
write_file(job_dir, "wrapper.ts", &wrapper_content)?;
Ok(()) as error::Result<()>
};
@@ -256,9 +257,8 @@ try {{
Ok(reserved_variables) as error::Result<(HashMap<String, String>, String)>
};
let ((reserved_variables, token), _, _, _) = tokio::try_join!(
let ((reserved_variables, token), _, _) = tokio::try_join!(
reserved_variables_args_out_f,
write_main_f,
write_wrapper_f,
write_import_map_f
)?;
@@ -288,7 +288,7 @@ try {{
args.push("--unstable-http");
if let Some(reqs) = requirements_o {
if !reqs.is_empty() {
let _ = write_file(job_dir, "lock.json", &reqs).await?;
let _ = write_file(job_dir, "lock.json", &reqs)?;
args.push("--lock=lock.json");
args.push("--lock-write");
}
@@ -388,7 +388,7 @@ async fn build_import_map(
}}
}}"#,
);
write_file(job_dir, "import_map.json", &import_map).await?;
write_file(job_dir, "import_map.json", &import_map)?;
Ok(()) as error::Result<()>
}
@@ -417,7 +417,7 @@ pub async fn start_worker(
use crate::common::build_envs_map;
let _ = write_file(job_dir, "main.ts", inner_content).await?;
let _ = write_file(job_dir, "main.ts", inner_content)?;
let common_deno_proc_envs = get_common_deno_proc_envs(&token, base_internal_url).await;
let context = variables::get_reserved_variables(
@@ -494,7 +494,7 @@ for await (const chunk of Deno.stdin.readable) {{
}}
"#,
);
write_file(job_dir, "wrapper.ts", &wrapper_content).await?;
write_file(job_dir, "wrapper.ts", &wrapper_content)?;
}
build_import_map(w_id, script_path, base_internal_url, job_dir).await?;
backend/windmill-worker/src/go_executor.rs
+9 -10
View File
@@ -12,7 +12,7 @@ use windmill_common::{
error::{self, Error},
jobs::QueuedJob,
utils::calculate_hash,
worker::save_cache,
worker::{save_cache, write_file},
};
use windmill_parser_go::{parse_go_imports, REQUIRE_PARSE};
use windmill_queue::{append_logs, CanceledBy};
@@ -20,7 +20,7 @@ use windmill_queue::{append_logs, CanceledBy};
use crate::{
common::{
capitalize, create_args_and_out_file, get_reserved_variables, handle_child, read_result,
start_child_process, write_file,
start_child_process,
},
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, GOPRIVATE, GOPROXY,
GO_BIN_CACHE_DIR, GO_CACHE_DIR, HOME_ENV, NSJAIL_PATH, PATH_ENV, TZ_ENV,
@@ -141,7 +141,7 @@ func main() {{
}}
}}"#;
write_file(job_dir, "main.go", WRAPPER_CONTENT).await?;
write_file(job_dir, "main.go", WRAPPER_CONTENT)?;
{
let spread = &sig
@@ -174,7 +174,7 @@ func Run(req Req) (interface{{}}, error){{
"#,
);
write_file(&format!("{job_dir}/inner"), "runner.go", &runner_content).await?;
write_file(&format!("{job_dir}/inner"), "runner.go", &runner_content)?;
}
}
@@ -247,8 +247,7 @@ func Run(req Req) (interface{{}}, error){{
.replace("{CACHE_DIR}", GO_CACHE_DIR)
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string())
.replace("{SHARED_MOUNT}", shared_mount),
)
.await?;
)?;
let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str());
nsjail_cmd
.current_dir(job_dir)
@@ -313,11 +312,11 @@ async fn gen_go_mod(
let md = requirements.split_once(GO_REQ_SPLITTER);
if let Some((req, sum)) = md {
write_file(job_dir, "go.mod", &req).await?;
write_file(job_dir, "go.sum", &sum).await?;
write_file(job_dir, "go.mod", &req)?;
write_file(job_dir, "go.sum", &sum)?;
Ok((true, true))
} else {
write_file(job_dir, "go.mod", &requirements).await?;
write_file(job_dir, "go.mod", &requirements)?;
Ok((true, false))
}
}
@@ -471,7 +470,7 @@ async fn gen_go_mymod(code: &str, job_dir: &str) -> error::Result<()> {
.await
.expect("could not create go's mymod dir");
write_file(&mymod_dir, "inner_main.go", &code).await?;
write_file(&mymod_dir, "inner_main.go", &code)?;
Ok(())
}
backend/windmill-worker/src/lib.rs
+2
View File
@@ -24,3 +24,5 @@ mod worker;
mod worker_flow;
mod worker_lockfiles;
pub use worker::*;
pub use deno_executor::generate_deno_lock;
backend/windmill-worker/src/php_executor.rs
+7 -7
View File
@@ -8,6 +8,7 @@ use uuid::Uuid;
use windmill_common::{
error::{self, to_anyhow, Result},
jobs::QueuedJob,
worker::write_file,
};
use windmill_parser::Typ;
use windmill_queue::{append_logs, CanceledBy};
@@ -15,7 +16,7 @@ use windmill_queue::{append_logs, CanceledBy};
use crate::{
common::{
create_args_and_out_file, get_main_override, get_reserved_variables, handle_child,
read_result, start_child_process, write_file,
read_result, start_child_process,
},
AuthedClientBackgroundTask, COMPOSER_CACHE_DIR, COMPOSER_PATH, DISABLE_NSJAIL, DISABLE_NUSER,
NSJAIL_PATH, PHP_PATH,
@@ -70,10 +71,10 @@ pub async fn composer_install(
requirements: String,
lock: Option<String>,
) -> Result<String> {
write_file(job_dir, "composer.json", &requirements).await?;
write_file(job_dir, "composer.json", &requirements)?;
if let Some(lock) = lock.as_ref() {
write_file(job_dir, "composer.lock", lock).await?;
write_file(job_dir, "composer.lock", lock)?;
}
let mut child_cmd = Command::new(&*COMPOSER_PATH);
@@ -178,7 +179,7 @@ pub async fn handle_php_job(
append_logs(&job.id, job.workspace_id.to_string(), init_logs, db).await;
let _ = write_file(job_dir, "main.php", inner_content).await?;
let _ = write_file(job_dir, "main.php", inner_content)?;
let main_override = get_main_override(job.args.as_ref());
@@ -241,7 +242,7 @@ try {{
}}
"#,
);
write_file(job_dir, "wrapper.php", &wrapper_content).await?;
write_file(job_dir, "wrapper.php", &wrapper_content)?;
Ok(()) as error::Result<()>
};
@@ -269,8 +270,7 @@ try {{
.replace("{JOB_DIR}", job_dir)
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string())
.replace("{SHARED_MOUNT}", shared_mount),
)
.await?;
)?;
let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str());
let args = vec![
backend/windmill-worker/src/python_executor.rs
+9 -11
View File
@@ -16,7 +16,7 @@ use windmill_common::{
error::{self, Error},
jobs::QueuedJob,
utils::calculate_hash,
worker::WORKER_CONFIG,
worker::{write_file, WORKER_CONFIG},
DB,
};
@@ -56,7 +56,7 @@ use windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS;
use crate::{
common::{
create_args_and_out_file, get_main_override, get_reserved_variables, handle_child,
read_result, start_child_process, write_file,
read_result, start_child_process,
},
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,
@@ -150,7 +150,7 @@ pub async fn pip_compile(
}
let file = "requirements.in";
write_file(job_dir, file, &requirements).await?;
write_file(job_dir, file, &requirements)?;
let mut args = vec![
"-q",
@@ -352,7 +352,7 @@ except BaseException as e:
sys.exit(1)
"#,
);
write_file(job_dir, "wrapper.py", &wrapper_content).await?;
write_file(job_dir, "wrapper.py", &wrapper_content)?;
let client = client.get_authed().await;
let mut reserved_variables = get_reserved_variables(job, &client.token, db).await?;
@@ -387,8 +387,7 @@ mount {{
"{ADDITIONAL_PYTHON_PATHS}",
additional_python_paths_folders.as_str(),
),
)
.await?;
)?;
} else {
reserved_variables.insert("PYTHONPATH".to_string(), additional_python_paths_folders);
}
@@ -506,9 +505,9 @@ async fn prepare_wrapper(
let module_dir = format!("{}/{}", job_dir, dirs);
tokio::fs::create_dir_all(format!("{module_dir}/")).await?;
let _ = write_file(&module_dir, &format!("{last}.py"), inner_content).await?;
let _ = write_file(&module_dir, &format!("{last}.py"), inner_content)?;
if relative_imports {
let _ = write_file(job_dir, "loader.py", RELATIVE_PYTHON_LOADER).await?;
let _ = write_file(job_dir, "loader.py", RELATIVE_PYTHON_LOADER)?;
}
let sig = windmill_parser_py::parse_python_signature(inner_content, main_override.clone())?;
@@ -779,8 +778,7 @@ pub async fn handle_python_reqs(
.replace("{WORKER_DIR}", &worker_dir)
.replace("{CACHE_DIR}", PIP_CACHE_DIR)
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()),
)
.await?;
)?;
};
let mut req_with_penv: Vec<(String, String)> = vec![];
@@ -1153,7 +1151,7 @@ for line in sys.stdin:
sys.stdout.flush()
"#,
);
write_file(job_dir, "wrapper.py", &wrapper_content).await?;
write_file(job_dir, "wrapper.py", &wrapper_content)?;
}
let reserved_variables = windmill_common::variables::get_reserved_variables(
backend/windmill-worker/src/worker.rs
+5 -28
View File
@@ -7,7 +7,7 @@
*/
use windmill_common::{
auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms, JWT_SECRET}, scripts::PREVIEW_IS_TAR_CODEBASE_HASH, worker::{get_memory, get_vcpus, get_windmill_memory_usage, get_worker_memory_usage, TMP_DIR}
auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms, JWT_SECRET}, scripts::PREVIEW_IS_TAR_CODEBASE_HASH, worker::{get_memory, get_vcpus, get_windmill_memory_usage, get_worker_memory_usage, write_file, ROOT_CACHE_DIR, TMP_DIR}
};
use anyhow::{Context, Result};
@@ -86,7 +86,7 @@ use crate::{
bun_executor::handle_bun_job,
common::{
build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args,
read_result, save_in_cache, write_file, NO_LOGS_AT_ALL, SLOW_LOGS,
read_result, save_in_cache, NO_LOGS_AT_ALL, SLOW_LOGS,
},
deno_executor::handle_deno_job,
go_executor::handle_go_job,
@@ -224,7 +224,6 @@ pub async fn create_token_for_owner(
pub const TMP_LOGS_DIR: &str = concatcp!(TMP_DIR, "/logs");
pub const ROOT_CACHE_DIR: &str = concatcp!(TMP_DIR, "/cache/");
pub const ROOT_CACHE_NOMOUNT_DIR: &str = concatcp!(TMP_DIR, "/cache_nomount/");
pub const LOCK_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "lock");
@@ -239,7 +238,6 @@ pub const BUN_CACHE_DIR: &str = concatcp!(ROOT_CACHE_NOMOUNT_DIR, "bun");
pub const BUN_BUNDLE_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "bun");
pub const BUN_DEPSTAR_CACHE_DIR: &str = concatcp!(ROOT_CACHE_NOMOUNT_DIR, "buntar");
pub const HUB_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "hub");
pub const GO_BIN_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "gobin");
pub const POWERSHELL_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "powershell");
pub const COMPOSER_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "composer");
@@ -780,7 +778,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
if let Some(ref netrc) = *NETRC {
tracing::info!("Writing netrc at {}/.netrc", HOME_ENV.as_str());
write_file(&HOME_ENV, ".netrc", netrc)
.await
.expect("could not write netrc");
}
@@ -795,8 +792,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
&worker_dir,
"download_deps.py.sh",
INCLUDE_DEPS_PY_SH_CONTENT,
)
.await;
);
}
let mut last_ping = Instant::now() - Duration::from_secs(NUM_SECS_PING + 1);
@@ -2743,28 +2739,9 @@ pub async fn get_hub_script_content_and_requirements(
let script_path = script_path
.clone()
.ok_or_else(|| Error::InternalErr(format!("expected script path for hub script")))?;
let mut script_path_iterator = script_path.split("/");
script_path_iterator.next();
let version = script_path_iterator
.next()
.ok_or_else(|| Error::InternalErr(format!("expected hub path to have version number")))?;
let cache_path = format!("{HUB_CACHE_DIR}/{version}");
let script;
if tokio::fs::metadata(&cache_path).await.is_err() {
script = get_full_hub_script_by_path(StripPath(script_path.to_string()), &HTTP_CLIENT, db)
let script = get_full_hub_script_by_path(StripPath(script_path.to_string()), &HTTP_CLIENT, db)
.await?;
write_file(
HUB_CACHE_DIR,
&version,
&serde_json::to_string(&script).map_err(to_anyhow)?,
)
.await?;
tracing::info!("wrote hub script {script_path} to cache");
} else {
let cache_content = tokio::fs::read_to_string(cache_path).await?;
script = serde_json::from_str(&cache_content).unwrap();
tracing::info!("read hub script {script_path} from cache");
}
Ok(ContentReqLangEnvs {
content: script.content,
lockfile: script.lockfile,
backend/windmill-worker/src/worker_lockfiles.rs
+3 -4
View File
@@ -12,7 +12,7 @@ use windmill_common::flows::{FlowModule, FlowModuleValue};
use windmill_common::get_latest_deployed_hash_for_path;
use windmill_common::jobs::JobPayload;
use windmill_common::scripts::ScriptHash;
use windmill_common::worker::{get_annotation, to_raw_value, to_raw_value_owned};
use windmill_common::worker::{get_annotation, to_raw_value, to_raw_value_owned, write_file};
use windmill_common::{
error::{self, to_anyhow},
flows::FlowValue,
@@ -28,7 +28,6 @@ use windmill_queue::{append_logs, CanceledBy, PushIsolationLevel};
use crate::python_executor::{create_dependencies_dir, handle_python_reqs, pip_compile};
use crate::{
bun_executor::gen_lockfile,
common::write_file,
deno_executor::generate_deno_lock,
go_executor::install_go_dependencies,
php_executor::{composer_install, parse_php_imports},
@@ -1328,7 +1327,7 @@ async fn capture_dependency_job(
mem_peak,
canceled_by,
job_dir,
db,
Some(db),
w_id,
worker_name,
base_internal_url,
@@ -1339,7 +1338,7 @@ async fn capture_dependency_job(
let npm_mode = npm_mode
.unwrap_or_else(|| windmill_common::worker::get_annotation(job_raw_code).npm_mode);
if !raw_deps {
let _ = write_file(job_dir, "main.ts", job_raw_code).await?;
let _ = write_file(job_dir, "main.ts", job_raw_code)?;
}
let req = gen_lockfile(
mem_peak,
frontend/src/lib/components/FlowStatusViewerInner.svelte
+5 -2
View File
@@ -938,6 +938,7 @@
if (rightColumnSelect != 'node_definition') {
rightColumnSelect = 'node_status'
}
console.log('BAR', e.detail)
if (typeof e.detail == 'string') {
if (e.detail == 'Input') {
selectedNode = 'start'
@@ -946,8 +947,10 @@
selectedNode = 'end'
stepDetail = 'end'
} else {
selectedNode = e.detail
stepDetail = e.detail
const mod = dfs(job?.raw_flow?.modules ?? [], (m) => m).find(
(m) => m?.id === e?.detail
)
stepDetail = mod
}
} else {
stepDetail = e.detail
frontend/src/lib/components/flows/pickers/PickHubScript.svelte
+1
View File
@@ -96,6 +96,7 @@
summary: `${x.summary} (${x.app})`
})
)
hubNotAvailable = false
} catch (err) {
hubNotAvailable = true
console.error('Hub not available')