diff --git a/backend/src/main.rs b/backend/src/main.rs index 2b08c7e980..ac424a62bb 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -150,6 +150,7 @@ Windmill Community Edition {GIT_VERSION} "BLACKLIST_WORKSPACES", "INSTANCE_EVENTS_WEBHOOK", "CLOUD_HOSTED", + "GLOBAL_CACHE_INTERVAL", ]); if server_mode || num_workers > 0 { diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index 088173bde4..cbde57502f 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -1,215 +1,142 @@ #[cfg(feature = "enterprise")] -use crate::{ - DENO_CACHE_DIR, DENO_TMP_CACHE_DIR, GO_CACHE_DIR, GO_TMP_CACHE_DIR, PIP_CACHE_DIR, - PIP_TMP_CACHE_DIR, -}; +use crate::{DENO_TMP_CACHE_DIR, GO_TMP_CACHE_DIR, PIP_TMP_CACHE_DIR}; -use crate::{ROOT_CACHE_DIR, ROOT_TMP_CACHE_DIR}; -#[cfg(feature = "enterprise")] +use crate::{ROOT_CACHE_DIR, ROOT_TMP_CACHE_DIR, TAR_CACHE_RATE, TMP_DIR}; +use itertools::Itertools; +use rand::Rng; use std::process::Stdio; + #[cfg(feature = "enterprise")] -use tokio::{fs::DirBuilder, process::Command, sync::mpsc::Sender, time::Instant}; +use tokio::{process::Command, sync::mpsc::Sender, time::Instant}; use windmill_common::error; #[cfg(feature = "enterprise")] const TAR_CACHE_FILENAME: &str = "entirecache.tar"; #[cfg(feature = "enterprise")] -pub async fn copy_cache_from_bucket( - bucket: &str, - tx: Option>, -) -> Option> { - tracing::info!("Copying cache from bucket in the background {bucket}"); - let bucket = bucket.to_string(); - let tx_is_some = tx.is_some(); - let f = async move { - let elapsed = Instant::now(); +pub async fn cache_global(bucket: &str, tx: Sender<()>) -> error::Result<()> { + copy_cache_from_bucket(bucket, tx).await?; + copy_cache_to_bucket(bucket).await?; - match Command::new("rclone") - .arg("copy") - .arg(format!(":s3,env_auth=true:{bucket}")) - .arg(if tx_is_some { - ROOT_TMP_CACHE_DIR - } else { - ROOT_CACHE_DIR - }) - .arg("--size-only") - .arg("--fast-list") - .arg("--exclude") - .arg(format!("\"{TAR_CACHE_FILENAME},/deno/gen/file/**\"")) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .spawn() - { - Ok(mut h) => { - h.wait().await.unwrap(); - } - Err(e) => tracing::warn!("Failed to run periodic job pull. Error: {:?}", e), - } - tracing::info!( - "Finished copying cache from bucket {bucket}, took {:?}s", - elapsed.elapsed().as_secs() - ); - - for x in if !tx_is_some { - [PIP_CACHE_DIR, DENO_CACHE_DIR, GO_CACHE_DIR] - } else { - [PIP_TMP_CACHE_DIR, DENO_TMP_CACHE_DIR, GO_TMP_CACHE_DIR] - } { - DirBuilder::new() - .recursive(true) - .create(x) - .await - .expect("could not create initial worker dir"); - } - - if let Some(tx) = tx { - tx.send(()).await.expect("can send copy cache signal"); - } - }; - if tx_is_some { - return Some(tokio::spawn(f)); - } else { - f.await; - return None; + // this is to prevent excessive tar upload. 1/100*15min = each worker sync its tar once per day on average + if rand::thread_rng().gen_range(0..*TAR_CACHE_RATE) == 0 { + copy_cache_to_bucket_as_tar(bucket).await; } + Ok(()) } #[cfg(feature = "enterprise")] -pub async fn copy_cache_to_bucket(bucket: &str) { - tracing::info!("Copying cache to bucket {bucket}"); - let elapsed = Instant::now(); - match Command::new("rclone") - .arg("copy") - .arg(ROOT_CACHE_DIR) - .arg(format!(":s3,env_auth=true:{bucket}")) - .arg("--size-only") - .arg("--fast-list") - .arg("--exclude") - .arg(format!("\"{TAR_CACHE_FILENAME},/deno/gen/file/**\"")) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .spawn() +pub async fn copy_cache_from_bucket(bucket: &str, tx: Sender<()>) -> error::Result<()> { + tracing::info!("Copying cache from bucket in the background {bucket}"); + let bucket = bucket.to_string(); + + let start = Instant::now(); + + if let Err(e) = execute_command( + ROOT_TMP_CACHE_DIR, + "rclone", + vec![ + "copy", + &format!(":s3,env_auth=true:{bucket}"), + &ROOT_TMP_CACHE_DIR, + "--size-only", + "--fast-list", + "--exclude", + &format!("\"{TAR_CACHE_FILENAME},/deno/gen/file/**\""), + ], + ) + .await { - Ok(mut h) => { - h.wait().await.unwrap(); - } - Err(e) => tracing::info!("Failed to run periodic job push. Error: {:?}", e), + tracing::info!("Failed to to copy cache from bucket. Error: {:?}", e); + return Err(e); + } + + tracing::info!( + "Finished copying cache from bucket {bucket}, took {:?}s", + start.elapsed().as_secs() + ); + + tx.send(()).await.expect("can send copy cache signal"); + + Ok(()) +} + +#[cfg(feature = "enterprise")] +pub async fn copy_cache_to_bucket(bucket: &str) -> error::Result<()> { + tracing::info!("Copying cache to bucket {bucket}"); + let start = Instant::now(); + + if let Err(e) = execute_command( + ROOT_TMP_CACHE_DIR, + "rclone", + vec![ + "copy", + &ROOT_TMP_CACHE_DIR, + &format!(":s3,env_auth=true:{bucket}"), + "--size-only", + "--fast-list", + "--exclude", + &format!("\"{TAR_CACHE_FILENAME},/deno/gen/file/**\""), + ], + ) + .await + { + tracing::info!("Failed to to copy cache to bucket. Error: {:?}", e); + return Err(e); } tracing::info!( "Finished copying cache to bucket {bucket}, took: {:?}s", - elapsed.elapsed().as_secs() + start.elapsed().as_secs() ); + Ok(()) } #[cfg(feature = "enterprise")] pub async fn copy_cache_to_bucket_as_tar(bucket: &str) { tracing::info!("Copying cache to bucket {bucket} as tar"); - let elapsed = Instant::now(); + let start = Instant::now(); - match Command::new("tar") - .current_dir(ROOT_CACHE_DIR) - .arg("-c") - .arg("-f") - .arg(format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}")) - .args(&["pip", "go", "deno"]) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .spawn() + if let Err(e) = execute_command( + ROOT_TMP_CACHE_DIR, + "tar", + vec![ + "-c", + "-f", + &format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}"), + "pip", + "go", + "deno", + ], + ) + .await { - Ok(mut h) => { - if !h.wait().await.unwrap().success() { - tracing::info!("Failed to tar cache"); - return; - } - } - Err(e) => { - tracing::info!("Failed tar cache. Error: {e:?}"); - return; - } + tracing::info!("Failed to tar cache. Error: {:?}", e); + return; } - let tar_metadata = tokio::fs::metadata(format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}")).await; + let tar_metadata = + tokio::fs::metadata(format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}")).await; if tar_metadata.is_err() || tar_metadata.as_ref().unwrap().len() == 0 { tracing::info!("Failed to tar cache"); return; } - match Command::new("rclone") - .current_dir(ROOT_CACHE_DIR) - .arg("copyto") - .arg(TAR_CACHE_FILENAME) - .arg(format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}")) - .arg("-vv") - .arg("--size-only") - .arg("--fast-list") - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .spawn() + if let Err(e) = execute_command( + ROOT_TMP_CACHE_DIR, + "rclone", + vec![ + "copyto", + &format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}"), + &format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}"), + "-vv", + "--size-only", + "--fast-list", + ], + ) + .await { - Ok(mut h) => { - h.wait().await.unwrap(); - } - Err(e) => tracing::info!("Failed to copy tar cache to bucket. Error: {:?}", e), - } - - if let Err(e) = tokio::fs::remove_file(format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}")).await { - tracing::info!("Failed to remove tar cache. Error: {:?}", e); - }; - - tracing::info!( - "Finished copying cache to bucket {bucket} as tar, took: {:?}s. Size of new tar: {}", - elapsed.elapsed().as_secs(), - tar_metadata.unwrap().len() - ); -} - -#[cfg(feature = "enterprise")] -pub async fn copy_cache_from_bucket_as_tar(bucket: &str) -> bool { - tracing::info!("Copying cache from bucket {bucket} as tar"); - let elapsed = Instant::now(); - - match Command::new("rclone") - .arg("copyto") - .arg(format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}")) - .arg(format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}")) - .arg("-vv") - .arg("--size-only") - .arg("--fast-list") - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .spawn() - { - Ok(mut h) => { - if !h.wait().await.unwrap().success() { - tracing::info!("Failed to download tar cache, continuing nonetheless"); - return false; - } - } - Err(e) => { - tracing::info!("Failed to download tar cache, continuing nonetheless. Error: {e:?}"); - return false; - } - } - - match Command::new("tar") - .current_dir(ROOT_TMP_CACHE_DIR) - .arg("-xpvf") - .arg(format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}")) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .spawn() - { - Ok(mut h) => { - if !h.wait().await.unwrap().success() { - tracing::info!("Failed to untar cache, continuing nonetheless"); - return false; - } - } - Err(e) => { - tracing::warn!("Failed to untar cache, continuing nonetheless. Error: {e:?}"); - return false; - } + tracing::info!("Failed to copy tar to bucket. Error: {:?}", e); + return; } if let Err(e) = @@ -218,13 +145,85 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) -> bool { tracing::info!("Failed to remove tar cache. Error: {:?}", e); }; - let r = move_tmp_cache_to_cache().await.is_ok(); + tracing::info!( + "Finished copying cache to bucket {bucket} as tar, took: {:?}s. Size of new tar: {}", + start.elapsed().as_secs(), + tar_metadata.unwrap().len() + ); +} + +#[cfg(feature = "enterprise")] +pub async fn copy_cache_from_bucket_as_tar(bucket: &str) { + use tokio::fs::metadata; + + tracing::info!("Copying cache from bucket {bucket} as tar"); + let elapsed = Instant::now(); + + if let Err(e) = execute_command( + ROOT_CACHE_DIR, + "rclone", + vec![ + "copyto", + &format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}"), + &format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}"), + "-vv", + "--size-only", + "--fast-list", + ], + ) + .await + { + tracing::info!("Failed copy tar from cache. Error: {:?}", e); + return; + } + + if let Err(e) = execute_command( + ROOT_CACHE_DIR, + "tar", + vec!["-xpvf", &format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}")], + ) + .await + { + tracing::info!("Failed to untar cache. Error: {:?}", e); + return; + } + + if let Err(e) = tokio::fs::remove_file(format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}")).await { + tracing::info!("Failed to remove tar cache. Error: {:?}", e); + return; + }; tracing::info!( - "Finished copying cache from bucket {bucket} as tar, took: {:?}s. copy success: {r}", + "Finished copying cache from bucket {bucket} as tar, took: {:?}s", elapsed.elapsed().as_secs() ); - return r; + + if metadata(&ROOT_TMP_CACHE_DIR).await.is_ok() { + if let Err(e) = tokio::fs::remove_dir_all(&ROOT_TMP_CACHE_DIR).await { + tracing::info!(error = %e, "Could not remove root tmp cache dir"); + } + } + tokio::fs::create_dir_all(&ROOT_TMP_CACHE_DIR) + .await + .expect("Could not create root tmp cache dir"); + + for x in ["deno", "go", "pip"] { + if let Err(e) = execute_command( + TMP_DIR, + "cp", + vec!["-e", &format!("{ROOT_CACHE_DIR}/{x}"), &ROOT_TMP_CACHE_DIR], + ) + .await + { + tracing::info!(error = %e, "Could not copy root dir to tmp root dir"); + } + } + tokio::fs::rename( + &format!("{ROOT_CACHE_DIR}/{TAR_CACHE_FILENAME}"), + &format!("{ROOT_CACHE_DIR}/{TAR_CACHE_FILENAME}"), + ) + .await + .expect("Could not rename tar cache"); } // async fn check_if_bucket_syncable(bucket: &str) -> bool { @@ -241,9 +240,58 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) -> bool { // return true; // } -pub async fn move_tmp_cache_to_cache() -> error::Result<()> { - tokio::fs::remove_dir_all(ROOT_CACHE_DIR).await?; - tokio::fs::rename(ROOT_TMP_CACHE_DIR, ROOT_CACHE_DIR).await?; - tracing::info!("Finished moving tmp cache to cache"); +pub async fn copy_tmp_cache_to_cache() -> error::Result<()> { + let start: Instant = Instant::now(); + execute_command( + TMP_DIR, + "rclone", + vec!["sync", ROOT_TMP_CACHE_DIR, ROOT_CACHE_DIR], + ) + .await?; + tracing::info!( + "Finished copying local tmp cache to local cache. Took {}ms", + start.elapsed().as_millis(), + ); + Ok(()) +} + +pub async fn copy_cache_to_tmp_cache() -> error::Result<()> { + let start: Instant = Instant::now(); + execute_command( + TMP_DIR, + "rclone", + vec!["sync", ROOT_CACHE_DIR, ROOT_TMP_CACHE_DIR], + ) + .await?; + tracing::info!( + "Finished copying local cache to local tmp cache. Took {}ms", + start.elapsed().as_millis() + ); + Ok(()) +} + +pub async fn execute_command(dir: &str, command: &str, args: Vec<&str>) -> error::Result<()> { + match Command::new(command) + .current_dir(dir) + .args(args.clone()) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .spawn() + { + Ok(mut h) => { + if !h.wait().await.unwrap().success() { + return Err(error::Error::ExecutionErr(format!( + "Failed to apply {command} with args: {}", + args.iter().join(" ") + ))); + } + } + Err(e) => { + return Err(error::Error::ExecutionErr(format!( + "Failed to apply {command} with args: {}. Error: {e:?}", + args.iter().join(" ") + ))); + } + } Ok(()) } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 5f24e5b4e6..ac492e2fa6 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -51,13 +51,13 @@ use async_recursion::async_recursion; use rand::Rng; #[cfg(feature = "enterprise")] -use crate::global_cache::{copy_cache_from_bucket_as_tar, copy_cache_from_bucket, copy_cache_to_bucket, copy_cache_to_bucket_as_tar}; +use crate::global_cache::{copy_cache_from_bucket_as_tar}; use crate::{ jobs::{add_completed_job, add_completed_job_error}, worker_flow::{ handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress, - }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs}, global_cache::{move_tmp_cache_to_cache}, go_executor::{handle_go_job, install_go_dependencies}, + }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs}, global_cache::{cache_global, copy_cache_to_tmp_cache, copy_tmp_cache_to_cache}, go_executor::{handle_go_job, install_go_dependencies}, }; @@ -119,7 +119,7 @@ pub async fn create_token_for_owner( Ok(token) } -const TMP_DIR: &str = "/tmp/windmill"; +pub const TMP_DIR: &str = "/tmp/windmill"; pub const ROOT_CACHE_DIR: &str = "/tmp/windmill/cache/"; pub const ROOT_TMP_CACHE_DIR: &str = "/tmp/windmill/tmpcache/"; pub const PIP_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "pip"); @@ -167,10 +167,6 @@ lazy_static::lazy_static! { .and_then(|x| x.parse::().ok()) .unwrap_or(false); - static ref S3_CACHE_BUCKET: Option = std::env::var("S3_CACHE_BUCKET") - .ok() - .map(|e| Some(e)) - .unwrap_or(None); pub static ref DENO_PATH: String = std::env::var("DENO_PATH").unwrap_or_else(|_| "/usr/bin/deno".to_string()); pub static ref NSJAIL_PATH: String = std::env::var("NSJAIL_PATH").unwrap_or_else(|_| "nsjail".to_string()); @@ -199,7 +195,7 @@ lazy_static::lazy_static! { .ok() .map(|x| x.split(',').map(|x| x.to_string()).collect()); - static ref TAR_CACHE_RATE: i32 = std::env::var("TAR_CACHE_RATE") + pub static ref TAR_CACHE_RATE: i32 = std::env::var("TAR_CACHE_RATE") .ok() .and_then(|x| x.parse::().ok()) .unwrap_or(100); @@ -229,6 +225,11 @@ lazy_static::lazy_static! { .and_then(|x| x.parse::().ok()) .unwrap_or(60 * 10); + pub static ref S3_CACHE_BUCKET: Option = std::env::var("S3_CACHE_BUCKET") + .ok() + .map(|e| Some(e)) + .unwrap_or(None); + } //only matter if CLOUD_HOSTED @@ -294,7 +295,7 @@ pub async fn run_worker(2); + let (copy_to_bucket_tx, mut copy_to_bucket_rx) = mpsc::channel::<()>(2); let mut copy_cache_from_bucket_handle: Option> = None; - let mut initialized_cache = false; - #[cfg(feature = "enterprise")] if i_worker == 1 { if let Some(ref s) = S3_CACHE_BUCKET.clone() { // We try to download the entire cache as a tar, it is much faster over S3 - if !copy_cache_from_bucket_as_tar(&s).await { - // We revert to copying the cache from the bucket - copy_cache_from_bucket_handle = copy_cache_from_bucket(&s, Some(_copy_bucket_tx.clone())).await; - } else { - initialized_cache = true; - } + copy_cache_from_bucket_as_tar(&s).await; } - } + }; IS_READY.store(true, Ordering::Relaxed); @@ -458,6 +452,7 @@ pub async fn run_worker NUM_SECS_PING { @@ -474,19 +469,26 @@ pub async fn run_worker *GLOBAL_CACHE_INTERVAL && i_worker == 1 { - if let Some(ref s) = S3_CACHE_BUCKET.clone() { - copy_cache_from_bucket(&s, None).await; - copy_cache_to_bucket(&s).await; - - // this is to prevent excessive tar upload. 1/100*15min = each worker sync its tar once per day on average - if rand::thread_rng().gen_range(0..*TAR_CACHE_RATE) == 0 { - copy_cache_to_bucket_as_tar(&s).await; + if i_worker == 1 && S3_CACHE_BUCKET.is_some() { + if last_sync.elapsed().as_secs() > *GLOBAL_CACHE_INTERVAL && + (copy_cache_from_bucket_handle.is_none() || copy_cache_from_bucket_handle.as_ref().unwrap().is_finished()) { + tracing::info!("Started syncing cache"); + last_sync = Instant::now(); + if let Err(e) = copy_cache_to_tmp_cache().await { + tracing::error!("failed to copy cache to tmp cache: {}", e); + } else { + copy_cache_from_bucket_handle = Some(tokio::task::spawn(async move { + if let Some(ref s) = S3_CACHE_BUCKET.clone() { + if let Err(e) = cache_global(s, copy_tx).await { + tracing::error!("failed to sync cache: {}", e); + } + } + })); } } - last_sync = Instant::now(); } + let (do_break, next_job) = if first_run { (false, Ok(Some(QueuedJob::default()))) } else { @@ -495,17 +497,17 @@ pub async fn run_worker { if let Some(copy_cache_from_bucket_handle) = copy_cache_from_bucket_handle.as_ref() { - copy_cache_from_bucket_handle.abort(); + if !copy_cache_from_bucket_handle.is_finished() { + copy_cache_from_bucket_handle.abort(); + } } println!("received killpill for worker {}", i_worker); (true, Ok(None)) }, - _ = _copy_bucket_rx.recv() => { - if let Err(e) = move_tmp_cache_to_cache().await { + _ = copy_to_bucket_rx.recv() => { + if let Err(e) = copy_tmp_cache_to_cache().await { tracing::error!(worker = %worker_name, "failed to sync tmp cache to cache: {}", e); } - copy_cache_from_bucket_handle = None; - initialized_cache = true; (false, Ok(None)) }, Some(job_id) = same_worker_rx.recv() => { @@ -1419,6 +1421,9 @@ async fn handle_deno_job( // start = Instant::now(); handle_child(&job.id, db, logs, child, false, worker_name, &job.workspace_id, "deno run").await?; // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str()); + if let Err(e) = tokio::fs::remove_dir_all(format!("{DENO_CACHE_DIR}/gen/file/{job_dir}")).await { + tracing::error!("failed to remove deno gen tmp cache dir: {}", e); + } read_result(job_dir).await }