From 16465e47c8acf6320694a821b375aa20ee51067c Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 27 Aug 2024 02:30:54 +0200 Subject: [PATCH] fix: cache hub scripts in more cases + pre-cache hub scripts deps in deno --- backend/Cargo.lock | 1 + backend/src/main.rs | 28 ++++++-- backend/windmill-common/Cargo.toml | 1 + backend/windmill-common/src/scripts.rs | 29 ++++++++ backend/windmill-common/src/worker.rs | 14 ++++ backend/windmill-worker/src/bash_executor.rs | 36 +++++----- backend/windmill-worker/src/bun_executor.rs | 53 +++++++-------- backend/windmill-worker/src/common.rs | 25 ++----- backend/windmill-worker/src/deno_executor.rs | 68 +++++++++---------- backend/windmill-worker/src/go_executor.rs | 19 +++--- backend/windmill-worker/src/lib.rs | 2 + backend/windmill-worker/src/php_executor.rs | 14 ++-- .../windmill-worker/src/python_executor.rs | 20 +++--- backend/windmill-worker/src/worker.rs | 33 ++------- .../windmill-worker/src/worker_lockfiles.rs | 7 +- .../components/FlowStatusViewerInner.svelte | 7 +- .../flows/pickers/PickHubScript.svelte | 1 + 17 files changed, 192 insertions(+), 166 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 757546715b..48a3c640e0 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -10630,6 +10630,7 @@ dependencies = [ "axum", "bytes", "chrono", + "const_format", "cron", "futures-core", "git-version", diff --git a/backend/src/main.rs b/backend/src/main.rs index ebc7f6dc3b..dc753f0aa4 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -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) -> 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(()) } diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index e7cd9d1432..c324dcf728 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -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 } diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index b10558e40f..d32ee3f720 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -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 { let hub_base_url = HUB_BASE_URL.read().await.clone(); let result = http_get_from_hub( diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 9f09852f43..263f764e67 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -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 { + 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!( diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 7d1ecb2ca0..10879e2a64 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -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::>(); let args = args_owned.iter().map(|s| &s[..]).collect::>(); - 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", diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index c7ff297ff1..6a67a79b3e 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -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 = @@ -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 = 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 = @@ -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() { diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index cfd00020b8..2d36a24742 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -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 { - 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?; diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 54462644de..1e3a2e77b6 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -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, job_dir: &str, - db: &sqlx::Pool, + db: Option<&sqlx::Pool>, w_id: &str, worker_name: &str, base_internal_url: &str, ) -> error::Result { - 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)> }; - 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?; diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 3fe964a211..da0c7a2e5a 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -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(()) } diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index ede73f08ec..336b2e5cf9 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -24,3 +24,5 @@ mod worker; mod worker_flow; mod worker_lockfiles; pub use worker::*; + +pub use deno_executor::generate_deno_lock; diff --git a/backend/windmill-worker/src/php_executor.rs b/backend/windmill-worker/src/php_executor.rs index 1093488341..c41af02dfd 100644 --- a/backend/windmill-worker/src/php_executor.rs +++ b/backend/windmill-worker/src/php_executor.rs @@ -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, ) -> Result { - 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![ diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 8d33a057b3..36de523eb3 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -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( diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 6a6ccc7daf..5bac0f789f 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -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 m).find( + (m) => m?.id === e?.detail + ) + stepDetail = mod } } else { stepDetail = e.detail diff --git a/frontend/src/lib/components/flows/pickers/PickHubScript.svelte b/frontend/src/lib/components/flows/pickers/PickHubScript.svelte index 8e1732952b..ccc79a0a11 100644 --- a/frontend/src/lib/components/flows/pickers/PickHubScript.svelte +++ b/frontend/src/lib/components/flows/pickers/PickHubScript.svelte @@ -96,6 +96,7 @@ summary: `${x.summary} (${x.app})` }) ) + hubNotAvailable = false } catch (err) { hubNotAvailable = true console.error('Hub not available')