kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: benchmarks graph (#2244)

Browse Source 
* feat: benchmarks graph

* feat: benchmarks graph gh action

* fix: gh action for testing

* fix: gh action

* Update benchmark.yml

* Update benchmark.yml

* Update benchmark.yml

* feat: deployed scripts + more langs

* fix: gh action regex

* fix: increase time

* fix: title + logs
This commit is contained in:
HugoCasa
2023-09-07 15:45:08 +02:00
committed by GitHub
parent 0868717a73
commit fc92d8049f
8 changed files with 1110 additions and 662 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
.github/workflows/benchmark.yml vendored
View File

@@ -32,10 +32,23 @@ jobs:
- uses: denoland/setup-deno@v1
with:
deno-version: v1.x
- uses: actions/checkout@v4
with:
ref: benchmarks
- name: benchmark
timeout-minutes: 10
run:
deno run --unstable -A
https://raw.githubusercontent.com/windmill-labs/windmill/main/benchmarks/benchmark_noop.ts
--host http://localhost:8000 -e admin@windmill.dev -p changeme -j
10000
run: deno run --unstable -A
https://raw.githubusercontent.com/windmill-labs/windmill/${GITHUB_REF##ref/head/}/benchmarks/benchmark_suite.ts
--host http://localhost:8000
-e admin@windmill.dev
-p changeme
-c https://raw.githubusercontent.com/windmill-labs/windmill/${GITHUB_REF##ref/head/}/benchmarks/suite_config.json
--branch ${GITHUB_REF##ref/head/}
- name: Push changes
run: |
pwd
git add .
git config --local user.email "41898282+github-actions[bot]@users.noreply.github.com"
git config --local user.name "github-actions[bot]"
git commit -m "Update benchmarks"
git push

2
benchmarks/action.ts
View File

@@ -2,7 +2,7 @@ import {
FlowValue,
JobService,
Preview as ScriptPreview,
} from "https://deno.land/x/windmill@v1.38.5/windmill-api/index.ts";
} from "https://deno.land/x/windmill@v1.167.0/windmill-api/index.ts";
export type Action = FlowAction | ScriptAction | RandomAction;

426
benchmarks/benchmark_noop.ts
View File

@@ -20,213 +20,231 @@ async function login(email: string, password: string): Promise<string> {
});
}
export const VERSION = "v1.125.1";
export const VERSION = "v1.167.0";
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);
export async function main({
host,
email,
password,
token,
workspace,
jobs,
batches,
}: {
host: string;
email?: string;
password?: string;
token?: string;
workspace: string;
jobs: number;
batches: number;
}) {
windmill.setClient("", host);
console.log(
"Started benchmark with NOOP jobs with options",
JSON.stringify(
{
host,
email,
workspace,
},
null,
4
)
);
console.log(
"Started benchmark with NOOP jobs with options",
JSON.stringify(
{
host,
email,
workspace,
},
null,
4
)
);
const config = {
token: "",
server: host,
workspace_id: workspace,
};
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;
let lastElapsed = 0;
let lastQueueLength = queue_length;
const updateState = setInterval(async () => {
const elapsed = start ? Date.now() - start : 0;
queue_length = (
await (
await fetch(
host + "/api/w/" + config.workspace_id + "/jobs/queue/count",
{ headers: { ["Authorization"]: "Bearer " + config.token } }
)
).json()
).database_length;
const avgThr = (((jobsSent - queue_length) / elapsed) * 1000).toFixed(
2
);
const instThr =
lastElapsed > 0
? (
((lastQueueLength - queue_length) / (elapsed - lastElapsed)) *
1000
).toFixed(2)
: 0;
lastElapsed = elapsed;
lastQueueLength = queue_length;
await Deno.stdout.write(
enc(
`elapsed: ${(elapsed / 1000).toFixed(2)} | jobs executed: ${
jobsSent - queue_length
}/${jobsSent} (thr: inst ${instThr} - avg ${avgThr}) | 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.0;
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");
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;
}
)
.command(
"upgrade",
new UpgradeCommand({
main: "main.ts",
args: [
"--allow-net",
"--allow-read",
"--allow-write",
"--allow-env",
"--unstable",
],
provider: new DenoLandProvider({ name: "wmillbench" }),
} 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;
let lastElapsed = 0;
let lastQueueLength = queue_length;
const updateState = setInterval(async () => {
const elapsed = start ? Date.now() - start : 0;
queue_length = (
await (
await fetch(
host + "/api/w/" + config.workspace_id + "/jobs/queue/count",
{ headers: { ["Authorization"]: "Bearer " + config.token } }
)
).json()
).database_length;
const avgThr = (((jobsSent - queue_length) / elapsed) * 1000).toFixed(2);
const instThr =
lastElapsed > 0
? (
((lastQueueLength - queue_length) / (elapsed - lastElapsed)) *
1000
).toFixed(2)
: 0;
lastElapsed = elapsed;
lastQueueLength = queue_length;
await Deno.stdout.write(
enc(
`elapsed: ${(elapsed / 1000).toFixed(2)} | jobs executed: ${
jobsSent - queue_length
}/${jobsSent} (thr: inst ${instThr} - avg ${avgThr}) | 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.0;
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");
return {
throughput: jobsSent / total_duration_sec,
};
}
if (import.meta.main) {
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",
})
)
.parse();
.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(main)
.command(
"upgrade",
new UpgradeCommand({
main: "main.ts",
args: [
"--allow-net",
"--allow-read",
"--allow-write",
"--allow-env",
"--unstable",
],
provider: new DenoLandProvider({ name: "wmillbench" }),
})
)
.parse();
}

194
benchmarks/benchmark_suite.ts Normal file
View File

@@ -0,0 +1,194 @@
import { Command } from "https://deno.land/x/cliffy@v0.25.7/command/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";
const VERSION = "1.167.0";
type Config = {
benchmarks: [
{
graph_title: string;
name: string;
jobs: number | undefined;
type: "noop" | "flow" | "deno" | "python" | "go" | "bash";
}
];
};
async function main({
host,
email,
password,
token,
workspace,
configPath,
branch,
}: {
host: string;
email?: string;
password?: string;
token?: string;
workspace: string;
configPath: string;
branch?: string;
}) {
const { main: runNoopBenchmark } = await import(
branch !== undefined
? `https://raw.githubusercontent.com/windmill-labs/windmill/${branch}/benchmarks/benchmark_noop.ts`
: "./benchmark_noop.ts"
);
const { main: runBenchmark } = await import(
branch !== undefined
? `https://raw.githubusercontent.com/windmill-labs/windmill/${branch}/benchmarks/main.ts`
: "./main.ts"
);
const { drawGraph } = await import(
branch !== undefined
? `https://raw.githubusercontent.com/windmill-labs/windmill/${branch}/benchmarks/graph.ts`
: "./graph.ts"
);
async function getConfig(configPath: string): Promise<Config> {
if (configPath.startsWith("http")) {
const response = await fetch(configPath);
return await response.json();
} else {
return JSON.parse(await Deno.readTextFile(configPath));
}
}
try {
const config = await getConfig(configPath);
for (const benchmark of config.benchmarks) {
try {
console.log(
"%cRunning benchmark " + benchmark.name,
"font-weight: bold;"
);
let result:
| {
throughput: number;
}
| undefined;
if (benchmark.type === "noop") {
result = await runNoopBenchmark({
host,
email,
password,
token,
workspace,
jobs: 1000,
batches: 1,
});
} else {
result = await runBenchmark({
host,
email,
password,
token,
workspace,
workers: 1,
seconds: benchmark.type === "flow" ? 2 : 5,
metrics: "http://localhost:8001/metrics",
maximumThroughput: Infinity,
zombieTimeout: 90000,
histogramBuckets: [],
scriptPattern: [
"deno",
"python",
"go",
"bash",
"dedicated",
].includes(benchmark.type)
? benchmark.type
: "deno",
useFlows: benchmark.type === "flow",
hideProgress: true,
});
}
if (!result) {
throw new Error("No result returned");
}
const stat = {
value: result.throughput,
ts: Date.now(),
};
let data: (typeof stat)[] = [];
const jsonFilePath = `${benchmark.name}.json`;
try {
const existing = await Deno.readTextFile(jsonFilePath);
data = JSON.parse(existing);
} catch (_) {
console.log("No existing data file found, creating new one.");
}
data.push(stat);
await Deno.writeTextFile(jsonFilePath, JSON.stringify(data, null, 4));
const svg = drawGraph(
data.slice(-10).map((d) => ({ ...d, date: new Date(d.ts) })),
benchmark.graph_title
);
await Deno.writeTextFile(`${benchmark.name}.svg`, svg);
} catch (err) {
console.error("Failed to run benchmark", benchmark.name, err);
}
}
Deno.exit(0); // JSDOM from drawGraph doesn't exit cleanly
} catch (err) {
return console.error(`Failed to read config file ${configPath}: ${err}`);
}
}
await new Command()
.name("wmillbenchsuite")
.description("Run benchmark suite 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("-c --config-path <config:string>", "The path of the config file", {
required: true,
})
.option(
"--branch <branch:string>",
"The branch to use when running remotely."
)
.action(main)
.command(
"upgrade",
new UpgradeCommand({
main: "main.ts",
args: [
"--allow-net",
"--allow-read",
"--allow-write",
"--allow-env",
"--unstable",
],
provider: new DenoLandProvider({ name: "wmillbench" }),
})
)
.parse();

