_OBSERVE_RUN_CACHE = {} _OBSERVE_SCAN_TRUNCATED = False _PROCESS_SCAN_TRUNCATED = False _OBSERVE_SCAN_ERROR = None _PROCESS_SCAN_ERROR = None _MEMORY_RECLAIM_CONFIG_ERROR = None _OBSERVE_ROOT_DIAGNOSTICS = {"configured": [], "readable": [], "missing": []} _PROCESS_PRESSURE_DIAGNOSTICS = {} _CGROUP_RECLAIM_DIAGNOSTICS = {"configuredCount": 0, "eligibleCount": 0, "protected": []} def configured_observe_roots(): roots = config_list(MEMORY_CONFIG, "observeStateRoots", config_list(MEMORY_CONFIG, "webObserveRoots", [])) return [os.path.abspath(item) for item in roots if isinstance(item, str) and item.startswith("/")] def configured_observe_runs(root): global _OBSERVE_SCAN_TRUNCATED, _OBSERVE_SCAN_ERROR resolved_root = os.path.abspath(root) if resolved_root in _OBSERVE_RUN_CACHE: return list(_OBSERVE_RUN_CACHE[resolved_root]) max_depth = config_int(MEMORY_CONFIG, "observeRunDepth", 1, minimum=1, maximum=12) scan_limit = config_int(MEMORY_CONFIG, "observeScanLimit", 1000, minimum=1, maximum=100000) markers = set(["manifest.json", "heartbeat.json", "pid", "observer.pid", "runner.pid"]) runs = [] scanned = 0 if not os.path.isdir(resolved_root) or os.path.islink(resolved_root): return runs def record_walk_error(error): global _OBSERVE_SCAN_ERROR _OBSERVE_SCAN_ERROR = "observe-state-walk-failed:%s" % type(error).__name__ for current, directories, files in os.walk(resolved_root, topdown=True, followlinks=False, onerror=record_walk_error): relative = os.path.relpath(current, resolved_root) depth = 0 if relative == "." else relative.count(os.sep) + 1 directories[:] = sorted(name for name in directories if not os.path.islink(os.path.join(current, name))) if depth >= max_depth: directories[:] = [] if depth == 0: continue scanned += 1 if scanned > scan_limit: _OBSERVE_SCAN_TRUNCATED = True break if markers.intersection(files): runs.append(os.path.abspath(current)) directories[:] = [] if not _OBSERVE_SCAN_TRUNCATED: _OBSERVE_RUN_CACHE[resolved_root] = list(runs) return runs def is_observe_run_path(path): resolved = os.path.abspath(path) for root in configured_observe_roots(): prefix = root.rstrip("/") + "/" if resolved.startswith(prefix) and resolved in set(configured_observe_runs(root)): return True return False def is_direct_observe_run_path(path): # Kept as a compatibility name for generic artifact retention. Validation is # now marker/depth based because the configured roots contain dated trees. return is_observe_run_path(path) def path_has_open_fd(path): resolved = os.path.realpath(path) prefix = resolved.rstrip("/") + "/" proc_root = "/proc" try: pids = [name for name in os.listdir(proc_root) if name.isdigit()] except OSError: return True for pid in pids: base = os.path.join(proc_root, pid) for name in ["cwd", "root"]: try: target = os.path.realpath(os.readlink(os.path.join(base, name))) except OSError: continue if target == resolved or target.startswith(prefix): return True fd_dir = os.path.join(base, "fd") try: fds = os.listdir(fd_dir) except OSError: continue for fd in fds: try: target = os.path.realpath(os.readlink(os.path.join(fd_dir, fd))) except OSError: continue if target == resolved or target.startswith(prefix): return True return False def assert_web_observe_candidate(path): resolved = os.path.abspath(path) if not is_observe_run_path(resolved): raise RuntimeError("refusing to remove web-observe path outside configured marker/depth run roots: %s" % path) if os.path.islink(resolved) or not os.path.isdir(resolved): raise RuntimeError("refusing to remove non-directory or symlink web-observe path: %s" % path) stale_hours = config_float(MEMORY_CONFIG, "staleRunMaxAgeHours", 6.0, minimum=0.0) record = observe_run_record(resolved, stale_hours) if record.get("pidAlive"): raise RuntimeError("refusing to remove active web-observe run with live pid: %s" % path) if not record.get("staleSignal"): raise RuntimeError("refusing to remove web-observe run without stale signal: %s" % path) if path_has_open_fd(resolved): raise RuntimeError("refusing to remove web-observe run with open fd/cwd reference: %s" % path) return record def cgroup_path_for_pid(pid): try: with open("/proc/%s/cgroup" % int(pid), "r", encoding="utf-8") as handle: for line in handle: parts = line.rstrip("\n").split(":", 2) if len(parts) == 3 and parts[0] == "0" and parts[1] == "": return parts[2] except (OSError, ValueError): pass return None def build_process_pressure_diagnostics(rows): zombies = [row for row in rows.values() if row.get("state") == "Z"] by_comm = {} by_parent = {} for row in zombies: comm = str(row.get("comm") or "unknown") by_comm[comm] = by_comm.get(comm, 0) + 1 ppid = safe_int(row.get("ppid")) parent = rows.get(ppid) or {} bucket = by_parent.setdefault(ppid, { "ppid": ppid, "parentComm": parent.get("comm"), "parentCommandPreview": parent.get("commandPreview"), "zombieCount": 0, }) bucket["zombieCount"] += 1 top_rss = [] for row in sorted(rows.values(), key=lambda item: safe_int(item.get("rssBytes")), reverse=True)[:5]: top_rss.append({ "pid": safe_int(row.get("pid")), "ppid": safe_int(row.get("ppid")), "comm": row.get("comm"), "state": row.get("state"), "rssBytes": safe_int(row.get("rssBytes")), "rssHuman": fmt_bytes(safe_int(row.get("rssBytes"))), "cgroupPath": cgroup_path_for_pid(row.get("pid")), }) parent_rows = sorted(by_parent.values(), key=lambda item: safe_int(item.get("zombieCount")), reverse=True)[:5] for item in parent_rows: item.pop("parentCommandPreview", None) item["cgroupPath"] = cgroup_path_for_pid(item.get("ppid")) if item.get("ppid") else None return { "processCount": len(rows), "zombieCount": len(zombies), "zombiesByComm": [ {"comm": comm, "zombieCount": count} for comm, count in sorted(by_comm.items(), key=lambda item: item[1], reverse=True)[:5] ], "zombiesByParent": parent_rows, "topRss": top_rss, } def read_proc_process_table(): global _PROCESS_SCAN_TRUNCATED, _PROCESS_SCAN_ERROR, _PROCESS_PRESSURE_DIAGNOSTICS scan_limit = config_int(MEMORY_CONFIG, "processScanLimit", 16384, minimum=1, maximum=1000000) try: clock_ticks = int(os.sysconf("SC_CLK_TCK")) uptime = float(open("/proc/uptime", "r", encoding="utf-8").read().split()[0]) page_size = int(os.sysconf("SC_PAGE_SIZE")) pid_names = sorted((name for name in os.listdir("/proc") if name.isdigit()), key=int) if len(pid_names) > scan_limit: _PROCESS_SCAN_TRUNCATED = True return {} except Exception as exc: _PROCESS_SCAN_ERROR = "proc-process-table-read-failed:%s" % type(exc).__name__ return {} rows = {} stat_read_failures = 0 for pid_name in pid_names: base = os.path.join("/proc", pid_name) try: with open(os.path.join(base, "stat"), "r", encoding="utf-8") as handle: stat_text = handle.read().strip() close = stat_text.rfind(")") if close < 0: continue comm = stat_text[stat_text.find("(") + 1:close] fields = stat_text[close + 2:].split() if len(fields) < 22: continue pid = int(pid_name) state = fields[0] ppid = int(fields[1]) sid = int(fields[3]) start_ticks = int(fields[19]) try: with open(os.path.join(base, "cmdline"), "rb") as handle: command_line = handle.read(65536).replace(b"\0", b" ").decode("utf-8", errors="replace").strip() except OSError: command_line = "" try: with open(os.path.join(base, "statm"), "r", encoding="utf-8") as handle: rss_bytes = safe_int(handle.read().split()[1]) * page_size except Exception: rss_bytes = 0 rows[pid] = { "pid": pid, "ppid": ppid, "sid": sid, "comm": comm, "state": state, "startTicks": start_ticks, "ageSeconds": max(0, int(uptime - (float(start_ticks) / float(clock_ticks)))), "rssBytes": rss_bytes, "commandPreview": redact_command_preview(command_line), "commandLine": command_line, } except (OSError, ValueError, IndexError): stat_read_failures += 1 continue if not rows: _PROCESS_SCAN_ERROR = "proc-process-table-empty" elif stat_read_failures > max(32, len(pid_names) // 2): _PROCESS_SCAN_ERROR = "proc-process-stat-read-failure-threshold:%s-of-%s" % (stat_read_failures, len(pid_names)) _PROCESS_PRESSURE_DIAGNOSTICS = build_process_pressure_diagnostics(rows) return rows def validate_observe_run_markers(path): for name in ["heartbeat.json", "manifest.json"]: marker = os.path.join(path, name) if not os.path.lexists(marker): continue if os.path.islink(marker) or not os.path.isfile(marker): raise OSError("observe marker is not a regular file: %s" % marker) with open(marker, "r", encoding="utf-8") as handle: value = json.load(handle) if not isinstance(value, dict): raise OSError("observe JSON marker is not an object: %s" % marker) for name in ["pid", "observer.pid", "browser.pid", "runner.pid"]: marker = os.path.join(path, name) if not os.path.lexists(marker): continue if os.path.islink(marker) or not os.path.isfile(marker): raise OSError("observe pid marker is not a regular file: %s" % marker) with open(marker, "r", encoding="utf-8") as handle: raw = handle.read().strip() if not re.match(r"^[1-9]\d*$", raw): raise OSError("observe pid marker is invalid: %s" % marker) def observer_run_records(): global _OBSERVE_SCAN_ERROR, _OBSERVE_ROOT_DIAGNOSTICS stale_hours = config_float(MEMORY_CONFIG, "staleRunMaxAgeHours", 6.0, minimum=0.0) records = [] roots = configured_observe_roots() accessible_roots = [] missing_roots = [] for root in roots: try: root_stat = os.stat(root) if os.path.isdir(root) and not os.path.islink(root): accessible_roots.append(root) elif root_stat: _OBSERVE_SCAN_ERROR = "observe-state-root-not-directory" except FileNotFoundError: missing_roots.append(root) continue except OSError as exc: _OBSERVE_SCAN_ERROR = "observe-state-root-read-failed:%s" % type(exc).__name__ minimum_readable = config_int(MEMORY_CONFIG, "minimumReadableObserveRoots", 1, minimum=1, maximum=100) _OBSERVE_ROOT_DIAGNOSTICS = {"configured": list(roots), "readable": list(accessible_roots), "missing": missing_roots} if len(accessible_roots) < minimum_readable: _OBSERVE_SCAN_ERROR = _OBSERVE_SCAN_ERROR or "minimum-readable-observe-roots-not-met:%s-of-%s" % (len(accessible_roots), minimum_readable) for root in accessible_roots: for path in configured_observe_runs(root): try: validate_observe_run_markers(path) records.append(observe_run_record(path, stale_hours)) except (OSError, ValueError, json.JSONDecodeError) as exc: _OBSERVE_SCAN_ERROR = "observe-run-marker-read-failed:%s" % type(exc).__name__ continue return records def process_descendants(processes, root_pid): selected = set([int(root_pid)]) changed = True while changed: changed = False for pid, row in processes.items(): if pid not in selected and safe_int(row.get("ppid")) in selected: selected.add(pid) changed = True return selected def session_closed_tree(processes, sid): members = {pid: row for pid, row in processes.items() if safe_int(row.get("sid")) == int(sid)} if not members: return None roots = [pid for pid, row in members.items() if safe_int(row.get("ppid")) not in members] if len(roots) != 1: return None root_pid = roots[0] if process_descendants(members, root_pid) != set(members.keys()): return None return {"rootPid": root_pid, "members": members} def command_matches(row, patterns): haystack = str(row.get("commandLine") or "").lower() return any(str(pattern).lower() in haystack for pattern in patterns if str(pattern)) def memory_process_candidate_id(kind, root): return "%s:%s:%s:%s" % (kind, safe_int(root.get("sid")), safe_int(root.get("pid")), safe_int(root.get("startTicks"))) def read_cgroup_integer(path, name): try: with open(os.path.join(path, name), "r", encoding="utf-8") as handle: return int(handle.read().strip()) except (OSError, ValueError): return None def read_cgroup_memory_stat(path): selected = {} try: with open(os.path.join(path, "memory.stat"), "r", encoding="utf-8") as handle: for line in handle: parts = line.split() if len(parts) == 2 and parts[0] in set(["anon", "file", "inactive_anon", "inactive_file"]): selected[parts[0]] = safe_int(parts[1]) except OSError: return {} return selected def configured_cgroup_reclaim_targets(): global _MEMORY_RECLAIM_CONFIG_ERROR policy = MEMORY_CONFIG.get("cgroupReclaim") if policy is None: return [] if not isinstance(policy, dict): _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-policy-invalid" return [] if policy.get("enabled") is not True: return [] target_available = policy.get("targetMemAvailableBytes") swap_reserve = policy.get("minimumSwapFreeBytesAfterReclaim") targets = policy.get("targets") if isinstance(target_available, bool) or not isinstance(target_available, int) or target_available <= 0: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-memory-invalid" return [] if isinstance(swap_reserve, bool) or not isinstance(swap_reserve, int) or swap_reserve < 0: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-swap-reserve-invalid" return [] if not isinstance(targets, list) or not targets: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-targets-invalid" return [] validated = [] seen_ids = set() for index, item in enumerate(targets): if not isinstance(item, dict): _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-invalid:%s" % index return [] target_id = item.get("id") path = item.get("path") workload_selector = item.get("workloadSelector") reclaim_bytes = item.get("reclaimBytes") swappiness = item.get("swappiness") minimum_inactive_anon = item.get("minimumInactiveAnonBytes", 0) if not isinstance(target_id, str) or not re.match(r"^[A-Za-z0-9._-]{1,80}$", target_id) or target_id in seen_ids: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-id-invalid:%s" % index return [] if (path is None) == (workload_selector is None): _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-identity-invalid:%s" % target_id return [] if path is not None: if not isinstance(path, str) or (path != "/sys/fs/cgroup" and not path.startswith("/sys/fs/cgroup/")) or os.path.normpath(path) != path: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-path-invalid:%s" % target_id return [] if workload_selector is not None: if not isinstance(workload_selector, dict): _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-workload-selector-invalid:%s" % target_id return [] namespace = workload_selector.get("namespace") match_labels = workload_selector.get("matchLabels") max_matched_pods = workload_selector.get("maxMatchedPods") if not isinstance(namespace, str) or not re.match(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$", namespace): _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-workload-namespace-invalid:%s" % target_id return [] if not isinstance(match_labels, dict) or not match_labels or any( not isinstance(key, str) or not re.match(r"^[A-Za-z0-9./_-]{1,253}$", key) or not isinstance(value, str) or not re.match(r"^[A-Za-z0-9._-]{1,63}$", value) for key, value in match_labels.items() ): _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-workload-labels-invalid:%s" % target_id return [] if isinstance(max_matched_pods, bool) or not isinstance(max_matched_pods, int) or max_matched_pods < 1 or max_matched_pods > 20: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-workload-max-pods-invalid:%s" % target_id return [] if isinstance(reclaim_bytes, bool) or not isinstance(reclaim_bytes, int) or reclaim_bytes <= 0 or reclaim_bytes > 8 * 1024 * 1024 * 1024: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-bytes-invalid:%s" % target_id return [] if isinstance(swappiness, bool) or not isinstance(swappiness, int) or swappiness < 0 or swappiness > 200: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-swappiness-invalid:%s" % target_id return [] if isinstance(minimum_inactive_anon, bool) or not isinstance(minimum_inactive_anon, int) or minimum_inactive_anon < 0: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-minimum-inactive-anon-invalid:%s" % target_id return [] if swappiness > 0 and minimum_inactive_anon < reclaim_bytes: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-minimum-inactive-anon-too-small:%s" % target_id return [] priority = item.get("priority", index + 1) if isinstance(priority, bool) or not isinstance(priority, int) or priority < 1 or priority > 1000: _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-priority-invalid:%s" % target_id return [] seen_ids.add(target_id) validated.append({ "id": target_id, "path": path, "workloadSelector": workload_selector, "reclaimBytes": reclaim_bytes, "swappiness": swappiness, "minimumInactiveAnonBytes": minimum_inactive_anon, "priority": priority, "targetMemAvailableBytes": target_available, "minimumSwapFreeBytesAfterReclaim": swap_reserve, }) return sorted(validated, key=lambda item: item.get("priority")) _CGROUP_RECLAIM_RUNTIME_INVENTORY = None def cgroup_reclaim_runtime_inventory(): global _CGROUP_RECLAIM_RUNTIME_INVENTORY if _CGROUP_RECLAIM_RUNTIME_INVENTORY is not None: return _CGROUP_RECLAIM_RUNTIME_INVENTORY attribution = MEMORY_CONFIG.get("attribution") or {} timeout = safe_int(attribution.get("commandTimeoutSeconds")) runtime = command(["k3s", "crictl", "pods", "-o", "json"], timeout) if runtime.get("exitCode") != 0: _CGROUP_RECLAIM_RUNTIME_INVENTORY = { "ok": False, "reason": "workload-runtime-inventory-failed", "command": bounded(runtime), "pods": [], } return _CGROUP_RECLAIM_RUNTIME_INVENTORY try: payload = json.loads(runtime.get("stdout") or "{}") except (TypeError, ValueError, json.JSONDecodeError): _CGROUP_RECLAIM_RUNTIME_INVENTORY = {"ok": False, "reason": "workload-runtime-inventory-invalid", "pods": []} return _CGROUP_RECLAIM_RUNTIME_INVENTORY cgroup_root = attribution.get("cgroupRoot") pod_paths = {} scan_limit = safe_int(attribution.get("cgroupScanLimit")) scanned = 0 for current, directories, files in os.walk(os.path.join(cgroup_root, "kubepods.slice"), topdown=True, followlinks=False): directories[:] = sorted(name for name in directories if not os.path.islink(os.path.join(current, name))) scanned += 1 if scanned > scan_limit: _CGROUP_RECLAIM_RUNTIME_INVENTORY = {"ok": False, "reason": "workload-cgroup-scan-truncated", "pods": []} return _CGROUP_RECLAIM_RUNTIME_INVENTORY match = re.search(r"pod([0-9a-f_]{36})\.slice$", os.path.basename(current)) if match is not None and "memory.current" in files: pod_paths[match.group(1).replace("_", "-")] = current directories[:] = [] pods = [] for pod in payload.get("items") or []: metadata = pod.get("metadata") if isinstance(pod, dict) else None uid = metadata.get("uid") if isinstance(metadata, dict) else None name = metadata.get("name") if isinstance(metadata, dict) else None namespace = metadata.get("namespace") if isinstance(metadata, dict) else None labels = pod.get("labels") if isinstance(pod, dict) else None if pod.get("state") != "SANDBOX_READY" or not all(isinstance(value, str) and value for value in [uid, name, namespace]): continue pods.append({ "namespace": namespace, "name": name, "uid": uid, "labels": labels if isinstance(labels, dict) else {}, "path": pod_paths.get(uid), }) _CGROUP_RECLAIM_RUNTIME_INVENTORY = {"ok": True, "pods": pods} return _CGROUP_RECLAIM_RUNTIME_INVENTORY def resolve_cgroup_reclaim_target(item): path = item.get("path") if isinstance(path, str): return {"ok": True, "matches": [{"path": path, "pod": None}]} selector = item.get("workloadSelector") or {} labels = selector.get("matchLabels") or {} label_selector = ",".join("%s=%s" % pair for pair in sorted(labels.items())) inventory = cgroup_reclaim_runtime_inventory() if inventory.get("ok") is not True: return {"ok": False, "reason": inventory.get("reason"), "command": inventory.get("command"), "matches": []} pods = [pod for pod in inventory.get("pods") or [] if pod.get("namespace") == selector.get("namespace") and all((pod.get("labels") or {}).get(key) == value for key, value in labels.items()) ] if len(pods) == 0: return {"ok": False, "reason": "workload-selector-no-running-pods", "matches": []} if len(pods) > safe_int(selector.get("maxMatchedPods")): return {"ok": False, "reason": "workload-selector-max-pods-exceeded", "matchedPodCount": len(pods), "matches": []} matches = [] for pod in pods: uid = pod.get("uid") name = pod.get("name") resolved_path = pod.get("path") if not isinstance(uid, str) or not isinstance(name, str) or not isinstance(resolved_path, str): return {"ok": False, "reason": "workload-pod-cgroup-unavailable", "matches": []} matches.append({ "path": resolved_path, "pod": {"namespace": selector.get("namespace"), "name": name, "uid": uid, "labelSelector": label_selector}, }) return {"ok": True, "matches": matches} def collect_cgroup_reclaim_candidates(): global _MEMORY_RECLAIM_CONFIG_ERROR, _CGROUP_RECLAIM_DIAGNOSTICS targets = configured_cgroup_reclaim_targets() diagnostics = {"configuredCount": len(targets), "eligibleCount": 0, "protected": []} _CGROUP_RECLAIM_DIAGNOSTICS = diagnostics if _MEMORY_RECLAIM_CONFIG_ERROR or not targets: return [] host = collect_memory_snapshot() if host.get("ok") is not True: _MEMORY_RECLAIM_CONFIG_ERROR = "host-memory-snapshot-unavailable" return [] available = safe_int((host.get("memory") or {}).get("availableBytes")) swap_free = safe_int((host.get("swap") or {}).get("freeBytes")) candidates = [] for item in targets: resolution = resolve_cgroup_reclaim_target(item) if resolution.get("ok") is not True: diagnostics["protected"].append({ "id": item.get("id"), "workloadSelector": item.get("workloadSelector"), "reason": resolution.get("reason"), "matchedPodCount": resolution.get("matchedPodCount"), }) continue for resolved in resolution.get("matches") or []: path = resolved.get("path") pod = resolved.get("pod") reclaim_path = os.path.join(path, "memory.reclaim") protected_reason = None if os.path.realpath(path) != path or not os.path.isdir(path) or os.path.islink(path): protected_reason = "cgroup-path-unavailable-or-redirected" elif not os.path.exists(reclaim_path) or not os.access(reclaim_path, os.W_OK): protected_reason = "memory-reclaim-unavailable" elif available > safe_int(item.get("targetMemAvailableBytes")): protected_reason = "target-memavailable-already-met" elif safe_int(item.get("swappiness")) > 0 and swap_free - safe_int(item.get("reclaimBytes")) < safe_int(item.get("minimumSwapFreeBytesAfterReclaim")): protected_reason = "minimum-swap-free-reserve" current = read_cgroup_integer(path, "memory.current") memory_stat = read_cgroup_memory_stat(path) inactive_anon = safe_int(memory_stat.get("inactive_anon")) if protected_reason is None and safe_int(item.get("swappiness")) > 0 and inactive_anon < safe_int(item.get("minimumInactiveAnonBytes")): protected_reason = "minimum-inactive-anon-not-met" if protected_reason is not None: diagnostics["protected"].append({ "id": item.get("id"), "path": path, "workloadSelector": item.get("workloadSelector"), "resolvedPod": pod, "reason": protected_reason, "memoryCurrentBytes": current, "inactiveAnonBytes": inactive_anon, }) continue reclaim_bytes = safe_int(item.get("reclaimBytes")) identity_suffix = (pod or {}).get("uid") candidate_id = "cgroup-memory-reclaim:%s" % item.get("id") if identity_suffix: candidate_id += ":%s" % identity_suffix candidates.append({ "id": candidate_id, "configId": item.get("id"), "kind": "cgroup-memory-reclaim", "risk": "medium", "description": "Request bounded cgroup v2 memory reclaim from a YAML-resolved workload", "reason": "host-memavailable-below-yaml-target", "path": path, "workloadSelector": item.get("workloadSelector"), "resolvedPod": pod, "priority": safe_int(item.get("priority")), "reclaimBytes": reclaim_bytes, "swappiness": safe_int(item.get("swappiness")), "minimumInactiveAnonBytes": safe_int(item.get("minimumInactiveAnonBytes")), "targetMemAvailableBytes": safe_int(item.get("targetMemAvailableBytes")), "minimumSwapFreeBytesAfterReclaim": safe_int(item.get("minimumSwapFreeBytesAfterReclaim")), "memoryCurrentBytes": current, "memorySwapCurrentBytes": read_cgroup_integer(path, "memory.swap.current"), "memoryStat": memory_stat, "requestedMemoryReclaimBytes": reclaim_bytes, "requestedMemoryReclaim": fmt_bytes(reclaim_bytes), "estimateKind": "kernel-request-upper-bound", "expectedMemoryRssBytes": reclaim_bytes, "expectedMemoryRss": fmt_bytes(reclaim_bytes), "estimatedDiskBytes": 0, "estimatedReclaimBytes": 0, "action": {"op": "cgroup-v2-memory-reclaim", "allowlist": "yaml-workload-selector-or-exact-path"}, }) diagnostics["eligibleCount"] = len(candidates) return candidates def collect_memory_reclaim_candidates(): global _OBSERVE_SCAN_TRUNCATED, _PROCESS_SCAN_TRUNCATED, _OBSERVE_SCAN_ERROR, _PROCESS_SCAN_ERROR, _MEMORY_RECLAIM_CONFIG_ERROR, _OBSERVE_ROOT_DIAGNOSTICS, _PROCESS_PRESSURE_DIAGNOSTICS, _CGROUP_RECLAIM_DIAGNOSTICS _OBSERVE_SCAN_TRUNCATED = False _PROCESS_SCAN_TRUNCATED = False _OBSERVE_SCAN_ERROR = None _PROCESS_SCAN_ERROR = None _MEMORY_RECLAIM_CONFIG_ERROR = None _OBSERVE_ROOT_DIAGNOSTICS = {"configured": [], "readable": [], "missing": []} _PROCESS_PRESSURE_DIAGNOSTICS = {} _CGROUP_RECLAIM_DIAGNOSTICS = {"configuredCount": 0, "eligibleCount": 0, "protected": []} _OBSERVE_RUN_CACHE.clear() required_keys = [ "observeStateRoots", "minimumReadableObserveRoots", "observeRunDepth", "observeScanLimit", "staleRunMaxAgeHours", "processScanLimit", "processStaleMinAgeSeconds", "maxMemoryReclaimCandidates", "terminationGraceSeconds", "terminationPollMilliseconds", "observerRootPatterns", "browserRootPatterns", "browserProcessPatterns", ] if any(key not in MEMORY_CONFIG for key in required_keys): _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-memory-pressure-policy-missing" return [] list_keys = ["observeStateRoots", "observerRootPatterns", "browserRootPatterns", "browserProcessPatterns"] integer_keys = [ "observeRunDepth", "observeScanLimit", "processScanLimit", "processStaleMinAgeSeconds", "maxMemoryReclaimCandidates", "terminationGraceSeconds", "terminationPollMilliseconds", ] if any(not isinstance(MEMORY_CONFIG.get(key), list) or not MEMORY_CONFIG.get(key) for key in list_keys): _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-memory-pressure-list-invalid" return [] if any(any(not isinstance(item, str) or not item.strip() or "\0" in item for item in MEMORY_CONFIG.get(key)) for key in list_keys): _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-memory-pressure-list-entry-invalid" return [] if any(not os.path.isabs(item) for item in MEMORY_CONFIG.get("observeStateRoots")): _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-observe-root-not-absolute" return [] integer_ranges = { "observeRunDepth": (1, 12), "minimumReadableObserveRoots": (1, 100), "observeScanLimit": (1, 100000), "processScanLimit": (1, 1000000), "processStaleMinAgeSeconds": (60, 604800), "maxMemoryReclaimCandidates": (1, 100), "terminationGraceSeconds": (1, 30), "terminationPollMilliseconds": (50, 1000), } if any(isinstance(MEMORY_CONFIG.get(key), bool) or not isinstance(MEMORY_CONFIG.get(key), int) or MEMORY_CONFIG.get(key) < minimum or MEMORY_CONFIG.get(key) > maximum for key, (minimum, maximum) in integer_ranges.items()): _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-memory-pressure-integer-invalid" return [] if MEMORY_CONFIG.get("minimumReadableObserveRoots") > len(MEMORY_CONFIG.get("observeStateRoots")): _MEMORY_RECLAIM_CONFIG_ERROR = "minimum-readable-observe-roots-exceeds-configured-roots" return [] stale_hours = MEMORY_CONFIG.get("staleRunMaxAgeHours") if isinstance(stale_hours, bool) or not isinstance(stale_hours, (int, float)) or stale_hours <= 0 or stale_hours > 8760: _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-stale-age-invalid" return [] browser_patterns = config_list(MEMORY_CONFIG, "browserProcessPatterns", []) root_patterns = config_list(MEMORY_CONFIG, "browserRootPatterns", []) observer_patterns = config_list(MEMORY_CONFIG, "observerRootPatterns", []) stale_seconds = config_int(MEMORY_CONFIG, "processStaleMinAgeSeconds", 3600, minimum=60, maximum=604800) if not browser_patterns or not root_patterns or not observer_patterns: return [] processes = read_proc_process_table() records = observer_run_records() if not processes and _PROCESS_SCAN_ERROR is None: _PROCESS_SCAN_ERROR = "proc-process-table-empty" if _PROCESS_SCAN_TRUNCATED or _OBSERVE_SCAN_TRUNCATED or _PROCESS_SCAN_ERROR or _OBSERVE_SCAN_ERROR: return [] fresh_observer_pids = set() stale_observer_pids = set() for record in records: pid = safe_int(record.get("pid")) if pid <= 0 or pid not in processes: continue if record.get("staleLiveSignal"): stale_observer_pids.add(pid) else: fresh_observer_pids.add(pid) protected_pids = set() for pid in fresh_observer_pids: protected_pids.update(process_descendants(processes, pid)) candidates = collect_cgroup_reclaim_candidates() for sid in sorted(set(safe_int(row.get("sid")) for row in processes.values() if safe_int(row.get("sid")) > 0)): closed = session_closed_tree(processes, sid) if closed is None: continue members = closed["members"] member_pids = set(members.keys()) if member_pids.intersection(protected_pids): continue root = members[closed["rootPid"]] if safe_int(root.get("ageSeconds")) < stale_seconds: continue has_browser = any(command_matches(row, browser_patterns) for row in members.values()) intersecting_stale = member_pids.intersection(stale_observer_pids) kind = None reason = None if intersecting_stale and command_matches(root, observer_patterns): kind = "stale-web-observer-process-tree" reason = "stale-observer-heartbeat" elif not member_pids.intersection(set(safe_int(item.get("pid")) for item in records if item.get("pidAlive"))): if command_matches(root, root_patterns) and has_browser: kind = "browser-orphan-process-tree" reason = "browser-session-without-observer-record" if kind is None: continue identities = [ { "pid": pid, "ppid": safe_int(row.get("ppid")), "sid": safe_int(row.get("sid")), "startTicks": safe_int(row.get("startTicks")), "rssBytes": safe_int(row.get("rssBytes")), "commandPreview": row.get("commandPreview"), } for pid, row in sorted(members.items()) ] expected_memory = sum(safe_int(item.get("rssBytes")) for item in identities) candidates.append({ "id": memory_process_candidate_id(kind, root), "kind": kind, "risk": "medium", "description": "Terminate a YAML-allowlisted stale WebProbe process session after identity and active-observer revalidation", "reason": reason, "rootPid": safe_int(root.get("pid")), "sid": sid, "rootStartTicks": safe_int(root.get("startTicks")), "processCount": len(identities), "processIdentities": identities, "expectedMemoryRssBytes": expected_memory, "expectedMemoryRss": fmt_bytes(expected_memory), "estimatedDiskBytes": 0, "estimatedReclaimBytes": 0, "activeObserverIntersection": False, "treeClosure": "unique-root-ppid-sid-closed", "action": {"op": "term-wait-kill-exact-process-tree", "allowlist": "yaml-web-probe-memory-pressure"}, }) max_candidates = config_int(MEMORY_CONFIG, "maxMemoryReclaimCandidates", 20, minimum=1, maximum=100) return sorted(candidates, key=lambda item: ( 0 if item.get("kind") == "cgroup-memory-reclaim" else 1, safe_int(item.get("priority"), 1000), -safe_int(item.get("expectedMemoryRssBytes")), ))[:max_candidates] def memory_reclaim_scan_diagnostics(): blocked = bool(_MEMORY_RECLAIM_CONFIG_ERROR or _OBSERVE_SCAN_TRUNCATED or _PROCESS_SCAN_TRUNCATED or _OBSERVE_SCAN_ERROR or _PROCESS_SCAN_ERROR) return { "ok": not blocked, "blocked": blocked, "configError": _MEMORY_RECLAIM_CONFIG_ERROR, "observeScanTruncated": bool(_OBSERVE_SCAN_TRUNCATED), "processScanTruncated": bool(_PROCESS_SCAN_TRUNCATED), "observeScanError": _OBSERVE_SCAN_ERROR, "processScanError": _PROCESS_SCAN_ERROR, "processPressure": _PROCESS_PRESSURE_DIAGNOSTICS, "cgroupReclaim": _CGROUP_RECLAIM_DIAGNOSTICS, "observeRoots": { "configuredCount": len(_OBSERVE_ROOT_DIAGNOSTICS.get("configured") or []), "readableCount": len(_OBSERVE_ROOT_DIAGNOSTICS.get("readable") or []), "missingCount": len(_OBSERVE_ROOT_DIAGNOSTICS.get("missing") or []), "minimumReadable": safe_int(MEMORY_CONFIG.get("minimumReadableObserveRoots")), "configured": _OBSERVE_ROOT_DIAGNOSTICS.get("configured") or [], "readable": _OBSERVE_ROOT_DIAGNOSTICS.get("readable") or [], "missing": _OBSERVE_ROOT_DIAGNOSTICS.get("missing") or [], }, "policySource": "config/unidesk-cli.yaml#gc.remote.targets.%s.memoryPressure" % PROVIDER_ID, } def matching_live_identity(identity, processes): pid = safe_int(identity.get("pid")) row = processes.get(pid) return row if row \ and safe_int(row.get("startTicks")) == safe_int(identity.get("startTicks")) \ and safe_int(row.get("ppid")) == safe_int(identity.get("ppid")) \ and safe_int(row.get("sid")) == safe_int(identity.get("sid")) else None def matching_surviving_identity(identity, processes): pid = safe_int(identity.get("pid")) row = processes.get(pid) return row if row \ and safe_int(row.get("startTicks")) == safe_int(identity.get("startTicks")) else None def require_complete_process_scan(processes, phase): if not processes or _PROCESS_SCAN_ERROR or _PROCESS_SCAN_TRUNCATED: raise RuntimeError("memory process identity scan blocked during %s: %s" % ( phase, _PROCESS_SCAN_ERROR or "process-scan-truncated", )) return processes def extend_with_allowlisted_session_members(identities, processes, sid, browser_patterns): known = {(safe_int(item.get("pid")), safe_int(item.get("startTicks"))) for item in identities} added = [] for pid, row in processes.items(): identity_key = (safe_int(pid), safe_int(row.get("startTicks"))) if identity_key in known or safe_int(row.get("sid")) != safe_int(sid) or not command_matches(row, browser_patterns): continue item = { "pid": safe_int(pid), "ppid": safe_int(row.get("ppid")), "sid": safe_int(row.get("sid")), "startTicks": safe_int(row.get("startTicks")), "rssBytes": safe_int(row.get("rssBytes")), "commandPreview": row.get("commandPreview"), "discoveredAfterTerm": True, } identities.append(item) added.append(item) known.add(identity_key) return added def execute_cgroup_reclaim_candidate(candidate): global _CGROUP_RECLAIM_RUNTIME_INVENTORY config_id = candidate.get("configId") configured = next((item for item in configured_cgroup_reclaim_targets() if item.get("id") == config_id), None) if configured is None: raise RuntimeError("cgroup reclaim candidate is no longer present in YAML allowlist") for key in ["workloadSelector", "reclaimBytes", "swappiness", "minimumInactiveAnonBytes", "targetMemAvailableBytes", "minimumSwapFreeBytesAfterReclaim"]: if configured.get(key) != candidate.get(key): raise RuntimeError("cgroup reclaim candidate no longer matches YAML: %s" % key) _CGROUP_RECLAIM_RUNTIME_INVENTORY = None resolution = resolve_cgroup_reclaim_target(configured) if resolution.get("ok") is not True: raise RuntimeError("cgroup reclaim workload selector no longer resolves: %s" % resolution.get("reason")) path = candidate.get("path") resolved_uid = ((candidate.get("resolvedPod") or {}).get("uid")) current_match = next((item for item in resolution.get("matches") or [] if item.get("path") == path and ((item.get("pod") or {}).get("uid")) == resolved_uid), None) if current_match is None: raise RuntimeError("cgroup reclaim candidate identity changed before execution") reclaim_path = os.path.join(path, "memory.reclaim") if os.path.realpath(path) != path or not os.path.isdir(path) or os.path.islink(path): raise RuntimeError("cgroup reclaim path is unavailable or redirected") if not os.path.exists(reclaim_path) or not os.access(reclaim_path, os.W_OK): raise RuntimeError("cgroup memory.reclaim is unavailable") before = collect_memory_snapshot() before_available = safe_int((before.get("memory") or {}).get("availableBytes"), None) if before_available is None: raise RuntimeError("MemAvailable is unavailable before cgroup reclaim") if before_available > safe_int(configured.get("targetMemAvailableBytes")): return { "status": "succeeded", "outcome": "target-memavailable-already-met", "reclaimedMemoryBytes": 0, "memoryBefore": before, "memoryAfter": before, "cgroupMemoryBeforeBytes": read_cgroup_integer(path, "memory.current"), "cgroupMemoryAfterBytes": read_cgroup_integer(path, "memory.current"), "kernelWriteAttempted": False, } swap_free = safe_int((before.get("swap") or {}).get("freeBytes"), None) if safe_int(configured.get("swappiness")) > 0: if swap_free is None: raise RuntimeError("SwapFree is unavailable before cgroup reclaim") required = safe_int(configured.get("reclaimBytes")) + safe_int(configured.get("minimumSwapFreeBytesAfterReclaim")) if swap_free < required: raise RuntimeError("cgroup reclaim would violate YAML minimum SwapFree reserve") inactive_anon = safe_int(read_cgroup_memory_stat(path).get("inactive_anon")) if inactive_anon < safe_int(configured.get("minimumInactiveAnonBytes")): raise RuntimeError("minimum inactive anonymous memory is no longer available") cgroup_before = read_cgroup_integer(path, "memory.current") write_error = None try: with open(reclaim_path, "w", encoding="utf-8") as handle: handle.write("%s swappiness=%s\n" % (safe_int(configured.get("reclaimBytes")), safe_int(configured.get("swappiness")))) except OSError as exc: write_error = "%s:%s" % (type(exc).__name__, getattr(exc, "errno", None)) time.sleep(0.2) after = collect_memory_snapshot() after_available = safe_int((after.get("memory") or {}).get("availableBytes"), None) cgroup_after = read_cgroup_integer(path, "memory.current") cgroup_delta = max(0, cgroup_before - cgroup_after) if cgroup_before is not None and cgroup_after is not None else 0 host_delta = max(0, after_available - before_available) if after_available is not None else 0 succeeded = write_error is None or cgroup_delta > 0 or host_delta > 0 partial = max(cgroup_delta, host_delta) < safe_int(configured.get("reclaimBytes")) if write_error and succeeded: outcome = "partial-reclaim-after-kernel-error" elif succeeded and partial: outcome = "partial-reclaim" else: outcome = "reclaim-requested" if succeeded else "reclaim-request-failed" return { "status": "succeeded" if succeeded else "failed", "outcome": outcome, "reclaimedMemoryBytes": max(cgroup_delta, host_delta), "hostMemAvailableDeltaBytes": (after_available - before_available) if after_available is not None else None, "cgroupMemoryDeltaBytes": cgroup_delta, "cgroupMemoryBeforeBytes": cgroup_before, "cgroupMemoryAfterBytes": cgroup_after, "kernelWriteAttempted": True, "kernelWriteError": write_error, "memoryBefore": before, "memoryAfter": after, } def execute_memory_process_candidate(candidate): current = next((item for item in collect_memory_reclaim_candidates() if item.get("id") == candidate.get("id")), None) if current is None: raise RuntimeError("memory candidate no longer has the same closed identity or is now protected") before = collect_memory_snapshot() identities = [dict(item) for item in (current.get("processIdentities") or [])] candidate_sid = safe_int(current.get("sid")) browser_patterns = config_list(MEMORY_CONFIG, "browserProcessPatterns", []) late_members = [] processes = require_complete_process_scan(read_proc_process_table(), "pre-term-revalidation") live = [item for item in identities if matching_live_identity(item, processes)] for item in sorted(live, key=lambda value: safe_int(value.get("pid")), reverse=True): try: os.kill(safe_int(item.get("pid")), signal.SIGTERM) except ProcessLookupError: pass grace_seconds = config_int(MEMORY_CONFIG, "terminationGraceSeconds", 5, minimum=1, maximum=30) poll_ms = config_int(MEMORY_CONFIG, "terminationPollMilliseconds", 200, minimum=50, maximum=1000) deadline = time.time() + grace_seconds residual = [] while time.time() < deadline: processes = require_complete_process_scan(read_proc_process_table(), "term-grace-poll") late_members.extend(extend_with_allowlisted_session_members(identities, processes, candidate_sid, browser_patterns)) residual = [item for item in identities if matching_surviving_identity(item, processes)] if not residual: break time.sleep(float(poll_ms) / 1000.0) processes = require_complete_process_scan(read_proc_process_table(), "pre-kill-revalidation") late_members.extend(extend_with_allowlisted_session_members(identities, processes, candidate_sid, browser_patterns)) residual = [item for item in identities if matching_surviving_identity(item, processes)] for item in sorted(residual, key=lambda value: safe_int(value.get("pid")), reverse=True): try: os.kill(safe_int(item.get("pid")), signal.SIGKILL) except ProcessLookupError: pass final_deadline = time.time() + min(2.0, float(grace_seconds)) remaining = residual while time.time() < final_deadline: processes = require_complete_process_scan(read_proc_process_table(), "kill-grace-poll") late_members.extend(extend_with_allowlisted_session_members(identities, processes, candidate_sid, browser_patterns)) remaining = [item for item in identities if matching_surviving_identity(item, processes)] if not remaining: break time.sleep(float(poll_ms) / 1000.0) after = collect_memory_snapshot() before_available = safe_int((before.get("memory") or {}).get("availableBytes"), None) after_available = safe_int((after.get("memory") or {}).get("availableBytes"), None) return { "status": "failed" if remaining else "succeeded", "reclaimedBytes": 0, "reclaimedMemoryBytes": max(0, after_available - before_available) if before_available is not None and after_available is not None else None, "expectedMemoryRssBytes": current.get("expectedMemoryRssBytes"), "termSignalCount": len(live), "killSignalCount": len(residual), "remainingIdentityCount": len(remaining), "lateAllowlistedIdentityCount": len(late_members), "remainingIdentities": [ { "pid": safe_int(item.get("pid")), "startTicks": safe_int(item.get("startTicks")), "sid": safe_int(item.get("sid")), "originalPpid": safe_int(item.get("ppid")), "observedPpid": safe_int((matching_surviving_identity(item, processes) or {}).get("ppid")), } for item in remaining ], "memoryBefore": before, "memoryAfter": after, "identityRevalidated": True, "activeObserverIntersection": False, }