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

worker: move ping to root execution

Browse Source
This commit is contained in:
Abel Lucas
2024-12-16 17:49:28 +01:00
parent 90ad56d582
commit d71a76f083
21 changed files with 126 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
backend/.sqlx/query-c6423de98ffedc4e58cca0bd7883ef87f04a3daa054882a0c4e81c723196dbb6.json generated Normal file
View File

@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE worker_ping SET\n ping_at = now()\n , current_job_id = $1\n , current_job_workspace_id = $2\n , memory_usage = $3\n , wm_memory_usage = $4\n , occupancy_rate = $6\n , occupancy_rate_15s = $7\n , occupancy_rate_5m = $8\n , occupancy_rate_30m = $9\n WHERE worker = $5",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Varchar",
"Int8",
"Int8",
"Text",
"Float4",
"Float4",
"Float4",
"Float4"
]
},
"nullable": []
},
"hash": "c6423de98ffedc4e58cca0bd7883ef87f04a3daa054882a0c4e81c723196dbb6"
}

3
backend/src/main.rs
View File

@@ -159,7 +159,6 @@ async fn cache_hub_scripts(file_path: Option<String>) -> anyhow::Result<()> {
"global",
"global",
"",
&mut None,
)
.await?;
tokio::fs::remove_dir_all(job_dir).await?;
@@ -179,7 +178,6 @@ async fn cache_hub_scripts(file_path: Option<String>) -> anyhow::Result<()> {
"cache_init",
envs.clone(),
false,
&mut None,
)
.await?;
@@ -196,7 +194,6 @@ async fn cache_hub_scripts(file_path: Option<String>) -> anyhow::Result<()> {
"",
"cache_init",
"",
&mut None,
)
.await
{

11
backend/windmill-worker/src/ansible_executor.rs
View File

@@ -25,7 +25,7 @@ use crate::{
bash_executor::BIN_BASH,
common::{
check_executor_binary_exists, get_reserved_variables, read_and_check_result,
start_child_process, transform_json, OccupancyMetrics,
start_child_process, transform_json,
},
handle_child::handle_child,
python_executor::{create_dependencies_dir, handle_python_reqs, uv_pip_compile},
@@ -53,7 +53,6 @@ async fn handle_ansible_python_deps(
worker_name: &str,
worker_dir: &str,
mem_peak: &mut i32,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Vec<String>> {
create_dependencies_dir(job_dir).await;
@@ -81,7 +80,6 @@ async fn handle_ansible_python_deps(
db,
worker_name,
w_id,
&mut Some(occupancy_metrics),
false,
false,
)
@@ -107,7 +105,6 @@ async fn handle_ansible_python_deps(
worker_name,
job_dir,
worker_dir,
&mut Some(occupancy_metrics),
false,
false,
)
@@ -125,7 +122,6 @@ async fn install_galaxy_collections(
w_id: &str,
mem_peak: &mut i32,
db: &sqlx::Pool<sqlx::Postgres>,
occupancy_metrics: &mut OccupancyMetrics,
) -> anyhow::Result<()> {
write_file(job_dir, "requirements.yml", collections_yml)?;
@@ -168,7 +164,6 @@ async fn install_galaxy_collections(
"ansible galaxy install",
None,
false,
&mut Some(occupancy_metrics),
)
.await?;
@@ -188,7 +183,6 @@ pub async fn handle_ansible_job(
shared_mount: &str,
base_internal_url: &str,
envs: HashMap<String, String>,
occupancy_metrics: &mut OccupancyMetrics,
) -> windmill_common::error::Result<Box<RawValue>> {
check_executor_binary_exists(
"ansible-playbook",
@@ -210,7 +204,6 @@ pub async fn handle_ansible_job(
worker_name,
worker_dir,
mem_peak,
occupancy_metrics,
)
.await?;
@@ -283,7 +276,6 @@ pub async fn handle_ansible_job(
&job.workspace_id,
mem_peak,
db,
occupancy_metrics,
)
.await?;
}
@@ -424,7 +416,6 @@ fi
"python run",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;
read_and_check_result(job_dir).await

9
backend/windmill-worker/src/bash_executor.rs
View File

@@ -44,7 +44,6 @@ use crate::handle_child::run_future_with_polling_update_job_poller;
use crate::{
common::{
build_args_map, get_reserved_variables, read_file, read_file_content, start_child_process,
OccupancyMetrics,
},
handle_child::handle_child,
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV,
@@ -71,7 +70,6 @@ pub async fn handle_bash_job(
base_internal_url: &str,
worker_name: &str,
envs: HashMap<String, String>,
occupancy_metrics: &mut OccupancyMetrics,
_killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
) -> Result<Box<RawValue>, Error> {
let annotation = windmill_common::worker::BashAnnotations::parse(&content);
@@ -212,7 +210,6 @@ exit $exit_status
"bash run",
job.timeout,
true,
&mut Some(occupancy_metrics),
)
.await?;
@@ -225,7 +222,6 @@ exit $exit_status
job.timeout,
mem_peak,
worker_name,
occupancy_metrics,
_killpill_rx,
)
.await;
@@ -269,7 +265,6 @@ async fn handle_docker_job(
job_timeout: Option<i32>,
mem_peak: &mut i32,
worker_name: &str,
occupancy_metrics: &mut OccupancyMetrics,
killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
) -> Result<Box<RawValue>, Error> {
let client = bollard::Docker::connect_with_unix_defaults().map_err(to_anyhow)?;
@@ -360,7 +355,6 @@ async fn handle_docker_job(
wait_f,
worker_name,
workspace_id,
&mut Some(occupancy_metrics),
Box::pin(match mem_client {
Ok(client) => client
.stats(
@@ -462,7 +456,6 @@ pub async fn handle_powershell_job(
base_internal_url: &str,
worker_name: &str,
envs: HashMap<String, String>,
occupancy_metrics: &mut OccupancyMetrics,
) -> Result<Box<RawValue>, Error> {
let pwsh_args = {
let args = build_args_map(job, client, db).await?.map(Json);
@@ -549,7 +542,6 @@ pub async fn handle_powershell_job(
"powershell install",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;
}
@@ -759,7 +751,6 @@ $env:PSModulePath = \"{};$PSModulePathBackup\"",
"powershell run",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;

2
backend/windmill-worker/src/bigquery_executor.rs
View File

@@ -209,7 +209,6 @@ pub async fn do_bigquery(
mem_peak: &mut i32,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
occupancy_metrics: &mut OccupancyMetrics,
) -> windmill_common::error::Result<Box<RawValue>> {
let bigquery_args = build_args_values(job, client, db).await?;
@@ -363,7 +362,6 @@ pub async fn do_bigquery(
result_f.map_err(to_anyhow),
worker_name,
&job.workspace_id,
&mut Some(occupancy_metrics),
Box::pin(futures::stream::once(async { 0 })),
)
.await?;

22
backend/windmill-worker/src/bun_executor.rs
View File

@@ -21,7 +21,6 @@ use crate::{
common::{
create_args_and_out_file, get_main_override, get_reserved_variables, parse_npm_config,
read_file, read_file_content, read_result, start_child_process, write_file_binary,
OccupancyMetrics,
},
handle_child::handle_child,
AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR,
@@ -96,7 +95,6 @@ pub async fn gen_bun_lockfile(
export_pkg: bool,
raw_deps: Option<String>,
npm_mode: bool,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> Result<Option<String>> {
let common_bun_proc_envs: HashMap<String, String> = get_common_bun_proc_envs(None).await;
@@ -155,7 +153,6 @@ pub async fn gen_bun_lockfile(
"bun build",
None,
false,
occupancy_metrics,
)
.await?;
} else {
@@ -179,7 +176,6 @@ pub async fn gen_bun_lockfile(
worker_name,
common_bun_proc_envs,
npm_mode,
occupancy_metrics,
)
.await?;
} else {
@@ -265,7 +261,6 @@ pub async fn install_bun_lockfile(
worker_name: &str,
common_bun_proc_envs: HashMap<String, String>,
npm_mode: bool,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> Result<()> {
let mut child_cmd = Command::new(if npm_mode { &*NPM_PATH } else { &*BUN_PATH });
child_cmd
@@ -335,7 +330,6 @@ pub async fn install_bun_lockfile(
"bun install",
None,
false,
occupancy_metrics,
)
.await?
} else {
@@ -475,7 +469,6 @@ pub async fn generate_wrapper_mjs(
timeout: Option<i32>,
mem_peak: &mut i32,
common_bun_proc_envs: &HashMap<String, String>,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> Result<()> {
let mut child = Command::new(&*BUN_PATH);
child
@@ -502,7 +495,6 @@ pub async fn generate_wrapper_mjs(
"bun build",
timeout,
false,
occupancy_metrics,
)
.await?;
fs::rename(
@@ -522,7 +514,6 @@ pub async fn generate_bun_bundle(
timeout: Option<i32>,
mem_peak: &mut i32,
common_bun_proc_envs: &HashMap<String, String>,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> Result<()> {
let mut child = Command::new(&*BUN_PATH);
child
@@ -550,7 +541,6 @@ pub async fn generate_bun_bundle(
"bun build",
timeout,
false,
occupancy_metrics,
)
.await?;
} else {
@@ -665,7 +655,6 @@ pub async fn prebundle_bun_script(
base_internal_url: &str,
worker_name: &str,
token: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> Result<()> {
let (local_path, remote_path) = compute_bundle_local_and_remote_path(
inner_content,
@@ -712,7 +701,6 @@ pub async fn prebundle_bun_script(
None,
&mut 0,
&common_bun_proc_envs,
occupancy_metrics,
)
.await?;
@@ -811,7 +799,6 @@ pub async fn handle_bun_job(
envs: HashMap<String, String>,
shared_mount: &str,
new_args: &mut Option<HashMap<String, Box<RawValue>>>,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<RawValue>> {
let mut annotation = windmill_common::worker::TypeScriptAnnotations::parse(inner_content);
@@ -935,7 +922,6 @@ pub async fn handle_bun_job(
worker_name,
common_bun_proc_envs.clone(),
annotation.npm,
&mut Some(occupancy_metrics),
)
.await?;
@@ -976,7 +962,6 @@ pub async fn handle_bun_job(
false,
None,
annotation.npm,
&mut Some(occupancy_metrics),
)
.await?;
@@ -1211,7 +1196,6 @@ try {{
job.timeout,
mem_peak,
&common_bun_proc_envs,
&mut Some(occupancy_metrics),
)
.await?;
if !local_path.is_empty() {
@@ -1252,7 +1236,6 @@ try {{
job.timeout,
mem_peak,
&common_bun_proc_envs,
&mut Some(occupancy_metrics),
)
.await?;
}
@@ -1300,7 +1283,6 @@ try {{
worker_name,
&job.workspace_id,
false,
occupancy_metrics,
)
.await?;
tracing::info!(
@@ -1452,7 +1434,6 @@ try {{
"bun run",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;
@@ -1601,7 +1582,6 @@ pub async fn start_worker(
worker_name,
common_bun_proc_envs.clone(),
annotation.npm,
&mut None,
)
.await?;
tracing::info!("dedicated worker requirements installed: {reqs}");
@@ -1621,7 +1601,6 @@ pub async fn start_worker(
false,
None,
annotation.npm,
&mut None,
)
.await?;
}
@@ -1717,7 +1696,6 @@ for await (const line of Readline.createInterface({{ input: process.stdin }})) {
None,
&mut mem_peak,
&common_bun_proc_envs,
&mut None,
)
.await?;
}

10
backend/windmill-worker/src/csharp_executor.rs
View File

@@ -34,7 +34,6 @@ use crate::{
NUGET_CONFIG, PATH_ENV, TZ_ENV,
};
use crate::common::OccupancyMetrics;
use crate::AuthedClientBackgroundTask;
#[cfg(windows)]
@@ -62,7 +61,6 @@ pub async fn generate_nuget_lockfile(
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
w_id: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<String> {
check_executor_binary_exists("dotnet", DOTNET_PATH.as_str(), "C#")?;
@@ -92,7 +90,6 @@ pub async fn generate_nuget_lockfile(
"dotnet restore",
None,
false,
&mut Some(occupancy_metrics),
)
.await?;
@@ -118,7 +115,6 @@ pub async fn generate_nuget_lockfile(
_db: &sqlx::Pool<sqlx::Postgres>,
_worker_name: &str,
_w_id: &str,
_occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<String> {
Err(anyhow!("C# is not available because the feature is not enabled").into())
}
@@ -262,7 +258,6 @@ async fn build_cs_proj(
w_id: &str,
base_internal_url: &str,
hash: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<String> {
if let Some(nuget_config) = NUGET_CONFIG.read().await.clone() {
write_file(job_dir, "nuget.config", &nuget_config)?;
@@ -311,7 +306,6 @@ async fn build_cs_proj(
"dotnet publish",
None,
false,
&mut Some(occupancy_metrics),
)
.await?;
append_logs(job_id, w_id, "\n\n", db).await;
@@ -366,7 +360,6 @@ pub async fn handle_csharp_job(
_base_internal_url: &str,
_worker_name: &str,
_envs: HashMap<String, String>,
_occupancy_metrics: &mut OccupancyMetrics,
) -> Result<Box<RawValue>, Error> {
Err(anyhow!("C# is not available because the feature is not enabled").into())
}
@@ -384,7 +377,6 @@ pub async fn handle_csharp_job(
base_internal_url: &str,
worker_name: &str,
envs: HashMap<String, String>,
occupancy_metrics: &mut OccupancyMetrics,
) -> Result<Box<RawValue>, Error> {
check_executor_binary_exists("dotnet", DOTNET_PATH.as_str(), "C#")?;
@@ -449,7 +441,6 @@ pub async fn handle_csharp_job(
&job.workspace_id,
base_internal_url,
&hash,
occupancy_metrics,
)
.await?
};
@@ -520,7 +511,6 @@ pub async fn handle_csharp_job(
"csharp run",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;
read_result(job_dir).await

6
backend/windmill-worker/src/deno_executor.rs
View File

@@ -8,7 +8,7 @@ use windmill_queue::append_logs;
use crate::{
common::{
create_args_and_out_file, get_main_override, get_reserved_variables, parse_npm_config,
read_file, read_result, start_child_process, OccupancyMetrics,
read_file, read_result, start_child_process,
},
handle_child::handle_child,
AuthedClientBackgroundTask, DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV,
@@ -105,7 +105,6 @@ pub async fn generate_deno_lock(
w_id: &str,
worker_name: &str,
base_internal_url: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> error::Result<String> {
let _ = write_file(job_dir, "main.ts", code)?;
@@ -158,7 +157,6 @@ pub async fn generate_deno_lock(
"deno cache",
None,
false,
occupancy_metrics,
)
.await?;
} else {
@@ -188,7 +186,6 @@ pub async fn handle_deno_job(
worker_name: &str,
envs: HashMap<String, String>,
new_args: &mut Option<HashMap<String, Box<RawValue>>>,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<RawValue>> {
// let mut start = Instant::now();
let logs1 = "\n\n--- DENO CODE EXECUTION ---\n".to_string();
@@ -416,7 +413,6 @@ try {{
"deno run",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;
// logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str());

9
backend/windmill-worker/src/go_executor.rs
View File

@@ -17,7 +17,7 @@ use windmill_queue::append_logs;
use crate::{
common::{
capitalize, create_args_and_out_file, get_reserved_variables, read_result,
start_child_process, OccupancyMetrics,
start_child_process,
},
handle_child::handle_child,
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, GOPRIVATE, GOPROXY,
@@ -45,7 +45,6 @@ pub async fn handle_go_job(
base_internal_url: &str,
worker_name: &str,
envs: HashMap<String, String>,
occupation_metrics: &mut OccupancyMetrics,
) -> Result<Box<RawValue>, Error> {
//go does not like executing modules at temp root
let job_dir = &format!("{job_dir}/go");
@@ -89,7 +88,6 @@ pub async fn handle_go_job(
skip_tidy,
worker_name,
&job.workspace_id,
occupation_metrics,
)
.await?;
@@ -208,7 +206,6 @@ func Run(req Req) (interface{{}}, error){{
"go build",
None,
false,
&mut Some(occupation_metrics),
)
.await?;
@@ -311,7 +308,6 @@ func Run(req Req) (interface{{}}, error){{
"go run",
job.timeout,
false,
&mut Some(occupation_metrics),
)
.await?;
@@ -350,7 +346,6 @@ pub async fn install_go_dependencies(
has_sum: bool,
worker_name: &str,
w_id: &str,
occupation_metrics: &mut OccupancyMetrics,
) -> error::Result<String> {
if !skip_go_mod {
gen_go_mymod(code, job_dir).await?;
@@ -373,7 +368,6 @@ pub async fn install_go_dependencies(
"go init",
None,
false,
&mut Some(occupation_metrics),
)
.await?;
@@ -438,7 +432,6 @@ pub async fn install_go_dependencies(
&format!("go {mod_command}"),
None,
false,
&mut Some(occupation_metrics),
)
.await?;

3
backend/windmill-worker/src/graphql_executor.rs
View File

@@ -12,7 +12,6 @@ use windmill_queue::HTTP_CLIENT;
use serde::Deserialize;
use crate::common::OccupancyMetrics;
use crate::handle_child::run_future_with_polling_update_job_poller;
use crate::{common::build_args_map, AuthedClientBackgroundTask};
@@ -41,7 +40,6 @@ pub async fn do_graphql(
db: &sqlx::Pool<sqlx::Postgres>,
mem_peak: &mut i32,
worker_name: &str,
occupation_metrics: &mut OccupancyMetrics,
) -> windmill_common::error::Result<Box<RawValue>> {
let args = build_args_map(job, client, db).await?.map(Json);
let job_args = if args.is_some() {
@@ -151,7 +149,6 @@ pub async fn do_graphql(
result_f,
worker_name,
&job.workspace_id,
&mut Some(occupation_metrics),
Box::pin(stream::once(async { 0 })),
)
.await?;

44
backend/windmill-worker/src/handle_child.rs
View File

@@ -15,7 +15,7 @@ use windmill_common::error::to_anyhow;
use windmill_common::error::{self, Error};
use windmill_common::worker::{get_windmill_memory_usage, get_worker_memory_usage, CLOUD_HOSTED};
use windmill_common::worker::CLOUD_HOSTED;
use windmill_queue::{append_logs, CanceledBy};
@@ -46,7 +46,7 @@ use futures::{
stream, StreamExt,
};
use crate::common::{resolve_job_timeout, OccupancyMetrics};
use crate::common::resolve_job_timeout;
use crate::job_logger::{append_job_logs, append_with_limit, LARGE_LOG_THRESHOLD_SIZE};
use crate::job_logger_ee::process_streaming_log_lines;
use crate::{MAX_RESULT_SIZE, MAX_WAIT_FOR_SIGINT, MAX_WAIT_FOR_SIGTERM};
@@ -99,7 +99,6 @@ pub async fn handle_child(
child_name: &str,
custom_timeout: Option<i32>,
sigterm: bool,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> error::Result<()> {
let start = Instant::now();
@@ -141,7 +140,6 @@ pub async fn handle_child(
worker,
w_id,
rx,
occupancy_metrics,
);
enum KillReason {
@@ -494,7 +492,6 @@ pub async fn run_future_with_polling_update_job_poller<Fut, T, S>(
result_f: Fut,
worker_name: &str,
w_id: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
get_mem: S,
) -> error::Result<T>
where
@@ -503,16 +500,7 @@ where
{
let (tx, rx) = broadcast::channel::<()>(3);
let update_job = update_job_poller(
job_id,
db,
mem_peak,
get_mem,
worker_name,
w_id,
rx,
occupancy_metrics,
);
let update_job = update_job_poller(job_id, db, mem_peak, get_mem, worker_name, w_id, rx);
let timeout_ms = u64::try_from(
resolve_job_timeout(&db, &w_id, job_id, timeout)
@@ -556,7 +544,6 @@ pub async fn update_job_poller<S>(
worker_name: &str,
w_id: &str,
mut rx: broadcast::Receiver<()>,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> UpdateJobPollingExit
where
S: stream::Stream<Item = i32> + Unpin,
@@ -581,37 +568,12 @@ where
_ = interval.tick() => {
// update the last_ping column every 5 seconds
i+=1;
if i == 1 || i % 10 == 0 {
let memory_usage = get_worker_memory_usage();
let wm_memory_usage = get_windmill_memory_usage();
tracing::info!("job {job_id} on {worker_name} in {w_id} worker memory snapshot {}kB/{}kB", memory_usage.unwrap_or_default()/1024, wm_memory_usage.unwrap_or_default()/1024);
let occupancy = occupancy_metrics.as_mut().map(|x| x.update_occupancy_metrics());
if job_id != Uuid::nil() {
sqlx::query!(
"UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2, memory_usage = $3, wm_memory_usage = $4,
occupancy_rate = $6, occupancy_rate_15s = $7, occupancy_rate_5m = $8, occupancy_rate_30m = $9 WHERE worker = $5",
&job_id,
&w_id,
memory_usage,
wm_memory_usage,
&worker_name,
occupancy.map(|x| x.0),
occupancy.and_then(|x| x.1),
occupancy.and_then(|x| x.2),
occupancy.and_then(|x| x.3),
)
.execute(&db)
.await
.expect("update worker ping");
}
}
let current_mem = get_mem.next().await.unwrap_or(0);
if current_mem > *mem_peak {
*mem_peak = current_mem
}
tracing::info!("job {job_id} on {worker_name} in {w_id} still running. mem: {current_mem}kB, peak mem: {mem_peak}kB");
let update_job_row = i == 2 || (!*SLOW_LOGS && (i < 20 || (i < 120 && i % 5 == 0) || i % 10 == 0)) || i % 20 == 0;
if update_job_row {
#[cfg(feature = "enterprise")]

5
backend/windmill-worker/src/js_eval.rs
View File

@@ -45,7 +45,7 @@ use windmill_common::error::Error;
use windmill_common::{flow_status::JobResult, DB};
use crate::{common::OccupancyMetrics, AuthedClient};
use crate::AuthedClient;
#[cfg(feature = "deno_core")]
use crate::{common::unsafe_raw, handle_child::run_future_with_polling_update_job_poller};
@@ -745,7 +745,6 @@ pub async fn eval_fetch_timeout(
_worker_name: &str,
_w_id: &str,
_load_client: bool,
_occupation_metrics: &mut OccupancyMetrics,
) -> anyhow::Result<Box<RawValue>> {
use serde_json::value::to_raw_value;
Ok(to_raw_value("require deno_core").unwrap())
@@ -764,7 +763,6 @@ pub async fn eval_fetch_timeout(
worker_name: &str,
w_id: &str,
load_client: bool,
occupation_metrics: &mut OccupancyMetrics,
) -> anyhow::Result<Box<RawValue>> {
use windmill_queue::append_logs;
@@ -913,7 +911,6 @@ pub async fn eval_fetch_timeout(
async { result_f.await? },
worker_name,
w_id,
&mut Some(occupation_metrics),
Box::pin(futures::stream::once(async { 0 })),
)
.await

2
backend/windmill-worker/src/mssql_executor.rs
View File

@@ -39,7 +39,6 @@ pub async fn do_mssql(
db: &sqlx::Pool<sqlx::Postgres>,
mem_peak: &mut i32,
worker_name: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<RawValue>> {
let mssql_args = build_args_values(job, client, db).await?;
@@ -157,7 +156,6 @@ pub async fn do_mssql(
result_f,
worker_name,
&job.workspace_id,
&mut Some(occupancy_metrics),
Box::pin(futures::stream::once(async { 0 })),
)
.await?;

2
backend/windmill-worker/src/mysql_executor.rs
View File

@@ -109,7 +109,6 @@ pub async fn do_mysql(
mem_peak: &mut i32,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
occupancy_metrics: &mut OccupancyMetrics,
) -> windmill_common::error::Result<Box<RawValue>> {
let args = build_args_map(job, client, db).await?.map(Json);
let job_args = if args.is_some() {
@@ -294,7 +293,6 @@ pub async fn do_mysql(
result_f,
worker_name,
&job.workspace_id,
&mut Some(occupancy_metrics),
Box::pin(futures::stream::once(async { 0 })),
)
.await?;

4
backend/windmill-worker/src/pg_executor.rs
View File

@@ -33,7 +33,7 @@ use windmill_parser_sql::{
parse_db_resource, parse_pg_statement_arg_indices, parse_pgsql_sig, parse_sql_blocks,
};
use crate::common::{build_args_values, sizeof_val, OccupancyMetrics};
use crate::common::{build_args_values, sizeof_val};
use crate::handle_child::run_future_with_polling_update_job_poller;
use crate::{AuthedClientBackgroundTask, MAX_RESULT_SIZE};
use bytes::Buf;
@@ -162,7 +162,6 @@ pub async fn do_postgresql(
mem_peak: &mut i32,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<RawValue>> {
let pg_args = build_args_values(job, client, db).await?;
@@ -351,7 +350,6 @@ pub async fn do_postgresql(
result_f,
worker_name,
&job.workspace_id,
&mut Some(occupancy_metrics),
Box::pin(futures::stream::once(async { 0 })),
)
.await?;

5
backend/windmill-worker/src/php_executor.rs
View File

@@ -70,7 +70,6 @@ pub async fn composer_install(
worker_name: &str,
requirements: String,
lock: Option<String>,
occupancy_metrics: &mut OccupancyMetrics,
) -> Result<String> {
check_executor_binary_exists("php", PHP_PATH.as_str(), "php")?;
@@ -101,7 +100,6 @@ pub async fn composer_install(
"composer install",
None,
false,
&mut Some(occupancy_metrics),
)
.await?;
@@ -142,7 +140,6 @@ pub async fn handle_php_job(
worker_name: &str,
envs: HashMap<String, String>,
shared_mount: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<RawValue>> {
check_executor_binary_exists("php", PHP_PATH.as_str(), "php")?;
@@ -172,7 +169,6 @@ pub async fn handle_php_job(
worker_name,
composer_json,
composer_lock,
occupancy_metrics,
)
.await?;
"require './vendor/autoload.php';"
@@ -328,7 +324,6 @@ try {{
"php run",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;
read_result(job_dir).await

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

@@ -80,7 +80,7 @@ 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,
read_result, start_child_process,
},
handle_child::{get_mem_peak, handle_child},
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, LOCK_CACHE_DIR,
@@ -128,7 +128,6 @@ pub async fn uv_pip_compile(
db: &Pool<Postgres>,
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,
// Debug-only flag
@@ -262,7 +261,6 @@ pub async fn uv_pip_compile(
"pip-compile",
None,
false,
occupancy_metrics,
)
.await
.map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?;
@@ -347,7 +345,6 @@ pub async fn uv_pip_compile(
"uv",
None,
false,
occupancy_metrics,
)
.await
.map_err(|e| {
@@ -518,7 +515,6 @@ pub async fn handle_python_job(
base_internal_url: &str,
envs: HashMap<String, String>,
new_args: &mut Option<HashMap<String, Box<RawValue>>>,
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(
@@ -532,7 +528,6 @@ pub async fn handle_python_job(
worker_name,
worker_dir,
mem_peak,
&mut Some(occupancy_metrics),
)
.await?;
@@ -784,7 +779,6 @@ mount {{
"python run",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;
@@ -1052,7 +1046,6 @@ async fn handle_python_deps(
worker_name: &str,
worker_dir: &str,
mem_peak: &mut i32,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
) -> error::Result<Vec<String>> {
create_dependencies_dir(job_dir).await;
@@ -1089,7 +1082,6 @@ async fn handle_python_deps(
db,
worker_name,
w_id,
occupancy_metrics,
annotations.no_uv || annotations.no_uv_compile,
annotations.no_cache,
)
@@ -1115,7 +1107,6 @@ async fn handle_python_deps(
worker_name,
job_dir,
worker_dir,
occupancy_metrics,
annotations.no_uv || annotations.no_uv_install,
false,
)
@@ -1326,7 +1317,6 @@ pub async fn handle_python_reqs(
_worker_name: &str,
job_dir: &str,
worker_dir: &str,
_occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
// TODO: Remove (Deprecated)
mut no_uv_install: bool,
is_ansible: bool,
@@ -1936,7 +1926,6 @@ pub async fn start_worker(
worker_name,
job_dir,
&mut mem_peak,
&mut None,
)
.await?;

9
backend/windmill-worker/src/rust_executor.rs
View File

@@ -16,7 +16,7 @@ use windmill_queue::append_logs;
use crate::{
common::{
check_executor_binary_exists, create_args_and_out_file, get_reserved_variables,
read_result, start_child_process, OccupancyMetrics,
read_result, start_child_process,
},
handle_child::handle_child,
AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV,
@@ -129,7 +129,6 @@ pub async fn generate_cargo_lockfile(
db: &sqlx::Pool<sqlx::Postgres>,
worker_name: &str,
w_id: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<String> {
check_executor_binary_exists("cargo", CARGO_PATH.as_str(), "rust")?;
@@ -161,7 +160,6 @@ pub async fn generate_cargo_lockfile(
"cargo generate-lockfile",
None,
false,
&mut Some(occupancy_metrics),
)
.await?;
@@ -181,7 +179,6 @@ pub async fn build_rust_crate(
w_id: &str,
base_internal_url: &str,
hash: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<String> {
let bin_path = format!("{}/{hash}", RUST_CACHE_DIR);
@@ -221,7 +218,6 @@ pub async fn build_rust_crate(
"rust build",
None,
false,
&mut Some(occupancy_metrics),
)
.await?;
append_logs(job_id, w_id, "\n\n", db).await;
@@ -280,7 +276,6 @@ pub async fn handle_rust_job(
base_internal_url: &str,
worker_name: &str,
envs: HashMap<String, String>,
occupancy_metrics: &mut OccupancyMetrics,
) -> Result<Box<RawValue>, Error> {
check_executor_binary_exists("cargo", CARGO_PATH.as_str(), "rust")?;
@@ -329,7 +324,6 @@ pub async fn handle_rust_job(
&job.workspace_id,
base_internal_url,
&hash,
occupancy_metrics,
)
.await?
};
@@ -397,7 +391,6 @@ pub async fn handle_rust_job(
"rust run",
job.timeout,
false,
&mut Some(occupancy_metrics),
)
.await?;
read_result(job_dir).await

4
backend/windmill-worker/src/snowflake_executor.rs
View File

@@ -17,7 +17,7 @@ use windmill_queue::HTTP_CLIENT;
use serde::{Deserialize, Serialize};
use crate::common::{resolve_job_timeout, OccupancyMetrics};
use crate::common::resolve_job_timeout;
use crate::handle_child::run_future_with_polling_update_job_poller;
use crate::{common::build_args_values, AuthedClientBackgroundTask};
@@ -246,7 +246,6 @@ pub async fn do_snowflake(
mem_peak: &mut i32,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
occupancy_metrics: &mut OccupancyMetrics,
) -> windmill_common::error::Result<Box<RawValue>> {
let snowflake_args = build_args_values(job, client, db).await?;
@@ -416,7 +415,6 @@ pub async fn do_snowflake(
result_f.map_err(to_anyhow),
worker_name,
&job.workspace_id,
&mut Some(occupancy_metrics),
Box::pin(futures::stream::once(async { 0 })),
)
.await?;

162
backend/windmill-worker/src/worker.rs
View File

@@ -35,12 +35,14 @@ use windmill_common::METRICS_DEBUG_ENABLED;
#[cfg(feature = "prometheus")]
use windmill_common::METRICS_ENABLED;
use futures::{future, future::Either};
use reqwest::Response;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use sqlx::{types::Json, Pool, Postgres};
use std::{
collections::HashMap,
fs::DirBuilder,
pin::pin,
sync::{
atomic::{AtomicBool, AtomicU16, Ordering},
Arc,
@@ -84,7 +86,7 @@ use tokio::{
RwLock,
},
task::JoinHandle,
time::Instant,
time::{interval, Instant, MissedTickBehavior},
};
use rand::Rng;
@@ -127,11 +129,10 @@ use crate::ansible_executor::handle_ansible_job;
#[cfg(feature = "mysql")]
use crate::mysql_executor::do_mysql;
use backon::ConstantBuilder;
use backon::{BackoffBuilder, Retryable};
#[cfg(feature = "enterprise")]
use crate::dedicated_worker::create_dedicated_worker_map;
use backon::ConstantBuilder;
use backon::{BackoffBuilder, Retryable};
#[cfg(feature = "enterprise")]
use crate::snowflake_executor::do_snowflake;
@@ -1591,28 +1592,90 @@ pub async fn run_worker(
windmill_common::otel_ee::set_span_parent(&span, &rj);
// span.context().span().add_event_with_timestamp("job created".to_string(), arc_job.created_at.into(), vec![]);
match handle_queued_job(
arc_job.clone(),
raw_code,
raw_lock,
raw_flow,
db,
&authed_client,
&hostname,
&worker_name,
&worker_dir,
&job_dir,
same_worker_tx.clone(),
base_internal_url,
job_completed_tx.clone(),
&mut occupancy_metrics,
&mut killpill_rx2,
#[cfg(feature = "benchmark")]
&mut bench,
let mut script_duration = 0f32;
let mut ping_interval = interval(Duration::from_secs(5));
ping_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
let Either::Left((result, _)) = future::select(
// Handle job:
pin!(handle_queued_job(
arc_job.clone(),
raw_code,
raw_lock,
raw_flow,
db,
&authed_client,
&hostname,
&worker_name,
&worker_dir,
&job_dir,
same_worker_tx.clone(),
base_internal_url,
job_completed_tx.clone(),
&mut script_duration,
&mut killpill_rx2,
#[cfg(feature = "benchmark")]
&mut bench,
)
.instrument(span)),
// Ping worker every 5 seconds:
pin!(async {
loop {
ping_interval.tick().await;
let memory_usage = get_worker_memory_usage();
let wm_memory_usage = get_windmill_memory_usage();
tracing::info!(
"job {} on {worker_name} in {} worker memory snapshot {}kB/{}kB",
&arc_job.id,
&arc_job.workspace_id,
memory_usage.unwrap_or_default() / 1024,
wm_memory_usage.unwrap_or_default() / 1024
);
let (
occupancy_rate,
occupancy_rate_15s,
occupancy_rate_5m,
occupancy_rate_30m,
) = occupancy_metrics.update_occupancy_metrics();
let ping_res = sqlx::query!(
"UPDATE worker_ping SET
ping_at = now()
, current_job_id = $1
, current_job_workspace_id = $2
, memory_usage = $3
, wm_memory_usage = $4
, occupancy_rate = $6
, occupancy_rate_15s = $7
, occupancy_rate_5m = $8
, occupancy_rate_30m = $9
WHERE worker = $5",
&arc_job.id,
&arc_job.workspace_id,
memory_usage,
wm_memory_usage,
&worker_name,
occupancy_rate,
occupancy_rate_15s,
occupancy_rate_5m,
occupancy_rate_30m
)
.execute(db)
.await;
if let Err(err) = ping_res {
tracing::error!("failed to update worker ping: {}", err);
}
last_ping = Instant::now();
}
}),
)
.instrument(span)
.await
{
else {
unreachable!()
};
occupancy_metrics.total_duration_of_running_jobs += script_duration;
match result {
Err(err) => {
handle_job_error(
db,
@@ -1672,7 +1735,7 @@ pub async fn run_worker(
if !KEEP_JOB_DIR.load(Ordering::Relaxed) && !(arc_job.is_flow() && same_worker)
{
let _ = tokio::fs::remove_dir_all(job_dir).await;
let _ = tokio::fs::remove_dir_all(&job_dir).await;
}
}
@@ -1844,7 +1907,6 @@ async fn do_nativets(
db: &Pool<Postgres>,
mem_peak: &mut i32,
worker_name: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> windmill_common::error::Result<Box<RawValue>> {
let args = build_args_map(job, client, db).await?.map(Json);
let job_args = if args.is_some() {
@@ -1865,7 +1927,6 @@ async fn do_nativets(
worker_name,
&job.workspace_id,
true,
occupancy_metrics,
)
.await?)
}
@@ -1890,7 +1951,7 @@ async fn handle_queued_job(
same_worker_tx: SameWorkerSender,
base_internal_url: &str,
job_completed_tx: JobCompletedSender,
occupancy_metrics: &mut OccupancyMetrics,
duration: &mut f32,
killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> windmill_common::error::Result<bool> {
@@ -2083,7 +2144,6 @@ async fn handle_queued_job(
worker_dir,
base_internal_url,
&client.get_token().await,
occupancy_metrics,
)
.await
}
@@ -2098,7 +2158,6 @@ async fn handle_queued_job(
worker_dir,
base_internal_url,
&client.get_token().await,
occupancy_metrics,
)
.await
}
@@ -2111,7 +2170,6 @@ async fn handle_queued_job(
worker_dir,
base_internal_url,
&client.get_token().await,
occupancy_metrics,
)
.await
.map(|()| serde_json::from_str("{}").unwrap()),
@@ -2140,12 +2198,10 @@ async fn handle_queued_job(
worker_name,
&mut column_order,
&mut new_args,
occupancy_metrics,
killpill_rx,
)
.await;
occupancy_metrics.total_duration_of_running_jobs +=
metric_timer.elapsed().as_secs_f32();
*duration = metric_timer.elapsed().as_secs_f32();
r
}
};
@@ -2264,7 +2320,6 @@ async fn handle_code_execution_job(
worker_name: &str,
column_order: &mut Option<Vec<String>>,
new_args: &mut Option<HashMap<String, Box<RawValue>>>,
occupancy_metrics: &mut OccupancyMetrics,
killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
) -> error::Result<Box<RawValue>> {
let script_hash = || {
@@ -2353,7 +2408,6 @@ async fn handle_code_execution_job(
mem_peak,
worker_name,
column_order,
occupancy_metrics,
)
.await;
} else if language == Some(ScriptLang::Mysql) {
@@ -2371,7 +2425,6 @@ async fn handle_code_execution_job(
mem_peak,
worker_name,
column_order,
occupancy_metrics,
)
.await;
} else if language == Some(ScriptLang::Bigquery) {
@@ -2400,7 +2453,6 @@ async fn handle_code_execution_job(
mem_peak,
worker_name,
column_order,
occupancy_metrics,
)
.await;
}
@@ -2422,7 +2474,6 @@ async fn handle_code_execution_job(
mem_peak,
worker_name,
column_order,
occupancy_metrics,
)
.await;
}
@@ -2444,28 +2495,10 @@ async fn handle_code_execution_job(
#[cfg(all(feature = "enterprise", feature = "mssql"))]
{
return do_mssql(
job,
&client,
&code,
db,
mem_peak,
worker_name,
occupancy_metrics,
)
.await;
return do_mssql(job, &client, &code, db, mem_peak, worker_name).await;
}
} else if language == Some(ScriptLang::Graphql) {
return do_graphql(
job,
&client,
&code,
db,
mem_peak,
worker_name,
occupancy_metrics,
)
.await;
return do_graphql(job, &client, &code, db, mem_peak, worker_name).await;
} else if language == Some(ScriptLang::Nativets) {
append_logs(
&job.id,
@@ -2493,7 +2526,6 @@ async fn handle_code_execution_job(
db,
mem_peak,
worker_name,
occupancy_metrics,
)
.await?;
return Ok(result);
@@ -2563,7 +2595,6 @@ mount {{
base_internal_url,
envs,
new_args,
occupancy_metrics,
)
.await
}
@@ -2580,7 +2611,6 @@ mount {{
worker_name,
envs,
new_args,
occupancy_metrics,
)
.await
}
@@ -2599,7 +2629,6 @@ mount {{
envs,
&shared_mount,
new_args,
occupancy_metrics,
)
.await
}
@@ -2616,7 +2645,6 @@ mount {{
base_internal_url,
worker_name,
envs,
occupancy_metrics,
)
.await
}
@@ -2632,7 +2660,6 @@ mount {{
base_internal_url,
worker_name,
envs,
occupancy_metrics,
killpill_rx,
)
.await
@@ -2649,7 +2676,6 @@ mount {{
base_internal_url,
worker_name,
envs,
occupancy_metrics,
)
.await
}
@@ -2672,7 +2698,6 @@ mount {{
worker_name,
envs,
&shared_mount,
occupancy_metrics,
)
.await
}
@@ -2695,7 +2720,6 @@ mount {{
base_internal_url,
worker_name,
envs,
occupancy_metrics,
)
.await
}
@@ -2719,7 +2743,6 @@ mount {{
&shared_mount,
base_internal_url,
envs,
occupancy_metrics,
)
.await
}
@@ -2736,7 +2759,6 @@ mount {{
base_internal_url,
worker_name,
envs,
occupancy_metrics,
)
.await
}

43
backend/windmill-worker/src/worker_lockfiles.rs
View File

@@ -32,7 +32,6 @@ use windmill_parser_py_imports::parse_relative_imports;
use windmill_parser_ts::parse_expr_for_imports;
use windmill_queue::{append_logs, PushIsolationLevel};
use crate::common::OccupancyMetrics;
use crate::csharp_executor::generate_nuget_lockfile;
#[cfg(feature = "php")]
@@ -226,7 +225,6 @@ pub async fn handle_dependency_job(
worker_dir: &str,
base_internal_url: &str,
token: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<RawValue>> {
let script_path = job.script_path();
let raw_deps = job
@@ -285,7 +283,6 @@ pub async fn handle_dependency_job(
script_path,
raw_deps,
npm_mode,
occupancy_metrics,
)
.await;
@@ -543,7 +540,6 @@ pub async fn handle_flow_dependency_job(
worker_dir: &str,
base_internal_url: &str,
token: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<serde_json::value::RawValue>> {
let job_path = job.script_path.clone().ok_or_else(|| {
error::Error::InternalErr(
@@ -621,7 +617,6 @@ pub async fn handle_flow_dependency_job(
base_internal_url,
token,
&nodes_to_relock,
occupancy_metrics,
)
.await?;
let new_flow_value = Json(serde_json::value::to_raw_value(&flow).map_err(to_anyhow)?);
@@ -744,7 +739,6 @@ async fn lock_modules<'c>(
base_internal_url: &str,
token: &str,
locks_to_reload: &Option<Vec<String>>,
occupancy_metrics: &mut OccupancyMetrics,
// (modules to replace old seq (even unmmodified ones), new transaction, modified ids) )
) -> Result<(
Vec<FlowModule>,
@@ -791,7 +785,6 @@ async fn lock_modules<'c>(
base_internal_url,
token,
locks_to_reload,
occupancy_metrics,
))
.await?;
e.value = FlowModuleValue::ForloopFlow {
@@ -823,7 +816,6 @@ async fn lock_modules<'c>(
base_internal_url,
token,
locks_to_reload,
occupancy_metrics,
))
.await?;
nmodified_ids.extend(inner_modified_ids);
@@ -847,7 +839,6 @@ async fn lock_modules<'c>(
base_internal_url,
token,
locks_to_reload,
occupancy_metrics,
))
.await?;
e.value = FlowModuleValue::WhileloopFlow {
@@ -877,7 +868,6 @@ async fn lock_modules<'c>(
base_internal_url,
token,
locks_to_reload,
occupancy_metrics,
))
.await?;
nmodified_ids.extend(inner_modified_ids);
@@ -898,7 +888,6 @@ async fn lock_modules<'c>(
base_internal_url,
token,
locks_to_reload,
occupancy_metrics,
))
.await?;
e.value = FlowModuleValue::BranchOne {
@@ -950,7 +939,6 @@ async fn lock_modules<'c>(
),
false,
None,
occupancy_metrics,
)
.await;
//
@@ -1313,7 +1301,6 @@ async fn lock_modules_app(
job_path: &str,
base_internal_url: &str,
token: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> Result<Value> {
match value {
Value::Object(mut m) => {
@@ -1357,7 +1344,6 @@ async fn lock_modules_app(
&format!("{}/app", job.script_path()),
false,
None,
occupancy_metrics,
)
.await;
match new_lock {
@@ -1414,7 +1400,6 @@ async fn lock_modules_app(
job_path,
base_internal_url,
token,
occupancy_metrics,
)
.await?,
);
@@ -1436,7 +1421,6 @@ async fn lock_modules_app(
job_path,
base_internal_url,
token,
occupancy_metrics,
)
.await?,
);
@@ -1456,7 +1440,6 @@ pub async fn handle_app_dependency_job(
worker_dir: &str,
base_internal_url: &str,
token: &str,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<()> {
let job_path = job.script_path.clone().ok_or_else(|| {
error::Error::InternalErr(
@@ -1486,7 +1469,6 @@ pub async fn handle_app_dependency_job(
&job_path,
base_internal_url,
token,
occupancy_metrics,
)
.await?;
@@ -1575,7 +1557,6 @@ async fn python_dep(
worker_name: &str,
w_id: &str,
worker_dir: &str,
occupancy_metrics: &mut Option<&mut OccupancyMetrics>,
no_uv_compile: bool,
no_uv_install: bool,
) -> std::result::Result<String, Error> {
@@ -1588,7 +1569,6 @@ async fn python_dep(
db,
worker_name,
w_id,
occupancy_metrics,
no_uv_compile,
false,
)
@@ -1604,7 +1584,6 @@ async fn python_dep(
worker_name,
job_dir,
worker_dir,
occupancy_metrics,
no_uv_install,
false,
)
@@ -1635,7 +1614,6 @@ async fn capture_dependency_job(
script_path: &str,
raw_deps: bool,
npm_mode: Option<bool>,
occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<String> {
match job_language {
ScriptLang::Python3 => {
@@ -1689,7 +1667,6 @@ async fn capture_dependency_job(
worker_name,
w_id,
worker_dir,
&mut Some(occupancy_metrics),
no_uv_compile | no_uv,
no_uv_install | no_uv,
)
@@ -1736,7 +1713,6 @@ async fn capture_dependency_job(
worker_name,
w_id,
worker_dir,
&mut Some(occupancy_metrics),
false,
false,
)
@@ -1760,7 +1736,6 @@ async fn capture_dependency_job(
false,
worker_name,
w_id,
occupancy_metrics,
)
.await
}
@@ -1779,7 +1754,6 @@ async fn capture_dependency_job(
w_id,
worker_name,
base_internal_url,
&mut Some(occupancy_metrics),
)
.await
}
@@ -1807,7 +1781,6 @@ async fn capture_dependency_job(
None
},
npm_mode,
&mut Some(occupancy_metrics),
)
.await?;
if req.is_some() && !raw_deps {
@@ -1822,7 +1795,6 @@ async fn capture_dependency_job(
base_internal_url,
worker_name,
&token,
&mut Some(occupancy_metrics),
)
.await?;
}
@@ -1849,18 +1821,7 @@ async fn capture_dependency_job(
}
}
};
composer_install(
mem_peak,
job_id,
w_id,
db,
job_dir,
worker_name,
reqs,
None,
occupancy_metrics,
)
.await
composer_install(mem_peak, job_id, w_id, db, job_dir, worker_name, reqs, None).await
}
}
ScriptLang::Rust => {
@@ -1884,7 +1845,6 @@ async fn capture_dependency_job(
db,
worker_name,
w_id,
occupancy_metrics,
)
.await?;
@@ -1906,7 +1866,6 @@ async fn capture_dependency_job(
db,
worker_name,
w_id,
occupancy_metrics,
)
.await
}