feat(benchmark): Initial Benchmarking Tool (#731)

* Deno setup * Very basic benchmarking * Collect metrics into CSV * Rip out CSV functionality * export to influxdb * remove old files * Move into subfolder in preparation for merge * Move settings to correct folder * Cleanup & Fix typing * Remove unused code * Add JSON export of specific metrics * Wait for all jobs to complete & check resutls * Delete output.json * Apply some review comments * Simplify some truthy expressions * Remove InfluxDB * Rewrite Stats calculation * Add CSV output * Fix stdev calculations * Add README * Add maximum-throughput option * Add flow option * Remove testing changes * Revert auto-format * Track zombie workers
2022-10-18 14:10:04 +02:00
parent 584f168f8f
commit 43e6f5c525
5 changed files with 565 additions and 0 deletions
--- a/benchmarks/.vscode/settings.json
+++ b/benchmarks/.vscode/settings.json
@@ -0,0 +1,5 @@
+{
+  "deno.enable": true,
+  "deno.unstable": true,
+  "editor.tabSize": 2
+}
--- a/benchmarks/README.md
+++ b/benchmarks/README.md
@@ -0,0 +1,47 @@
+# Benchmarks
+
+This folder includes a small deno/ts utility to benchmark execution of jobs & flows.
+
+```
+Usage:   windmill-bench
+
+Description:
+
+  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/")
+  --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:
+
+  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 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.
--- a/benchmarks/main.ts
+++ b/benchmarks/main.ts
@@ -0,0 +1,293 @@
+/// <reference no-default-lib="true" />
+/// <reference lib="deno.window" />
+
+import { Command } from "https://deno.land/x/cliffy@v0.25.2/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.37.0/mod.ts";
+import * as api from "https://deno.land/x/windmill@v1.37.0/windmill-api/index.ts";
+
+async function login(
+  config: api.Configuration,
+  email: string,
+  password: string
+): Promise<string> {
+  return await new windmill.UserApi(config).login({
+    email: email,
+    password: password,
+  });
+}
+
+await new Command()
+  .name("windmill-bench")
+  .description("Run Benchmark to measure throughput of windmill.")
+  .version("v0.0.0")
+  .option("--host <url:string>", "The windmill host to benchmark.", {
+    default: "http://127.0.0.1/",
+  })
+  .option(
+    "--workers <workers:number>",
+    "The number of workers to run at once.",
+    {
+      default: 1,
+    }
+  )
+  .option(
+    "-s --seconds <seconds:number>",
+    "How long to run the benchmark for (in seconds).",
+    {
+      default: 30,
+    }
+  )
+  .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: "starter" }
+  )
+  .option("-m --metrics <metrics:string>", "The url to scrape metrics from.", {
+    default: "http://localhost:8001/metrics",
+  })
+  .option(
+    "--export-json <export_json:string>",
+    "If set, exports will be into a JSON file."
+  )
+  .option(
+    "--export-csv <export_csv:string>",
+    "If set, exports will be into a csv file."
+  )
+  .option(
+    "--export-histograms [histograms...:string]",
+    "Mark metrics (without label) that are reported as histograms to export."
+  )
+  .option(
+    "--export-simple [simple...:string]",
+    "Mark metrics (without label) that are reported as simple values."
+  )
+  .option(
+    "--maximum-throughput <maximum_throughput:number>",
+    "Maximum number of jobs/flows to start in one second.",
+    {
+      default: Infinity,
+    }
+  )
+  .option("--use-flows", "Run flows instead of jobs.")
+  .option(
+    "--zombie-timeout",
+    "The maximum time in ms to wait for jobs to complete.",
+    {
+      default: 90000,
+    }
+  )
+  .option(
+    "--histogram-buckets [buckets...:string]",
+    "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",
+      ],
+    }
+  )
+  .action(
+    async ({
+      host,
+      workers: num_workers,
+      seconds,
+      email,
+      password,
+      token,
+      workspace,
+      metrics,
+      exportJson,
+      exportCsv,
+      exportHistograms,
+      exportSimple,
+      histogramBuckets,
+      maximumThroughput,
+      useFlows,
+      zombieTimeout,
+    }) => {
+      const metrics_worker = new Worker(
+        new URL("./scraper.ts", import.meta.url).href,
+        {
+          type: "module",
+        }
+      );
+
+      if (!Array.isArray(histogramBuckets)) {
+        histogramBuckets = [];
+      }
+
+      if (!Array.isArray(exportHistograms)) {
+        exportHistograms = [];
+      }
+
+      if (!Array.isArray(exportSimple)) {
+        exportSimple = [];
+      }
+
+      metrics_worker.postMessage({
+        exportHistograms,
+        histogramBuckets,
+        exportSimple,
+        host: metrics,
+      });
+
+      console.log("collecting samples...");
+      host = host.endsWith("/") ? host.substring(0, host.length - 1) : host;
+      host = `${host}/api`;
+
+      let config = {
+        ...api.createConfiguration({
+          baseServer: new api.ServerConfiguration(host, {}),
+        }),
+        workspace_id: workspace,
+      };
+
+      let final_token: string;
+      if (!token) {
+        if (email && password) {
+          final_token = await login(config, email, password);
+        } else {
+          console.error("Token or email with password are required.");
+          return;
+        }
+      } else {
+        final_token = token;
+      }
+
+      config = {
+        ...api.createConfiguration({
+          baseServer: config.baseServer,
+          authMethods: {
+            bearerAuth: {
+              tokenProvider: {
+                getToken() {
+                  return final_token;
+                },
+              },
+            },
+          },
+        }),
+        workspace_id: config.workspace_id,
+      };
+
+      const per_worker_throughput = maximumThroughput / num_workers;
+      const shared_config = {
+        server: host,
+        token: final_token,
+        workspace_id: config.workspace_id,
+        per_worker_throughput,
+        useFlows,
+      };
+
+      let workers: Worker[] = new Array(num_workers);
+      for (let i = 0; i < num_workers; i++) {
+        workers[i] = new Worker(new URL("./worker.ts", import.meta.url).href, {
+          type: "module",
+        });
+      }
+
+      workers.forEach((worker, i) => {
+        worker.postMessage({ ...shared_config, i });
+      });
+
+      await sleep(seconds);
+
+      let zombie_jobs = 0;
+      workers.forEach((worker) => {
+        const l = (evt: MessageEvent<any>) => {
+          zombie_jobs += evt.data;
+          worker.removeEventListener("message", l);
+          workers = workers.filter((w) => w != worker);
+          worker.terminate();
+        };
+        worker.addEventListener("message", l);
+        worker.postMessage(
+          Number.isSafeInteger(zombieTimeout) ? zombieTimeout : 90000
+        );
+      });
+
+      console.log("waiting for shutdown");
+      while (workers.length > 0) {
+        await sleep(0.1);
+      }
+
+      console.log("zombie jobs: ", zombie_jobs);
+
+      metrics_worker.postMessage("stop");
+      console.log("waiting for metrics");
+      const { columns, transfer_values } = await new Promise<{
+        columns: string[];
+        transfer_values: ArrayBufferLike[];
+      }>((resolve, _reject) => {
+        metrics_worker.onmessage = (e) => {
+          resolve(e.data);
+          metrics_worker.terminate();
+        };
+      });
+      const values = transfer_values.map((x) => new Float32Array(x));
+
+      if (exportJson) {
+        console.log("exporting mean & stdev to json");
+        const obj: any = {};
+        for (let i = 0; i < columns.length; i++) {
+          const name = columns[i]!;
+          const value = values[i]!;
+          const mean = value.reduce((acc, e) => acc + e, 0) / values.length;
+          const stdev = Math.sqrt(
+            value.reduce((acc, e) => acc + (e - mean) ** 2) / values.length
+          );
+          obj[name] = { mean, stdev };
+        }
+
+        await Deno.writeTextFile(exportJson, JSON.stringify(obj));
+      }
+
+      if (exportCsv) {
+        const f = await Deno.open(exportCsv, {
+          write: true,
+          create: true,
+          truncate: true,
+        });
+        const encoder = new TextEncoder();
+        const newline = new Uint8Array(1);
+        newline[0] = 0x0a;
+        await f.write(encoder.encode(columns.join(",")));
+        await f.write(newline);
+
+        for (let i = 0; i < values.length; i++) {
+          await f.write(encoder.encode(values[i].join(",")));
+          await f.write(newline);
+        }
+
+        f.close();
+      }
+      console.log("done");
+    }
+  )
+  .parse();
--- a/benchmarks/scraper.ts
+++ b/benchmarks/scraper.ts
@@ -0,0 +1,88 @@
+/// <reference no-default-lib="true" />
+/// <reference lib="deno.worker" />
+
+import { sleep } from "https://deno.land/x/sleep@v1.2.1/sleep.ts";
+import parsePrometheusTextFormat from "npm:parse-prometheus-text-format";
+
+const promise = new Promise<{
+  host: string;
+  histogramBuckets: string[];
+  exportHistograms: string[];
+  exportSimple: string[];
+}>((resolve, _reject) => {
+  self.onmessage = (evt) => {
+    resolve(evt.data);
+    self.onmessage = null;
+  };
+});
+const { host, histogramBuckets, exportHistograms, exportSimple } =
+  await promise;
+
+const columns = exportHistograms
+  .flatMap((x) => (histogramBuckets as string[]).map((b) => x + "_" + b))
+  .concat(exportSimple.map((x) => x + "_value"));
+
+const values: Float32Array[] = [];
+
+let cont = true;
+self.onmessage = (_evt) => {
+  cont = false;
+};
+
+while (cont) {
+  const start = Date.now();
+  const response = await fetch(host);
+  const text = await response.text();
+  // TODO: parsePrometheusTextFormat seems incomplete. Consider rewriting for deno with actual completeness.
+  // Specifically histogram labels seem to not be reported.
+  const prometheusValues: [
+    {
+      name: string;
+      help: string;
+    } & (
+      | {
+          type: "COUNTER" | "GAUGE";
+          metrics: [{ value: string; labels: Record<string, string> }];
+        }
+      | {
+          type: "HISTOGRAM";
+          metrics: [{ buckets: Record<string, number> }];
+        }
+    )
+  ] = parsePrometheusTextFormat(text);
+  const new_values: Float32Array = new Float32Array(columns.length);
+
+  prometheusValues.forEach((x) => {
+    if (
+      x.type == "HISTOGRAM" &&
+      exportHistograms.findIndex((e) => e == x.name) !== -1
+    ) {
+      x.metrics.forEach((m) => {
+        histogramBuckets.forEach((e) => {
+          new_values[columns.indexOf(x.name + "_" + e)] = m.buckets[e];
+        });
+      });
+    }
+    if (
+      (x.type == "GAUGE" || x.type == "COUNTER") &&
+      exportSimple.findIndex((e) => e == x.name) !== -1
+    ) {
+      // TODO: is there something smarter we can do then take the mean of all labels?
+      let v = 0;
+      let n = 0;
+      x.metrics.forEach((m) => {
+        n++;
+        v += Number(m.value);
+      });
+      new_values[columns.indexOf(x.name + "_value")] = v / n;
+    }
+  });
+
+  values.push(new_values);
+
+  const timeTaken = Date.now() - start;
+  await sleep((100 - timeTaken) / 1000);
+}
+
+const transfer_values = values.map((x) => x.buffer);
+self.postMessage({ columns, transfer_values }, transfer_values);
--- a/benchmarks/worker.ts
+++ b/benchmarks/worker.ts
@@ -0,0 +1,132 @@
+/// <reference no-default-lib="true" />
+/// <reference lib="deno.worker" />
+import { sleep } from "https://deno.land/x/sleep@v1.2.1/sleep.ts";
+import * as windmill from "https://deno.land/x/windmill@v1.37.0/mod.ts";
+import * as api from "https://deno.land/x/windmill@v1.37.0/windmill-api/index.ts";
+
+const promise = new Promise<
+  api.Configuration & {
+    workspace_id: string;
+    per_worker_throughput: number;
+    useFlows: boolean;
+  }
+>((resolve, _reject) => {
+  self.onmessage = (evt) => {
+    const sharedConfig = evt.data;
+    const config = {
+      ...api.createConfiguration({
+        baseServer: new api.ServerConfiguration(sharedConfig.server, {}),
+        authMethods: {
+          bearerAuth: {
+            tokenProvider: {
+              getToken() {
+                return sharedConfig.token;
+              },
+            },
+          },
+        },
+      }),
+      workspace_id: sharedConfig.workspace_id,
+      per_worker_throughput: sharedConfig.per_worker_throughput,
+      useFlows: sharedConfig.useFlows,
+    };
+    self.name = "Worker " + sharedConfig.i;
+    resolve(config);
+    self.onmessage = null;
+  };
+});
+const config = await promise;
+const jobApi = new windmill.JobApi(config);
+const outstanding: string[] = [];
+let cont = true;
+let total_spawned = 0;
+const start_time = Date.now();
+let complete_timeout = Infinity;
+
+self.onmessage = (evt) => {
+  cont = false;
+  complete_timeout = evt.data;
+};
+
+while (cont) {
+  if ((await jobApi.listQueue(config.workspace_id)).length > 500) {
+    console.log("queue very long. 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;
+  }
+  let uuid: string;
+  if (config.useFlows) {
+    uuid = await jobApi.runFlowPreview(config.workspace_id, {
+      args: {},
+      value: {
+        modules: [
+          {
+            inputTransforms: {},
+            value: {
+              language: "deno",
+              type: "rawscript",
+              content:
+                'export function main(){ return Deno.env.get("WM_JOB_ID"); }',
+            },
+          },
+          {
+            inputTransforms: {},
+            value: {
+              language: "deno",
+              type: "rawscript",
+              content:
+                'export function main(){ return Deno.env.get("WM_JOB_ID"); }',
+            },
+          },
+        ],
+      },
+    });
+  } else {
+    uuid = await jobApi.runScriptPreview(config.workspace_id, {
+      language: "deno",
+      content: 'export function main(){ return Deno.env.get("WM_JOB_ID"); }',
+      args: {},
+    });
+  }
+  outstanding.push(uuid);
+  total_spawned++;
+}
+
+const end_time = Date.now() + complete_timeout;
+while (outstanding.length > 0 && Date.now() < end_time) {
+  const uuid = outstanding.shift()!;
+  const r = await jobApi.getJob(config.workspace_id, uuid);
+  if (r.running) {
+    outstanding.push(uuid);
+    continue;
+  } else if (!config.useFlows) {
+    try {
+      let result: string;
+      if (r.result) {
+        result = r.result;
+      } else {
+        const j = await jobApi.getCompletedJob(config.workspace_id, uuid);
+        result = j.result;
+      }
+      if (result != uuid) {
+        console.log(
+          "job did not return correct UUID: " + result + " != " + uuid
+        );
+      }
+    } catch (e) {
+      console.log("error during wait: ", e);
+      outstanding.push(uuid);
+      continue;
+    }
+  }
+}
+
+self.postMessage(outstanding.length);