fix: 修复 Decision Center 子进程泄露并增强内存归因

2026-07-16 06:20:09 +02:00
parent 8298c5ce74
commit 03e2a760c2
7 changed files with 208 additions and 44 deletions
@@ -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