From 556dbce239f39bf137c954c53b2b576227ee3d1f Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 13 Apr 2023 22:24:36 +0200 Subject: [PATCH] feat(backend): install python scripts on save --- README.md | 3 +- .../windmill-worker/src/python_executor.rs | 36 ++++++++----------- backend/windmill-worker/src/worker.rs | 33 +++++++++++++---- 3 files changed, 44 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index fedd669e71..70ecf2f7b8 100644 --- a/README.md +++ b/README.md @@ -334,7 +334,8 @@ it being synced automatically everyday. | INCLUDE_HEADERS | None | Whitelist of headers that are passed to jobs as args (separated by a comma) | Server | | WHITELIST_WORKSPACES | None | Whitelist of workspaces this worker takes job from | Worker | | BLACKLIST_WORKSPACES | None | Blacklist of workspaces this worker takes job from | Worker | -| INSTANCE_EVENTS_WEBHOOK | None | Webhook to notify of events such as new user added, signup/invite. Can hook back to windmill to send emails | Server | +| INSTANCE_EVENTS_WEBHOOK | None | Webhook to notify of events such as new user added, signup/invite. Can hook back to windmill to send emails | +| GLOBAL_CACHE_INTERVAL | 10\*60 | (Enterprise Edition only) Interval in seconds in between bucket sync of the cache. This interval \* 2 is the time at which you're guaranteed all the worker's caches are synced together. | Worker | ## Run a local dev setup diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 101183c3c5..bf4d2e5f15 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -11,7 +11,8 @@ use tokio::{ use uuid::Uuid; use windmill_common::{ error::{self, Error}, - utils::calculate_hash, jobs::QueuedJob, + jobs::QueuedJob, + utils::calculate_hash, }; lazy_static::lazy_static! { @@ -192,7 +193,8 @@ pub async fn handle_python_job( .split("\n") .filter(|x| !x.starts_with("--")) .collect(), - job, + &job.id, + &job.workspace_id, logs, db, worker_name, @@ -423,9 +425,10 @@ mount {{ read_result(job_dir).await } -async fn handle_python_reqs( +pub async fn handle_python_reqs( requirements: Vec<&str>, - job: &QueuedJob, + job_id: &Uuid, + w_id: &str, logs: &mut String, db: &sqlx::Pool, worker_name: &str, @@ -458,16 +461,16 @@ async fn handle_python_reqs( tracing::info!( worker_name = %worker_name, - job_id = %job.id, - workspace_id = %job.workspace_id, + job_id = %job_id, + workspace_id = %w_id, "started setup python dependencies" ); let child = if !*DISABLE_NSJAIL { tracing::info!( worker_name = %worker_name, - job_id = %job.id, - workspace_id = %job.workspace_id, + job_id = %job_id, + workspace_id = %w_id, "starting nsjail" ); let mut vars = vars.clone(); @@ -515,23 +518,14 @@ async fn handle_python_reqs( .spawn()? }; - let child = handle_child( - &job.id, - db, - logs, - child, - false, - worker_name, - &job.workspace_id, - ) - .await; + let child = handle_child(&job_id, db, logs, child, false, worker_name, &w_id).await; tracing::info!( worker_name = %worker_name, - job_id = %job.id, - workspace_id = %job.workspace_id, + job_id = %job_id, + workspace_id = %w_id, is_ok = child.is_ok(), "finished setting up python dependencies {}", - job.id + job_id ); child?; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 0130026684..8460b58b93 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -57,7 +57,7 @@ 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}, 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::{move_tmp_cache_to_cache}, go_executor::{handle_go_job, install_go_dependencies}, }; @@ -131,8 +131,6 @@ pub const GO_TMP_CACHE_DIR: &str = concatcp!(ROOT_TMP_CACHE_DIR, "go"); const NUM_SECS_PING: u64 = 5; -#[cfg(feature = "enterprise")] -const NUM_SECS_SYNC: u64 = 60 * 10; const INCLUDE_DEPS_PY_SH_CONTENT: &str = include_str!("../nsjail/download_deps.py.sh"); const NSJAIL_CONFIG_RUN_BASH_CONTENT: &str = include_str!("../nsjail/run.bash.config.proto"); @@ -225,6 +223,12 @@ lazy_static::lazy_static! { static ref TIMEOUT_DURATION: Duration = Duration::from_secs(*TIMEOUT as u64); pub static ref SESSION_TOKEN_EXPIRY: i32 = (*TIMEOUT as i32) * 2; + + pub static ref GLOBAL_CACHE_INTERVAL: u64 = std::env::var("GLOBAL_CACHE_INTERVAL") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(60 * 10); + } //only matter if CLOUD_HOSTED @@ -421,7 +425,7 @@ pub async fn run_worker( #[cfg(feature = "enterprise")] let mut last_sync = - Instant::now() + Duration::from_secs(rand::thread_rng().gen_range(0..NUM_SECS_SYNC)); + Instant::now() + Duration::from_secs(rand::thread_rng().gen_range(0..*GLOBAL_CACHE_INTERVAL)); let (same_worker_tx, mut same_worker_rx) = mpsc::channel::(5); @@ -456,7 +460,7 @@ pub async fn run_worker( } #[cfg(feature = "enterprise")] - if initialized_cache && last_sync.elapsed().as_secs() > NUM_SECS_SYNC { + if initialized_cache && last_sync.elapsed().as_secs() > *GLOBAL_CACHE_INTERVAL { if let Some(ref s) = S3_CACHE_BUCKET.clone() { copy_cache_from_bucket(&s, None).await; copy_cache_to_bucket(&s).await; @@ -1572,7 +1576,24 @@ async fn capture_dependency_job( match job_language { ScriptLang::Python3 => { create_dependencies_dir(job_dir).await; - pip_compile(job_id, job_raw_code, logs, job_dir, db, worker_name, w_id).await + let req = pip_compile(job_id, job_raw_code, logs, job_dir, db, worker_name, w_id).await; + // install the dependencies to pre-fill the cache + if let Ok(req) = req.as_ref() { + handle_python_reqs( + req + .split("\n") + .filter(|x| !x.starts_with("--")) + .collect(), + job_id, + w_id, + logs, + db, + worker_name, + job_dir, + ) + .await?; + } + req } ScriptLang::Go => { install_go_dependencies(