From a5f65a7943a6089ec113c20a55c5d842705dbf9b Mon Sep 17 00:00:00 2001 From: lyc8503 Date: Mon, 19 Jan 2026 00:52:32 +0800 Subject: [PATCH] feat(worker): check all targets in parellel, reduces ~15% cpu time usage in test, also try to a fix rare race condition --- worker/package-lock.json | 43 +++++++++++++++++++++++ worker/package.json | 3 ++ worker/src/index.ts | 73 ++++++++++------------------------------ worker/src/monitor.ts | 60 +++++++++++++++++++++++++++++++++ 4 files changed, 124 insertions(+), 55 deletions(-) diff --git a/worker/package-lock.json b/worker/package-lock.json index 3e97b6a..f9aa531 100644 --- a/worker/package-lock.json +++ b/worker/package-lock.json @@ -7,6 +7,9 @@ "": { "name": "uptimeworker", "version": "0.0.0", + "dependencies": { + "p-limit": "^7.2.0" + }, "devDependencies": { "@cloudflare/workers-types": "^4.20250410.0", "typescript": "^5.0.4", @@ -1299,6 +1302,21 @@ "node": ">=18.0.0" } }, + "node_modules/p-limit": { + "version": "7.2.0", + "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-7.2.0.tgz", + "integrity": "sha512-ATHLtwoTNDloHRFFxFJdHnG6n2WUeFjaR8XQMFdKIv0xkXjrER8/iG9iu265jOM95zXHAfv9oTkqhrfbIzosrQ==", + "license": "MIT", + "dependencies": { + "yocto-queue": "^1.2.1" + }, + "engines": { + "node": ">=20" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, "node_modules/path-to-regexp": { "version": "6.3.0", "resolved": "https://registry.npmjs.org/path-to-regexp/-/path-to-regexp-6.3.0.tgz", @@ -1519,6 +1537,18 @@ } } }, + "node_modules/yocto-queue": { + "version": "1.2.2", + "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-1.2.2.tgz", + "integrity": "sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ==", + "license": "MIT", + "engines": { + "node": ">=12.20" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, "node_modules/youch": { "version": "4.1.0-beta.10", "resolved": "https://registry.npmjs.org/youch/-/youch-4.1.0-beta.10.tgz", @@ -2196,6 +2226,14 @@ "zod": "3.22.3" } }, + "p-limit": { + "version": "7.2.0", + "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-7.2.0.tgz", + "integrity": "sha512-ATHLtwoTNDloHRFFxFJdHnG6n2WUeFjaR8XQMFdKIv0xkXjrER8/iG9iu265jOM95zXHAfv9oTkqhrfbIzosrQ==", + "requires": { + "yocto-queue": "^1.2.1" + } + }, "path-to-regexp": { "version": "6.3.0", "resolved": "https://registry.npmjs.org/path-to-regexp/-/path-to-regexp-6.3.0.tgz", @@ -2330,6 +2368,11 @@ "dev": true, "requires": {} }, + "yocto-queue": { + "version": "1.2.2", + "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-1.2.2.tgz", + "integrity": "sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ==" + }, "youch": { "version": "4.1.0-beta.10", "resolved": "https://registry.npmjs.org/youch/-/youch-4.1.0-beta.10.tgz", diff --git a/worker/package.json b/worker/package.json index 9ba07bf..e9e08ac 100644 --- a/worker/package.json +++ b/worker/package.json @@ -11,5 +11,8 @@ "@cloudflare/workers-types": "^4.20250410.0", "typescript": "^5.0.4", "wrangler": "^4.54.0" + }, + "dependencies": { + "p-limit": "^7.2.0" } } diff --git a/worker/src/index.ts b/worker/src/index.ts index b2cef75..64e4e30 100644 --- a/worker/src/index.ts +++ b/worker/src/index.ts @@ -1,12 +1,12 @@ import { DurableObject } from 'cloudflare:workers' import { MonitorTarget } from '../../types/config' import { workerConfig } from '../../uptime.config' -import { getStatus, getStatusWithGlobalPing } from './monitor' +import { doMonitor, getStatus } from './monitor' import { formatAndNotify, getWorkerLocation } from './util' import { CompactedMonitorStateWrapper, getFromStore, setToStore } from './store' +import pLimit from 'p-limit' export interface Env { - UPTIMEFLARE_STATE: KVNamespace REMOTE_CHECKER_DO: DurableObjectNamespace UPTIMEFLARE_D1: D1Database } @@ -24,62 +24,25 @@ const Worker = { let statusChanged = false const currentTimeSecond = Math.round(Date.now() / 1000) - // Check each monitor - // TODO: concurrent status check + // Parallel check multiple monitors + // Max concurrent connection is 6 limited by Cloudflare Workers, we use 5 here to be safe + type CheckResult = { id: string; location: string; status: { ping: number; up: boolean; err: string } } + let checkQueue: Promise[] = [] + let checkResult: Record = {}; + const limit = pLimit(5); for (const monitor of workerConfig.monitors) { - console.log(`[${workerLocation}] Checking ${monitor.name}...`) + checkQueue.push(limit(() => doMonitor(monitor, workerLocation, env))) + } + for (const result of await Promise.all(checkQueue)) { + checkResult[result.id] = result + } + + // Update each monitor's state based on check results + for (const monitor of workerConfig.monitors) { + console.log(`Processing monitor result: ${monitor.name} (${monitor.id})`) let monitorStatusChanged = false - let checkLocation = workerLocation - let status - - if (monitor.checkProxy) { - // Initiate a check using proxy (Geo-specific monitoring) - try { - console.log('Calling check proxy: ' + monitor.checkProxy) - let resp - if (monitor.checkProxy.startsWith('worker://')) { - const doLoc = monitor.checkProxy.replace('worker://', '') - const doId = env.REMOTE_CHECKER_DO.idFromName(doLoc) - const doStub = env.REMOTE_CHECKER_DO.get(doId, { - locationHint: doLoc as DurableObjectLocationHint, - }) - resp = await doStub.getLocationAndStatus(monitor) - try { - // Kill the DO instance after use, to avoid extra resource usage - await doStub.kill() - } catch (err) { - // An error here is expected, ignore it - } - } else if (monitor.checkProxy.startsWith('globalping://')) { - resp = await getStatusWithGlobalPing(monitor) - } else { - resp = await ( - await fetch(monitor.checkProxy, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(monitor), - }) - ).json<{ location: string; status: { ping: number; up: boolean; err: string } }>() - } - checkLocation = resp.location - status = resp.status - } catch (err) { - console.log('Error calling proxy: ' + err) - if (monitor.checkProxyFallback) { - console.log('Falling back to local check...') - status = await getStatus(monitor) - } else { - // TODO: more consistent error handling (throw or return?) - status = { ping: 0, up: false, err: 'Unknown check proxy error' } - } - } - } else { - // Initiate a check from the current location - status = await getStatus(monitor) - } - - const currentTimeSecond = Math.round(Date.now() / 1000) + const { location: checkLocation, status } = checkResult[monitor.id] // Update counters status.up ? state.data.overallUp++ : state.data.overallDown++ diff --git a/worker/src/monitor.ts b/worker/src/monitor.ts index d0ce142..8293483 100644 --- a/worker/src/monitor.ts +++ b/worker/src/monitor.ts @@ -1,3 +1,4 @@ +import { Env } from '.' import { MonitorTarget } from '../../types/config' import { withTimeout, fetchTimeout } from './util' @@ -352,3 +353,62 @@ export async function getStatus( return status } + +export async function doMonitor(monitor: MonitorTarget, defaultLocation: string, env: Env) { + let checkLocation = defaultLocation + let status + + if (monitor.checkProxy) { + // Initiate a check using proxy (Geo-specific monitoring) + try { + console.log(`[${monitor.id}] Calling check proxy: ${monitor.checkProxy}`) + let resp + if (monitor.checkProxy.startsWith('worker://')) { + const doLoc = monitor.checkProxy.replace('worker://', '') + const doId = env.REMOTE_CHECKER_DO.idFromName(monitor.id) + const doStub = env.REMOTE_CHECKER_DO.get(doId, { + locationHint: doLoc as DurableObjectLocationHint, + }) + resp = await doStub.getLocationAndStatus(monitor) + try { + // Kill the DO instance after use, to avoid extra resource usage + await doStub.kill() + } catch (err) { + // An error here is expected, ignore it + } + } else if (monitor.checkProxy.startsWith('globalping://')) { + resp = await getStatusWithGlobalPing(monitor) + } else { + resp = await ( + await fetch(monitor.checkProxy, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(monitor), + }) + ).json<{ location: string; status: { ping: number; up: boolean; err: string } }>() + } + checkLocation = resp.location + status = resp.status + } catch (err) { + console.log(`[${monitor.id}] Error calling proxy: ${err}`) + if (monitor.checkProxyFallback) { + console.log('Falling back to local check...') + status = await getStatus(monitor) + } else { + // TODO: more consistent error handling (throw or return?) + status = { ping: 0, up: false, err: 'Unknown check proxy error' } + } + } + } else { + // Initiate a check from the current location + status = await getStatus(monitor) + } + + console.log(`[${monitor.id}] Check result from ${checkLocation}: up=${status.up}, ping=${status.ping}, err=${status.err}`) + + return { + location: checkLocation, + status, + id: monitor.id, + } +}