From 2ce18416284cc7fae321ab0c4c4bf48cd85e62af Mon Sep 17 00:00:00 2001 From: root Date: Tue, 6 Oct 2026 17:58:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20=E8=B5=84=E6=BA=90=E7=9B=91?= =?UTF-8?q?=E6=8E=A7=E2=80=94=E2=80=94/proc=20=E4=B8=8E=20cgroup=20?= =?UTF-8?q?=E9=87=87=E6=A0=B7=E3=80=81/system/resources=20=E8=B6=8B?= =?UTF-8?q?=E5=8A=BF=EF=BC=8C=E5=BF=83=E8=B7=B3=E5=B8=A6=E8=BF=9B=E7=A8=8B?= =?UTF-8?q?=20CPU?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- apps/server/src/modules/system/heartbeat.ts | 18 +- apps/server/src/modules/system/index.ts | 14 + apps/server/src/modules/system/resources.ts | 425 ++++++++++++++++++++ apps/server/test/system-resources.test.ts | 206 ++++++++++ apps/server/test/system/system.test.ts | 50 ++- apps/server/vitest.config.ts | 1 + 6 files changed, 712 insertions(+), 2 deletions(-) create mode 100644 apps/server/src/modules/system/resources.ts create mode 100644 apps/server/test/system-resources.test.ts diff --git a/apps/server/src/modules/system/heartbeat.ts b/apps/server/src/modules/system/heartbeat.ts index 00cf5f42..860f52d3 100644 --- a/apps/server/src/modules/system/heartbeat.ts +++ b/apps/server/src/modules/system/heartbeat.ts @@ -3,6 +3,7 @@ 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" }; +import { createProcessCpuMeter } from "./resources.ts"; // 进程心跳写入(W9-I):每个常驻进程每 15s upsert process_heartbeat 一行 + 追加一条采样。 // 写失败只记 debug(库暂时不可达不影响业务;读取方按 last_beat_at 判 stale)。 @@ -86,6 +87,11 @@ function memory(): Record { } } +/** 本进程两次心跳之间的 CPU 占用(% of 一个核);取不到时不写 */ +function cpuExtra(pct: number | null): Record { + return pct == null ? {} : { cpuPercent: pct }; +} + /** 写一次心跳(upsert + 采样);返回写入的 instanceId */ export async function writeHeartbeat( db: Kysely, @@ -160,6 +166,8 @@ export function startHeartbeatWriter(o: HeartbeatWriterOptions): HeartbeatWriter const intervalMs = o.intervalMs ?? HEARTBEAT_WRITE_MS; const lag = o.lag ?? createLagSampler(); const log = o.logger.child({ component: "heartbeat", kind: o.kind }); + const cpu = createProcessCpuMeter(); + cpu(); let inflight: Promise | null = null; let stopped = false; @@ -181,7 +189,15 @@ export function startHeartbeatWriter(o: HeartbeatWriterOptions): HeartbeatWriter startedAt, lagMs: l.lastMs, lagMaxMs: l.windowMaxMs, - sample: { ...sample, extra: { ...memory(), lifetimeMaxLagMs: l.lifetimeMaxMs, ...(sample.extra ?? {}) } }, + sample: { + ...sample, + extra: { + ...memory(), + ...cpuExtra(cpu()), + lifetimeMaxLagMs: l.lifetimeMaxMs, + ...(sample.extra ?? {}), + }, + }, }); } catch (err) { log.debug({ err }, "心跳写入失败"); diff --git a/apps/server/src/modules/system/index.ts b/apps/server/src/modules/system/index.ts index 1fbbd4f0..1d38ccca 100644 --- a/apps/server/src/modules/system/index.ts +++ b/apps/server/src/modules/system/index.ts @@ -25,6 +25,7 @@ import { safeHostname, startHeartbeatWriter, } from "./heartbeat.ts"; +import { createResourceSampler, type ResourceSampler } from "./resources.ts"; import { type BatchOptions, CLEANUP_JOB_IDS, RETENTION_JOBS } from "./retention.ts"; import { buildStorage, type StoragePaths, storageHealth } from "./storage.ts"; @@ -58,6 +59,8 @@ export interface SystemModuleOptions { retentionJobs?: boolean; /** 保留清理的分批参数(测试) */ retentionBatch?: BatchOptions; + /** 资源采样器(测试注入);false 不起;缺省 api / all 角色注册路由时起一个每 5 秒读 /proc 的采样器 */ + resourceSampler?: ResourceSampler | false; } type Infer = T extends z.ZodType ? z.infer : Record; @@ -100,6 +103,8 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule } { const startedAt = new Date(); let writer: HeartbeatWriter | null = null; + let sampler: ResourceSampler | null = null; + const ownSampler = !opts.resourceSampler; return { name: "system", apiHeartbeat: () => writer, @@ -113,6 +118,8 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule sample: () => ({ sseConnections: sseConnectionCount() }), }); } + if (opts.resourceSampler) sampler = opts.resourceSampler; + else if (opts.resourceSampler !== false && !sampler) sampler = createResourceSampler(); const selfKind = ctx.role === "all" ? "all" : ctx.role === "api" ? "api" : null; const selfId = selfKind ? instanceIdOf(selfKind, safeHostname(), Deno.pid) : undefined; const paths = (): StoragePaths => { @@ -140,6 +147,11 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule }); }); route(app, Api.system.healthHistory, async ({ query }) => await loadHistory(ctx.db, query.hours)); + route(app, Api.system.resources, async ({ query }) => { + // 采样器只在注册路由的进程里起;测试关掉时临时起一个不自动采样的 + sampler ??= createResourceSampler({ autoStart: false }); + return await sampler.snapshot(query.minutes); + }); route(app, Api.system.storage, async () => buildStorage({ db: ctx.db, @@ -226,6 +238,8 @@ export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule async stop() { await writer?.stop(); writer = null; + if (ownSampler) sampler?.stop(); + sampler = null; }, }; } diff --git a/apps/server/src/modules/system/resources.ts b/apps/server/src/modules/system/resources.ts new file mode 100644 index 00000000..a6f17796 --- /dev/null +++ b/apps/server/src/modules/system/resources.ts @@ -0,0 +1,425 @@ +import { availableParallelism, cpus } from "node:os"; +import process from "node:process"; +import type { ResourceSample, ResourceScope, SystemResources } from "@umt/contracts"; + +// 资源监控(界面第二轮 V 线,参考 MP 仪表盘 CPU / 内存 / 网络组件):api 进程每 5 秒读一次 /proc 与 cgroup v2, +// 相邻两次读数的差算速率,采样只放内存(环形,保留 6 小时)。读不到的文件按 null 降级,不报错。 +// 口径:容器里优先 cgroup(本容器 CPU 占用、内存、磁盘读写);网络读 /proc/net/dev(容器网络命名空间即本容器); +// 宿主没有 cgroup 文件时退回 /proc 整机。 + +export const RESOURCE_SAMPLE_MS = 5_000; +export const RESOURCE_RETENTION_MS = 6 * 3600_000; +/** 趋势最多返回的点数 */ +const MAX_POINTS = 360; + +export type ReadText = (path: string) => Promise; + +export const defaultReadText: ReadText = async (path) => { + try { + return await Deno.readTextFile(path); + } catch { + return null; + } +}; + +// ── 解析(纯函数) ── + +/** /proc/stat 第一行 cpu:total = 前 8 项之和,idle = idle + iowait */ +export function parseProcStat(text: string | null): { total: number; idle: number } | null { + const line = text?.split("\n").find((l) => l.startsWith("cpu ")); + if (!line) return null; + const v = line.trim().split(/\s+/).slice(1, 9).map(Number); + if (v.length < 4 || v.some((n) => !Number.isFinite(n))) return null; + const total = v.reduce((s, n) => s + n, 0); + return { total, idle: (v[3] ?? 0) + (v[4] ?? 0) }; +} + +/** /proc/meminfo(kB)→ 字节 */ +export function parseMeminfo(text: string | null): { + totalBytes: number; + availableBytes: number; + swapUsedBytes: number | null; +} | null { + if (!text) return null; + const kv = new Map(); + for (const l of text.split("\n")) { + const m = /^(\w+):\s+(\d+)/.exec(l); + if (m) kv.set(m[1]!, Number(m[2]) * 1024); + } + const total = kv.get("MemTotal"); + const avail = kv.get("MemAvailable") ?? (kv.get("MemFree") ?? 0) + (kv.get("Cached") ?? 0); + if (!total) return null; + const st = kv.get("SwapTotal"); + const sf = kv.get("SwapFree"); + return { + totalBytes: total, + availableBytes: Math.min(total, avail), + swapUsedBytes: st != null && sf != null ? Math.max(0, st - sf) : null, + }; +} + +export function parseLoadavg(text: string | null): [number, number, number] | null { + const v = text?.trim().split(/\s+/).slice(0, 3).map(Number); + if (!v || v.length < 3 || v.some((n) => !Number.isFinite(n))) return null; + return [v[0]!, v[1]!, v[2]!]; +} + +/** /proc/net/dev:除 lo 外各接口收 / 发字节合计 */ +export function parseNetDev(text: string | null): { rx: number; tx: number } | null { + if (!text) return null; + let rx = 0; + let tx = 0; + let seen = false; + for (const l of text.split("\n")) { + const m = /^\s*([^:\s]+):\s*(.*)$/.exec(l); + if (!m || m[1] === "lo") continue; + const f = m[2]!.trim().split(/\s+/).map(Number); + if (f.length < 9) continue; + rx += f[0] ?? 0; + tx += f[8] ?? 0; + seen = true; + } + return seen ? { rx, tx } : null; +} + +/** cgroup v2 cpu.stat 的 usage_usec */ +export function parseCgroupCpuUsage(text: string | null): number | null { + const m = text ? /^usage_usec\s+(\d+)/m.exec(text) : null; + return m ? Number(m[1]) : null; +} + +/** cgroup v2 cpu.max:"max 100000" → null(不限);"200000 100000" → 2 核 */ +export function parseCgroupCpuMax(text: string | null): number | null { + const [quota, period] = text?.trim().split(/\s+/) ?? []; + if (!quota || quota === "max" || !period) return null; + const q = Number(quota); + const p = Number(period); + return q > 0 && p > 0 ? q / p : null; +} + +/** memory.max:"max" → null */ +export function parseCgroupBytes(text: string | null): number | null { + const t = text?.trim(); + if (!t || t === "max") return null; + const n = Number(t); + return Number.isFinite(n) ? n : null; +} + +/** memory.stat 的某个字段 */ +export function parseStatField(text: string | null, key: string): number | null { + const m = text ? new RegExp(`^${key}\\s+(\\d+)`, "m").exec(text) : null; + return m ? Number(m[1]) : null; +} + +/** cgroup v2 io.stat:各设备 rbytes / wbytes 合计 */ +export function parseIoStat(text: string | null): { read: number; write: number } | null { + if (!text?.trim()) return null; + let read = 0; + let write = 0; + let seen = false; + for (const l of text.split("\n")) { + const r = /\brbytes=(\d+)/.exec(l); + const w = /\bwbytes=(\d+)/.exec(l); + if (r || w) seen = true; + read += r ? Number(r[1]) : 0; + write += w ? Number(w[1]) : 0; + } + return seen ? { read, write } : null; +} + +const WHOLE_DISK = /^(sd[a-z]+|vd[a-z]+|xvd[a-z]+|hd[a-z]+|nvme\d+n\d+|mmcblk\d+)$/; + +/** /proc/diskstats:整盘(不含分区、loop、ram)读 / 写扇区 × 512 */ +export function parseDiskstats(text: string | null): { read: number; write: number } | null { + if (!text) return null; + let read = 0; + let write = 0; + let seen = false; + for (const l of text.split("\n")) { + const f = l.trim().split(/\s+/); + if (f.length < 10 || !WHOLE_DISK.test(f[2] ?? "")) continue; + read += Number(f[5]) * 512; + write += Number(f[9]) * 512; + seen = true; + } + return seen ? { read, write } : null; +} + +// ── 读数与采样 ── + +export interface RawReading { + atMs: number; + host: { total: number; idle: number } | null; + cgroupCpuUsec: number | null; + cgroupCores: number | null; + onlineCores: number; + load: [number, number, number] | null; + meminfo: ReturnType; + cgroupMem: { current: number; inactiveFile: number; max: number | null } | null; + net: { rx: number; tx: number } | null; + io: { read: number; write: number; scope: ResourceScope } | null; + uptimeSec: number | null; +} + +const CG = "/sys/fs/cgroup"; + +function onlineCores(): number { + try { + return availableParallelism(); + } catch { + return cpus().length || 1; + } +} + +export async function readResources(read: ReadText = defaultReadText, now = Date.now()): Promise { + const [stat, cpuStat, cpuMax, load, mem, memCur, memMax, memStat, net, ioStat, diskstats, uptime] = await Promise.all( + [ + read("/proc/stat"), + read(`${CG}/cpu.stat`), + read(`${CG}/cpu.max`), + read("/proc/loadavg"), + read("/proc/meminfo"), + read(`${CG}/memory.current`), + read(`${CG}/memory.max`), + read(`${CG}/memory.stat`), + read("/proc/net/dev"), + read(`${CG}/io.stat`), + read("/proc/diskstats"), + read("/proc/uptime"), + ], + ); + const current = parseCgroupBytes(memCur); + const cgIo = parseIoStat(ioStat); + const hostIo = cgIo ? null : parseDiskstats(diskstats); + const up = Number(uptime?.trim().split(/\s+/)[0]); + return { + atMs: now, + host: parseProcStat(stat), + cgroupCpuUsec: parseCgroupCpuUsage(cpuStat), + cgroupCores: parseCgroupCpuMax(cpuMax), + onlineCores: onlineCores(), + load: parseLoadavg(load), + meminfo: parseMeminfo(mem), + cgroupMem: + current == null + ? null + : { + current, + inactiveFile: parseStatField(memStat, "inactive_file") ?? 0, + max: parseCgroupBytes(memMax), + }, + net: parseNetDev(net), + io: cgIo ? { ...cgIo, scope: "cgroup" } : hostIo ? { ...hostIo, scope: "host" } : null, + uptimeSec: Number.isFinite(up) ? up : null, + }; +} + +const round1 = (n: number) => Math.round(n * 10) / 10; +const clampPct = (n: number) => round1(Math.min(100, Math.max(0, n))); +const rate = (cur: number, prev: number, sec: number) => (sec > 0 && cur >= prev ? (cur - prev) / sec : null); + +export interface ResourcePoint { + sample: ResourceSample; + current: Omit; +} + +/** 由相邻两次读数算一个点(prev 为 null 时速率与 CPU 为 null) */ +export function computePoint(prev: RawReading | null, cur: RawReading): ResourcePoint { + const sec = prev ? (cur.atMs - prev.atMs) / 1000 : 0; + const hostPercent = + prev?.host && cur.host && cur.host.total > prev.host.total + ? clampPct((1 - (cur.host.idle - prev.host.idle) / (cur.host.total - prev.host.total)) * 100) + : null; + const cores = cur.cgroupCores ?? cur.onlineCores; + const cgPercent = + prev && prev.cgroupCpuUsec != null && cur.cgroupCpuUsec != null && sec > 0 + ? clampPct(((cur.cgroupCpuUsec - prev.cgroupCpuUsec) / (sec * 1e6 * cores)) * 100) + : null; + const cpuScope: ResourceScope = cur.cgroupCpuUsec != null ? "cgroup" : cur.host ? "host" : "none"; + const cpuPercent = cpuScope === "cgroup" ? cgPercent : hostPercent; + + const mi = cur.meminfo; + const hostUsed = mi ? mi.totalBytes - mi.availableBytes : null; + const cg = cur.cgroupMem; + const memScope: ResourceScope = cg ? "cgroup" : mi ? "host" : "none"; + const used = cg ? Math.max(0, cg.current - cg.inactiveFile) : hostUsed; + const limit = cg ? (cg.max ?? mi?.totalBytes ?? null) : (mi?.totalBytes ?? null); + const memPercent = used != null && limit ? clampPct((used / limit) * 100) : null; + + const rx = prev?.net && cur.net ? rate(cur.net.rx, prev.net.rx, sec) : null; + const tx = prev?.net && cur.net ? rate(cur.net.tx, prev.net.tx, sec) : null; + const sameIo = prev?.io && cur.io && prev.io.scope === cur.io.scope; + const rd = sameIo ? rate(cur.io!.read, prev!.io!.read, sec) : null; + const wr = sameIo ? rate(cur.io!.write, prev!.io!.write, sec) : null; + + return { + sample: { + at: new Date(cur.atMs).toISOString(), + cpuPercent, + memoryUsedBytes: used, + memoryPercent: memPercent, + netRxBps: rx == null ? null : Math.round(rx), + netTxBps: tx == null ? null : Math.round(tx), + diskReadBps: rd == null ? null : Math.round(rd), + diskWriteBps: wr == null ? null : Math.round(wr), + }, + current: { + cpu: { scope: cpuScope, percent: cpuPercent, cores: round1(cores), hostPercent, load: cur.load }, + memory: { + scope: memScope, + usedBytes: used, + limitBytes: limit, + percent: memPercent, + hostTotalBytes: mi?.totalBytes ?? null, + hostAvailableBytes: mi?.availableBytes ?? null, + swapUsedBytes: mi?.swapUsedBytes ?? null, + }, + network: { + scope: cur.net ? (cg ? "cgroup" : "host") : "none", + rxBps: rx == null ? null : Math.round(rx), + txBps: tx == null ? null : Math.round(tx), + rxTotalBytes: cur.net?.rx ?? null, + txTotalBytes: cur.net?.tx ?? null, + }, + diskIo: { + scope: cur.io?.scope ?? "none", + readBps: rd == null ? null : Math.round(rd), + writeBps: wr == null ? null : Math.round(wr), + }, + hostUptimeSec: cur.uptimeSec, + }, + }; +} + +/** 按桶取平均(null 不参与);点数不超过 MAX_POINTS */ +export function bucketSamples( + samples: ResourceSample[], + intervalSec: number, + minutes: number, +): { bucketSec: number; points: ResourceSample[] } { + const bucketSec = Math.max(intervalSec, Math.ceil((minutes * 60) / MAX_POINTS / intervalSec) * intervalSec); + if (bucketSec <= intervalSec) return { bucketSec: intervalSec, points: samples }; + const keys: (keyof ResourceSample)[] = [ + "cpuPercent", + "memoryUsedBytes", + "memoryPercent", + "netRxBps", + "netTxBps", + "diskReadBps", + "diskWriteBps", + ]; + const buckets = new Map(); + for (const s of samples) { + const k = Math.floor(Date.parse(s.at) / 1000 / bucketSec) * bucketSec; + const arr = buckets.get(k) ?? []; + arr.push(s); + buckets.set(k, arr); + } + const points = [...buckets.entries()] + .sort((a, b) => a[0] - b[0]) + .map(([k, arr]) => { + const out = { at: new Date(k * 1000).toISOString() } as ResourceSample; + for (const key of keys) { + const vals = arr.map((s) => s[key]).filter((v): v is number => typeof v === "number"); + (out as Record)[key] = vals.length + ? Math.round((vals.reduce((a, b) => a + b, 0) / vals.length) * 10) / 10 + : null; + } + return out; + }); + return { bucketSec, points }; +} + +export interface ResourceSampler { + /** 立即采一次(测试 / 首次请求用) */ + tick(): Promise; + snapshot(minutes: number): Promise; + stop(): void; +} + +export function createResourceSampler( + o: { read?: ReadText; intervalMs?: number; retentionMs?: number; now?: () => number; autoStart?: boolean } = {}, +): ResourceSampler { + const read = o.read ?? defaultReadText; + const intervalMs = o.intervalMs ?? RESOURCE_SAMPLE_MS; + const retentionMs = o.retentionMs ?? RESOURCE_RETENTION_MS; + const now = o.now ?? Date.now; + let prev: RawReading | null = null; + let last: ResourcePoint | null = null; + const ring: ResourceSample[] = []; + let inflight: Promise | null = null; + + const tick = () => { + inflight ??= (async () => { + const cur = await readResources(read, now()); + const p = computePoint(prev, cur); + prev = cur; + last = p; + // 第一次读数没有速率,不进趋势 + if (p.sample.cpuPercent != null || p.sample.netRxBps != null || ring.length) ring.push(p.sample); + const cutoff = now() - retentionMs; + while (ring.length && Date.parse(ring[0]!.at) < cutoff) ring.shift(); + })().finally(() => { + inflight = null; + }); + return inflight; + }; + + let timer: ReturnType | null = null; + if (o.autoStart !== false) { + void tick(); + timer = setInterval(() => void tick(), intervalMs); + try { + Deno.unrefTimer(timer); + } catch { + // ignore + } + } + + return { + tick, + async snapshot(minutes) { + // 还没有两次读数(刚启动):现采两次,间隔 300ms,让当前值有速率 + if (!last || !ring.length) { + await tick(); + if (!ring.length) { + await new Promise((r) => setTimeout(r, 300)); + await tick(); + } + } + const intervalSec = intervalMs / 1000; + const from = now() - minutes * 60_000; + const { bucketSec, points } = bucketSamples( + ring.filter((s) => Date.parse(s.at) >= from), + intervalSec, + minutes, + ); + return { at: new Date(now()).toISOString(), intervalSec, bucketSec, ...last!.current, samples: points }; + }, + stop() { + if (timer) clearInterval(timer); + timer = null; + }, + }; +} + +// ── 进程自身 CPU(心跳里带上,运行状态的进程表展示) ── + +/** 返回一个取样函数:每次调用给出距上次调用期间本进程 CPU 占用(% of 一个核);首次为 null */ +export function createProcessCpuMeter(): () => number | null { + let last: { usage: number; at: number } | null = null; + return () => { + try { + // node:process 兼容层的 cpuUsage(微秒) + const u = process.cpuUsage(); + const usage = u.user + u.system; + const at = performance.now(); + const prev = last; + last = { usage, at }; + if (!prev || at <= prev.at) return null; + return round1(((usage - prev.usage) / ((at - prev.at) * 1000)) * 100); + } catch { + return null; + } + }; +} diff --git a/apps/server/test/system-resources.test.ts b/apps/server/test/system-resources.test.ts new file mode 100644 index 00000000..c368d13c --- /dev/null +++ b/apps/server/test/system-resources.test.ts @@ -0,0 +1,206 @@ +import { SystemResources } from "@umt/contracts"; +import { describe, expect, it } from "vitest"; +import { + bucketSamples, + computePoint, + createProcessCpuMeter, + createResourceSampler, + parseCgroupBytes, + parseCgroupCpuMax, + parseCgroupCpuUsage, + parseDiskstats, + parseIoStat, + parseLoadavg, + parseMeminfo, + parseNetDev, + parseProcStat, + type ReadText, + readResources, +} from "../src/modules/system/resources.ts"; + +// 资源监控采集:/proc 与 cgroup v2 的解析、速率计算、降级与采样器(不碰 PG;读文件用注入的假实现)。 + +const STAT = (user: number, idle: number) => `cpu ${user} 0 100 ${idle} 50 0 0 0 0 0\ncpu0 1 2 3 4 5 6 7 8 0 0\n`; +const MEMINFO = `MemTotal: 16000000 kB +MemFree: 2000000 kB +MemAvailable: 8000000 kB +SwapTotal: 1000000 kB +SwapFree: 600000 kB +`; +const NETDEV = (rx: number, tx: number) => `Inter-| Receive | Transmit + face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed + lo: 999999 10 0 0 0 0 0 0 999999 10 0 0 0 0 0 0 + eth0: ${rx} 100 0 0 0 0 0 0 ${tx} 80 0 0 0 0 0 0 +`; +const IOSTAT = (r: number, w: number) => + `8:0 rbytes=${r} wbytes=${w} rios=10 wios=5 dbytes=0 dios=0\n259:0 rbytes=0 wbytes=0\n`; + +/** 假文件系统:按路径给内容,没给的读不到 */ +function fakeFs(files: Record): ReadText { + return (path) => Promise.resolve(files[path] ?? null); +} + +describe("解析", () => { + it("/proc 各文件", () => { + expect(parseProcStat(STAT(1000, 8000))).toEqual({ total: 1000 + 100 + 8000 + 50, idle: 8050 }); + expect(parseProcStat("garbage")).toBeNull(); + expect(parseMeminfo(MEMINFO)).toEqual({ + totalBytes: 16000000 * 1024, + availableBytes: 8000000 * 1024, + swapUsedBytes: 400000 * 1024, + }); + expect(parseMeminfo(null)).toBeNull(); + expect(parseLoadavg("0.52 0.40 0.31 2/345 6789")).toEqual([0.52, 0.4, 0.31]); + expect(parseLoadavg("")).toBeNull(); + // lo 不计 + expect(parseNetDev(NETDEV(5000, 3000))).toEqual({ rx: 5000, tx: 3000 }); + // 整盘才计(分区、loop 不计) + expect( + parseDiskstats( + " 8 0 sda 100 0 2000 0 50 0 4000 0 0 0 0\n 8 1 sda1 100 0 2000 0 50 0 4000 0 0 0 0\n 7 0 loop0 1 0 8 0 0 0 0 0 0 0 0\n", + ), + ).toEqual({ read: 2000 * 512, write: 4000 * 512 }); + }); + + it("cgroup v2", () => { + expect(parseCgroupCpuUsage("usage_usec 123456\nuser_usec 1\n")).toBe(123456); + expect(parseCgroupCpuMax("max 100000")).toBeNull(); + expect(parseCgroupCpuMax("200000 100000\n")).toBe(2); + expect(parseCgroupBytes("max\n")).toBeNull(); + expect(parseCgroupBytes("1048576\n")).toBe(1048576); + expect(parseIoStat(IOSTAT(4096, 8192))).toEqual({ read: 4096, write: 8192 }); + expect(parseIoStat("")).toBeNull(); + }); +}); + +describe("读数与速率", () => { + it("容器:CPU 按 cgroup 用量 / 核数,内存扣掉 inactive_file,磁盘读写取 io.stat", async () => { + const files = (usage: number, user: number, idle: number, rx: number, io: number) => + fakeFs({ + "/proc/stat": STAT(user, idle), + "/sys/fs/cgroup/cpu.stat": `usage_usec ${usage}\n`, + "/sys/fs/cgroup/cpu.max": "200000 100000", + "/proc/loadavg": "1.00 0.50 0.25 1/2 3", + "/proc/meminfo": MEMINFO, + "/sys/fs/cgroup/memory.current": String(600 * 1024 * 1024), + "/sys/fs/cgroup/memory.max": String(1024 * 1024 * 1024), + "/sys/fs/cgroup/memory.stat": `anon 1\ninactive_file ${100 * 1024 * 1024}\n`, + "/proc/net/dev": NETDEV(rx, rx / 2), + "/sys/fs/cgroup/io.stat": IOSTAT(io, io * 2), + "/proc/uptime": "3600.5 7000.1", + }); + const a = await readResources(files(0, 1000, 9000, 0, 0), 0); + const b = await readResources(files(5_000_000, 2000, 10000, 10_000_000, 1_000_000), 5000); + const first = computePoint(null, a); + expect(first.sample.cpuPercent).toBeNull(); + expect(first.sample.netRxBps).toBeNull(); + const p = computePoint(a, b); + // 5 秒用了 5 秒 CPU,限额 2 核 → 50% + expect(p.current.cpu).toMatchObject({ scope: "cgroup", percent: 50, cores: 2, load: [1, 0.5, 0.25] }); + // 整机:total 增 2000、idle 增 1000 → 50% + expect(p.current.cpu.hostPercent).toBe(50); + expect(p.current.memory).toMatchObject({ + scope: "cgroup", + usedBytes: 500 * 1024 * 1024, + limitBytes: 1024 * 1024 * 1024, + percent: 48.8, + }); + expect(p.sample).toMatchObject({ netRxBps: 2_000_000, netTxBps: 1_000_000, diskReadBps: 200_000 }); + expect(p.current.diskIo).toMatchObject({ scope: "cgroup", writeBps: 400_000 }); + expect(p.current.hostUptimeSec).toBe(3600.5); + }); + + it("宿主(没有 cgroup 文件):退回 /proc 整机口径", async () => { + const files = (user: number, idle: number) => + fakeFs({ "/proc/stat": STAT(user, idle), "/proc/meminfo": MEMINFO, "/proc/net/dev": NETDEV(0, 0) }); + const p = computePoint(await readResources(files(1000, 9000), 0), await readResources(files(1500, 9500), 1000)); + expect(p.current.cpu).toMatchObject({ scope: "host", percent: 50 }); + expect(p.current.memory).toMatchObject({ scope: "host", usedBytes: 8000000 * 1024, percent: 50 }); + expect(p.current.diskIo).toMatchObject({ scope: "none", readBps: null }); + expect(p.current.network.scope).toBe("host"); + }); + + it("什么都读不到:全部为 null,不抛错", async () => { + const none = fakeFs({}); + const p = computePoint(await readResources(none, 0), await readResources(none, 1000)); + expect(p.current.cpu).toMatchObject({ scope: "none", percent: null, hostPercent: null, load: null }); + expect(p.current.memory).toMatchObject({ scope: "none", usedBytes: null, percent: null }); + expect(p.current.network).toMatchObject({ scope: "none", rxBps: null }); + }); + + it("计数器回绕 / 重置时速率为 null,不出负数", async () => { + const f = (rx: number) => fakeFs({ "/proc/net/dev": NETDEV(rx, rx) }); + const p = computePoint(await readResources(f(1000), 0), await readResources(f(10), 1000)); + expect(p.sample.netRxBps).toBeNull(); + }); +}); + +describe("采样器", () => { + it("环形保留、按桶平均、快照符合契约", async () => { + let now = Date.parse("2026-10-06T00:00:00Z"); + let usage = 0; + const read: ReadText = (path) => { + if (path === "/sys/fs/cgroup/cpu.stat") return Promise.resolve(`usage_usec ${usage}\n`); + if (path === "/sys/fs/cgroup/cpu.max") return Promise.resolve("100000 100000"); + if (path === "/proc/meminfo") return Promise.resolve(MEMINFO); + return Promise.resolve(null); + }; + const s = createResourceSampler({ read, autoStart: false, intervalMs: 5000, retentionMs: 60_000, now: () => now }); + for (let i = 0; i < 30; i++) { + // 每 5 秒用 2.5 秒 CPU(1 核)→ 50% + usage += 2_500_000; + now += 5000; + await s.tick(); + } + const snap = SystemResources.parse(await s.snapshot(5)); + // 保留 60 秒(含边界)→ 13 个点;第一次读数没有速率,不进趋势 + expect(snap.samples).toHaveLength(13); + expect(snap.samples.every((x) => x.cpuPercent === 50)).toBe(true); + expect(snap.cpu).toMatchObject({ scope: "cgroup", percent: 50, cores: 1 }); + expect(snap.intervalSec).toBe(5); + s.stop(); + + // 长窗口:按桶取平均,点数不超过 360 + const many = Array.from({ length: 4000 }, (_, i) => ({ + at: new Date(Date.parse("2026-10-06T00:00:00Z") + i * 5000).toISOString(), + cpuPercent: i % 2 ? 40 : 60, + memoryUsedBytes: 100, + memoryPercent: 10, + netRxBps: null, + netTxBps: null, + diskReadBps: null, + diskWriteBps: null, + })); + const b = bucketSamples(many, 5, 360); + expect(b.bucketSec).toBe(60); + expect(b.points.length).toBeLessThanOrEqual(360); + expect(b.points[0]).toMatchObject({ cpuPercent: 50, memoryUsedBytes: 100, netRxBps: null }); + }); + + it("刚启动没有趋势时,快照现采两次给出当前值", async () => { + let usage = 0; + const read: ReadText = (path) => { + if (path === "/sys/fs/cgroup/cpu.stat") { + usage += 100_000; + return Promise.resolve(`usage_usec ${usage}\n`); + } + return Promise.resolve(null); + }; + const s = createResourceSampler({ read, autoStart: false }); + const snap = await s.snapshot(60); + expect(snap.cpu.scope).toBe("cgroup"); + expect(snap.cpu.percent).not.toBeNull(); + expect(snap.samples.length).toBeGreaterThanOrEqual(1); + }); + + it("进程 CPU:首次为 null,之后为非负数", () => { + const m = createProcessCpuMeter(); + expect(m()).toBeNull(); + let x = 0; + for (let i = 0; i < 1e5; i++) x += Math.sqrt(i); + expect(x).toBeGreaterThan(0); + const v = m(); + expect(v).not.toBeNull(); + expect(v!).toBeGreaterThanOrEqual(0); + }); +}); diff --git a/apps/server/test/system/system.test.ts b/apps/server/test/system/system.test.ts index 31751f4d..490b2eff 100644 --- a/apps/server/test/system/system.test.ts +++ b/apps/server/test/system/system.test.ts @@ -2,7 +2,7 @@ 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 { SystemHealth, SystemHealthHistory, SystemResources } 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"; @@ -16,6 +16,7 @@ import { writeHeartbeat, } from "../../src/modules/system/heartbeat.ts"; import { createSystemModule } from "../../src/modules/system/index.ts"; +import { createResourceSampler } from "../../src/modules/system/resources.ts"; import type { ServerContext } from "../../src/modules/types.ts"; import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; import { DEV_KEY, json } from "../helpers.ts"; @@ -217,6 +218,16 @@ describe.skipIf(!hasPg)("W9-I 进程与健康(真实 PG)", () => { engineSchema: h.engineSchema, apiHeartbeat: false, healthDeps: { dataDir: dir, diskUsage: usage, ioTimeoutMs: 200 }, + // 资源采样:只给 meminfo 与 cgroup cpu.stat,其余读不到 → 降级为 null + resourceSampler: createResourceSampler({ + autoStart: false, + read: (path) => { + if (path === "/proc/meminfo") return Promise.resolve("MemTotal: 1000 kB\nMemAvailable: 250 kB\n"); + if (path === "/sys/fs/cgroup/cpu.stat") + return Promise.resolve(`usage_usec ${Math.round(performance.now() * 500)}\n`); + return Promise.resolve(null); + }, + }), }), ], ctx: h.ctx, @@ -295,6 +306,43 @@ describe.skipIf(!hasPg)("W9-I 进程与健康(真实 PG)", () => { expect(d.sse.connections).toBe(0); }); + it("资源监控:GET /system/resources 给当前值与趋势,读不到的字段为 null", async () => { + const res = await admin("GET", "/system/resources?minutes=30"); + expect(res.status).toBe(200); + const d = SystemResources.parse((await json(res)).data); + expect(d.memory).toMatchObject({ scope: "host", usedBytes: 750 * 1024, percent: 75 }); + expect(d.cpu.scope).toBe("cgroup"); + expect(d.cpu.percent).not.toBeNull(); + expect(d.network).toMatchObject({ scope: "none", rxBps: null }); + expect(d.diskIo.scope).toBe("none"); + expect(d.samples.length).toBeGreaterThanOrEqual(1); + expect((await admin("GET", "/system/resources?minutes=1")).status).toBe(422); + }); + + it("心跳带上本进程 CPU 占用", async () => { + const w = startHeartbeatWriter({ + db: h.db, + logger: h.ctx.logger, + kind: "worker:agent", + host: "cpu-host", + pid: 7, + intervalMs: 60_000, + lag: { take: () => ({ lastMs: 0, windowMaxMs: 0, lifetimeMaxMs: 0 }), stop() {} }, + }); + await w.beat(); + await new Promise((r) => setTimeout(r, 20)); + await w.beat(); + const row = await h.db + .selectFrom("process_heartbeat") + .select("extra") + .where("instance_id", "=", w.instanceId) + .executeTakeFirstOrThrow(); + expect(row.extra).toMatchObject({ cpuPercent: expect.any(Number), rssBytes: expect.any(Number) }); + await w.stop(); + await sql`delete from process_heartbeat where instance_id = ${w.instanceId}`.execute(h.db); + await sql`delete from process_heartbeat_sample where instance_id = ${w.instanceId}`.execute(h.db); + }); + 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); diff --git a/apps/server/vitest.config.ts b/apps/server/vitest.config.ts index 73054598..7c6b5034 100644 --- a/apps/server/vitest.config.ts +++ b/apps/server/vitest.config.ts @@ -22,6 +22,7 @@ const UNIT = [ "test/modules.test.ts", "test/sse.test.ts", "test/static.test.ts", + "test/system-resources.test.ts", ]; const SERIAL = [