fix(system): retired heartbeat rows no longer alarm on status page
- GET /system/health splits processes into current instances (hosts with
a beat in the last 10 min, newest per host+kind, not stopped) and
retiredProcesses[]; 45s stale and level only consider current ones
- heartbeat writer marks stopped_at on graceful exit (migration 0043),
upsert clears it; pure api role now stops its writer on SIGTERM
- scheduled job system.heartbeat_cleanup (maintain, 1h) deletes rows
without a beat for 24h and samples older than 7 days
- status page: per-role alarms ("io worker 45 秒没有心跳"), collapsible
retired instances; glossary/runbook updated
- bump all members to 0.2.2, CHANGELOG [0.2.2]
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
1 parent
fa261eb6cd
commit
b4c9b1ead2
32 files changed
+747
-91
No files matched your search
@@ -1,6 +1,9 @@
|
||||
import { join, resolve } from "node:path";
|
||||
import { z } from "zod";
|
||||
|
||||
/** compose 里 `${X:-}` 展开成空串:按未设置处理 */
|
||||
const emptyToUndef = (v: unknown) => (v === "" ? undefined : v);
|
||||
|
||||
const LogLevel = z.enum(["fatal", "error", "warn", "info", "debug", "trace", "silent"]);
|
||||
|
||||
export const ConfigSchema = z.object({
|
||||
@@ -66,6 +69,22 @@ export const ConfigSchema = z.object({
|
||||
ORGANIZE_ON_EXISTING: z.enum(["conflict", "skip", "replace_with_verify"]).default("conflict"),
|
||||
/** 随行字幕中此语言加 .default 标记(对照 MP DEFAULT_SUB);不设 = 不加 */
|
||||
ORGANIZE_DEFAULT_SUBTITLE_LANG: z.enum(["zh-cn", "zh-tw", "en", "ja", "ko"]).optional(),
|
||||
// ── 浏览器服务(camofox-browser,docs/runbook/browser.md)──
|
||||
/**
|
||||
* 部署自带的浏览器服务地址(compose 缺省 http://camofox:9377;接自有实例时改成它的地址)。
|
||||
* 设置后启动时登记/更新名为 camofox-env 的浏览器服务;不设则只用设置页登记的。
|
||||
*/
|
||||
BROWSER_PROVIDER_URL: z.preprocess(emptyToUndef, z.string().url().optional()),
|
||||
/** 该服务的访问凭据(camofox CAMOFOX_ACCESS_KEY);只在进程内使用,不入库 */
|
||||
BROWSER_ACCESS_KEY: z.preprocess(emptyToUndef, z.string().min(1).optional()),
|
||||
/** false = 停用 camofox-env(容器可照常运行,界面标"浏览器不可用") */
|
||||
BROWSER_ENABLED: z.preprocess(
|
||||
emptyToUndef,
|
||||
z
|
||||
.enum(["true", "false", "1", "0"])
|
||||
.default("true")
|
||||
.transform((v) => v === "true" || v === "1"),
|
||||
),
|
||||
});
|
||||
export type Config = z.infer<typeof ConfigSchema>;
|
||||
|
||||
|
||||
@@ -237,6 +237,11 @@ if (role === "api" || role === "all") {
|
||||
ac.abort();
|
||||
await server.finished;
|
||||
});
|
||||
if (role === "api") {
|
||||
// 纯 api 进程不走 worker 的模块 stop:这里停 system 模块的心跳写入器,把本进程的心跳行标 stopped_at
|
||||
const sys = SERVER_MODULES.find((m) => m.name === "system");
|
||||
if (sys?.stop) stops.push(async () => await sys.stop?.());
|
||||
}
|
||||
}
|
||||
if (role === "worker" || role === "worker:agent" || role === "all") {
|
||||
const worker = await startWorker({
|
||||
|
||||
@@ -85,6 +85,13 @@ export function registerBrowserRoutes(app: ModuleApp, svc: () => BrowserService)
|
||||
const B = Api.browser;
|
||||
|
||||
route(app, B.providers, async ({ query }) => ({ items: await svc().listProviders(query.probe === "true") }));
|
||||
route(app, B.createProvider, async ({ body }) => svc().createProvider(body));
|
||||
route(app, B.updateProvider, async ({ params, body }) => svc().updateProvider(params.id, body));
|
||||
route(app, B.probeProvider, async ({ params }) => {
|
||||
const [v] = await svc().listProviders(true, params.id);
|
||||
if (!v) throw new BrowserServiceError("not_found", "浏览器服务不存在");
|
||||
return v;
|
||||
});
|
||||
|
||||
route(app, B.sessions, async ({ query }) => ({
|
||||
items: await svc().list({
|
||||
|
||||
@@ -114,8 +114,13 @@ export interface RecoveryState {
|
||||
downloads?: { sha256: string; artifactId: string; sourceReleaseId: string | null; planId: string | null }[];
|
||||
}
|
||||
|
||||
/** 环境变量(BROWSER_PROVIDER_URL)登记的浏览器服务 */
|
||||
export const ENV_PROVIDER_NAME = "camofox-env";
|
||||
const ENV_MANAGED = "env";
|
||||
|
||||
export class BrowserService {
|
||||
private readonly providers = new Map<string, { sig: string; provider: BrowserProvider }>();
|
||||
private envSync: Promise<void> | undefined;
|
||||
readonly now: () => Date;
|
||||
|
||||
constructor(
|
||||
@@ -163,7 +168,11 @@ export class BrowserService {
|
||||
const { db, secrets } = this.ctx;
|
||||
return {
|
||||
resolve: async (req) => {
|
||||
if (req.kind === "provider_access_key") return row.secret_ref ? secrets.resolve(row.secret_ref) : null;
|
||||
if (req.kind === "provider_access_key") {
|
||||
if (row.secret_ref) return secrets.resolve(row.secret_ref);
|
||||
// 环境变量登记的服务:凭据只在进程内(BROWSER_ACCESS_KEY),不入库
|
||||
return obj(row.config).managedBy === ENV_MANAGED ? (this.ctx.config?.BROWSER_ACCESS_KEY ?? null) : null;
|
||||
}
|
||||
let q = db
|
||||
.selectFrom("site_account")
|
||||
.selectAll()
|
||||
@@ -210,7 +219,126 @@ export class BrowserService {
|
||||
return provider;
|
||||
}
|
||||
|
||||
/**
|
||||
* BROWSER_PROVIDER_URL / BROWSER_ACCESS_KEY / BROWSER_ENABLED → 名为 camofox-env 的浏览器服务(每个进程同步一次)。
|
||||
* 地址与启用由环境变量决定;config 里的其它键(verified、超时等)保留。没设地址时已有的 camofox-env 置为停用。
|
||||
*/
|
||||
syncEnvProvider(): Promise<void> {
|
||||
this.envSync ??= this.doSyncEnvProvider().catch((e) => {
|
||||
this.envSync = undefined;
|
||||
this.ctx.logger.warn({ component: "browser", err: String(e) }, "登记环境变量浏览器服务失败");
|
||||
});
|
||||
return this.envSync;
|
||||
}
|
||||
|
||||
private async doSyncEnvProvider(): Promise<void> {
|
||||
const c = this.ctx.config;
|
||||
const url = c?.BROWSER_PROVIDER_URL;
|
||||
const enabled = !!url && c.BROWSER_ENABLED !== false;
|
||||
const db = this.ctx.db;
|
||||
const row = await db
|
||||
.selectFrom("browser_provider")
|
||||
.selectAll()
|
||||
.where("name", "=", ENV_PROVIDER_NAME)
|
||||
.executeTakeFirst();
|
||||
if (!url) {
|
||||
if (row?.enabled)
|
||||
await db.updateTable("browser_provider").set({ enabled: false }).where("id", "=", row.id).execute();
|
||||
return;
|
||||
}
|
||||
const config = JSON.stringify({ ...obj(row?.config), managedBy: ENV_MANAGED });
|
||||
if (row)
|
||||
await db
|
||||
.updateTable("browser_provider")
|
||||
.set({ base_url: url, enabled, kind: "camofox", secret_ref: null, config })
|
||||
.where("id", "=", row.id)
|
||||
.execute();
|
||||
else
|
||||
await db
|
||||
.insertInto("browser_provider")
|
||||
.values({ name: ENV_PROVIDER_NAME, kind: "camofox", base_url: url, enabled, config })
|
||||
.onConflict((oc) => oc.column("name").doNothing())
|
||||
.execute();
|
||||
}
|
||||
|
||||
/** 设置页登记浏览器服务:访问凭据加密存 secret;名称 camofox-env 保留给环境变量 */
|
||||
async createProvider(body: {
|
||||
name: string;
|
||||
baseUrl: string;
|
||||
accessKey?: string | undefined;
|
||||
enabled?: boolean | undefined;
|
||||
}): Promise<BrowserProviderView> {
|
||||
if (body.name === ENV_PROVIDER_NAME)
|
||||
throw new BrowserServiceError("bad_request", `名称 ${ENV_PROVIDER_NAME} 保留给环境变量`);
|
||||
const exists = await this.ctx.db
|
||||
.selectFrom("browser_provider")
|
||||
.select("id")
|
||||
.where("name", "=", body.name)
|
||||
.executeTakeFirst();
|
||||
if (exists) throw new BrowserServiceError("conflict", `浏览器服务「${body.name}」已存在`);
|
||||
const secretRef = body.accessKey
|
||||
? await this.ctx.secrets.create({ purpose: "other", plaintext: body.accessKey, label: `browser:${body.name}` })
|
||||
: null;
|
||||
const row = await this.ctx.db
|
||||
.insertInto("browser_provider")
|
||||
.values({
|
||||
name: body.name,
|
||||
kind: "camofox",
|
||||
base_url: body.baseUrl,
|
||||
secret_ref: secretRef,
|
||||
enabled: body.enabled ?? true,
|
||||
})
|
||||
.returning("id")
|
||||
.executeTakeFirstOrThrow();
|
||||
return this.providerView(row.id);
|
||||
}
|
||||
|
||||
/** 改地址 / 凭据 / 启用;accessKey = null 清除凭据。camofox-env 的地址与凭据归环境变量管,只能改启用 */
|
||||
async updateProvider(
|
||||
id: string,
|
||||
body: { baseUrl?: string | undefined; accessKey?: string | null | undefined; enabled?: boolean | undefined },
|
||||
): Promise<BrowserProviderView> {
|
||||
const row = await this.ctx.db.selectFrom("browser_provider").selectAll().where("id", "=", id).executeTakeFirst();
|
||||
if (!row) throw new BrowserServiceError("not_found", "浏览器服务不存在");
|
||||
if (obj(row.config).managedBy === ENV_MANAGED && (body.baseUrl !== undefined || body.accessKey !== undefined))
|
||||
throw new BrowserServiceError(
|
||||
"conflict",
|
||||
"该浏览器服务由环境变量 BROWSER_PROVIDER_URL / BROWSER_ACCESS_KEY 管理,改部署配置后重启",
|
||||
);
|
||||
let secretRef = row.secret_ref;
|
||||
if (body.accessKey === null) secretRef = null;
|
||||
else if (body.accessKey) {
|
||||
if (secretRef) await this.ctx.secrets.replace(secretRef, body.accessKey);
|
||||
else
|
||||
secretRef = await this.ctx.secrets.create({
|
||||
purpose: "other",
|
||||
plaintext: body.accessKey,
|
||||
label: `browser:${row.name}`,
|
||||
});
|
||||
}
|
||||
await this.ctx.db
|
||||
.updateTable("browser_provider")
|
||||
.set({
|
||||
...(body.baseUrl !== undefined ? { base_url: body.baseUrl } : {}),
|
||||
...(body.enabled !== undefined ? { enabled: body.enabled } : {}),
|
||||
secret_ref: secretRef,
|
||||
})
|
||||
.where("id", "=", id)
|
||||
.execute();
|
||||
if (row.secret_ref && secretRef === null) await this.ctx.secrets.remove(row.secret_ref).catch(() => {});
|
||||
// 凭据换了但 secret_ref 不变:签名不变,显式丢弃缓存的适配器
|
||||
this.providers.delete(id);
|
||||
return this.providerView(id);
|
||||
}
|
||||
|
||||
private async providerView(id: string): Promise<BrowserProviderView> {
|
||||
const v = (await this.listProviders(false, id))[0];
|
||||
if (!v) throw new BrowserServiceError("not_found", "浏览器服务不存在");
|
||||
return v;
|
||||
}
|
||||
|
||||
async providerRow(id?: string): Promise<BrowserProviderRow> {
|
||||
await this.syncEnvProvider();
|
||||
let q = this.ctx.db
|
||||
.selectFrom("browser_provider")
|
||||
.selectAll()
|
||||
@@ -223,12 +351,14 @@ export class BrowserService {
|
||||
return row;
|
||||
}
|
||||
|
||||
async listProviders(probe: boolean): Promise<BrowserProviderView[]> {
|
||||
const rows = await this.ctx.db.selectFrom("browser_provider").selectAll().orderBy("created_at").execute();
|
||||
async listProviders(probe: boolean, onlyId?: string): Promise<BrowserProviderView[]> {
|
||||
await this.syncEnvProvider();
|
||||
let rows = await this.ctx.db.selectFrom("browser_provider").selectAll().orderBy("created_at").execute();
|
||||
if (onlyId) rows = rows.filter((r) => r.id === onlyId);
|
||||
const out: BrowserProviderView[] = [];
|
||||
for (let row of rows) {
|
||||
let health: BrowserHealth | null = null;
|
||||
if (probe && row.enabled && row.kind !== "none") {
|
||||
if (probe && (row.enabled || onlyId) && row.kind !== "none") {
|
||||
const p = this.providerFor(row);
|
||||
health = await p.health();
|
||||
let caps: BrowserCapabilities | null = null;
|
||||
@@ -255,6 +385,9 @@ export class BrowserService {
|
||||
kind: row.kind as "camofox" | "none",
|
||||
baseUrl: row.base_url,
|
||||
secretRef: row.secret_ref,
|
||||
managedBy: obj(row.config).managedBy === ENV_MANAGED ? "env" : null,
|
||||
hasAccessKey:
|
||||
!!row.secret_ref || (obj(row.config).managedBy === ENV_MANAGED && !!this.ctx.config?.BROWSER_ACCESS_KEY),
|
||||
enabled: row.enabled,
|
||||
capabilities: Object.keys(caps).length ? (caps as unknown as BrowserCapabilities) : null,
|
||||
verifiedAt: iso(row.verified_at),
|
||||
|
||||
@@ -8,8 +8,8 @@ import type {
|
||||
SystemMount,
|
||||
SystemProcess,
|
||||
} from "@umt/contracts";
|
||||
import { type DB, type Kysely, sql } from "@umt/db";
|
||||
import { HEARTBEAT_STALE_MS } from "./heartbeat.ts";
|
||||
import { type DB, type Kysely, type Selectable, sql } from "@umt/db";
|
||||
import { CURRENT_HOST_WINDOW_MS, HEARTBEAT_ROW_RETENTION_MS, HEARTBEAT_STALE_MS } from "./heartbeat.ts";
|
||||
|
||||
// GET /system/health 各段的取数(W9-I)。每段独立 try/catch:一段失败只把该段标 unknown,不让整页 500。
|
||||
// 慢 IO(statfs / 哨兵检查)都带超时:挂载卡死时只标 timeout,不拖住请求。
|
||||
@@ -53,44 +53,84 @@ const num = (v: unknown) => (v == null ? null : Number(v));
|
||||
|
||||
// ── 进程 ──
|
||||
|
||||
export async function loadProcesses(db: Kysely<DB>, now: Date): Promise<SystemProcess[]> {
|
||||
type HeartbeatRow = Selectable<DB["process_heartbeat"]>;
|
||||
|
||||
function toProcess(r: HeartbeatRow, now: Date): SystemProcess {
|
||||
const beat = toDate(r.last_beat_at);
|
||||
const age = Math.max(0, now.getTime() - beat.getTime());
|
||||
const extra = (r.extra ?? {}) as Record<string, unknown>;
|
||||
const conc = (r.concurrency ?? {}) as { io?: number; agent?: number };
|
||||
const sse = extra.sseConnections;
|
||||
return {
|
||||
instanceId: r.instance_id,
|
||||
processKind: r.process_kind,
|
||||
pid: r.pid,
|
||||
host: r.host,
|
||||
version: r.version,
|
||||
startedAt: toDate(r.started_at).toISOString(),
|
||||
lastBeatAt: beat.toISOString(),
|
||||
beatAgeSec: Math.round(age / 100) / 10,
|
||||
stale: age > HEARTBEAT_STALE_MS,
|
||||
stoppedAt: r.stopped_at ? toDate(r.stopped_at).toISOString() : null,
|
||||
eventLoopLagMs: r.event_loop_lag_ms,
|
||||
eventLoopLagMaxMs: r.event_loop_lag_max_ms,
|
||||
concurrency: {
|
||||
...(typeof conc.io === "number" ? { io: conc.io } : {}),
|
||||
...(typeof conc.agent === "number" ? { agent: conc.agent } : {}),
|
||||
},
|
||||
runningRuns: r.running_runs,
|
||||
sseConnections: typeof sse === "number" ? sse : null,
|
||||
extra,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 把心跳行分成当前实例与已下线实例:
|
||||
* - 当前 host = 最近 10 分钟内有心跳且没标 stopped_at 的行所在 host(容器重建后 host 会变);
|
||||
* - 当前实例 = 当前 host 上、没标 stopped_at、且是同 host 同 kind 里最新启动的那一行;
|
||||
* - 其余(已正常退出、host 已不在、被同 host 新进程替换)都算已下线实例,不参与 stale 判定与告警。
|
||||
*/
|
||||
export function splitProcesses(
|
||||
ps: SystemProcess[],
|
||||
now: Date,
|
||||
): { processes: SystemProcess[]; retiredProcesses: SystemProcess[] } {
|
||||
const alive = ps.filter((p) => !p.stoppedAt);
|
||||
const hosts = new Set(
|
||||
alive.filter((p) => now.getTime() - Date.parse(p.lastBeatAt) <= CURRENT_HOST_WINDOW_MS).map((p) => p.host),
|
||||
);
|
||||
const newest = new Map<string, SystemProcess>();
|
||||
for (const p of alive) {
|
||||
if (!hosts.has(p.host)) continue;
|
||||
const k = `${p.host}\u0000${p.processKind}`;
|
||||
const cur = newest.get(k);
|
||||
if (!cur || Date.parse(p.startedAt) > Date.parse(cur.startedAt)) newest.set(k, p);
|
||||
}
|
||||
const current = new Set([...newest.values()].map((p) => p.instanceId));
|
||||
const processes = ps.filter((p) => current.has(p.instanceId));
|
||||
const retiredProcesses = ps
|
||||
.filter((p) => !current.has(p.instanceId))
|
||||
.sort((a, b) => Date.parse(b.lastBeatAt) - Date.parse(a.lastBeatAt));
|
||||
return { processes, retiredProcesses };
|
||||
}
|
||||
|
||||
export async function loadProcesses(
|
||||
db: Kysely<DB>,
|
||||
now: Date,
|
||||
): Promise<{ processes: SystemProcess[]; retiredProcesses: SystemProcess[] }> {
|
||||
const rows = await db
|
||||
.selectFrom("process_heartbeat")
|
||||
.selectAll()
|
||||
.where("last_beat_at", ">", new Date(now.getTime() - 24 * 3600_000))
|
||||
.where("last_beat_at", ">", new Date(now.getTime() - HEARTBEAT_ROW_RETENTION_MS))
|
||||
.orderBy("process_kind")
|
||||
.orderBy("instance_id")
|
||||
.execute();
|
||||
return rows.map((r) => {
|
||||
const beat = toDate(r.last_beat_at);
|
||||
const age = Math.max(0, now.getTime() - beat.getTime());
|
||||
const extra = (r.extra ?? {}) as Record<string, unknown>;
|
||||
const conc = (r.concurrency ?? {}) as { io?: number; agent?: number };
|
||||
const sse = extra.sseConnections;
|
||||
return {
|
||||
instanceId: r.instance_id,
|
||||
processKind: r.process_kind,
|
||||
pid: r.pid,
|
||||
host: r.host,
|
||||
version: r.version,
|
||||
startedAt: toDate(r.started_at).toISOString(),
|
||||
lastBeatAt: beat.toISOString(),
|
||||
beatAgeSec: Math.round(age / 100) / 10,
|
||||
stale: age > HEARTBEAT_STALE_MS,
|
||||
eventLoopLagMs: r.event_loop_lag_ms,
|
||||
eventLoopLagMaxMs: r.event_loop_lag_max_ms,
|
||||
concurrency: {
|
||||
...(typeof conc.io === "number" ? { io: conc.io } : {}),
|
||||
...(typeof conc.agent === "number" ? { agent: conc.agent } : {}),
|
||||
},
|
||||
runningRuns: r.running_runs,
|
||||
sseConnections: typeof sse === "number" ? sse : null,
|
||||
extra,
|
||||
};
|
||||
});
|
||||
return splitProcesses(
|
||||
rows.map((r) => toProcess(r, now)),
|
||||
now,
|
||||
);
|
||||
}
|
||||
|
||||
/** 进程段的档位:有 stale 或延迟 ≥ 5s 为 critical;延迟 ≥ 1s 为 warn;没有任何 worker 心跳为 warn */
|
||||
/** 进程段的档位(只看当前实例):有 stale 或延迟 ≥ 1s 为 warn;延迟 ≥ 5s 为 critical;没有任何 worker 心跳为 warn */
|
||||
export function processLevel(ps: SystemProcess[]): HealthLevel {
|
||||
const live = ps.filter((p) => !p.stale);
|
||||
const levels: HealthLevel[] = [];
|
||||
@@ -344,8 +384,11 @@ export async function loadSchedules(db: Kysely<DB>, now: Date): Promise<SystemHe
|
||||
|
||||
export async function buildHealth(d: HealthDeps): Promise<SystemHealth> {
|
||||
const now = d.now?.() ?? new Date();
|
||||
const [processes, db, paths, engine, schedules] = await Promise.all([
|
||||
loadProcesses(d.db, now).catch(() => [] as SystemProcess[]),
|
||||
const [{ processes, retiredProcesses }, db, paths, engine, schedules] = await Promise.all([
|
||||
loadProcesses(d.db, now).catch(() => ({
|
||||
processes: [] as SystemProcess[],
|
||||
retiredProcesses: [] as SystemProcess[],
|
||||
})),
|
||||
loadDb(d.db),
|
||||
loadDisksAndMounts(d).catch(() => ({ disks: [] as SystemDisk[], mounts: [] as SystemMount[] })),
|
||||
loadEngine(d.db, d.engineSchema, now).catch((): SystemHealth["engine"] => ({
|
||||
@@ -388,6 +431,7 @@ export async function buildHealth(d: HealthDeps): Promise<SystemHealth> {
|
||||
uptimeSec: Math.max(0, Math.round((now.getTime() - d.self.startedAt.getTime()) / 1000)),
|
||||
},
|
||||
processes,
|
||||
retiredProcesses,
|
||||
sse: { connections: sse },
|
||||
db,
|
||||
disks: paths.disks,
|
||||
|
||||
@@ -6,12 +6,17 @@ import serverDeno from "../../../deno.json" with { type: "json" };
|
||||
|
||||
// 进程心跳写入(W9-I):每个常驻进程每 15s upsert process_heartbeat 一行 + 追加一条采样。
|
||||
// 写失败只记 debug(库暂时不可达不影响业务;读取方按 last_beat_at 判 stale)。
|
||||
// 正常退出(stop)时把本行标 stopped_at,读取方立即把它归入已下线实例;裁剪由定时任务 system.heartbeat_cleanup 负责。
|
||||
|
||||
export const HEARTBEAT_WRITE_MS = 15_000;
|
||||
/** 超过这个时长没有心跳即 stale(3 个写周期) */
|
||||
export const HEARTBEAT_STALE_MS = 45_000;
|
||||
/** 采样保留天数 */
|
||||
export const SAMPLE_RETENTION_DAYS = 7;
|
||||
/** 进程行保留:超过这个时长没心跳的行删除 */
|
||||
export const HEARTBEAT_ROW_RETENTION_MS = 24 * 3600_000;
|
||||
/** 当前实例判定窗口:最近这段时间内有心跳的 host 视为当前在跑的容器 */
|
||||
export const CURRENT_HOST_WINDOW_MS = 10 * 60_000;
|
||||
|
||||
export const SERVER_VERSION: string = serverDeno.version;
|
||||
|
||||
@@ -109,7 +114,7 @@ export async function writeHeartbeat(
|
||||
host = excluded.host, version = excluded.version, started_at = excluded.started_at,
|
||||
last_beat_at = excluded.last_beat_at, event_loop_lag_ms = excluded.event_loop_lag_ms,
|
||||
event_loop_lag_max_ms = excluded.event_loop_lag_max_ms, concurrency = excluded.concurrency,
|
||||
running_runs = excluded.running_runs, extra = excluded.extra`.execute(db);
|
||||
running_runs = excluded.running_runs, extra = excluded.extra, stopped_at = null`.execute(db);
|
||||
await sql`
|
||||
insert into process_heartbeat_sample (instance_id, process_kind, at, event_loop_lag_ms, event_loop_lag_max_ms,
|
||||
running_runs, sse_connections)
|
||||
@@ -117,10 +122,22 @@ export async function writeHeartbeat(
|
||||
${s.sseConnections ?? null})`.execute(db);
|
||||
}
|
||||
|
||||
/** 裁剪过期采样与 7 天没心跳的进程行 */
|
||||
export async function pruneHeartbeats(db: Kysely<DB>, days = SAMPLE_RETENTION_DAYS): Promise<void> {
|
||||
await sql`delete from process_heartbeat_sample where at < now() - make_interval(days => ${days})`.execute(db);
|
||||
await sql`delete from process_heartbeat where last_beat_at < now() - make_interval(days => ${days})`.execute(db);
|
||||
/** 正常退出:把本实例的行标 stopped_at(之后读取方不再当活进程) */
|
||||
export async function markHeartbeatStopped(db: Kysely<DB>, instanceId: string, at = new Date()): Promise<void> {
|
||||
await sql`update process_heartbeat set stopped_at = ${at} where instance_id = ${instanceId}`.execute(db);
|
||||
}
|
||||
|
||||
/** 裁剪:超过 24 小时没心跳的进程行、超过 7 天的采样;返回删除行数 */
|
||||
export async function pruneHeartbeats(
|
||||
db: Kysely<DB>,
|
||||
o: { now?: Date; rowRetentionMs?: number; sampleDays?: number } = {},
|
||||
): Promise<{ rows: number; samples: number }> {
|
||||
const now = o.now ?? new Date();
|
||||
const rowCutoff = new Date(now.getTime() - (o.rowRetentionMs ?? HEARTBEAT_ROW_RETENTION_MS));
|
||||
const sampleCutoff = new Date(now.getTime() - (o.sampleDays ?? SAMPLE_RETENTION_DAYS) * 24 * 3600_000);
|
||||
const rows = await db.deleteFrom("process_heartbeat").where("last_beat_at", "<", rowCutoff).executeTakeFirst();
|
||||
const samples = await db.deleteFrom("process_heartbeat_sample").where("at", "<", sampleCutoff).executeTakeFirst();
|
||||
return { rows: Number(rows.numDeletedRows), samples: Number(samples.numDeletedRows) };
|
||||
}
|
||||
|
||||
export interface HeartbeatWriter {
|
||||
@@ -130,7 +147,7 @@ export interface HeartbeatWriter {
|
||||
stop(): Promise<void>;
|
||||
}
|
||||
|
||||
/** 起一个周期写入器:立即写一次,之后每 intervalMs 写一次;每约 1 小时裁剪一次 */
|
||||
/** 起一个周期写入器:立即写一次,之后每 intervalMs 写一次;stop 时把本行标 stopped_at */
|
||||
export function startHeartbeatWriter(o: HeartbeatWriterOptions): HeartbeatWriter {
|
||||
const host = o.host ?? safeHostname();
|
||||
const pid = o.pid ?? Deno.pid;
|
||||
@@ -140,8 +157,6 @@ export function startHeartbeatWriter(o: HeartbeatWriterOptions): HeartbeatWriter
|
||||
const intervalMs = o.intervalMs ?? HEARTBEAT_WRITE_MS;
|
||||
const lag = o.lag ?? createLagSampler();
|
||||
const log = o.logger.child({ component: "heartbeat", kind: o.kind });
|
||||
const pruneEvery = Math.max(1, Math.round(3_600_000 / intervalMs));
|
||||
let n = 0;
|
||||
let inflight: Promise<void> | null = null;
|
||||
let stopped = false;
|
||||
|
||||
@@ -165,8 +180,6 @@ export function startHeartbeatWriter(o: HeartbeatWriterOptions): HeartbeatWriter
|
||||
lagMaxMs: l.windowMaxMs,
|
||||
sample: { ...sample, extra: { ...memory(), lifetimeMaxLagMs: l.lifetimeMaxMs, ...(sample.extra ?? {}) } },
|
||||
});
|
||||
if (n % pruneEvery === 0) await pruneHeartbeats(o.db);
|
||||
n += 1;
|
||||
} catch (err) {
|
||||
log.debug({ err }, "心跳写入失败");
|
||||
}
|
||||
@@ -194,6 +207,11 @@ export function startHeartbeatWriter(o: HeartbeatWriterOptions): HeartbeatWriter
|
||||
clearInterval(timer);
|
||||
lag.stop();
|
||||
await inflight?.catch(() => {});
|
||||
try {
|
||||
await markHeartbeatStopped(o.db, instanceId);
|
||||
} catch (err) {
|
||||
log.debug({ err }, "心跳退出标记失败");
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -4,9 +4,17 @@ import type { z } from "zod";
|
||||
import { dataPaths } from "../../config.ts";
|
||||
import { requireScope } from "../../http/auth.ts";
|
||||
import { sseConnectionCount } from "../../http/sse.ts";
|
||||
import { registerScheduledJob } from "../schedules/registry.ts";
|
||||
import type { ModuleApp, ServerContext, ServerModule } from "../types.ts";
|
||||
import { buildHealth, type HealthDeps, loadHistory } from "./health.ts";
|
||||
import { type HeartbeatWriter, instanceIdOf, SERVER_VERSION, safeHostname, startHeartbeatWriter } from "./heartbeat.ts";
|
||||
import {
|
||||
type HeartbeatWriter,
|
||||
instanceIdOf,
|
||||
pruneHeartbeats,
|
||||
SERVER_VERSION,
|
||||
safeHostname,
|
||||
startHeartbeatWriter,
|
||||
} from "./heartbeat.ts";
|
||||
|
||||
export { HEARTBEAT_STALE_MS, HEARTBEAT_WRITE_MS, startHeartbeatWriter } from "./heartbeat.ts";
|
||||
|
||||
@@ -15,6 +23,7 @@ export { HEARTBEAT_STALE_MS, HEARTBEAT_WRITE_MS, startHeartbeatWriter } from "./
|
||||
* - GET /system/health:进程心跳 / 事件循环延迟 / SSE 连接 / DB / 磁盘 / 挂载根 / 引擎 / 定时任务,仅 admin;
|
||||
* - GET /system/health/history?hours=:心跳采样时间序列。
|
||||
* 心跳写入:worker 进程在 worker/index.ts 里起写入器;纯 api 进程(role=api)在注册路由时起(role=all 由 worker 写 kind=all)。
|
||||
* 清理:定时任务 system.heartbeat_cleanup(维护组,1 小时)删 24 小时没心跳的进程行与 7 天前的采样。
|
||||
*/
|
||||
export interface SystemModuleOptions {
|
||||
/** 引擎 schema(测试覆盖);缺省 openworkflow */
|
||||
@@ -23,6 +32,8 @@ export interface SystemModuleOptions {
|
||||
healthDeps?: Partial<Pick<HealthDeps, "diskUsage" | "checkRoot" | "now" | "ioTimeoutMs" | "dataDir">>;
|
||||
/** 缺省 true;false 时 role=api 也不起心跳写入器(测试) */
|
||||
apiHeartbeat?: boolean;
|
||||
/** 定时任务 system.heartbeat_cleanup 的默认周期,默认 1 小时;<=0 不登记 */
|
||||
cleanupIntervalMs?: number;
|
||||
}
|
||||
|
||||
type Infer<T> = T extends z.ZodType ? z.infer<T> : Record<string, never>;
|
||||
@@ -93,6 +104,23 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule
|
||||
route(app, Api.system.healthHistory, async ({ query }) => await loadHistory(ctx.db, query.hours));
|
||||
},
|
||||
registerTasks() {},
|
||||
start(ctx) {
|
||||
const everyMs = opts.cleanupIntervalMs ?? 3600_000;
|
||||
if (everyMs <= 0) return;
|
||||
registerScheduledJob(ctx, {
|
||||
key: "system.heartbeat_cleanup",
|
||||
defaults: {
|
||||
name: "进程心跳清理",
|
||||
description: "删除 24 小时没有心跳的进程行和 7 天前的心跳采样",
|
||||
group: "maintain",
|
||||
schedule: { type: "interval", everyMs },
|
||||
},
|
||||
run: async ({ now }) => {
|
||||
const r = await pruneHeartbeats(ctx.db, { now });
|
||||
return { summary: `删除 ${r.rows} 个进程行、${r.samples} 条采样` };
|
||||
},
|
||||
});
|
||||
},
|
||||
async stop() {
|
||||
await writer?.stop();
|
||||
writer = null;
|
||||
|
||||
Reference in new issue
Block a user