From fb0983bcb35a3ed0fac75e14fa15fe48af378495 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 19 Apr 2023 00:57:13 +0200 Subject: [PATCH] feat(ee): sync cache in background --- backend/windmill-worker/src/global_cache.rs | 23 +++++++++++---------- backend/windmill-worker/src/worker.rs | 9 ++++++-- 2 files changed, 19 insertions(+), 13 deletions(-) diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index cbde57502f..78acc7c56a 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -1,13 +1,16 @@ #[cfg(feature = "enterprise")] -use crate::{DENO_TMP_CACHE_DIR, GO_TMP_CACHE_DIR, PIP_TMP_CACHE_DIR}; - use crate::{ROOT_CACHE_DIR, ROOT_TMP_CACHE_DIR, TAR_CACHE_RATE, TMP_DIR}; +#[cfg(feature = "enterprise")] use itertools::Itertools; +#[cfg(feature = "enterprise")] use rand::Rng; +#[cfg(feature = "enterprise")] use std::process::Stdio; #[cfg(feature = "enterprise")] use tokio::{process::Command, sync::mpsc::Sender, time::Instant}; + +#[cfg(feature = "enterprise")] use windmill_common::error; #[cfg(feature = "enterprise")] @@ -42,7 +45,7 @@ pub async fn copy_cache_from_bucket(bucket: &str, tx: Sender<()>) -> error::Resu "--size-only", "--fast-list", "--exclude", - &format!("\"{TAR_CACHE_FILENAME},/deno/gen/file/**\""), + &format!("\"/{TAR_CACHE_FILENAME},/deno/gen/file/**\""), ], ) .await @@ -76,7 +79,7 @@ pub async fn copy_cache_to_bucket(bucket: &str) -> error::Result<()> { "--size-only", "--fast-list", "--exclude", - &format!("\"{TAR_CACHE_FILENAME},/deno/gen/file/**\""), + &format!("\"/{TAR_CACHE_FILENAME},/deno/gen/file/**\""), ], ) .await @@ -203,6 +206,7 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) { 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"); @@ -211,19 +215,13 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) { if let Err(e) = execute_command( TMP_DIR, "cp", - vec!["-e", &format!("{ROOT_CACHE_DIR}/{x}"), &ROOT_TMP_CACHE_DIR], + vec!["-r", &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 { @@ -240,6 +238,7 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) { // return true; // } +#[cfg(feature = "enterprise")] pub async fn copy_tmp_cache_to_cache() -> error::Result<()> { let start: Instant = Instant::now(); execute_command( @@ -255,6 +254,7 @@ pub async fn copy_tmp_cache_to_cache() -> error::Result<()> { Ok(()) } +#[cfg(feature = "enterprise")] pub async fn copy_cache_to_tmp_cache() -> error::Result<()> { let start: Instant = Instant::now(); execute_command( @@ -270,6 +270,7 @@ pub async fn copy_cache_to_tmp_cache() -> error::Result<()> { Ok(()) } +#[cfg(feature = "enterprise")] pub async fn execute_command(dir: &str, command: &str, args: Vec<&str>) -> error::Result<()> { match Command::new(command) .current_dir(dir) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index ac492e2fa6..cd89ee4400 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -47,21 +47,23 @@ use futures::{ }; use async_recursion::async_recursion; + #[cfg(feature = "enterprise")] use rand::Rng; #[cfg(feature = "enterprise")] -use crate::global_cache::{copy_cache_from_bucket_as_tar}; +use crate::global_cache::{copy_cache_from_bucket_as_tar, copy_cache_to_tmp_cache, cache_global, copy_tmp_cache_to_cache}; 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::{cache_global, copy_cache_to_tmp_cache, copy_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}, go_executor::{handle_go_job, install_go_dependencies}, }; + pub async fn create_token_for_owner_in_bg(db: &Pool, job: &QueuedJob) -> Arc> { let rw_lock = Arc::new(RwLock::new(String::new())); // skipping test runs @@ -403,6 +405,7 @@ pub async fn run_worker(2); let mut copy_cache_from_bucket_handle: Option> = None; @@ -452,6 +455,7 @@ pub async fn run_worker { + #[cfg(feature = "enterprise")] if let Err(e) = copy_tmp_cache_to_cache().await { tracing::error!(worker = %worker_name, "failed to sync tmp cache to cache: {}", e); }