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

feat(ee): sync cache in background

Browse Source
This commit is contained in:
Ruben Fiszel
2023-04-19 00:55:35 +02:00
parent 7f9f314b4a
commit 89d0bea9fa
3 changed files with 271 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/src/main.rs
View File

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

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

@@ -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<Sender<()>>,
) -> Option<tokio::task::JoinHandle<()>> {
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(())
}

71
backend/windmill-worker/src/worker.rs
View File

@@ -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::<bool>().ok())
.unwrap_or(false);
static ref S3_CACHE_BUCKET: Option<String> = 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::<i32>().ok())
.unwrap_or(100);
@@ -229,6 +225,11 @@ lazy_static::lazy_static! {
.and_then(|x| x.parse::<u64>().ok())
.unwrap_or(60 * 10);
pub static ref S3_CACHE_BUCKET: Option<String> = 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let worker_dir = format!("{TMP_DIR}/{worker_name}");
tracing::debug!(worker_dir = %worker_dir, worker_name = %worker_name, "Creating worker dir");
for x in [&worker_dir, ROOT_TMP_CACHE_DIR, PIP_CACHE_DIR, DENO_CACHE_DIR, GO_CACHE_DIR] {
for x in [&worker_dir, PIP_CACHE_DIR, DENO_CACHE_DIR, GO_CACHE_DIR] {
DirBuilder::new()
.recursive(true)
.create(x)
@@ -402,24 +403,17 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
WORKER_STARTED.inc();
}
let (_copy_bucket_tx, mut _copy_bucket_rx) = mpsc::channel::<()>(2);
let (copy_to_bucket_tx, mut copy_to_bucket_rx) = mpsc::channel::<()>(2);
let mut copy_cache_from_bucket_handle: Option<tokio::task::JoinHandle<()>> = 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
);
}
let copy_tx = copy_to_bucket_tx.clone();
let do_break = async {
if last_ping.elapsed().as_secs() > NUM_SECS_PING {
@@ -474,19 +469,26 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
#[cfg(feature = "enterprise")]
if initialized_cache && last_sync.elapsed().as_secs() > *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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
biased;
_ = rx.recv() => {
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
}