From e8745115f47a5a570afc80fe959d52e6a5020947 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 15:23:37 +0800 Subject: [PATCH] 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 --- apps/server/src/http/sse.ts | 14 + .../modules/auth-settings/engine-business.ts | 246 ++++++++++ .../modules/auth-settings/engine-routes.ts | 11 +- apps/server/src/modules/index.ts | 2 + apps/server/src/modules/system/health.ts | 437 ++++++++++++++++++ apps/server/src/modules/system/heartbeat.ts | 207 +++++++++ apps/server/src/modules/system/index.ts | 103 +++++ apps/server/src/worker/index.ts | 32 +- apps/server/test/modules.test.ts | 3 +- apps/server/test/system/system.test.ts | 360 +++++++++++++++ docs/runbook/ops.md | 89 ++++ packages/contracts/src/api.ts | 15 + packages/contracts/src/ops.ts | 175 +++++++ packages/contracts/test/db-sync.test.ts | 4 + .../db/migrations/0034_process_heartbeat.ts | 54 +++ packages/db/src/types.ts | 35 ++ packages/db/test/schema.test.ts | 6 + 17 files changed, 1789 insertions(+), 4 deletions(-) create mode 100644 apps/server/src/modules/auth-settings/engine-business.ts create mode 100644 apps/server/src/modules/system/health.ts create mode 100644 apps/server/src/modules/system/heartbeat.ts create mode 100644 apps/server/src/modules/system/index.ts create mode 100644 apps/server/test/system/system.test.ts create mode 100644 docs/runbook/ops.md create mode 100644 packages/db/migrations/0034_process_heartbeat.ts diff --git a/apps/server/src/http/sse.ts b/apps/server/src/http/sse.ts index b97be08d..db02d32c 100644 --- a/apps/server/src/http/sse.ts +++ b/apps/server/src/http/sse.ts @@ -75,6 +75,12 @@ export class EventBus implements EventBusPort { const SINCE_PAGE = 500; +/** 本进程当前打开的 SSE 流数(进程与健康页用;不含 JSON 补读请求) */ +let openStreams = 0; +export function sseConnectionCount(): number { + return openStreams; +} + export interface SseOptions { bus: EventBusPort; keepaliveMs?: number; @@ -135,6 +141,13 @@ export function sseRoutes({ bus, keepaliveMs = 15_000 }: SseOptions) { } return streamSSE(c, async (stream) => { + openStreams += 1; + let counted = true; + const release = () => { + if (counted) openStreams = Math.max(0, openStreams - 1); + counted = false; + }; + stream.onAbort(release); const queue: EventEnvelope[] = []; let wake: (() => void) | null = null; // 先订阅再读缓冲,避免两者之间的事件丢失 @@ -185,6 +198,7 @@ export function sseRoutes({ bus, keepaliveMs = 15_000 }: SseOptions) { } finally { clearInterval(timer); unsubscribe(); + release(); } }); }) as never); diff --git a/apps/server/src/modules/auth-settings/engine-business.ts b/apps/server/src/modules/auth-settings/engine-business.ts new file mode 100644 index 00000000..3cc1716e --- /dev/null +++ b/apps/server/src/modules/auth-settings/engine-business.ts @@ -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 / 各业务表)。 +// 表达式引用外层 "".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) => + sql`jsonb_build_object('kind', ${sql.lit(kind)}::text, 'refId', (${ref})::text)`; +const whenKey = (key: string, kind: EngineBusinessKind, ref: RawBuilder = 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 = { + 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): RawBuilder[] { + const out: RawBuilder[] = []; + 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 这类 ":" 串 → 触发来源;认不出返回 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 & { business: EngineRunBusiness | null }; + +const TITLE_SQL: Record 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( + db: Kysely, + schema: string, + items: T[], +): Promise[]> { + 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, + schema: string, + runIds: string[], +): Promise> { + const out = new Map(); + 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: + const keys = found.map((r) => r.idempotency_key).filter((k): k is string => !!k); + const outboxKeys = new Set(); + 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>(); + for (const r of found) { + const set = byKind.get(r.b!.kind) ?? new Set(); + set.add(r.b!.refId); + byKind.set(r.b!.kind, set); + } + const titles = new Map(); + 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(); + 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; +} diff --git a/apps/server/src/modules/auth-settings/engine-routes.ts b/apps/server/src/modules/auth-settings/engine-routes.ts index 85c741af..b98b0d74 100644 --- a/apps/server/src/modules/auth-settings/engine-routes.ts +++ b/apps/server/src/modules/auth-settings/engine-routes.ts @@ -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, 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 }) => { diff --git a/apps/server/src/modules/index.ts b/apps/server/src/modules/index.ts index d71f1b49..809edcf1 100644 --- a/apps/server/src/modules/index.ts +++ b/apps/server/src/modules/index.ts @@ -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, ]; diff --git a/apps/server/src/modules/system/health.ts b/apps/server/src/modules/system/health.ts new file mode 100644 index 00000000..e6f116d4 --- /dev/null +++ b/apps/server/src/modules/system/health.ts @@ -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; + 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; + 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(p: Promise, ms: number): Promise { + let t: ReturnType | undefined; + return Promise.race([ + p, + new Promise((_, 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, now: Date): Promise { + 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; + 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): Promise { + 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) { + 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 => { + 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 => { + 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 => { + 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, schema: string, now: Date): Promise { + 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, now: Date): Promise { + 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 { + 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, hours: number, now = new Date()): Promise { + 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(); + 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()] }; +} diff --git a/apps/server/src/modules/system/heartbeat.ts b/apps/server/src/modules/system/heartbeat.ts new file mode 100644 index 00000000..207dd95c --- /dev/null +++ b/apps/server/src/modules/system/heartbeat.ts @@ -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; +} + +export interface HeartbeatWriterOptions { + db: Kysely; + kind: ProcessKind; + logger: Logger; + /** 每次写入前采集(可异步);抛错时按空采样写 */ + sample?: () => HeartbeatSample | Promise; + 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 { + try { + const m = Deno.memoryUsage(); + return { rssBytes: m.rss, heapUsedBytes: m.heapUsed }; + } catch { + return {}; + } +} + +/** 写一次心跳(upsert + 采样);返回写入的 instanceId */ +export async function writeHeartbeat( + db: Kysely, + row: { + instanceId: string; + kind: ProcessKind; + pid: number; + host: string; + version: string; + startedAt: Date; + lagMs: number; + lagMaxMs: number; + sample: HeartbeatSample; + at?: Date; + }, +): Promise { + 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, days = SAMPLE_RETENTION_DAYS): Promise { + 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; + stop(): Promise; +} + +/** 起一个周期写入器:立即写一次,之后每 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 | 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(); + } +} diff --git a/apps/server/src/modules/system/index.ts b/apps/server/src/modules/system/index.ts new file mode 100644 index 00000000..d6851f1c --- /dev/null +++ b/apps/server/src/modules/system/index.ts @@ -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>; + /** 缺省 true;false 时 role=api 也不起心跳写入器(测试) */ + apiHeartbeat?: boolean; +} + +type Infer = T extends z.ZodType ? z.infer : Record; + +function route( + app: ModuleApp, + def: D, + handler: (i: { query: Infer }) => Promise, +) { + 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(); diff --git a/apps/server/src/worker/index.ts b/apps/server/src/worker/index.ts index 4665d33a..bbe53d9b 100644 --- a/apps/server/src/worker/index.ts +++ b/apps/server/src/worker/index.ts @@ -2,7 +2,8 @@ import { EventType } from "@umt/contracts"; import { OutboxRelay } from "@umt/tasks"; import type { Logger } from "pino"; import { collectOutboxRoutes, tasksLogSink } from "../context.ts"; -import type { EventBusPort } from "../http/sse.ts"; +import { type EventBusPort, sseConnectionCount } from "../http/sse.ts"; +import { type HeartbeatWriter, startHeartbeatWriter } from "../modules/system/heartbeat.ts"; import type { ServerContext, ServerModule } from "../modules/types.ts"; /** agent 组任务:整段会话占一槽数分钟,与文件复制等 io 任务分池 */ @@ -155,6 +156,34 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe } } + // 进程与健康页:每 15s 把心跳/延迟/并发/运行中 run 数写进 process_heartbeat(写失败不影响 worker) + let hbWriter: HeartbeatWriter | null = null; + if (runtime) { + const { ctx } = runtime; + const groupTasks = new Set(plan.flatMap((p) => p.versions ?? [])); + const countAll = plan.some((p) => !p.versions); + hbWriter = startHeartbeatWriter({ + db: ctx.db, + logger, + kind: group === "split" ? "all" : group === "agent" ? "worker:agent" : "worker", + sample: async () => { + const running = await ctx.engine.list({ status: "running", limit: 500 }).catch(() => null); + const s = probe.stats(); + return { + concurrency: { + ...(plan.some((p) => p.label !== "agent") ? { io: runtime.concurrency ?? 4 } : {}), + ...(plan.some((p) => p.label === "agent") ? { agent: runtime.agentConcurrency ?? 4 } : {}), + }, + runningRuns: running + ? running.filter((r) => r.status === "running" && (countAll || groupTasks.has(r.taskName))).length + : null, + sseConnections: group === "split" ? sseConnectionCount() : null, + extra: { group, lagWarnings: s.loopLagWarnings, probeMaxLagMs: s.maxLoopLagMs }, + }; + }, + }); + } + log.info( { heartbeatMs, @@ -170,6 +199,7 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe async stop() { clearInterval(timer); probe.stop(); + await hbWriter?.stop(); for (const m of started.reverse()) { try { await m.stop?.(); diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts index 0f2bc42d..d30c5e5d 100644 --- a/apps/server/test/modules.test.ts +++ b/apps/server/test/modules.test.ts @@ -43,7 +43,7 @@ const pingModule: ServerModule = { }; describe("ServerModule 骨架", () => { - it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索(末尾),名字唯一", () => { + it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 进程与健康(末尾),名字唯一", () => { expect(SERVER_MODULES.map((m) => m.name)).toEqual([ "auth-settings", "sources-identity", @@ -53,6 +53,7 @@ describe("ServerModule 骨架", () => { "pi", "schedules", "search", + "system", ]); }); diff --git a/apps/server/test/system/system.test.ts b/apps/server/test/system/system.test.ts new file mode 100644 index 00000000..8724dc1a --- /dev/null +++ b/apps/server/test/system/system.test.ts @@ -0,0 +1,360 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { DiskUsage } from "@umt/adapters"; +import { SystemHealth, SystemHealthHistory } from "@umt/contracts"; +import { sql } from "@umt/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { inferTriggeredBy, triggeredByFromActor } from "../../src/modules/auth-settings/engine-business.ts"; +import { diskLevel, processLevel, worst } from "../../src/modules/system/health.ts"; +import { HEARTBEAT_STALE_MS, startHeartbeatWriter, writeHeartbeat } from "../../src/modules/system/heartbeat.ts"; +import { createSystemModule } from "../../src/modules/system/index.ts"; +import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; +import { DEV_KEY, json } from "../helpers.ts"; + +// W9-I 进程与健康 + 引擎 runs 业务字段(真实 PG;磁盘 / 哨兵检查注入假实现)。 + +const hasPg = await pgAvailable(); + +describe("纯函数", () => { + it("档位合并忽略 unknown;磁盘阈值", () => { + expect(worst(["ok", "unknown", "warn"])).toBe("warn"); + expect(worst(["unknown"])).toBe("ok"); + expect(worst(["warn", "critical", "ok"])).toBe("critical"); + const t = { warnPercent: 85, criticalPercent: 90 }; + expect(diskLevel(0.5, t)).toBe("ok"); + expect(diskLevel(0.86, t)).toBe("warn"); + expect(diskLevel(0.9, t)).toBe("critical"); + }); + + it("进程档位:stale / 无 worker → warn;延迟 ≥5s → critical", () => { + const p = (kind: "api" | "worker", o: { stale?: boolean; lag?: number } = {}) => + ({ + instanceId: `${kind}:h:1`, + processKind: kind, + pid: 1, + host: "h", + version: "t", + startedAt: new Date().toISOString(), + lastBeatAt: new Date().toISOString(), + beatAgeSec: 0, + stale: o.stale ?? false, + eventLoopLagMs: 0, + eventLoopLagMaxMs: o.lag ?? 0, + concurrency: {}, + runningRuns: null, + sseConnections: null, + extra: {}, + }) as const; + expect(processLevel([p("api"), p("worker")])).toBe("ok"); + expect(processLevel([p("api")])).toBe("warn"); + expect(processLevel([p("api"), p("worker", { stale: true })])).toBe("warn"); + expect(processLevel([p("api"), p("worker", { lag: 1200 })])).toBe("warn"); + expect(processLevel([p("api"), p("worker", { lag: 6000 })])).toBe("critical"); + }); + + it("触发来源推断", () => { + expect(triggeredByFromActor("api_key:abc")).toBe("api_key"); + expect(triggeredByFromActor("admin_session:u1")).toBe("user"); + expect(triggeredByFromActor("system")).toBeNull(); + const base = { + kind: "download" as const, + idempotencyKey: null, + taskTriggeredBy: null, + agentTrigger: null, + agentCreatedBy: null, + fromOutbox: false, + }; + expect(inferTriggeredBy({ ...base, kind: "agent", agentTrigger: "subagent" })).toBe("agent"); + expect(inferTriggeredBy({ ...base, kind: "agent", agentTrigger: "user", agentCreatedBy: "api_key:k" })).toBe( + "api_key", + ); + expect(inferTriggeredBy({ ...base, taskTriggeredBy: "agent" })).toBe("agent"); + expect(inferTriggeredBy({ ...base, idempotencyKey: "retry:r1:1" })).toBe("user"); + expect(inferTriggeredBy({ ...base, idempotencyKey: "ingestFeed:manual:f:1" })).toBe("user"); + expect(inferTriggeredBy({ ...base, idempotencyKey: "release.observed:x", fromOutbox: true })).toBe("event"); + expect(inferTriggeredBy({ ...base, idempotencyKey: "snapshotSite:s1:123" })).toBe("schedule"); + expect(inferTriggeredBy({ ...base, kind: "schedule", idempotencyKey: "purgeTrash:1" })).toBe("schedule"); + expect(inferTriggeredBy(base)).toBeNull(); + }); +}); + +describe.skipIf(!hasPg)("W9-I 进程与健康(真实 PG)", () => { + let h: Harness; + let admin: ReturnType; + const dir = mkdtempSync(join(tmpdir(), "umt-w9i-")); + const libOk = join(dir, "lib-ok"); + const usage = (path: string): Promise => + Promise.resolve( + path.includes("full") + ? { + status: "ok", + path, + totalBytes: 1000, + freeBytes: 20, + availableBytes: 20, + usedBytes: 980, + usedRatio: 0.98, + } + : path.includes("hang") + ? new Promise(() => {}) + : { + status: "ok", + path, + totalBytes: 1000, + freeBytes: 500, + availableBytes: 500, + usedBytes: 500, + usedRatio: 0.5, + }, + ); + + beforeAll(async () => { + h = await createHarness({ + engine: true, + extraModules: [], + }); + // harness 在 createHarness 时已建 app;这里再建一个挂了 system 模块的 app(同一 ctx / db) + const { createApp } = await import("../../src/http/app.ts"); + const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts"); + h.app = createApp({ + config: h.ctx.config, + logger: h.ctx.logger, + bus: h.bus, + db: h.db, + role: "all", + version: "0.0.0-test", + modules: [ + createAuthSettingsModule({ setupTokenOnStart: false, engineSchema: h.engineSchema }), + createSystemModule({ + engineSchema: h.engineSchema, + apiHeartbeat: false, + healthDeps: { dataDir: dir, diskUsage: usage, ioTimeoutMs: 200 }, + }), + ], + ctx: h.ctx, + }) as Harness["app"]; + admin = bearer(h.app, DEV_KEY); + await Deno.mkdir(libOk, { recursive: true }); + await Deno.writeTextFile(join(libOk, ".umt-root"), "x"); + }, 60_000); + afterAll(async () => { + await h?.close(); + rmSync(dir, { recursive: true, force: true }); + }); + + it("心跳写入:upsert 同一 instance_id + 追加采样;超过 45s 标 stale", async () => { + const w = startHeartbeatWriter({ + db: h.db, + logger: h.ctx.logger, + kind: "worker", + host: "test-host", + pid: 4242, + intervalMs: 60_000, + lag: { take: () => ({ lastMs: 3, windowMaxMs: 12, lifetimeMaxMs: 40 }), stop() {} }, + sample: () => ({ concurrency: { io: 4 }, runningRuns: 2, extra: { group: "io" } }), + }); + await w.beat(); + await w.beat(); + await w.stop(); + expect(w.instanceId).toBe("worker:test-host:4242"); + const rows = await h.db.selectFrom("process_heartbeat").selectAll().execute(); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + process_kind: "worker", + pid: 4242, + event_loop_lag_ms: 3, + event_loop_lag_max_ms: 12, + running_runs: 2, + concurrency: { io: 4 }, + }); + expect(rows[0]!.extra).toMatchObject({ group: "io", lifetimeMaxLagMs: 40 }); + const samples = await h.db.selectFrom("process_heartbeat_sample").selectAll().execute(); + expect(samples.length).toBeGreaterThanOrEqual(2); + + // 一个 api 进程 60s 前的心跳 → stale + await writeHeartbeat(h.db, { + instanceId: "api:old-host:1", + kind: "api", + pid: 1, + host: "old-host", + version: "0.1.0", + startedAt: new Date(Date.now() - 3600_000), + lagMs: 1, + lagMaxMs: 2, + sample: { sseConnections: 3 }, + at: new Date(Date.now() - HEARTBEAT_STALE_MS - 15_000), + }); + + const res = await admin("GET", "/system/health"); + expect(res.status).toBe(200); + const d = SystemHealth.parse((await json(res)).data); + const byId = new Map(d.processes.map((p) => [p.instanceId, p])); + expect(byId.get("worker:test-host:4242")).toMatchObject({ stale: false, runningRuns: 2, concurrency: { io: 4 } }); + expect(byId.get("api:old-host:1")).toMatchObject({ stale: true, sseConnections: 3 }); + // stale 进程的 SSE 不计入 + expect(d.sse.connections).toBe(0); + }); + + it("health 各段形状:DB / 磁盘阈值 / 挂载根哨兵 / 引擎 / 定时任务", async () => { + await sql`insert into library_root (name, local_path, sentinel_file) values + ('电影', ${libOk}, '.umt-root'), ('剧集(掉线)', ${join(dir, "missing")}, null)`.execute(h.db); + await sql`insert into downloader_instance (name, base_url) values ('qb1', 'http://qb')`.execute(h.db); + await sql`insert into path_mapping (view_kind, downloader_instance_id, remote_prefix, local_prefix, priority) + select 'downloader', id, '/downloads/full', ${join(dir, "full")}, 0 from downloader_instance where name = 'qb1'`.execute( + h.db, + ); + const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`); + const run = (id: string, status: string, name: string, input: unknown, at: Date, finished: Date | null = null) => + sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, attempts, + worker_id, available_at, started_at, finished_at, created_at, updated_at) + values ('default', ${id}, ${name}, 'v1', ${status}, '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1, + ${status === "running" ? "w1" : null}, ${at}, ${at}, ${finished}, ${at}, ${at})`.execute(h.db); + const now = Date.now(); + await run("h1", "pending", "ingestFeed", { input: {} }, new Date(now - 600_000)); + await run("h2", "failed", "ingestFeed", { input: {} }, new Date(now - 3600_000), new Date(now - 3500_000)); + await run("h3", "completed", "ingestFeed", { input: {} }, new Date(now - 3600_000), new Date(now - 3500_000)); + await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule, + last_status, last_error, last_run_at) + values ('feeds.poll', 'Feed 抓取', 'collect', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb, + '{"type":"interval","everyMs":60000}'::jsonb, 'failed', 'boom', now())`.execute(h.db); + await sql`insert into scheduled_job_run (job_id, trigger, status, error) values ('feeds.poll', 'schedule', 'failed', 'boom')`.execute( + h.db, + ); + + const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data); + expect(d.db).toMatchObject({ level: expect.any(String), error: null }); + expect(d.db.latencyMs).toBeGreaterThanOrEqual(0); + expect(d.db.sizeBytes).toBeGreaterThan(0); + expect(d.db.connections!.max).toBeGreaterThan(0); + + expect(d.diskThresholds).toEqual({ warnPercent: 85, criticalPercent: 90 }); + const disk = (scope: string, name?: string) => d.disks.find((x) => x.scope === scope && (!name || x.name === name)); + expect(disk("data_dir")).toMatchObject({ path: dir, level: "ok", usedRatio: 0.5 }); + expect(disk("download_dir")).toMatchObject({ name: "qb1 /downloads/full", level: "critical", usedRatio: 0.98 }); + expect(disk("library_root", "电影")).toMatchObject({ level: "ok" }); + + const mount = (name: string) => d.mounts.find((m) => m.name === name); + expect(mount("电影")).toMatchObject({ + kind: "library_root", + status: "online", + reason: null, + sentinelChecked: true, + }); + expect(mount("剧集(掉线)")).toMatchObject({ status: "offline", reason: "root_missing" }); + expect(mount("qb1 /downloads/full")).toMatchObject({ + kind: "path_mapping", + status: "offline", + sentinelChecked: false, + }); + + expect(d.engine).toMatchObject({ pending: 1, finished24h: 2, failed24h: 1, failureRate24h: 0.5, level: "warn" }); + expect(d.engine.oldestPendingAgeSec).toBeGreaterThanOrEqual(590); + expect(d.schedules).toMatchObject({ level: "warn", enabled: 1, failedRuns24h: 1 }); + expect(d.schedules.failing).toEqual([ + expect.objectContaining({ id: "feeds.poll", name: "Feed 抓取", lastError: "boom" }), + ]); + expect(d.level).toBe("critical"); + expect(d.api.version).toMatch(/\d+\.\d+\.\d+/); + }); + + it("磁盘 statfs 卡死 → timeout 标 critical,不拖住请求", async () => { + await sql`insert into library_root (name, local_path) values ('卡死', ${join(dir, "hang")})`.execute(h.db); + const t0 = Date.now(); + const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data); + expect(Date.now() - t0).toBeLessThan(3000); + expect(d.disks.find((x) => x.name === "卡死")).toMatchObject({ level: "critical", error: "timeout" }); + await sql`delete from library_root where name = '卡死'`.execute(h.db); + }); + + it("history:按桶聚合的时间序列;hours 越界 422", async () => { + const res = await admin("GET", "/system/health/history?hours=1"); + expect(res.status).toBe(200); + const d = SystemHealthHistory.parse((await json(res)).data); + expect(d.bucketSec).toBe(15); + const s = d.series.find((x) => x.instanceId === "worker:test-host:4242"); + expect(s?.processKind).toBe("worker"); + expect(s!.points[0]).toMatchObject({ lagMaxMs: 12, runningRuns: 2 }); + expect((await admin("GET", "/system/health/history?hours=500")).status).toBe(422); + expect((await json(await admin("GET", "/system/health/history?hours=168"))).data.bucketSec).toBe(1680); + }); + + it("引擎 runs 补 business 字段 + businessKind / businessRefId 筛选", async () => { + const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`); + const at = new Date("2026-09-10T00:00:00Z"); + const run = (id: string, name: string, input: unknown, sec: number) => + sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, attempts, + available_at, created_at, updated_at) + values ('default', ${id}, ${name}, 'v1', 'completed', '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1, + ${at}, ${new Date(at.getTime() + sec * 1000)}, ${at})`.execute(h.db); + const meta = (runId: string, name: string, key: string) => + sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key) values (${runId}, ${name}, 1, ${key})`.execute( + h.db, + ); + + const feed = await sql<{ + id: string; + }>`insert into feed_source (name, url_display) values ('HDSky RSS', 'https://x/rss') + returning id::text`.execute(h.db); + const feedId = feed.rows[0]!.id; + const ar = await sql<{ id: string }>`insert into agent_run (trigger, goal, created_by) + values ('schedule', '巡检:待处理', 'schedule') returning id::text`.execute(h.db); + const agentId = ar.rows[0]!.id; + + // b1 事件链触发的 Feed 抓取(outbox 幂等键) + await run("b1", "ingestFeed", { input: { feedSourceId: feedId }, traceId: "t" }, 1); + await meta("b1", "ingestFeed", "feed.due:x:1"); + await sql`insert into outbox_event (type, payload, idempotency_key) values ('feed.due', '{}'::jsonb, 'feed.due:x:1')`.execute( + h.db, + ); + // b2 Agent 会话(定时巡检) + await run("b2", "runAgent", { input: { agentRunId: agentId } }, 2); + await meta("b2", "runAgent", `agent_run:${agentId}`); + // b3 API Key 发起的整理(只有业务任务行能关联) + await run("b3", "verifyAndImport", { input: { taskId: "00000000-0000-0000-0000-000000000001" } }, 3); + await meta("b3", "verifyAndImport", "verifyAndImport:files:p1"); + await sql`insert into task (kind, idempotency_key, input, subject_type, subject_id, triggered_by, engine_run_id) + values ('verifyAndImport', 'files:p1', '{}'::jsonb, 'organization_record', '00000000-0000-0000-0000-0000000000aa', 'api_key:k1', 'b3')`.execute( + h.db, + ); + // b4 定时清理回收站(无业务 id) + await run("b4", "purgeTrash", { input: {} }, 4); + await meta("b4", "purgeTrash", "purgeTrash:slot-1"); + await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule) + values ('trash.purge', '清理回收站', 'maintain', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb, + '{"type":"interval","everyMs":60000}'::jsonb)`.execute(h.db); + // b5 手动立即执行的清理 → 不算 schedule,也没有业务对象 + await run("b5", "purgeTrash", { input: {} }, 5); + await meta("b5", "purgeTrash", "purgeTrash:manual:r1"); + + const list = await json(await admin("GET", "/engine/runs?from=2026-09-10T00:00:00Z&to=2026-09-11T00:00:00Z")); + const by = new Map(list.data.items.map((r: { id: string; business: unknown }) => [r.id, r.business])); + expect(by.get("b1")).toEqual({ kind: "feed", refId: feedId, title: "HDSky RSS", triggeredBy: "event" }); + expect(by.get("b2")).toEqual({ kind: "agent", refId: agentId, title: "巡检:待处理", triggeredBy: "schedule" }); + expect(by.get("b3")).toEqual({ + kind: "organization", + refId: "00000000-0000-0000-0000-0000000000aa", + title: null, + triggeredBy: "api_key", + }); + expect(by.get("b4")).toEqual({ + kind: "schedule", + refId: "trash.purge", + title: "清理回收站", + triggeredBy: "schedule", + }); + expect(by.get("b5")).toBeNull(); + + const f = await json(await admin("GET", "/engine/runs?businessKind=feed")); + expect(f.data.items.map((r: { id: string }) => r.id)).toEqual(["b1"]); + const g = await json(await admin("GET", `/engine/runs?businessRefId=${agentId}`)); + expect(g.data.items.map((r: { id: string }) => r.id)).toEqual(["b2"]); + const s = await json(await admin("GET", "/engine/runs?businessKind=schedule&businessRefId=trash.purge")); + expect(s.data.items.map((r: { id: string }) => r.id)).toEqual(["b4"]); + expect((await admin("GET", "/engine/runs?businessKind=bogus")).status).toBe(422); + + const detail = (await json(await admin("GET", "/engine/runs/b2"))).data; + expect(detail.business).toMatchObject({ kind: "agent", refId: agentId }); + expect(detail.stepAttempts).toEqual([]); + }); +}); diff --git a/docs/runbook/ops.md b/docs/runbook/ops.md new file mode 100644 index 00000000..70357728 --- /dev/null +++ b/docs/runbook/ops.md @@ -0,0 +1,89 @@ +# 进程与健康 + +控制台「运行日志 → 进程与健康」的数据来源。后端在 `apps/server/src/modules/system/`,两个端点都只给 admin。 + +## 心跳怎么写 + +| 进程 | 谁写 | `process_kind` | +| --- | --- | --- | +| `--role api` | system 模块注册路由时起写入器 | `api` | +| `--role worker`(io 组) | `worker/index.ts` | `worker` | +| `--role worker:agent` | `worker/index.ts` | `worker:agent` | +| `--role all` | `worker/index.ts`(同进程的 API 不再单写) | `all` | + +- 每 15 秒 upsert `process_heartbeat` 一行,同时往 `process_heartbeat_sample` 追加一条采样。 +- 行主键 `instance_id = ::`,容器重启覆盖同一行。 +- 写入方每小时裁剪一次:采样保留 7 天,7 天没心跳的进程行删除。 +- 写库失败只记 debug 日志,不影响业务。 + +## GET /system/health 字段 + +| 字段 | 含义 | +| --- | --- | +| `level` | 各段最差档:`ok` / `warn` / `critical`。取不到数据的段是 `unknown`,不参与汇总 | +| `api` | 处理本次请求的 API 进程:版本、启动时间、运行秒数 | +| `processes[]` | 最近 24 小时有心跳的进程 | +| `processes[].stale` | 超过 45 秒没心跳 | +| `processes[].eventLoopLagMs` | 最近一次采样的事件循环延迟(1 秒采一次) | +| `processes[].eventLoopLagMaxMs` | 最近一个心跳周期(15 秒)内的最大延迟 | +| `processes[].concurrency` | 引擎 worker 并发 `{io, agent}`;api 进程为 `{}` | +| `processes[].runningRuns` | 本组任务正在执行的 run 数。同组多实例时是整组合计;api 进程为 null | +| `processes[].sseConnections` | 本进程打开的 SSE 流数。不托管 HTTP 的进程为 null | +| `processes[].extra` | `rssBytes`、`heapUsedBytes`、`lifetimeMaxLagMs`,worker 另有 `group`、`lagWarnings`、`probeMaxLagMs` | +| `sse.connections` | 存活 api/all 进程的 SSE 连接合计 | +| `db` | `select 1` 延迟、库大小、本库连接数(按 active / idle / idle in transaction 分)与 `max_connections` | +| `disks[]` | DATA_DIR、每个启用的库根、每个下载器路径映射的 statfs 用量 | +| `diskThresholds` | 取自设置 `cache.resumePercent`(warn)和 `cache.hardThresholdPercent`(critical) | +| `mounts[]` | 库根做哨兵检查(`checkRoot`),下载目录只查目录可读。`status`:`online` / `offline` / `timeout` | +| `engine` | 待处理 run 数、执行中 run 数、最老待处理时长、24 小时结束数 / 失败数 / 失败率 | +| `schedules` | 定时任务总数、启用数、执行中数、最近一次失败的启用任务、24 小时失败运行数 | + +单个路径的 statfs 或哨兵检查超过 3 秒记 `timeout`:磁盘段标 critical,挂载段标 `timeout`。请求本身不会被卡住。 + +## GET /system/health/history?hours= + +- `hours` 取 1–168,缺省 6。 +- 按桶聚合,每条序列最多约 360 个点。桶宽最小 15 秒,168 小时时是 1680 秒。 +- 每个进程一条序列,每个点有 `lagMs`(桶内平均)、`lagMaxMs`(桶内最大)、`runningRuns`、`sseConnections`。 + +## 告警阈值建议 + +- 进程 `stale`:warn。连续 2 分钟仍 stale 就去查容器状态和日志。 +- 没有任何存活的 worker / all 进程:warn。任务不会被执行。 +- 事件循环最大延迟 ≥ 1 秒记 warn,≥ 5 秒记 critical。延迟接近 15 秒时,引擎租约有丢失风险。 +- DB 延迟 ≥ 100 ms 或连接数 ≥ 75% `max_connections` 记 warn;≥ 500 ms 或 ≥ 90% 记 critical。 +- `idle in transaction` 连接超过 5 个记 warn,通常是有事务没提交。 +- 磁盘用量 ≥ `cache.resumePercent`(缺省 85%)记 warn,≥ `cache.hardThresholdPercent`(缺省 90%)记 critical。 +- 挂载根 `offline` 或 `timeout`:critical。先查宿主机挂载,再查哨兵文件。 +- 最老待处理 run 等待 ≥ 5 分钟记 warn,≥ 30 分钟记 critical。通常是 worker 没起来或并发被占满。 +- 24 小时失败率 ≥ 10% 记 warn;失败率 ≥ 30% 且结束数 ≥ 10 记 critical。 +- 有启用的定时任务最近一次运行失败:warn。 + +## 引擎 runs 的业务字段 + +`GET /engine/runs` 和 `GET /engine/runs/{id}` 的每个 run 多了 `business`,推断不出时为 null: + +```json +{ "kind": "feed", "refId": "", "title": "HDSky RSS", "triggeredBy": "event" } +``` + +- `kind`:`subscription` / `download` / `organization` / `agent` / `schedule` / `feed` / `site` / `decision` / `media`。 +- `refId`:业务对象 id;`schedule` 是定时任务 key(如 `trash.purge`)。`download` 优先指向下载任务,计划还没提交时指向收录计划 id。 +- `title`:业务对象的展示名。对象已删除时为 null。 +- `triggeredBy`:`user` / `api_key` / `schedule` / `agent` / `event`,推断不出为 null。 + +`kind` 与 `refId` 的推断顺序(`auth-settings/engine-business.ts`,列表筛选和展示用同一条 SQL 表达式): + +1. run 输入里的业务 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`、`refreshPiTokens`、`cleanupDownloadCache`、`reconcileLibrary`),且幂等键不是手动或重跑。 + +`triggeredBy` 的推断顺序: + +1. Agent 会话看 `agent_run.trigger`。 +2. 看 `task.triggered_by`。 +3. 幂等键是 `retry:` 或含 `manual` 的记为 `user`。 +4. 引擎幂等键能对上 outbox 事件的记为 `event`。 +5. 调度器时间槽键记为 `schedule`。 + +筛选参数:`businessKind`、`businessRefId`,可以一起用。这两个参数对每行都要算一次推断表达式,runs 很多时要带上时间范围。 diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index f8b409fa..f07482a5 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -1654,6 +1654,21 @@ export const Api = { summary: "服务健康状态(免鉴权)", response: HealthStatus, }), + healthDetail: ep({ + method: "get", + path: "/system/health", + scope: "admin", + summary: "进程与健康:进程心跳 / 事件循环延迟 / SSE 连接 / DB / 磁盘 / 挂载根 / 引擎 / 定时任务", + response: ops.SystemHealth, + }), + healthHistory: ep({ + method: "get", + path: "/system/health/history", + scope: "admin", + summary: "心跳采样时间序列(事件循环延迟、运行中 run、SSE 连接;按桶聚合)", + query: ops.SystemHealthHistoryQuery, + response: ops.SystemHealthHistory, + }), }, // ── 引擎面板(仅 admin) ── diff --git a/packages/contracts/src/ops.ts b/packages/contracts/src/ops.ts index c98d6d5a..05d613ec 100644 --- a/packages/contracts/src/ops.ts +++ b/packages/contracts/src/ops.ts @@ -225,6 +225,32 @@ export type EngineStepKind = z.infer; /** 引擎 ID 是 openworkflow 自己生成的文本 ID(非本地 UUID) */ const EngineId = z.string().min(1).max(200); +/** run 关联的业务对象(W9-I):由 run 输入 / 业务任务行 / agent_run / tasks_run_meta / outbox 事件推断,推断不出为 null */ +export const EngineBusinessKind = z.enum([ + "subscription", + "download", + "organization", + "agent", + "schedule", + "feed", + "site", + "decision", + "media", +]); +export type EngineBusinessKind = z.infer; +/** 触发来源:user=控制台会话 / 手动按钮 / 重跑;api_key=API Key 调用;schedule=定时任务;agent=Agent 或子会话;event=outbox 事件链 */ +export const EngineTriggeredBy = z.enum(["user", "api_key", "schedule", "agent", "event"]); +export type EngineTriggeredBy = z.infer; +export const EngineRunBusiness = z.object({ + kind: EngineBusinessKind, + /** 业务对象 id(schedule 为定时任务 key,如 trash.purge) */ + refId: z.string(), + /** 展示名(订阅/作品标题、下载任务名、Feed/站点名、会话目标等);对象已删除时为 null */ + title: z.string().nullable(), + triggeredBy: EngineTriggeredBy.nullable(), +}); +export type EngineRunBusiness = z.infer; + export const EngineRun = z.object({ id: EngineId, workflowName: z.string(), @@ -239,6 +265,8 @@ export const EngineRun = z.object({ finishedAt: Timestamp.nullable(), createdAt: Timestamp, updatedAt: Timestamp, + /** 关联业务对象;推断不出为 null(旧客户端可忽略) */ + business: EngineRunBusiness.nullable().optional(), }); export type EngineRun = z.infer; @@ -251,6 +279,10 @@ export const EngineRunQuery = z.object({ from: Timestamp.optional(), /** created_at 上界(不含) */ to: Timestamp.optional(), + /** 按关联业务对象类型筛选 */ + businessKind: EngineBusinessKind.optional(), + /** 按关联业务对象 id 筛选(可与 businessKind 同用) */ + businessRefId: z.string().min(1).max(200).optional(), }); export type EngineRunQuery = z.infer; @@ -304,3 +336,146 @@ export type EngineStats = z.infer; export const EngineCancelRequest = z.object({ reason: z.string().max(500).optional() }); export const EngineCancelResult = z.object({ canceled: z.boolean() }); + +// ── 进程与健康(W9-I;GET /system/health、/system/health/history,仅 admin) ── + +export const ProcessKind = z.enum(["api", "worker", "worker:agent", "all"]); +export type ProcessKind = z.infer; +/** 三档状态:ok / warn(需要关注)/ critical(需要处理);unknown = 取不到数据 */ +export const HealthLevel = z.enum(["ok", "warn", "critical", "unknown"]); +export type HealthLevel = z.infer; + +export const SystemProcess = z.object({ + /** ::;同一容器重启覆盖同一行 */ + instanceId: z.string(), + processKind: ProcessKind, + pid: z.number().int(), + host: z.string(), + version: z.string(), + startedAt: Timestamp, + lastBeatAt: Timestamp, + /** 距上次心跳秒数 */ + beatAgeSec: z.number().nonnegative(), + /** 超过 45s 没心跳 */ + stale: z.boolean(), + /** 最近一次采样的事件循环延迟 */ + eventLoopLagMs: NonNegInt, + /** 最近一个心跳周期(15s)内的最大延迟 */ + eventLoopLagMaxMs: NonNegInt, + /** 引擎 worker 并发;api 进程为 {} */ + concurrency: z.object({ io: NonNegInt.optional(), agent: NonNegInt.optional() }), + /** 本组任务正在执行的 run 数(同组多实例时是组合计);api 进程为 null */ + runningRuns: NonNegInt.nullable(), + /** 本进程 SSE 长连接数;不托管 HTTP 的进程为 null */ + sseConnections: NonNegInt.nullable(), + /** 其它采样:rssBytes/heapUsedBytes/lagWarnings/lifetimeMaxLagMs/group 等 */ + extra: JsonObject, +}); +export type SystemProcess = z.infer; + +export const SystemDisk = z.object({ + scope: z.enum(["data_dir", "library_root", "download_dir"]), + /** data_dir 为 null */ + id: z.string().nullable(), + name: z.string(), + path: z.string(), + level: HealthLevel, + totalBytes: NonNegInt.nullable(), + usedBytes: NonNegInt.nullable(), + availableBytes: NonNegInt.nullable(), + usedRatio: z.number().min(0).max(1).nullable(), + /** 取不到用量时的错误码/说明 */ + error: z.string().nullable(), +}); +export type SystemDisk = z.infer; + +export const SystemMount = z.object({ + kind: z.enum(["library_root", "path_mapping"]), + id: z.string(), + name: z.string(), + path: z.string(), + /** online=目录与哨兵都在;offline=目录缺失/无权限/哨兵缺失;timeout=检查超时(挂载可能卡死) */ + status: z.enum(["online", "offline", "timeout"]), + /** root_missing / sentinel_missing / permission_denied / error / timeout;online 为 null */ + reason: z.string().nullable(), + /** 是否做了哨兵检查(下载目录映射没有哨兵,只查目录可读) */ + sentinelChecked: z.boolean(), +}); +export type SystemMount = z.infer; + +export const SystemHealth = z.object({ + /** 各段最差档(unknown 不参与) */ + level: HealthLevel, + checkedAt: Timestamp, + /** 处理本请求的 api 进程 */ + api: z.object({ version: z.string(), startedAt: Timestamp, uptimeSec: z.number().nonnegative() }), + processes: z.array(SystemProcess), + sse: z.object({ connections: NonNegInt }), + db: z.object({ + level: HealthLevel, + latencyMs: z.number().nonnegative().nullable(), + sizeBytes: NonNegInt.nullable(), + /** 本库连接数(pg_stat_activity),按状态分 */ + connections: z + .object({ total: NonNegInt, active: NonNegInt, idle: NonNegInt, idleInTransaction: NonNegInt, max: NonNegInt }) + .nullable(), + error: z.string().nullable(), + }), + disks: z.array(SystemDisk), + /** 磁盘阈值(取自设置 cache.resumePercent / cache.hardThresholdPercent) */ + diskThresholds: z.object({ warnPercent: z.number(), criticalPercent: z.number() }), + mounts: z.array(SystemMount), + engine: z.object({ + level: HealthLevel, + pending: NonNegInt, + running: NonNegInt, + oldestPendingAgeSec: z.number().nonnegative().nullable(), + finished24h: NonNegInt, + failed24h: NonNegInt, + /** failed / (completed + failed),24h 内无结束 run 为 null */ + failureRate24h: z.number().min(0).max(1).nullable(), + }), + schedules: z.object({ + level: HealthLevel, + total: NonNegInt, + enabled: NonNegInt, + running: NonNegInt, + /** 最近一次运行失败的已启用定时任务 */ + failing: z.array( + z.object({ id: z.string(), name: z.string(), lastRunAt: Timestamp.nullable(), lastError: z.string().nullable() }), + ), + failedRuns24h: NonNegInt, + }), +}); +export type SystemHealth = z.infer; + +export const SystemHealthHistoryQuery = z.object({ + /** 回看小时数,1–168,缺省 6 */ + hours: z.coerce.number().int().min(1).max(168).default(6), +}); +export type SystemHealthHistoryQuery = z.infer; + +export const SystemHealthHistory = z.object({ + from: Timestamp, + to: Timestamp, + /** 聚合桶宽(秒);点数约 ≤ 360 */ + bucketSec: z.number().int().positive(), + series: z.array( + z.object({ + instanceId: z.string(), + processKind: z.string(), + points: z.array( + z.object({ + at: Timestamp, + /** 桶内平均 */ + lagMs: z.number().nonnegative(), + /** 桶内最大 */ + lagMaxMs: z.number().nonnegative(), + runningRuns: z.number().nonnegative().nullable(), + sseConnections: z.number().nonnegative().nullable(), + }), + ), + }), + ), +}); +export type SystemHealthHistory = z.infer; diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index b1d08e49..80357ad3 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -127,6 +127,9 @@ const INTERNAL_TABLES = [ "AdminSession", "LoginAttempt", "SseEvent", + // W9-I 进程心跳(0034):服务端内部,经 /system/health 的专用 DTO 暴露 + "ProcessHeartbeat", + "ProcessHeartbeatSample", ]; const INTERNAL_ENUMS = ["TasksOpLedgerStatus"]; @@ -209,6 +212,7 @@ const ENUMS: Record = { ReconcileRunMode: C.ReconcileMode, ReconcileRunStatus: C.ReconcileStatus, SecretPurpose: C.SecretPurpose, + ProcessHeartbeatProcessKind: C.ProcessKind, ScheduledJobBoardGroup: C.ScheduleGroup, ScheduledJobKind: C.ScheduleKind, ScheduledJobLastStatus: C.ScheduleRunStatus, diff --git a/packages/db/migrations/0034_process_heartbeat.ts b/packages/db/migrations/0034_process_heartbeat.ts new file mode 100644 index 00000000..5cd14988 --- /dev/null +++ b/packages/db/migrations/0034_process_heartbeat.ts @@ -0,0 +1,54 @@ +import type { Kysely } from "kysely"; +import { dropTables, execDdl, oneOf } from "../src/migration-helpers.ts"; + +// W9-I 进程与健康:各常驻进程(api / worker / worker:agent / all)每 15s upsert 一行心跳, +// 同时追加一条采样到 process_heartbeat_sample 供 GET /system/health/history 画图(保留 7 天,写入方顺手裁剪)。 +// instance_id = :::同一容器重启覆盖旧行,不会越积越多;超过 45s 没心跳由读取方标 stale。 +// 另给 task.engine_run_id 补索引:引擎 runs 列表按业务字段筛选时按 run id 反查业务任务行。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +create table process_heartbeat ( + instance_id text primary key, + process_kind text not null ${oneOf("process_kind", ["api", "worker", "worker:agent", "all"])}, + pid integer not null, + host text not null, + version text not null, + started_at timestamptz not null, + last_beat_at timestamptz not null default now(), + event_loop_lag_ms integer not null default 0 check (event_loop_lag_ms >= 0), + event_loop_lag_max_ms integer not null default 0 check (event_loop_lag_max_ms >= 0), + concurrency jsonb not null default '{}'::jsonb, + running_runs integer check (running_runs is null or running_runs >= 0), + extra jsonb not null default '{}'::jsonb +); +create index process_heartbeat_beat_idx on process_heartbeat (last_beat_at); + +create table process_heartbeat_sample ( + id bigserial primary key, + instance_id text not null, + process_kind text not null, + at timestamptz not null default now(), + event_loop_lag_ms integer not null default 0, + event_loop_lag_max_ms integer not null default 0, + running_runs integer, + sse_connections integer +); +create index process_heartbeat_sample_at_idx on process_heartbeat_sample (at, instance_id); + +create index if not exists task_engine_run_idx on task (engine_run_id) where engine_run_id is not null; +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `drop index if exists task_engine_run_idx; +${dropTables(["process_heartbeat_sample", "process_heartbeat"])}`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index d2f35280..3b48547f 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -194,6 +194,7 @@ export type PolicyRevisionStatus = "draft" | "published" | "superseded" | "rolle export type PreferenceTagKind = "soft" | "hard" | "version" | "target"; export type PreferenceTagOrigin = "manual" | "conversation" | "agent_inferred"; export type PreferenceTagStatus = "active" | "disabled" | "deleted"; +export type ProcessHeartbeatProcessKind = "api" | "worker" | "worker:agent" | "all"; export type ReconcileRunKind = "identity" | "metadata" | "asset" | "acquisition" | "server"; export type ReconcileRunMode = "normal" | "read_only"; export type ReconcileRunStatus = "queued" | "running" | "succeeded" | "partial" | "failed" | "cancelled"; @@ -1079,6 +1080,32 @@ export interface PreferenceTagTable { updated_at: GeneratedTimestamptz; } +export interface ProcessHeartbeatTable { + instance_id: string; + process_kind: ProcessHeartbeatProcessKind; + pid: number; + host: string; + version: string; + started_at: Timestamptz; + last_beat_at: GeneratedTimestamptz; + event_loop_lag_ms: Generated; + event_loop_lag_max_ms: Generated; + concurrency: Generated; + running_runs: number | null; + extra: Generated; +} + +export interface ProcessHeartbeatSampleTable { + id: Generated; + instance_id: string; + process_kind: string; + at: GeneratedTimestamptz; + event_loop_lag_ms: Generated; + event_loop_lag_max_ms: Generated; + running_runs: number | null; + sse_connections: number | null; +} + export interface ReconcileRunTable { id: Generated; kind: ReconcileRunKind; @@ -1558,6 +1585,8 @@ export interface DB { policy_revision: PolicyRevisionTable; preference_suppression: PreferenceSuppressionTable; preference_tag: PreferenceTagTable; + process_heartbeat: ProcessHeartbeatTable; + process_heartbeat_sample: ProcessHeartbeatSampleTable; reconcile_run: ReconcileRunTable; resolution_attempt: ResolutionAttemptTable; rule_atom: RuleAtomTable; @@ -1723,6 +1752,12 @@ export type PreferenceSuppressionUpdate = Updateable export type PreferenceTagRow = Selectable; export type NewPreferenceTag = Insertable; export type PreferenceTagUpdate = Updateable; +export type ProcessHeartbeatRow = Selectable; +export type NewProcessHeartbeat = Insertable; +export type ProcessHeartbeatUpdate = Updateable; +export type ProcessHeartbeatSampleRow = Selectable; +export type NewProcessHeartbeatSample = Insertable; +export type ProcessHeartbeatSampleUpdate = Updateable; export type ReconcileRunRow = Selectable; export type NewReconcileRun = Insertable; export type ReconcileRunUpdate = Updateable; diff --git a/packages/db/test/schema.test.ts b/packages/db/test/schema.test.ts index 5e31c387..92fc935b 100644 --- a/packages/db/test/schema.test.ts +++ b/packages/db/test/schema.test.ts @@ -102,6 +102,9 @@ const EXPECTED_TABLES = [ // W9-D 规则组(0033) "rule_group", "rule_atom", + // W9-I 进程心跳(0034) + "process_heartbeat", + "process_heartbeat_sample", ].sort(); /** 不走 entityTable() 的表:追加型/引擎自有表,没有 uuid id + created_at/updated_at 三件套 */ @@ -118,6 +121,9 @@ const NON_ENTITY_TABLES = [ // 稳定 key 作主键 / 只追加的运行历史 "scheduled_job", "scheduled_job_run", + // 进程心跳:instance_id 主键 / 只追加的采样 + "process_heartbeat", + "process_heartbeat_sample", ]; describe.skipIf(skip)("schema migrations (real PG)", () => {