kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

backend: use v2 tables through views where possible (v2 phase 3) (#5119)

Browse Source
This commit is contained in:
Lucas Abel
2025-02-06 12:43:24 +01:00
committed by GitHub
parent 8374377958
commit 6ded2c598e
217 changed files with 5864 additions and 1075 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
backend/windmill-worker/src/python_executor.rs
View File

@@ -10,7 +10,7 @@ use anyhow::anyhow;
use itertools::Itertools;
use regex::Regex;
use serde_json::value::RawValue;
use sqlx::{types::Json, Pool, Postgres};
use sqlx::{Pool, Postgres};
use tokio::{
fs::{metadata, DirBuilder, File},
io::AsyncReadExt,
@@ -26,7 +26,7 @@ use windmill_common::{
self,
Error::{self},
},
jobs::{QueuedJob, PREPROCESSOR_FAKE_ENTRYPOINT},
jobs::QueuedJob,
utils::calculate_hash,
worker::{write_file, PythonAnnotations, WORKER_CONFIG},
DB,
@@ -35,8 +35,8 @@ use windmill_common::{
#[cfg(feature = "enterprise")]
use windmill_common::variables::get_secret_value_as_admin;
use windmill_queue::{append_logs, CanceledBy};
use std::env::var;
use windmill_queue::{append_logs, CanceledBy};
lazy_static::lazy_static! {
static ref PYTHON_PATH: String =
@@ -84,8 +84,8 @@ use windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS;
use crate::{
common::{
create_args_and_out_file, get_main_override, get_reserved_variables, read_file,
read_result, start_child_process, OccupancyMetrics,
create_args_and_out_file, get_reserved_variables, read_file, read_result,
start_child_process, OccupancyMetrics,
},
handle_child::handle_child,
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, INSTANCE_PYTHON_VERSION,
@@ -931,8 +931,6 @@ pub async fn handle_python_job(
tracing::debug!("Finished deps postinstall stage");
}
if no_uv {
append_logs(
&job.id,
@@ -966,10 +964,11 @@ pub async fn handle_python_job(
pre_spread,
) = prepare_wrapper(
job_dir,
job.is_flow_step,
job.preprocessed,
job.script_entrypoint_override.as_deref(),
inner_content,
&script_path,
job.args.as_ref(),
false,
)
.await?;
@@ -1086,7 +1085,7 @@ except BaseException as e:
// Usefull if certain wheels needs to be preinstalled before execution.
let global_site_packages_path = py_version.to_cache_dir() + "/global-site-packages";
let additional_python_paths_folders = {
let mut paths= additional_python_paths.clone();
let mut paths = additional_python_paths.clone();
if std::fs::metadata(&global_site_packages_path).is_ok() {
// We want global_site_packages_path to be included in additonal_python_paths_folders, but
// we don't want it to be included in global_site_packages_path.
@@ -1236,10 +1235,11 @@ mount {{
async fn prepare_wrapper(
job_dir: &str,
job_is_flow_step: bool,
job_preprocessed: Option<bool>,
job_script_entrypoint_override: Option<&str>,
inner_content: &str,
script_path: &str,
args: Option<&Json<HashMap<String, Box<RawValue>>>>,
skip_preprocessor: bool,
) -> error::Result<(
&'static str,
&'static str,
@@ -1252,16 +1252,8 @@ async fn prepare_wrapper(
Option<String>,
Option<String>,
)> {
let (main_override, apply_preprocessor) = match get_main_override(args) {
Some(main_override) => {
if !skip_preprocessor && main_override == PREPROCESSOR_FAKE_ENTRYPOINT {
(None, true)
} else {
(Some(main_override), false)
}
}
None => (None, false),
};
let main_override = job_script_entrypoint_override.as_deref();
let apply_preprocessor = !job_is_flow_step && job_preprocessed == Some(false);
let relative_imports = RELATIVE_IMPORT_REGEX.is_match(&inner_content);
@@ -1301,7 +1293,10 @@ async fn prepare_wrapper(
let _ = write_file(job_dir, "loader.py", RELATIVE_PYTHON_LOADER)?;
}
let sig = windmill_parser_py::parse_python_signature(inner_content, main_override.clone())?;
let sig = windmill_parser_py::parse_python_signature(
inner_content,
main_override.map(ToString::to_string),
)?;
let pre_sig = if apply_preprocessor {
Some(windmill_parser_py::parse_python_signature(
@@ -1418,7 +1413,7 @@ async fn prepare_wrapper(
last,
transforms,
spread,
main_override,
main_override.map(ToString::to_string),
pre_spread,
))
}
@@ -1768,7 +1763,7 @@ async fn spawn_uv_install(
// Track https://github.com/astral-sh/uv/issues/6715
if let Some(cert_path) = INDEX_CERT.as_ref() {
// Once merged --cert can be used instead
//
//
// command_args.extend(["--cert", cert_path]);
envs.push(("SSL_CERT_FILE", cert_path));
}
@@ -2045,29 +2040,26 @@ pub async fn handle_python_reqs(
// Notify server that we are still alive
// Detect if job has been canceled
let canceled =
sqlx::query_scalar::<_, bool>
(r#"
UPDATE queue
SET last_ping = now()
, mem_peak = $1
WHERE id = $2
RETURNING canceled
"#)
.bind(mem_peak_actual)
.bind(job_id_2)
.fetch_optional(&db_2)
.await
.unwrap_or_else(|e| {
tracing::error!(%e, "error updating job {job_id_2}: {e:#}");
Some(false)
})
.unwrap_or_else(|| {
// if the job is not in queue, it can only be in the completed_job so it is already complete
false
});
let canceled = sqlx::query_scalar!(
"UPDATE v2_job_runtime r SET
memory_peak = $1,
ping = now()
FROM v2_job_queue q
WHERE r.id = $2 AND q.id = r.id
RETURNING canceled_by IS NOT NULL AS \"canceled!\"",
mem_peak_actual,
job_id_2
)
.fetch_optional(&db_2)
.await
.unwrap_or_else(|e| {
tracing::error!(%e, "error updating job {job_id_2}: {e:#}");
Some(false)
})
.unwrap_or_else(|| {
// if the job is not in queue, it can only be in the completed_job so it is already complete
false
});
if canceled {
@@ -2289,12 +2281,12 @@ pub async fn handle_python_reqs(
uv_install_proccess.kill().await?;
pids.lock().await.get_mut(i).and_then(|e| e.take());
return Err(anyhow::anyhow!("uv pip install was canceled"));
},
},
(_, exitstatus) = async {
// See tokio::process::Child::wait_with_output() for more context
// Sometimes uv_install_proccess.wait() is not exiting if stderr is not awaited before it :/
(stderr_future.await, uv_install_proccess.wait().await)
} => match exitstatus {
} => match exitstatus {
Ok(status) => if !status.success() {
tracing::warn!(
workspace_id = %w_id,
@@ -2479,7 +2471,6 @@ pub async fn start_worker(
)
.await?;
let _args = None;
let (
import_loader,
import_base64,
@@ -2491,7 +2482,7 @@ pub async fn start_worker(
spread,
_,
_,
) = prepare_wrapper(job_dir, inner_content, script_path, _args.as_ref(), true).await?;
) = prepare_wrapper(job_dir, false, None, None, inner_content, script_path).await?;
{
let indented_transforms = transforms