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

feat(backend): install python scripts on save

Browse Source
This commit is contained in:
Ruben Fiszel
2023-04-13 22:24:36 +02:00
parent c2b66fef4f
commit 556dbce239
3 changed files with 44 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2 -1
View File
@@ -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
backend/windmill-worker/src/python_executor.rs
+15 -21
View File
@@ -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<sqlx::Postgres>,
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?;
backend/windmill-worker/src/worker.rs
+27 -6
View File
@@ -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::<u64>().ok())
.unwrap_or(60 * 10);
}
//only matter if CLOUD_HOSTED
@@ -421,7 +425,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
#[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::<Uuid>(5);
@@ -456,7 +460,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
}
#[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(