diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 96fe402d7f..67540c0720 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -36,6 +36,7 @@ incremental = true [features] enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise"] +benchmark = ["windmill-api/benchmark"] [dependencies] anyhow.workspace = true diff --git a/backend/src/main.rs b/backend/src/main.rs index 13e4e1b850..48bc518cdc 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -12,7 +12,7 @@ use monitor::handle_zombie_jobs_periodically; use sqlx::{Pool, Postgres}; use std::{ net::{IpAddr, Ipv4Addr, SocketAddr}, - sync::Arc, + sync::{Arc}, }; use tokio::{ fs::{metadata, DirBuilder}, diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index d074e681b8..49920821dc 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::sync::{Arc}; use futures::{stream, Stream}; use serde::Deserialize; diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 8756692058..1dd09a0ad5 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -10,6 +10,7 @@ path = "src/lib.rs" [features] enterprise = ["windmill-queue/enterprise", "async-stripe", "windmill-audit/enterprise"] +benchmark = [] [dependencies] windmill-queue.workspace = true diff --git a/backend/windmill-api/src/workers.rs b/backend/windmill-api/src/workers.rs index a6f68f91cb..816eeb1388 100644 --- a/backend/windmill-api/src/workers.rs +++ b/backend/windmill-api/src/workers.rs @@ -20,12 +20,27 @@ use windmill_common::{ utils::{paginate, Pagination}, }; +#[cfg(feature = "benchmark")] +use windmill_queue::IDLE_WORKERS; +#[cfg(feature = "benchmark")] +use std::sync::atomic::Ordering; + + +#[cfg(not(feature = "benchmark"))] pub fn global_service() -> Router { Router::new() .route("/list", get(list_worker_pings)) .route("/custom_tags", get(get_custom_tags)) } +#[cfg(feature = "benchmark")] +pub fn global_service() -> Router { + Router::new() + .route("/toggle", get(toggle)) + .route("/list", get(list_worker_pings)) + .route("/custom_tags", get(get_custom_tags)) +} + lazy_static::lazy_static! { pub static ref CUSTOM_TAGS: Vec = std::env::var("CUSTOM_TAGS") .ok() @@ -43,6 +58,11 @@ struct WorkerPing { jobs_executed: i32, } +#[derive(Serialize, Deserialize)] +struct EnableWorkerQuery { + disable: bool, +} + async fn list_worker_pings( authed: Authed, Extension(user_db): Extension, @@ -64,6 +84,14 @@ async fn list_worker_pings( Ok(Json(rows)) } +#[cfg(feature = "benchmark")] +async fn toggle( + Query(query): Query, +) -> JsonResult { + IDLE_WORKERS.store(query.disable, Ordering::Relaxed); + Ok(Json(IDLE_WORKERS.load(Ordering::Relaxed))) +} + async fn get_custom_tags() -> Json> { Json(CUSTOM_TAGS.clone()) } diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index f9d4a12945..cd861e317e 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -35,4 +35,4 @@ rsmq_async.workspace = true tokio.workspace = true futures-core.workspace = true itertools.workspace = true -async-recursion.workspace = true \ No newline at end of file +async-recursion.workspace = true diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index f0e4dc3d83..1155622fc1 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6,7 +6,7 @@ * LICENSE-AGPL for a copy of the license. */ -use std::{collections::HashMap, vec}; +use std::{collections::HashMap, vec, sync::atomic::AtomicBool}; use anyhow::Context; use async_recursion::async_recursion; @@ -81,6 +81,11 @@ lazy_static::lazy_static! { pub static ref ACCEPTED_TAGS_FILTER: String = format!(" AND ({})", ACCEPTED_TAGS.clone().into_iter().map(|x| format!("(tag = '{x}')")).join(" OR ")); + + // When compiled in 'benchmark' mode, this flags is exposed via the /workers/toggle endpoint + // and make it possible to disable to current active workers (such that they don't pull any) + // jobs from the queue + pub static ref IDLE_WORKERS: AtomicBool = AtomicBool::new(false); } #[cfg(feature = "enterprise")] diff --git a/backend/windmill-queue/src/queue_transaction.rs b/backend/windmill-queue/src/queue_transaction.rs index edf30784e7..01bd021807 100644 --- a/backend/windmill-queue/src/queue_transaction.rs +++ b/backend/windmill-queue/src/queue_transaction.rs @@ -1,4 +1,4 @@ -use std::fmt::Debug; +use std::{fmt::{Debug}}; use futures_core::{future::BoxFuture, stream::BoxStream}; use rsmq_async::{RedisBytes, RsmqConnection}; @@ -9,6 +9,8 @@ pub enum RedisOp { DeleteMessage(String), } +unsafe impl Send for RedisOp {} + impl RedisOp { pub async fn apply(self, rsmq: &mut R) -> Result<(), rsmq_async::RsmqError> { match self { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 24e016f8c7..b7d784d642 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -62,7 +62,7 @@ use rand::Rng; #[cfg(feature = "enterprise")] use crate::global_cache::{copy_cache_to_tmp_cache, cache_global, copy_tmp_cache_to_cache, copy_denogo_cache_from_bucket_as_tar, copy_all_piptars_from_bucket}; -use windmill_queue::{add_completed_job, add_completed_job_error}; +use windmill_queue::{add_completed_job, add_completed_job_error,IDLE_WORKERS}; use crate::{ worker_flow::{ @@ -70,9 +70,6 @@ use crate::{ }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, }; - - - pub async fn create_token_for_owner_in_bg(db: &Pool, job: &QueuedJob) -> Arc> { let rw_lock = Arc::new(RwLock::new(String::new())); // skipping test runs @@ -259,7 +256,6 @@ lazy_static::lazy_static! { .and_then(|x| x.parse::().ok()); pub static ref CAN_PULL: Arc> = Arc::new(RwLock::new(())); - } //only matter if CLOUD_HOSTED @@ -563,6 +559,11 @@ pub async fn run_worker { diff --git a/benchmarks/README.md b/benchmarks/README.md index 16470112f2..c358471626 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -10,20 +10,23 @@ Install the `wmill` CLI tool using Update to the latest version using `wmillbench upgrade`. +To build a local version, you can just run: +``` +deno install -A main.ts +``` + ## Quickstart -Have your instance expose prometheus metrics (METRICS_ADDR=1). +Have your instance expose prometheus metrics (METRICS_ADDR=true). Then ``` -wmillbench -s 1 -e admin@windmill.dev -p changeme --host YOUR_HOST +wmillbench -e admin@windmill.dev -p changeme --host YOUR_HOST ``` ## Usage -``` - Usage: wmillbench Description: @@ -32,25 +35,23 @@ Run Benchmark to measure throughput of windmill. Options: --h, --help - Show this help. -V, --version - Show the version number for this -program. --host - The windmill host to benchmark. (Default: -"http://127.0.0.1:8000/") --workers - The number of workers to run at -once. (Default: 1) -s, --seconds - How long to run the benchmark for -(in seconds). (Default: 30) -e, --email - The email to use to login. -p, ---password - The password to use to login. -t, --token - The -token to use when talking to the API server. Preferred over manual login. -w, ---workspace - The workspace to spawn scripts from. (Default: -"starter") -m, --metrics - The url to scrape metrics from. (Default: -"http://localhost:8001/metrics") --export-json - If set, exports -will be into a JSON file. --export-csv - If set, exports will be -into a csv file. --export-histograms [histograms...] - Mark metrics (without -label) that are reported as histograms to export. --export-simple [simple...] - -Mark metrics (without label) that are reported as simple values. ---maximum-throughput - Maximum number of jobs/flows to -start in one second. (Default: Infinity) --use-flows - Run flows instead of -jobs. --histogram-buckets [buckets...] - Define what buckets to collect from -histograms. (Default: [ "+Inf", "10", "5", "2.5", "2.5", "1", "0.5", "0.25", -"0.1", "0.05", "0.025", "0.01", "0.005" ]) +-h, --help - Show this help. +-V, --version - Show the version number for this program. +--host - The windmill host to benchmark. (Default: "http://127.0.0.1:8000/") +--workers - The number of workers to run at once. (Default: 1) +-s, --seconds - How long to run the benchmark for (in seconds). (Default: 30) +-e, --email - The email to use to login. +-p, --password - The password to use to login. +-t, --token - The token to use when talking to the API server. Preferred over manual login. +-w, --workspace - The workspace to spawn scripts from. (Default: "starter") +-m, --metrics - The url to scrape metrics from. (Default: "http://localhost:8001/metrics") +--export-json - If set, exports will be into a JSON file. +--export-csv - If set, exports will be into a csv file. +--export-histograms [histograms...] - Mark metrics (without label) that are reported as histograms to export. +--export-simple [simple...] - Mark metrics (without label) that are reported as simple values. +--maximum-throughput - Maximum number of jobs/flows to start in one second. (Default: Infinity) +--use-flows - Run flows instead of jobs. +--histogram-buckets [buckets...] - Define what buckets to collect from histograms. (Default: [ "+Inf", "10", "5", "2.5", "2.5", "1", "0.5", "0.25", "0.1", "0.05", "0.025", "0.01", "0.005" ]) Environment variables: @@ -58,7 +59,7 @@ WM_TOKEN - The token to use when talking to the API server. Preferred over manual login. WM_WORKSPACE - The workspace to spawn scripts from. -``` + This will run a simple benchmark against localhost (the default admin email + password are set above), all execution is done in the "bench" workspace (as set @@ -67,6 +68,18 @@ via `--workspace`). Metrics are exported to JSON will only include mean & stdev, histograms get one entry for each bucket. CSV will include a full list of all values scraped. +## NOOP jobs benchmark + +A specific benchmark creating a set of NOOP jobs all at once in windmill is also available. +in `benchmarks_noop.ts` + +You can build it locally with: +``` +deno install -A benchmarks_noop.ts +``` +and then +``` +benchmarks_noop -e admin@windmill.dev -p changeme --host YOUR_HOST ``` -``` +By default it creates 10000 jobs in Windmill in a single batch, but this is parametrizable. \ No newline at end of file diff --git a/benchmarks/benchmark_noop.ts b/benchmarks/benchmark_noop.ts new file mode 100644 index 0000000000..2a7334f476 --- /dev/null +++ b/benchmarks/benchmark_noop.ts @@ -0,0 +1,218 @@ +/// +/// + +import { Command } from "https://deno.land/x/cliffy@v0.25.7/command/mod.ts"; +import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts"; +import * as windmill from "https://deno.land/x/windmill@v1.38.5/mod.ts"; +import { UpgradeCommand } from "https://deno.land/x/cliffy@v0.25.7/command/upgrade/upgrade_command.ts"; +import { DenoLandProvider } from "https://deno.land/x/cliffy@v0.25.7/command/upgrade/mod.ts"; +export { + DenoLandProvider, + UpgradeCommand, +} from "https://deno.land/x/cliffy@v0.25.7/command/upgrade/mod.ts"; + +async function login(email: string, password: string): Promise { + return await windmill.UserService.login({ + requestBody: { + email: email, + password: password, + }, + }); +} + +export const VERSION = "v1.125.1"; + +await new Command() + .name("wmillbench") + .description("Run Benchmark to measure throughput of windmill.") + .version(VERSION) + .option("--host ", "The windmill host to benchmark.", { + default: "http://127.0.0.1:8000", + }) + .option("-e --email ", "The email to use to login.") + .option("-p --password ", "The password to use to login.") + .env( + "WM_TOKEN=", + "The token to use when talking to the API server. Preferred over manual login." + ) + .option( + "-t --token ", + "The token to use when talking to the API server. Preferred over manual login." + ) + .env( + "WM_WORKSPACE=", + "The workspace to spawn scripts from." + ) + .option( + "-w --workspace ", + "The workspace to spawn scripts from.", + { default: "admins" } + ) + .option( + "-j --jobs ", + "Number of NOOP jobs to create.", + { default: 10000 } + ) + .option( + "-b --batches ", + "Number of batches to create all the jobs.", + { default: 1 } + ) + .action( + async ({ + host, + email, + password, + token, + workspace, + jobs, + batches, + }) => { + windmill.setClient("", host); + + console.log( + "Started benchmark with NOOP jobs with options", + JSON.stringify( + { + host, + email, + workspace, + }, + null, + 4 + ) + ); + + const config = { + token: "", + server: host, + workspace_id: workspace, + }; + + let final_token: string; + if (!token) { + if (email && password) { + final_token = await login(email, password); + } else { + console.error("Token or email with password are required."); + return; + } + } else { + final_token = token; + } + + config.token = final_token; + windmill.setClient(final_token, host); + const enc = (s: string) => new TextEncoder().encode(s); + + console.log("Disabling workers before loading jobs") + const disable_workers = await fetch( + config.server + "/api/workers/toggle?disable=true", + { + method: "GET", + headers: { ["Authorization"]: "Bearer " + config.token }, + } + ) + if (!disable_workers.ok) { + console.error("Unable to disable workers. Is the Windmill server running in benchmark mode?") + } + + const jobsSent = jobs; + const batch_num = batches; + console.log(`Bulk creating ${jobsSent} jobs in ${batch_num} batches`) + + const start_create = Date.now() + const all_create_operations = [] + for (let i = 0; i < batch_num; i++) { + all_create_operations.push(fetch( + config.server + + "/api/w/" + + config.workspace_id + + `/jobs/add_noop_jobs/${jobsSent / batch_num}`, + { + method: "POST", + headers: { ["Authorization"]: "Bearer " + config.token }, + } + )); + } + await Promise.all(all_create_operations) + + const end_create = Date.now() + const create_duration = end_create - start_create + console.log(`Jobs successfully added to the queue in ${create_duration}s. Windmill will start pulling them\n`) + const start = Date.now() + + let queue_length = jobsSent + const updateState = setInterval(async () => { + const elapsed = start ? Math.ceil((Date.now() - start) / 1000) : 0; + queue_length = ( + await ( + await fetch( + host + "/api/w/" + config.workspace_id + "/jobs/queue/count", + { headers: { ["Authorization"]: "Bearer " + config.token } } + ) + ).json() + ).database_length; + await Deno.stdout.write( + enc( + `elapsed: ${elapsed} | jobs executed: ${JSON.stringify( + jobsSent - queue_length + )}/${jobsSent} (thr: ${((jobsSent - queue_length) / elapsed).toFixed( + 2 + )}) | queue: ${queue_length} \r` + ) + ); + }, 100); + + console.log("Enabling workers to start processing jobs") + const enable_workers = await fetch( + config.server + "/api/workers/toggle?disable=false", + { + method: "GET", + headers: { ["Authorization"]: "Bearer " + config.token }, + } + ) + if (!enable_workers.ok) { + console.error("Unable to disable workers. Is the Windmill server running in benchmark mode?") + } + + while (queue_length > 0) { + await sleep(0.1); + } + + clearInterval(updateState); + + const total_duration_sec = (Date.now() - start) / 1000; + console.log(`jobs: ${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("done"); + } + ) + .command( + "upgrade", + new UpgradeCommand({ + main: "main.ts", + args: [ + "--allow-net", + "--allow-read", + "--allow-write", + "--allow-env", + "--unstable", + ], + provider: new DenoLandProvider({ name: "wmillbench" }), + }) + ) + .parse(); diff --git a/benchmarks/worker.ts b/benchmarks/worker.ts index ea96b1754a..f800d30d3d 100644 --- a/benchmarks/worker.ts +++ b/benchmarks/worker.ts @@ -46,284 +46,261 @@ let total_spawned = 0; let start_time: number; let complete_timeout = Infinity; -if (config.scriptPattern == "noop") { - const n = 10000; - const res = await fetch( - config.server + - "/api/w/" + - config.workspace_id + - `/jobs/add_noop_jobs/${n}`, - { - method: "POST", - headers: { ["Authorization"]: "Bearer " + config.token }, - } - ); - const uuids = await res.json(); - outstanding.push(...uuids); - total_spawned += n; - self.postMessage({ type: "jobs_sent", jobs_sent: total_spawned }); +start_time = Date.now(); + +self.onmessage = (evt) => { cont = false; - self.onmessage = (evt) => { - cont = false; - complete_timeout = evt.data; - start_time = Date.now(); - }; -} else { - start_time = Date.now(); + complete_timeout = evt.data; +}; - self.onmessage = (evt) => { - cont = false; - complete_timeout = evt.data; - }; +const updateStatusInterval = setInterval(() => { + self.postMessage({ type: "jobs_sent", jobs_sent: total_spawned }); +}, 100); - const updateStatusInterval = setInterval(() => { - self.postMessage({ type: "jobs_sent", jobs_sent: total_spawned }); - }, 100); +while (cont) { + const queue_length = ( + await ( + await fetch( + config.server + "/api/w/" + config.workspace_id + "/jobs/queue/count", + { headers: { ["Authorization"]: "Bearer " + config.token } } + ) + ).json() + ).database_length; + if (queue_length > 2500) { + console.log( + `queue length: ${queue_length} > 2500. waiting... ` + ); + await sleep(0.5); + continue; + } - while (cont) { - const queue_length = ( - await ( - await fetch( - config.server + "/api/w/" + config.workspace_id + "/jobs/queue/count", - { headers: { ["Authorization"]: "Bearer " + config.token } } - ) - ).json() - ).database_length; - if (queue_length > 2500) { - console.log( - `queue length: ${queue_length} > 2500. waiting... ` - ); - await sleep(0.5); - continue; - } - - if ( - (total_spawned * 1000) / (Date.now() - start_time) > - config.per_worker_throughput - ) { - console.log("at maximum throughput. waiting..."); - await sleep(0.1); - continue; - } - total_spawned++; - if (total_spawned > config.max_per_worker) { - break; - } - let uuid: string; - if (config.custom) { - await evaluate(config.custom); - continue; - } else if (config.useFlows) { - let payload: api.FlowPreview; - if (config.flowPattern == "branchone") { - payload = { - path: "branchone", - args: {}, - value: { - modules: [ - { - id: "a", - value: { - input_transforms: {}, - language: api.RawScript.language.DENO, - type: "rawscript", - content: - 'export function main(){ return Deno.env.get("WM_FLOW_JOB_ID"); }', - }, + if ( + (total_spawned * 1000) / (Date.now() - start_time) > + config.per_worker_throughput + ) { + console.log("at maximum throughput. waiting..."); + await sleep(0.1); + continue; + } + total_spawned++; + if (total_spawned > config.max_per_worker) { + break; + } + let uuid: string; + if (config.custom) { + await evaluate(config.custom); + continue; + } else if (config.useFlows) { + let payload: api.FlowPreview; + if (config.flowPattern == "branchone") { + payload = { + path: "branchone", + args: {}, + value: { + modules: [ + { + id: "a", + value: { + input_transforms: {}, + language: api.RawScript.language.DENO, + type: "rawscript", + content: + 'export function main(){ return Deno.env.get("WM_FLOW_JOB_ID"); }', }, - { - id: "b", - value: { - type: "branchone", - branches: [], - default: [ - { - id: "c", - value: { - input_transforms: { - x: { - type: "javascript", - expr: "results.a", - }, + }, + { + id: "b", + value: { + type: "branchone", + branches: [], + default: [ + { + id: "c", + value: { + input_transforms: { + x: { + type: "javascript", + expr: "results.a", }, - language: api.RawScript.language.DENO, - type: "rawscript", - content: "export function main(x: string){ return x; }", }, + language: api.RawScript.language.DENO, + type: "rawscript", + content: "export function main(x: string){ return x; }", }, - ], - }, + }, + ], }, - ], - }, - }; - } else if (config.flowPattern == "branchallparrallel") { - payload = { - path: "branchall", - args: {}, - value: { - modules: [ - { - id: "a", - value: { - input_transforms: {}, - language: api.RawScript.language.DENO, - type: "rawscript", - content: - 'export function main(){ return Deno.env.get("WM_FLOW_JOB_ID"); }', - }, + }, + ], + }, + }; + } else if (config.flowPattern == "branchallparrallel") { + payload = { + path: "branchall", + args: {}, + value: { + modules: [ + { + id: "a", + value: { + input_transforms: {}, + language: api.RawScript.language.DENO, + type: "rawscript", + content: + 'export function main(){ return Deno.env.get("WM_FLOW_JOB_ID"); }', }, - { - id: "b", - value: { - type: "branchall", - parallel: true, - branches: [ - { - modules: [ - { - id: "c", - value: { - input_transforms: { - x: { - type: "javascript", - expr: "results.a", - }, + }, + { + id: "b", + value: { + type: "branchall", + parallel: true, + branches: [ + { + modules: [ + { + id: "c", + value: { + input_transforms: { + x: { + type: "javascript", + expr: "results.a", }, - language: api.RawScript.language.DENO, - type: "rawscript", - content: - "export function main(x: string){ return x; }", }, + language: api.RawScript.language.DENO, + type: "rawscript", + content: + "export function main(x: string){ return x; }", }, - ], - }, - { - modules: [ - { - id: "d", - value: { - input_transforms: { - x: { - type: "javascript", - expr: "results.a", - }, + }, + ], + }, + { + modules: [ + { + id: "d", + value: { + input_transforms: { + x: { + type: "javascript", + expr: "results.a", }, - language: api.RawScript.language.DENO, - type: "rawscript", - content: - "export function main(x: string){ return x; }", }, + language: api.RawScript.language.DENO, + type: "rawscript", + content: + "export function main(x: string){ return x; }", }, - ], - }, - ], - }, + }, + ], + }, + ], }, - ], - }, - }; - } else { - payload = { - path: "2steps", - args: {}, - value: { - modules: [ - { - id: "a", - value: { - input_transforms: {}, - language: api.RawScript.language.DENO, - type: "rawscript", - content: - 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', - }, + }, + ], + }, + }; + } else { + payload = { + path: "2steps", + args: {}, + value: { + modules: [ + { + id: "a", + value: { + input_transforms: {}, + language: api.RawScript.language.DENO, + type: "rawscript", + content: + 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', }, - { - id: "b", - value: { - input_transforms: {}, - language: api.RawScript.language.DENO, - type: "rawscript", - content: - 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', - }, + }, + { + id: "b", + value: { + input_transforms: {}, + language: api.RawScript.language.DENO, + type: "rawscript", + content: + 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', }, - ], - }, - }; - } - uuid = await windmill.JobService.runFlowPreview({ + }, + ], + }, + }; + } + uuid = await windmill.JobService.runFlowPreview({ + workspace: config.workspace_id, + requestBody: payload, + }); + } else { + let payload: api.Preview; + if (config.scriptPattern == "httpversion") { + payload = { + path: "httpversion", + kind: "http", + args: { + url: "http://localhost:8000/api/version", + }, + }; + } else if (config.scriptPattern == "httpslow") { + payload = { + path: "httpversion", + kind: "http", + args: { + url: "https://hub.dummyapis.com/delay?seconds=10", + }, + }; + } else if (config.scriptPattern == "noop") { + payload = { + path: "noop", + kind: "noop", + args: {}, + }; + } else if (config.scriptPattern == "identity") { + payload = { + path: "identity", + kind: "identity", + args: { + identity: "itsme", + }, + }; + } else if (config.scriptPattern == "postgresql") { + payload = { + path: "postgresql", + language: "postgresql", + args: { + query: "SELECT email FROM usr", + database_url: + "postgres://postgres:changeme@localhost:5432/windmill", + }, + }; + } else { + payload = { + path: "denosimple", + language: api.Preview.language.DENO, + content: + 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', + args: {}, + }; + } + try { + uuid = await windmill.JobService.runScriptPreview({ workspace: config.workspace_id, requestBody: payload, }); - } else { - let payload: api.Preview; - if (config.scriptPattern == "httpversion") { - payload = { - path: "httpversion", - kind: "http", - args: { - url: "http://localhost:8000/api/version", - }, - }; - } else if (config.scriptPattern == "httpslow") { - payload = { - path: "httpversion", - kind: "http", - args: { - url: "https://hub.dummyapis.com/delay?seconds=10", - }, - }; - } else if (config.scriptPattern == "noop") { - payload = { - path: "noop", - kind: "noop", - args: {}, - }; - } else if (config.scriptPattern == "identity") { - payload = { - path: "identity", - kind: "identity", - args: { - identity: "itsme", - }, - }; - } else if (config.scriptPattern == "postgresql") { - payload = { - path: "postgresql", - language: "postgresql", - args: { - query: "SELECT email FROM usr", - database_url: - "postgres://postgres:changeme@localhost:5432/windmill", - }, - }; - } else { - payload = { - path: "denosimple", - language: api.Preview.language.DENO, - content: - 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', - args: {}, - }; - } - try { - uuid = await windmill.JobService.runScriptPreview({ - workspace: config.workspace_id, - requestBody: payload, - }); - } catch (e) { - console.error("error running script: " + e.body); - Deno.exit(1); - } + } catch (e) { + console.error("error running script: " + e.body); + Deno.exit(1); } - if (!config.continous) outstanding.push(uuid); } - - clearInterval(updateStatusInterval); + if (!config.continous) outstanding.push(uuid); } +clearInterval(updateStatusInterval); + + const end_time = Date.now() + complete_timeout; let incorrect_results = 0; @@ -340,59 +317,51 @@ async function getQueueCount() { ).database_length; } -if (config.scriptPattern == "noop") { - let queue_length = await getQueueCount(); - while (queue_length > 0 && Date.now() < end_time) { - await Deno.stdout.write(enc(`queue length: ${queue_length}\r`)); - queue_length = await getQueueCount(); - } -} else { - while (outstanding.length > 0 && Date.now() < end_time) { - await Deno.stdout.write( - enc("\rwaiting for jobs to complete: " + outstanding.length + "\n") - ); - const uuid = outstanding.shift()!; +while (outstanding.length > 0 && Date.now() < end_time) { + await Deno.stdout.write( + enc("\rwaiting for jobs to complete: " + outstanding.length + "\n") + ); + const uuid = outstanding.shift()!; - let r: Job; + let r: Job; + try { + r = await windmill.JobService.getJob({ + workspace: config.workspace_id, + id: uuid, + }); + } catch (e) { + console.log("job not found: " + uuid + " " + e.message); + continue; + } + if (r.type == "QueuedJob") { + outstanding.push(uuid); + await Deno.stdout.write( + enc(`uuid: ${uuid}, queue length: ${await getQueueCount()}\r`) + ); + } else { + r = r as api.CompletedJob; try { - r = await windmill.JobService.getJob({ - workspace: config.workspace_id, - id: uuid, - }); - } catch (e) { - console.log("job not found: " + uuid + " " + e.message); - continue; - } - if (r.type == "QueuedJob") { - outstanding.push(uuid); - await Deno.stdout.write( - enc(`uuid: ${uuid}, queue length: ${await getQueueCount()}\r`) - ); - } else { - r = r as api.CompletedJob; - try { - if ( - !["httpversion", "identity", "httpslow", "noop"].includes( - config.scriptPattern - ) && - r.result != uuid - ) { - console.log( - "job did not return correct UUID: " + - r.result + - " != " + - uuid + - "job: \n" + - JSON.stringify(r, null, 2) - ); - incorrect_results++; - } else { - // console.log(r.result); - } - } catch (e) { - console.log("error during wait: ", e); - outstanding.push(uuid); + if ( + !["httpversion", "identity", "httpslow", "noop"].includes( + config.scriptPattern + ) && + r.result != uuid + ) { + console.log( + "job did not return correct UUID: " + + r.result + + " != " + + uuid + + "job: \n" + + JSON.stringify(r, null, 2) + ); + incorrect_results++; + } else { + // console.log(r.result); } + } catch (e) { + console.log("error during wait: ", e); + outstanding.push(uuid); } } }