feat: use flock to avoid concurrency issues on pip and shared volume

backend/parsers/windmill-parser-py-imports/src/lib.rs

@@ -49,9 +49,9 @@ lazy_static! {
 
 pub async fn parse_python_imports(
     code: &str,
-    w_id: &str,
-    path: &str,
-    db: &Pool<Postgres>,
+    _w_id: &str,
+    _path: &str,
+    _db: &Pool<Postgres>,
 ) -> error::Result<Vec<String>> {
     let find_requirements = code
         .lines()

backend/src/main.rs

@@ -23,8 +23,8 @@ use windmill_api::{LICENSE_KEY, OAUTH_CLIENTS, SMTP_CLIENT};
 use windmill_common::{utils::rd_string, METRICS_ADDR};
 use windmill_worker::{
     BUN_CACHE_DIR, BUN_TMP_CACHE_DIR, DENO_CACHE_DIR, DENO_TMP_CACHE_DIR, GO_CACHE_DIR,
-    GO_TMP_CACHE_DIR, HUB_CACHE_DIR, HUB_TMP_CACHE_DIR, PIP_CACHE_DIR, ROOT_TMP_CACHE_DIR,
-    TAR_PIP_TMP_CACHE_DIR,
+    GO_TMP_CACHE_DIR, HUB_CACHE_DIR, HUB_TMP_CACHE_DIR, LOCK_CACHE_DIR, PIP_CACHE_DIR,
+    ROOT_TMP_CACHE_DIR, TAR_PIP_TMP_CACHE_DIR,
 };
 
 const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
@@ -340,6 +340,7 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
     }
 
     for x in [
+        LOCK_CACHE_DIR,
         PIP_CACHE_DIR,
         DENO_CACHE_DIR,
         BUN_CACHE_DIR,

backend/windmill-worker/src/python_executor.rs

@@ -19,6 +19,10 @@ lazy_static::lazy_static! {
     static ref PYTHON_PATH: String =
     std::env::var("PYTHON_PATH").unwrap_or_else(|_| "/usr/local/bin/python3".to_string());
 
+    static ref FLOCK_PATH: String =
+    std::env::var("FLOCK_PATH").unwrap_or_else(|_| "/usr/bin/flock".to_string());
+
+
 
     static ref PIP_INDEX_URL: Option<String> = std::env::var("PIP_INDEX_URL").ok();
     static ref PIP_EXTRA_INDEX_URL: Option<String> = std::env::var("PIP_EXTRA_INDEX_URL").ok();
@@ -55,8 +59,8 @@ use crate::S3_CACHE_BUCKET;
 use crate::{
     common::{read_result, set_logs},
     create_args_and_out_file, get_reserved_variables, handle_child, write_file,
-    AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HTTPS_PROXY, HTTP_PROXY, NO_PROXY,
-    NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR,
+    AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HTTPS_PROXY, HTTP_PROXY,
+    LOCK_CACHE_DIR, NO_PROXY, NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR,
 };
 
 pub async fn create_dependencies_dir(job_dir: &str) {
@@ -569,7 +573,8 @@ pub async fn handle_python_reqs(
                 .stderr(Stdio::piped())
                 .spawn()?
         } else {
-            let mut args = vec![
+            let mut command_args = vec![
+                PYTHON_PATH.as_str(),
                 "-m",
                 "pip",
                 "install",
@@ -584,13 +589,13 @@ pub async fn handle_python_reqs(
                 venv_p.as_str(),
             ];
             if let Some(url) = PIP_EXTRA_INDEX_URL.as_ref() {
-                args.extend(["--extra-index-url", url]);
+                command_args.extend(["--extra-index-url", url]);
             }
             if let Some(url) = PIP_INDEX_URL.as_ref() {
-                args.extend(["--index-url", url]);
+                command_args.extend(["--index-url", url]);
             }
             if let Some(host) = PIP_TRUSTED_HOST.as_ref() {
-                args.extend(["--trusted-host", &host]);
+                command_args.extend(["--trusted-host", &host]);
             }
             let mut envs = vec![("PATH", PATH_ENV.as_str())];
             if let Some(http_proxy) = HTTP_PROXY.as_ref() {
@@ -603,10 +608,15 @@ pub async fn handle_python_reqs(
                 envs.push(("NO_PROXY", no_proxy));
             }
 
-            Command::new(PYTHON_PATH.as_str())
+            Command::new(FLOCK_PATH.as_str())
                 .env_clear()
                 .envs(envs)
-                .args(args)
+                .args([
+                    "-x",
+                    &format!("{}/pip-{}.lock", LOCK_CACHE_DIR, req),
+                    "--command",
+                    &command_args.join(" "),
+                ])
                 .stdout(Stdio::piped())
                 .stderr(Stdio::piped())
                 .spawn()?

backend/windmill-worker/src/worker.rs

@@ -130,6 +130,7 @@ pub async fn create_token_for_owner(
 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 LOCK_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "lock");
 pub const PIP_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "pip");
 pub const DENO_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "deno");
 pub const GO_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "go");

docker-compose.yml

@@ -72,6 +72,7 @@ services:
       - /var/run/docker.sock:/var/run/docker.sock
       # See Oauth (https://docs.windmill.dev/docs/misc/setup_oauth)
       - ./oauth.json:/usr/src/app/oauth.json
+      - worker_dependency_cache:/tmp/windmill/cache
 
   ## This worker is specialized for "native" jobs. Jobs that are lightweight and run "in-process" and can thus be parallelized to more than 1 at a time on a given worker
   windmill_worker_native: