Merge remote-tracking branch 'origin/master' into docs/selfmedia-voice-selector-mdtodo
Pipelines as Code CI / hwlab-web-probe-sentinel-nc01- Success
Pipelines as Code CI / platform-infra-gitea-nc01- Success
Pipelines as Code CI / unidesk-host- Success

This commit is contained in:
Codex
2026-07-16 06:25:30 +02:00
7 changed files with 208 additions and 44 deletions
+7 -1
View File
@@ -128,10 +128,16 @@ rm -rf -- /tmp/unidesk-cli-output
bun scripts/cli.ts gc remote <providerId> 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
+15 -1
View File
@@ -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
@@ -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 <node> 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 对象数量。
@@ -19,6 +19,9 @@
#### R1.2.1 [completed]
为 UniDesk #1710 增加 `devops-infra` Kubernetes 历史对象留存:手动 `gc remote <node> 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 <node> 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)。
+23 -2
View File
@@ -31,8 +31,10 @@
- 不执行 cluster preflight、候选扫描或 mutation
- 固定披露 `metric=MemAvailable``source=/proc/meminfo``swapExcluded=true` 和可用字节数。
- `gc remote <providerId> memory-distribution`
- 只读输出物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布;
- 默认返回紧凑 overview`--full` 才展开完整明细
- 一次只读输出物理内存、PSS、进程 swap、cgroup、zombie 和 Kubernetes 对象分布;
- `processes.topSwap``/proc/<pid>/status``VmSwap` 排序,识别低 RSS、高 swap 的抖动风险
- `processes.zombies.byParent` 披露父进程、所属 Kubernetes Pod、泄露分类和处置类别;
- 默认返回紧凑 overview,只有缺少必要字段时才用 `--full` 展开完整明细;
- swap 单独披露,不计入 WebProbe 启动资格。
- `gc remote <providerId> retention plan|run --confirm|status --job-id <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 输出,不得被这个入口删除或截断。
+76 -13
View File
@@ -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
@@ -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<string, string> } = {}): { 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<void> {
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<void> {
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<Json
GIT_COMMITTER_EMAIL: config.storage.authorEmail,
...(config.storage.sshCommand ? { GIT_SSH_COMMAND: config.storage.sshCommand } : {}),
};
const commit = spawnSync("git", ["commit", "-m", message], {
cwd: config.storage.worktreePath,
env: { ...process.env, ...env },
encoding: "utf8",
maxBuffer: 8 * 1024 * 1024,
});
if ((commit.status ?? 1) !== 0) throw new HttpError(503, "git commit failed", { stderr: String(commit.stderr).slice(-2000) });
const commit = runGit(["commit", "-m", message], { allowFailure: true, env });
if (!commit.ok) throw new HttpError(503, "git commit failed", { stderr: commit.stderr.slice(-2000) });
let pushAttempts = 0;
let pushSucceeded = false;
let pushStderr = "";
@@ -2191,7 +2224,11 @@ if (import.meta.main && !isCheckMode) {
void waitForSchema()
.then(async () => {
await refreshDatabaseHealth();
await updateStorageHealth();
try {
ensureGitWorktree();
} finally {
await updateStorageHealth();
}
})
.catch((error) => {
log("error", "schema_init_failed", { error: errorToJson(error) });