kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add standalone bundle support on CE

Browse Source
This commit is contained in:
Ruben Fiszel
2025-03-04 18:05:49 +01:00
parent ebbb1a61ec
commit 9095ca7b5b
5 changed files with 94 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
backend/windmill-api/src/scripts.rs
View File

@@ -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());

2
backend/windmill-common/src/worker.rs
View File

@@ -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 {

89
backend/windmill-worker/src/bun_executor.rs
View File

@@ -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>,

9
backend/windmill-worker/src/global_cache.rs
View File

@@ -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}"
)));

2
frontend/src/lib/components/common/table/ScriptRow.svelte
View File

@@ -79,7 +79,7 @@
async function deleteScript(path: string): Promise<void> {
await ScriptService.deleteScriptByPath({ workspace: $workspaceStore!, path })
dispatch('change')
sendUserToast(`Delete script ${path}`)
sendUserToast(`Deleted script ${path}`)
}
let scheduleEditor: ScheduleEditor