starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-21 08:02:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

migrating benchmark to deno client 1.38.5

Browse Source
This commit is contained in:
Ruben Fiszel
2022-10-18 23:16:58 +02:00
parent f468d6d870
commit a34e514e4e
2 changed files with 71 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/main.ts
+26 -34
View File
@@ -3,17 +3,17 @@
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";
import * as windmill from "https://deno.land/x/windmill@v1.38.5/mod.ts";
async function login(
config: api.Configuration,
email: string,
password: string
): Promise<string> {
return await new windmill.UserApi(config).login({
email: email,
password: password,
return await windmill.UserService.login({
requestBody: {
email: email,
password: password,
}
});
}
@@ -131,6 +131,7 @@ await new Command()
useFlows,
zombieTimeout,
}) => {
windmill.setClient("", host);
const metrics_worker = new Worker(
new URL("./scraper.ts", import.meta.url).href,
{
@@ -173,20 +174,17 @@ await new Command()
zombieTimeout,
}, null, 4));
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, {}),
}),
const config = {
token: "",
server: host,
workspace_id: workspace,
};
let final_token: string;
if (!token) {
if (email && password) {
final_token = await login(config, email, password);
final_token = await login(email, password);
} else {
console.error("Token or email with password are required.");
return;
@@ -195,21 +193,8 @@ await new Command()
final_token = token;
}
config = {
...api.createConfiguration({
baseServer: config.baseServer,
authMethods: {
bearerAuth: {
tokenProvider: {
getToken() {
return final_token;
},
},
},
},
}),
workspace_id: config.workspace_id,
};
config.token = final_token;
windmill.setClient(final_token, host);
const per_worker_throughput = maximumThroughput / num_workers;
const shared_config = {
@@ -230,19 +215,21 @@ await new Command()
let start: number | undefined = undefined;
let jobSent = Array(num_workers).fill(0);
const jobsSent = Array(num_workers).fill(0);
const enc = (s: string) => new TextEncoder().encode(s);
const i = setInterval(async () => {
const updateState = setInterval(async () => {
const elapsed = start ? Math.ceil((Date.now() - start) / 1000) : 0;
await Deno.stdout.write(enc(`elapsed: ${elapsed}/${seconds} | jobs sent: ${JSON.stringify(jobSent)}\r`))
const sum = jobsSent.reduce((a, b) => a + b, 0)
const queue_length = (await windmill.JobService.listQueue({ workspace: config.workspace_id })).length
await Deno.stdout.write(enc(`elapsed: ${elapsed}/${seconds} | jobs sent: ${JSON.stringify(jobsSent)} (sum: ${sum}) | queue: ${queue_length} \r`))
}, 100);
workers.forEach((worker, i) => {
worker.postMessage({ ...shared_config, i });
worker.addEventListener("message", (evt: MessageEvent<any>) => {
if (evt.data.type === "jobs_sent") {
jobSent[i] = evt.data.jobs_sent
jobsSent[i] = evt.data.jobs_sent
}
})
});
@@ -250,14 +237,17 @@ await new Command()
await sleep(seconds);
clearInterval(i);
await Deno.stdout.write(enc(` \rduration: ${seconds} | jobs sent: ${JSON.stringify(jobSent)}\n`));
clearInterval(updateState);
await Deno.stdout.write(enc(" ".padStart(30) + `\rduration: ${seconds} | jobs sent: ${JSON.stringify(jobsSent)}\n`));
let zombie_jobs = 0;
let incorrect_results = 0;
workers.forEach((worker) => {
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);
worker.terminate();
@@ -275,6 +265,8 @@ await new Command()
}
console.log("zombie jobs: ", zombie_jobs);
console.log("incorrect results: ", incorrect_results);
console.log("queue length:", (await windmill.JobService.listQueue({ workspace: config.workspace_id })).length)
metrics_worker.postMessage("stop");
console.log("waiting for metrics");
benchmarks/worker.ts
+45 -49
View File
@@ -1,11 +1,11 @@
/// <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";
import * as windmill from "https://deno.land/x/windmill@v1.38.5/mod.ts";
import * as api from "https://deno.land/x/windmill@v1.38.5/windmill-api/index.ts";
const promise = new Promise<
api.Configuration & {
{
workspace_id: string;
per_worker_throughput: number;
useFlows: boolean;
@@ -13,19 +13,8 @@ const promise = new Promise<
>((resolve, _reject) => {
self.onmessage = (evt) => {
const sharedConfig = evt.data;
windmill.setClient(sharedConfig.token, sharedConfig.server)
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,
@@ -36,13 +25,11 @@ const promise = new Promise<
};
});
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;
@@ -54,9 +41,11 @@ const updateStatusInterval = setInterval(() => {
}, 100)
while (cont) {
if ((await jobApi.listQueue(config.workspace_id)).length > 500) {
console.log("queue very long. waiting...");
const queue_length = (await windmill.JobService.listQueue({ workspace: config.workspace_id })).length
if (queue_length > 500) {
console.log(`queue length: ${queue_length} > 500. waiting...`);
await sleep(0.5);
continue;
}
@@ -70,36 +59,40 @@ while (cont) {
}
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"); }',
uuid = await windmill.JobService.runFlowPreview({
workspace: config.workspace_id, requestBody: {
args: {},
value: {
modules: [
{
input_transforms: {},
value: {
language: api.RawScript.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"); }',
{
input_transforms: {},
value: {
language: api.RawScript.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: {},
uuid = await windmill.JobService.runScriptPreview({
workspace: config.workspace_id, requestBody: {
language: api.Preview.language.DENO,
content: 'export function main(){ return Deno.env.get("WM_JOB_ID"); }',
args: {},
}
});
}
outstanding.push(uuid);
@@ -109,18 +102,21 @@ while (cont) {
clearInterval(updateStatusInterval);
const end_time = Date.now() + complete_timeout;
let incorrect_results = 0;
while (outstanding.length > 0 && Date.now() < end_time) {
const uuid = outstanding.shift()!;
const r = await jobApi.getJob(config.workspace_id, uuid);
let r = await windmill.JobService.getJob({ workspace: config.workspace_id, id: uuid });
if (r.type == 'QueuedJob') {
outstanding.push(uuid);
console.log(uuid)
console.log(uuid, (await windmill.JobService.listQueue({ workspace: config.workspace_id })).length)
} else if (!config.useFlows) {
r = r as api.CompletedJob;
try {
if (r.result != uuid) {
console.log(
"job did not return correct UUID: " + r.result + " != " + uuid
"job did not return correct UUID: " + r.result + " != " + uuid + "job: \n" + JSON.stringify(r, null, 2),
);
incorrect_results++;
}
} catch (e) {
console.log("error during wait: ", e);
@@ -129,4 +125,4 @@ while (outstanding.length > 0 && Date.now() < end_time) {
}
}
self.postMessage({ type: "zombie_jobs", zombie_jobs: outstanding.length });
self.postMessage({ type: "zombie_jobs", zombie_jobs: outstanding.length, incorrect_results });