feat(system): process heartbeat + /system/health, engine runs business fields (W9-I)

- process_heartbeat / process_heartbeat_sample (0034): api / worker / worker:agent / all
  upsert every 15s (event loop lag last/max, concurrency, running runs, SSE connections, memory)
- GET /system/health (admin): processes with 45s stale flag, SSE, DB latency/size/connections,
  disks (DATA_DIR / library roots / download dirs, cache thresholds), mount sentinel checks
  with IO timeout, engine pending/oldest/24h failure rate, schedules
- GET /system/health/history?hours=: bucketed lag / running / SSE series
- /engine/runs and /engine/runs/{id}: business {kind, refId, title, triggeredBy} joined from
  run input / task / agent_run / tasks_run_meta / outbox; businessKind / businessRefId filters
- docs/runbook/ops.md: field meanings and alert thresholds

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-28 15:23:37 +08:00
1 parent 46d37d87fe
commit e8745115f4
17 files changed
+1789 -4

No files matched your search

@@ -0,0 +1,246 @@
import type { EngineBusinessKind, EngineRunBusiness, EngineRunQuery, EngineTriggeredBy } from "@umt/contracts";
import { type DB, type Kysely, sql } from "@umt/db";
import type { RawBuilder } from "kysely";
// 引擎 runs 的业务字段(W9-I):run → 关联业务对象 {kind, refId, title, triggeredBy}。
// kind/refId 用一条 SQL 表达式推断(列表筛选与展示共用,保证一致),按优先级取第一个命中:
// 1. run 输入(信封 input.input)里的业务 id 键:agentRunId / recordId / downloadTaskId / acquisitionPlanId /
// subscriptionId / acquisitionSlotId / feedSourceId / siteId / accountId / decisionId / mediaEntityId
// 2. 业务任务行 task(engine_run_id = run.id)的 subject_type / subject_id
// 3. 无业务 id 的定时任务(purgeTrash 等,非手动触发)→ schedule + 定时任务 key
// title 与 triggeredBy 在 TS 里按页批量查(agent_run / task / tasks_run_meta / outbox_event / 各业务表)。
// 表达式引用外层 "<engine schema>".workflow_runs 的列,必须在以 workflow_runs 为 FROM 的查询里使用。
const I = sql.raw(`(workflow_runs.input->'input')`);
const RUN_ID = sql.raw(`workflow_runs.id`);
const RUN_NAME = sql.raw(`workflow_runs.workflow_name`);
/** 输入里的文本值转 uuid(不是 uuid 时为 null,避免 cast 报错) */
const uuidOf = (key: string) =>
sql`(case when (${I}->>${key}) ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then (${I}->>${key})::uuid end)`;
const obj = (kind: EngineBusinessKind, ref: RawBuilder<unknown>) =>
sql`jsonb_build_object('kind', ${sql.lit(kind)}::text, 'refId', (${ref})::text)`;
const whenKey = (key: string, kind: EngineBusinessKind, ref: RawBuilder<unknown> = sql`${I}->>${key}`) =>
sql`(case when jsonb_typeof(${I}) = 'object' and ${I} ? ${key} and (${ref}) is not null then ${obj(kind, ref)} end)`;
/** 下载:优先指向下载任务(plan 已提交时),否则指向收录计划 id */
const planRef = sql`coalesce((select dt.id from download_task dt where dt.acquisition_plan_id = ${uuidOf("acquisitionPlanId")} order by dt.created_at limit 1), ${uuidOf("acquisitionPlanId")})`;
const slotSubRef = sql`(select s.subscription_id from acquisition_slot s where s.id = ${uuidOf("acquisitionSlotId")})`;
const accountSiteRef = sql`(select a.site_id from site_account a where a.id = ${uuidOf("accountId")})`;
/** 无业务 id 的任务 → 定时任务 key(只在非手动 / 非重跑时认作定时触发) */
export const SCHEDULE_TASK_JOBS: Record<string, string> = {
purgeTrash: "trash.purge",
refreshPiTokens: "pi.token_refresh",
cleanupDownloadCache: "cache.threshold",
reconcileLibrary: "library.reconcile",
};
const taskSubject = sql`(select case t.subject_type
when 'agent_run' then jsonb_build_object('kind', 'agent', 'refId', t.subject_id::text)
when 'organization_record' then jsonb_build_object('kind', 'organization', 'refId', t.subject_id::text)
when 'download_task' then jsonb_build_object('kind', 'download', 'refId', t.subject_id::text)
when 'acquisition_plan' then jsonb_build_object('kind', 'download', 'refId',
coalesce((select dt.id::text from download_task dt where dt.acquisition_plan_id::text = t.subject_id::text
order by dt.created_at limit 1), t.subject_id::text))
when 'acquisition_slot' then (select jsonb_build_object('kind', 'subscription', 'refId', s.subscription_id::text)
from acquisition_slot s where s.id::text = t.subject_id::text and s.subscription_id is not null)
when 'media_entity' then jsonb_build_object('kind', 'media', 'refId', t.subject_id::text)
end
from task t where t.engine_run_id = ${RUN_ID} and t.subject_id is not null order by t.created_at desc limit 1)`;
const manualKey = sql`coalesce((select m.idempotency_key ~ '^retry:|(^|:)manual(:|$)' from tasks_run_meta m where m.run_id = ${RUN_ID}), false)`;
const scheduleBranch = sql`(case when ${RUN_NAME} in (${sql.join(
Object.keys(SCHEDULE_TASK_JOBS).map((k) => sql.lit(k)),
)}) and not ${manualKey} then jsonb_build_object('kind', 'schedule', 'refId', case ${RUN_NAME} ${sql.join(
Object.entries(SCHEDULE_TASK_JOBS).map(([t, j]) => sql`when ${sql.lit(t)} then ${sql.lit(j)}`),
sql` `,
)} end) end)`;
/** run → {kind, refId} 的 jsonb 表达式(可能为 null) */
export const BUSINESS_EXPR = sql<{ kind: EngineBusinessKind; refId: string } | null>`coalesce(
${whenKey("agentRunId", "agent")},
${whenKey("recordId", "organization")},
${whenKey("downloadTaskId", "download")},
${whenKey("acquisitionPlanId", "download", planRef)},
${whenKey("subscriptionId", "subscription")},
${whenKey("acquisitionSlotId", "subscription", slotSubRef)},
${whenKey("feedSourceId", "feed")},
${whenKey("siteId", "site")},
${whenKey("accountId", "site", accountSiteRef)},
${whenKey("decisionId", "decision")},
${whenKey("mediaEntityId", "media")},
${taskSubject},
${scheduleBranch}
)`;
/** EngineRunQuery 的业务筛选条件(加到 workflow_runs 查询的 where) */
export function businessConds(q: Pick<EngineRunQuery, "businessKind" | "businessRefId">): RawBuilder<unknown>[] {
const out: RawBuilder<unknown>[] = [];
if (q.businessKind) out.push(sql`(${BUSINESS_EXPR})->>'kind' = ${q.businessKind}`);
if (q.businessRefId) out.push(sql`(${BUSINESS_EXPR})->>'refId' = ${q.businessRefId}`);
return out;
}
// ── triggeredBy ──
/** task.triggered_by / agent_run.created_by 这类 "<kind>:<id>" 串 → 触发来源;认不出返回 null */
export function triggeredByFromActor(s: string | null | undefined): EngineTriggeredBy | null {
if (!s) return null;
const head = s.split(":")[0]!.toLowerCase();
if (head === "api_key") return "api_key";
if (["admin_session", "dev_admin", "user", "admin", "console"].includes(head)) return "user";
if (head === "agent" || head === "subagent") return "agent";
if (head === "schedule" || head === "scheduler") return "schedule";
if (head === "event") return "event";
return null;
}
/** 调度器按时间槽入队的幂等键前缀(见各模块 scheduler / start) */
const SCHEDULE_KEY =
/^(snapshotSite:(?!manual:)|refreshSiteAccount:(relogin|userData):|slot-due:|refreshMetadata:tracking:|refreshMetadata:subscription:|search-eval:|refreshPiTokens:|purgeTrash:|reconcileLibrary:auto:|cleanupDownloadCache:threshold:|auto:|threshold:)/;
export interface TriggerFacts {
kind: EngineBusinessKind;
idempotencyKey: string | null;
taskTriggeredBy: string | null;
agentTrigger: string | null;
agentCreatedBy: string | null;
fromOutbox: boolean;
}
/** 触发来源推断(纯函数,便于测试):agent_run.trigger → task.triggered_by → 幂等键(手动/重跑)→ outbox → 调度槽 */
export function inferTriggeredBy(f: TriggerFacts): EngineTriggeredBy | null {
if (f.agentTrigger) {
if (f.agentTrigger === "subagent") return "agent";
if (f.agentTrigger === "schedule") return "schedule";
if (f.agentTrigger === "event") return "event";
if (f.agentTrigger === "user") return triggeredByFromActor(f.agentCreatedBy) === "api_key" ? "api_key" : "user";
}
const byTask = triggeredByFromActor(f.taskTriggeredBy);
if (byTask) return byTask;
const key = f.idempotencyKey ?? "";
if (/^retry:|(^|:)manual(:|$)/.test(key)) return "user";
if (f.fromOutbox) return "event";
if (f.kind === "schedule" || SCHEDULE_KEY.test(key)) return "schedule";
return null;
}
// ── 批量补字段 ──
type Row = { id: string };
type Enriched<T> = T & { business: EngineRunBusiness | null };
const TITLE_SQL: Record<EngineBusinessKind, (ids: string[]) => RawBuilder<{ id: string; title: string | null }>> = {
agent: (ids) => sql`select id::text as id, goal as title from agent_run where id::text = any(${ids}::text[])`,
subscription: (ids) => sql`select s.id::text as id, coalesce(e.title, s.title_query) as title
from subscription s left join media_entity e on e.id = s.media_entity_id where s.id::text = any(${ids}::text[])`,
download: (ids) => sql`select id::text as id, name as title from download_task where id::text = any(${ids}::text[])`,
organization: (ids) => sql`select r.id::text as id, coalesce(e.title, r.target_path, r.source_path) as title
from organization_record r left join media_entity e on e.id = r.media_entity_id where r.id::text = any(${ids}::text[])`,
feed: (ids) => sql`select id::text as id, name as title from feed_source where id::text = any(${ids}::text[])`,
site: (ids) => sql`select id::text as id, name as title from site where id::text = any(${ids}::text[])`,
decision: (ids) => sql`select id::text as id, title from decision where id::text = any(${ids}::text[])`,
media: (ids) => sql`select id::text as id, title from media_entity where id::text = any(${ids}::text[])`,
schedule: (ids) => sql`select id, name as title from scheduled_job where id = any(${ids}::text[])`,
};
const tableOf = (schema: string) => {
if (!/^[a-z_][a-z0-9_]*$/.test(schema)) throw new Error(`非法 schema 名 ${schema}`);
return sql.raw(`"${schema}".workflow_runs`);
};
/** 给一页 run 补 business 字段(缺则 null);查询失败时整页降级为 null,不影响原响应 */
export async function attachBusiness<T extends Row>(
db: Kysely<DB>,
schema: string,
items: T[],
): Promise<Enriched<T>[]> {
if (!items.length) return [];
try {
const map = await resolveBusiness(
db,
schema,
items.map((i) => i.id),
);
return items.map((i) => ({ ...i, business: map.get(i.id) ?? null }));
} catch {
return items.map((i) => ({ ...i, business: null }));
}
}
export async function resolveBusiness(
db: Kysely<DB>,
schema: string,
runIds: string[],
): Promise<Map<string, EngineRunBusiness>> {
const out = new Map<string, EngineRunBusiness>();
if (!runIds.length) return out;
const rows = await sql<{
id: string;
b: { kind: EngineBusinessKind; refId: string } | null;
idempotency_key: string | null;
task_triggered_by: string | null;
}>`select workflow_runs.id, ${BUSINESS_EXPR} as b,
(select m.idempotency_key from tasks_run_meta m where m.run_id = workflow_runs.id) as idempotency_key,
(select t.triggered_by from task t where t.engine_run_id = workflow_runs.id order by t.created_at desc limit 1)
as task_triggered_by
from ${tableOf(schema)} where workflow_runs.id = any(${runIds}::text[])`.execute(db);
const found = rows.rows.filter((r) => r.b?.kind && r.b.refId);
if (!found.length) return out;
// outbox 来源:引擎幂等键 = outbox_event.idempotency_key 或 outbox:<event id>
const keys = found.map((r) => r.idempotency_key).filter((k): k is string => !!k);
const outboxKeys = new Set<string>();
if (keys.length) {
const ev = await sql<{ k: string }>`
select idempotency_key as k from outbox_event where idempotency_key = any(${keys}::text[])
union all
select 'outbox:' || id::text as k from outbox_event
where id::text = any(${keys.filter((k) => k.startsWith("outbox:")).map((k) => k.slice(7))}::text[])`.execute(db);
for (const r of ev.rows) outboxKeys.add(r.k);
}
// 标题 + agent 触发信息:按 kind 分组批量查
const byKind = new Map<EngineBusinessKind, Set<string>>();
for (const r of found) {
const set = byKind.get(r.b!.kind) ?? new Set<string>();
set.add(r.b!.refId);
byKind.set(r.b!.kind, set);
}
const titles = new Map<string, string | null>();
await Promise.all(
[...byKind].map(async ([kind, ids]) => {
const q = TITLE_SQL[kind];
const r = await q([...ids]).execute(db);
for (const t of r.rows) titles.set(`${kind}:${t.id}`, t.title);
}),
);
const agentIds = [...(byKind.get("agent") ?? [])];
const agents = new Map<string, { trigger: string; created_by: string | null }>();
if (agentIds.length) {
const r = await sql<{ id: string; trigger: string; created_by: string | null }>`
select id::text as id, trigger, created_by from agent_run where id::text = any(${agentIds}::text[])`.execute(db);
for (const a of r.rows) agents.set(a.id, a);
}
for (const r of found) {
const b = r.b!;
const agent = b.kind === "agent" ? agents.get(b.refId) : undefined;
out.set(r.id, {
kind: b.kind,
refId: b.refId,
title: titles.get(`${b.kind}:${b.refId}`) ?? null,
triggeredBy: inferTriggeredBy({
kind: b.kind,
idempotencyKey: r.idempotency_key,
taskTriggeredBy: r.task_triggered_by,
agentTrigger: agent?.trigger ?? null,
agentCreatedBy: agent?.created_by ?? null,
fromOutbox: !!r.idempotency_key && outboxKeys.has(r.idempotency_key),
}),
});
}
return out;
}
@@ -13,6 +13,7 @@ import { ApiException } from "../../http/errors.ts";
import { redact } from "../../logger.ts";
import type { ModuleApp, ServerContext } from "../types.ts";
import { audit, type Cursor, type Def, decodeCursor, encodeCursor, endpoint } from "./endpoint.ts";
import { attachBusiness, businessConds } from "./engine-business.ts";
// 引擎面板只读 API(仅 admin):直查 openworkflow schema(docs/design/openworkflow-dashboard-reference.md)。
// 不返回 config/context;input/output/error/signal data 过敏感字段脱敏;取消走 TaskEngine.cancel(先写业务侧取消标志)。
@@ -130,6 +131,7 @@ export function engineQueries(db: Kysely<DB>, schema = "openworkflow"): EngineQu
if (q.workflowName) conds.push(sql`workflow_name = ${q.workflowName}`);
if (q.from) conds.push(sql`created_at >= ${new Date(q.from)}`);
if (q.to) conds.push(sql`created_at < ${new Date(q.to)}`);
conds.push(...businessConds(q));
if (cur) {
conds.push(sql`(created_at, id) < (timestamptz 'epoch' + ${cur.us} * interval '1 microsecond', ${cur.id})`);
}
@@ -246,11 +248,16 @@ export function registerEngineRoutes(app: ModuleApp, ctx: ServerContext, opts: {
q ??= engineQueries(ctx.db, opts.schema);
return q;
};
endpoint(app, D.runs, async ({ query }) => await queries().listRuns(query));
// W9-I:列表与详情补 business 字段(engine-business.ts;查询失败降级为 null)
const schema = opts.schema ?? "openworkflow";
endpoint(app, D.runs, async ({ query }) => {
const page = await queries().listRuns(query);
return { ...page, items: await attachBusiness(ctx.db, schema, page.items) };
});
endpoint(app, D.run, async ({ params }) => {
const r = await queries().getRun(params.id);
if (!r) throw new ApiException("not_found", `run ${params.id} 不存在`);
return r;
return (await attachBusiness(ctx.db, schema, [r]))[0];
});
endpoint(app, D.stats, async () => await queries().stats());
endpoint(app, D.cancel, async ({ params, body, actor }) => {
+2
View File
@@ -6,6 +6,7 @@ import { piModule } from "./pi/index.ts";
import { schedulesModule } from "./schedules/index.ts";
import { searchModule } from "./search/index.ts";
import { sourcesIdentityModule } from "./sources-identity/index.ts";
import { systemModule } from "./system/index.ts";
import type { ServerModule } from "./types.ts";
export type * from "./types.ts";
@@ -23,4 +24,5 @@ export const SERVER_MODULES: readonly ServerModule[] = [
piModule,
schedulesModule,
searchModule,
systemModule,
];
+437
View File
@@ -0,0 +1,437 @@
import { readdir, stat } from "node:fs/promises";
import { checkRoot, DEFAULT_ROOT_SENTINEL, type DiskUsage, diskUsage } from "@umt/adapters";
import type {
HealthLevel,
SystemDisk,
SystemHealth,
SystemHealthHistory,
SystemMount,
SystemProcess,
} from "@umt/contracts";
import { type DB, type Kysely, sql } from "@umt/db";
import { HEARTBEAT_STALE_MS } from "./heartbeat.ts";
// GET /system/health 各段的取数(W9-I)。每段独立 try/catch:一段失败只把该段标 unknown,不让整页 500。
// 慢 IO(statfs / 哨兵检查)都带超时:挂载卡死时只标 timeout,不拖住请求。
export interface HealthDeps {
db: Kysely<DB>;
engineSchema: string;
dataDir: string;
/** 本进程(处理请求的 api)信息 */
self: { version: string; startedAt: Date; sseConnections: number; instanceId?: string };
/** 磁盘阈值(百分比):warn / critical */
thresholds: { warnPercent: number; criticalPercent: number };
/** 测试注入 */
now?: () => Date;
diskUsage?: (path: string) => Promise<DiskUsage>;
checkRoot?: typeof checkRoot;
/** 单个路径 IO 检查超时(ms),缺省 3000 */
ioTimeoutMs?: number;
}
const LEVEL_ORDER: HealthLevel[] = ["ok", "warn", "critical"];
export function worst(levels: HealthLevel[]): HealthLevel {
let w: HealthLevel = "ok";
for (const l of levels) if (l !== "unknown" && LEVEL_ORDER.indexOf(l) > LEVEL_ORDER.indexOf(w)) w = l;
return w;
}
class Timeout extends Error {}
function withTimeout<T>(p: Promise<T>, ms: number): Promise<T> {
let t: ReturnType<typeof setTimeout> | undefined;
return Promise.race([
p,
new Promise<T>((_, reject) => {
t = setTimeout(() => reject(new Timeout("timeout")), ms);
}),
]).finally(() => clearTimeout(t));
}
const toDate = (v: Date | string) => (v instanceof Date ? v : new Date(v));
const num = (v: unknown) => (v == null ? null : Number(v));
// ── 进程 ──
export async function loadProcesses(db: Kysely<DB>, now: Date): Promise<SystemProcess[]> {
const rows = await db
.selectFrom("process_heartbeat")
.selectAll()
.where("last_beat_at", ">", new Date(now.getTime() - 24 * 3600_000))
.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,
};
});
}
/** 进程段的档位:有 stale 或延迟 ≥ 5s 为 critical;延迟 ≥ 1s 为 warn;没有任何 worker 心跳为 warn */
export function processLevel(ps: SystemProcess[]): HealthLevel {
const live = ps.filter((p) => !p.stale);
const levels: HealthLevel[] = [];
if (ps.some((p) => p.stale)) levels.push("warn");
if (!live.some((p) => p.processKind !== "api")) levels.push("warn");
for (const p of live) {
if (p.eventLoopLagMaxMs >= 5000) levels.push("critical");
else if (p.eventLoopLagMaxMs >= 1000) levels.push("warn");
}
return worst(levels);
}
// ── DB ──
export async function loadDb(db: Kysely<DB>): Promise<SystemHealth["db"]> {
try {
const t0 = performance.now();
await sql`select 1`.execute(db);
const latencyMs = Math.round((performance.now() - t0) * 10) / 10;
const size = await sql<{ size: string }>`select pg_database_size(current_database())::text as size`.execute(db);
const conns = await sql<{ total: number; active: number; idle: number; iit: number; max: number }>`
select count(*)::int as total,
count(*) filter (where state = 'active')::int as active,
count(*) filter (where state = 'idle')::int as idle,
count(*) filter (where state like 'idle in transaction%')::int as iit,
current_setting('max_connections')::int as max
from pg_stat_activity where datname = current_database()`.execute(db);
const c = conns.rows[0]!;
const usage = c.max > 0 ? c.total / c.max : 0;
const level: HealthLevel =
latencyMs >= 500 || usage >= 0.9 ? "critical" : latencyMs >= 100 || usage >= 0.75 || c.iit > 5 ? "warn" : "ok";
return {
level,
latencyMs,
sizeBytes: Number(size.rows[0]?.size ?? 0),
connections: { total: c.total, active: c.active, idle: c.idle, idleInTransaction: c.iit, max: c.max },
error: null,
};
} catch (err) {
return { level: "critical", latencyMs: null, sizeBytes: null, connections: null, error: (err as Error).message };
}
}
// ── 磁盘与挂载根 ──
interface PathTarget {
scope: SystemDisk["scope"];
id: string | null;
name: string;
path: string;
}
export async function loadPathTargets(db: Kysely<DB>) {
const roots = await db
.selectFrom("library_root")
.select(["id", "name", "local_path", "sentinel_file"])
.where("enabled", "=", true)
.orderBy("name")
.execute();
const maps = await db
.selectFrom("path_mapping")
.leftJoin("downloader_instance", "downloader_instance.id", "path_mapping.downloader_instance_id")
.select([
"path_mapping.id as id",
"path_mapping.remote_prefix as remote",
"path_mapping.local_prefix as local",
"downloader_instance.name as downloader",
])
.where("path_mapping.enabled", "=", true)
.where("path_mapping.view_kind", "=", "downloader")
.orderBy("path_mapping.local_prefix")
.execute();
return { roots, maps };
}
export function diskLevel(ratio: number, t: HealthDeps["thresholds"]): HealthLevel {
const pct = ratio * 100;
return pct >= t.criticalPercent ? "critical" : pct >= t.warnPercent ? "warn" : "ok";
}
export async function loadDisksAndMounts(d: HealthDeps): Promise<{ disks: SystemDisk[]; mounts: SystemMount[] }> {
const usage = d.diskUsage ?? diskUsage;
const check = d.checkRoot ?? checkRoot;
const ms = d.ioTimeoutMs ?? 3000;
const { roots, maps } = await loadPathTargets(d.db);
const targets: PathTarget[] = [
{ scope: "data_dir", id: null, name: "DATA_DIR", path: d.dataDir },
...roots.map((r) => ({ scope: "library_root" as const, id: r.id, name: r.name, path: r.local_path })),
...maps.map((m) => ({
scope: "download_dir" as const,
id: m.id,
name: m.downloader ? `${m.downloader} ${m.remote}` : m.remote,
path: m.local,
})),
];
const disks = await Promise.all(
targets.map(async (t): Promise<SystemDisk> => {
const base = { scope: t.scope, id: t.id, name: t.name, path: t.path };
try {
const u = await withTimeout(usage(t.path), ms);
if (u.status !== "ok")
return {
...base,
level: "unknown",
totalBytes: null,
usedBytes: null,
availableBytes: null,
usedRatio: null,
error: u.code,
};
return {
...base,
level: diskLevel(u.usedRatio, d.thresholds),
totalBytes: u.totalBytes,
usedBytes: u.usedBytes,
availableBytes: u.availableBytes,
usedRatio: Math.min(1, Math.max(0, u.usedRatio)),
error: null,
};
} catch (err) {
return {
...base,
level: err instanceof Timeout ? "critical" : "unknown",
totalBytes: null,
usedBytes: null,
availableBytes: null,
usedRatio: null,
error: err instanceof Timeout ? "timeout" : (err as Error).message,
};
}
}),
);
const mounts = await Promise.all([
...roots.map(async (r): Promise<SystemMount> => {
const base = { kind: "library_root" as const, id: r.id, name: r.name, path: r.local_path, sentinelChecked: true };
try {
const h = await withTimeout(check(r.local_path, r.sentinel_file ?? DEFAULT_ROOT_SENTINEL), ms);
return h.kind === "online"
? { ...base, status: "online", reason: null }
: { ...base, status: "offline", reason: h.reason };
} catch (err) {
return err instanceof Timeout
? { ...base, status: "timeout", reason: "timeout" }
: { ...base, status: "offline", reason: "error" };
}
}),
...maps.map(async (m): Promise<SystemMount> => {
const base = {
kind: "path_mapping" as const,
id: m.id,
name: m.downloader ? `${m.downloader} ${m.remote}` : m.remote,
path: m.local,
sentinelChecked: false,
};
try {
await withTimeout(
(async () => {
const st = await stat(m.local);
if (!st.isDirectory()) throw Object.assign(new Error("不是目录"), { code: "ENOTDIR" });
await readdir(m.local);
})(),
ms,
);
return { ...base, status: "online", reason: null };
} catch (err) {
if (err instanceof Timeout) return { ...base, status: "timeout", reason: "timeout" };
const code = (err as { code?: string }).code;
const reason =
code === "ENOENT" ? "root_missing" : code === "EACCES" || code === "EPERM" ? "permission_denied" : "error";
return { ...base, status: "offline", reason };
}
}),
]);
return { disks, mounts };
}
// ── 引擎 ──
const schemaTable = (schema: string, table: string) => {
if (!/^[a-z_][a-z0-9_]*$/.test(schema)) throw new Error(`非法 schema 名 ${schema}`);
return sql.raw(`"${schema}".${table}`);
};
export async function loadEngine(db: Kysely<DB>, schema: string, now: Date): Promise<SystemHealth["engine"]> {
const t = schemaTable(schema, "workflow_runs");
const since = new Date(now.getTime() - 24 * 3600_000);
const r = await sql<{
pending: number;
running: number;
oldest: Date | null;
finished: number;
failed: number;
}>`select
count(*) filter (where status = 'pending')::int as pending,
count(*) filter (where status = 'running' and worker_id is not null)::int as running,
min(coalesce(available_at, created_at)) filter (where status = 'pending' and coalesce(available_at, created_at) <= ${now}) as oldest,
count(*) filter (where status in ('completed', 'succeeded', 'failed') and finished_at >= ${since})::int as finished,
count(*) filter (where status = 'failed' and finished_at >= ${since})::int as failed
from ${t} where namespace_id = 'default'`.execute(db);
const row = r.rows[0]!;
const oldestAge = row.oldest ? Math.max(0, (now.getTime() - toDate(row.oldest).getTime()) / 1000) : null;
const rate = row.finished > 0 ? row.failed / row.finished : null;
const level: HealthLevel = worst([
oldestAge != null && oldestAge >= 1800 ? "critical" : oldestAge != null && oldestAge >= 300 ? "warn" : "ok",
rate != null && row.finished >= 10 && rate >= 0.3 ? "critical" : rate != null && rate >= 0.1 ? "warn" : "ok",
]);
return {
level,
pending: row.pending,
running: row.running,
oldestPendingAgeSec: oldestAge == null ? null : Math.round(oldestAge),
finished24h: row.finished,
failed24h: row.failed,
failureRate24h: rate,
};
}
// ── 定时任务 ──
export async function loadSchedules(db: Kysely<DB>, now: Date): Promise<SystemHealth["schedules"]> {
const jobs = await db
.selectFrom("scheduled_job")
.select(["id", "name", "enabled", "available", "last_status", "last_run_at", "last_error", "running_since"])
.orderBy("id")
.execute();
const failedRuns = await sql<{ n: number }>`
select count(*)::int as n from scheduled_job_run
where status = 'failed' and started_at >= ${new Date(now.getTime() - 24 * 3600_000)}`.execute(db);
const enabled = jobs.filter((j) => j.enabled && j.available);
const failing = enabled
.filter((j) => j.last_status === "failed")
.map((j) => ({
id: j.id,
name: j.name,
lastRunAt: j.last_run_at ? toDate(j.last_run_at).toISOString() : null,
lastError: j.last_error,
}));
return {
level: failing.length ? "warn" : "ok",
total: jobs.length,
enabled: enabled.length,
running: jobs.filter((j) => j.running_since).length,
failing,
failedRuns24h: failedRuns.rows[0]?.n ?? 0,
};
}
// ── 汇总 ──
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[]),
loadDb(d.db),
loadDisksAndMounts(d).catch(() => ({ disks: [] as SystemDisk[], mounts: [] as SystemMount[] })),
loadEngine(d.db, d.engineSchema, now).catch((): SystemHealth["engine"] => ({
level: "unknown",
pending: 0,
running: 0,
oldestPendingAgeSec: null,
finished24h: 0,
failed24h: 0,
failureRate24h: null,
})),
loadSchedules(d.db, now).catch((): SystemHealth["schedules"] => ({
level: "unknown",
total: 0,
enabled: 0,
running: 0,
failing: [],
failedRuns24h: 0,
})),
]);
// SSE:存活的 api/all 进程上报的连接数合计;本进程若还没写过心跳,用实时值
const selfCounted = processes.some((p) => !p.stale && p.instanceId === d.self.instanceId && p.sseConnections != null);
const sse =
processes.filter((p) => !p.stale).reduce((s, p) => s + (p.sseConnections ?? 0), 0) +
(selfCounted ? 0 : d.self.sseConnections);
const mountLevel: HealthLevel = paths.mounts.some((m) => m.status !== "online") ? "critical" : "ok";
return {
level: worst([
processLevel(processes),
db.level,
...paths.disks.map((x) => x.level),
mountLevel,
engine.level,
schedules.level,
]),
checkedAt: now.toISOString(),
api: {
version: d.self.version,
startedAt: d.self.startedAt.toISOString(),
uptimeSec: Math.max(0, Math.round((now.getTime() - d.self.startedAt.getTime()) / 1000)),
},
processes,
sse: { connections: sse },
db,
disks: paths.disks,
diskThresholds: d.thresholds,
mounts: paths.mounts,
engine,
schedules,
};
}
// ── 历史 ──
export async function loadHistory(db: Kysely<DB>, hours: number, now = new Date()): Promise<SystemHealthHistory> {
const from = new Date(now.getTime() - hours * 3600_000);
// 约 360 个点;最细 15s(一个心跳周期)
const bucketSec = Math.max(15, Math.ceil((hours * 3600) / 360 / 15) * 15);
const r = await sql<{
instance_id: string;
process_kind: string;
bucket: Date;
lag: number;
lag_max: number;
running: number | null;
sse: number | null;
}>`select instance_id, max(process_kind) as process_kind,
date_bin(make_interval(secs => ${bucketSec}), at, timestamptz 'epoch') as bucket,
avg(event_loop_lag_ms)::float8 as lag, max(event_loop_lag_max_ms)::float8 as lag_max,
max(running_runs)::float8 as running, max(sse_connections)::float8 as sse
from process_heartbeat_sample where at >= ${from} and at <= ${now}
group by instance_id, bucket order by instance_id, bucket`.execute(db);
const series = new Map<string, SystemHealthHistory["series"][number]>();
for (const row of r.rows) {
let s = series.get(row.instance_id);
if (!s) {
s = { instanceId: row.instance_id, processKind: row.process_kind, points: [] };
series.set(row.instance_id, s);
}
s.points.push({
at: toDate(row.bucket).toISOString(),
lagMs: Math.round(Number(row.lag) * 10) / 10,
lagMaxMs: Number(row.lag_max),
runningRuns: num(row.running),
sseConnections: num(row.sse),
});
}
return { from: from.toISOString(), to: now.toISOString(), bucketSec, series: [...series.values()] };
}
+207
View File
@@ -0,0 +1,207 @@
import { hostname } from "node:os";
import type { ProcessKind } from "@umt/contracts";
import { type DB, type Kysely, sql } from "@umt/db";
import type { Logger } from "pino";
import serverDeno from "../../../deno.json" with { type: "json" };
// 进程心跳写入(W9-I):每个常驻进程每 15s upsert process_heartbeat 一行 + 追加一条采样。
// 写失败只记 debug(库暂时不可达不影响业务;读取方按 last_beat_at 判 stale)。
export const HEARTBEAT_WRITE_MS = 15_000;
/** 超过这个时长没有心跳即 stale(3 个写周期) */
export const HEARTBEAT_STALE_MS = 45_000;
/** 采样保留天数 */
export const SAMPLE_RETENTION_DAYS = 7;
export const SERVER_VERSION: string = serverDeno.version;
/** 事件循环延迟采样器:每 intervalMs 测一次 setInterval 实际触发偏差;window 在每次心跳读取后重置 */
export function createLagSampler(intervalMs = 1000) {
let expected = performance.now() + intervalMs;
let last = 0;
let windowMax = 0;
let lifetimeMax = 0;
const timer = setInterval(() => {
const now = performance.now();
const lag = Math.max(0, Math.round(now - expected));
expected = now + intervalMs;
last = lag;
if (lag > windowMax) windowMax = lag;
if (lag > lifetimeMax) lifetimeMax = lag;
}, intervalMs);
try {
Deno.unrefTimer(timer);
} catch {
// 非 Deno 运行时(vitest 走 deno,理论上不会到这里)
}
return {
/** 读出 {last, windowMax, lifetimeMax} 并重置窗口 */
take() {
const r = { lastMs: last, windowMaxMs: Math.max(windowMax, last), lifetimeMaxMs: lifetimeMax };
windowMax = 0;
return r;
},
stop: () => clearInterval(timer),
};
}
export interface HeartbeatSample {
concurrency?: { io?: number; agent?: number };
runningRuns?: number | null;
sseConnections?: number | null;
extra?: Record<string, unknown>;
}
export interface HeartbeatWriterOptions {
db: Kysely<DB>;
kind: ProcessKind;
logger: Logger;
/** 每次写入前采集(可异步);抛错时按空采样写 */
sample?: () => HeartbeatSample | Promise<HeartbeatSample>;
intervalMs?: number;
version?: string;
host?: string;
pid?: number;
startedAt?: Date;
/** 测试注入:替代内置延迟采样器 */
lag?: { take(): { lastMs: number; windowMaxMs: number; lifetimeMaxMs: number }; stop(): void };
}
export function instanceIdOf(kind: ProcessKind, host: string, pid: number) {
return `${kind}:${host}:${pid}`;
}
function memory(): Record<string, number> {
try {
const m = Deno.memoryUsage();
return { rssBytes: m.rss, heapUsedBytes: m.heapUsed };
} catch {
return {};
}
}
/** 写一次心跳(upsert + 采样);返回写入的 instanceId */
export async function writeHeartbeat(
db: Kysely<DB>,
row: {
instanceId: string;
kind: ProcessKind;
pid: number;
host: string;
version: string;
startedAt: Date;
lagMs: number;
lagMaxMs: number;
sample: HeartbeatSample;
at?: Date;
},
): Promise<void> {
const s = row.sample;
const at = row.at ?? new Date();
const extra = { ...(s.extra ?? {}), ...(s.sseConnections != null ? { sseConnections: s.sseConnections } : {}) };
await sql`
insert into process_heartbeat (instance_id, process_kind, pid, host, version, started_at, last_beat_at,
event_loop_lag_ms, event_loop_lag_max_ms, concurrency, running_runs, extra)
values (${row.instanceId}, ${row.kind}, ${row.pid}, ${row.host}, ${row.version}, ${row.startedAt}, ${at},
${row.lagMs}, ${row.lagMaxMs}, ${JSON.stringify(s.concurrency ?? {})}::jsonb, ${s.runningRuns ?? null},
${JSON.stringify(extra)}::jsonb)
on conflict (instance_id) do update set process_kind = excluded.process_kind, pid = excluded.pid,
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);
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)
values (${row.instanceId}, ${row.kind}, ${at}, ${row.lagMs}, ${row.lagMaxMs}, ${s.runningRuns ?? null},
${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);
}
export interface HeartbeatWriter {
readonly instanceId: string;
/** 立即写一次(测试 / 启动时用) */
beat(): Promise<void>;
stop(): Promise<void>;
}
/** 起一个周期写入器:立即写一次,之后每 intervalMs 写一次;每约 1 小时裁剪一次 */
export function startHeartbeatWriter(o: HeartbeatWriterOptions): HeartbeatWriter {
const host = o.host ?? safeHostname();
const pid = o.pid ?? Deno.pid;
const startedAt = o.startedAt ?? new Date();
const version = o.version ?? SERVER_VERSION;
const instanceId = instanceIdOf(o.kind, host, pid);
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;
const once = async () => {
let sample: HeartbeatSample = {};
try {
sample = (await o.sample?.()) ?? {};
} catch (err) {
log.debug({ err }, "心跳采样失败");
}
const l = lag.take();
try {
await writeHeartbeat(o.db, {
instanceId,
kind: o.kind,
pid,
host,
version,
startedAt,
lagMs: l.lastMs,
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 }, "心跳写入失败");
}
};
const beat = () => {
if (stopped) return Promise.resolve();
// 上一次还没写完(库慢)就跳过这一拍,不堆积
inflight ??= once().finally(() => {
inflight = null;
});
return inflight;
};
void beat();
const timer = setInterval(() => void beat(), intervalMs);
try {
Deno.unrefTimer(timer);
} catch {
// ignore
}
return {
instanceId,
beat,
async stop() {
stopped = true;
clearInterval(timer);
lag.stop();
await inflight?.catch(() => {});
},
};
}
export function safeHostname(): string {
try {
return Deno.hostname();
} catch {
return hostname();
}
}
+103
View File
@@ -0,0 +1,103 @@
import { createRoute, type RouteConfig } from "@hono/zod-openapi";
import { Api, apiResponse, DEFAULT_SETTINGS, type EndpointDef } from "@umt/contracts";
import type { z } from "zod";
import { dataPaths } from "../../config.ts";
import { requireScope } from "../../http/auth.ts";
import { sseConnectionCount } from "../../http/sse.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";
export { HEARTBEAT_STALE_MS, HEARTBEAT_WRITE_MS, startHeartbeatWriter } from "./heartbeat.ts";
/**
* W9-I 进程与健康(docs/runbook/ops.md):
* - GET /system/health:进程心跳 / 事件循环延迟 / SSE 连接 / DB / 磁盘 / 挂载根 / 引擎 / 定时任务,仅 admin;
* - GET /system/health/history?hours=:心跳采样时间序列。
* 心跳写入:worker 进程在 worker/index.ts 里起写入器;纯 api 进程(role=api)在注册路由时起(role=all 由 worker 写 kind=all)。
*/
export interface SystemModuleOptions {
/** 引擎 schema(测试覆盖);缺省 openworkflow */
engineSchema?: string;
/** 测试注入:替换 health 取数依赖 */
healthDeps?: Partial<Pick<HealthDeps, "diskUsage" | "checkRoot" | "now" | "ioTimeoutMs" | "dataDir">>;
/** 缺省 true;false 时 role=api 也不起心跳写入器(测试) */
apiHeartbeat?: boolean;
}
type Infer<T> = T extends z.ZodType ? z.infer<T> : Record<string, never>;
function route<D extends EndpointDef>(
app: ModuleApp,
def: D,
handler: (i: { query: Infer<D["query"]> }) => Promise<unknown>,
) {
const request: RouteConfig["request"] = {};
if (def.query) request.query = def.query as never;
const r = createRoute({
method: def.method,
path: def.path,
summary: def.summary,
tags: ["system"],
middleware: [requireScope(def.scope)] as never,
request,
responses: {
200: { description: def.summary, content: { "application/json": { schema: apiResponse(def.response) } } },
},
});
// biome-ignore lint/suspicious/noExplicitAny: 路由泛型在封装里擦除,入参已由 zod-openapi 校验
app.openapi(r, (async (c: any) => {
const data = def.response.parse(await handler({ query: def.query ? c.req.valid("query") : {} } as never));
return c.json({ ok: true, data }, 200);
// biome-ignore lint/suspicious/noExplicitAny: 同上
}) as any);
}
export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule & {
apiHeartbeat(): HeartbeatWriter | null;
} {
const startedAt = new Date();
let writer: HeartbeatWriter | null = null;
return {
name: "system",
apiHeartbeat: () => writer,
registerRoutes(app, ctx: ServerContext) {
if (ctx.role === "api" && opts.apiHeartbeat !== false && !writer) {
writer = startHeartbeatWriter({
db: ctx.db,
logger: ctx.logger,
kind: "api",
startedAt,
sample: () => ({ sseConnections: sseConnectionCount() }),
});
}
const selfKind = ctx.role === "all" ? "all" : ctx.role === "api" ? "api" : null;
const selfId = selfKind ? instanceIdOf(selfKind, safeHostname(), Deno.pid) : undefined;
route(app, Api.system.healthDetail, async () => {
const cache = await ctx.settings.get("cache").catch(() => DEFAULT_SETTINGS.cache);
return await buildHealth({
db: ctx.db,
engineSchema: opts.engineSchema ?? "openworkflow",
dataDir: opts.healthDeps?.dataDir ?? dataPaths(ctx.config).dataDir,
self: {
version: SERVER_VERSION,
startedAt,
sseConnections: sseConnectionCount(),
...(selfId ? { instanceId: selfId } : {}),
},
thresholds: { warnPercent: cache.resumePercent, criticalPercent: cache.hardThresholdPercent },
...(opts.healthDeps ?? {}),
});
});
route(app, Api.system.healthHistory, async ({ query }) => await loadHistory(ctx.db, query.hours));
},
registerTasks() {},
async stop() {
await writer?.stop();
writer = null;
},
};
}
export const systemModule = createSystemModule();