From 28afe59a4ab4e97a18e4ca7a893d83c329ee4078 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 13 Jun 2024 19:09:16 +0200 Subject: [PATCH] feat: accelerate bun through caches (#3909) * all * all * all * all * all * simplify extract --- backend/src/main.rs | 7 +- backend/windmill-api/src/jobs.rs | 23 +-- backend/windmill-worker/Cargo.toml | 8 +- backend/windmill-worker/src/bun_executor.rs | 186 ++++++++---------- backend/windmill-worker/src/worker.rs | 15 +- .../windmill-worker/src/worker_lockfiles.rs | 11 +- benchmarks/benchmark_oneoff.ts | 34 +++- 7 files changed, 142 insertions(+), 142 deletions(-) diff --git a/backend/src/main.rs b/backend/src/main.rs index bc597e0ba7..708c612ed8 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -53,9 +53,9 @@ use windmill_common::METRICS_ADDR; use windmill_common::global_settings::OBJECT_STORE_CACHE_CONFIG_SETTING; use windmill_worker::{ - BUN_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, + BUN_CACHE_DIR, BUN_TAR_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, }; use crate::monitor::{ @@ -709,6 +709,7 @@ pub async fn run_workers error::JsonResult> { require_admin(authed.is_admin, &authed.username)?; - let mut sqlb = SqlBuilder::select_from("queue") - .fields(&["id"]) - .clone(); + let mut sqlb = SqlBuilder::select_from("queue").fields(&["id"]).clone(); sqlb = join_concurrency_key(lq.concurrency_key.as_ref(), sqlb); @@ -3935,13 +3933,16 @@ async fn add_batch_jobs( ) .fetch_all(&db) .await?; - sqlx::query!( - "INSERT INTO concurrency_key (job_id, key) SELECT id, $1 FROM unnest($2::uuid[]) as id", - custom_concurrency_key, - &uuids - ) - .execute(&db) - .await?; + + if let Some(custom_concurrency_key) = custom_concurrency_key { + sqlx::query!( + "INSERT INTO concurrency_key (job_id, key) SELECT id, $1 FROM unnest($2::uuid[]) as id", + custom_concurrency_key, + &uuids + ) + .execute(&db) + .await?; + } Ok(Json(uuids)) } diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 0888670ce9..a8c4b7919c 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -11,10 +11,10 @@ path = "src/lib.rs" [features] default = [] prometheus = ["dep:prometheus", "windmill-common/prometheus"] -enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:sha2", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"] +enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"] benchmark = ["windmill-queue/benchmark"] flamegraph = [] -parquet = ["windmill-common/parquet", "dep:object_store", "dep:tar"] +parquet = ["windmill-common/parquet", "dep:object_store"] flow_testing = [] cloud = [] @@ -71,7 +71,7 @@ base64.workspace = true gcp_auth = { workspace = true, optional = true } rust_decimal.workspace = true jsonwebtoken = { workspace = true, optional = true } -sha2 = { workspace = true, optional = true } +sha2.workspace = true pem = { workspace = true, optional = true } urlencoding.workspace = true nix.workspace = true @@ -81,7 +81,7 @@ hex.workspace = true tiberius = { workspace = true, optional = true } tokio-util = { workspace = true, optional = true } openidconnect = { workspace = true, optional = true} -tar = { workspace = true, optional = true} +tar.workspace = true object_store = { workspace = true, optional = true} convert_case.workspace = true diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 4a6cd7bc34..514bbf3d3f 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -2,8 +2,8 @@ use std::{collections::HashMap, process::Stdio}; use base64::Engine; use itertools::Itertools; -use regex::Regex; use serde_json::value::RawValue; +use sha2::Digest; use uuid::Uuid; use windmill_parser_ts::remove_pinned_imports; use windmill_queue::{append_logs, CanceledBy}; @@ -16,9 +16,9 @@ use crate::{ create_args_and_out_file, get_main_override, get_reserved_variables, handle_child, parse_npm_config, read_result, start_child_process, write_file, write_file_binary, }, - AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL, - DISABLE_NUSER, HOME_ENV, NODE_PATH, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_PATH, PATH_ENV, - TZ_ENV, + AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_CACHE_DIR, BUN_PATH, BUN_TAR_CACHE_DIR, + DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NODE_PATH, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_PATH, + PATH_ENV, TZ_ENV, }; use tokio::{fs::File, process::Command}; @@ -32,7 +32,7 @@ use tokio::sync::mpsc::Receiver; use windmill_common::variables; use windmill_common::{ - error::{self, to_anyhow, Result}, + error::{self, Result}, jobs::QueuedJob, }; @@ -50,11 +50,6 @@ const NSJAIL_CONFIG_RUN_BUN_CONTENT: &str = include_str!("../nsjail/run.bun.conf pub const BUN_LOCKB_SPLIT: &str = "\n//bun.lockb\n"; pub const EMPTY_FILE: &str = ""; -lazy_static::lazy_static! { - pub static ref TRUSTED_DEP: Regex = Regex::new(r"//\s?trustedDependencies:(.*)\n").unwrap(); - -} - pub async fn gen_lockfile( mem_peak: &mut i32, canceled_by: &mut Option, @@ -67,7 +62,6 @@ pub async fn gen_lockfile( base_internal_url: &str, worker_name: &str, export_pkg: bool, - trusted_deps: Vec, raw_deps: Option, npm_mode: bool, ) -> Result> { @@ -123,33 +117,6 @@ pub async fn gen_lockfile( false, ) .await?; - - if trusted_deps.len() > 0 { - let logs1 = format!( - "\ndetected trustedDependencies: {}\n", - trusted_deps.join(", ") - ); - append_logs(&job_id, w_id, logs1, db).await; - - let mut content = "".to_string(); - { - let mut file = File::open(format!("{job_dir}/package.json")).await?; - file.read_to_string(&mut content).await?; - } - let mut value = - serde_json::from_str::>(&content) - .map_err(to_anyhow)?; - value.insert( - "trustedDependencies".to_string(), - serde_json::json!(trusted_deps), - ); - write_file( - job_dir, - "package.json", - &serde_json::to_string(&value).map_err(to_anyhow)?, - ) - .await?; - } } install_lockfile( @@ -309,23 +276,6 @@ pub async fn install_lockfile( Ok(()) } -pub fn get_trusted_deps(code: &str) -> Vec { - // postinstall not allowed with nsjail - if !*DISABLE_NSJAIL { - return vec![]; - } - TRUSTED_DEP - .captures(code) - .map(|x| { - x[1].to_string() - .trim() - .split(',') - .map(|y| y.trim().to_string()) - .collect_vec() - }) - .unwrap_or_default() -} - struct Annotations { npm_mode: bool, nodejs_mode: bool, @@ -491,6 +441,18 @@ pub async fn pull_codebase(_w_id: &str, _id: &str, _job_dir: &str) -> Result<()> )); } +fn untar_file(file_path: &str, output_dir: &str) -> anyhow::Result<()> { + // Open the tar file + let file = std::fs::File::open(file_path)?; + let file = std::io::BufReader::new(file); + + // For a plain tar file, use it directly + let mut archive = tar::Archive::new(file); + archive.unpack(output_dir)?; + + Ok(()) +} + #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_bun_job( requirements_o: Option, @@ -511,7 +473,7 @@ pub async fn handle_bun_job( let _ = write_file(job_dir, "main.ts", inner_content).await?; } else { let _ = write_file(job_dir, "package.json", r#"{ "type": "module" }"#).await?; - } + }; let common_bun_proc_envs: HashMap = get_common_bun_proc_envs(&base_internal_url).await; @@ -530,6 +492,7 @@ pub async fn handle_bun_job( )); } + let mut gbuntar_name = None; if let Some(codebase) = codebase.as_ref() { pull_codebase(&job.workspace_id, codebase, job_dir).await?; } else if let Some(reqs) = requirements_o { @@ -542,38 +505,58 @@ pub async fn handle_bun_job( let _ = write_file(job_dir, "package.json", &splitted[0]).await?; let lockb = splitted[1]; if lockb != EMPTY_FILE { - let has_trusted_deps = &splitted[0].contains("trustedDependencies"); - - if !has_trusted_deps { - let _ = write_file_binary( - job_dir, - "bun.lockb", - &base64::engine::general_purpose::STANDARD - .decode(&splitted[1]) - .map_err(|_| { - error::Error::InternalErr("Could not decode bun.lockb".to_string()) - })?, - ) - .await?; - } - - install_lockfile( - mem_peak, - canceled_by, - &job.id, - &job.workspace_id, - db, + let _ = write_file_binary( job_dir, - worker_name, - common_bun_proc_envs.clone(), - annotation.npm_mode, + "bun.lockb", + &base64::engine::general_purpose::STANDARD + .decode(&splitted[1]) + .map_err(|_| { + error::Error::InternalErr("Could not decode bun.lockb".to_string()) + })?, ) .await?; + + let mut sha_path = sha2::Sha256::new(); + sha_path.update(lockb.as_bytes()); + + let buntar_name = base64::engine::general_purpose::STANDARD.encode(sha_path.finalize()); + let buntar_path = format!("{BUN_TAR_CACHE_DIR}/{buntar_name}.tar"); + + let mut skip_install = false; + let mut create_buntar = false; + if tokio::fs::metadata(&buntar_path).await.is_ok() { + if let Err(e) = untar_file(&buntar_path, job_dir) { + tracing::error!("Could not untar buntar: {e}"); + } else { + gbuntar_name = Some(buntar_name.clone()); + skip_install = true; + } + } else { + create_buntar = true; + } + if !skip_install { + install_lockfile( + mem_peak, + canceled_by, + &job.id, + &job.workspace_id, + db, + job_dir, + worker_name, + common_bun_proc_envs.clone(), + annotation.npm_mode, + ) + .await?; + if create_buntar { + if let Err(e) = tar::Builder::new(std::fs::File::create(&buntar_path)?) + .append_dir_all(".", job_dir) + { + tracing::error!("Could not create buntar: {e}"); + } + } + } } } else { - // TODO: remove once bun implement a reasonable set of trusted deps - let trusted_deps = get_trusted_deps(inner_content); - // if !*DISABLE_NSJAIL || !empty_trusted_deps || has_custom_config_registry { let logs1 = "\n\n--- BUN INSTALL ---\n".to_string(); append_logs(&job.id, &job.workspace_id, logs1, db).await; @@ -590,17 +573,17 @@ pub async fn handle_bun_job( base_internal_url, worker_name, false, - trusted_deps, None, annotation.npm_mode, ) .await?; + // } } let _ = write_file(job_dir, "main.ts", &remove_pinned_imports(inner_content)?).await?; - let init_logs = if codebase.is_some() { + let mut init_logs = if codebase.is_some() { "\n\n--- NODE SNAPSHOT EXECUTION ---\n".to_string() } else if annotation.nodejs_mode { "\n\n--- NODE CODE EXECUTION ---\n".to_string() @@ -608,6 +591,13 @@ pub async fn handle_bun_job( "\n\n--- BUN CODE EXECUTION ---\n".to_string() }; + if let Some(gbuntar_name) = gbuntar_name { + init_logs = format!( + "\nskipping install, using cached buntar based on lockfile hash: {gbuntar_name}{}", + init_logs + ); + } + append_logs(&job.id, &job.workspace_id, init_logs, db).await; let write_wrapper_f = async { @@ -963,20 +953,16 @@ pub async fn start_worker( let _ = write_file(job_dir, "package.json", &splitted[0]).await?; let lockb = splitted[1]; if lockb != EMPTY_FILE { - let has_trusted_deps = &splitted[0].contains("trustedDependencies"); - - if !has_trusted_deps { - let _ = write_file_binary( - job_dir, - "bun.lockb", - &base64::engine::general_purpose::STANDARD - .decode(&splitted[1]) - .map_err(|_| { - error::Error::InternalErr("Could not decode bun.lockb".to_string()) - })?, - ) - .await?; - } + let _ = write_file_binary( + job_dir, + "bun.lockb", + &base64::engine::general_purpose::STANDARD + .decode(&splitted[1]) + .map_err(|_| { + error::Error::InternalErr("Could not decode bun.lockb".to_string()) + })?, + ) + .await?; install_lockfile( &mut mem_peak, @@ -993,7 +979,6 @@ pub async fn start_worker( tracing::info!("dedicated worker requirements installed: {reqs}"); } } else if !*DISABLE_NSJAIL { - let trusted_deps = get_trusted_deps(inner_content); logs.push_str("\n\n--- BUN INSTALL ---\n"); let _ = gen_lockfile( &mut mem_peak, @@ -1007,7 +992,6 @@ pub async fn start_worker( base_internal_url, worker_name, false, - trusted_deps, None, annotation.npm_mode, ) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index f61edf6742..4fbdcfe7b3 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -204,6 +204,7 @@ pub const DENO_CACHE_DIR_NPM: &str = concatcp!(ROOT_CACHE_DIR, "deno/npm"); pub const GO_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "go"); pub const BUN_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "bun"); +pub const BUN_TAR_CACHE_DIR: &str = concatcp!(ROOT_CACHE_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"); @@ -723,12 +724,14 @@ pub async fn run_worker { - let trusted_deps = if !raw_deps { + if !raw_deps { let _ = write_file(job_dir, "main.ts", job_raw_code).await?; - //TODO: remove once bun provides sane default fot it - get_trusted_deps(job_raw_code) - } else { - vec![] - }; + } let req = gen_lockfile( mem_peak, canceled_by, @@ -1219,7 +1215,6 @@ async fn capture_dependency_job( base_internal_url, worker_name, true, - trusted_deps, if raw_deps { Some(job_raw_code.to_string()) } else { diff --git a/benchmarks/benchmark_oneoff.ts b/benchmarks/benchmark_oneoff.ts index 50e8f43369..da8c8b9fea 100644 --- a/benchmarks/benchmark_oneoff.ts +++ b/benchmarks/benchmark_oneoff.ts @@ -119,7 +119,11 @@ export async function main({ return completedJobs - pastJobs; } - if (["deno", "python", "go", "bash", "dedicated", "bun", "nativets"].includes(kind)) { + if ( + ["deno", "python", "go", "bash", "dedicated", "bun", "nativets"].includes( + kind + ) + ) { await createBenchScript(kind, workspace); } @@ -135,7 +139,9 @@ export async function main({ kind: "noop", }); } else if ( - ["deno", "python", "go", "bash", "dedicated", "bun", "nativets"].includes(kind) + ["deno", "python", "go", "bash", "dedicated", "bun", "nativets"].includes( + kind + ) ) { body = JSON.stringify({ kind: "script", @@ -148,22 +154,21 @@ export async function main({ flow_value: payload.value, }); } else if (kind.startsWith("flow:")) { - console.log("Detected custom flow ") + console.log("Detected custom flow "); body = JSON.stringify({ kind: "flow", - path: kind.substr(5) + path: kind.substr(5), }); } else if (kind.startsWith("script:")) { - console.log("Detected custom script") + console.log("Detected custom script"); body = JSON.stringify({ kind: "script", - path: kind.substr(7) + path: kind.substr(7), }); } else { throw new Error("Unknown script pattern " + kind); } - const response = await fetch( config.server + "/api/w/" + @@ -179,7 +184,12 @@ export async function main({ } ); if (!response.ok) { - throw new Error("Failed to create jobs: " + response.statusText); + throw new Error( + "Failed to create jobs: " + + response.statusText + + " " + + (await response.text()) + ); } const uuids = await response.json(); const end_create = Date.now(); @@ -247,7 +257,13 @@ export async function main({ console.log("completed jobs", completedJobs); console.log("queue length:", await getQueueCount()); - if (!noVerify && kind !== "noop" && kind !== 'nativets' && !kind.startsWith("flow:") && !kind.startsWith("script:")) { + if ( + !noVerify && + kind !== "noop" && + kind !== "nativets" && + !kind.startsWith("flow:") && + !kind.startsWith("script:") + ) { await verifyOutputs(uuids, config.workspace_id); }