Specific NOOP jobs benchmark (#1810)

* Refactor and fix NOOP benchmark * Add endpoint to toggle workers on/off when compiled in benchmark mode * Improve noop benchmark
2026-08-18 16:02:10 +00:00 · 2023-07-10 14:16:34 +02:00
parent 69f4a0a771
commit 96df8900cb
12 changed files with 575 additions and 337 deletions
@@ -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
@@ -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},
@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::sync::{Arc};

 use futures::{stream, Stream};
 use serde::Deserialize;
@@ -10,6 +10,7 @@ path = "src/lib.rs"

 [features]
 enterprise = ["windmill-queue/enterprise", "async-stripe", "windmill-audit/enterprise"]
+benchmark = []

 [dependencies]
 windmill-queue.workspace = true
@@ -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<String> = 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<UserDB>,
@@ -64,6 +84,14 @@ async fn list_worker_pings(
    Ok(Json(rows))
 }

+#[cfg(feature = "benchmark")]
+async fn toggle(
+    Query(query): Query<EnableWorkerQuery>,
+) -> JsonResult<bool> {
+    IDLE_WORKERS.store(query.disable, Ordering::Relaxed);
+    Ok(Json(IDLE_WORKERS.load(Ordering::Relaxed)))
+}
+
 async fn get_custom_tags() -> Json<Vec<String>> {
    Json(CUSTOM_TAGS.clone())
 }
@@ -35,4 +35,4 @@ rsmq_async.workspace = true
 tokio.workspace = true
 futures-core.workspace = true
 itertools.workspace = true
-async-recursion.workspace = true
+async-recursion.workspace = true
@@ -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")]
@@ -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<R: RsmqConnection>(self, rsmq: &mut R) -> Result<(), rsmq_async::RsmqError> {
        match self {
@@ -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<Postgres>, job: &QueuedJob) -> Arc<RwLock<String>> {
    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::<u64>().ok());

    pub static ref CAN_PULL: Arc<RwLock<()>> = Arc::new(RwLock::new(()));
-
 }

 //only matter if CLOUD_HOSTED
@@ -563,6 +559,11 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
            } else {
                // println!("2: {:?}",  instant.elapsed());
                async {
+                    if IDLE_WORKERS.load(Ordering::Relaxed) {
+                        // TODO: Need to sleep for a little time before re-checking, maybe?
+                        // tracing::warn!("Worker is marked as idle. Not pulling any job for now");
+                        return (false, Ok(None));
+                    }
                    tokio::select! {
                        biased;
                        _ = rx.recv() => {
@@ -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 <url> - The windmill host to benchmark. (Default:
-"http://127.0.0.1:8000/") --workers <workers> - The number of workers to run at
-once. (Default: 1) -s, --seconds <seconds> - How long to run the benchmark for
-(in seconds). (Default: 30) -e, --email <email> - The email to use to login. -p,
--password <password> - The password to use to login. -t, --token <token> - The
-token to use when talking to the API server. Preferred over manual login. -w,
--workspace <workspace> - The workspace to spawn scripts from. (Default:
-"starter") -m, --metrics <metrics> - The url to scrape metrics from. (Default:
-"http://localhost:8001/metrics") --export-json <export_json> - If set, exports
-will be into a JSON file. --export-csv <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_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 <url> - The windmill host to benchmark. (Default: "http://127.0.0.1:8000/") 
+--workers <workers> - The number of workers to run at once. (Default: 1) 
+-s, --seconds <seconds> - How long to run the benchmark for (in seconds). (Default: 30) 
+-e, --email <email> - The email to use to login. 
+-p, --password <password> - The password to use to login. 
+-t, --token <token> - The token to use when talking to the API server. Preferred over manual login. 
+-w, --workspace <workspace> - The workspace to spawn scripts from. (Default: "starter") 
+-m, --metrics <metrics> - The url to scrape metrics from. (Default: "http://localhost:8001/metrics") 
+--export-json <export_json> - If set, exports will be into a JSON file. 
+--export-csv <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_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 <token> - The token to use when talking to the API server. Preferred
 over manual login. WM_WORKSPACE <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.
@@ -0,0 +1,218 @@
+/// <reference no-default-lib="true" />
+/// <reference lib="deno.window" />
+
+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<string> {
+  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 <url:string>", "The windmill host to benchmark.", {
+    default: "http://127.0.0.1:8000",
+  })
+  .option("-e --email <email:string>", "The email to use to login.")
+  .option("-p --password <password:string>", "The password to use to login.")
+  .env(
+    "WM_TOKEN=<token:string>",
+    "The token to use when talking to the API server. Preferred over manual login."
+  )
+  .option(
+    "-t --token <token:string>",
+    "The token to use when talking to the API server. Preferred over manual login."
+  )
+  .env(
+    "WM_WORKSPACE=<workspace:string>",
+    "The workspace to spawn scripts from."
+  )
+  .option(
+    "-w --workspace <workspace:string>",
+    "The workspace to spawn scripts from.",
+    { default: "admins" }
+  )
+  .option(
+    "-j --jobs <jobs:number>",
+    "Number of NOOP jobs to create.",
+    { default: 10000 }
+  )
+  .option(
+    "-b --batches <batches:number>",
+    "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();
@@ -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);
    }
  }
 }