starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-08 00:03:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

removing pulling all piptars at start

Browse Source
This commit is contained in:
Ruben Fiszel
2024-03-17 11:21:43 +01:00
parent e0235d9e9e
commit b3e53de947
3 changed files with 2 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/global_settings.rs
-1
View File
@@ -1,4 +1,3 @@
pub const WORKER_S3_BUCKET_SYNC: &str = "worker_s3_bucket_sync";
pub const CUSTOM_TAGS_SETTING: &str = "custom_tags";
pub const DEFAULT_TAGS_PER_WORKSPACE_SETTING: &str = "default_tags_per_workspace";
pub const BASE_URL_SETTING: &str = "base_url";
backend/windmill-worker/src/global_cache.rs
+1 -407
View File
@@ -1,16 +1,11 @@
#[cfg(feature = "enterprise")]
use crate::{ROOT_CACHE_DIR, ROOT_TMP_CACHE_DIR, TAR_PIP_TMP_CACHE_DIR};
use crate::{ROOT_TMP_CACHE_DIR, TAR_PIP_TMP_CACHE_DIR};
#[cfg(feature = "enterprise")]
use itertools::Itertools;
// #[cfg(feature = "enterprise")]
// use rand::Rng;
#[cfg(feature = "enterprise")]
use std::process::Stdio;
#[cfg(feature = "enterprise")]
use windmill_common::DB;
#[cfg(feature = "enterprise")]
use windmill_common::global_settings::WORKER_S3_BUCKET_SYNC;
#[cfg(feature = "enterprise")]
use tokio::{process::Command, time::Instant};
@@ -18,9 +13,6 @@ use tokio::{process::Command, time::Instant};
#[cfg(feature = "enterprise")]
use windmill_common::error;
// #[cfg(feature = "enterprise")]
// const TAR_CACHE_FILENAME: &str = "denogocache.tar";
#[cfg(feature = "enterprise")]
pub async fn build_tar_and_push(bucket: String, folder: String) -> error::Result<()> {
tracing::info!("Started building and pushing piptar {folder}");
@@ -130,373 +122,6 @@ pub async fn pull_from_tar(bucket: &str, folder: String) -> error::Result<()> {
Ok(())
}
// #[cfg(feature = "enterprise")]
// pub async fn cache_global(bucket: &str, _tx: Sender<()>) -> error::Result<()> {
// // copy_cache_from_bucket(bucket, tx).await?;
// // copy_cache_to_bucket(bucket).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(bucket).await;
// }
// Ok(())
// }
// #[cfg(feature = "enterprise")]
// 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,
// // "-l",
// "--size-only",
// "--fast-list",
// "--filter",
// "+ deno/npm/**",
// "--filter",
// "+ deno/deps/**",
// // "--filter",
// // "+ bun/**",
// "--filter",
// "+ go/**",
// "--filter",
// "+ tar/**",
// "--filter",
// "- *",
// ],
// )
// .await
// {
// tracing::info!("Failed 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}"),
// // "-l",
// "--size-only",
// "--fast-list",
// "--filter",
// "+ deno/npm/**",
// "--filter",
// "+ deno/deps/**",
// // "--filter",
// // "+ bun/**",
// "--filter",
// "+ go/**",
// "--filter",
// "- *",
// ],
// )
// .await
// {
// tracing::info!("Failed to copy cache to bucket. Error: {:?}", e);
// return Err(e);
// }
// tracing::info!(
// "Finished copying cache to bucket {bucket}, took: {:?}s",
// start.elapsed().as_secs()
// );
// Ok(())
// }
#[cfg(feature = "enterprise")]
pub async fn worker_s3_bucket_sync_enabled(db: &DB) -> bool {
let q = sqlx::query!(
"SELECT value FROM global_settings WHERE name = $1",
WORKER_S3_BUCKET_SYNC
)
.fetch_optional(db)
.await;
if let Ok(q) = q {
let r = q.map(|x| x.value.as_bool().unwrap_or(true)).unwrap_or(true);
tracing::info!("Got global setting {WORKER_S3_BUCKET_SYNC}: {}", r);
r
} else {
tracing::info!("Failed to get global setting {WORKER_S3_BUCKET_SYNC}");
false
}
}
// #[cfg(feature = "enterprise")]
// pub async fn copy_cache_to_bucket_as_tar(bucket: &str) {
// tracing::info!("Copying cache to bucket {bucket} as tar");
// let start = Instant::now();
// if let Err(e) = execute_command(
// ROOT_TMP_CACHE_DIR,
// "tar",
// vec![
// "-c",
// "-f",
// &format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}"),
// "go",
// "deno/npm",
// "deno/deps", // "bun",
// ],
// )
// .await
// {
// tracing::info!("Failed to tar cache. Error: {:?}", e);
// return;
// }
// 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;
// }
// 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}"),
// "-v",
// "--size-only",
// "--fast-list",
// "--s3-no-check-bucket",
// ],
// )
// .await
// {
// tracing::info!("Failed to copy tar to bucket. Error: {:?}", e);
// return;
// }
// if let Err(e) =
// tokio::fs::remove_file(format!("{ROOT_TMP_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: {}",
// start.elapsed().as_secs(),
// tar_metadata.unwrap().len()
// );
// }
// #[cfg(feature = "enterprise")]
// pub async fn copy_denogo_cache_from_bucket_as_tar(bucket: &str) {
// tracing::info!("Copying deno,go,bun cache from bucket {bucket} as tar");
// let mut start: Instant = Instant::now();
// if let Err(e) = execute_command(
// ROOT_TMP_CACHE_DIR,
// "rclone",
// vec![
// "copyto",
// &format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}"),
// &format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}"),
// "-v",
// "--size-only",
// "--fast-list",
// ],
// )
// .await
// {
// tracing::info!("Failed copying deno,go,bun tar from cache. Error: {:?}", e);
// return;
// }
// tracing::info!(
// "Finished copying denogobun tar for from bucket as tar. took {}s",
// start.elapsed().as_secs()
// );
// start = Instant::now();
// if let Err(e) = execute_command(
// ROOT_CACHE_DIR,
// "tar",
// vec![
// "-xpvf",
// &format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}"),
// ],
// )
// .await
// {
// tracing::info!("Failed to untar denogobun tar to cache. Error: {:?}", e);
// return;
// }
// tracing::info!(
// "Finished untaring denogobun tar to cache. took: {}s",
// start.elapsed().as_secs()
// );
// start = Instant::now();
// if let Err(e) = execute_command(
// ROOT_TMP_CACHE_DIR,
// "tar",
// vec![
// "-xpvf",
// &format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}"),
// ],
// )
// .await
// {
// tracing::info!("Failed to untar denogobun tar to tmpcache. Error: {:?}", e);
// return;
// }
// tracing::info!(
// "Finished untaring denogobun tar to /tmpcache. took: {}s",
// start.elapsed().as_secs()
// );
// if let Err(e) =
// tokio::fs::remove_file(format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}")).await
// {
// tracing::info!("Failed to remove denogobuntar cache. Error: {:?}", e);
// };
// }
#[cfg(feature = "enterprise")]
pub async fn copy_all_piptars_from_bucket(bucket: &str) {
tracing::info!("Copying all piptars cache from bucket {bucket}");
let start = Instant::now();
let bucket = bucket.trim_start_matches("s3://");
if let Err(e) = execute_command(
ROOT_CACHE_DIR,
"rclone",
vec![
"copy",
&format!(":s3,env_auth=true:{bucket}/tar/pip/"),
&TAR_PIP_TMP_CACHE_DIR,
"-v",
"--size-only",
"--fast-list",
],
)
.await
{
tracing::info!("Failed transferring all piptars from cache. Error: {:?}", e);
return;
}
tracing::info!(
"Finished transferring piptars from bucket {bucket} as tar, took: {:?}s",
start.elapsed().as_secs()
);
}
// #[cfg(feature = "enterprise")]
// 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,
// // "-l",
// "--filter",
// "+ deno/npm/**",
// "--filter",
// "+ deno/deps/**",
// // "--filter",
// // "+ bun/**",
// "--filter",
// "+ go/**",
// "--filter",
// "- *",
// ],
// )
// .await?;
// tracing::info!(
// "Finished copying local tmp cache to local cache. Took {}ms",
// start.elapsed().as_millis(),
// );
// if let Err(e) = untar_all_piptars().await {
// tracing::info!("Failed to untar piptars. Error: {:?}", e);
// }
// Ok(())
// }
#[cfg(feature = "enterprise")]
pub async fn untar_all_piptars() -> error::Result<()> {
use tokio::fs::{self, metadata};
use crate::PIP_CACHE_DIR;
let start: Instant = Instant::now();
let mut entries = fs::read_dir(TAR_PIP_TMP_CACHE_DIR).await?;
while let Some(entry) = entries.next_entry().await? {
if let Err(e) = {
let entry_path = entry.path();
let path = entry_path.to_str().expect("Could not convert path to str");
let folder = format!(
"{PIP_CACHE_DIR}/{}",
path.split('/')
.last()
.unwrap()
.strip_suffix(".tar")
.ok_or_else(|| error::Error::InternalErr(format!(
"Unexpected path file not ending in .tar for: {}",
path
)))?
);
if metadata(&folder).await.is_ok() {
continue;
}
// tracing::info!("A: {path} {folder}");
extract_pip_tar(&path, &folder).await?;
Ok(()) as error::Result<()>
} {
tracing::info!("Failed to extract pip tar. Error: {:?}", e);
}
}
tracing::info!(
"Finished untarring all piptars. Took {}ms",
start.elapsed().as_millis(),
);
Ok(())
}
#[cfg(feature = "enterprise")]
pub async fn extract_pip_tar(tar: &str, folder: &str) -> error::Result<()> {
use tokio::fs;
@@ -515,37 +140,6 @@ pub async fn extract_pip_tar(tar: &str, folder: &str) -> error::Result<()> {
Ok(())
}
// #[cfg(feature = "enterprise")]
// 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,
// // "-l",
// "--filter",
// "+ deno/npm/**",
// "--filter",
// "+ deno/deps/**",
// // "--filter",
// // "+ bun/**",
// "--filter",
// "+ go/**",
// "--filter",
// "- *",
// ],
// )
// .await?;
// tracing::info!(
// "Finished copying local cache to local tmp cache. Took {}ms",
// start.elapsed().as_millis()
// );
// Ok(())
// }
#[cfg(feature = "enterprise")]
pub async fn execute_command(dir: &str, command: &str, args: Vec<&str>) -> error::Result<()> {
tracing::info!("Executing command: {command} {}", args.iter().join(" "));
backend/windmill-worker/src/worker.rs
+1 -22
View File
@@ -34,8 +34,7 @@ use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
use windmill_parser_py_imports::parse_relative_imports;
use uuid::Uuid;
#[cfg(feature = "enterprise")]
use windmill_common::ee::{get_license_plan, LicensePlan};
use windmill_common::{
error::{self, to_anyhow, Error},
flows::{FlowModule, FlowModuleValue, FlowValue},
@@ -59,9 +58,6 @@ use windmill_queue::{
#[cfg(feature = "prometheus")]
use windmill_queue::register_metric;
#[cfg(feature = "enterprise")]
use windmill_common::s3_helpers::S3_CACHE_BUCKET;
use serde_json::{json, value::RawValue, Value};
#[cfg(any(target_os = "linux", target_os = "macos"))]
@@ -86,9 +82,6 @@ use async_recursion::async_recursion;
use rand::Rng;
#[cfg(feature = "enterprise")]
use crate::global_cache::{copy_all_piptars_from_bucket, untar_all_piptars};
use windmill_queue::{add_completed_job, add_completed_job_error};
use crate::{
@@ -940,20 +933,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
ws.inc();
}
#[cfg(feature = "enterprise")]
if i_worker == 1 {
if let Some(bucket) = S3_CACHE_BUCKET.read().await.clone() {
if matches!(get_license_plan().await, LicensePlan::Pro) {
tracing::warn!("S3 cache not available in the pro plan");
} else if crate::global_cache::worker_s3_bucket_sync_enabled(&db).await {
copy_all_piptars_from_bucket(&bucket).await;
if let Err(e) = untar_all_piptars().await {
tracing::error!("Failed to untar pip tarballs: {:?}", e);
}
}
}
}
let (same_worker_tx, mut same_worker_rx) = mpsc::channel::<SameWorkerPayload>(5);
let (job_completed_tx, mut job_completed_rx) = mpsc::channel::<SendResult>(3);