Files
openclaw/scripts/lib/gateway-bench-probes.ts
Vincent Koc 98feafe785 fix(scripts): prevent gateway benchmarks from stalling on probes (#119063)
* fix(scripts): bound gateway benchmark probes

* fix(scripts): bound concurrency response deadlines

* fix(perf): preserve benchmark socket reuse
2026-08-04 16:23:45 +08:00

188 lines
5.6 KiB
TypeScript

// Gateway Bench Probes script supports OpenClaw repository automation.
import { spawnSync } from "node:child_process";
import { request } from "node:http";
import { createServer } from "node:net";
import { expectDefined } from "../../packages/normalization-core/src/expect.ts";
const PROBE_REQUEST_TIMEOUT_MS = 100;
export async function getFreePort(): Promise<number> {
return new Promise((resolve, reject) => {
const server = createServer();
server.on("error", reject);
server.listen(0, "127.0.0.1", () => {
const address = server.address();
if (!address || typeof address === "string") {
server.close(() => reject(new Error("failed to allocate port")));
return;
}
const { port } = address;
server.close(() => resolve(port));
});
});
}
export async function requestProbeStatus(
port: number,
pathname: string,
): Promise<{ errorKind: string | null; status: number | null }> {
try {
const status = await requestStatus(port, pathname);
return {
errorKind: status === 200 ? null : `http-${status}`,
status,
};
} catch (error) {
return {
errorKind: classifyProbeErrorKind(error),
status: null,
};
}
}
function classifyProbeErrorKind(error: unknown): string {
if (typeof error === "object" && error !== null) {
const code = (error as { code?: unknown }).code;
if (typeof code === "string" && code.trim()) {
return code.trim().toLowerCase();
}
const message = (error as { message?: unknown }).message;
if (typeof message === "string" && message.toLowerCase().includes("probe timeout")) {
return "timeout";
}
const name = (error as { name?: unknown }).name;
if (typeof name === "string" && name.trim()) {
return name.trim().toLowerCase();
}
}
return "error";
}
export function readProcessRssMb(pid: number | undefined): number | null {
if (!pid || process.platform === "win32") {
return null;
}
const result = spawnSync("ps", ["-o", "rss=", "-p", String(pid)], {
encoding: "utf8",
stdio: ["ignore", "pipe", "ignore"],
});
if (result.status !== 0) {
return null;
}
const rssKb = parseProcessRssKb(result.stdout);
return rssKb === null ? null : rssKb / 1024;
}
export function parseProcessRssKb(raw: string): number | null {
const value = raw.trim();
if (!/^[1-9][0-9]*$/u.test(value)) {
return null;
}
const rssKb = Number(value);
return Number.isSafeInteger(rssKb) ? rssKb : null;
}
export function readProcessTreeCpuMs(rootPid: number | undefined): number | null {
if (!rootPid || process.platform === "win32") {
return null;
}
const result = spawnSync("ps", ["-eo", "pid=,ppid=,time="], {
encoding: "utf8",
stdio: ["ignore", "pipe", "ignore"],
});
if (result.status !== 0) {
return null;
}
const childrenByParent = new Map<number, number[]>();
const cpuByPid = new Map<number, number>();
for (const line of result.stdout.split("\n")) {
const match = line.trim().match(/^(\d+)\s+(\d+)\s+(\S+)$/u);
if (!match) {
continue;
}
const pid = Number(expectDefined(match[1], "process id from ps output"));
const ppid = Number(expectDefined(match[2], "parent process id from ps output"));
const cpuMs = parsePsCpuTimeMs(expectDefined(match[3], "CPU time from ps output"));
if (!Number.isInteger(pid) || !Number.isInteger(ppid) || cpuMs === null) {
continue;
}
cpuByPid.set(pid, cpuMs);
const children = childrenByParent.get(ppid) ?? [];
children.push(pid);
childrenByParent.set(ppid, children);
}
if (!cpuByPid.has(rootPid)) {
return null;
}
let totalCpuMs = 0;
const seen = new Set<number>();
const stack = [rootPid];
while (stack.length > 0) {
const pid = stack.pop();
if (!pid || seen.has(pid)) {
continue;
}
seen.add(pid);
totalCpuMs += cpuByPid.get(pid) ?? 0;
for (const childPid of childrenByParent.get(pid) ?? []) {
stack.push(childPid);
}
}
return totalCpuMs;
}
function requestStatus(port: number, pathname: string): Promise<number> {
return new Promise((resolve, reject) => {
let settled = false;
const settle = (run: () => void) => {
if (settled) {
return;
}
settled = true;
clearTimeout(timer);
run();
};
const req = request({ host: "127.0.0.1", method: "HEAD", path: pathname, port }, (res) => {
const status = res.statusCode ?? 0;
// Gateway probe HEAD responses carry the same status without a body to drain.
settle(() => resolve(status));
});
req.on("error", (error) => settle(() => reject(error)));
// Socket timeouts reset on activity, so enforce the attempt budget as wall-clock time.
const timer = setTimeout(() => {
const error = new Error("probe timeout");
settle(() => reject(error));
req.destroy(error);
}, PROBE_REQUEST_TIMEOUT_MS);
timer.unref?.();
req.end();
});
}
function parsePsCpuTimeMs(raw: string): number | null {
const parts = raw.trim().split(":").map(Number);
if (parts.some((part) => !Number.isFinite(part) || part < 0)) {
return null;
}
if (parts.length === 2) {
const [minutes, seconds] = parts;
return Math.round(
(expectDefined(minutes, "process CPU minutes") * 60 +
expectDefined(seconds, "process CPU seconds")) *
1000,
);
}
if (parts.length === 3) {
const [hours, minutes, seconds] = parts;
return Math.round(
(expectDefined(hours, "process CPU hours") * 60 * 60 +
expectDefined(minutes, "process CPU minutes") * 60 +
expectDefined(seconds, "process CPU seconds")) *
1000,
);
}
return null;
}