490 lines
18 KiB
TypeScript
490 lines
18 KiB
TypeScript
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, type Selectable, sql } from "@umt/db";
|
|
import { CURRENT_HOST_WINDOW_MS, HEARTBEAT_ROW_RETENTION_MS, HEARTBEAT_STALE_MS } from "./heartbeat.ts";
|
|
|
|
// GET /system/health 各段的取数(W9-I)。每段独立 try/catch:一段失败只把该段标 unknown,不让整页 500。
|
|
// 慢 IO(statfs / 哨兵检查)都带超时:挂载卡死时只标 timeout,不拖住请求。
|
|
|
|
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;
|
|
/** 存储段(CACHE-2);不给时该段为 unknown */
|
|
storage?: () => Promise<{ level: HealthLevel; alerts: string[] }>;
|
|
}
|
|
|
|
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));
|
|
|
|
// ── 进程 ──
|
|
|
|
type HeartbeatRow = Selectable<DB["process_heartbeat"]>;
|
|
|
|
function toProcess(r: HeartbeatRow, now: Date): SystemProcess {
|
|
const beat = toDate(r.last_beat_at);
|
|
const age = Math.max(0, now.getTime() - beat.getTime());
|
|
const extra = (r.extra ?? {}) as Record<string, unknown>;
|
|
const conc = (r.concurrency ?? {}) as { io?: number; agent?: number };
|
|
const sse = extra.sseConnections;
|
|
return {
|
|
instanceId: r.instance_id,
|
|
processKind: r.process_kind,
|
|
pid: r.pid,
|
|
host: r.host,
|
|
version: r.version,
|
|
startedAt: toDate(r.started_at).toISOString(),
|
|
lastBeatAt: beat.toISOString(),
|
|
beatAgeSec: Math.round(age / 100) / 10,
|
|
stale: age > HEARTBEAT_STALE_MS,
|
|
stoppedAt: r.stopped_at ? toDate(r.stopped_at).toISOString() : null,
|
|
eventLoopLagMs: r.event_loop_lag_ms,
|
|
eventLoopLagMaxMs: r.event_loop_lag_max_ms,
|
|
concurrency: {
|
|
...(typeof conc.io === "number" ? { io: conc.io } : {}),
|
|
...(typeof conc.agent === "number" ? { agent: conc.agent } : {}),
|
|
},
|
|
runningRuns: r.running_runs,
|
|
sseConnections: typeof sse === "number" ? sse : null,
|
|
extra,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* 把心跳行分成当前实例与已下线实例:
|
|
* - 当前 host = 最近 10 分钟内有心跳且没标 stopped_at 的行所在 host(容器重建后 host 会变);
|
|
* - 当前实例 = 当前 host 上、没标 stopped_at、且是同 host 同 kind 里最新启动的那一行;
|
|
* - 其余(已正常退出、host 已不在、被同 host 新进程替换)都算已下线实例,不参与 stale 判定与告警。
|
|
*/
|
|
export function splitProcesses(
|
|
ps: SystemProcess[],
|
|
now: Date,
|
|
): { processes: SystemProcess[]; retiredProcesses: SystemProcess[] } {
|
|
const alive = ps.filter((p) => !p.stoppedAt);
|
|
const hosts = new Set(
|
|
alive.filter((p) => now.getTime() - Date.parse(p.lastBeatAt) <= CURRENT_HOST_WINDOW_MS).map((p) => p.host),
|
|
);
|
|
const newest = new Map<string, SystemProcess>();
|
|
for (const p of alive) {
|
|
if (!hosts.has(p.host)) continue;
|
|
const k = `${p.host}\u0000${p.processKind}`;
|
|
const cur = newest.get(k);
|
|
if (!cur || Date.parse(p.startedAt) > Date.parse(cur.startedAt)) newest.set(k, p);
|
|
}
|
|
const current = new Set([...newest.values()].map((p) => p.instanceId));
|
|
const processes = ps.filter((p) => current.has(p.instanceId));
|
|
const retiredProcesses = ps
|
|
.filter((p) => !current.has(p.instanceId))
|
|
.sort((a, b) => Date.parse(b.lastBeatAt) - Date.parse(a.lastBeatAt));
|
|
return { processes, retiredProcesses };
|
|
}
|
|
|
|
export async function loadProcesses(
|
|
db: Kysely<DB>,
|
|
now: Date,
|
|
): Promise<{ processes: SystemProcess[]; retiredProcesses: SystemProcess[] }> {
|
|
const rows = await db
|
|
.selectFrom("process_heartbeat")
|
|
.selectAll()
|
|
.where("last_beat_at", ">", new Date(now.getTime() - HEARTBEAT_ROW_RETENTION_MS))
|
|
.orderBy("process_kind")
|
|
.orderBy("instance_id")
|
|
.execute();
|
|
return splitProcesses(
|
|
rows.map((r) => toProcess(r, now)),
|
|
now,
|
|
);
|
|
}
|
|
|
|
/** 进程段的档位(只看当前实例):有 stale 或延迟 ≥ 1s 为 warn;延迟 ≥ 5s 为 critical;没有任何 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, retiredProcesses }, db, paths, engine, schedules, storage] = await Promise.all([
|
|
loadProcesses(d.db, now).catch(() => ({
|
|
processes: [] as SystemProcess[],
|
|
retiredProcesses: [] as SystemProcess[],
|
|
})),
|
|
loadDb(d.db),
|
|
loadDisksAndMounts(d).catch(() => ({ disks: [] as SystemDisk[], mounts: [] as SystemMount[] })),
|
|
loadEngine(d.db, d.engineSchema, now).catch((): SystemHealth["engine"] => ({
|
|
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,
|
|
})),
|
|
(d.storage?.() ?? Promise.resolve({ level: "unknown" as HealthLevel, alerts: [] })).catch(() => ({
|
|
level: "unknown" as HealthLevel,
|
|
alerts: [] as string[],
|
|
})),
|
|
]);
|
|
// 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,
|
|
storage.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,
|
|
retiredProcesses,
|
|
sse: { connections: sse },
|
|
db,
|
|
disks: paths.disks,
|
|
diskThresholds: d.thresholds,
|
|
mounts: paths.mounts,
|
|
engine,
|
|
schedules,
|
|
storage,
|
|
};
|
|
}
|
|
|
|
// ── 历史 ──
|
|
|
|
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()] };
|
|
}
|