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

feat(python): Multiple runtime versions (#4579)

Browse Source 
* feat: Handle `pip install` by `uv`

Dirty and untested, but already something working

* Integrate with NSJAIL and prepare fallbacks

* Refactor fallback
no_uv disable compile and install
where no_uv_install and no_uv_compile are a bit more specific

* Remove `--disable-pip-version-check`
Reason:
   warning: pip's `--disable-pip-version-check` has no effect

* Fix backend compilation error

* Pip fallback overwrite UV's cache

* Initially refactor cache (No S3)

* Support S3

* Remove unused import

* Handle flags for NSJAIL

* Return deleted flag

* Remove verbose mode and enable link-mode=copy

* Granural migration of lockfiles

Before i realized we dont need it :)

* Initial draft (not-working)

* Add fallback

* Fix bug preventing uv from installing deps

'\n' - Love it

* Add verbosity indicator

* Iterate on feature
- Added instance python version
- Rework logic

* Fix EE build error
error[E0599]: no method named `iter` found for tuple `(PyVersion, std::vec::Vec<std::string::String>)` in the current scope

* Support S3

* Support NSJAIL

* Refactor `get_python`

* Make NSJAIL work [Unsafe]

config file missed /proc mount causing install phase to fail

* Trigger CI

* Clean up

* Make Actions build it

* Trigger CI #2

* Update Dockerfile and clean up

* Change fallbacks
now there is only no_uv and NOUV

* Expose INSTANCE_PYTHON_VERSION through env variable

* Change namings

* Include py-version to requirements.in

Also add comments and make code much cleaner

* Use const for python installation dir

It was hardcoded before

* Pin preinstalled version

* Update python_executor.rs

* Up to date branch

* Create PYCACHE dirs

TODO: PY_TAR_DIRS

* Fix after merge

* Make it safer

* Implement USE_SYSTEM_PYTHON

* Implement latest_stable option

* Load INSTANCE_PYTHON_VERSION on startup

* Check for multiple annotations used

* Fix Latest Stable button not pressed if selected

* Proper error handling for conflict on multiple annotations

* Fix merge conflicts

* Preinstall 3.11 and Latest Stable

* Preinstall latest stable in non-blocking manner

* Fix Warning

* Gate preinstall logic behind "python" feature

* Handle raw_deps properly

* Make it work with nsjail

* Revert docker-image.yml

* Revert Dockerfile

* Cleanup + Fixing

