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

feat(python, windows): enable S3 to cache wheels (#5199)

Browse Source 
* add .minio-data to .gitignore

* feat(python): make S3 cache arch specific

Current schema:

S3-Bucket
├── python_311
│   ├── wheel==1.0
│   └── wheel2==1.0
└── python_312
    ├── wheel==1.0
    └── wheel2==1.0

New schema:

S3-Bucket
├── linux_aarch64
│   └── ...
└── linux_x86_64
    ├── python_311
    │   ├── wheel==1.0
    │   └── wheel2==1.0
    └── python_312
        ├── wheel==1.0
        └── wheel2==1.0

* remove .minio-data from .gitignore

* remove unneeded tracing::error

* feat(python, windows): enable S3 to cache wheels

* fix(python, windows): drop residual unix cfg gates on S3 cache items

Post-merge, PIPTAR_UPLOAD_CHANNEL and its call sites were already
cross-platform, but the types/functions they reference (PiptarUploadTask,
handle_piptar_uploads, pull_from_tar, OBJECT_STORE_SETTINGS) remained
unix-gated, breaking the Windows build.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Pyra
2026-06-19 13:11:35 +00:00
committed by GitHub
co-authored by Claude Opus 4.7
parent 496e770264
commit ab3bc97cd9
1 changed files with 16 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/python_executor.rs
+16 -14
View File
@@ -24,8 +24,10 @@ use tokio::{
use windmill_queue::MiniPulledJob;
use uuid::Uuid;
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use windmill_common::ee_oss::{get_license_plan, LicensePlan};
use windmill_common::{
error::{
self,
@@ -72,7 +74,7 @@ lazy_static::lazy_static! {
static ref EPHEMERAL_TOKEN_CMD: Option<String> = var("EPHEMERAL_TOKEN_CMD").ok();
}
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
lazy_static::lazy_static! {
static ref PIPTAR_UPLOAD_CHANNEL: tokio::sync::mpsc::UnboundedSender<PiptarUploadTask> = {
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
@@ -84,14 +86,14 @@ lazy_static::lazy_static! {
};
}
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
#[derive(Debug)]
struct PiptarUploadTask {
venv_path: String,
cache_dir: String,
}
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
async fn handle_piptar_uploads(mut rx: tokio::sync::mpsc::UnboundedReceiver<PiptarUploadTask>) {
use crate::global_cache::build_tar_and_push;
use windmill_object_store::get_object_store;
@@ -137,10 +139,10 @@ pub fn has_relative_imports(content: &str) -> bool {
RELATIVE_IMPORT_REGEX.is_match(content)
}
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use crate::global_cache::pull_from_tar;
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use windmill_object_store::OBJECT_STORE_SETTINGS;
use crate::{
@@ -2388,12 +2390,12 @@ pub async fn handle_python_reqs(
instant: std::time::Instant,
conn: &Connection,
) {
#[cfg(not(all(feature = "enterprise", feature = "parquet", unix)))]
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
{
(s3_pull, s3_push) = (false, false);
}
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
if OBJECT_STORE_SETTINGS.read().await.is_none() {
(s3_pull, s3_push) = (false, false);
}
@@ -2648,7 +2650,7 @@ pub async fn handle_python_reqs(
let mut handles = Vec::with_capacity(total_to_install);
// let mem_peak_thread_safe = Arc::new(tokio::sync::Mutex::new(0));
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
let is_not_pro = !matches!(get_license_plan().await, LicensePlan::Pro);
let total_time = std::time::Instant::now();
@@ -2696,7 +2698,7 @@ pub async fn handle_python_reqs(
let pids = pids.clone();
let worker_dir = worker_dir.clone();
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
let py_version = py_version.clone();
handles.push(task::spawn(async move {
@@ -2714,7 +2716,7 @@ pub async fn handle_python_reqs(
);
let start = std::time::Instant::now();
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
if is_not_pro {
if let Some(os) = windmill_object_store::get_object_store().await {
tokio::select! {
@@ -2896,10 +2898,10 @@ pub async fn handle_python_reqs(
}
};
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
let s3_push = is_not_pro;
#[cfg(not(all(feature = "enterprise", feature = "parquet", unix)))]
#[cfg(not(all(feature = "enterprise", feature = "parquet")))]
let s3_push = false;
if is_sandboxing_enabled() {
@@ -2953,7 +2955,7 @@ pub async fn handle_python_reqs(
)
.await;
#[cfg(all(feature = "enterprise", feature = "parquet", unix))]
#[cfg(all(feature = "enterprise", feature = "parquet"))]
if s3_push {
// Send to upload channel for sequential processing
let upload_task = PiptarUploadTask {