118
benchmarks/graph.ts Normal file
View File

@@ -0,0 +1,118 @@
import * as d3 from "https://cdn.jsdelivr.net/npm/d3@7/+esm";
import { JSDOM } from "https://jspm.dev/jsdom";
type DataPoint = {
value: number;
date: Date;
};
export function drawGraph(data: DataPoint[], title: string) {
const context = {
jsdom: new JSDOM(""),
};
const { window } = context.jsdom;
const { document } = window;
const body = d3.select(document).select("body");
const width = 400;
const height = 200;
const marginTop = 20;
const marginRight = 30;
const marginBottom = 30;
const marginLeft = 60;
const svg = body
.append("svg")
.attr("width", width + marginLeft + marginRight)
.attr("height", height + marginTop + marginBottom)
.append("g")
.attr("transform", "translate(" + marginLeft + "," + marginTop + ")");
const x = d3
.scaleTime()
.domain(
d3.extent(data, function (d: DataPoint) {
return d.date;
})
)
.nice()
.range([0, width]);
const xAxis = d3.axisBottom(x).ticks(5);
svg
.append("g")
.attr("transform", "translate(0," + height + ")")
.call(xAxis);
// Add Y axis
const y = d3
.scaleLinear()
.domain([
0,
d3.max(data, function (d: DataPoint) {
return +d.value;
}) * 1.5,
])
.range([height, 0])
.nice();
svg.append("g").call(d3.axisLeft(y));
svg
.append("text")
.attr("text-anchor", "middle")
.attr("style", "font-size: 12px")
.attr("transform", "rotate(-90)")
.attr("y", -marginLeft + 20)
.attr("x", -height / 2)
.text("[jobs/s]");
svg
.append("text")
.attr("text-anchor", "middle")
.attr("style", "font-size: 16px")
.attr("y", 0)
.attr("x", width / 2)
.text(title);
// Add the line
svg
.append("path")
.datum(data)
.attr("fill", "none")
.attr("stroke", "steelblue")
.attr("stroke-width", 1.5)
.attr(
"d",
d3.line(
function (d: DataPoint) {
return x(d.date);
},
function (d: DataPoint) {
return y(d.value);
}
)
);
return body.node().innerHTML;
}
if (import.meta.main) {
const svg = drawGraph(
[
{
value: 10,
date: new Date(86400000),
},
{
value: 12,
date: new Date(86400000 * 2),
},
],
"test"
);
console.log(svg);
Deno.exit(0);
}

