diff --git a/backend/.sqlx/query-c6423de98ffedc4e58cca0bd7883ef87f04a3daa054882a0c4e81c723196dbb6.json b/backend/.sqlx/query-c6423de98ffedc4e58cca0bd7883ef87f04a3daa054882a0c4e81c723196dbb6.json new file mode 100644 index 0000000000..d519197cde --- /dev/null +++ b/backend/.sqlx/query-c6423de98ffedc4e58cca0bd7883ef87f04a3daa054882a0c4e81c723196dbb6.json @@ -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" +} diff --git a/backend/src/main.rs b/backend/src/main.rs index 45aa87e99a..216e39ace7 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -159,7 +159,6 @@ async fn cache_hub_scripts(file_path: Option) -> 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) -> anyhow::Result<()> { "cache_init", envs.clone(), false, - &mut None, ) .await?; @@ -196,7 +194,6 @@ async fn cache_hub_scripts(file_path: Option) -> anyhow::Result<()> { "", "cache_init", "", - &mut None, ) .await { diff --git a/backend/windmill-worker/src/ansible_executor.rs b/backend/windmill-worker/src/ansible_executor.rs index 2f68ebb2b8..b8635032e6 100644 --- a/backend/windmill-worker/src/ansible_executor.rs +++ b/backend/windmill-worker/src/ansible_executor.rs @@ -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> { 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, - 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, - occupancy_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { 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 diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 84d79f665c..6bc84a4fa9 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -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, - occupancy_metrics: &mut OccupancyMetrics, _killpill_rx: &mut tokio::sync::broadcast::Receiver<()>, ) -> Result, 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, mem_peak: &mut i32, worker_name: &str, - occupancy_metrics: &mut OccupancyMetrics, killpill_rx: &mut tokio::sync::broadcast::Receiver<()>, ) -> Result, 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, - occupancy_metrics: &mut OccupancyMetrics, ) -> Result, 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?; diff --git a/backend/windmill-worker/src/bigquery_executor.rs b/backend/windmill-worker/src/bigquery_executor.rs index 1b60785bee..0f53cb6218 100644 --- a/backend/windmill-worker/src/bigquery_executor.rs +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -209,7 +209,6 @@ pub async fn do_bigquery( mem_peak: &mut i32, worker_name: &str, column_order: &mut Option>, - occupancy_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { 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?; diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 7c33d6d9e3..820ea6d42d 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -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, npm_mode: bool, - occupancy_metrics: &mut Option<&mut OccupancyMetrics>, ) -> Result> { let common_bun_proc_envs: HashMap = 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, 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, mem_peak: &mut i32, common_bun_proc_envs: &HashMap, - 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, mem_peak: &mut i32, common_bun_proc_envs: &HashMap, - 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, shared_mount: &str, new_args: &mut Option>>, - occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { 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?; } diff --git a/backend/windmill-worker/src/csharp_executor.rs b/backend/windmill-worker/src/csharp_executor.rs index 419cea1bf0..6999fdb44f 100644 --- a/backend/windmill-worker/src/csharp_executor.rs +++ b/backend/windmill-worker/src/csharp_executor.rs @@ -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, worker_name: &str, w_id: &str, - occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result { 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, _worker_name: &str, _w_id: &str, - _occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result { 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 { 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, - _occupancy_metrics: &mut OccupancyMetrics, ) -> Result, 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, - occupancy_metrics: &mut OccupancyMetrics, ) -> Result, 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 diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 76c1f3bf7c..b0ce9203c0 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -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 { 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, new_args: &mut Option>>, - occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { // 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()); diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index f7ac8f93d9..3514e38a5e 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -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, - occupation_metrics: &mut OccupancyMetrics, ) -> Result, 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 { 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?; diff --git a/backend/windmill-worker/src/graphql_executor.rs b/backend/windmill-worker/src/graphql_executor.rs index 38c9c8f71d..32a2aa6840 100644 --- a/backend/windmill-worker/src/graphql_executor.rs +++ b/backend/windmill-worker/src/graphql_executor.rs @@ -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, mem_peak: &mut i32, worker_name: &str, - occupation_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { 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?; diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 0b351ced99..33820eae6b 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -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, 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( result_f: Fut, worker_name: &str, w_id: &str, - occupancy_metrics: &mut Option<&mut OccupancyMetrics>, get_mem: S, ) -> error::Result 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( worker_name: &str, w_id: &str, mut rx: broadcast::Receiver<()>, - occupancy_metrics: &mut Option<&mut OccupancyMetrics>, ) -> UpdateJobPollingExit where S: stream::Stream + 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")] diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index a3327775a8..fe80d5b1d1 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -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> { 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> { 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 diff --git a/backend/windmill-worker/src/mssql_executor.rs b/backend/windmill-worker/src/mssql_executor.rs index 860ea29309..59b374aa32 100644 --- a/backend/windmill-worker/src/mssql_executor.rs +++ b/backend/windmill-worker/src/mssql_executor.rs @@ -39,7 +39,6 @@ pub async fn do_mssql( db: &sqlx::Pool, mem_peak: &mut i32, worker_name: &str, - occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { 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?; diff --git a/backend/windmill-worker/src/mysql_executor.rs b/backend/windmill-worker/src/mysql_executor.rs index b9c97dc276..1d381379ea 100644 --- a/backend/windmill-worker/src/mysql_executor.rs +++ b/backend/windmill-worker/src/mysql_executor.rs @@ -109,7 +109,6 @@ pub async fn do_mysql( mem_peak: &mut i32, worker_name: &str, column_order: &mut Option>, - occupancy_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { 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?; diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 015f12bbc7..65c216af94 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -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>, - occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { 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?; diff --git a/backend/windmill-worker/src/php_executor.rs b/backend/windmill-worker/src/php_executor.rs index c13cc35fdc..b840815351 100644 --- a/backend/windmill-worker/src/php_executor.rs +++ b/backend/windmill-worker/src/php_executor.rs @@ -70,7 +70,6 @@ pub async fn composer_install( worker_name: &str, requirements: String, lock: Option, - occupancy_metrics: &mut OccupancyMetrics, ) -> Result { 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, shared_mount: &str, - occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result> { 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 diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index e19e44073e..962edc5beb 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -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, 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, new_args: &mut Option>>, - occupancy_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { 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> { 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?; diff --git a/backend/windmill-worker/src/rust_executor.rs b/backend/windmill-worker/src/rust_executor.rs index 0630468ec7..0f78084ca3 100644 --- a/backend/windmill-worker/src/rust_executor.rs +++ b/backend/windmill-worker/src/rust_executor.rs @@ -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, worker_name: &str, w_id: &str, - occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result { 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 { 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, - occupancy_metrics: &mut OccupancyMetrics, ) -> Result, 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 diff --git a/backend/windmill-worker/src/snowflake_executor.rs b/backend/windmill-worker/src/snowflake_executor.rs index 43368f1503..e7fdc4e623 100644 --- a/backend/windmill-worker/src/snowflake_executor.rs +++ b/backend/windmill-worker/src/snowflake_executor.rs @@ -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>, - occupancy_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { 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?; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 4b6fb16c99..8f015d4436 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -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, mem_peak: &mut i32, worker_name: &str, - occupancy_metrics: &mut OccupancyMetrics, ) -> windmill_common::error::Result> { 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 { @@ -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>, new_args: &mut Option>>, - occupancy_metrics: &mut OccupancyMetrics, killpill_rx: &mut tokio::sync::broadcast::Receiver<()>, ) -> error::Result> { 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 } diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 48a7c43d25..76af72980f 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -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> { 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> { 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>, - occupancy_metrics: &mut OccupancyMetrics, // (modules to replace old seq (even unmmodified ones), new transaction, modified ids) ) ) -> Result<( Vec, @@ -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 { 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 { @@ -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, - occupancy_metrics: &mut OccupancyMetrics, ) -> error::Result { 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 }