feat: add process resource monitor

src/components/provider-gateway/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@unidesk/provider-gateway",
-  "version": "0.2.5",
+  "version": "0.2.6",
   "private": true,
   "type": "module",
   "scripts": {

src/components/provider-gateway/src/index.ts

@@ -1,4 +1,4 @@
-import { appendFileSync, existsSync, mkdirSync, readFileSync } from "node:fs";
+import { appendFileSync, existsSync, mkdirSync, readFileSync, readdirSync } from "node:fs";
 import { dirname } from "node:path";
 import {
   type CoreDispatchMessage,
@@ -13,6 +13,7 @@ import {
   type DockerStatusSnapshot,
   type DockerVolumeSummary,
   type JsonValue,
+  type ProcessResourceSummary,
   type ProviderLabels,
   type ProviderTaskStatusMessage,
   type SystemStatusSnapshot,
@@ -56,6 +57,10 @@ let systemStatusTimer: ReturnType<typeof setInterval> | null = null;
 let dockerStatusRunning = false;
 let systemStatusRunning = false;
 let previousCpuSample: { idle: number; total: number } | null = null;
+let previousProcessSamples = new Map<number, { totalTicks: number; sampledAtMs: number; readBytes: number; writeBytes: number }>();
+let passwdCache: Map<number, string> | null = null;
+let clockTicksCache: number | null = null;
+let pageSizeCache: number | null = null;
 let reconnectAttempt = 0;
 let stopping = false;
 let upgradeSleepUntil = 0;
@@ -261,6 +266,7 @@ async function sendSystemStatus(): Promise<void> {
       cpuPercent: status.cpu.percent,
       memoryPercent: status.memory.percent,
       diskPercent: status.disk.percent,
+      processCount: status.processes?.length ?? 0,
     });
   } catch (error) {
     logger("error", "system_status_failed", { error: error instanceof Error ? error.message : String(error) });
@@ -384,6 +390,11 @@ function clampPercent(value: number): number {
   return Math.max(0, Math.min(100, Number(value.toFixed(1))));
 }
 
+function roundMetric(value: number, digits = 1): number {
+  if (!Number.isFinite(value)) return 0;
+  return Math.max(0, Number(value.toFixed(digits)));
+}
+
 function readCpuSample(): { idle: number; total: number; cores: number } {
   const stat = readFileSync("/proc/stat", "utf8");
   const lines = stat.split("\n");
@@ -415,6 +426,211 @@ function readMemInfo(): Record<string, number> {
   return result;
 }
 
+async function readGetconfNumber(name: string, fallback: number): Promise<number> {
+  const result = await runProcessCommand("getconf", [name], 1000);
+  const parsed = Number(result.stdout.trim());
+  return result.ok && Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
+}
+
+async function systemClockTicks(): Promise<number> {
+  if (clockTicksCache !== null) return clockTicksCache;
+  clockTicksCache = await readGetconfNumber("CLK_TCK", 100);
+  return clockTicksCache;
+}
+
+async function systemPageSize(): Promise<number> {
+  if (pageSizeCache !== null) return pageSizeCache;
+  pageSizeCache = await readGetconfNumber("PAGESIZE", 4096);
+  return pageSizeCache;
+}
+
+function usernameForUid(uid: number): string {
+  if (passwdCache === null) {
+    passwdCache = new Map<number, string>();
+    try {
+      for (const line of readFileSync("/etc/passwd", "utf8").split("\n")) {
+        const parts = line.split(":");
+        const parsedUid = Number(parts[2]);
+        if (parts[0] && Number.isFinite(parsedUid)) passwdCache.set(parsedUid, parts[0]);
+      }
+    } catch {
+      // Provider containers may not have the host passwd database; fall back to uid labels.
+    }
+  }
+  return passwdCache.get(uid) ?? `uid:${uid}`;
+}
+
+function readProcessStatus(pid: number): { uid: number; threads: number; name: string } {
+  const status = readFileSync(`/proc/${pid}/status`, "utf8");
+  let uid = -1;
+  let threads = 0;
+  let name = "";
+  for (const line of status.split("\n")) {
+    if (line.startsWith("Uid:")) {
+      const parsed = Number(line.trim().split(/\s+/)[1]);
+      if (Number.isFinite(parsed)) uid = parsed;
+    } else if (line.startsWith("Threads:")) {
+      const parsed = Number(line.trim().split(/\s+/)[1]);
+      if (Number.isFinite(parsed)) threads = parsed;
+    } else if (line.startsWith("Name:")) {
+      name = line.slice("Name:".length).trim();
+    }
+  }
+  return { uid, threads, name };
+}
+
+function readProcessCommand(pid: number, fallback: string): string {
+  try {
+    const raw = readFileSync(`/proc/${pid}/cmdline`, "utf8");
+    const command = raw.split("\0").filter(Boolean).join(" ").trim();
+    return (command || fallback).slice(0, 500);
+  } catch {
+    return fallback;
+  }
+}
+
+function readProcessIo(pid: number): { readBytes: number; writeBytes: number } {
+  try {
+    const result = { readBytes: 0, writeBytes: 0 };
+    for (const line of readFileSync(`/proc/${pid}/io`, "utf8").split("\n")) {
+      const match = line.match(/^(read_bytes|write_bytes):\s+(\d+)/);
+      if (!match) continue;
+      const parsed = Number(match[2]);
+      if (!Number.isFinite(parsed)) continue;
+      if (match[1] === "read_bytes") result.readBytes = parsed;
+      if (match[1] === "write_bytes") result.writeBytes = parsed;
+    }
+    return result;
+  } catch {
+    return { readBytes: 0, writeBytes: 0 };
+  }
+}
+
+function readUptimeSeconds(): number {
+  const uptime = Number(readFileSync("/proc/uptime", "utf8").trim().split(/\s+/)[0]);
+  return Number.isFinite(uptime) ? uptime : 0;
+}
+
+function parseProcessStat(raw: string): {
+  pid: number;
+  name: string;
+  state: string;
+  ppid: number;
+  totalTicks: number;
+  startTicks: number;
+  vmsBytes: number;
+  rssPages: number;
+  threads: number;
+} | null {
+  const open = raw.indexOf("(");
+  const close = raw.lastIndexOf(")");
+  if (open < 0 || close <= open) return null;
+  const pid = Number(raw.slice(0, open).trim());
+  const name = raw.slice(open + 1, close);
+  const fields = raw.slice(close + 1).trim().split(/\s+/);
+  if (!Number.isFinite(pid) || fields.length < 22) return null;
+  const value = (index: number): number => {
+    const parsed = Number(fields[index]);
+    return Number.isFinite(parsed) ? parsed : 0;
+  };
+  return {
+    pid,
+    name,
+    state: fields[0] || "?",
+    ppid: value(1),
+    totalTicks: value(11) + value(12),
+    threads: value(17),
+    startTicks: value(19),
+    vmsBytes: value(20),
+    rssPages: value(21),
+  };
+}
+
+async function collectProcessResources(totalMemoryBytes: number, cpuCores: number): Promise<{
+  processes: ProcessResourceSummary[];
+  summary: Record<string, JsonValue>;
+}> {
+  const [clockTicks, pageSize] = await Promise.all([systemClockTicks(), systemPageSize()]);
+  const uptimeSeconds = readUptimeSeconds();
+  const sampledAtMs = Date.now();
+  const rows: ProcessResourceSummary[] = [];
+  const seen = new Set<number>();
+  let skipped = 0;
+  const hasPreviousProcessSample = previousProcessSamples.size > 0;
+
+  for (const entry of readdirSync("/proc")) {
+    if (!/^\d+$/.test(entry)) continue;
+    const pid = Number(entry);
+    try {
+      const stat = parseProcessStat(readFileSync(`/proc/${entry}/stat`, "utf8"));
+      if (stat === null) {
+        skipped += 1;
+        continue;
+      }
+      const status = readProcessStatus(pid);
+      const io = readProcessIo(pid);
+      const previous = previousProcessSamples.get(pid);
+      const elapsedSampleSeconds = previous ? Math.max(0.001, (sampledAtMs - previous.sampledAtMs) / 1000) : 0;
+      const elapsedProcessSeconds = Math.max(0, uptimeSeconds - stat.startTicks / clockTicks);
+      const cpuPercent = previous
+        ? ((Math.max(0, stat.totalTicks - previous.totalTicks) / clockTicks) / elapsedSampleSeconds) * 100
+        : elapsedProcessSeconds > 0
+          ? ((stat.totalTicks / clockTicks) / elapsedProcessSeconds) * 100
+          : 0;
+      const ioSeconds = previous ? elapsedSampleSeconds : 0;
+      const readBytesPerSecond = previous && ioSeconds > 0 ? Math.max(0, io.readBytes - previous.readBytes) / ioSeconds : 0;
+      const writeBytesPerSecond = previous && ioSeconds > 0 ? Math.max(0, io.writeBytes - previous.writeBytes) / ioSeconds : 0;
+      const rssBytes = Math.max(0, stat.rssPages * pageSize);
+      const name = status.name || stat.name;
+      const uid = status.uid >= 0 ? status.uid : 0;
+      rows.push({
+        pid: stat.pid,
+        ppid: stat.ppid,
+        uid,
+        user: usernameForUid(uid),
+        name,
+        command: readProcessCommand(pid, name),
+        state: stat.state,
+        cpuPercent: roundMetric(Math.min(cpuPercent, Math.max(1, cpuCores) * 100)),
+        memoryPercent: totalMemoryBytes > 0 ? clampPercent((rssBytes / totalMemoryBytes) * 100) : 0,
+        rssBytes,
+        vmsBytes: Math.max(0, stat.vmsBytes),
+        threads: status.threads || stat.threads,
+        readBytes: io.readBytes,
+        writeBytes: io.writeBytes,
+        readBytesPerSecond: roundMetric(readBytesPerSecond, 0),
+        writeBytesPerSecond: roundMetric(writeBytesPerSecond, 0),
+        elapsedSeconds: roundMetric(elapsedProcessSeconds, 0),
+        startedAt: new Date(Date.now() - elapsedProcessSeconds * 1000).toISOString(),
+      });
+      previousProcessSamples.set(pid, {
+        totalTicks: stat.totalTicks,
+        sampledAtMs,
+        readBytes: io.readBytes,
+        writeBytes: io.writeBytes,
+      });
+      seen.add(pid);
+    } catch {
+      skipped += 1;
+    }
+  }
+
+  previousProcessSamples = new Map([...previousProcessSamples].filter(([pid]) => seen.has(pid)));
+  rows.sort((a, b) => b.rssBytes - a.rssBytes || b.cpuPercent - a.cpuPercent || a.pid - b.pid);
+  return {
+    processes: rows.slice(0, 120),
+    summary: {
+      total: rows.length,
+      visible: Math.min(rows.length, 120),
+      skipped,
+      defaultSort: "memory_desc",
+      scope: "provider_pid_namespace",
+      cpuPercentMode: hasPreviousProcessSample ? "delta_ticks_per_sample" : "lifetime_average_first_sample",
+      diskIoMode: "proc_pid_io_delta_bytes_per_second",
+    },
+  };
+}
+
 async function readDiskUsage(path: string): Promise<{ mount: string; totalBytes: number; usedBytes: number; availableBytes: number; percent: number }> {
   const result = await runProcessCommand("df", ["-PB1", path], 3000);
   if (!result.ok) throw new Error(`df failed with exit ${result.exitCode}: ${result.stderr}`);
@@ -485,6 +701,18 @@ async function collectSystemStatus(): Promise<SystemStatusSnapshot> {
     errors.push({ source: "proc.meminfo", error: error instanceof Error ? error.message : String(error) });
   }
 
+  let processes: ProcessResourceSummary[] = [];
+  let processSummary: Record<string, JsonValue> = { total: 0, visible: 0, defaultSort: "memory_desc" };
+  try {
+    const totalMemoryBytes = typeof memory.totalBytes === "number" ? memory.totalBytes : 0;
+    const cpuCores = typeof cpu.cores === "number" ? cpu.cores : 1;
+    const collected = await collectProcessResources(totalMemoryBytes, cpuCores);
+    processes = collected.processes;
+    processSummary = collected.summary;
+  } catch (error) {
+    errors.push({ source: "proc.processes", error: error instanceof Error ? error.message : String(error) });
+  }
+
   let disk: Record<string, JsonValue> = { path: config.monitorDiskPath, mount: "", totalBytes: 0, usedBytes: 0, availableBytes: 0, percent: 0 };
   try {
     disk = { path: config.monitorDiskPath, ...(await readDiskUsage(config.monitorDiskPath)) };
@@ -492,7 +720,7 @@ async function collectSystemStatus(): Promise<SystemStatusSnapshot> {
     errors.push({ source: "df", path: config.monitorDiskPath, error: error instanceof Error ? error.message : String(error) });
   }
 
-  return { ok: errors.length === 0, collectedAt, cpu, memory, disk, errors };
+  return { ok: errors.length === 0, collectedAt, cpu, memory, disk, processes, processSummary, errors };
 }
 
 async function collectDockerStatus(): Promise<DockerStatusSnapshot> {
@@ -975,6 +1203,7 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
       "docker run -d",
       `--name ${shellQuote(candidateName)}`,
       "--restart no",
+      "--pid host",
       "$network_arg",
       `--env-file "$candidate_env_file"`,
       `--label ${shellQuote(`com.docker.compose.project=${config.upgradeComposeProject}`)}`,
@@ -1053,6 +1282,7 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
       candidateUsesOldContainerNetworks: true,
       candidateUsesOldContainerExtraHosts: true,
       candidateUsesOldContainerEnvironment: true,
+      candidateUsesHostPidNamespace: true,
       removeScope: {
         projectLabel: config.upgradeComposeProject,
         serviceLabel: config.upgradeService,