diff --git a/benchmarks/.vscode/settings.json b/benchmarks/.vscode/settings.json new file mode 100644 index 0000000000..07ce4d0d9d --- /dev/null +++ b/benchmarks/.vscode/settings.json @@ -0,0 +1,5 @@ +{ + "deno.enable": true, + "deno.unstable": true, + "editor.tabSize": 2 +} diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 0000000000..9500c3defe --- /dev/null +++ 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 - The windmill host to benchmark. (Default: "http://127.0.0.1/") + --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: + + 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 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. diff --git a/benchmarks/main.ts b/benchmarks/main.ts new file mode 100644 index 0000000000..7357adf278 --- /dev/null +++ b/benchmarks/main.ts @@ -0,0 +1,293 @@ +/// +/// + +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 { + 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 ", "The windmill host to benchmark.", { + default: "http://127.0.0.1/", + }) + .option( + "--workers ", + "The number of workers to run at once.", + { + default: 1, + } + ) + .option( + "-s --seconds ", + "How long to run the benchmark for (in seconds).", + { + default: 30, + } + ) + .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: "starter" } + ) + .option("-m --metrics ", "The url to scrape metrics from.", { + default: "http://localhost:8001/metrics", + }) + .option( + "--export-json ", + "If set, exports will be into a JSON file." + ) + .option( + "--export-csv ", + "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 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) => { + 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(); diff --git a/benchmarks/scraper.ts b/benchmarks/scraper.ts new file mode 100644 index 0000000000..cbbd8499a1 --- /dev/null +++ b/benchmarks/scraper.ts @@ -0,0 +1,88 @@ +/// +/// + +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 }]; + } + | { + type: "HISTOGRAM"; + metrics: [{ buckets: Record }]; + } + ) + ] = 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); diff --git a/benchmarks/worker.ts b/benchmarks/worker.ts new file mode 100644 index 0000000000..c69aa91436 --- /dev/null +++ b/benchmarks/worker.ts @@ -0,0 +1,132 @@ +/// +/// +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);