diff --git a/backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json b/backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json index 1b8084742c..7718e05ccf 100644 --- a/backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json +++ b/backend/.sqlx/query-123c0608e229c29187009b7961355ddd99c4ad1f46b876dd86e372b84d806ecd.json @@ -37,7 +37,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c.json b/backend/.sqlx/query-25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c.json index bf591ef11c..9d082a6772 100644 --- a/backend/.sqlx/query-25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c.json +++ b/backend/.sqlx/query-25bef6a248f3ee0ea2cbcc376c217cbcf1013ae311c36b42d423bf6a02fa016c.json @@ -67,7 +67,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json b/backend/.sqlx/query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json index a5dee163e5..1166260449 100644 --- a/backend/.sqlx/query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json +++ b/backend/.sqlx/query-438b5b5d29b05846c2e074cad2404e797527841cd97dba80c271cbefafae65cc.json @@ -28,7 +28,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json b/backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json index 1517c8d1d4..eabf671894 100644 --- a/backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json +++ b/backend/.sqlx/query-5cd89ab614d3cac80fb81627267ee85b191263989b0c78b2bfce77e796e96825.json @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json b/backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json index bfe7c41f64..c52efca4c0 100644 --- a/backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json +++ b/backend/.sqlx/query-65835f2e5ad38f7cc6b147dadfef6f580f15bca96d9746c9359e98ca793f8f1f.json @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-2de52e1f3226ca9281b6e25f74d4d05f4509cb87c875234bfc7b310a012e4d40.json b/backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json similarity index 73% rename from backend/.sqlx/query-2de52e1f3226ca9281b6e25f74d4d05f4509cb87c875234bfc7b310a012e4d40.json rename to backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json index 9c386c3509..73e6e17c76 100644 --- a/backend/.sqlx/query-2de52e1f3226ca9281b6e25f74d4d05f4509cb87c875234bfc7b310a012e4d40.json +++ b/backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id", + "query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $11)\n )\n INSERT INTO queue \n (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10 FROM uuid_table) \n RETURNING id", "describe": { "columns": [ { @@ -46,7 +46,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", @@ -70,5 +69,5 @@ false ] }, - "hash": "2de52e1f3226ca9281b6e25f74d4d05f4509cb87c875234bfc7b310a012e4d40" + "hash": "6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d" } diff --git a/backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json b/backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json index 6308bf3bb2..fead4ba250 100644 --- a/backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json +++ b/backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json @@ -60,7 +60,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json b/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json index 38f81da395..c90719118a 100644 --- a/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json +++ b/backend/.sqlx/query-b224cdd1221fc9e7227ef8e8c025eedc09bceb82bca349f3e31c8513ebbf0192.json @@ -42,7 +42,6 @@ "bash", "postgresql", "nativets", - "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 659f9213b9..b2ba7d2b13 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -63,7 +63,7 @@ pub async fn initial_load( db: &Pool, tx: tokio::sync::broadcast::Sender<()>, worker_mode: bool, - server_mode: bool, + server_mode: bool ) { let reload_worker_config_f = async { if worker_mode { diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index b97a06c747..2457e543a9 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -2332,7 +2332,6 @@ struct BatchInfo { kind: String, flow_value: Option, path: Option, - dedicated_worker: Option, } #[tracing::instrument(level = "trace", skip_all)] @@ -2362,7 +2361,7 @@ async fn add_batch_jobs( batch_info.path, JobKind::Script, Some(script.language), - batch_info.dedicated_worker, + script.dedicated_worker, ) } "flow" => { @@ -2433,15 +2432,22 @@ async fn add_batch_jobs( format!("{}", language.as_str()) }; - let uuids = sqlx::query_scalar!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id", + let uuids = sqlx::query_scalar!( + r#"WITH uuid_table as ( + select gen_random_uuid() as uuid from generate_series(1, $11) + ) + INSERT INTO queue + (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) + (SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10 FROM uuid_table) + RETURNING id"#, hash.map(|h| h.0), path, job_kind.clone() as JobKind, language as ScriptLang, tag, authed.username, - authed.email, username_to_permissioned_as(&authed.username), + authed.email, Utc::now(), w_id, n diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 960cdf3a98..af17eae543 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -161,6 +161,7 @@ pub struct Script { pub concurrent_limit: Option, #[serde(skip_serializing_if = "Option::is_none")] pub concurrency_time_window_s: Option, + pub dedicated_worker: Option, } #[derive(Serialize)] diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index d46ab7bbb6..df9e4721d8 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -464,6 +464,10 @@ pub async fn start_worker( mut jobs_rx: Receiver, mut killpill_rx: tokio::sync::broadcast::Receiver<()>, ) -> Result<()> { + use std::task::Poll; + + use futures::{future, Future}; + let mut logs = "".to_string(); let _ = write_file(job_dir, "main.ts", inner_content).await?; let common_bun_proc_envs: HashMap = @@ -684,6 +688,21 @@ plugin(p) // let mut i = 0; // let mut j = 0; let mut alive = true; + + fn conditional_polling( + fut: impl Future, + predicate: bool, + ) -> impl Future { + let mut fut = Box::pin(fut); + future::poll_fn(move |cx| { + if predicate { + fut.as_mut().poll(cx) + } else { + Poll::Pending + } + }) + } + loop { tokio::select! { biased; @@ -711,8 +730,8 @@ plugin(p) tracing::info!("dedicated worker process exited"); break; } - } - job = jobs_rx.recv(), if alive && jobs.len() < MAX_BUFFERED_DEDICATED_JOBS => { + }, + job = conditional_polling(jobs_rx.recv(), alive && jobs.len() < MAX_BUFFERED_DEDICATED_JOBS) => { // i += 1; if let Some(job) = job { tracing::debug!("received job"); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index b3fa1fe62e..072202134b 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -10,12 +10,16 @@ use anyhow::Result; use const_format::concatcp; use itertools::Itertools; use once_cell::sync::OnceCell; +use prometheus::core::{AtomicU64, GenericCounter}; #[cfg(feature = "benchmark")] use serde::Serialize; use sqlx::{Pool, Postgres}; use std::{ collections::HashMap, - sync::{atomic::Ordering, Arc}, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, time::Duration, }; use windmill_api_client::Client; @@ -337,6 +341,52 @@ macro_rules! add_time { }; } +async fn handle_receive_completed_job< + R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static, +>( + jc: JobCompleted, + worker_execution_failed: HashMap, GenericCounter>, + base_internal_url: String, + db: Pool, + worker_dir: String, + same_worker_tx: Sender, + rsmq: Option, +) { + let metrics = build_language_metrics(&worker_execution_failed.clone(), &jc.job.language); + let token = jc.token.clone(); + let workspace = jc.job.workspace_id.clone(); + let client = AuthedClient { + base_internal_url: base_internal_url.to_string(), + workspace, + token, + client: OnceCell::new(), + }; + if let Err(err) = process_completed_job( + &jc, + &client, + &db, + &worker_dir, + metrics.clone(), + same_worker_tx.clone(), + rsmq.clone(), + ) + .await + { + handle_job_error( + &db, + &client, + &jc.job, + err, + metrics, + false, + same_worker_tx.clone(), + &worker_dir, + rsmq.clone(), + ) + .await; + } +} + pub async fn run_worker( db: &Pool, worker_instance: &str, @@ -549,43 +599,168 @@ pub async fn run_worker, + None::, + JobKind::Noop as JobKind, + ScriptLang::Deno as ScriptLang, + "deno", + "admin", + "u/admin", + "admin@windmill.dev", + chrono::Utc::now(), + "admins", + jobs + ) + .execute(db) + .await.unwrap_or_else(|_e| panic!("failed to insert noop jobs")); + } + } + + #[cfg(feature = "benchmark")] + let completed_jobs = Arc::new(AtomicUsize::new(0)); + #[cfg(feature = "benchmark")] + let start = Instant::now(); + #[cfg(feature = "benchmark")] + let main_duration = Arc::new(AtomicUsize::new(0)); + #[cfg(feature = "benchmark")] + let send_duration = Arc::new(AtomicUsize::new(0)); + #[cfg(feature = "benchmark")] + let process_duration = Arc::new(AtomicUsize::new(0)); + + #[cfg(feature = "benchmark")] + let main_duration2 = main_duration.clone(); + #[cfg(feature = "benchmark")] + let send_duration2 = send_duration.clone(); + let send_result = tokio::spawn(async move { while let Some(jc) = job_completed_rx.recv().await { - let metrics = build_language_metrics(&worker_execution_failed2, &jc.job.language); - let token = jc.token.clone(); - let workspace = jc.job.workspace_id.clone(); - let client = AuthedClient { - base_internal_url: base_internal_url2.to_string(), - workspace, - token, - client: OnceCell::new(), - }; - if let Err(err) = process_completed_job( - &jc, - &client, - &db2, - &worker_dir2, - metrics.clone(), - same_worker_tx2.clone(), - rsmq2.clone(), - ) - .await - { - handle_job_error( - &db2, - &client, - &jc.job, - err, - metrics, - false, - same_worker_tx2.clone(), - &worker_dir2, - rsmq2.clone(), + let base_internal_url2 = base_internal_url2.clone(); + let worker_execution_failed2 = worker_execution_failed2.clone(); + let worker_dir2 = worker_dir2.clone(); + let db2 = db2.clone(); + let same_worker_tx2 = same_worker_tx2.clone(); + let rsmq2 = rsmq2.clone(); + + if matches!(jc.job.job_kind, JobKind::Noop) || is_dedicated_worker { + thread_count.fetch_add(1, Ordering::SeqCst); + let thread_count = thread_count.clone(); + + #[cfg(feature = "benchmark")] + let send_duration = send_duration2.clone(); + #[cfg(feature = "benchmark")] + let process_duration = process_duration.clone(); + #[cfg(feature = "benchmark")] + let completed_jobs = completed_jobs.clone(); + #[cfg(feature = "benchmark")] + let main_duration = main_duration2.clone(); + + tokio::spawn(async move { + #[cfg(feature = "benchmark")] + let process_start = Instant::now(); + + handle_receive_completed_job( + jc, + worker_execution_failed2, + base_internal_url2, + db2, + worker_dir2, + same_worker_tx2, + rsmq2, + ) + .await; + #[cfg(feature = "benchmark")] + { + let n = completed_jobs.fetch_add(1, Ordering::SeqCst); + if (n + 1) % 1000 == 0 || n == (jobs - 1) as usize { + let duration_s = start.elapsed().as_secs_f64(); + let jobs_per_sec = n as f64 / duration_s; + tracing::info!( + "completed {} jobs in {}s, {} jobs/s", + n + 1, + duration_s, + jobs_per_sec + ); + + tracing::info!( + "main loop without send {}s", + main_duration.load(Ordering::SeqCst) as f64 / 1000.0 + ); + + tracing::info!( + "send job completed / send dedicated job duration {}s", + send_duration.load(Ordering::SeqCst) as f64 / 1000.0 + ); + + tracing::info!( + "job completed process duration {}s", + process_duration.load(Ordering::SeqCst) as f64 / 1000.0 + ); + } + + process_duration.fetch_add( + process_start.elapsed().as_millis() as usize, + Ordering::SeqCst, + ); + } + + thread_count.fetch_sub(1, Ordering::SeqCst); + }); + } else { + handle_receive_completed_job( + jc, + worker_execution_failed2, + base_internal_url2, + db2, + worker_dir2, + same_worker_tx2, + rsmq2, ) .await; } } + tracing::info!("stopped processing new completed jobs"); + while thread_count.load(Ordering::SeqCst) > 0 { + tokio::time::sleep(Duration::from_millis(100)).await; + } + tracing::info!("finished processing all completed jobs"); + // if let Err(e) = // add_completed_job(&db2, &job, success, false, result, logs, rsmq2.clone()).await // { @@ -719,6 +894,9 @@ pub async fn run_worker>, Option>) }; + #[cfg(feature = "benchmark")] + tracing::info!("pre loop time {}s", start.elapsed().as_secs_f64()); + loop { #[cfg(feature = "benchmark")] let loop_start = Instant::now(); @@ -889,11 +1067,27 @@ pub async fn run_worker new TextEncoder().encode(s); + async function getQueueCount() { + return ( + await ( + await fetch( + config.server + "/api/w/" + config.workspace_id + "/jobs/queue/count", + { headers: { ["Authorization"]: "Bearer " + config.token } } + ) + ).json() + ).database_length; + } + + let pastJobs = 0; + async function getCompletedJobsCount(): Promise { + const completedJobs = ( + await ( + await fetch( + host + "/api/w/" + config.workspace_id + "/jobs/completed/count", + { headers: { ["Authorization"]: "Bearer " + config.token } } + ) + ).json() + ).database_length; + return completedJobs - pastJobs; + } + if (["deno", "python", "go", "bash", "dedicated", "bun"].includes(kind)) { await createBenchScript(kind, workspace); } - let jobsSent = jobs; + pastJobs = await getCompletedJobsCount(); + + const jobsSent = jobs; console.log(`Bulk creating ${jobsSent} jobs`); const start_create = Date.now(); @@ -84,9 +137,8 @@ export async function main({ body = JSON.stringify({ kind: "script", path: "f/benchmarks/" + kind, - dedicated_worker: kind === "dedicated", }); - } else if (["2steps", "onebranch", "branchallparrallel"].includes(kind)) { + } else if (["2steps"].includes(kind)) { const payload = getFlowPayload(kind); body = JSON.stringify({ kind: "flow", @@ -113,6 +165,7 @@ export async function main({ if (!response.ok) { throw new Error("Failed to create jobs: " + response.statusText); } + const uuids = await response.json(); const end_create = Date.now(); const create_duration = end_create - start_create; console.log( @@ -122,69 +175,61 @@ export async function main({ ); let start = Date.now(); - let queue_length = jobsSent; + let completedJobs = 0; let lastElapsed = 0; - let lastQueueLength = queue_length; - const updateState = setInterval(async () => { - const elapsed = start ? Date.now() - start : 0; - queue_length = ( - await ( - await fetch( - host + "/api/w/" + config.workspace_id + "/jobs/queue/count", - { headers: { ["Authorization"]: "Bearer " + config.token } } + let lastCompletedJobs = 0; + + let didStart = false; + while (completedJobs < jobsSent) { + if (!didStart) { + const actual_queue = await getQueueCount(); + if (actual_queue < jobsSent) { + start = Date.now(); + didStart = true; + } + } else { + const elapsed = start ? Date.now() - start : 0; + completedJobs = await getCompletedJobsCount(); + if (kind === "2steps") { + completedJobs = Math.floor(completedJobs / 3); + } + const avgThr = ((completedJobs / elapsed) * 1000).toFixed(2); + const instThr = + lastElapsed > 0 + ? ( + ((completedJobs - lastCompletedJobs) / (elapsed - lastElapsed)) * + 1000 + ).toFixed(2) + : 0; + + lastElapsed = elapsed; + lastCompletedJobs = completedJobs; + + await Deno.stdout.write( + enc( + `elapsed: ${(elapsed / 1000).toFixed( + 2 + )} | jobs executed: ${completedJobs}/${jobsSent} (thr: inst ${instThr} - avg ${avgThr}) | remaining: ${ + jobsSent - completedJobs + } \r` ) - ).json() - ).database_length; - const avgThr = (((jobsSent - queue_length) / elapsed) * 1000).toFixed(2); - const instThr = - lastElapsed > 0 - ? ( - ((lastQueueLength - queue_length) / (elapsed - lastElapsed)) * - 1000 - ).toFixed(2) - : 0; - - lastElapsed = elapsed; - lastQueueLength = queue_length; - - await Deno.stdout.write( - enc( - `elapsed: ${(elapsed / 1000).toFixed(2)} | jobs executed: ${ - jobsSent - queue_length - }/${jobsSent} (thr: inst ${instThr} - avg ${avgThr}) | queue: ${queue_length} \r` - ) - ); - }, 10); - - while (queue_length > 0) { - if (queue_length < jobsSent && jobsSent === jobs) { - // reset start time to when the first job was picked up - start = Date.now(); - jobsSent = queue_length; + ); } - await sleep(0.01); } - clearInterval(updateState); - const total_duration_sec = (Date.now() - start) / 1000.0; - await sleep(0.1); console.log(`\njobs: ${jobsSent}`); console.log(`duration: ${total_duration_sec}s`); console.log(`avg. throughput (jobs/time): ${jobsSent / total_duration_sec}`); - console.log( - "queue length:", - ( - await ( - await fetch( - host + "/api/w/" + config.workspace_id + "/jobs/queue/count", - { headers: { ["Authorization"]: "Bearer " + config.token } } - ) - ).json() - ).database_length - ); + console.log("completed jobs", await getCompletedJobsCount()); + console.log("queue length:", await getQueueCount()); + + if (!noVerify && kind !== "noop") { + await verifyOutputs(uuids, config.workspace_id); + } + console.log("done"); return { @@ -229,7 +274,7 @@ if (import.meta.main) { ) .option( "--kind ", - "Specifiy the benchmark kind among: deno, identity, python, go, bash, dedicated, bun, noop, 2steps, onebranch, branchallparrallel", + "Specifiy the benchmark kind among: deno, identity, python, go, bash, dedicated, bun, noop, 2steps", { required: true, } @@ -237,6 +282,7 @@ if (import.meta.main) { .option("-j --jobs ", "Number of jobs to create.", { default: 10000, }) + .option("--no-verify", "Do not verify the output of the jobs.") .action(main) .command( "upgrade", diff --git a/benchmarks/lib.ts b/benchmarks/lib.ts index e40040c351..f9ab85ea83 100644 --- a/benchmarks/lib.ts +++ b/benchmarks/lib.ts @@ -32,16 +32,16 @@ async function waitForDedicatedWorker(workspace: string, path: string) { const query = windmill.JobService.runWaitResultScriptByPath({ workspace, path, - requestBody: { - args: {}, - }, + requestBody: {}, }); - const timeout = new Promise((_, reject) => { - setTimeout(() => { + let timeout; + const timeoutPromise = new Promise((_, reject) => { + timeout = setTimeout(() => { reject("Timeout"); }, 15000); }); - await Promise.race([query, timeout]); + await Promise.race([query, timeoutPromise]); + clearTimeout(timeout); } export async function createBenchScript( @@ -63,6 +63,7 @@ export async function createBenchScript( let scriptContent: string; let language: string; + let schemaProperties = {}; if (scriptPattern === "python") { scriptContent = 'import os\n\ndef main():\n return os.environ.get("WM_JOB_ID")'; @@ -74,9 +75,15 @@ export async function createBenchScript( } else if (scriptPattern === "bash") { scriptContent = "echo $WM_JOB_ID"; language = "bash"; - } else if (scriptPattern === "dedicated" || scriptPattern === "bun") { + } else if (scriptPattern === "bun") { scriptContent = 'export function main(){ return Bun.env["WM_JOB_ID"]; }'; language = "bun"; + } else if (scriptPattern === "dedicated") { + scriptContent = "export function main(uuid){ return uuid; }"; + language = "bun"; + schemaProperties = { + uuid: { default: null, description: "", type: "string" }, + }; } else if (scriptPattern === "deno") { scriptContent = 'export function main(){ return Deno.env.get("WM_JOB_ID"); }'; @@ -96,6 +103,12 @@ export async function createBenchScript( description: "", language: language as api.NewScript.language, dedicated_worker: scriptPattern === "dedicated", + schema: { + $schema: "https://json-schema.org/draft/2020-12/schema", + properties: schemaProperties, + required: [], + type: "object", + }, }, });