feat(server): 资源监控——/proc 与 cgroup 采样、/system/resources 趋势，心跳带进程 CPU

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei

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<string, number> {
   }
 }
 
+/** 本进程两次心跳之间的 CPU 占用(% of 一个核);取不到时不写 */
+function cpuExtra(pct: number | null): Record<string, number> {
+  return pct == null ? {} : { cpuPercent: pct };
+}
+
 /** 写一次心跳(upsert + 采样);返回写入的 instanceId */
 export async function writeHeartbeat(
   db: Kysely<DB>,
@@ -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<void> | 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 }, "心跳写入失败");

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> = T extends z.ZodType ? z.infer<T> : Record<string, never>;
@@ -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;
     },
   };
 }

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<string | null>;
+
+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<string, number>();
+  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<typeof parseMeminfo>;
+  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<RawReading> {
+  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<SystemResources, "at" | "intervalSec" | "bucketSec" | "samples">;
+}
+
+/** 由相邻两次读数算一个点(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<number, ResourceSample[]>();
+  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<string, unknown>)[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<void>;
+  snapshot(minutes: number): Promise<SystemResources>;
+  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<void> | 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<typeof setInterval> | 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;
+    }
+  };
+}