* Add windows support

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
pyranota
2025-01-24 13:42:48 +03:00
committed by GitHub
parent fe96d2be3a
commit 3af3fc898b
19 changed files with 802 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+12
View File
@@ -95,6 +95,14 @@ ARG WITH_KUBECTL=true
ARG WITH_HELM=true
ARG WITH_GIT=true
# To change latest stable version:
# 1. Change placeholder in instanceSettings.ts
# 2. Change LATEST_STABLE_PY in dockerfile
# 3. Change #[default] annotation for PyVersion in backend
ARG LATEST_STABLE_PY=3.11.10
ENV UV_PYTHON_INSTALL_DIR=/tmp/windmill/cache/py_runtime
ENV UV_PYTHON_PREFERENCE=only-managed
RUN pip install --upgrade pip==24.2
RUN apt-get update \
@@ -161,6 +169,10 @@ ENV GO_PATH=/usr/local/go/bin/go
# Install UV
RUN curl --proto '=https' --tlsv1.2 -LsSf https://github.com/astral-sh/uv/releases/download/0.5.15/uv-installer.sh | sh && mv /root/.local/bin/uv /usr/local/bin/uv
# Preinstall python runtimes
RUN uv python install 3.11.10
RUN uv python install $LATEST_STABLE_PY
RUN curl -sL https://deb.nodesource.com/setup_20.x | bash -
RUN apt-get -y update && apt-get install -y curl procps nodejs awscli && apt-get clean \
&& rm -rf /var/lib/apt/lists/*
backend/parsers/windmill-parser-py-imports/src/lib.rs
+74 -3
View File
@@ -21,7 +21,7 @@ use rustpython_parser::{
Parse,
};
use sqlx::{Pool, Postgres};
use windmill_common::error;
use windmill_common::{error, worker::PythonAnnotations};
const DEF_MAIN: &str = "def main(";
@@ -171,14 +171,75 @@ fn parse_code_for_imports(code: &str, path: &str) -> error::Result<Vec<String>>
return Ok(nimports);
}
#[async_recursion]
pub async fn parse_python_imports(
code: &str,
w_id: &str,
path: &str,
db: &Pool<Postgres>,
already_visited: &mut Vec<String>,
annotated_pyv_numeric: &mut Option<u32>,
) -> error::Result<Vec<String>> {
parse_python_imports_inner(
code,
w_id,
path,
db,
already_visited,
annotated_pyv_numeric,
&mut annotated_pyv_numeric.and_then(|_| Some(path.to_owned())),
)
.await
}
#[async_recursion]
async fn parse_python_imports_inner(
code: &str,
w_id: &str,
path: &str,
db: &Pool<Postgres>,
already_visited: &mut Vec<String>,
annotated_pyv_numeric: &mut Option<u32>,
path_where_annotated_pyv: &mut Option<String>,
) -> error::Result<Vec<String>> {
let PythonAnnotations { py310, py311, py312, py313, .. } = PythonAnnotations::parse(&code);
// we pass only if there is none or only one annotation
// Naive:
// 1. Check if there are multiple annotated version
// 2. If no, take one and compare with annotated version
// 3. We continue if same or replace none with new one
// Optimized:
// 1. Iterate over all annotations compare each with annotated_pyv and replace on flight
// 2. If annotated_pyv is different version, throw and error
// This way we make sure there is no multiple annotations for same script
// and we get detailed span on conflicting versions
let mut check = |is_py_xyz, numeric| -> error::Result<()> {
if is_py_xyz {
if let Some(v) = annotated_pyv_numeric {
if *v != numeric {
return Err(error::Error::from(anyhow::anyhow!(
"Annotated 2 or more different python versions: \n - py{v} at {}\n - py{numeric} at {path}\nIt is possible to use only one.",
path_where_annotated_pyv.clone().unwrap_or("Unknown".to_owned())
)));
}
} else {
*annotated_pyv_numeric = Some(numeric);
}
*path_where_annotated_pyv = Some(path.to_owned());
}
Ok(())
};
check(py310, 310)?;
check(py311, 311)?;
check(py312, 312)?;
check(py313, 313)?;
let find_requirements = code
.lines()
.find_position(|x| x.starts_with("#requirements:") || x.starts_with("# requirements:"));
@@ -225,11 +286,21 @@ pub async fn parse_python_imports(
.fetch_optional(db)
.await?
.unwrap_or_else(|| "".to_string());
if already_visited.contains(&rpath) {
vec![]
} else {
already_visited.push(rpath.clone());
parse_python_imports(&code, w_id, &rpath, db, already_visited).await?
parse_python_imports_inner(
&code,
w_id,
&rpath,
db,
already_visited,
annotated_pyv_numeric,
path_where_annotated_pyv,
)
.await?
}
} else {
vec![replace_import(n.to_string())]
backend/parsers/windmill-parser-py-imports/tests/tests.rs
+3
View File
@@ -25,6 +25,7 @@ def main():
"f/foo/bar",
&db,
&mut already_visited,
&mut None,
)
.await?;
// println!("{}", serde_json::to_string(&r)?);
@@ -57,6 +58,7 @@ def main():
"f/foo/bar",
&db,
&mut already_visited,
&mut None,
)
.await?;
println!("{}", serde_json::to_string(&r)?);
@@ -87,6 +89,7 @@ def main():
"f/foo/bar",
&db,
&mut already_visited,
&mut None,
)
.await?;
println!("{}", serde_json::to_string(&r)?);
backend/src/main.rs
+17 -6
View File
@@ -9,8 +9,9 @@
use anyhow::Context;
use monitor::{
load_base_url, load_otel, reload_delete_logs_periodically_setting, reload_indexer_config,
reload_nuget_config_setting, reload_timeout_wait_result_setting,
send_current_log_file_to_object_store, send_logs_to_object_store,
reload_instance_python_version_setting, reload_nuget_config_setting,
reload_timeout_wait_result_setting, send_current_log_file_to_object_store,
send_logs_to_object_store,
};
use rand::Rng;
use sqlx::{postgres::PgListener, Pool, Postgres};
@@ -35,7 +36,7 @@ use windmill_common::{
CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
DEFAULT_TAGS_WORKSPACES_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING,
EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING,
JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING,
INSTANCE_PYTHON_VERSION_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING,
LICENSE_KEY_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING,
NUGET_CONFIG_SETTING, OAUTH_SETTING, OTEL_SETTING, PIP_INDEX_URL_SETTING,
REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
@@ -69,8 +70,9 @@ use windmill_worker::{
get_hub_script_content_and_requirements, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR,
BUN_DEPSTAR_CACHE_DIR, CSHARP_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS,
DENO_CACHE_DIR_NPM, GO_BIN_CACHE_DIR, GO_CACHE_DIR, LOCK_CACHE_DIR, PIP_CACHE_DIR,
POWERSHELL_CACHE_DIR, PY311_CACHE_DIR, RUST_CACHE_DIR, TAR_PIP_CACHE_DIR, TAR_PY311_CACHE_DIR,
TMP_LOGS_DIR, UV_CACHE_DIR,
POWERSHELL_CACHE_DIR, PY310_CACHE_DIR, PY311_CACHE_DIR, PY312_CACHE_DIR, PY313_CACHE_DIR,
RUST_CACHE_DIR, TAR_PIP_CACHE_DIR, TAR_PY310_CACHE_DIR, TAR_PY311_CACHE_DIR,
TAR_PY312_CACHE_DIR, TAR_PY313_CACHE_DIR, TMP_LOGS_DIR, UV_CACHE_DIR,
};
use crate::monitor::{
@@ -765,6 +767,9 @@ Windmill Community Edition {GIT_VERSION}
PIP_INDEX_URL_SETTING => {
reload_pip_index_url_setting(&db).await
},
INSTANCE_PYTHON_VERSION_SETTING => {
reload_instance_python_version_setting(&db).await
},
NPM_CONFIG_REGISTRY_SETTING => {
reload_npm_config_registry_setting(&db).await
},
@@ -1016,12 +1021,18 @@ pub async fn run_workers(
TMP_LOGS_DIR,
UV_CACHE_DIR,
TAR_PIP_CACHE_DIR,
TAR_PY311_CACHE_DIR,
DENO_CACHE_DIR,
DENO_CACHE_DIR_DEPS,
DENO_CACHE_DIR_NPM,
BUN_CACHE_DIR,
PY310_CACHE_DIR,
PY311_CACHE_DIR,
PY312_CACHE_DIR,
PY313_CACHE_DIR,
TAR_PY310_CACHE_DIR,
TAR_PY311_CACHE_DIR,
TAR_PY312_CACHE_DIR,
TAR_PY313_CACHE_DIR,
PIP_CACHE_DIR,
BUN_DEPSTAR_CACHE_DIR,
BUN_BUNDLE_CACHE_DIR,
backend/src/monitor.rs
+17 -6
View File
@@ -41,10 +41,10 @@ use windmill_common::{
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
CRITICAL_ERROR_CHANNELS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
DEFAULT_TAGS_WORKSPACES_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING,
MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING, NUGET_CONFIG_SETTING,
OTEL_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, INSTANCE_PYTHON_VERSION_SETTING,
JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING,
LICENSE_KEY_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING,
NUGET_CONFIG_SETTING, OTEL_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
},
@@ -68,8 +68,8 @@ use windmill_common::{
use windmill_queue::cancel_job;
use windmill_worker::{
create_token_for_owner, handle_job_error, AuthedClient, SameWorkerPayload, SameWorkerSender,
SendResult, BUNFIG_INSTALL_SCOPES, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, NPM_CONFIG_REGISTRY,
NUGET_CONFIG, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, SCRIPT_TOKEN_EXPIRY,
SendResult, BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR,
NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, SCRIPT_TOKEN_EXPIRY,
};
#[cfg(feature = "parquet")]
@@ -198,6 +198,7 @@ pub async fn initial_load(
reload_pip_index_url_setting(&db).await;
reload_npm_config_registry_setting(&db).await;
reload_bunfig_install_scopes_setting(&db).await;
reload_instance_python_version_setting(&db).await;
reload_nuget_config_setting(&db).await;
}
}
@@ -908,6 +909,16 @@ pub async fn reload_pip_index_url_setting(db: &DB) {
.await;
}
pub async fn reload_instance_python_version_setting(db: &DB) {
reload_option_setting_with_tracing(
db,
INSTANCE_PYTHON_VERSION_SETTING,
"INSTANCE_PYTHON_VERSION",
INSTANCE_PYTHON_VERSION.clone(),
)
.await;
}
pub async fn reload_npm_config_registry_setting(db: &DB) {
reload_option_setting_with_tracing(
db,
backend/windmill-common/src/error.rs
+2
View File
@@ -66,6 +66,8 @@ pub enum Error {
AiError(String),
#[error("{0}")]
AlreadyCompleted(String),
#[error("Find python error: {0}")]
FindPythonError(String),
#[error("{0}")]
Utf8(#[from] std::string::FromUtf8Error),
#[error("Encoding/decoding error: {0}")]
backend/windmill-common/src/global_settings.rs
+3 -1
View File
@@ -14,6 +14,7 @@ pub const NUGET_CONFIG_SETTING: &str = "nuget_config";
pub const EXTRA_PIP_INDEX_URL_SETTING: &str = "pip_extra_index_url";
pub const PIP_INDEX_URL_SETTING: &str = "pip_index_url";
pub const INSTANCE_PYTHON_VERSION_SETTING: &str = "instance_python_version";
pub const SCIM_TOKEN_SETTING: &str = "scim_token";
pub const SAML_METADATA_SETTING: &str = "saml_metadata";
pub const SMTP_SETTING: &str = "smtp_settings";
@@ -39,7 +40,7 @@ pub const JWT_SECRET_SETTING: &str = "jwt_secret";
pub const EMAIL_DOMAIN_SETTING: &str = "email_domain";
pub const OTEL_SETTING: &str = "otel";
pub const ENV_SETTINGS: [&str; 54] = [
pub const ENV_SETTINGS: [&str; 55] = [
"DISABLE_NSJAIL",
"MODE",
"NUM_WORKERS",
@@ -62,6 +63,7 @@ pub const ENV_SETTINGS: [&str; 54] = [
"GOPRIVATE",
"GOPROXY",
"NETRC",
"INSTANCE_PYTHON_VERSION",
"PIP_INDEX_URL",
"PIP_EXTRA_INDEX_URL",
"PIP_TRUSTED_HOST",
backend/windmill-common/src/worker.rs
+5 -1
View File
@@ -333,14 +333,18 @@ fn parse_file<T: FromStr>(path: &str) -> Option<T> {
.flatten()
}
#[derive(Copy, Clone)]
#[annotations("#")]
pub struct PythonAnnotations {
pub no_cache: bool,
pub no_uv: bool,
pub no_uv_install: bool,
pub no_uv_compile: bool,
pub no_postinstall: bool,
pub py310: bool,
pub py311: bool,
pub py312: bool,
pub py313: bool,
}
#[annotations("//")]
backend/windmill-worker/nsjail/download.py.config.proto
+6
View File
@@ -20,6 +20,7 @@ clone_newuser: {CLONE_NEWUSER}
keep_caps: true
keep_env: true
mount_proc: true
mount {
src: "/bin"
@@ -79,6 +80,11 @@ mount {
is_bind: true
rw: true
}
mount {
src: "{PY_INSTALL_DIR}"
dst: "{PY_INSTALL_DIR}"
is_bind: true
}
mount {
src: "/dev/urandom"
backend/windmill-worker/nsjail/download_deps.py.sh
+1
View File
@@ -27,6 +27,7 @@ CMD="/usr/local/bin/uv pip install
--no-color
--no-deps
--link-mode=copy
$PY_PATH
$INDEX_URL_ARG $EXTRA_INDEX_URL_ARG $TRUSTED_HOST_ARG
--index-strategy unsafe-best-match
--system
backend/windmill-worker/nsjail/run.python3.config.proto
+7
View File
@@ -16,6 +16,7 @@ clone_newuser: {CLONE_NEWUSER}
keep_caps: false
keep_env: true
mount_proc: true
mount {
src: "/bin"
@@ -110,6 +111,12 @@ mount {
is_bind: true
}
mount {
src: "{PY_INSTALL_DIR}"
dst: "{PY_INSTALL_DIR}"
is_bind: true
}
mount {
src: "/dev/urandom"
dst: "/dev/urandom"
backend/windmill-worker/src/ansible_executor.rs
+3 -2
View File
@@ -32,7 +32,7 @@ use crate::{
check_executor_binary_exists, get_reserved_variables, read_and_check_result, start_child_process, transform_json, OccupancyMetrics
},
handle_child::handle_child,
python_executor::{create_dependencies_dir, handle_python_reqs, uv_pip_compile},
python_executor::{create_dependencies_dir, handle_python_reqs, uv_pip_compile, PyVersion},
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV,
PROXY_ENVS, TZ_ENV,
};
@@ -88,6 +88,7 @@ async fn handle_ansible_python_deps(
worker_name,
w_id,
&mut Some(occupancy_metrics),
PyVersion::Py311,
false,
false,
)
@@ -115,7 +116,7 @@ async fn handle_ansible_python_deps(
job_dir,
worker_dir,
&mut Some(occupancy_metrics),
false,
crate::python_executor::PyVersion::Py311,
false,
)
.await?;
backend/windmill-worker/src/global_cache.rs
+12 -6
View File
@@ -17,19 +17,24 @@ use std::sync::Arc;
pub async fn build_tar_and_push(
s3_client: Arc<dyn ObjectStore>,
folder: String,
// python_311
python_xyz: String,
no_uv: bool,
) -> error::Result<()> {
use object_store::path::Path;
use crate::{TAR_PIP_CACHE_DIR, TAR_PY311_CACHE_DIR};
use crate::{TAR_PIP_CACHE_DIR, TAR_PYBASE_CACHE_DIR};
tracing::info!("Started building and pushing piptar {folder}");
let start = Instant::now();
// e.g. tiny==1.0.0
let folder_name = folder.split("/").last().unwrap();
let prefix = if no_uv {
TAR_PIP_CACHE_DIR
} else {
TAR_PY311_CACHE_DIR
&format!("{TAR_PYBASE_CACHE_DIR}/{}", python_xyz)
};
let tar_path = format!("{prefix}/{folder_name}_tar.tar",);
@@ -53,7 +58,7 @@ pub async fn build_tar_and_push(
.put(
&Path::from(format!(
"/tar/{}/{folder_name}.tar",
if no_uv { "pip" } else { "python_311" }
if no_uv { "pip" } else { &python_xyz }
)),
std::fs::read(&tar_path)?.into(),
)
@@ -82,6 +87,8 @@ pub async fn build_tar_and_push(
pub async fn pull_from_tar(
client: Arc<dyn ObjectStore>,
folder: String,
// python_311
python_xyz: String,
no_uv: bool,
) -> error::Result<()> {
use windmill_common::s3_helpers::attempt_fetch_bytes;
@@ -91,14 +98,13 @@ pub async fn pull_from_tar(
tracing::info!("Attempting to pull piptar {folder_name} from bucket");
let start = Instant::now();
let tar_path = format!(
"tar/{}/{folder_name}.tar",
if no_uv { "pip" } else { "python_311" }
if no_uv { "pip".to_owned() } else { python_xyz }
);
let bytes = attempt_fetch_bytes(client, &tar_path).await?;
// tracing::info!("B: {target} {folder}");
extract_tar(bytes, &folder).await.map_err(|e| {
tracing::error!("Failed to extract piptar {folder_name}. Error: {:?}", e);
e
backend/windmill-worker/src/python_executor.rs
+489 -62
View File
@@ -22,7 +22,10 @@ use uuid::Uuid;
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
use windmill_common::ee::{get_license_plan, LicensePlan};
use windmill_common::{
error::{self, Error},
error::{
self,
Error::{self},
},
jobs::{QueuedJob, PREPROCESSOR_FAKE_ENTRYPOINT},
utils::calculate_hash,
worker::{write_file, PythonAnnotations, WORKER_CONFIG},
@@ -51,18 +54,18 @@ lazy_static::lazy_static! {
static ref PIP_TRUSTED_HOST: Option<String> = std::env::var("PIP_TRUSTED_HOST").ok();
static ref PIP_INDEX_CERT: Option<String> = std::env::var("PIP_INDEX_CERT").ok();
pub static ref USE_SYSTEM_PYTHON: bool = std::env::var("USE_SYSTEM_PYTHON")
.ok().map(|flag| flag == "true").unwrap_or(false);
pub static ref USE_PIP_COMPILE: bool = std::env::var("USE_PIP_COMPILE")
.ok().map(|flag| flag == "true").unwrap_or(false);
/// Use pip install
pub static ref USE_PIP_INSTALL: bool = std::env::var("USE_PIP_INSTALL")
.ok().map(|flag| flag == "true").unwrap_or(false);
static ref RELATIVE_IMPORT_REGEX: Regex = Regex::new(r#"(import|from)\s(((u|f)\.)|\.)"#).unwrap();
static ref EPHEMERAL_TOKEN_CMD: Option<String> = std::env::var("EPHEMERAL_TOKEN_CMD").ok();
}
const NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT: &str = include_str!("../nsjail/download.py.config.proto");
@@ -82,12 +85,292 @@ use crate::{
create_args_and_out_file, get_main_override, get_reserved_variables, read_file,
read_result, start_child_process, OccupancyMetrics,
},
handle_child::{get_mem_peak, handle_child},
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, LOCK_CACHE_DIR,
NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, PROXY_ENVS,
PY311_CACHE_DIR, TZ_ENV, UV_CACHE_DIR,
handle_child::handle_child,
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, INSTANCE_PYTHON_VERSION,
LOCK_CACHE_DIR, NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL,
PROXY_ENVS, PY_INSTALL_DIR, TZ_ENV, UV_CACHE_DIR,
};
// To change latest stable version:
// 1. Change placeholder in instanceSettings.ts
// 2. Change LATEST_STABLE_PY in dockerfile
// 3. Change #[default] annotation for PyVersion in backend
#[derive(Eq, PartialEq, Clone, Copy, Default, Debug)]
pub enum PyVersion {
Py310,
#[default]
Py311,
Py312,
Py313,
}
impl PyVersion {
pub async fn from_instance_version() -> Self {
match INSTANCE_PYTHON_VERSION.read().await.clone() {
Some(v) => PyVersion::from_string_with_dots(&v).unwrap_or_else(|| {
let v = PyVersion::default();
tracing::error!(
"Cannot parse INSTANCE_PYTHON_VERSION ({:?}), fallback to latest_stable ({v:?})",
*INSTANCE_PYTHON_VERSION
);
v
}),
// Use latest stable
None => PyVersion::default(),
}
}
/// e.g.: `/tmp/windmill/cache/python_3xy`
pub fn to_cache_dir(&self) -> String {
use windmill_common::worker::ROOT_CACHE_DIR;
format!("{ROOT_CACHE_DIR}{}", &self.to_cache_dir_top_level())
}
/// e.g.: `python_3xy`
pub fn to_cache_dir_top_level(&self) -> String {
format!("python_{}", self.to_string_no_dot())
}
/// e.g.: `3xy`
pub fn to_string_no_dot(&self) -> String {
self.to_string_with_dot().replace('.', "")
}
/// e.g.: `3.xy`
pub fn to_string_with_dot(&self) -> &str {
use PyVersion::*;
match self {
Py310 => "3.10",
Py311 => "3.11",
Py312 => "3.12",
Py313 => "3.13",
}
}
pub fn from_string_with_dots(value: &str) -> Option<Self> {
use PyVersion::*;
match value {
"3.10" => Some(Py310),
"3.11" => Some(Py311),
"3.12" => Some(Py312),
"3.13" => Some(Py313),
"default" => Some(PyVersion::default()),
_ => {
tracing::warn!(
"Cannot convert string (\"{value}\") to PyVersion\nExpected format x.yz"
);
None
}
}
}
pub fn from_string_no_dots(value: &str) -> Option<Self> {
use PyVersion::*;
match value {
"310" => Some(Py310),
"311" => Some(Py311),
"312" => Some(Py312),
"313" => Some(Py313),
"default" => Some(PyVersion::default()),
_ => {
tracing::warn!(
"Cannot convert string (\"{value}\") to PyVersion\nExpected format xyz"
);
None
}
}
}
/// e.g.: `# py3xy` -> `PyVersion::Py3XY`
pub fn parse_version(line: &str) -> Option<Self> {
Self::from_string_no_dots(line.replace(" ", "").replace("#py", "").as_str())
}
pub fn from_py_annotations(a: PythonAnnotations) -> Option<Self> {
let PythonAnnotations { py310, py311, py312, py313, .. } = a;
use PyVersion::*;
if py313 {
Some(Py313)
} else if py312 {
Some(Py312)
} else if py311 {
Some(Py311)
} else if py310 {
Some(Py310)
} else {
None
}
}
pub fn from_numeric(n: u32) -> Option<Self> {
use PyVersion::*;
match n {
310 => Some(Py310),
311 => Some(Py311),
312 => Some(Py312),
313 => Some(Py313),
_ => None,
}
}
pub fn to_numeric(&self) -> u32 {
use PyVersion::*;
match self {
Py310 => 310,
Py311 => 311,
Py312 => 312,
Py313 => 313,
}
}
pub async fn get_python(
&self,
job_id: &Uuid,
mem_peak: &mut i32,
// canceled_by: &mut Option<CanceledBy>,
db: &Pool<Postgres>,
worker_name: &str,
w_id: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> error::Result<Option<String>> {
// lazy_static::lazy_static! {
// static ref PYTHON_PATHS: Arc<RwLock<HashMap<PyVersion, String>>> = Arc::new(RwLock::new(HashMap::new()));
// }
let res = self
.get_python_inner(job_id, mem_peak, db, worker_name, w_id, occupancy_metrics)
.await;
if let Err(ref e) = res {
tracing::error!(
"worker_name: {worker_name}, w_id: {w_id}, job_id: {job_id}\n
Error while getting python from uv, falling back to system python: {e:?}"
);
}
res
}
async fn get_python_inner(
self,
job_id: &Uuid,
mem_peak: &mut i32,
// canceled_by: &mut Option<CanceledBy>,
db: &Pool<Postgres>,
worker_name: &str,
w_id: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> error::Result<Option<String>> {
let py_path = self.find_python().await;
// Runtime is not installed
if py_path.is_err() {
// Install it
if let Err(err) = self
.install_python(job_id, mem_peak, db, worker_name, w_id, occupancy_metrics)
.await
{
tracing::error!("Cannot install python: {err}");
return Err(err);
} else {
// Try to find one more time
let py_path = self.find_python().await;
if let Err(err) = py_path {
tracing::error!("Cannot find python version {err}");
return Err(err);
}
// TODO: Cache the result
py_path
}
} else {
py_path
}
}
async fn install_python(
self,
job_id: &Uuid,
mem_peak: &mut i32,
// canceled_by: &mut Option<CanceledBy>,
db: &Pool<Postgres>,
worker_name: &str,
w_id: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> error::Result<()> {
let v = self.to_string_with_dot();
append_logs(job_id, w_id, format!("\nINSTALLING PYTHON ({})", v), db).await;
// Create dirs for newly installed python
// If we dont do this, NSJAIL will not be able to mount cache
// For the default version directory created during startup (main.rs)
DirBuilder::new()
.recursive(true)
.create(self.to_cache_dir())
.await
.expect("could not create initial worker dir");
let logs = String::new();
#[cfg(windows)]
let uv_cmd = "uv";
#[cfg(unix)]
let uv_cmd = UV_PATH.as_str();
let mut child_cmd = Command::new(uv_cmd);
child_cmd
.args(["python", "install", v, "--python-preference=only-managed"])
// TODO: Do we need these?
.envs([("UV_PYTHON_INSTALL_DIR", PY_INSTALL_DIR)])
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let child_process = start_child_process(child_cmd, "uv").await?;
append_logs(&job_id, &w_id, logs, db).await;
handle_child(
job_id,
db,
mem_peak,
&mut None,
child_process,
false,
worker_name,
&w_id,
"uv",
None,
false,
occupancy_metrics,
)
.await
}
async fn find_python(self) -> error::Result<Option<String>> {
#[cfg(windows)]
let uv_cmd = "uv";
#[cfg(unix)]
let uv_cmd = UV_PATH.as_str();
let mut child_cmd = Command::new(uv_cmd);
let output = child_cmd
// .current_dir(job_dir)
.args([
"python",
"find",
self.to_string_with_dot(),
"--python-preference=only-managed",
])
.envs([
("UV_PYTHON_INSTALL_DIR", PY_INSTALL_DIR),
("UV_PYTHON_PREFERENCE", "only-managed"),
])
// .stdout(Stdio::piped())
.stderr(Stdio::piped())
.output()
.await?;
// Check if the command was successful
if output.status.success() {
// Convert the output to a String
let stdout =
String::from_utf8(output.stdout).expect("Failed to convert output to String");
return Ok(Some(stdout.replace('\n', "")));
} else {
// If the command failed, print the error
let stderr =
String::from_utf8(output.stderr).expect("Failed to convert error output to String");
return Err(error::Error::FindPythonError(stderr));
}
}
}
#[cfg(windows)]
use crate::SYSTEM_ROOT;
@@ -120,6 +403,10 @@ pub fn handle_ephemeral_token(x: String) -> String {
x
}
// This function only invoked during deployment of script or test run.
// And never for already deployed scripts, these have their lockfiles in PostgreSQL
// thus this function call is skipped.
/// Returns lockfile and python version
pub async fn uv_pip_compile(
job_id: &Uuid,
requirements: &str,
@@ -130,14 +417,16 @@ pub async fn uv_pip_compile(
worker_name: &str,
w_id: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
// Fallback to pip-compile. Will be removed in future
mut no_uv: bool,
py_version: PyVersion,
// Debug-only flag
no_cache: bool,
// Fallback to pip-compile. Will be removed in future
mut no_uv: bool,
) -> error::Result<String> {
let mut logs = String::new();
logs.push_str(&format!("\nresolving dependencies..."));
logs.push_str(&format!("\ncontent of requirements:\n{}\n", requirements));
let requirements = if let Some(pip_local_dependencies) =
WORKER_CONFIG.read().await.pip_local_dependencies.as_ref()
{
@@ -167,6 +456,11 @@ pub async fn uv_pip_compile(
requirements.to_string()
};
// Include python version to requirements.in
// We need it because same hash based on requirements.in can get calculated even for different python versions
// To prevent from overwriting same requirements.in but with different python versions, we include version to hash
let requirements = format!("# py{}\n{}", py_version.to_string_no_dot(), requirements);
#[cfg(feature = "enterprise")]
let requirements = replace_pip_secret(db, w_id, &requirements, worker_name, job_id).await?;
@@ -186,15 +480,21 @@ pub async fn uv_pip_compile(
if !no_cache {
if let Some(cached) = sqlx::query_scalar!(
"SELECT lockfile FROM pip_resolution_cache WHERE hash = $1",
// Python version is included in hash,
// hash will be the different for every python version
req_hash
)
.fetch_optional(db)
.await?
{
logs.push_str(&format!("\nfound cached resolution: {req_hash}"));
logs.push_str(&format!(
"\nFound cached resolution: {req_hash}, on python version: {}",
py_version.to_string_with_dot()
));
return Ok(cached);
}
}
let file = "requirements.in";
write_file(job_dir, file, &requirements)?;
@@ -269,6 +569,11 @@ pub async fn uv_pip_compile(
.await
.map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?;
} else {
// Make sure we have python runtime installed
py_version
.get_python(job_id, mem_peak, db, worker_name, w_id, occupancy_metrics)
.await?;
let mut args = vec![
"pip",
"compile",
@@ -286,11 +591,15 @@ pub async fn uv_pip_compile(
// Target to /tmp/windmill/cache/uv
"--cache-dir",
UV_CACHE_DIR,
// We dont want UV to manage python installations
"--python-preference",
"only-system",
"--no-python-downloads",
];
args.extend([
"-p",
&py_version.to_string_with_dot(),
"--python-preference",
"only-managed",
]);
if no_cache {
args.extend(["--no-cache"]);
}
@@ -332,6 +641,7 @@ pub async fn uv_pip_compile(
.env_clear()
.env("HOME", HOME_ENV.to_string())
.env("PATH", PATH_ENV.to_string())
.env("UV_PYTHON_INSTALL_DIR", PY_INSTALL_DIR.to_string())
.envs(PROXY_ENVS.clone())
.args(&args)
.stdout(Stdio::piped())
@@ -378,17 +688,22 @@ pub async fn uv_pip_compile(
let mut file = File::open(path_lock).await?;
let mut req_content = "".to_string();
file.read_to_string(&mut req_content).await?;
let lockfile = req_content
.lines()
.filter(|x| !x.trim_start().starts_with('#'))
.map(|x| x.to_string())
.collect::<Vec<String>>()
.join("\n");
let lockfile = format!(
"# py{}\n{}",
py_version.to_string_no_dot(),
req_content
.lines()
.filter(|x| !x.trim_start().starts_with('#'))
.map(|x| x.to_string())
.collect::<Vec<String>>()
.join("\n")
);
sqlx::query!(
"INSERT INTO pip_resolution_cache (hash, lockfile, expiration) VALUES ($1, $2, now() + ('3 days')::interval) ON CONFLICT (hash) DO UPDATE SET lockfile = $2",
req_hash,
lockfile
).fetch_optional(db).await?;
Ok(lockfile)
}
@@ -538,7 +853,8 @@ pub async fn handle_python_job(
occupancy_metrics: &mut OccupancyMetrics,
) -> windmill_common::error::Result<Box<RawValue>> {
let script_path = crate::common::use_flow_root_path(job.script_path());
let mut additional_python_paths = handle_python_deps(
let (py_version, mut additional_python_paths) = handle_python_deps(
job_dir,
requirements_o,
inner_content,
@@ -554,23 +870,53 @@ pub async fn handle_python_job(
)
.await?;
let PythonAnnotations { no_uv, no_postinstall, .. } = PythonAnnotations::parse(inner_content);
tracing::debug!("Finished handling python dependencies");
let python_path = if no_uv {
PYTHON_PATH.clone()
} else if let Some(python_path) = py_version
.get_python(
&job.id,
mem_peak,
db,
worker_name,
&job.workspace_id,
&mut Some(occupancy_metrics),
)
.await?
{
python_path
} else {
PYTHON_PATH.clone()
};
if !PythonAnnotations::parse(inner_content).no_postinstall {
if !no_postinstall {
if let Err(e) = postinstall(&mut additional_python_paths, job_dir, job, db).await {
tracing::error!("Postinstall stage has failed. Reason: {e}");
}
tracing::debug!("Finished deps postinstall stage");
}
append_logs(
&job.id,
&job.workspace_id,
"\n\n--- PYTHON CODE EXECUTION ---\n".to_string(),
db,
)
.await;
if no_uv {
append_logs(
&job.id,
&job.workspace_id,
format!("\n\n--- SYSTEM PYTHON (Fallback) CODE EXECUTION ---\n",),
db,
)
.await;
} else {
append_logs(
&job.id,
&job.workspace_id,
format!(
"\n\n--- PYTHON ({}) CODE EXECUTION ---\n",
py_version.to_string_with_dot()
),
db,
)
.await;
}
let (
import_loader,
import_base64,
@@ -725,6 +1071,7 @@ mount {{
"run.config.proto",
&NSJAIL_CONFIG_RUN_PYTHON3_CONTENT
.replace("{JOB_DIR}", job_dir)
.replace("{PY_INSTALL_DIR}", PY_INSTALL_DIR)
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string())
.replace("{SHARED_MOUNT}", shared_mount)
.replace("{SHARED_DEPENDENCIES}", shared_deps.as_str())
@@ -743,6 +1090,7 @@ mount {{
"started python code execution {}",
job.id
);
let child = if !*DISABLE_NSJAIL {
let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str());
nsjail_cmd
@@ -759,7 +1107,7 @@ mount {{
"--config",
"run.config.proto",
"--",
PYTHON_PATH.as_str(),
&python_path,
"-u",
"-m",
"wrapper",
@@ -768,7 +1116,9 @@ mount {{
.stderr(Stdio::piped());
start_child_process(nsjail_cmd, NSJAIL_PATH.as_str()).await?
} else {
let mut python_cmd = Command::new(PYTHON_PATH.as_str());
let mut python_cmd = Command::new(&python_path);
let args = vec!["-u", "-m", "wrapper"];
python_cmd
.current_dir(job_dir)
.env_clear()
@@ -778,7 +1128,7 @@ mount {{
.env("TZ", TZ_ENV.as_str())
.env("BASE_INTERNAL_URL", base_internal_url)
.env("HOME", HOME_ENV.as_str())
.args(vec!["-u", "-m", "wrapper"])
.args(args)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
@@ -788,7 +1138,7 @@ mount {{
python_cmd.env("USERPROFILE", crate::USERPROFILE_ENV.as_str());
}
start_child_process(python_cmd, PYTHON_PATH.as_str()).await?
start_child_process(python_cmd, &python_path).await?
};
handle_child(
@@ -1073,7 +1423,7 @@ async fn handle_python_deps(
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> error::Result<Vec<String>> {
) -> error::Result<(PyVersion, Vec<String>)> {
create_dependencies_dir(job_dir).await;
let mut additional_python_paths: Vec<String> = WORKER_CONFIG
@@ -1084,8 +1434,12 @@ async fn handle_python_deps(
.unwrap_or_else(|| vec![])
.clone();
let annotations = windmill_common::worker::PythonAnnotations::parse(inner_content);
let mut requirements;
let mut annotated_pyv = None;
let mut annotated_pyv_numeric = None;
let is_deployed = requirements_o.is_some();
let instance_pyv = PyVersion::from_instance_version().await;
let annotations = windmill_common::worker::PythonAnnotations::parse(inner_content);
let requirements = match requirements_o {
Some(r) => r,
None => {
@@ -1097,9 +1451,13 @@ async fn handle_python_deps(
script_path,
db,
&mut already_visited,
&mut annotated_pyv_numeric,
)
.await?
.join("\n");
annotated_pyv = annotated_pyv_numeric.and_then(|v| PyVersion::from_numeric(v));
if !requirements.is_empty() {
requirements = uv_pip_compile(
job_id,
@@ -1111,8 +1469,9 @@ async fn handle_python_deps(
worker_name,
w_id,
occupancy_metrics,
annotations.no_uv || annotations.no_uv_compile,
annotated_pyv.unwrap_or(instance_pyv),
annotations.no_cache,
annotations.no_uv || annotations.no_uv_compile,
)
.await
.map_err(|e| {
@@ -1123,12 +1482,47 @@ async fn handle_python_deps(
}
};
let requirements_lines: Vec<&str> = if requirements.len() > 0 {
requirements
.split("\n")
.filter(|x| !x.starts_with("--") && !x.trim().is_empty())
.collect()
} else {
vec![]
};
/*
For deployed scripts we want to find out version in following order:
1. Assigned version (written in lockfile)
2. 3.11
For Previews:
1. Annotated version
2. Instance version
3. Latest Stable
*/
let final_version = if is_deployed {
// If script is deployed we can try to parse first line to get assigned version
if let Some(v) = requirements_lines
.get(0)
.and_then(|line| PyVersion::parse_version(line))
{
// We have valid assigned version, we use it
v
} else {
// If there is no assigned version in lockfile we automatically fallback to 3.11
// In this case we have dependencies, but no associated python version
// This is the case for old deployed scripts
PyVersion::Py311
}
} else {
// This is not deployed script, meaning we test run it (Preview)
annotated_pyv.unwrap_or(instance_pyv)
};
// If len > 0 it means there is atleast one dependency or assigned python version
if requirements.len() > 0 {
let mut venv_path = handle_python_reqs(
requirements
.split("\n")
.filter(|x| !x.starts_with("--") && !x.trim().is_empty())
.collect(),
requirements_lines,
job_id,
w_id,
mem_peak,
@@ -1138,13 +1532,14 @@ async fn handle_python_deps(
job_dir,
worker_dir,
occupancy_metrics,
final_version,
annotations.no_uv || annotations.no_uv_install,
false,
)
.await?;
additional_python_paths.append(&mut venv_path);
}
Ok(additional_python_paths)
Ok((final_version, additional_python_paths))
}
lazy_static::lazy_static! {
@@ -1160,6 +1555,8 @@ async fn spawn_uv_install(
venv_p: &str,
job_dir: &str,
(pip_extra_index_url, pip_index_url): (Option<String>, Option<String>),
// If none, it is system python
py_path: Option<String>,
no_uv_install: bool,
) -> Result<tokio::process::Child, Error> {
if !*DISABLE_NSJAIL {
@@ -1181,7 +1578,14 @@ async fn spawn_uv_install(
if let Some(host) = PIP_TRUSTED_HOST.as_ref() {
vars.push(("TRUSTED_HOST", host));
}
let _owner;
if let Some(py_path) = py_path.as_ref() {
_owner = format!(
"-p {} --python-preference only-managed",
py_path.as_str() //
);
vars.push(("PY_PATH", &_owner));
}
vars.push(("REQ", &req));
vars.push(("TARGET", venv_p));
@@ -1231,8 +1635,6 @@ async fn spawn_uv_install(
&req,
"--no-deps",
"--no-color",
// "-p",
// "3.11",
// Prevent uv from discovering configuration files.
"--no-config",
"--link-mode=copy",
@@ -1250,6 +1652,22 @@ async fn spawn_uv_install(
]
};
if !no_uv_install {
if let Some(py_path) = py_path.as_ref() {
command_args.extend([
"-p",
py_path.as_str(),
"--python-preference",
"only-managed", //
]);
} else {
command_args.extend([
"--python-preference",
"only-system", //
]);
}
}
if let Some(url) = pip_extra_index_url.as_ref() {
url.split(",").for_each(|url| {
command_args.extend(["--extra-index-url", url]);
@@ -1339,7 +1757,7 @@ fn pad_string(value: &str, total_length: usize) -> String {
}
}
/// pip install, include cached or pull from S3
/// uv pip install, include cached or pull from S3
pub async fn handle_python_reqs(
requirements: Vec<&str>,
job_id: &Uuid,
@@ -1351,9 +1769,9 @@ pub async fn handle_python_reqs(
job_dir: &str,
worker_dir: &str,
_occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
py_version: PyVersion,
// TODO: Remove (Deprecated)
mut no_uv_install: bool,
is_ansible: bool,
) -> error::Result<Vec<String>> {
let counter_arc = Arc::new(tokio::sync::Mutex::new(0));
// Append logs with line like this:
@@ -1405,7 +1823,7 @@ pub async fn handle_python_reqs(
}
no_uv_install |= *USE_PIP_INSTALL;
if no_uv_install && !is_ansible {
if no_uv_install {
append_logs(&job_id, w_id, "\nFallback to pip (Deprecated!)\n", db).await;
tracing::warn!("Fallback to pip");
}
@@ -1449,13 +1867,14 @@ pub async fn handle_python_reqs(
NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT
})
.replace("{WORKER_DIR}", &worker_dir)
.replace("{PY_INSTALL_DIR}", &PY_INSTALL_DIR)
.replace(
"{CACHE_DIR}",
if no_uv_install {
PIP_CACHE_DIR
&(if no_uv_install {
PIP_CACHE_DIR.to_owned()
} else {
PY311_CACHE_DIR
},
py_version.to_cache_dir()
}),
)
.replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()),
)?;
@@ -1473,11 +1892,10 @@ pub async fn handle_python_reqs(
if req.starts_with('#') || req.starts_with('-') || req.trim().is_empty() {
continue;
}
// TODO: Remove
let py_prefix = if no_uv_install {
PIP_CACHE_DIR
} else {
PY311_CACHE_DIR
&py_version.to_cache_dir()
};
let venv_p = format!(
@@ -1528,7 +1946,7 @@ pub async fn handle_python_reqs(
let mut local_mem_peak = 0;
for pid_o in pids.lock().await.iter() {
if pid_o.is_some(){
let mem = get_mem_peak(*pid_o, !*DISABLE_NSJAIL).await;
let mem = crate::handle_child::get_mem_peak(*pid_o, !*DISABLE_NSJAIL).await;
if mem < 0 {
tracing::warn!(
workspace_id = %w_id_2,
@@ -1659,6 +2077,14 @@ pub async fn handle_python_reqs(
let is_not_pro = !matches!(get_license_plan().await, LicensePlan::Pro);
let total_time = std::time::Instant::now();
let py_path = if no_uv_install {
None
} else {
py_version
.get_python(job_id, mem_peak, db, _worker_name, w_id, _occupancy_metrics)
.await?
};
let has_work = req_with_penv.len() > 0;
for ((i, (req, venv_p)), mut kill_rx) in
req_with_penv.iter().enumerate().zip(kill_rxs.into_iter())
@@ -1688,6 +2114,7 @@ pub async fn handle_python_reqs(
let venv_p = venv_p.clone();
let counter_arc = counter_arc.clone();
let pip_indexes = pip_indexes.clone();
let py_path = py_path.clone();
let pids = pids.clone();
handles.push(task::spawn(async move {
@@ -1711,7 +2138,7 @@ pub async fn handle_python_reqs(
tokio::select! {
// Cancel was called on the job
_ = kill_rx.recv() => return Err(anyhow::anyhow!("S3 pull was canceled")),
pull = pull_from_tar(os, venv_p.clone(), no_uv_install) => {
pull = pull_from_tar(os, venv_p.clone(), py_version.to_cache_dir_top_level(), no_uv_install) => {
if let Err(e) = pull {
tracing::info!(
workspace_id = %w_id,
@@ -1744,7 +2171,8 @@ pub async fn handle_python_reqs(
&venv_p,
&job_dir,
pip_indexes,
no_uv_install,
py_path,
no_uv_install
).await {
Ok(r) => r,
Err(e) => {
@@ -1828,7 +2256,6 @@ pub async fn handle_python_reqs(
#[cfg(not(all(feature = "enterprise", feature = "parquet", unix)))]
let s3_push = false;
print_success(
false,
s3_push,
@@ -1846,7 +2273,7 @@ pub async fn handle_python_reqs(
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
if s3_push {
if let Some(os) = OBJECT_STORE_CACHE_SETTINGS.read().await.clone() {
tokio::spawn(build_tar_and_push(os, venv_p.clone(), no_uv_install));
tokio::spawn(build_tar_and_push(os, venv_p.clone(), py_version.to_cache_dir_top_level(), no_uv_install));
}
}
@@ -1955,7 +2382,7 @@ pub async fn start_worker(
.to_vec();
let context_envs = build_envs_map(context).await;
let additional_python_paths = handle_python_deps(
let (_, additional_python_paths) = handle_python_deps(
job_dir,
requirements_o,
inner_content,
backend/windmill-worker/src/worker.rs
+47 -5
View File
@@ -105,7 +105,7 @@ use crate::rust_executor::handle_rust_job;
use crate::php_executor::handle_php_job;
#[cfg(feature = "python")]
use crate::python_executor::handle_python_job;
use crate::python_executor::{handle_python_job, PyVersion};
#[cfg(feature = "python")]
use crate::ansible_executor::handle_ansible_job;
@@ -256,14 +256,20 @@ pub const LOCK_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "lock");
// Used as fallback now
pub const PIP_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "pip");
// pub const PY310_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_310");
pub const PY310_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_310");
pub const PY311_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_311");
// pub const PY312_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_312");
// pub const PY313_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_313");
pub const PY312_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_312");
pub const PY313_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "python_313");
pub const TAR_PY310_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "tar/python_310");
pub const TAR_PY311_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "tar/python_311");
pub const TAR_PY312_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "tar/python_312");
pub const TAR_PY313_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "tar/python_313");
pub const UV_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "uv");
pub const PY_INSTALL_DIR: &str = concatcp!(ROOT_CACHE_DIR, "py_runtime");
pub const TAR_PYBASE_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "tar");
pub const TAR_PIP_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "tar/pip");
pub const TAR_PY311_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "tar/python_311");
pub const DENO_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "deno");
pub const DENO_CACHE_DIR_DEPS: &str = concatcp!(ROOT_CACHE_DIR, "deno/deps");
pub const DENO_CACHE_DIR_NPM: &str = concatcp!(ROOT_CACHE_DIR, "deno/npm");
@@ -399,6 +405,7 @@ lazy_static::lazy_static! {
pub static ref PIP_EXTRA_INDEX_URL: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
pub static ref PIP_INDEX_URL: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
pub static ref INSTANCE_PYTHON_VERSION: Arc<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
pub static ref JOB_DEFAULT_TIMEOUT: Arc<RwLock<Option<i32>>> = Arc::new(RwLock::new(None));
static ref MAX_TIMEOUT: u64 = std::env::var("TIMEOUT")
@@ -765,6 +772,41 @@ pub async fn run_worker(
let worker_dir = format!("{TMP_DIR}/{worker_name}");
tracing::debug!(worker = %worker_name, hostname = %hostname, worker_dir = %worker_dir, "Creating worker dir");
#[cfg(feature = "python")]
{
let (db, worker_name, hostname, worker_dir) = (
db.clone(),
worker_name.clone(),
hostname.to_owned(),
worker_dir.clone(),
);
tokio::spawn(async move {
if let Err(e) = PyVersion::from_instance_version()
.await
.get_python(&Uuid::nil(), &mut 0, &db, &worker_name, "", &mut None)
.await
{
tracing::error!(
worker = %worker_name,
hostname = %hostname,
worker_dir = %worker_dir,
"Cannot preinstall or find Instance Python version to worker: {e}"//
);
}
if let Err(e) = PyVersion::Py311
.get_python(&Uuid::nil(), &mut 0, &db, &worker_name, "", &mut None)
.await
{
tracing::error!(
worker = %worker_name,
hostname = %hostname,
worker_dir = %worker_dir,
"Cannot preinstall or find default 311 version to worker: {e}"//
);
}
});
}
if let Some(ref netrc) = *NETRC {
tracing::info!(worker = %worker_name, hostname = %hostname, "Writing netrc at {}/.netrc", HOME_ENV.as_str());
write_file(&HOME_ENV, ".netrc", netrc).expect("could not write netrc");
backend/windmill-worker/src/worker_lockfiles.rs
+38 -8
View File
@@ -39,7 +39,8 @@ use crate::csharp_executor::generate_nuget_lockfile;
use crate::php_executor::{composer_install, parse_php_imports};
#[cfg(feature = "python")]
use crate::python_executor::{
create_dependencies_dir, handle_python_reqs, uv_pip_compile, USE_PIP_COMPILE, USE_PIP_INSTALL,
create_dependencies_dir, handle_python_reqs, uv_pip_compile, PyVersion, USE_PIP_COMPILE,
USE_PIP_INSTALL,
};
#[cfg(feature = "rust")]
use crate::rust_executor::generate_cargo_lockfile;
@@ -1595,10 +1596,28 @@ async fn python_dep(
w_id: &str,
worker_dir: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
annotated_pyv_numeric: Option<u32>,
annotations: PythonAnnotations,
no_uv_compile: bool,
no_uv_install: bool,
) -> std::result::Result<String, Error> {
create_dependencies_dir(job_dir).await;
/*
Unlike `handle_python_deps` which we use for running scripts (deployed and drafts)
This one used specifically for deploying scripts
So we can get final_version right away and include in lockfile
And the precendence is following:
1. Annotation version
2. Instance version
3. Latest Stable
*/
let final_version = annotated_pyv_numeric
.and_then(|pyv| PyVersion::from_numeric(pyv))
.unwrap_or(PyVersion::from_instance_version().await);
let req: std::result::Result<String, Error> = uv_pip_compile(
job_id,
&reqs,
@@ -1609,8 +1628,9 @@ async fn python_dep(
worker_name,
w_id,
occupancy_metrics,
final_version,
annotations.no_cache,
no_uv_compile,
false,
)
.await;
// install the dependencies to pre-fill the cache
@@ -1626,8 +1646,8 @@ async fn python_dep(
job_dir,
worker_dir,
occupancy_metrics,
final_version,
no_uv_install,
false,
)
.await;
@@ -1665,10 +1685,18 @@ async fn capture_dependency_job(
return Err(Error::InternalErr(
"Python requires the python feature to be enabled".to_string(),
));
#[cfg(feature = "python")]
{
let anns = PythonAnnotations::parse(job_raw_code);
let mut annotated_pyv_numeric = None;
let reqs = if raw_deps {
// `wmill script generate-metadata`
// should also respect annotated pyversion
// can be annotated in script itself
// or in requirements.txt if present
annotated_pyv_numeric =
PyVersion::from_py_annotations(anns).map(|v| v.to_numeric());
job_raw_code.to_string()
} else {
let mut already_visited = vec![];
@@ -1679,14 +1707,12 @@ async fn capture_dependency_job(
script_path,
&db,
&mut already_visited,
&mut annotated_pyv_numeric,
)
.await?
.join("\n")
};
let PythonAnnotations { no_uv, no_uv_install, no_uv_compile, .. } =
PythonAnnotations::parse(job_raw_code);
let PythonAnnotations { no_uv, no_uv_install, no_uv_compile, .. } = anns;
if no_uv || no_uv_install || no_uv_compile || *USE_PIP_COMPILE || *USE_PIP_INSTALL {
if let Err(e) = sqlx::query!(
r#"
@@ -1713,6 +1739,8 @@ async fn capture_dependency_job(
w_id,
worker_dir,
&mut Some(occupancy_metrics),
annotated_pyv_numeric,
anns,
no_uv_compile | no_uv,
no_uv_install | no_uv,
)
@@ -1761,6 +1789,8 @@ async fn capture_dependency_job(
w_id,
worker_dir,
&mut Some(occupancy_metrics),
None,
PythonAnnotations::default(),
false,
false,
)
frontend/src/lib/components/InstanceSetting.svelte
+26 -2
View File
@@ -27,15 +27,20 @@
import { createEventDispatcher } from 'svelte'
import { fade } from 'svelte/transition'
import { base } from '$lib/base'
import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from './common/toggleButton-v2/ToggleButton.svelte'
import SimpleEditor from './SimpleEditor.svelte'
export let setting: Setting
export let version: string
export let values: Writable<Record<string, any>>
export let loading = true
const dispatch = createEventDispatcher()
if (setting.fieldType == 'select' && $values[setting.key] == undefined){
$values[setting.key] = "default";
}
let latestKeyRenewalAttempt: {
result: string
attempted_at: string
@@ -166,6 +171,24 @@
EE only {#if setting.ee_only != ''}<Tooltip>{setting.ee_only}</Tooltip>{/if}
</div>
{/if}
{#if setting.fieldType == 'select'}
<div>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">{setting.label}</span>
{#if setting.description}
<span class="text-secondary text-xs">
{@html setting.description}
</span>
{/if}
</label>
<ToggleButtonGroup bind:selected={$values[setting.key]}>
{#each (setting.select_items ?? []) as item }
<ToggleButton value={item.value ?? item.label} label={item.label} tooltip={item.tooltip} />
{/each}
</ToggleButtonGroup>
</div>
{:else}
<!-- svelte-ignore a11y-label-has-associated-control -->
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">{setting.label}</span>
@@ -881,8 +904,8 @@
bind:seconds={$values[setting.key]}
/>
</div>
{:else if setting.fieldType == 'select'}
{/if}
{#if hasError}
<span class="text-red-500 dark:text-red-400 text-sm">
{setting.error ?? ''}
@@ -892,4 +915,5 @@
<input disabled placeholder="Loading..." />
{/if}
</label>
{/if}
{/if}
frontend/src/lib/components/instanceSettings.ts
+39 -3
View File
@@ -6,6 +6,13 @@ export interface Setting {
ee_only?: string
tooltip?: string
key: string
// If value is not specified for first element, it will automatcally use undefined
select_items?: {
label: string,
tooltip?: string,
// If not specified, label will be used
value?: any,
}[],
fieldType:
| 'text'
| 'number'
@@ -74,9 +81,9 @@ export const settings: Record<string, Setting[]> = {
isValid: (value: string | undefined) =>
value
? value?.startsWith('http') &&
value.includes('://') &&
!value?.endsWith('/') &&
!value?.endsWith(' ')
value.includes('://') &&
!value?.endsWith('/') &&
!value?.endsWith(' ')
: false
},
{
@@ -219,6 +226,35 @@ export const settings: Record<string, Setting[]> = {
],
'Auth/OAuth': [],
Registries: [
{
label: 'Instance Python Version',
description: 'Default python version for newly deployed scripts',
key: 'instance_python_version',
fieldType: 'select',
// To change latest stable version:
// 1. Change placeholder in instanceSettings.ts
// 2. Change LATEST_STABLE_PY in dockerfile
// 3. Change #[default] annotation for PyVersion in backend
placeholder: '3.10,3.11,3.12,3.13',
select_items: [{
label: "Latest Stable",
value: "default",
tooltip: "python-3.11",
},
{
label: "3.10",
},
{
label: "3.11",
},
{
label: "3.12",
},
{
label: "3.13",
}],
storage: 'setting',
},
{
label: 'Pip index url',
description: 'Add private Pip registry',
shell.nix
+1 -1
View File
@@ -29,7 +29,7 @@ in pkgs.mkShell {
postgresql
watchexec # used in client's dev.nu
poetry # for python client
uv
# uv
python312Packages.pip-tools # pip-compile
];