From 9095ca7b5bc2ca743bc2d96da4367bd264e23608 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 4 Mar 2025 18:05:49 +0100 Subject: [PATCH] add standalone bundle support on CE --- backend/windmill-api/src/scripts.rs | 62 ++++++------- backend/windmill-common/src/worker.rs | 2 + backend/windmill-worker/src/bun_executor.rs | 89 ++++++++++++------- backend/windmill-worker/src/global_cache.rs | 9 +- .../components/common/table/ScriptRow.svelte | 2 +- 5 files changed, 94 insertions(+), 70 deletions(-) diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 7c4810363c..320c8b147f 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -396,44 +396,46 @@ async fn create_snapshot_script( })?; uploaded = true; - let path = windmill_common::s3_helpers::bundle(&w_id, &hash); + + #[cfg(all(feature = "enterprise", feature = "parquet"))] + let object_store = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS + .read() + .await + .clone(); + + #[cfg(not(all(feature = "enterprise", feature = "parquet")))] + let object_store: Option<()> = None; if &windmill_common::utils::MODE_AND_ADDONS.mode == &windmill_common::utils::Mode::Standalone + && object_store.is_none() { - std::fs::create_dir_all(format!( - "{}/script_bundle/{}", - windmill_common::worker::ROOT_CACHE_NOMOUNT_DIR, - w_id - ))?; + std::fs::create_dir_all(windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR)?; windmill_common::worker::write_file( - windmill_common::worker::ROOT_CACHE_NOMOUNT_DIR, - &path, + windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR, + &hash, &String::from_utf8_lossy(&data), )?; - return Ok((StatusCode::CREATED, format!("{}", script_hash.unwrap()))); - } - - #[cfg(not(all(feature = "enterprise", feature = "parquet")))] - { - return Err(Error::ExecutionErr("codebase is an EE feature".to_string())); - } - - #[cfg(all(feature = "enterprise", feature = "parquet"))] - if let Some(os) = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS - .read() - .await - .clone() - { - if let Err(e) = os - .put(&object_store::path::Path::from(path.clone()), data.into()) - .await - { - tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e); - return Err(Error::ExecutionErr(format!("Failed to put {path} to s3"))); - } } else { - return Err(Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string())); + #[cfg(not(all(feature = "enterprise", feature = "parquet")))] + { + return Err(Error::ExecutionErr("codebase is an EE feature".to_string())); + } + + #[cfg(all(feature = "enterprise", feature = "parquet"))] + if let Some(os) = object_store { + let path = windmill_common::s3_helpers::bundle(&w_id, &hash); + + if let Err(e) = os + .put(&object_store::path::Path::from(path.clone()), data.into()) + .await + { + tracing::info!("Failed to put snapshot to s3 at {path}: {:?}", e); + return Err(Error::ExecutionErr(format!("Failed to put {path} to s3"))); + } + } else { + return Err(Error::BadConfig("Object store is required for snapshot script and is not configured for servers".to_string())); + } } } // println!("Length of `{}` is {} bytes", name, data.len()); diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 973398dcde..e0af4ea5fd 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -105,6 +105,8 @@ lazy_static::lazy_static! { } pub const ROOT_CACHE_NOMOUNT_DIR: &str = concatcp!(TMP_DIR, "/cache_nomount/"); +pub const ROOT_STANDALONE_BUNDLE_DIR: &str = concatcp!(TMP_DIR, "/standalone_bundle/"); + pub static MIN_VERSION_IS_LATEST: AtomicBool = AtomicBool::new(false); fn format_pull_query(peek: String) -> String { diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 6fc0424f11..094755cdb0 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -580,8 +580,6 @@ pub async fn generate_bun_bundle( } pub async fn pull_codebase(w_id: &str, id: &str, job_dir: &str) -> Result<()> { - use crate::global_cache::extract_tar; - let path = windmill_common::s3_helpers::bundle(&w_id, &id); let bun_cache_path = format!( "{}/{}", @@ -595,60 +593,83 @@ pub async fn pull_codebase(w_id: &str, id: &str, job_dir: &str) -> Result<()> { if is_tar { "codebase.tar" } else { "main.js" } ); - if tokio::fs::metadata(&bun_cache_path).await.is_ok() { + if std::fs::metadata(&bun_cache_path).is_ok() { tracing::info!("loading {bun_cache_path} from cache"); - if is_tar { - extract_tar(fs::read(bun_cache_path)?.into(), job_dir).await?; - } else { - #[cfg(unix)] - tokio::fs::symlink(&bun_cache_path, dst).await?; - #[cfg(windows)] - std::os::windows::fs::symlink_dir(&bun_cache_path, &dst)?; - } + extract_saved_codebase(job_dir, &bun_cache_path, is_tar, &dst, false)?; } else { + #[cfg(all(feature = "enterprise", feature = "parquet"))] + let object_store = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS + .read() + .await + .clone(); + #[cfg(not(all(feature = "enterprise", feature = "parquet")))] + let object_store: Option<()> = None; + + if &windmill_common::utils::MODE_AND_ADDONS.mode + == &windmill_common::utils::Mode::Standalone + && object_store.is_none() { - if &windmill_common::utils::MODE_AND_ADDONS.mode - == &windmill_common::utils::Mode::Standalone - { + let bun_cache_path = format!( + "{}{}", + windmill_common::worker::ROOT_STANDALONE_BUNDLE_DIR, + id + ); + if std::fs::metadata(&bun_cache_path).is_ok() { + tracing::info!("loading {bun_cache_path} from standalone bundle cache"); + extract_saved_codebase(job_dir, &bun_cache_path, is_tar, &dst, true)?; + } else { return Err(error::Error::ExecutionErr(format!( "(standalone bundle test mode) could not find codebase at {bun_cache_path}" ))); - } else { - return Err(error::Error::ExecutionErr( - "codebase is an EE feature".to_string(), - )); } + } else { + return Err(error::Error::ExecutionErr( + "codebase is an EE feature".to_string(), + )); } #[cfg(all(feature = "enterprise", feature = "parquet"))] - if let Some(os) = windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS - .read() - .await - .clone() - { + if let Some(os) = object_store { let dirs_splitted = bun_cache_path.split("/").collect_vec(); - tokio::fs::create_dir_all(dirs_splitted[..dirs_splitted.len() - 1].join("/")).await?; + std::fs::create_dir_all(dirs_splitted[..dirs_splitted.len() - 1].join("/"))?; let bytes = attempt_fetch_bytes(os, &path).await?; + tracing::info!("loading {bun_cache_path} from object store"); - tokio::fs::write(&bun_cache_path, &bytes).await?; - if is_tar { - extract_tar(bytes, job_dir).await?; - } else { - #[cfg(unix)] - tokio::fs::symlink(bun_cache_path, dst).await?; - - #[cfg(windows)] - std::os::windows::fs::symlink_dir(&bun_cache_path, &dst)?; - } + std::fs::write(&bun_cache_path, &bytes)?; + extract_saved_codebase(job_dir, &bun_cache_path, is_tar, &dst, false)?; } } Ok(()) } +fn extract_saved_codebase( + job_dir: &str, + bun_cache_path: &String, + is_tar: bool, + dst: &str, + copy: bool, +) -> Result<()> { + use crate::global_cache::extract_tar; + + Ok(if is_tar { + extract_tar(fs::read(bun_cache_path)?.into(), job_dir)?; + } else { + if copy { + std::fs::copy(bun_cache_path, dst)?; + } else { + #[cfg(unix)] + std::os::unix::fs::symlink(bun_cache_path, dst)?; + + #[cfg(windows)] + std::os::windows::fs::symlink_dir(bun_cache_path, dst)?; + } + }) +} + pub async fn prebundle_bun_script( inner_content: &str, lockfile: Option<&String>, diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index e037d7d9ec..c45bb5075a 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -93,7 +93,7 @@ pub async fn pull_from_tar( let tar_path = format!("tar/{TARGET}/{python_xyz}/{folder_name}.tar"); let bytes = attempt_fetch_bytes(client, &tar_path).await?; - extract_tar(bytes, &folder).await.map_err(|e| { + extract_tar(bytes, &folder).map_err(|e| { tracing::error!("Failed to extract piptar {folder_name}. Error: {:?}", e); e })?; @@ -106,18 +106,17 @@ pub async fn pull_from_tar( Ok(()) } -pub async fn extract_tar(tar: bytes::Bytes, folder: &str) -> error::Result<()> { +pub fn extract_tar(tar: bytes::Bytes, folder: &str) -> error::Result<()> { use bytes::Buf; - use tokio::fs::{self}; let start: Instant = Instant::now(); - fs::create_dir_all(&folder).await?; + std::fs::create_dir_all(&folder)?; let mut ar = tar::Archive::new(tar.reader()); if let Err(e) = ar.unpack(folder) { tracing::info!("Failed to untar to {folder}. Error: {:?}", e); - fs::remove_dir_all(&folder).await?; + std::fs::remove_dir_all(&folder)?; return Err(error::Error::ExecutionErr(format!( "Failed to untar tar {folder}" ))); diff --git a/frontend/src/lib/components/common/table/ScriptRow.svelte b/frontend/src/lib/components/common/table/ScriptRow.svelte index 086091f05c..63f113d5f1 100644 --- a/frontend/src/lib/components/common/table/ScriptRow.svelte +++ b/frontend/src/lib/components/common/table/ScriptRow.svelte @@ -79,7 +79,7 @@ async function deleteScript(path: string): Promise { await ScriptService.deleteScriptByPath({ workspace: $workspaceStore!, path }) dispatch('change') - sendUserToast(`Delete script ${path}`) + sendUserToast(`Deleted script ${path}`) } let scheduleEditor: ScheduleEditor