diff --git a/.agents/skills/unidesk-gc/SKILL.md b/.agents/skills/unidesk-gc/SKILL.md index 07b4ec11..be95d33d 100644 --- a/.agents/skills/unidesk-gc/SKILL.md +++ b/.agents/skills/unidesk-gc/SKILL.md @@ -128,10 +128,16 @@ rm -rf -- /tmp/unidesk-cli-output bun scripts/cli.ts gc remote memory-distribution ``` -- 默认输出物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布； +- 默认一次输出物理内存、PSS、进程 swap、cgroup、zombie 和 Kubernetes 对象分布； +- 同一结果必须直接披露 `processes.topSwap`，以及 zombie 父进程的 Kubernetes Pod 归属、 + 泄露分类和处置类别，避免额外执行 `ps`、cgroup 映射和 `kubectl` 拼接； - swap 单独披露，不能计入 WebProbe 启动资格； - 需要完整明细时才使用 `--full`。 +发现大量 zombie 时不得尝试终止 zombie。先按 overview 的 owner 定位长期父进程， +修复子进程回收路径，再通过正常自动交付滚动 owner。运行面重启只能清除既有 zombie， +不能替代源码修复。 + Kubernetes 历史对象手动回收固定使用： ```bash diff --git a/config/unidesk-cli.yaml b/config/unidesk-cli.yaml index 05ee80b2..3f8ce422 100644 --- a/config/unidesk-cli.yaml +++ b/config/unidesk-cli.yaml @@ -169,6 +169,7 @@ gc: podLimit: 4 processLimit: 4 processPssLimit: 4 + processSwapLimit: 4 commandTimeoutSeconds: 20 kubeconfigPath: /etc/rancher/k3s/k3s.yaml targetMemAvailableBytes: 4294967296 @@ -208,7 +209,20 @@ gc: - /opt/google/chrome/ - /headless_shell cgroupReclaim: - enabled: false + enabled: true + targetMemAvailableBytes: 4294967296 + minimumSwapFreeBytesAfterReclaim: 1073741824 + targets: + - id: k3s-file-pages + path: /sys/fs/cgroup/system.slice/k3s.service + reclaimBytes: 402653184 + swappiness: 0 + priority: 10 + - id: kubepods-file-pages + path: /sys/fs/cgroup/kubepods.slice + reclaimBytes: 536870912 + swappiness: 0 + priority: 20 kubernetesObjectRetention: enabled: true configId: devops-infra-tekton-history diff --git a/docs/MDTODO/details/web-observe-runtime-reliability/R1.2.2_Task_Report.md b/docs/MDTODO/details/web-observe-runtime-reliability/R1.2.2_Task_Report.md new file mode 100644 index 00000000..82042ccb --- /dev/null +++ b/docs/MDTODO/details/web-observe-runtime-reliability/R1.2.2_Task_Report.md @@ -0,0 +1,20 @@ +# R1.2.2 任务报告 + +## 根因 + +NC01 约 1 万个 zombie 中，Decision Center Pod 的长期 Bun 父进程拥有 99% 以上；每次存储健康刷新执行多条 Git 子进程，并通过 Bun 的 Node 兼容 `spawnSync` 路径遗留未回收的直接子进程。健康刷新同时承担远程 fetch、checkout 和 pull，扩大了故障频率与外部依赖面。 + +## 实现 + +- Decision Center 改用 Bun 原生 `spawnSync`；200 次 Git smoke 的 zombie 子进程数为 0。 +- 远程 Git 同步只在启动和写入路径执行；周期健康刷新只解析本地 `.git/HEAD`、ref 与 `packed-refs`。 +- `gc remote memory-distribution` 默认一次披露进程 swap、Docker/Pod owner、zombie 泄露分类、cgroup 和 Kubernetes 对象压力，并保持在 10 KiB 输出预算内。 +- NC01 YAML 增加两个 `swappiness=0` 的精确 cgroup 文件页回收候选；不终止 workload、不重启 k3s、不把 swap 计入可用物理内存。 + +## 运行证据 + +修复前 Decision Center 父进程 zombie 从 9993 持续增长至 10113。受控 cgroup GC job `nc01-memory-pressure-1784175371-858647` 两阶段均成功，cgroup 分别下降约 1.20 GiB 与 453 MiB，但并发工作负载快速重新占用，`MemAvailable` 净增约 62 MiB 后回落，证明反复 reclaim 不能替代源头修复和对象留存。 + +## 待验收 + +合并后等待正常自动 CI/CD 发布，确认 Decision Center Pod 滚动、既有 zombie 清除且跨一个 YAML 健康周期不再增长；同时复核 `MemAvailable`、k3s/PSS、swap 和 Kubernetes 对象数量。 diff --git a/docs/MDTODO/web-observe-runtime-reliability.md b/docs/MDTODO/web-observe-runtime-reliability.md index 5f252bbb..20857b9a 100644 --- a/docs/MDTODO/web-observe-runtime-reliability.md +++ b/docs/MDTODO/web-observe-runtime-reliability.md @@ -19,6 +19,9 @@ #### R1.2.1 [completed] 为 UniDesk #1710 增加 `devops-infra` Kubernetes 历史对象留存：手动 `gc remote retention plan|run|status` 与 YAML 定时滚动回收必须复用同一 fail-closed 筛选器；只处理终态 PipelineRun，通过 ownerRef 后台级联，保留活跃、过新、未知分组和每组最新对象；完成 dry-run、受控执行、timer 安装和 `MemAvailable` 实证后收口，完成任务后将详细报告写入[任务报告](./details/web-observe-runtime-reliability/R1.2.1_Task_Report.md)。 +#### R1.2.2 [in_progress] + +完成 [UniDesk #1710](https://github.com/pikasTech/unidesk/issues/1710) 的 Decision Center 子进程泄露与 GC 诊断收敛：修复周期性 Git 子进程未回收并把远程同步移出健康轮询；增强 `gc remote memory-distribution`，一次披露物理内存、PSS、swap、僵尸 Pod 归属、Kubernetes 对象压力和处置分类；经自动交付后验证僵尸不再增长并记录内存变化，完成任务后将详细报告写入[任务报告](./details/web-observe-runtime-reliability/R1.2.2_Task_Report.md)。 ### R1.3 测试与验收：https://github.com/pikasTech/unidesk/issues/1842；覆盖生命周期、stale/protected 分类和失败路径，提交独立 PR；合并后只等待自动 CI/CD 发布链，使用正式入口验证 observer/Chrome 退出、内存压力缓解且健康运行面不受扰动，完成任务后将详细报告写入[任务报告](./details/web-observe-runtime-reliability/R1.3_Task_Report.md)。 diff --git a/docs/reference/gc.md b/docs/reference/gc.md index b8d30b76..ad2950d2 100644 --- a/docs/reference/gc.md +++ b/docs/reference/gc.md @@ -31,8 +31,10 @@ - 不执行 cluster preflight、候选扫描或 mutation； - 固定披露 `metric=MemAvailable`、`source=/proc/meminfo`、`swapExcluded=true` 和可用字节数。 - `gc remote memory-distribution`： - - 只读输出物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布； - - 默认返回紧凑 overview，`--full` 才展开完整明细； + - 一次只读输出物理内存、PSS、进程 swap、cgroup、zombie 和 Kubernetes 对象分布； + - `processes.topSwap` 用 `/proc//status` 的 `VmSwap` 排序，识别低 RSS、高 swap 的抖动风险； + - `processes.zombies.byParent` 披露父进程、所属 Kubernetes Pod、泄露分类和处置类别； + - 默认返回紧凑 overview，只有缺少必要字段时才用 `--full` 展开完整明细； - swap 单独披露，不计入 WebProbe 启动资格。 - `gc remote retention plan|run --confirm|status --job-id `： - 按目标节点 `kubernetesObjectRetention` YAML 规划和异步回收 Kubernetes 历史对象； @@ -42,6 +44,25 @@ 所有成功和失败输出都必须是 JSON。`plan` 必须标记 `dryRun=true`、`mutation=false`；`run` 必须要求 `--confirm` 并报告 `diskBefore`、`diskAfter`、`summary`、`results` 和 `protected`。远端 GC 可用 `--target-use-percent N` 显式表达目标根盘水位；`summary.target` 必须给出目标所需释放量、候选估算、预计水位、缺口和 `safeStop` 决策，避免靠人工心算判断是否应该继续扩大清理范围。 +内存分布调查规则： + +- 先执行一次紧凑 overview； +- 禁止先堆叠 `ps`、`jq`、cgroup 和 Kubernetes 查询后再人工拼接同一归属； +- 僵尸进程不能直接终止； +- 固定处置是修复父进程的子进程回收，再通过正常自动交付受控滚动所属 workload； +- 运行面重启只能清除既有 zombie，不能替代源码修复； +- 数量、超时、资源和 namespace 范围全部由 `config/unidesk-cli.yaml` 的 + `memoryPressure.attribution` 拥有。 + +NC01 的内存压力文件页回收规则： + +- 只从 YAML 精确允许的 k3s cgroup 请求回收； +- 固定使用 `swappiness=0`； +- 不把 swap 当作可用物理内存； +- 不终止 workload，也不重启 k3s； +- 执行前由 overview 证明目标 cgroup 存在足够的非匿名页； +- 按 `plan`、`run --confirm`、`status`、`memory` 的顺序复核实际结果。 + 默认 `/tmp` GC 只包含 allowlisted 诊断目录和已知低风险路径。非 allowlist 的 stale `/tmp` 一级子项必须显式 `--include-stale-tmp` 才能进入候选；扫描按 `--limit` 有界枚举候选，执行时仍只允许删除 `/tmp` 直接子项，并避开 X11/ICE/font socket、systemd private、tmux、ssh、vscode 等系统/session 前缀。该入口不能递归扩大成通用 `/tmp` 清空器，也不能为了估算全量临时目录而长时间阻塞。 主 server VPN 诊断日志默认不清理。`/root/vpn-server/logs` 中由长期 `tcpdump -G` 产生的 `hy2-udp-ring-*.pcap` 和 `hy2-monitor-ring-*.pcap` 可通过显式 `--include-vpn-diagnostic-logs` 进入候选，默认只选择超过 `--vpn-diagnostic-log-keep-hours 24` 的普通 pcap 文件。执行前必须重新校验路径、文件名、非 symlink/regular file，并用 active-file 检查确认没有进程仍打开该文件。`hy2-server-evidence.jsonl`、stdout/stderr log、最新 pcap 和整个日志根目录始终作为 protected 输出，不得被这个入口删除或截断。 diff --git a/scripts/src/gc-remote-memory-distribution.py b/scripts/src/gc-remote-memory-distribution.py index 51e4777b..b139ee16 100644 --- a/scripts/src/gc-remote-memory-distribution.py +++ b/scripts/src/gc-remote-memory-distribution.py @@ -9,6 +9,7 @@ def memory_distribution_config(): "podLimit": (1, 100), "processLimit": (1, 100), "processPssLimit": (1, 50), + "processSwapLimit": (1, 100), "commandTimeoutSeconds": (1, 60), "targetMemAvailableBytes": (1, 1 << 50), "namespaceCountConcurrency": (1, 32), @@ -181,7 +182,24 @@ def memory_distribution_pss(pid): "swapPssBytes": values.get("SwapPss"), } -def collect_memory_distribution_processes(config): +def memory_distribution_process_swap(pid): + values = read_memory_distribution_key_values("/proc/%s/status" % int(pid), 1024) + return values.get("VmSwap") + +def memory_distribution_process_owner(cgroup_path, docker_names, pod_names): + if not isinstance(cgroup_path, str): + return None + docker_match = re.search(r"docker-([a-f0-9]{64})\.scope(?:/|$)", cgroup_path) + if docker_match is not None: + container_id = docker_match.group(1) + return {"kind": "docker", "id": container_id, "name": docker_names.get(container_id)} + match = re.search(r"pod([0-9a-f_]{36})\.slice(?:/|$)", cgroup_path) + if match is None: + return None + uid = match.group(1).replace("_", "-") + return {"kind": "kubernetes-pod", "uid": uid, **(pod_names.get(uid) or {})} + +def collect_memory_distribution_processes(config, docker_names, pod_names): result = command(["ps", "-eo", "pid=,ppid=,rss=,stat=,comm=,args="], safe_int(config.get("commandTimeoutSeconds"))) if result.get("exitCode") != 0: return {"ok": False, "command": memory_distribution_command_summary(result), "top": [], "zombies": {}} @@ -198,33 +216,53 @@ def collect_memory_distribution_processes(config): state = parts[3] comm = parts[4] args = parts[5] if len(parts) > 5 else comm - row = {"pid": pid, "ppid": ppid, "comm": comm, "state": state, "rssBytes": rss, "commandPreview": redact_command_preview(args)[:120]} + row = { + "pid": pid, + "ppid": ppid, + "comm": comm, + "state": state, + "rssBytes": rss, + "swapBytes": memory_distribution_process_swap(pid), + "commandPreview": redact_command_preview(args)[:120], + } rows.append(row) if state.startswith("Z"): zombie_by_comm[comm] = zombie_by_comm.get(comm, 0) + 1 zombie_by_parent[ppid] = zombie_by_parent.get(ppid, 0) + 1 by_pid = {item.get("pid"): item for item in rows} top = sorted(rows, key=lambda item: safe_int(item.get("rssBytes")), reverse=True)[:memory_distribution_limit(config, "processLimit")] + top_swap = sorted(rows, key=lambda item: safe_int(item.get("swapBytes")), reverse=True)[:memory_distribution_limit(config, "processSwapLimit")] pss_limit = memory_distribution_limit(config, "processPssLimit") for index, item in enumerate(top): item["cgroupPath"] = memory_distribution_cgroup_for_pid(item.get("pid")) + item["owner"] = memory_distribution_process_owner(item.get("cgroupPath"), docker_names, pod_names) if index < pss_limit: item.update(memory_distribution_pss(item.get("pid")) or {}) + for item in top_swap: + item["cgroupPath"] = memory_distribution_cgroup_for_pid(item.get("pid")) + item["owner"] = memory_distribution_process_owner(item.get("cgroupPath"), docker_names, pod_names) parent_rows = [] for ppid, count in sorted(zombie_by_parent.items(), key=lambda item: item[1], reverse=True)[:5]: parent = by_pid.get(ppid) or {} + cgroup_path = memory_distribution_cgroup_for_pid(ppid) + owner = memory_distribution_process_owner(cgroup_path, docker_names, pod_names) parent_rows.append({ "ppid": ppid, "parentComm": parent.get("comm"), "zombieCount": count, - "cgroupPath": memory_distribution_cgroup_for_pid(ppid), + "cgroupPath": cgroup_path, + "owner": owner, + "classification": "unreaped-child-process-leak", + "remediation": "fix-child-reaping-and-rollout-owner", }) return { "ok": True, "command": memory_distribution_command_summary(result), "processCount": len(rows), "rssTotalBytes": sum(safe_int(item.get("rssBytes")) for item in rows), + "swapTotalBytes": sum(safe_int(item.get("swapBytes")) for item in rows), "top": top, + "topSwap": top_swap, "zombies": { "count": sum(zombie_by_comm.values()), "byComm": [{"comm": comm, "count": count} for comm, count in sorted(zombie_by_comm.items(), key=lambda item: item[1], reverse=True)[:5]], @@ -332,7 +370,7 @@ def collect_memory_distribution_namespace_counts(config): def compact_memory_distribution_cgroup(item): compact = { key: item.get(key) - for key in ["path", "currentBytes", "swapCurrentBytes", "anonBytes"] + for key in ["currentBytes", "swapCurrentBytes", "anonBytes"] if item.get(key) is not None } owner = item.get("owner") if isinstance(item.get("owner"), dict) else None @@ -342,14 +380,26 @@ def compact_memory_distribution_cgroup(item): for key in ["kind", "namespace", "name"] if owner.get(key) is not None } + else: + compact["path"] = item.get("path") return compact def compact_memory_distribution_process(item): - return { + compact = { key: item.get(key) - for key in ["pid", "comm", "rssBytes", "pssBytes", "pssAnonBytes", "swapPssBytes", "cgroupPath"] + for key in ["pid", "comm", "rssBytes", "swapBytes", "pssBytes"] if item.get(key) is not None } + owner = item.get("owner") if isinstance(item.get("owner"), dict) else None + if owner is not None: + compact["owner"] = { + key: owner.get(key) + for key in ["kind", "namespace", "name"] + if owner.get(key) is not None + } + else: + compact["cgroupPath"] = item.get("cgroupPath") + return compact def compact_memory_distribution_pressure(pressure): return { @@ -365,11 +415,19 @@ def compact_memory_distribution_pressure(pressure): def compact_memory_distribution_zombies(zombies): by_parent = [] for item in zombies.get("byParent") or []: - by_parent.append({ + compact = { key: item.get(key) - for key in ["ppid", "parentComm", "zombieCount"] + for key in ["ppid", "parentComm", "zombieCount", "classification", "remediation"] if item.get(key) is not None - }) + } + owner = item.get("owner") if isinstance(item.get("owner"), dict) else None + if owner is not None: + compact["owner"] = { + key: owner.get(key) + for key in ["kind", "namespace", "name"] + if owner.get(key) is not None + } + by_parent.append(compact) return { "count": zombies.get("count"), "byComm": (zombies.get("byComm") or [])[:3], @@ -386,7 +444,7 @@ def collect_memory_distribution(observed_at): host_snapshot = collect_memory_snapshot() meminfo = read_memory_distribution_key_values("/proc/meminfo", 1024) cgroups = collect_memory_distribution_cgroups(config, docker_names, pod_names) - processes = collect_memory_distribution_processes(config) + processes = collect_memory_distribution_processes(config, docker_names, pod_names) metrics = collect_memory_distribution_metrics(config) namespace_counts = collect_memory_distribution_namespace_counts(config) available = safe_int((host_snapshot.get("memory") or {}).get("availableBytes")) @@ -442,7 +500,6 @@ def collect_memory_distribution(observed_at): if bool(OPTIONS.get("full")): return payload payload["host"] = { - "snapshot": host_snapshot, "pressure": {"memory": compact_memory_distribution_pressure(parse_memory_distribution_pressure("/proc/pressure/memory"))}, } payload["cgroups"] = { @@ -456,19 +513,25 @@ def collect_memory_distribution(observed_at): "ok": processes.get("ok"), "processCount": processes.get("processCount"), "rssTotalBytes": processes.get("rssTotalBytes"), + "swapTotalBytes": processes.get("swapTotalBytes"), "top": [compact_memory_distribution_process(item) for item in processes.get("top") or []], + "topSwap": [compact_memory_distribution_process(item) for item in processes.get("topSwap") or []], "zombies": compact_memory_distribution_zombies(processes.get("zombies") or {}), } payload["kubernetes"] = { "objects": metrics.get("objects"), "namespaceCounts": namespace_counts, - "command": metrics.get("command"), } + if (metrics.get("command", {}).get("ok") is not True): + payload["kubernetes"]["commandError"] = metrics.get("command") payload["collection"] = { "configSource": payload.get("collection", {}).get("configSource"), "podRuntimeMapCount": len(pod_names), "dockerRuntimeMapCount": len(docker_names), - "runtimeMaps": runtime_commands, + "runtimeMapErrors": { + key: value for key, value in runtime_commands.items() + if value.get("ok") is not True + }, "full": False, } return payload diff --git a/src/components/microservices/decision-center/src/index.ts b/src/components/microservices/decision-center/src/index.ts index 4e7c5198..ea00404d 100644 --- a/src/components/microservices/decision-center/src/index.ts +++ b/src/components/microservices/decision-center/src/index.ts @@ -1,7 +1,6 @@ import { createHash, randomUUID } from "node:crypto"; import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; -import { dirname, join } from "node:path"; -import { spawnSync } from "node:child_process"; +import { dirname, join, resolve } from "node:path"; import postgres from "postgres"; import { createHourlyJsonlWriter, logRetentionBytesForService } from "../../../shared/src/rotating-jsonl"; import { @@ -885,18 +884,18 @@ function health(): JsonRecord { }; } -function runGit(args: string[], options: { cwd?: string; allowFailure?: boolean } = {}): { ok: boolean; stdout: string; stderr: string; exitCode: number } { - const env = { ...process.env }; +function runGit(args: string[], options: { cwd?: string; allowFailure?: boolean; env?: Record } = {}): { ok: boolean; stdout: string; stderr: string; exitCode: number } { + const env = { ...process.env, ...options.env }; if (config.storage.sshCommand) env.GIT_SSH_COMMAND = config.storage.sshCommand; - const result = spawnSync("git", args, { + const result = Bun.spawnSync(["git", ...args], { cwd: options.cwd ?? config.storage.worktreePath, env, - encoding: "utf8", - maxBuffer: 8 * 1024 * 1024, + stdout: "pipe", + stderr: "pipe", }); - const exitCode = result.status ?? 1; - const stdout = typeof result.stdout === "string" ? result.stdout : ""; - const stderr = typeof result.stderr === "string" ? result.stderr : ""; + const exitCode = result.exitCode; + const stdout = new TextDecoder().decode(result.stdout); + const stderr = new TextDecoder().decode(result.stderr); if (exitCode !== 0 && options.allowFailure !== true) { throw new HttpError(503, "git storage command failed", { command: ["git", ...args].join(" "), @@ -1014,15 +1013,53 @@ function ensureGitWorktree(): void { } } +function gitDirectoryPath(): string | null { + const marker = join(config.storage.worktreePath, ".git"); + if (!existsSync(marker)) return null; + try { + const value = readFileSync(marker, "utf8").trim(); + if (value.startsWith("gitdir:")) return resolve(config.storage.worktreePath, value.slice("gitdir:".length).trim()); + } catch { + return marker; + } + return marker; +} + +function localGitRef(gitDirectory: string, ref: string): string { + try { + return readFileSync(join(gitDirectory, ...ref.split("/")), "utf8").trim(); + } catch { + try { + const packedRefs = readFileSync(join(gitDirectory, "packed-refs"), "utf8"); + for (const line of packedRefs.split("\n")) { + if (line.startsWith("#") || line.startsWith("^") || !line.trim()) continue; + const [commit, packedRef] = line.trim().split(" ", 2); + if (packedRef === ref) return commit ?? ""; + } + } catch { + return ""; + } + return ""; + } +} + function gitHeadSummary(): JsonRecord { - if (!existsSync(join(config.storage.worktreePath, ".git"))) return { present: false }; - const head = runGit(["rev-parse", "HEAD"], { allowFailure: true }); - const branch = runGit(["rev-parse", "--abbrev-ref", "HEAD"], { allowFailure: true }); + const gitDirectory = gitDirectoryPath(); + if (gitDirectory === null) return { present: false }; + let head = ""; + try { + head = readFileSync(join(gitDirectory, "HEAD"), "utf8").trim(); + } catch { + return { present: false }; + } + const symbolic = head.startsWith("ref:") ? head.slice("ref:".length).trim() : ""; + const commit = symbolic ? localGitRef(gitDirectory, symbolic) : head; + const present = /^[0-9a-f]{40}$/u.test(commit); return { - present: head.ok, - branch: branch.stdout.trim(), - commit: head.stdout.trim().slice(0, 40), - shortCommit: head.stdout.trim().slice(0, 12), + present, + branch: symbolic.startsWith("refs/heads/") ? symbolic.slice("refs/heads/".length) : symbolic ? symbolic : "HEAD", + commit: present ? commit : "", + shortCommit: present ? commit.slice(0, 12) : "", }; } @@ -1033,7 +1070,8 @@ async function updateStorageHealth(): Promise { return; } try { - ensureGitWorktree(); + const head = gitHeadSummary(); + if (head.present !== true) throw new Error("local Git storage worktree is not initialized"); storageHealth = { ok: true, primary: "github-repo", @@ -1042,7 +1080,7 @@ async function updateStorageHealth(): Promise { sshUrl: config.storage.repoSshUrl, branch: config.storage.branch, basePath: config.storage.basePath, - head: gitHeadSummary(), + head, }, error: null, }; @@ -1116,13 +1154,8 @@ async function writeSnapshotToGit(reason: string, dryRun: boolean): Promise { await refreshDatabaseHealth(); - await updateStorageHealth(); + try { + ensureGitWorktree(); + } finally { + await updateStorageHealth(); + } }) .catch((error) => { log("error", "schema_init_failed", { error: errorToJson(error) });