874
benchmarks/main.ts
View File

@@ -3,7 +3,8 @@
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.145.0/mod.ts";
import * as windmill from "https://deno.land/x/windmill@v1.167.0/mod.ts";
import * as api from "https://deno.land/x/windmill@v1.167.0/windmill-api/index.ts";
import { Action } from "./action.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";
@@ -23,430 +24,517 @@ async function login(email: string, password: string): Promise<string> {
export const VERSION = "v1.168.3";
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(
"--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("--max <max:number>", "Maximum number of operations performed.")
.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("-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(
"--flow-pattern <pattern:string>",
"Use a different flow pattern among: 2steps, onebranch (Default 2steps)"
)
.option(
"--script-pattern <pattern:string>",
"Use a different script pattern among: denotrivial, identity, httpversion, httpslow (Default denotrivial)"
)
.option("--custom <custom_path:string>", "Use custom actions during bench")
.option(
"--zombie-timeout",
"The maximum time in ms to wait for jobs to complete.",
{
default: 90000,
}
)
.option(
"--continous",
"Run the benchmark forever. This effectively disables metric collection & exports. No zombie jobs will be tracked."
)
.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,
flowPattern,
scriptPattern,
zombieTimeout,
continous,
max,
custom,
}) => {
windmill.setClient("", host);
console.log("Backend version: " + (await fetch(`${host}/api/version`)));
export async function main({
host,
workers: num_workers,
seconds,
email,
password,
token,
workspace,
metrics,
exportJson,
exportCsv,
exportHistograms,
exportSimple,
histogramBuckets,
maximumThroughput,
useFlows,
flowPattern,
scriptPattern,
zombieTimeout,
continous,
max,
custom,
hideProgress,
}: {
host: string;
workers: number;
seconds: number;
email?: string;
password?: string;
token?: string;
workspace: string;
metrics: string;
exportJson?: string;
exportCsv?: string;
exportHistograms?: string[];
exportSimple?: string[];
histogramBuckets: string[];
maximumThroughput: number;
useFlows?: boolean;
flowPattern?: string;
scriptPattern?: string;
zombieTimeout: number;
continous?: boolean;
max?: number;
custom?: string;
hideProgress?: boolean;
}) {
windmill.setClient("", host);
const versionResp = await fetch(`${host}/api/version`);
console.log("Backend version: " + (await versionResp.text()));
const custom_content: Action | undefined = custom
? JSON.parse(await Deno.readTextFile(custom))
: undefined;
const custom_content: Action | undefined = custom
? JSON.parse(await Deno.readTextFile(custom))
: undefined;
if (!Array.isArray(histogramBuckets)) {
histogramBuckets = [];
}
if (!Array.isArray(histogramBuckets)) {
histogramBuckets = [];
}
if (!Array.isArray(exportHistograms)) {
exportHistograms = [];
}
if (!Array.isArray(exportHistograms)) {
exportHistograms = [];
}
if (!Array.isArray(exportSimple)) {
exportSimple = [];
}
if (!Array.isArray(exportSimple)) {
exportSimple = [];
}
let metrics_worker: Worker | undefined = undefined;
if (!continous) {
if (exportJson || exportCsv) {
metrics_worker = new Worker(
new URL("./scraper.ts", import.meta.url).href,
{
type: "module",
}
);
metrics_worker.postMessage({
exportHistograms,
histogramBuckets,
exportSimple,
host: metrics,
});
let metrics_worker: Worker | undefined = undefined;
if (!continous) {
if (exportJson || exportCsv) {
metrics_worker = new Worker(
new URL("./scraper.ts", import.meta.url).href,
{
type: "module",
}
}
console.log(
"Started with options",
JSON.stringify(
{
host,
num_workers,
seconds,
email,
workspace,
metrics,
exportJson,
exportCsv,
exportHistograms,
exportSimple,
maximumThroughput,
useFlows,
flowPattern,
scriptPattern,
zombieTimeout,
continous,
},
null,
4
)
);
const config = {
token: "",
server: host,
workspace_id: workspace,
};
metrics_worker.postMessage({
exportHistograms,
histogramBuckets,
exportSimple,
host: metrics,
});
}
}
let final_token: string;
if (!token) {
if (email && password) {
console.log("Logging in with email and password...");
final_token = await login(email, password);
console.log("Logged in!");
} else {
console.error("Token or email with password are required.");
return;
}
} else {
final_token = token;
}
console.log("Using token", final_token);
config.token = final_token;
windmill.setClient(final_token, host);
const per_worker_throughput = maximumThroughput / num_workers;
const max_per_worker = max ? max / num_workers : undefined;
const shared_config = {
server: host,
token: final_token,
workspace_id: config.workspace_id,
per_worker_throughput,
max_per_worker,
console.log(
"Started with options",
JSON.stringify(
{
host,
num_workers,
seconds,
email,
workspace,
metrics,
exportJson,
exportCsv,
exportHistograms,
exportSimple,
maximumThroughput,
useFlows,
flowPattern,
scriptPattern,
zombieTimeout,
continous,
custom: custom_content,
};
hideProgress,
},
null,
4
)
);
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",
});
}
const config = {
token: "",
server: host,
workspace_id: workspace,
};
let start: number | undefined = undefined;
let final_token: string;
if (!token) {
if (email && password) {
console.log("Logging in with email and password...");
final_token = await login(email, password);
console.log("Logged in!");
} else {
console.error("Token or email with password are required.");
return;
}
} else {
final_token = token;
}
const jobsSent = Array(num_workers).fill(0);
const enc = (s: string) => new TextEncoder().encode(s);
console.log("Using token", final_token);
const updateState = setInterval(async () => {
const elapsed = start ? Math.ceil((Date.now() - start) / 1000) : 0;
const sum = jobsSent.reduce((a, b) => a + b, 0);
let queue_length = -1;
while (queue_length === -1) {
try {
queue_length = (
await (
await fetch(
host + "/api/w/" + config.workspace_id + "/jobs/queue/count",
{ headers: { ["Authorization"]: "Bearer " + config.token } }
)
).json()
).database_length;
} catch (e) {
console.log(
`queue count not reachable. waiting... `
);
await sleep(0.5);
continue;
}
}
await Deno.stdout.write(
enc(
`elapsed: ${elapsed}/${seconds} | jobs sent: ${JSON.stringify(
jobsSent
)} (sum: ${sum} thr: ${(sum / elapsed).toFixed(
2
)}) | queue: ${queue_length} \r`
)
);
}, 100);
config.token = final_token;
windmill.setClient(final_token, host);
workers.forEach((worker, i) => {
worker.addEventListener("message", (evt: MessageEvent<any>) => {
if (evt.data.type === "jobs_sent") {
jobsSent[i] = evt.data.jobs_sent;
}
});
worker.postMessage({ ...shared_config, i });
const per_worker_throughput = maximumThroughput / num_workers;
const max_per_worker = max ? max / num_workers : undefined;
const shared_config = {
server: host,
token: final_token,
workspace_id: config.workspace_id,
per_worker_throughput,
max_per_worker,
useFlows,
flowPattern,
scriptPattern,
continous,
custom: custom_content,
hideProgress,
};
if (
!useFlows &&
(scriptPattern === undefined ||
["deno", "python", "go", "bash"].includes(scriptPattern))
) {
console.log("Creating benchmark script...");
const path = `f/benchmarks/${scriptPattern || "deno"}`;
const exists = await windmill.ScriptService.existsScriptByPath({
workspace,
path,
});
if (exists) {
await windmill.ScriptService.deleteScriptByPath({
workspace,
path,
});
start = Date.now();
}
console.log("collecting samples...");
if (continous) {
while (true) {
await sleep(Infinity);
}
}
let scriptContent: string;
let language: string;
if (scriptPattern === "python") {
scriptContent =
'import os\n\ndef main():\n return os.environ.get("WM_JOB_ID")';
language = "python3";
} else if (scriptPattern === "go") {
scriptContent =
'package inner\nimport "os"\nfunc main() (string, error) { return os.Getenv("WM_JOB_ID"), nil }';
language = "go";
} else if (scriptPattern === "bash") {
scriptContent = "echo $WM_JOB_ID";
language = "bash";
} else {
scriptContent =
'export function main(){ return Deno.env.get("WM_JOB_ID"); }';
language = "deno";
}
await sleep(seconds);
await windmill.ScriptService.createScript({
workspace,
requestBody: {
path,
content: scriptContent,
summary: (scriptPattern || "deno") + " benchmark",
description: "",
language: language as api.NewScript.language,
},
});
clearInterval(updateState);
await sleep(5); // make sure script is created
}
let sum = jobsSent.reduce((a, b) => a + b, 0);
await Deno.stdout.write(
enc(" ".padStart(30) + `\rduration: ${seconds} | jobs sent: ${sum}\n`)
);
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",
});
}
const shutdown_start = Date.now();
let zombie_jobs = 0;
let incorrect_results = 0;
workers.forEach((worker, i) => {
const l = (evt: MessageEvent<any>) => {
if (evt.data.type === "zombie_jobs") {
zombie_jobs += evt.data.zombie_jobs;
incorrect_results += evt.data.incorrect_results;
worker.removeEventListener("message", l);
workers = workers.filter((w) => w != worker);
jobsSent[i] = evt.data.jobs_sent;
worker.terminate();
}
};
worker.addEventListener("message", l);
worker.postMessage(
Number.isSafeInteger(zombieTimeout) ? zombieTimeout : 90000
);
});
let start: number | undefined = undefined;
console.log("waiting for shutdown\n");
while (workers.length > 0) {
await sleep(0.1);
}
sum = jobsSent.reduce((a, b) => a + b, 0);
const jobsSent = Array(num_workers).fill(0);
const enc = (s: string) => new TextEncoder().encode(s);
const tts = (Date.now() - shutdown_start) / 1000;
const time = seconds + tts;
console.log("\ntime to shutdown:", tts);
console.log("jobs:", sum);
console.log("time (s + tts):", time);
console.log("throughput /s (jobs/time):", sum / time);
console.log("zombie jobs: ", zombie_jobs);
console.log("incorrect results: ", incorrect_results);
console.log(
"queue length:",
(
const updateState = setInterval(async () => {
const elapsed = start ? Math.ceil((Date.now() - start) / 1000) : 0;
const sum = jobsSent.reduce((a, b) => a + b, 0);
let queue_length = -1;
while (queue_length === -1) {
try {
queue_length = (
await (
await fetch(
host + "/api/w/" + config.workspace_id + "/jobs/queue/count",
{ headers: { ["Authorization"]: "Bearer " + config.token } }
)
).json()
).database_length
);
if (metrics_worker) {
metrics_worker.postMessage("stop");
console.log("waiting for metrics");
const { columns, transfer_values } = await new Promise<{
columns: string[];
transfer_values: ArrayBufferLike[];
}>((resolve, _reject) => {
if (metrics_worker) {
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();
}
} else {
return;
).database_length;
} catch (e) {
console.log(
`queue count not reachable. waiting... `
);
await sleep(0.5);
continue;
}
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" }),
await Deno.stdout.write(
enc(
`elapsed: ${elapsed}/${seconds} | jobs sent: ${JSON.stringify(
jobsSent
)} (sum: ${sum} thr: ${(sum / elapsed).toFixed(
2
)}) | queue: ${queue_length} \r`
)
);
}, 100);
workers.forEach((worker, i) => {
worker.addEventListener("message", (evt: MessageEvent<any>) => {
if (evt.data.type === "jobs_sent") {
jobsSent[i] = evt.data.jobs_sent;
}
});
worker.postMessage({ ...shared_config, i });
});
start = Date.now();
console.log("collecting samples...");
if (continous) {
while (true) {
await sleep(Infinity);
}
}
await sleep(seconds);
clearInterval(updateState);
let sum = jobsSent.reduce((a, b) => a + b, 0);
await Deno.stdout.write(
enc(" ".padStart(30) + `\rduration: ${seconds} | jobs sent: ${sum}\n`)
);
const shutdown_start = Date.now();
let zombie_jobs = 0;
let incorrect_results = 0;
workers.forEach((worker, i) => {
const l = (evt: MessageEvent<any>) => {
if (evt.data.type === "zombie_jobs") {
zombie_jobs += evt.data.zombie_jobs;
incorrect_results += evt.data.incorrect_results;
worker.removeEventListener("message", l);
workers = workers.filter((w) => w != worker);
jobsSent[i] = evt.data.jobs_sent;
worker.terminate();
}
};
worker.addEventListener("message", l);
worker.postMessage(
Number.isSafeInteger(zombieTimeout) ? zombieTimeout : 90000
);
});
console.log("waiting for shutdown\n");
while (workers.length > 0) {
await sleep(0.1);
}
sum = jobsSent.reduce((a, b) => a + b, 0);
const tts = (Date.now() - shutdown_start) / 1000;
const time = seconds + tts;
console.log("\ntime to shutdown:", tts);
console.log("jobs:", sum);
console.log("time (s + tts):", time);
console.log("throughput /s (jobs/time):", sum / time);
console.log("zombie jobs: ", zombie_jobs);
console.log("incorrect results: ", incorrect_results);
console.log(
"queue length:",
(
await (
await fetch(
host + "/api/w/" + config.workspace_id + "/jobs/queue/count",
{ headers: { ["Authorization"]: "Bearer " + config.token } }
)
).json()
).database_length
);
if (metrics_worker) {
metrics_worker.postMessage("stop");
console.log("waiting for metrics");
const { columns, transfer_values } = await new Promise<{
columns: string[];
transfer_values: ArrayBufferLike[];
}>((resolve, _reject) => {
if (metrics_worker) {
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");
return {
throughput: sum / time,
};
}
if (import.meta.main) {
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",
})
)
.parse();
.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("--max <max:number>", "Maximum number of operations performed.")
.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(
"-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 <export_histograms:string[]>",
"Mark metrics (without label) that are reported as histograms to export."
)
.option(
"--export-simple <export_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(
"--flow-pattern <pattern:string>",
"Use a different flow pattern among: 2steps, onebranch (Default 2steps)"
)
.option(
"--script-pattern <pattern:string>",
"Use a different script pattern among: deno, identity, python, go, bash (Default deno)"
)
.option("--custom <custom_path:string>", "Use custom actions during bench")
.option(
"--zombie-timeout <zombie_timeout:number>",
"The maximum time in ms to wait for jobs to complete.",
{
default: 90000,
}
)
.option(
"--continous",
"Run the benchmark forever. This effectively disables metric collection & exports. No zombie jobs will be tracked."
)
.option(
"--histogram-buckets <histogram_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",
],
}
)
.option("--hide-progress", "Hide worker progress logs")
.action(main)
.command(
"upgrade",
new UpgradeCommand({
main: "main.ts",
args: [
"--allow-net",
"--allow-read",
"--allow-write",
"--allow-env",
"--unstable",
],
provider: new DenoLandProvider({ name: "wmillbench" }),
})
)
.parse();
}

34
benchmarks/suite_config.json Normal file
View File

@@ -0,0 +1,34 @@
{
"benchmarks": [
{
"name": "noop_benchmark",
"graph_title": "noop throughput benchmark (single worker)",
"type": "noop"
},
{
"name": "flow_benchmark",
"graph_title": "flow throughput benchmark (single worker)",
"type": "flow"
},
{
"name": "deno_benchmark",
"graph_title": "deno throughput benchmark (single worker)",
"type": "deno"
},
{
"name": "python_benchmark",
"graph_title": "python throughput benchmark (single worker)",
"type": "python"
},
{
"name": "go_benchmark",
"graph_title": "go throughput benchmark (single worker)",
"type": "go"
},
{
"name": "bash_benchmark",
"graph_title": "bash throughput benchmark (single worker)",
"type": "bash"
}
]
}

101
benchmarks/worker.ts
View File

@@ -1,9 +1,9 @@
/// <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.151.0/mod.ts";
import * as api from "https://deno.land/x/windmill@v1.151.0/windmill-api/index.ts";
import { Job } from "https://deno.land/x/windmill@v1.151.0/windmill-api/index.ts";
import * as windmill from "https://deno.land/x/windmill@v1.167.0/mod.ts";
import * as api from "https://deno.land/x/windmill@v1.167.0/windmill-api/index.ts";
import { Job } from "https://deno.land/x/windmill@v1.167.0/windmill-api/index.ts";
import { Action, evaluate } from "./action.ts";
const promise = new Promise<{
@@ -17,6 +17,7 @@ const promise = new Promise<{
custom: Action | undefined;
server: string;
token: string;
hideProgress: boolean;
}>((resolve, _reject) => {
self.onmessage = (evt) => {
const sharedConfig = evt.data;
@@ -32,6 +33,7 @@ const promise = new Promise<{
custom: sharedConfig.custom,
server: sharedConfig.server,
token: sharedConfig.token,
hideProgress: sharedConfig.hideProgress,
};
self.name = "Worker " + sharedConfig.i;
resolve(config);
@@ -205,17 +207,13 @@ while (cont) {
language: api.RawScript.language.DENO,
type: "rawscript",
content:
'export function main(){ return Deno.env.get("WM_JOB_ID"); }',
'export function main(){ return Deno.env.get("WM_FLOW_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"); }',
type: "identity",
},
},
],
@@ -227,45 +225,25 @@ while (cont) {
requestBody: payload,
});
} else {
let payload: api.Preview;
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,
});
if (config.scriptPattern === "identity") {
uuid = await windmill.JobService.runScriptPreview({
workspace: config.workspace_id,
requestBody: {
path: "identity",
kind: api.Preview.kind.IDENTITY,
args: {
identity: "itsme",
},
},
});
} else {
uuid = await windmill.JobService.runScriptByPath({
workspace: config.workspace_id,
path: "f/benchmarks/" + (config.scriptPattern || "deno"),
requestBody: {},
});
}
} catch (e) {
console.error("error running script: " + e.body);
Deno.exit(1);
@@ -308,15 +286,17 @@ while (
Date.now() < end_time
) {
try {
await Deno.stdout.write(
enc(
"\rwaiting for jobs to complete: outstanding " +
outstanding.length +
" - queue" +
last_queue_length +
"\n"
)
);
if (!config.hideProgress) {
await Deno.stdout.write(
enc(
"\rwaiting for jobs to complete: outstanding " +
outstanding.length +
" - queue" +
last_queue_length +
"\n"
)
);
}
last_queue_length = await getQueueCount();
const uuid = outstanding.shift()!;
@@ -333,9 +313,12 @@ while (
}
if (r.type == "QueuedJob") {
outstanding.push(uuid);
await Deno.stdout.write(
enc(`uuid: ${uuid}, queue length: ${last_queue_length}\r`)
);
if (!config.hideProgress) {
await Deno.stdout.write(
enc(`uuid: ${uuid}, queue length: ${last_queue_length}\r`)
);
}
} else {
r = r as api.CompletedJob;
try {