284 lines
14 KiB
TypeScript
284 lines
14 KiB
TypeScript
/** # 脚本说明
|
|
* 压测发射器(在 37 服务器以 root 运行):
|
|
* 从 pool.json 逐个(并发 workers)执行:
|
|
* 节点导入(幂等)→ POST lease(ttlMs=2h,规避 10 分钟空闲自动关闭)
|
|
* → POST browser/start → 按标签数经 CDP /json/new 开真实页面。
|
|
* 直到「资源耗尽」触发:MemAvailable<1.5GB / swap已用≥7.2GB /
|
|
* 连续启动失败≥10 / 已开浏览器被 OOM 杀≥10 / 达到 --max 封顶。
|
|
* 触发后停止发射(已开浏览器保持常驻),结论写 data/<label>-stop.json。
|
|
* 断点续跑:进度在 data/<label>-launched.jsonl,重启脚本会跳过已开指纹。
|
|
*
|
|
* 用法:bun run-stress.ts --pool data/pool.json --tabs 3 --label test1 [--max 500] [--concurrency 6]
|
|
*/
|
|
/** # 成员变量 辅助:进程内同步读文本(供 /proc 扫描) **/
|
|
const fs = require("fs") as typeof import("fs");
|
|
|
|
/** # 成员变量 服务基址(同机) **/
|
|
const ADMIN = "http://127.0.0.1:5310";
|
|
const NODE = "http://127.0.0.1:3106";
|
|
|
|
/** # 成员变量 基目录与数据目录 **/
|
|
const BASE = new URL(".", import.meta.url).pathname;
|
|
const DATA = `${BASE}data`;
|
|
|
|
/** # 成员变量 3 标签站点;1 标签测试只取第一个(google) **/
|
|
const URLS3 = ["https://www.google.com/", "https://www.yahoo.com/", "https://x.com/"];
|
|
|
|
/** # 成员变量 Lease TTL:2 小时(远大于 10 分钟空闲关闭窗口) **/
|
|
const LEASE_TTL_MS = 7_200_000;
|
|
|
|
/** # 成员变量 资源耗尽触发阈值(连续失败放宽,配合冷却防瞬时抖动误停) **/
|
|
const MEM_FLOOR_MB = 1500;
|
|
const SWAP_CEIL_MB = 7200;
|
|
const MAX_CONSEC_FAIL = 20;
|
|
const MAX_OOM_KILLED = 10;
|
|
/** # 成员变量 发射节流:load1 超过核数×3(37 为 8 核 → 24)暂停发射 **/
|
|
const LOAD_GATE = 24;
|
|
|
|
// ---- 参数解析 ----
|
|
const args = process.argv.slice(2);
|
|
function argOf(k: string, d: string): string { const i = args.indexOf(k); return i >= 0 ? args[i + 1] : d; }
|
|
const TABS = Number(argOf("--tabs", "3"));
|
|
const LABEL = argOf("--label", "test1");
|
|
const POOL_FILE = argOf("--pool", `${DATA}/pool.json`);
|
|
const MAX = Number(argOf("--max", "0"));
|
|
const CONCURRENCY = Number(argOf("--concurrency", "6"));
|
|
/** # 成员变量 本轮要开的真实站点列表 **/
|
|
const OPEN_URLS = TABS === 1 ? URLS3.slice(0, 1) : URLS3;
|
|
|
|
type Entry = {
|
|
fp: string; name: string; startedAt: string;
|
|
cdpPort: number | null; mainPid: number | null; tabs: string[];
|
|
leaseId?: string; userDataDir?: string;
|
|
};
|
|
|
|
// ---- 密钥(仅内存,不落盘不回显) ----
|
|
/** # 函数:通过 docker exec 查询 admin DB 取单值文本 **/
|
|
function sqlOne(sql: string): string {
|
|
const p = Bun.spawnSync(["docker", "exec", "serpmax-pg", "psql", "-U", "postgres", "-d", "serpmax_admin", "-tAc", sql], { stdout: "pipe", stderr: "pipe" });
|
|
const out = p.stdout.toString().trim();
|
|
if (p.exitCode !== 0 || !out) throw new Error(`SQL 失败: ${p.stderr.toString().slice(0, 300)}`);
|
|
return out;
|
|
}
|
|
const TOKEN = sqlOne("SELECT api_token FROM users WHERE id=2");
|
|
const envText = fs.readFileSync("/opt/serpmax/node/.env", "utf8");
|
|
const NODE_KEY = /^NODE_KEY=(.*)$/m.exec(envText)?.[1]?.trim() ?? "";
|
|
/** # 成员变量 ipmart.io-US 代理四元组(DB 直取) **/
|
|
const PROXY_ID = "19e3ab5c-c61e-4936-8a7f-d78f6635f57c";
|
|
const prow = sqlOne(`SELECT type,host,port,coalesce(username,''),coalesce(password,'') FROM proxies WHERE id='${PROXY_ID}'`).split("|");
|
|
|
|
/** # 成员变量 节点公共头(与 admin relayHeaders 一致:Token+节点密钥双证) **/
|
|
function nodeHeaders(extra: Record<string, string> = {}): Record<string, string> {
|
|
return { "content-type": "application/json", authorization: `Bearer ${TOKEN}`, "x-node-key": NODE_KEY, ...extra };
|
|
}
|
|
|
|
// ---- 状态 ----
|
|
const pool = (await Bun.file(POOL_FILE).json()) as { fingerprint: string; name: string }[];
|
|
const LAUNCHED_FILE = `${DATA}/${LABEL}-launched.json`;
|
|
/** # 成员变量 已启动记录(含续跑读取) **/
|
|
let launched: Entry[] = [];
|
|
try { launched = JSON.parse(await Bun.file(LAUNCHED_FILE).text() || "[]"); } catch { launched = []; }
|
|
const doneSet = new Set(launched.map((e) => e.fp));
|
|
console.log(`[run] pool=${pool.length} tabs=${TABS} 续跑已开=${launched.length}`);
|
|
const persist = () => Bun.write(LAUNCHED_FILE, JSON.stringify(launched));
|
|
|
|
// ---- 统计/控制 ----
|
|
let consecutiveFail = 0;
|
|
let totalFail = 0;
|
|
let oomKilled = 0;
|
|
let stopReason: string | null = null;
|
|
const t0 = Date.now();
|
|
|
|
/** # 函数:读内存水位(MB) **/
|
|
function meminfo(): { availMB: number; swapUsedMB: number } {
|
|
const txt = fs.readFileSync("/proc/meminfo", "utf8");
|
|
const get = (k: string) => Number(new RegExp(`${k}:\\s+(\\d+)`).exec(txt)?.[1] ?? 0);
|
|
return { availMB: get("MemAvailable") / 1024, swapUsedMB: (get("SwapTotal") - get("SwapFree")) / 1024 };
|
|
}
|
|
|
|
/** # 函数:统计已开浏览器中被杀死的数量。
|
|
* 判据:该 Profile 的 user-data-dir 下已无任何 chrome 进程,且记录时长超过 60s
|
|
* (startBrowser 返回的 processId 是包装进程,会自然退出,不能逐 PID 判)。
|
|
**/
|
|
function countKilled(): number {
|
|
const alive = new Set<string>();
|
|
for (const pid of fs.readdirSync("/proc").filter((d) => /^\d+$/.test(d))) {
|
|
try {
|
|
const cmd = fs.readFileSync(`/proc/${pid}/cmdline`, "utf8");
|
|
const m = /--user-data-dir=\S*\/profiles\/([\w.]+)\/browser/.exec(cmd);
|
|
if (m) alive.add(m[1]);
|
|
} catch { /* 进程已退出 */ }
|
|
}
|
|
const now = Date.now();
|
|
let dead = 0;
|
|
for (const e of launched) {
|
|
if (now - Date.parse(e.startedAt) < 60_000) continue;
|
|
if (!alive.has(e.fp)) dead++;
|
|
}
|
|
return dead;
|
|
}
|
|
|
|
/** # 函数:能否继续发射;返回触发原因或 null,顺带刷新 oomKilled **/
|
|
function budget(): string | null {
|
|
if (stopReason) return stopReason;
|
|
if (consecutiveFail >= MAX_CONSEC_FAIL) return `连续启动失败 ${consecutiveFail} 次(共失败 ${totalFail})`;
|
|
oomKilled = countKilled();
|
|
if (oomKilled >= MAX_OOM_KILLED) return `浏览器主进程消失(疑似 OOM)${oomKilled} 个`;
|
|
if (MAX > 0 && launched.length >= MAX) return `达到封顶数 ${MAX}`;
|
|
const m = meminfo();
|
|
if (m.availMB < MEM_FLOOR_MB) return `MemAvailable ${m.availMB | 0}MB < ${MEM_FLOOR_MB}MB`;
|
|
if (m.swapUsedMB >= SWAP_CEIL_MB) return `swap 已用 ${m.swapUsedMB | 0}MB ≥ ${SWAP_CEIL_MB}MB`;
|
|
return null;
|
|
}
|
|
|
|
async function nodeFetch(path: string, init: RequestInit = {}, timeout = 120000): Promise<Response> {
|
|
return fetch(`${NODE}${path}`, { ...init, headers: nodeHeaders(init.headers as Record<string, string> | undefined), signal: AbortSignal.timeout(timeout) });
|
|
}
|
|
|
|
/** # 函数:解开 {success, data} 包装,拿 payload **/
|
|
function unwrap<T>(j: T | { data: T }): T {
|
|
return (j as any)?.data ?? (j as any);
|
|
}
|
|
|
|
/** # 函数:节点导入(幂等;载荷与 admin relay 的 importProfileToNode 等价) **/
|
|
async function ensureImported(fp: string, name: string): Promise<void> {
|
|
const chk = await nodeFetch(`/profiles/${fp}`, {}, 10000);
|
|
if (chk.ok) return;
|
|
if (chk.status !== 404) throw new Error(`节点查询 ${fp} 状态 ${chk.status}`);
|
|
const res = await nodeFetch("/profiles", {
|
|
method: "POST",
|
|
headers: { ...nodeHeaders(), "idempotency-key": `stress-import-${fp}` },
|
|
body: JSON.stringify({
|
|
id: fp, name,
|
|
browser: { kind: "cloak", theme: "light" },
|
|
network: {},
|
|
proxy: { id: PROXY_ID, protocol: prow[0], host: prow[1], port: Number(prow[2]), username: prow[3], password: prow[4] },
|
|
metadata: { importedBy: "stress-run" },
|
|
}),
|
|
}, 15000);
|
|
if (!res.ok && res.status !== 409) throw new Error(`导入 ${fp} 失败 ${res.status}`);
|
|
}
|
|
|
|
/** # 函数
|
|
* 通过 WS 对指定标签发送 Page.navigate(内核 /json/new?url= 不生效,需逐标签导航)。
|
|
* 参数 port:CDP 端口;targetId:标签 id;url:目标网址
|
|
* 返回值:是否发出成功(10s 超时视为失败,不阻塞)
|
|
*/
|
|
async function navigateTab(port: number, targetId: string, url: string): Promise<boolean> {
|
|
return new Promise((resolve) => {
|
|
let ws: WebSocket | null = null;
|
|
const done = (ok: boolean) => { try { ws?.close(); } catch { /* ignore */ } resolve(ok); };
|
|
const timer = setTimeout(() => done(false), 10000);
|
|
try {
|
|
ws = new WebSocket(`ws://127.0.0.1:${port}/devtools/page/${targetId}`);
|
|
ws.onopen = () => ws!.send(JSON.stringify({ id: 1, method: "Page.navigate", params: { url } }));
|
|
ws.onmessage = (ev) => {
|
|
try {
|
|
const m = JSON.parse(String(ev.data));
|
|
if (m.id === 1) { clearTimeout(timer); done(!m.error); }
|
|
} catch { /* ignore */ }
|
|
};
|
|
ws.onerror = () => { clearTimeout(timer); done(false); };
|
|
} catch { clearTimeout(timer); done(false); }
|
|
});
|
|
}
|
|
|
|
/** # 函数:lease(2h)→ start → 开标签;返回每 Profile 记录 **/
|
|
async function launchOne(fp: string, name: string): Promise<Entry> {
|
|
await ensureImported(fp, name);
|
|
// 已有未过期 lease(如脚本重启续跑)则复用其 leaseId,否则新拿一个
|
|
let leaseId = "";
|
|
const detail = await nodeFetch(`/profiles/${fp}`, {}, 10000);
|
|
if (detail.ok) {
|
|
const d = unwrap<{ lease?: { leaseId: string; expiresAt: string } | null }>(await detail.json());
|
|
if (d.lease && new Date(d.lease.expiresAt).getTime() > Date.now()) leaseId = d.lease.leaseId;
|
|
}
|
|
if (!leaseId) {
|
|
const r = await nodeFetch(`/profiles/${fp}/lease`, { method: "POST", body: JSON.stringify({ consumerId: "stress", ttlMs: LEASE_TTL_MS }) }, 15000);
|
|
if (!r.ok) throw new Error(`lease ${r.status} ${(await r.text()).slice(0, 100)}`);
|
|
leaseId = unwrap<{ leaseId: string }>(await r.json()).leaseId;
|
|
}
|
|
const start = await nodeFetch(`/profiles/${fp}/browser/start`, { method: "POST", headers: { "x-lease-id": leaseId }, body: "{}" }, 150000);
|
|
if (!start.ok) throw new Error(`start ${start.status} ${(await start.text()).slice(0, 130)}`);
|
|
const s = unwrap<{ cdp?: { endpoint?: string; port?: number }; processId?: number; userDataDir?: string }>(await start.json());
|
|
const entry: Entry = { fp, name, startedAt: new Date().toISOString(), cdpPort: s.cdp?.port ?? null, mainPid: s.processId ?? null, tabs: [], leaseId, userDataDir: s.userDataDir };
|
|
// OOM 角色反转:浏览器主进程调「易杀」,系统其它进程由采样器置「受保护」
|
|
if (entry.mainPid) { try { fs.writeFileSync(`/proc/${entry.mainPid}/oom_score_adj`, "200"); } catch { /* 进程已退出 */ } }
|
|
// 开标签:CDP /json/new 建空页(Chrome 111+ 要求 PUT;405 回退 POST)→ WS Page.navigate 逐个导航
|
|
if (entry.cdpPort) {
|
|
for (const u of OPEN_URLS) {
|
|
const url = `http://127.0.0.1:${entry.cdpPort}/json/new`;
|
|
try {
|
|
let t = await fetch(url, { method: "PUT", signal: AbortSignal.timeout(15000) });
|
|
if (t.status === 405) t = await fetch(url, { method: "POST", signal: AbortSignal.timeout(15000) });
|
|
if (t.ok) {
|
|
const j: any = unwrap(await t.json().catch(() => null) ?? {});
|
|
const tid = j?.id ?? "";
|
|
entry.tabs.push(tid);
|
|
if (tid && !j?.url?.startsWith(u)) void navigateTab(entry.cdpPort, tid, u);
|
|
}
|
|
} catch { /* 单个标签失败不阻塞 */ }
|
|
}
|
|
}
|
|
return entry;
|
|
}
|
|
|
|
// ---- 主 workers ----
|
|
let cursor = 0;
|
|
/** # 成员变量 每 Profile 启动失败计数(≥2 判代理/数据坏,节点强删后跳过) **/
|
|
const failByFp = new Map<string, number>();
|
|
/** # 函数:读 1 分钟负载(发射节流用) **/
|
|
function load1(): number {
|
|
try { return Number(fs.readFileSync("/proc/loadavg", "utf8").split(/\s+/)[0]) || 0; } catch { return 0; }
|
|
}
|
|
async function worker(): Promise<void> {
|
|
while (!stopReason) {
|
|
if (cursor >= pool.length) return;
|
|
// 发射节流:系统负载超过门限则等待(防 CPU 打满导致的启动超时雪崩)
|
|
while (load1() > LOAD_GATE && !stopReason) await new Promise((r) => setTimeout(r, 10_000));
|
|
const i = cursor++;
|
|
const item = pool[i];
|
|
if (!item || doneSet.has(item.fingerprint)) continue;
|
|
const b = budget();
|
|
if (b) { stopReason = stopReason ?? b; return; }
|
|
try {
|
|
const entry = await launchOne(item.fingerprint, item.name);
|
|
doneSet.add(item.fingerprint);
|
|
launched.push(entry);
|
|
consecutiveFail = 0;
|
|
if (launched.length % 10 === 0) {
|
|
const m = meminfo();
|
|
console.log(`[run] launched=${launched.length} avail=${m.availMB | 0}MB swapUsed=${m.swapUsedMB | 0}MB t=${((Date.now() - t0) / 60000).toFixed(1)}min`);
|
|
persist();
|
|
}
|
|
} catch (e) {
|
|
consecutiveFail++;
|
|
totalFail++;
|
|
console.error(`[run] fail(${consecutiveFail}) ${item.fingerprint}: ${String(e).slice(0, 160)}`);
|
|
// 坏 Profile(嫌疑代理不通/数据脏,非超时类错误)失败两次 → 节点强删后跳过
|
|
// 启动超时(Timeout)不删:属系统瞬时拥塞,续跑时还会重试
|
|
const msg = String(e);
|
|
if (!msg.includes("Timeout")) {
|
|
const n = (failByFp.get(item.fingerprint) ?? 0) + 1;
|
|
failByFp.set(item.fingerprint, n);
|
|
if (n >= 2) {
|
|
try { await nodeFetch(`/profiles/${item.fingerprint}?force=true`, { method: "DELETE" }, 30000); } catch { /* best-effort */ }
|
|
doneSet.add(item.fingerprint);
|
|
console.error(`[run] drop ${item.fingerprint}(失败 ${n} 次,节点已强删)`);
|
|
}
|
|
}
|
|
// 连续失败冷却:6 次后等 45s 再继续(躲开发射抖动期)
|
|
if (consecutiveFail === 6) { console.error(`[run] 连续失败,冷却 45s`); await new Promise((r) => setTimeout(r, 45_000)); }
|
|
}
|
|
const b2 = budget();
|
|
if (b2 && !stopReason) stopReason = b2;
|
|
}
|
|
}
|
|
await persist();
|
|
await Promise.all(Array.from({ length: CONCURRENCY }, () => worker()));
|
|
await persist();
|
|
const m = meminfo();
|
|
const final = { stopReason, launchedCount: launched.length, totalFail, oomKilled, tabs: TABS, minutes: +((Date.now() - t0) / 60000).toFixed(1), availMB: m.availMB | 0, swapUsedMB: m.swapUsedMB | 0, stoppedAt: new Date().toISOString() };
|
|
await Bun.write(`${DATA}/${LABEL}-stop.json`, JSON.stringify(final, null, 2));
|
|
console.log(`[run] STOP: ${JSON.stringify(final)}`);
|