windmill/benchmarks/benchmark_noop.ts

/// <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();