diff --git a/.agents/skills/unidesk-gc/SKILL.md b/.agents/skills/unidesk-gc/SKILL.md index 10752c86..07b4ec11 100644 --- a/.agents/skills/unidesk-gc/SKILL.md +++ b/.agents/skills/unidesk-gc/SKILL.md @@ -120,6 +120,32 @@ rm -rf -- /tmp/unidesk-cli-output - provider host 统一使用 `bun scripts/cli.ts gc remote ...`。 - 远端长任务必须异步执行，并用 `status --job-id` 短查询，不得维持长 SSH 会话。 +### 内存分布与 Kubernetes 留存 + +先用紧凑 overview 归因真实内存压力： + +```bash +bun scripts/cli.ts gc remote memory-distribution +``` + +- 默认输出物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布； +- swap 单独披露，不能计入 WebProbe 启动资格； +- 需要完整明细时才使用 `--full`。 + +Kubernetes 历史对象手动回收固定使用： + +```bash +bun scripts/cli.ts gc remote retention plan --limit 500 +bun scripts/cli.ts gc remote retention run --confirm --limit 500 +bun scripts/cli.ts gc remote status --job-id +``` + +- 筛选、保留窗口、分组、批量和级联策略只来自目标节点的 `kubernetesObjectRetention` YAML； +- 手动入口与 `policyTimer.includeKubernetesObjectRetention` 自动阶段必须复用同一筛选器； +- 活跃、过新、未知分组、时间不可解析、扫描截断和 Table 形状漂移全部保护； +- 只删除 YAML 指定的根对象，通过 ownerRef 级联，不直接删除下游 TaskRun、Pod、Secret 或 Event； +- 自动回收由 `gc remote policy plan|install --confirm|status` 管理，周期和单批上限由 YAML 控制。 + ## WebProbe 低内存人工处置 - 唯一资格指标： diff --git a/config/unidesk-cli.yaml b/config/unidesk-cli.yaml index de5d1975..05ee80b2 100644 --- a/config/unidesk-cli.yaml +++ b/config/unidesk-cli.yaml @@ -160,7 +160,33 @@ gc: observeRunDepth: 6 observeScanLimit: 2000 staleRunMaxAgeHours: 1 - processScanLimit: 16384 + processScanLimit: 32768 + attribution: + cgroupRoot: /sys/fs/cgroup + cgroupScanLimit: 32768 + topLevelLimit: 4 + systemServiceLimit: 4 + podLimit: 4 + processLimit: 4 + processPssLimit: 4 + commandTimeoutSeconds: 20 + kubeconfigPath: /etc/rancher/k3s/k3s.yaml + targetMemAvailableBytes: 4294967296 + objectCountResources: + - pods + - pipelineruns.tekton.dev + - taskruns.tekton.dev + - events + - secrets + - persistentvolumeclaims + - persistentvolumes + namespaceCountNamespaces: + - devops-infra + - agentrun-ci + - hwlab-ci + - pikaoa-ci + - selfmedia-ci + namespaceCountConcurrency: 8 processStaleMinAgeSeconds: 3600 maxMemoryReclaimCandidates: 20 terminationGraceSeconds: 5 @@ -181,6 +207,59 @@ gc: - /usr/bin/chromium-browser - /opt/google/chrome/ - /headless_shell + cgroupReclaim: + enabled: false + kubernetesObjectRetention: + enabled: true + configId: devops-infra-tekton-history + kubeconfigPath: /etc/rancher/k3s/k3s.yaml + namespace: devops-infra + resource: pipelineruns.tekton.dev + listMode: server-table + countPath: /apis/tekton.dev/v1/namespaces/{namespace}/pipelineruns + terminalConditionType: Succeeded + terminalStatuses: + - "True" + - "False" + minAgeHours: 24 + keepLatestPerGroup: 3 + manualMaxDeletePerRun: 500 + automaticMaxDeletePerRun: 100 + previewLimit: 10 + scanLimit: 5000 + requestTimeoutSeconds: 45 + deleteBatchSize: 50 + deleteTimeoutSeconds: 60 + postDeleteSettleSeconds: 5 + propagationPolicy: Background + groupSelectors: + - id: outer-pac-event + groupByLabels: + - pipelinesascode.tekton.dev/repository + - pipelinesascode.tekton.dev/original-prname + - id: inner-deterministic-publish + groupByLabels: + - unidesk.ai/node + - unidesk.ai/lane + - unidesk.ai/spec-ref + policyTimer: + enabled: true + name: unidesk-devops-infra-retention + onCalendar: "*:0/15" + randomizedDelaySec: 2min + stateDir: /var/lib/unidesk-gc + configDir: /etc/unidesk-gc + journalTargetBytes: 256MiB + tmpMinAgeHours: 24 + includeJournal: false + includeTmp: false + includeAptCache: false + includeToolCaches: false + includeWebObserveArtifacts: false + includeK3sImageCache: false + includeHostContainerdCache: false + includeLocalPathOrphans: false + includeKubernetesObjectRetention: true JD01: hostDockerGc: dockerJsonLogs: @@ -249,12 +328,15 @@ gc: configDir: /etc/unidesk-gc journalTargetBytes: 256MiB tmpMinAgeHours: 24 + includeJournal: true + includeTmp: true includeAptCache: true includeToolCaches: false includeWebObserveArtifacts: true includeK3sImageCache: true includeHostContainerdCache: true includeLocalPathOrphans: true + includeKubernetesObjectRetention: false agentrunSessionPvcs: enabled: true namespace: agentrun-v02 diff --git a/config/unidesk-host-k8s.yaml b/config/unidesk-host-k8s.yaml index 2941ed46..5048143a 100644 --- a/config/unidesk-host-k8s.yaml +++ b/config/unidesk-host-k8s.yaml @@ -162,6 +162,7 @@ services: secretKey: DATABASE_URL storage: primary: github-repo + healthRefreshSeconds: 300 github: repo: pikasTech/decision-center-data sshUrl: git@github.com:pikasTech/decision-center-data.git diff --git a/docs/MDTODO/details/web-observe-runtime-reliability/R1.2.1_Task_Report.md b/docs/MDTODO/details/web-observe-runtime-reliability/R1.2.1_Task_Report.md new file mode 100644 index 00000000..89a1cbe4 --- /dev/null +++ b/docs/MDTODO/details/web-observe-runtime-reliability/R1.2.1_Task_Report.md @@ -0,0 +1,28 @@ +# R1.2.1 任务报告 + +## 结论 + +已为 NC01 `devops-infra` 增加 YAML-first Kubernetes 历史对象留存。手动 `retention plan|run|status` 与 systemd policy timer 复用 `scripts/src/gc-remote-kubernetes-retention.py` 的同一个 fail-closed 筛选器；只删除终态、超过保留窗口、分组可识别且不属于每组最新保留数量的 PipelineRun，通过 Background ownerRef 级联回收下游对象。 + +## 配置与保护 + +- 配置真相：`config/unidesk-cli.yaml#gc.remote.targets.NC01.kubernetesObjectRetention` 与 `.policyTimer`。 +- 手动单批上限 500，自动单批上限 100，自动周期每 15 分钟。 +- 固定保护活跃、排队、过新、未知分组、时间不可解析、扫描截断和 Table 形状漂移对象。 +- 不直接删除 TaskRun、Pod、Secret、Event、PVC/PV 或 runtime workload。 +- NC01 policy 仅启用 `kubernetes-object-retention`，journal、tmp、cache、image 和 PVC 阶段全部关闭。 + +## 真实运行证据 + +- 初始 `devops-infra`：PipelineRun 1923、TaskRun 1922、Pod 932、Secret 1613。 +- 三个异步 job 分别成功删除 500、500、484 个 PipelineRun，全部批次成功。 +- 终态 overview：PipelineRun 451、TaskRun 450、Pod 444、Secret 455；剩余对象主要为活跃、24 小时内对象和每组最新保留对象。 +- timer `unidesk-devops-infra-retention.timer` 已 `loaded/active/waiting`，配置、runner 和状态文件均由 YAML 渲染。 +- 首次自动轮次 `failedCount=0`，只由 retention 阶段回收 6 个到龄 PipelineRun，下一轮已自动排期。 +- Python `py_compile`、`git diff --check` 与 `bun scripts/cli.ts check --syntax-only` 通过；未运行 Vitest。 + +## 内存判定 + +- `memory-distribution` 只以 `/proc/meminfo` 的 `MemAvailable` 判定，swap 单独披露。 +- 清理后 `MemAvailable` 约 2.8 GiB，距离 4 GiB 仍差约 1.25 GiB；WebProbe 不具备启动资格，未启动 Chromium。 +- 剩余压力主要来自 k3s 常驻 PSS、Kafka/Tempo/Tekton 控制面和 Decision Center zombie；不得继续扩大为 raw delete、k3s 重启或无证据 reclaim。 diff --git a/docs/MDTODO/web-observe-runtime-reliability.md b/docs/MDTODO/web-observe-runtime-reliability.md index 2c06329a..5f252bbb 100644 --- a/docs/MDTODO/web-observe-runtime-reliability.md +++ b/docs/MDTODO/web-observe-runtime-reliability.md @@ -16,6 +16,9 @@ 实现长期修复：https://github.com/pikasTech/unidesk/issues/1842；补齐所有调用方终态 stop，runner 从 owning YAML 获取最大运行时长/max samples/artifact cap 并自停，GC snapshot/plan/status 披露 active/stale、Chrome RSS、last cleanup 与 stop failure，完成任务后将详细报告写入[任务报告](./details/web-observe-runtime-reliability/R1.2_Task_Report.md)。 +#### 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.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 fc543f1e..b8d30b76 100644 --- a/docs/reference/gc.md +++ b/docs/reference/gc.md @@ -30,6 +30,14 @@ - 只读读取 `/proc/meminfo` 的 `MemAvailable`； - 不执行 cluster preflight、候选扫描或 mutation； - 固定披露 `metric=MemAvailable`、`source=/proc/meminfo`、`swapExcluded=true` 和可用字节数。 +- `gc remote memory-distribution`： + - 只读输出物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布； + - 默认返回紧凑 overview，`--full` 才展开完整明细； + - swap 单独披露，不计入 WebProbe 启动资格。 +- `gc remote retention plan|run --confirm|status --job-id `： + - 按目标节点 `kubernetesObjectRetention` YAML 规划和异步回收 Kubernetes 历史对象； + - `plan` 与手动、自动 `run` 必须复用同一个 fail-closed 筛选器； + - 表格形状、时间或分组无法识别时保护对象并停止删除。 - `--include-tool-caches`：本机和远端都必须是显式 opt-in，只清理固定 allowlist 的可重建 npm/npx/Bun 缓存；不得扩大成清理 `~/.npm`、`~/.bun`、`node_modules`、auth/config 或运行面依赖。 所有成功和失败输出都必须是 JSON。`plan` 必须标记 `dryRun=true`、`mutation=false`；`run` 必须要求 `--confirm` 并报告 `diskBefore`、`diskAfter`、`summary`、`results` 和 `protected`。远端 GC 可用 `--target-use-percent N` 显式表达目标根盘水位；`summary.target` 必须给出目标所需释放量、候选估算、预计水位、缺口和 `safeStop` 决策，避免靠人工心算判断是否应该继续扩大清理范围。 @@ -104,6 +112,25 @@ PK01 是腾讯云 Docker provider，不是 G14 k3s/registry 节点；长期运 PK01 pikanode temp retention 只允许清理 `/home/ubuntu/pikanode/html/temp` 下超过保留窗口的直接子目录，并必须保护 `html/download/`、`html/upload/`、`files/`、证书、Git state、直接日志文件和近期 temp workspace。该策略已固化为 PK01 节点本地 systemd timer 与 logrotate；人工排障时优先查看 `systemctl status unidesk-pk01-pikanode-temp-gc.timer` 和 `/var/log/unidesk-pk01/pikanode-temp-gc.log`。如果 PK01 高水位仍无法通过 temp retention 和通用低风险 GC 降下来，必须停止并进入 pikanode 下载产物留存、Docker image retention 或容量扩容决策，不能把 `download/`、`files/` 或 Docker overlay 当作普通临时目录删除。 +## NC01 远端策略 + +- NC01 的 `devops-infra` Tekton 历史对象留存由以下 YAML 唯一拥有： + - 手动筛选策略位于 `config/unidesk-cli.yaml#gc.remote.targets.NC01.kubernetesObjectRetention`； + - 自动周期、单批上限和启用阶段位于 `config/unidesk-cli.yaml#gc.remote.targets.NC01.policyTimer`。 +- 允许回收的对象必须同时满足： + - 资源类型是 YAML 指定的 PipelineRun； + - `Succeeded` 为 `True` 或 `False`； + - 对象年龄不小于 YAML 保留窗口； + - 分组标签完整，并且不属于每组最新保留数量。 +- 以下情况固定保护： + - 活跃、排队或状态未知； + - 过新、时间不可解析或分组未知； + - API 扫描截断、Table 形状漂移或命令超时。 +- 执行只删除 PipelineRun，并使用 YAML 指定的 ownerRef 级联策略： + - 不直接删除 TaskRun、Pod、Secret 或 Event； + - 手动入口提交异步 job 后必须用 `status --job-id` 查询终态； + - 自动 systemd timer 复用同一筛选器，并把每轮结果写入 YAML 指定的状态文件。 + ## JD01 远端策略 JD01 是 YAML-first k3s provider，承载 AgentRun、HWLAB v0.3、Web probe sentinel 和相关 PVC/state artifact。`gc remote JD01 ...` 的目标节点、lane、namespace 集合、state root、保护路径、扫描预算、artifact cap、retention 窗口、observer stop 策略和输出 limit 必须来自 `config/unidesk-cli.yaml#gc.remote.targets.JD01` 或等价 source of truth；CLI 参数只作为一次性覆盖，不得把 JD01 namespace、路径或阈值写成脚本隐藏默认。 diff --git a/scripts/native/deploy/render-unidesk-host-k8s.mjs b/scripts/native/deploy/render-unidesk-host-k8s.mjs index 3ff7b8a7..ef002c42 100755 --- a/scripts/native/deploy/render-unidesk-host-k8s.mjs +++ b/scripts/native/deploy/render-unidesk-host-k8s.mjs @@ -206,15 +206,22 @@ function readServiceStorage(service, servicePath) { if (storage === null) return null; const primary = required(storage.primary, `${servicePath}.storage.primary`); if (primary !== "postgres" && primary !== "github-repo") throw new Error(`${servicePath}.storage.primary must be postgres or github-repo`); + const healthRefreshSeconds = storage.healthRefreshSeconds === undefined + ? null + : positiveInteger(storage.healthRefreshSeconds, `${servicePath}.storage.healthRefreshSeconds`); + const resolved = { + primary, + ...(healthRefreshSeconds === null ? {} : { healthRefreshSeconds }), + }; const github = primary === "github-repo" ? record(storage.github, `${servicePath}.storage.github`) : null; - if (github === null) return { primary }; + if (github === null) return resolved; const ssh = record(github.ssh, `${servicePath}.storage.github.ssh`); const privateKey = record(ssh.privateKey, `${servicePath}.storage.github.ssh.privateKey`); const knownHosts = record(ssh.knownHosts, `${servicePath}.storage.github.ssh.knownHosts`); const privateKeySource = readFileSource(record(privateKey.sourceRef, `${servicePath}.storage.github.ssh.privateKey.sourceRef`), `${servicePath}.storage.github.ssh.privateKey.sourceRef`); const knownHostsSource = readFileSource(record(knownHosts.sourceRef, `${servicePath}.storage.github.ssh.knownHosts.sourceRef`), `${servicePath}.storage.github.ssh.knownHosts.sourceRef`); return { - primary, + ...resolved, github: { repo: required(github.repo, `${servicePath}.storage.github.repo`), sshUrl: required(github.sshUrl, `${servicePath}.storage.github.sshUrl`), @@ -629,6 +636,12 @@ function managedStorageEnv(envPrefix, storage) { `- name: ${envPrefix}_STORAGE_PRIMARY`, ` value: ${yamlScalar(storage?.primary ?? "postgres")}`, ]; + if (storage?.healthRefreshSeconds !== undefined) { + lines.push( + `- name: ${envPrefix}_STORAGE_HEALTH_REFRESH_MS`, + ` value: ${yamlScalar(positiveInteger(storage.healthRefreshSeconds, "storage.healthRefreshSeconds") * 1000)}`, + ); + } if (storage?.github === undefined) return lines.join("\n"); lines.push( `- name: ${envPrefix}_GIT_REPO`, diff --git a/scripts/src/gc-remote-kubernetes-retention.py b/scripts/src/gc-remote-kubernetes-retention.py new file mode 100644 index 00000000..7f34b08c --- /dev/null +++ b/scripts/src/gc-remote-kubernetes-retention.py @@ -0,0 +1,490 @@ +import hashlib +import json +import re +import subprocess +import time +import urllib.parse + + +def kubernetes_retention_config(config): + if not isinstance(config, dict): + raise RuntimeError("required YAML kubernetesObjectRetention policy is missing") + if not isinstance(config.get("enabled"), bool): + raise RuntimeError("kubernetesObjectRetention.enabled must be a YAML boolean") + required_strings = [ + "configId", + "kubeconfigPath", + "namespace", + "resource", + "listMode", + "countPath", + "terminalConditionType", + "propagationPolicy", + ] + for key in required_strings: + value = config.get(key) + if not isinstance(value, str) or not value: + raise RuntimeError("kubernetesObjectRetention.%s must be a non-empty YAML string" % key) + if not config.get("kubeconfigPath").startswith("/"): + raise RuntimeError("kubernetesObjectRetention.kubeconfigPath must be absolute") + if not re.match(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$", config.get("namespace")): + raise RuntimeError("kubernetesObjectRetention.namespace is invalid") + if config.get("listMode") != "server-table": + raise RuntimeError("kubernetesObjectRetention.listMode must be server-table") + if "{namespace}" not in config.get("countPath") or not config.get("countPath").startswith("/apis/"): + raise RuntimeError("kubernetesObjectRetention.countPath must be a namespaced API path template") + if config.get("propagationPolicy") not in ["Background", "Foreground", "Orphan"]: + raise RuntimeError("kubernetesObjectRetention.propagationPolicy is invalid") + positive_integers = [ + "minAgeHours", + "keepLatestPerGroup", + "manualMaxDeletePerRun", + "automaticMaxDeletePerRun", + "previewLimit", + "scanLimit", + "requestTimeoutSeconds", + "deleteBatchSize", + "deleteTimeoutSeconds", + ] + for key in positive_integers: + value = config.get(key) + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise RuntimeError("kubernetesObjectRetention.%s must be a positive YAML integer" % key) + settle_seconds = config.get("postDeleteSettleSeconds") + if isinstance(settle_seconds, bool) or not isinstance(settle_seconds, int) or settle_seconds < 0: + raise RuntimeError("kubernetesObjectRetention.postDeleteSettleSeconds must be a non-negative YAML integer") + terminal_statuses = config.get("terminalStatuses") + if not isinstance(terminal_statuses, list) or not terminal_statuses or any( + not isinstance(item, str) or not item for item in terminal_statuses + ): + raise RuntimeError("kubernetesObjectRetention.terminalStatuses must be a non-empty YAML string list") + selectors = config.get("groupSelectors") + if not isinstance(selectors, list) or not selectors: + raise RuntimeError("kubernetesObjectRetention.groupSelectors must be a non-empty YAML list") + selector_ids = set() + for selector in selectors: + if not isinstance(selector, dict): + raise RuntimeError("kubernetesObjectRetention.groupSelectors items must be YAML objects") + selector_id = selector.get("id") + labels = selector.get("groupByLabels") + if not isinstance(selector_id, str) or not re.match(r"^[a-z0-9][a-z0-9-]{0,63}$", selector_id): + raise RuntimeError("kubernetesObjectRetention group selector id is invalid") + if selector_id in selector_ids: + raise RuntimeError("kubernetesObjectRetention group selector ids must be unique") + selector_ids.add(selector_id) + if not isinstance(labels, list) or not labels or any( + not isinstance(item, str) or not item or len(item) > 253 for item in labels + ): + raise RuntimeError("kubernetesObjectRetention groupByLabels must be a non-empty YAML string list") + return config + + +def kubernetes_retention_command(config, args, timeout_key): + started = time.time() + timeout = int(config.get(timeout_key)) + try: + process = subprocess.run( + args, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout, + ) + return { + "exitCode": process.returncode, + "stdout": process.stdout, + "stderr": process.stderr, + "timedOut": False, + "elapsedMs": int((time.time() - started) * 1000), + } + except subprocess.TimeoutExpired as exc: + stdout = exc.stdout.decode("utf-8", errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "") + stderr = exc.stderr.decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "") + return { + "exitCode": 124, + "stdout": stdout, + "stderr": stderr, + "timedOut": True, + "elapsedMs": int((time.time() - started) * 1000), + } + + +def kubernetes_retention_command_summary(result): + return { + "exitCode": result.get("exitCode"), + "timedOut": result.get("timedOut") is True, + "elapsedMs": result.get("elapsedMs"), + "stdoutTail": str(result.get("stdout") or "")[-800:], + "stderrTail": str(result.get("stderr") or "")[-800:], + } + + +def kubernetes_retention_duration_hours(value): + if not isinstance(value, str) or not value or value in ["", ""]: + return None + matches = list(re.finditer(r"([0-9]+)([ysmhd])", value)) + if not matches or "".join(match.group(0) for match in matches) != value: + return None + multipliers = {"y": 8760.0, "s": 1.0 / 3600.0, "m": 1.0 / 60.0, "h": 1.0, "d": 24.0} + return sum(int(match.group(1)) * multipliers[match.group(2)] for match in matches) + + +def kubernetes_retention_group(labels, selectors): + for selector in selectors: + keys = selector.get("groupByLabels") or [] + values = [labels.get(key) for key in keys] + if all(isinstance(value, str) and value for value in values): + return { + "selectorId": selector.get("id"), + "key": "%s:%s" % (selector.get("id"), "|".join(values)), + "labels": {key: labels.get(key) for key in keys}, + } + return None + + +def kubernetes_retention_row(item, config, now_timestamp): + labels = item.get("labels") or {} + terminal_status = str(item.get("terminalStatus") or "") + terminal = terminal_status in set(config.get("terminalStatuses") or []) + group = kubernetes_retention_group(labels, config.get("groupSelectors") or []) + age_hours = kubernetes_retention_duration_hours(item.get("age")) + return { + "name": item.get("name"), + "createdAt": None, + "createdTimestamp": now_timestamp - age_hours * 3600.0 if age_hours is not None else None, + "age": item.get("age"), + "ageHours": round(age_hours, 2) if age_hours is not None else None, + "terminal": terminal, + "terminalStatus": terminal_status or None, + "terminalReason": item.get("terminalReason"), + "group": group, + } + + +def kubernetes_retention_list(config): + scan_limit = int(config.get("scanLimit")) + result = kubernetes_retention_command(config, [ + "kubectl", + "--kubeconfig", config.get("kubeconfigPath"), + "--namespace", config.get("namespace"), + "get", config.get("resource"), + "--no-headers=true", + "--show-labels=true", + ], "requestTimeoutSeconds") + command_summary = kubernetes_retention_command_summary(result) + if result.get("exitCode") != 0: + return { + "ok": False, + "error": "kubernetes-retention-list-failed", + "items": [], + "scan": {"scannedCount": 0, "limit": scan_limit, "truncated": True, "commands": [command_summary]}, + } + items = [] + invalid_rows = [] + sparse_non_terminal_rows = 0 + terminal_statuses = set(config.get("terminalStatuses") or []) + for line_number, line in enumerate((result.get("stdout") or "").splitlines(), start=1): + if not line.strip(): + continue + parts = line.split() + terminal = len(parts) >= 2 and parts[1] in terminal_statuses + valid_column_count = len(parts) == 6 if terminal else 4 <= len(parts) <= 6 + if not valid_column_count: + invalid_rows.append({ + "line": line_number, + "columnCount": len(parts), + "rowPreview": line[:600], + }) + continue + if not terminal and len(parts) < 6: + sparse_non_terminal_rows += 1 + labels = {} + labels_field = parts[-1] + if labels_field not in ["", ""]: + for label in labels_field.split(","): + key, separator, value = label.partition("=") + if separator and key and value: + labels[key] = value + if not labels: + invalid_rows.append({ + "line": line_number, + "columnCount": len(parts), + "rowPreview": line[:600], + "reason": "labels-column-invalid", + }) + continue + age = parts[3] if len(parts) >= 5 else None + completion_age = parts[4] if len(parts) == 6 else None + items.append({ + "name": parts[0], + "terminalStatus": parts[1], + "terminalReason": parts[2], + "age": age, + "completionAge": completion_age, + "labels": labels, + }) + truncated = len(items) > scan_limit + if truncated: + items = items[:scan_limit] + if invalid_rows: + return { + "ok": False, + "error": "kubernetes-retention-table-shape-invalid", + "items": [], + "scan": { + "scannedCount": len(items), + "limit": scan_limit, + "truncated": True, + "invalidRows": invalid_rows[:10], + "sparseNonTerminalRowCount": sparse_non_terminal_rows, + "commands": [command_summary], + }, + } + return { + "ok": not truncated, + "error": "kubernetes-retention-scan-limit-reached" if truncated else None, + "items": items, + "scan": { + "scannedCount": len(items), + "totalHint": len(items), + "limit": scan_limit, + "mode": config.get("listMode"), + "truncated": truncated, + "sparseNonTerminalRowCount": sparse_non_terminal_rows, + "commands": [command_summary], + }, + } + + +def kubernetes_retention_plan(config, mode, limit_override=None, full=False): + config = kubernetes_retention_config(config) + if not config.get("enabled"): + return { + "ok": False, + "error": "kubernetes-object-retention-disabled", + "configId": config.get("configId"), + "mutation": False, + } + listed = kubernetes_retention_list(config) + if not listed.get("ok"): + return { + "ok": False, + "error": listed.get("error"), + "message": listed.get("message"), + "configId": config.get("configId"), + "namespace": config.get("namespace"), + "mutation": False, + "scan": listed.get("scan"), + } + now_timestamp = time.time() + rows = [kubernetes_retention_row(item, config, now_timestamp) for item in listed.get("items") or []] + rows = [item for item in rows if isinstance(item.get("name"), str) and item.get("name")] + groups = {} + for row in rows: + group = row.get("group") + if group is not None: + groups.setdefault(group.get("key"), []).append(row) + protected_latest = set() + for group_rows in groups.values(): + terminal_rows = [item for item in group_rows if item.get("terminal")] + terminal_rows.sort(key=lambda item: item.get("createdTimestamp") or 0, reverse=True) + for item in terminal_rows[:int(config.get("keepLatestPerGroup"))]: + protected_latest.add(item.get("name")) + eligible = [] + protected = { + "active": 0, + "young": 0, + "latestPerGroup": 0, + "unknownGroup": 0, + "invalidTimestamp": 0, + } + for row in rows: + if not row.get("terminal"): + protected["active"] += 1 + elif row.get("group") is None: + protected["unknownGroup"] += 1 + elif row.get("createdTimestamp") is None or row.get("ageHours") is None: + protected["invalidTimestamp"] += 1 + elif row.get("ageHours") < int(config.get("minAgeHours")): + protected["young"] += 1 + elif row.get("name") in protected_latest: + protected["latestPerGroup"] += 1 + else: + eligible.append(row) + eligible.sort(key=lambda item: item.get("createdTimestamp") or 0) + configured_limit = int(config.get("automaticMaxDeletePerRun" if mode == "automatic" else "manualMaxDeletePerRun")) + effective_limit = configured_limit + if limit_override is not None: + effective_limit = min(configured_limit, max(1, int(limit_override))) + selected = eligible[:effective_limit] + selected_names = [item.get("name") for item in selected] + fingerprint = hashlib.sha256("\n".join(selected_names).encode("utf-8")).hexdigest() + group_summary = [] + for group_key, group_rows in groups.items(): + group_selected = [item for item in selected if (item.get("group") or {}).get("key") == group_key] + group_summary.append({ + "groupKey": group_key, + "totalCount": len(group_rows), + "terminalCount": len([item for item in group_rows if item.get("terminal")]), + "selectedCount": len(group_selected), + }) + group_summary.sort(key=lambda item: item.get("totalCount"), reverse=True) + preview_limit = int(config.get("previewLimit")) + preview = selected if full else selected[:preview_limit] + candidates = [ + { + "name": item.get("name"), + "createdAt": item.get("createdAt"), + "age": item.get("age"), + "ageHours": item.get("ageHours"), + "terminalStatus": item.get("terminalStatus"), + "terminalReason": item.get("terminalReason"), + "group": item.get("group"), + } + for item in preview + ] + return { + "ok": True, + "configId": config.get("configId"), + "mode": mode, + "namespace": config.get("namespace"), + "resource": config.get("resource"), + "dryRun": True, + "mutation": False, + "criteria": { + "terminalConditionType": config.get("terminalConditionType"), + "terminalStatuses": config.get("terminalStatuses"), + "minAgeHours": config.get("minAgeHours"), + "keepLatestPerGroup": config.get("keepLatestPerGroup"), + "effectiveDeleteLimit": effective_limit, + "propagationPolicy": config.get("propagationPolicy"), + }, + "scan": listed.get("scan"), + "summary": { + "objectCount": len(rows), + "terminalCount": len([item for item in rows if item.get("terminal")]), + "eligibleCount": len(eligible), + "selectedCount": len(selected), + "omittedEligibleCount": max(0, len(eligible) - len(selected)), + "protected": protected, + "groupCount": len(groups), + }, + "groups": group_summary if full else group_summary[:preview_limit], + "candidates": candidates, + "returnedCandidateCount": len(candidates), + "omittedCandidateCount": max(0, len(selected) - len(candidates)), + "selectedNames": selected_names, + "planFingerprint": fingerprint, + } + + +def kubernetes_retention_count(config): + base_path = config.get("countPath").replace( + "{namespace}", urllib.parse.quote(config.get("namespace"), safe="") + ) + separator = "&" if "?" in base_path else "?" + result = kubernetes_retention_command(config, [ + "kubectl", "--kubeconfig", config.get("kubeconfigPath"), "get", "--raw", "%s%slimit=1" % (base_path, separator), + ], "requestTimeoutSeconds") + command_summary = kubernetes_retention_command_summary(result) + command_summary.pop("stdoutTail", None) + if result.get("exitCode") != 0: + return {"ok": False, "command": command_summary} + try: + payload = json.loads(result.get("stdout") or "{}") + metadata = payload.get("metadata") or {} + return { + "ok": True, + "count": len(payload.get("items") or []) + int(metadata.get("remainingItemCount") or 0), + "command": command_summary, + } + except (TypeError, ValueError, json.JSONDecodeError) as exc: + return {"ok": False, "error": str(exc), "command": command_summary} + + +def kubernetes_retention_execute(config, mode, limit_override=None, progress=None, full=False): + config = kubernetes_retention_config(config) + before = kubernetes_retention_count(config) + plan = kubernetes_retention_plan(config, mode, limit_override=limit_override, full=True) + if not plan.get("ok"): + return {"ok": False, "plan": plan, "before": before, "mutation": False, "deletedCount": 0} + selected_names = list(plan.get("selectedNames") or []) + if not selected_names: + return {"ok": True, "plan": plan, "before": before, "after": before, "mutation": False, "deletedCount": 0, "batches": []} + batch_size = int(config.get("deleteBatchSize")) + batches = [] + deleted_count = 0 + for offset in range(0, len(selected_names), batch_size): + names = selected_names[offset:offset + batch_size] + result = kubernetes_retention_command(config, [ + "kubectl", + "--kubeconfig", config.get("kubeconfigPath"), + "--namespace", config.get("namespace"), + "delete", config.get("resource"), + "--ignore-not-found=true", + "--wait=false", + "--cascade=%s" % config.get("propagationPolicy").lower(), + ] + names, "deleteTimeoutSeconds") + ok = result.get("exitCode") == 0 + if ok: + deleted_count += len(names) + batches.append({ + "index": len(batches), + "requestedCount": len(names), + "ok": ok, + "command": kubernetes_retention_command_summary(result), + }) + if progress is not None: + progress(deleted_count, len(selected_names)) + if not ok: + break + settle_seconds = int(config.get("postDeleteSettleSeconds")) + if settle_seconds > 0: + time.sleep(settle_seconds) + after = kubernetes_retention_count(config) + ok = all(item.get("ok") for item in batches) + compact_plan = dict(plan) + if not full: + compact_plan.pop("selectedNames", None) + compact_plan["candidates"] = compact_plan.get("candidates", [])[:int(config.get("previewLimit"))] + return { + "ok": ok, + "configId": config.get("configId"), + "mode": mode, + "namespace": config.get("namespace"), + "resource": config.get("resource"), + "mutation": deleted_count > 0, + "requestedDeleteCount": len(selected_names), + "deletedCount": deleted_count, + "before": before, + "after": after, + "objectCountDelta": ( + int(before.get("count")) - int(after.get("count")) + if before.get("ok") and after.get("ok") else None + ), + "plan": compact_plan, + "batches": batches, + } + + +def run_kubernetes_object_retention_policy(config): + retention = config.get("kubernetesObjectRetention") if isinstance(config.get("kubernetesObjectRetention"), dict) else {} + if not retention.get("enabled"): + return {"id": "kubernetes-object-retention", "ok": True, "skipped": True} + result = kubernetes_retention_execute(retention, "automatic", full=False) + return { + "id": "kubernetes-object-retention", + "ok": result.get("ok") is True, + "skipped": result.get("mutation") is not True, + "configId": result.get("configId"), + "namespace": result.get("namespace"), + "resource": result.get("resource"), + "candidateCount": ((result.get("plan") or {}).get("summary") or {}).get("eligibleCount"), + "requestedDeleteCount": result.get("requestedDeleteCount"), + "deletedCount": result.get("deletedCount"), + "objectCountDelta": result.get("objectCountDelta"), + "before": result.get("before"), + "after": result.get("after"), + "batches": result.get("batches"), + "error": ((result.get("plan") or {}).get("error")) if not result.get("ok") else None, + } diff --git a/scripts/src/gc-remote-memory-distribution.py b/scripts/src/gc-remote-memory-distribution.py new file mode 100644 index 00000000..51e4777b --- /dev/null +++ b/scripts/src/gc-remote-memory-distribution.py @@ -0,0 +1,474 @@ +def memory_distribution_config(): + config = MEMORY_CONFIG.get("attribution") + if not isinstance(config, dict): + raise RuntimeError("required YAML memory attribution policy is missing") + required_integers = { + "cgroupScanLimit": (1, 1000000), + "topLevelLimit": (1, 100), + "systemServiceLimit": (1, 100), + "podLimit": (1, 100), + "processLimit": (1, 100), + "processPssLimit": (1, 50), + "commandTimeoutSeconds": (1, 60), + "targetMemAvailableBytes": (1, 1 << 50), + "namespaceCountConcurrency": (1, 32), + } + for key, (minimum, maximum) in required_integers.items(): + value = config.get(key) + if isinstance(value, bool) or not isinstance(value, int) or value < minimum or value > maximum: + raise RuntimeError("invalid YAML memory attribution integer: %s" % key) + cgroup_root = config.get("cgroupRoot") + kubeconfig = config.get("kubeconfigPath") + resources = config.get("objectCountResources") + namespaces = config.get("namespaceCountNamespaces") + if not isinstance(cgroup_root, str) or not cgroup_root.startswith("/") or os.path.normpath(cgroup_root) != cgroup_root: + raise RuntimeError("invalid YAML memory attribution cgroupRoot") + if not isinstance(kubeconfig, str) or not kubeconfig.startswith("/") or os.path.normpath(kubeconfig) != kubeconfig: + raise RuntimeError("invalid YAML memory attribution kubeconfigPath") + if not isinstance(resources, list) or not resources or any( + not isinstance(item, str) or not re.match(r"^[a-z0-9.-]{1,100}$", item) for item in resources + ): + raise RuntimeError("invalid YAML memory attribution objectCountResources") + if not isinstance(namespaces, list) or not namespaces or any( + not isinstance(item, str) or not re.match(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$", item) for item in namespaces + ): + raise RuntimeError("invalid YAML memory attribution namespaceCountNamespaces") + return config + +def read_memory_distribution_key_values(path, multiplier=1): + result = {} + try: + with open(path, "r", encoding="utf-8") as handle: + for line in handle: + parts = line.replace(":", "", 1).split() + if len(parts) < 2: + continue + try: + result[parts[0]] = int(parts[1]) * multiplier + except ValueError: + continue + except OSError: + return {} + return result + +def read_memory_distribution_integer(path): + try: + with open(path, "r", encoding="utf-8") as handle: + return int(handle.read().strip()) + except (OSError, ValueError): + return None + +def memory_distribution_cgroup_row(root, path): + stat = read_memory_distribution_key_values(os.path.join(path, "memory.stat")) + file_bytes = safe_int(stat.get("file")) + slab_reclaimable = safe_int(stat.get("slab_reclaimable")) + return { + "path": os.path.relpath(path, root), + "currentBytes": read_memory_distribution_integer(os.path.join(path, "memory.current")), + "swapCurrentBytes": read_memory_distribution_integer(os.path.join(path, "memory.swap.current")), + "anonBytes": stat.get("anon"), + "fileBytes": stat.get("file"), + "inactiveAnonBytes": stat.get("inactive_anon"), + "inactiveFileBytes": stat.get("inactive_file"), + "reclaimableFileSlabUpperBoundBytes": file_bytes + slab_reclaimable, + } + +def memory_distribution_limit(config, key): + configured = safe_int(config.get(key)) + return min(100, configured * 5) if bool(OPTIONS.get("full")) else configured + +def memory_distribution_top(rows, limit): + return sorted(rows, key=lambda item: safe_int(item.get("currentBytes")), reverse=True)[:limit] + +def memory_distribution_command_summary(result): + return { + "ok": result.get("exitCode") == 0, + "exitCode": result.get("exitCode"), + "timedOut": result.get("timedOut") is True, + "durationMs": result.get("durationMs"), + } + +def memory_distribution_runtime_maps(config): + timeout = safe_int(config.get("commandTimeoutSeconds")) + docker = command(["docker", "ps", "--no-trunc", "--format", "{{.ID}}\t{{.Names}}"], timeout) + docker_names = {} + if docker.get("exitCode") == 0: + for line in docker.get("stdout", "").splitlines(): + parts = line.split("\t", 1) + if len(parts) == 2 and re.match(r"^[a-f0-9]{64}$", parts[0]): + docker_names[parts[0]] = parts[1] + cri = command(["k3s", "crictl", "pods", "-o", "json"], timeout) + pod_names = {} + if cri.get("exitCode") == 0: + try: + payload = json.loads(cri.get("stdout") or "{}") + for item in payload.get("items") or []: + metadata = item.get("metadata") or {} + uid = str(metadata.get("uid") or "") + if uid: + pod_names[uid] = { + "namespace": metadata.get("namespace"), + "name": metadata.get("name"), + "state": item.get("state"), + } + except (TypeError, ValueError, json.JSONDecodeError): + pass + return docker_names, pod_names, { + "docker": memory_distribution_command_summary(docker), + "cri": memory_distribution_command_summary(cri), + } + +def collect_memory_distribution_cgroups(config, docker_names, pod_names): + root = config.get("cgroupRoot") + scan_limit = safe_int(config.get("cgroupScanLimit")) + top_level = [] + system_services = [] + pods = [] + scanned = 0 + truncated = False + pod_pattern = re.compile(r"pod([0-9a-f_]{36})\.slice$") + docker_pattern = re.compile(r"docker-([a-f0-9]{64})\.scope$") + for current, directories, files in os.walk(root, 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: + truncated = True + break + if "memory.current" not in files: + continue + row = memory_distribution_cgroup_row(root, current) + parent = os.path.dirname(current) + if parent == root: + top_level.append(row) + if parent == os.path.join(root, "system.slice"): + match = docker_pattern.search(os.path.basename(current)) + if match: + row["owner"] = {"kind": "docker", "name": docker_names.get(match.group(1)), "id": match.group(1)} + system_services.append(row) + match = pod_pattern.search(os.path.basename(current)) + if match: + uid = match.group(1).replace("_", "-") + row["owner"] = {"kind": "kubernetes-pod", "uid": uid, **(pod_names.get(uid) or {})} + pods.append(row) + directories[:] = [] + return { + "scan": {"root": root, "scannedCount": scanned, "limit": scan_limit, "truncated": truncated}, + "topLevel": memory_distribution_top(top_level, memory_distribution_limit(config, "topLevelLimit")), + "systemServices": memory_distribution_top(system_services, memory_distribution_limit(config, "systemServiceLimit")), + "pods": memory_distribution_top(pods, memory_distribution_limit(config, "podLimit")), + "podCgroupCount": len(pods), + } + +def memory_distribution_cgroup_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 memory_distribution_pss(pid): + values = read_memory_distribution_key_values("/proc/%s/smaps_rollup" % int(pid), 1024) + if not values: + return None + return { + "pssBytes": values.get("Pss"), + "pssAnonBytes": values.get("Pss_Anon"), + "pssFileBytes": values.get("Pss_File"), + "swapPssBytes": values.get("SwapPss"), + } + +def collect_memory_distribution_processes(config): + 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": {}} + rows = [] + zombie_by_comm = {} + zombie_by_parent = {} + for line in result.get("stdout", "").splitlines(): + parts = line.strip().split(None, 5) + if len(parts) < 5: + continue + pid = safe_int(parts[0]) + ppid = safe_int(parts[1]) + rss = safe_int(parts[2]) * 1024 + 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]} + 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")] + pss_limit = memory_distribution_limit(config, "processPssLimit") + for index, item in enumerate(top): + item["cgroupPath"] = memory_distribution_cgroup_for_pid(item.get("pid")) + if index < pss_limit: + item.update(memory_distribution_pss(item.get("pid")) or {}) + 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 {} + parent_rows.append({ + "ppid": ppid, + "parentComm": parent.get("comm"), + "zombieCount": count, + "cgroupPath": memory_distribution_cgroup_for_pid(ppid), + }) + return { + "ok": True, + "command": memory_distribution_command_summary(result), + "processCount": len(rows), + "rssTotalBytes": sum(safe_int(item.get("rssBytes")) for item in rows), + "top": top, + "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]], + "byParent": parent_rows, + }, + } + +def parse_memory_distribution_pressure(path): + result = {} + try: + with open(path, "r", encoding="utf-8") as handle: + for line in handle: + parts = line.split() + if not parts: + continue + result[parts[0]] = { + key: float(value) if key != "total" else safe_int(value) + for key, value in (item.split("=", 1) for item in parts[1:] if "=" in item) + } + except OSError: + pass + return result + +def collect_memory_distribution_metrics(config): + timeout = safe_int(config.get("commandTimeoutSeconds")) + kubeconfig = config.get("kubeconfigPath") + result = command(["kubectl", "--kubeconfig", kubeconfig, "get", "--raw", "/metrics"], timeout) + selected = set(config.get("objectCountResources") or []) + objects = {} + runtime = {} + if result.get("exitCode") == 0: + object_pattern = re.compile(r'^apiserver_storage_objects\{resource="([^"]+)"\}\s+([0-9.eE+-]+)$') + runtime_names = set([ + "process_resident_memory_bytes", "process_virtual_memory_bytes", "go_goroutines", + "go_memstats_heap_alloc_bytes", "go_memstats_heap_inuse_bytes", "go_memstats_heap_sys_bytes", + ]) + for line in result.get("stdout", "").splitlines(): + match = object_pattern.match(line) + if match and match.group(1) in selected: + objects[match.group(1)] = int(float(match.group(2))) + continue + parts = line.split() + if len(parts) == 2 and parts[0] in runtime_names: + try: + runtime[parts[0]] = float(parts[1]) + except ValueError: + pass + return {"command": memory_distribution_command_summary(result), "objects": objects, "runtime": runtime} + +def memory_distribution_namespace_resource_path(resource, namespace): + encoded_namespace = urllib.parse.quote(namespace, safe="") + paths = { + "pods": "/api/v1/namespaces/%s/pods?limit=1" % encoded_namespace, + "events": "/api/v1/namespaces/%s/events?limit=1" % encoded_namespace, + "secrets": "/api/v1/namespaces/%s/secrets?limit=1" % encoded_namespace, + "persistentvolumeclaims": "/api/v1/namespaces/%s/persistentvolumeclaims?limit=1" % encoded_namespace, + "pipelineruns.tekton.dev": "/apis/tekton.dev/v1/namespaces/%s/pipelineruns?limit=1" % encoded_namespace, + "taskruns.tekton.dev": "/apis/tekton.dev/v1/namespaces/%s/taskruns?limit=1" % encoded_namespace, + } + return paths.get(resource) + +def memory_distribution_namespace_count(config, namespace, resource): + path = memory_distribution_namespace_resource_path(resource, namespace) + if path is None: + return namespace, resource, None, "unsupported-resource" + result = command([ + "kubectl", "--kubeconfig", config.get("kubeconfigPath"), "get", "--raw", path, + ], safe_int(config.get("commandTimeoutSeconds"))) + if result.get("exitCode") != 0: + return namespace, resource, None, "read-failed" + try: + payload = json.loads(result.get("stdout") or "{}") + metadata = payload.get("metadata") or {} + count = len(payload.get("items") or []) + safe_int(metadata.get("remainingItemCount")) + return namespace, resource, count, None + except (TypeError, ValueError, json.JSONDecodeError): + return namespace, resource, None, "parse-failed" + +def collect_memory_distribution_namespace_counts(config): + import concurrent.futures + + namespaces = list(config.get("namespaceCountNamespaces") or []) + resources = [ + item for item in (config.get("objectCountResources") or []) + if memory_distribution_namespace_resource_path(item, namespaces[0]) is not None + ] + jobs = [(namespace, resource) for namespace in namespaces for resource in resources] + rows = {namespace: {"namespace": namespace} for namespace in namespaces} + errors = [] + with concurrent.futures.ThreadPoolExecutor(max_workers=safe_int(config.get("namespaceCountConcurrency"))) as pool: + futures = [pool.submit(memory_distribution_namespace_count, config, namespace, resource) for namespace, resource in jobs] + for future in concurrent.futures.as_completed(futures): + namespace, resource, count, error = future.result() + if error is None: + rows[namespace][resource] = count + else: + errors.append({"namespace": namespace, "resource": resource, "error": error}) + ordered = sorted( + rows.values(), + key=lambda item: sum(safe_int(value) for key, value in item.items() if key != "namespace"), + reverse=True, + ) + return {"rows": ordered, "errors": errors, "complete": not errors} + +def compact_memory_distribution_cgroup(item): + compact = { + key: item.get(key) + for key in ["path", "currentBytes", "swapCurrentBytes", "anonBytes"] + 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 + } + return compact + +def compact_memory_distribution_process(item): + return { + key: item.get(key) + for key in ["pid", "comm", "rssBytes", "pssBytes", "pssAnonBytes", "swapPssBytes", "cgroupPath"] + if item.get(key) is not None + } + +def compact_memory_distribution_pressure(pressure): + return { + level: { + key: values.get(key) + for key in ["avg10", "avg60"] + if values.get(key) is not None + } + for level, values in pressure.items() + if isinstance(values, dict) + } + +def compact_memory_distribution_zombies(zombies): + by_parent = [] + for item in zombies.get("byParent") or []: + by_parent.append({ + key: item.get(key) + for key in ["ppid", "parentComm", "zombieCount"] + if item.get(key) is not None + }) + return { + "count": zombies.get("count"), + "byComm": (zombies.get("byComm") or [])[:3], + "byParent": by_parent[:3], + } + +def collect_memory_distribution(observed_at): + try: + config = memory_distribution_config() + except RuntimeError as exc: + return {"ok": False, "action": "gc remote memory-distribution", "providerId": PROVIDER_ID, "dryRun": True, "mutation": False, "error": str(exc)} + started = time.time() + docker_names, pod_names, runtime_commands = memory_distribution_runtime_maps(config) + 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) + metrics = collect_memory_distribution_metrics(config) + namespace_counts = collect_memory_distribution_namespace_counts(config) + available = safe_int((host_snapshot.get("memory") or {}).get("availableBytes")) + target = safe_int(config.get("targetMemAvailableBytes")) + cached = safe_int(meminfo.get("Cached")) + safe_int(meminfo.get("Buffers")) + slab_reclaimable = safe_int(meminfo.get("SReclaimable")) + payload = { + "ok": cgroups.get("scan", {}).get("truncated") is not True, + "action": "gc remote memory-distribution", + "providerId": PROVIDER_ID, + "dryRun": True, + "mutation": False, + "observedAt": observed_at, + "durationMs": int((time.time() - started) * 1000), + "summary": { + "metric": "MemAvailable", + "source": "/proc/meminfo", + "swapExcludedFromAvailability": True, + "availableBytes": available, + "targetAvailableBytes": target, + "targetGapBytes": max(0, target - available + 1), + "anonPagesBytes": meminfo.get("AnonPages"), + "cachedAndBuffersBytes": cached, + "slabBytes": meminfo.get("Slab"), + "reclaimableFileSlabUpperBoundBytes": cached + slab_reclaimable, + "swapFreeBytes": safe_int((host_snapshot.get("swap") or {}).get("freeBytes")), + "memoryPsiFullAvg60": (parse_memory_distribution_pressure("/proc/pressure/memory").get("full") or {}).get("avg60"), + }, + "host": { + "snapshot": host_snapshot, + "pages": {key: meminfo.get(key) for key in [ + "MemFree", "Buffers", "Cached", "Active", "Inactive", "Active(anon)", "Inactive(anon)", + "Active(file)", "Inactive(file)", "AnonPages", "Mapped", "Shmem", "Slab", "SReclaimable", + "SUnreclaim", "KernelStack", "PageTables", "SwapCached", + ]}, + "pressure": {resource: parse_memory_distribution_pressure("/proc/pressure/%s" % resource) for resource in ["memory", "io", "cpu"]}, + }, + "cgroups": cgroups, + "processes": processes, + "kubernetes": {**metrics, "namespaceCounts": namespace_counts}, + "collection": { + "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.memoryPressure.attribution" % PROVIDER_ID, + "runtimeMaps": runtime_commands, + "podRuntimeMapCount": len(pod_names), + "dockerRuntimeMapCount": len(docker_names), + "full": bool(OPTIONS.get("full")), + }, + "policy": { + "analysisOnly": True, + "next": "Use this distribution before designing or running memory-pressure GC candidates.", + }, + } + 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"] = { + "scan": cgroups.get("scan"), + "podCgroupCount": cgroups.get("podCgroupCount"), + "topLevel": [compact_memory_distribution_cgroup(item) for item in cgroups.get("topLevel") or []], + "systemServices": [compact_memory_distribution_cgroup(item) for item in cgroups.get("systemServices") or []], + "pods": [compact_memory_distribution_cgroup(item) for item in cgroups.get("pods") or []], + } + payload["processes"] = { + "ok": processes.get("ok"), + "processCount": processes.get("processCount"), + "rssTotalBytes": processes.get("rssTotalBytes"), + "top": [compact_memory_distribution_process(item) for item in processes.get("top") or []], + "zombies": compact_memory_distribution_zombies(processes.get("zombies") or {}), + } + payload["kubernetes"] = { + "objects": metrics.get("objects"), + "namespaceCounts": namespace_counts, + "command": metrics.get("command"), + } + payload["collection"] = { + "configSource": payload.get("collection", {}).get("configSource"), + "podRuntimeMapCount": len(pod_names), + "dockerRuntimeMapCount": len(docker_names), + "runtimeMaps": runtime_commands, + "full": False, + } + return payload diff --git a/scripts/src/gc-remote-memory-pressure.test.ts b/scripts/src/gc-remote-memory-pressure.test.ts index 9dac61c7..5bf6c751 100644 --- a/scripts/src/gc-remote-memory-pressure.test.ts +++ b/scripts/src/gc-remote-memory-pressure.test.ts @@ -121,3 +121,41 @@ test("memory-pressure async runner uses a parent-child barrier and locked worker assert.match(runnerSource, /workerAlive/u); assert.match(runnerSource, /worker-identity-unavailable|worker-exited-without-terminal-state/u); }); + +test("memory-pressure plan includes only YAML allowlisted cgroup reclaim stages with swap reserve", () => { + const result = runPythonFixture(String.raw` +MEMORY_CONFIG = { + "observeStateRoots": ["/tmp/observe"], "minimumReadableObserveRoots": 1, + "observeRunDepth": 6, "observeScanLimit": 100, "staleRunMaxAgeHours": 1, + "processScanLimit": 100, "processStaleMinAgeSeconds": 3600, + "maxMemoryReclaimCandidates": 20, "terminationGraceSeconds": 1, + "terminationPollMilliseconds": 50, "observerRootPatterns": ["observer-runner.mjs"], + "browserRootPatterns": ["cliDaemon.mjs"], "browserProcessPatterns": ["chromium"], + "cgroupReclaim": { + "enabled": True, "targetMemAvailableBytes": 4294967296, + "minimumSwapFreeBytesAfterReclaim": 402653184, + "targets": [ + {"id": "file-pages", "path": "/sys/fs/cgroup/system.slice/k3s.service", "reclaimBytes": 402653184, "swappiness": 0, "priority": 10}, + {"id": "inactive-anon", "path": "/sys/fs/cgroup/system.slice/k3s.service", "reclaimBytes": 1342177280, "swappiness": 100, "priority": 20}, + ], + }, +} +read_proc_process_table = lambda: {1: {"pid": 1, "ppid": 0, "sid": 1, "startTicks": 1, "ageSeconds": 1, "rssBytes": 1, "commandLine": "init", "commandPreview": "init"}} +observer_run_records = lambda: [] +collect_memory_snapshot = lambda: {"ok": True, "memory": {"availableBytes": 3 * 1024 * 1024 * 1024}, "swap": {"freeBytes": 2 * 1024 * 1024 * 1024}} +os.path.realpath = lambda path: path +os.path.isdir = lambda path: True +os.path.islink = lambda path: False +os.path.exists = lambda path: True +os.access = lambda path, mode: True +read_cgroup_integer = lambda path, name: 6 * 1024 * 1024 * 1024 +read_cgroup_memory_stat = lambda path: {"inactive_anon": 2 * 1024 * 1024 * 1024} +candidates = collect_memory_reclaim_candidates() +print(json.dumps({"ids": [item["id"] for item in candidates], "swappiness": [item["swappiness"] for item in candidates], "diagnostics": memory_reclaim_scan_diagnostics()})) +`); + assert.deepEqual(result.ids, ["cgroup-memory-reclaim:file-pages", "cgroup-memory-reclaim:inactive-anon"]); + assert.deepEqual(result.swappiness, [0, 100]); + assert.equal(((result.diagnostics as Record).cgroupReclaim as Record).eligibleCount, 2); + assert.match(webObserveSource, /minimum SwapFree reserve/u); + assert.match(webObserveSource, /memory\.reclaim/u); +}); diff --git a/scripts/src/gc-remote-policy-runner.py b/scripts/src/gc-remote-policy-runner.py index 038934fc..50c85b42 100644 --- a/scripts/src/gc-remote-policy-runner.py +++ b/scripts/src/gc-remote-policy-runner.py @@ -8,6 +8,8 @@ import sys import time from datetime import datetime, timezone +# __UNIDESK_GC_REMOTE_POLICY_KUBERNETES_RETENTION_HELPERS__ + def now_iso(): return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") @@ -135,6 +137,8 @@ def write_state(config, payload): def run_journal(config): + if config.get("includeJournal") is not True: + return {"id": "journal", "ok": True, "skipped": True} target = safe_int(config.get("journalTargetBytes"), 268435456) result = command(["journalctl", "--vacuum-size=%d" % target], 30) return {"id": "journal", "ok": result["exitCode"] == 0, "command": result} @@ -150,6 +154,8 @@ def run_apt(config): def run_tmp(config): + if config.get("includeTmp") is not True: + return {"id": "tmp-allowlist", "ok": True, "skipped": True} prefixes = list(config.get("tmpPrefixAllowlist") or []) protected = set(config.get("tmpExactProtect") or []) cutoff = time.time() - float(config.get("tmpMinAgeHours") or 24) * 3600.0 @@ -181,6 +187,8 @@ def run_tmp(config): def run_tool_caches(config): + if config.get("includeToolCaches") is not True: + return {"id": "tool-caches", "ok": True, "skipped": True} paths = list(config.get("toolCachePaths") or []) reclaimed = 0 deleted = 0 @@ -453,7 +461,7 @@ def main(): config_path = sys.argv[1] if len(sys.argv) > 1 else "" config = load_config(config_path) results = [] - for fn in [run_journal, run_apt, run_tmp, run_tool_caches, run_web_observe, run_session_pvcs, run_k3s_images, run_host_containerd, run_local_path_orphans]: + for fn in [run_journal, run_apt, run_tmp, run_tool_caches, run_web_observe, run_session_pvcs, run_k3s_images, run_host_containerd, run_local_path_orphans, run_kubernetes_object_retention_policy]: try: results.append(fn(config)) except Exception as exc: diff --git a/scripts/src/gc-remote-runner.py b/scripts/src/gc-remote-runner.py index 554966fa..2a6932fe 100644 --- a/scripts/src/gc-remote-runner.py +++ b/scripts/src/gc-remote-runner.py @@ -23,6 +23,7 @@ PVC_CONFIG = REMOTE_TARGET.get("pvcAttribution") if isinstance(REMOTE_TARGET.get CONTAINERD_CONFIG = REMOTE_TARGET.get("containerdImageCache") if isinstance(REMOTE_TARGET.get("containerdImageCache"), dict) else {} HOST_CONTAINERD_CONFIG = REMOTE_TARGET.get("hostContainerdCache") if isinstance(REMOTE_TARGET.get("hostContainerdCache"), dict) else {} LOCAL_PATH_CONFIG = REMOTE_TARGET.get("localPathStorage") if isinstance(REMOTE_TARGET.get("localPathStorage"), dict) else {} +KUBERNETES_RETENTION_CONFIG = REMOTE_TARGET.get("kubernetesObjectRetention") if isinstance(REMOTE_TARGET.get("kubernetesObjectRetention"), dict) else {} POLICY_TIMER_CONFIG = REMOTE_TARGET.get("policyTimer") if isinstance(REMOTE_TARGET.get("policyTimer"), dict) else {} HOST_DOCKER_GC_CONFIG = REMOTE_TARGET.get("hostDockerGc") if isinstance(REMOTE_TARGET.get("hostDockerGc"), dict) else {} POLICY_RUNNER_SOURCE = base64.b64decode("__UNIDESK_GC_REMOTE_POLICY_RUNNER_BASE64__").decode("utf-8") @@ -348,6 +349,67 @@ def emit_json(payload, persist_large=True): compact = compact_payload_for_stdout(payload, full_size, job_id or None, paths) print(json.dumps(compact, ensure_ascii=False, indent=2)) +def compact_kubernetes_retention_result(result): + if not isinstance(result, dict): + return result + plan = result.get("plan") if isinstance(result.get("plan"), dict) else {} + scan = plan.get("scan") if isinstance(plan.get("scan"), dict) else {} + batches = [] + for item in result.get("batches") or []: + if not isinstance(item, dict): + continue + command_payload = item.get("command") if isinstance(item.get("command"), dict) else {} + batches.append({ + "index": item.get("index"), + "requestedCount": item.get("requestedCount"), + "ok": item.get("ok") is True, + "command": { + key: command_payload.get(key) + for key in ["exitCode", "timedOut", "elapsedMs"] + if command_payload.get(key) is not None + }, + }) + compact = { + key: result.get(key) + for key in [ + "ok", + "configId", + "mode", + "namespace", + "resource", + "mutation", + "requestedDeleteCount", + "deletedCount", + "objectCountDelta", + ] + if result.get(key) is not None + } + for count_key in ["before", "after"]: + count_payload = result.get(count_key) if isinstance(result.get(count_key), dict) else {} + command_payload = count_payload.get("command") if isinstance(count_payload.get("command"), dict) else {} + compact[count_key] = { + key: count_payload.get(key) + for key in ["ok", "count", "error"] + if count_payload.get(key) is not None + } + compact[count_key]["command"] = { + key: command_payload.get(key) + for key in ["exitCode", "timedOut", "elapsedMs"] + if command_payload.get(key) is not None + } + compact["plan"] = { + "criteria": plan.get("criteria"), + "summary": plan.get("summary"), + "planFingerprint": plan.get("planFingerprint"), + "scan": { + key: scan.get(key) + for key in ["scannedCount", "totalHint", "limit", "mode", "truncated", "sparseNonTerminalRowCount"] + if scan.get(key) is not None + }, + } + compact["batches"] = batches + return compact + def remote_gc_job_status(): job_id = job_id_or_none() if not job_id: @@ -366,7 +428,8 @@ def remote_gc_job_status(): descriptor = acquire_job_lock(paths) with open(paths["state"], "r", encoding="utf-8") as handle: payload = json.load(handle) - if payload.get("scope") == "memory-pressure-only" and payload.get("status") == "running": + worker_scopes = set(["memory-pressure-only", "kubernetes-object-retention"]) + if payload.get("scope") in worker_scopes and payload.get("status") == "running": worker_pid = safe_int(payload.get("workerPid")) expected_start_ticks = safe_int(payload.get("workerStartTicks")) actual_identity = process_identity(worker_pid) if worker_pid > 0 else None @@ -374,10 +437,15 @@ def remote_gc_job_status(): worker_alive = actual_identity is not None and actual_identity.get("state") not in set(["Z", "X"]) \ and actual_start_ticks == expected_start_ticks if not worker_alive: + outcome = ( + "memory-reclaim-worker-exited-without-terminal-state" + if payload.get("scope") == "memory-pressure-only" + else "kubernetes-retention-worker-exited-without-terminal-state" + ) payload.update({ "ok": False, "status": "failed", - "outcome": "memory-reclaim-worker-exited-without-terminal-state", + "outcome": outcome, "workerAlive": False, "workerObservedState": actual_identity.get("state") if actual_identity else None, "workerObservedStartTicks": actual_start_ticks, @@ -398,6 +466,8 @@ def remote_gc_job_status(): finally: if descriptor is not None: release_job_lock(descriptor) + if payload.get("scope") == "kubernetes-object-retention" and not bool(OPTIONS.get("full")): + payload["retention"] = compact_kubernetes_retention_result(payload.get("retention")) payload["logTail"] = read_file_tail(paths["log"]) return payload @@ -666,14 +736,27 @@ def collect_memory_snapshot(): "totalHuman": fmt_bytes(total_bytes) if total_bytes is not None else None, "availableHuman": fmt_bytes(available_bytes), } + swap_total_bytes = fields.get("SwapTotal") + swap_free_bytes = fields.get("SwapFree") + swap = { + "totalBytes": swap_total_bytes, + "freeBytes": swap_free_bytes, + "totalHuman": fmt_bytes(swap_total_bytes) if swap_total_bytes is not None else None, + "freeHuman": fmt_bytes(swap_free_bytes) if swap_free_bytes is not None else None, + } return { "ok": True, "metric": "MemAvailable", "source": source, "swapExcluded": True, "memory": memory, + "swap": swap, } +# __UNIDESK_GC_REMOTE_MEMORY_DISTRIBUTION_HELPERS__ + +# __UNIDESK_GC_REMOTE_KUBERNETES_RETENTION_HELPERS__ + def observe_run_record(path, stale_hours): stat = os.stat(path) heartbeat = read_json_file(os.path.join(path, "heartbeat.json")) or {} @@ -1524,6 +1607,8 @@ def execute(candidate): return {"reclaimedBytes": before, "stale": record} if kind in set(["browser-orphan-process-tree", "stale-web-observer-process-tree"]): return execute_memory_process_candidate(candidate) + if kind == "cgroup-memory-reclaim": + return execute_cgroup_reclaim_candidate(candidate) if kind == "k3s-cri-image-prune": return execute_k3s_image_cache_prune() if kind == "host-containerd-cache-prune": @@ -1577,6 +1662,14 @@ def returned_results(results): "estimatedDiskBytes", "expectedMemoryRssBytes", "reclaimedMemoryBytes", + "outcome", + "configId", + "reclaimBytes", + "swappiness", + "cgroupMemoryDeltaBytes", + "hostMemAvailableDeltaBytes", + "kernelWriteAttempted", + "kernelWriteError", "processCount", "termSignalCount", "killSignalCount", @@ -1626,8 +1719,8 @@ def plan_payload(observed_at, preflight, protected, candidates, visible): "requiresRunConfirm": True, "planCommand": "bun scripts/cli.ts gc remote %s plan --memory-pressure-only" % PROVIDER_ID, "runCommand": "bun scripts/cli.ts gc remote %s run --confirm --memory-pressure-only" % PROVIDER_ID, - "candidateKinds": ["browser-orphan-process-tree", "stale-web-observer-process-tree"], - "neverTouches": ["disk candidates", "apt cache", "journal", "tmp", "container images", "active observer process trees"], + "candidateKinds": ["cgroup-memory-reclaim", "browser-orphan-process-tree", "stale-web-observer-process-tree"], + "neverTouches": ["disk candidates", "apt cache", "journal", "tmp", "container images", "active observer process trees", "workload lifecycle"], "termination": "SIGTERM, bounded wait, identity reclose, SIGKILL only same-tree residual", }, } @@ -1710,18 +1803,27 @@ def safe_unit_name(value): raw = "unidesk-%s-low-risk-gc" % re.sub(r"[^a-z0-9]+", "-", PROVIDER_ID.lower()).strip("-") return raw[:80] +def required_policy_bool(key): + value = POLICY_TIMER_CONFIG.get(key) if isinstance(POLICY_TIMER_CONFIG, dict) else None + if not isinstance(value, bool): + raise RuntimeError("gc.remote.targets.%s.policyTimer.%s must be a YAML boolean" % (PROVIDER_ID, key)) + return value + def render_remote_policy(): unit_name = safe_unit_name(config_str(POLICY_TIMER_CONFIG, "name", "unidesk-%s-low-risk-gc" % PROVIDER_ID.lower())) on_calendar = config_str(POLICY_TIMER_CONFIG, "onCalendar", "daily") randomized_delay_sec = config_str(POLICY_TIMER_CONFIG, "randomizedDelaySec", "15min") journal_target = parse_size_value(POLICY_TIMER_CONFIG.get("journalTargetBytes"), int(OPTIONS.get("journalTargetBytes") or 536870912)) tmp_min_age_hours = config_float(POLICY_TIMER_CONFIG, "tmpMinAgeHours", float(OPTIONS.get("tmpMinAgeHours") or 24), minimum=0.0) - include_apt_cache = config_bool(POLICY_TIMER_CONFIG, "includeAptCache", bool(OPTIONS.get("aptCache", True))) - include_tool_caches = config_bool(POLICY_TIMER_CONFIG, "includeToolCaches", False) - include_web_observe = config_bool(POLICY_TIMER_CONFIG, "includeWebObserveArtifacts", False) - include_k3s_images = config_bool(POLICY_TIMER_CONFIG, "includeK3sImageCache", False) - include_host_containerd = config_bool(POLICY_TIMER_CONFIG, "includeHostContainerdCache", False) - include_local_path_orphans = config_bool(POLICY_TIMER_CONFIG, "includeLocalPathOrphans", False) + include_journal = required_policy_bool("includeJournal") + include_tmp = required_policy_bool("includeTmp") + include_apt_cache = required_policy_bool("includeAptCache") + include_tool_caches = required_policy_bool("includeToolCaches") + include_web_observe = required_policy_bool("includeWebObserveArtifacts") + include_k3s_images = required_policy_bool("includeK3sImageCache") + include_host_containerd = required_policy_bool("includeHostContainerdCache") + include_local_path_orphans = required_policy_bool("includeLocalPathOrphans") + include_kubernetes_retention = required_policy_bool("includeKubernetesObjectRetention") state_dir = config_str(POLICY_TIMER_CONFIG, "stateDir", "/var/lib/unidesk-gc") config_dir = config_str(POLICY_TIMER_CONFIG, "configDir", "/etc/unidesk-gc") script_path = "/usr/local/sbin/%s.py" % unit_name @@ -1733,6 +1835,9 @@ def render_remote_policy(): "providerId": PROVIDER_ID, "unitName": unit_name, "statePath": state_path, + "includeJournal": include_journal, + "includeTmp": include_tmp, + "includeToolCaches": include_tool_caches, "journalTargetBytes": int(journal_target), "tmpMinAgeHours": tmp_min_age_hours, "tmpPrefixAllowlist": TMP_PREFIX_ALLOWLIST, @@ -1743,18 +1848,23 @@ def render_remote_policy(): "k3sImageCache": {"enabled": include_k3s_images, **CONTAINERD_CONFIG}, "hostContainerdCache": {"enabled": include_host_containerd, **HOST_CONTAINERD_CONFIG}, "localPathStorage": {"enabled": include_local_path_orphans, **LOCAL_PATH_CONFIG}, + "kubernetesObjectRetention": ( + {**KUBERNETES_RETENTION_CONFIG, "enabled": True} + if include_kubernetes_retention else {"enabled": False} + ), "agentrunSessionPvcs": POLICY_TIMER_CONFIG.get("agentrunSessionPvcs") if isinstance(POLICY_TIMER_CONFIG.get("agentrunSessionPvcs"), dict) else {"enabled": False}, } stages = [ - {"id": "journal", "enabled": True, "risk": "low", "mode": "journalctl-vacuum"}, + {"id": "journal", "enabled": include_journal, "risk": "low", "mode": "journalctl-vacuum"}, {"id": "apt-cache", "enabled": include_apt_cache, "risk": "low", "mode": "apt-get-clean"}, - {"id": "tmp-allowlist", "enabled": True, "risk": "low", "mode": "direct-child-prefix-retention"}, + {"id": "tmp-allowlist", "enabled": include_tmp, "risk": "low", "mode": "direct-child-prefix-retention"}, {"id": "tool-caches", "enabled": include_tool_caches, "risk": "medium", "mode": "fixed-path-rebuildable-cache"}, {"id": "web-observe-artifacts", "enabled": include_web_observe, "risk": "medium", "mode": "manifest-heartbeat-pid-openfd-stale-run"}, {"id": "agentrun-session-pvcs", "enabled": bool(policy_config["agentrunSessionPvcs"].get("enabled")), "risk": "medium", "mode": "kubectl-delete-pvc-wait-false-owner-aware"}, {"id": "k3s-image-cache", "enabled": include_k3s_images, "risk": "medium", "mode": "crictl-rmi-prune-ci-idle"}, {"id": "host-containerd-orphans", "enabled": include_host_containerd, "risk": "medium", "mode": "ctr-metadata-empty-yaml-orphan-state"}, {"id": "local-path-orphans", "enabled": include_local_path_orphans, "risk": "medium", "mode": "pv-unreferenced-direct-child"}, + {"id": "kubernetes-object-retention", "enabled": include_kubernetes_retention, "risk": "medium", "mode": "terminal-owner-cascade-bounded"}, ] service = "\n".join([ "[Unit]", @@ -1796,12 +1906,15 @@ def render_remote_policy(): "journalTargetBytes": int(journal_target), "journalTarget": fmt_bytes(journal_target), "tmpMinAgeHours": tmp_min_age_hours, + "includeJournal": include_journal, + "includeTmp": include_tmp, "includeAptCache": include_apt_cache, "includeToolCaches": include_tool_caches, "includeWebObserveArtifacts": include_web_observe, "includeK3sImageCache": include_k3s_images, "includeHostContainerdCache": include_host_containerd, "includeLocalPathOrphans": include_local_path_orphans, + "includeKubernetesObjectRetention": include_kubernetes_retention, "stages": stages, "policyConfig": policy_config, "script": POLICY_RUNNER_SOURCE, @@ -1820,7 +1933,7 @@ def remote_policy_plan_payload(observed_at): "observedAt": observed_at, "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.policyTimer" % PROVIDER_ID, "enabled": config_bool(POLICY_TIMER_CONFIG, "enabled", False), - "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec", "journalTargetBytes", "journalTarget", "tmpMinAgeHours", "includeAptCache", "includeToolCaches", "includeWebObserveArtifacts", "includeK3sImageCache", "includeHostContainerdCache", "includeLocalPathOrphans"]}, + "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec", "journalTargetBytes", "journalTarget", "tmpMinAgeHours", "includeJournal", "includeTmp", "includeAptCache", "includeToolCaches", "includeWebObserveArtifacts", "includeK3sImageCache", "includeHostContainerdCache", "includeLocalPathOrphans", "includeKubernetesObjectRetention"]}, "stages": rendered.get("stages"), "servicePreview": rendered["service"] if bool(OPTIONS.get("full")) else None, "timerPreview": rendered["timer"] if bool(OPTIONS.get("full")) else None, @@ -1888,6 +2001,12 @@ def compact_policy_last_run(last): "deletedPvcCount", "protectedCount", "candidateCount", + "requestedDeleteCount", + "deletedCount", + "objectCountDelta", + "configId", + "namespace", + "resource", "reclaimedBytes", ] if item.get(key) is not None @@ -1947,7 +2066,7 @@ def remote_policy_install_payload(observed_at): "mutation": True, "observedAt": observed_at, "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.policyTimer" % PROVIDER_ID, - "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec", "journalTargetBytes", "journalTarget", "tmpMinAgeHours", "includeAptCache", "includeToolCaches", "includeWebObserveArtifacts", "includeK3sImageCache", "includeHostContainerdCache", "includeLocalPathOrphans"]}, + "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec", "journalTargetBytes", "journalTarget", "tmpMinAgeHours", "includeJournal", "includeTmp", "includeAptCache", "includeToolCaches", "includeWebObserveArtifacts", "includeK3sImageCache", "includeHostContainerdCache", "includeLocalPathOrphans", "includeKubernetesObjectRetention"]}, "stages": rendered.get("stages"), "systemd": { "daemonReload": bounded(daemon), @@ -1956,6 +2075,247 @@ def remote_policy_install_payload(observed_at): }, } +def kubernetes_retention_plan_payload(observed_at): + try: + plan = kubernetes_retention_plan( + KUBERNETES_RETENTION_CONFIG, + "manual", + limit_override=int(OPTIONS.get("limit") or 1), + full=bool(OPTIONS.get("full")), + ) + except Exception as exc: + return { + "ok": False, + "action": "gc remote retention plan", + "providerId": PROVIDER_ID, + "dryRun": True, + "mutation": False, + "observedAt": observed_at, + "error": "kubernetes-retention-plan-failed", + "message": str(exc), + "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.kubernetesObjectRetention" % PROVIDER_ID, + } + plan = dict(plan) + plan.pop("selectedNames", None) + plan.update({ + "action": "gc remote retention plan", + "providerId": PROVIDER_ID, + "observedAt": observed_at, + "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.kubernetesObjectRetention" % PROVIDER_ID, + "runCommand": "bun scripts/cli.ts gc remote %s retention run --confirm --limit %s" % ( + PROVIDER_ID, + int(OPTIONS.get("limit") or 1), + ), + "statusCommandTemplate": "bun scripts/cli.ts gc remote %s status --job-id " % PROVIDER_ID, + }) + return plan + +def update_kubernetes_retention_job_progress(paths, completed_count, candidate_count, worker_pid, worker_start_ticks): + descriptor = acquire_job_lock(paths) + try: + with open(paths["state"], "r", encoding="utf-8") as handle: + payload = json.load(handle) + if payload.get("status") != "running" or safe_int(payload.get("workerPid")) != worker_pid \ + or safe_int(payload.get("workerStartTicks")) != worker_start_ticks: + raise RuntimeError("kubernetes retention job state no longer belongs to this worker") + payload.update({ + "workerHeartbeatAt": now_iso(), + "progress": {"completedCount": completed_count, "candidateCount": candidate_count}, + }) + write_json_atomic(paths["state"], payload) + finally: + release_job_lock(descriptor) + +def finish_kubernetes_retention_job(job_id, observed_at, limit_override, worker_pid, worker_start_ticks): + paths = job_paths(job_id) + memory_before = collect_memory_snapshot() + + def progress(completed_count, candidate_count): + update_kubernetes_retention_job_progress( + paths, + completed_count, + candidate_count, + worker_pid, + worker_start_ticks, + ) + + result = kubernetes_retention_execute( + KUBERNETES_RETENTION_CONFIG, + "manual", + limit_override=limit_override, + progress=progress, + full=bool(OPTIONS.get("full")), + ) + memory_after = collect_memory_snapshot() + requested_count = safe_int(result.get("requestedDeleteCount")) + deleted_count = safe_int(result.get("deletedCount")) + payload = { + "ok": result.get("ok") is True, + "action": "gc remote retention run", + "providerId": PROVIDER_ID, + "scope": "kubernetes-object-retention", + "kind": "kubernetes-object-retention", + "status": "succeeded" if result.get("ok") is True else "failed", + "jobId": job_id, + "statusCommand": status_command(job_id), + "statePath": paths["state"], + "dryRun": False, + "mutation": result.get("mutation") is True, + "observedAt": observed_at, + "startedAt": observed_at, + "finishedAt": now_iso(), + "workerPid": worker_pid, + "workerStartTicks": worker_start_ticks, + "workerAlive": False, + "workerCompleted": True, + "workerHeartbeatAt": now_iso(), + "progress": {"completedCount": deleted_count, "candidateCount": requested_count}, + "memoryBefore": memory_before, + "memoryAfter": memory_after, + "memoryAvailableDeltaBytes": ( + safe_int((memory_after.get("memory") or {}).get("availableBytes")) + - safe_int((memory_before.get("memory") or {}).get("availableBytes")) + ) if memory_before.get("ok") and memory_after.get("ok") else None, + "summary": { + "configId": result.get("configId"), + "namespace": result.get("namespace"), + "resource": result.get("resource"), + "requestedDeleteCount": requested_count, + "deletedCount": deleted_count, + "objectCountDelta": result.get("objectCountDelta"), + "beforeCount": (result.get("before") or {}).get("count"), + "afterCount": (result.get("after") or {}).get("count"), + "failedBatchCount": len([item for item in result.get("batches") or [] if item.get("ok") is not True]), + }, + "retention": result, + "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.kubernetesObjectRetention" % PROVIDER_ID, + } + write_memory_job_state(paths, payload) + +def start_kubernetes_retention_job(observed_at): + try: + retention_config = kubernetes_retention_config(KUBERNETES_RETENTION_CONFIG) + except Exception as exc: + return { + "ok": False, + "action": "gc remote retention run", + "providerId": PROVIDER_ID, + "dryRun": True, + "mutation": False, + "status": "blocked", + "error": "kubernetes-retention-config-invalid", + "message": str(exc), + } + limit_override = int(OPTIONS.get("limit") or 1) + job_id = "%s-kubernetes-retention-%s-%s" % ( + re.sub(r"[^A-Za-z0-9._-]+", "-", PROVIDER_ID.lower()).strip("-") or "provider", + int(time.time()), + os.getpid(), + ) + paths = job_paths(job_id) + base = { + "ok": True, + "action": "gc remote retention run", + "providerId": PROVIDER_ID, + "scope": "kubernetes-object-retention", + "kind": "kubernetes-object-retention", + "jobId": job_id, + "statusCommand": status_command(job_id), + "statePath": paths["state"], + "dryRun": False, + "mutation": False, + "mutationPending": True, + "observedAt": observed_at, + "startedAt": observed_at, + "configId": retention_config.get("configId"), + "namespace": retention_config.get("namespace"), + "resource": retention_config.get("resource"), + "limit": min(limit_override, int(retention_config.get("manualMaxDeletePerRun"))), + "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.kubernetesObjectRetention" % PROVIDER_ID, + } + read_descriptor, write_descriptor = os.pipe() + try: + child_pid = os.fork() + except OSError as exc: + os.close(read_descriptor) + os.close(write_descriptor) + failed = dict(base) + failed.update({"ok": False, "status": "failed", "error": "kubernetes-retention-fork-failed", "message": str(exc), "finishedAt": now_iso()}) + write_memory_job_state(paths, failed) + return failed + if child_pid == 0: + try: + os.close(write_descriptor) + released = os.read(read_descriptor, 1) + os.close(read_descriptor) + if released != b"1": + os._exit(1) + os.setsid() + with open("/dev/null", "rb", buffering=0) as stdin_handle, open(paths["log"], "ab", buffering=0) as log_handle: + os.dup2(stdin_handle.fileno(), 0) + os.dup2(log_handle.fileno(), 1) + os.dup2(log_handle.fileno(), 2) + finish_kubernetes_retention_job( + job_id, + observed_at, + limit_override, + os.getpid(), + process_start_ticks(os.getpid()), + ) + except Exception as exc: + failed = dict(base) + failed.update({ + "ok": False, + "status": "failed", + "error": "kubernetes-retention-worker-failed", + "message": str(exc), + "workerPid": os.getpid(), + "workerStartTicks": process_start_ticks(os.getpid()), + "finishedAt": now_iso(), + }) + try: + write_memory_job_state(paths, failed) + except Exception: + pass + finally: + os._exit(0) + os.close(read_descriptor) + worker_start_ticks = process_start_ticks(child_pid) + if worker_start_ticks is None: + os.close(write_descriptor) + try: + os.kill(child_pid, signal.SIGKILL) + except ProcessLookupError: + pass + failed = dict(base) + failed.update({"ok": False, "status": "failed", "error": "kubernetes-retention-worker-identity-unavailable", "workerPid": child_pid, "finishedAt": now_iso()}) + write_memory_job_state(paths, failed) + return failed + running = dict(base) + running.update({ + "status": "running", + "workerPid": child_pid, + "workerStartTicks": worker_start_ticks, + "workerAlive": True, + "workerHeartbeatAt": now_iso(), + "progress": {"completedCount": 0, "candidateCount": None}, + }) + try: + write_memory_job_state(paths, running) + os.write(write_descriptor, b"1") + except Exception as exc: + try: + os.kill(child_pid, signal.SIGKILL) + except ProcessLookupError: + pass + failed = dict(base) + failed.update({"ok": False, "status": "failed", "error": "kubernetes-retention-worker-release-failed", "message": str(exc), "workerPid": child_pid, "workerStartTicks": worker_start_ticks, "finishedAt": now_iso()}) + write_memory_job_state(paths, failed) + return failed + finally: + os.close(write_descriptor) + return running + def update_memory_pressure_job_progress(paths, completed_count, candidate_count, worker_pid, worker_start_ticks): descriptor = acquire_job_lock(paths) try: @@ -1983,8 +2343,15 @@ def finish_memory_pressure_job(job_id, candidates, observed_at, worker_pid, work item = dict(candidate) item.update({ "status": execution.get("status") or "succeeded", + "outcome": execution.get("outcome"), "reclaimedBytes": 0, "reclaimedMemoryBytes": execution.get("reclaimedMemoryBytes"), + "hostMemAvailableDeltaBytes": execution.get("hostMemAvailableDeltaBytes"), + "cgroupMemoryDeltaBytes": execution.get("cgroupMemoryDeltaBytes"), + "cgroupMemoryBeforeBytes": execution.get("cgroupMemoryBeforeBytes"), + "cgroupMemoryAfterBytes": execution.get("cgroupMemoryAfterBytes"), + "kernelWriteAttempted": execution.get("kernelWriteAttempted"), + "kernelWriteError": execution.get("kernelWriteError"), "termSignalCount": execution.get("termSignalCount"), "killSignalCount": execution.get("killSignalCount"), "remainingIdentityCount": execution.get("remainingIdentityCount"), @@ -2157,6 +2524,9 @@ def start_memory_pressure_job(observed_at, candidates): def main(): observed_at = now_iso() + if ACTION == "memory-distribution": + emit_json(collect_memory_distribution(observed_at), persist_large=True) + return 0 if ACTION == "memory": memory = collect_memory_snapshot() emit_json({ @@ -2170,6 +2540,7 @@ def main(): "source": memory.get("source"), "swapExcluded": memory.get("swapExcluded") is True, "memory": memory.get("memory"), + "swap": memory.get("swap"), "error": memory.get("error"), "valuesRedacted": True, }, persist_large=False) @@ -2177,6 +2548,12 @@ def main(): if ACTION == "status" and job_id_or_none(): emit_json(remote_gc_job_status(), persist_large=False) return 0 + if ACTION == "retention-plan": + emit_json(kubernetes_retention_plan_payload(observed_at), persist_large=True) + return 0 + if ACTION == "retention-run": + emit_json(start_kubernetes_retention_job(observed_at), persist_large=False) + return 0 if bool(OPTIONS.get("memoryPressureOnly")) and ACTION in set(["plan", "run"]): candidates = collect_memory_reclaim_candidates() visible = visible_items(candidates) diff --git a/scripts/src/gc-remote-web-observe.py b/scripts/src/gc-remote-web-observe.py index a3525efc..4493e8e2 100644 --- a/scripts/src/gc-remote-web-observe.py +++ b/scripts/src/gc-remote-web-observe.py @@ -5,6 +5,8 @@ _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", [])) @@ -105,8 +107,61 @@ def assert_web_observe_candidate(path): 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 + 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")) @@ -129,10 +184,12 @@ def read_proc_process_table(): 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]) @@ -150,6 +207,8 @@ def read_proc_process_table(): "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, @@ -163,6 +222,7 @@ def read_proc_process_table(): _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): @@ -250,14 +310,157 @@ def command_matches(row, patterns): 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") + reclaim_bytes = item.get("reclaimBytes") + swappiness = item.get("swappiness") + 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 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 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 [] + 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, + "reclaimBytes": reclaim_bytes, + "swappiness": swappiness, + "priority": priority, + "targetMemAvailableBytes": target_available, + "minimumSwapFreeBytesAfterReclaim": swap_reserve, + }) + return sorted(validated, key=lambda item: item.get("priority")) + +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: + path = item.get("path") + 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") + if protected_reason is not None: + diagnostics["protected"].append({ + "id": item.get("id"), + "path": path, + "reason": protected_reason, + "memoryCurrentBytes": current, + }) + continue + memory_stat = read_cgroup_memory_stat(path) + reclaim_bytes = safe_int(item.get("reclaimBytes")) + candidates.append({ + "id": "cgroup-memory-reclaim:%s" % item.get("id"), + "configId": item.get("id"), + "kind": "cgroup-memory-reclaim", + "risk": "medium", + "description": "Request bounded cgroup v2 memory reclaim from an exact YAML allowlist path", + "reason": "host-memavailable-below-yaml-target", + "path": path, + "priority": safe_int(item.get("priority")), + "reclaimBytes": reclaim_bytes, + "swappiness": safe_int(item.get("swappiness")), + "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, + "expectedMemoryRssBytes": reclaim_bytes, + "expectedMemoryRss": fmt_bytes(reclaim_bytes), + "estimatedDiskBytes": 0, + "estimatedReclaimBytes": 0, + "action": {"op": "cgroup-v2-memory-reclaim", "allowlist": "yaml-exact-cgroup-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 + 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", @@ -329,7 +532,7 @@ def collect_memory_reclaim_candidates(): protected_pids = set() for pid in fresh_observer_pids: protected_pids.update(process_descendants(processes, pid)) - candidates = [] + 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: @@ -386,7 +589,11 @@ def collect_memory_reclaim_candidates(): "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: safe_int(item.get("expectedMemoryRssBytes")), reverse=True)[:max_candidates] + 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) @@ -398,6 +605,8 @@ def memory_reclaim_scan_diagnostics(): "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 []), @@ -453,6 +662,70 @@ def extend_with_allowlisted_session_members(identities, processes, sid, browser_ known.add(identity_key) return added +def execute_cgroup_reclaim_candidate(candidate): + 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 ["path", "reclaimBytes", "swappiness", "targetMemAvailableBytes", "minimumSwapFreeBytesAfterReclaim"]: + if configured.get(key) != candidate.get(key): + raise RuntimeError("cgroup reclaim candidate no longer matches YAML: %s" % key) + path = configured.get("path") + 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") + 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 + return { + "status": "succeeded" if succeeded else "failed", + "outcome": "partial-reclaim-after-kernel-error" if write_error and succeeded else ("reclaim-requested" if succeeded else "reclaim-request-failed"), + "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: diff --git a/scripts/src/gc-remote.ts b/scripts/src/gc-remote.ts index 94dd540b..3100d6a6 100644 --- a/scripts/src/gc-remote.ts +++ b/scripts/src/gc-remote.ts @@ -5,7 +5,7 @@ import { type UniDeskConfig, rootPath } from "./config"; import { remoteGcDegradedFailure } from "./gc-remote-degraded"; import { runSshCommandCapture } from "./ssh"; -type RemoteGcAction = "plan" | "memory" | "snapshot" | "trend" | "run" | "status" | "policy-plan" | "policy-install" | "policy-status"; +type RemoteGcAction = "plan" | "memory" | "memory-distribution" | "retention-plan" | "retention-run" | "snapshot" | "trend" | "run" | "status" | "policy-plan" | "policy-install" | "policy-status"; interface RemoteGcOptions { confirm: boolean; @@ -75,6 +75,11 @@ const GC_REMOTE_RUNNER_RELATIVE_PATH = "scripts/src/gc-remote-runner.py"; const GC_REMOTE_RUNNER_CONFIG_PLACEHOLDER = "__UNIDESK_GC_REMOTE_CONFIG_BASE64__"; const GC_REMOTE_WEB_OBSERVE_RELATIVE_PATH = "scripts/src/gc-remote-web-observe.py"; const GC_REMOTE_WEB_OBSERVE_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_WEB_OBSERVE_HELPERS__"; +const GC_REMOTE_MEMORY_DISTRIBUTION_RELATIVE_PATH = "scripts/src/gc-remote-memory-distribution.py"; +const GC_REMOTE_MEMORY_DISTRIBUTION_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_MEMORY_DISTRIBUTION_HELPERS__"; +const GC_REMOTE_KUBERNETES_RETENTION_RELATIVE_PATH = "scripts/src/gc-remote-kubernetes-retention.py"; +const GC_REMOTE_KUBERNETES_RETENTION_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_KUBERNETES_RETENTION_HELPERS__"; +const GC_REMOTE_POLICY_KUBERNETES_RETENTION_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_POLICY_KUBERNETES_RETENTION_HELPERS__"; const GC_REMOTE_CONTAINERD_RELATIVE_PATH = "scripts/src/gc-remote-containerd.py"; const GC_REMOTE_CONTAINERD_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_CONTAINERD_HELPERS__"; const GC_REMOTE_PVC_RELATIVE_PATH = "scripts/src/gc-remote-pvc.py"; @@ -91,10 +96,40 @@ export async function runRemoteGcCommand(config: UniDeskConfig, providerId: stri return { ok: false, error: "gc-remote-provider-required", - usage: "bun scripts/cli.ts gc remote memory|plan|snapshot|trend|run|status|policy [--confirm]", + usage: "bun scripts/cli.ts gc remote memory|memory-distribution|retention|plan|snapshot|trend|run|status|policy [--confirm]", }; } const subaction = action ?? "plan"; + if (subaction === "retention") { + const [retentionAction = "plan", ...retentionArgs] = args; + const options = parseRemoteGcOptions(retentionArgs); + if (retentionAction === "plan" || retentionAction === "dry-run") { + return await runRemoteGc(config, providerId, "retention-plan", options); + } + if (retentionAction === "run") { + if (!options.confirm) { + return { + ok: false, + error: "gc-remote-retention-run-requires-confirm", + dryRun: true, + mutation: false, + requiredFlag: "--confirm", + planCommand: `bun scripts/cli.ts gc remote ${providerId} retention plan --limit ${options.limit}`, + runCommand: `bun scripts/cli.ts gc remote ${providerId} retention run --confirm --limit ${options.limit}`, + }; + } + return await runRemoteGc(config, providerId, "retention-run", options); + } + if (retentionAction === "status") { + return await runRemoteGc(config, providerId, "status", options); + } + return { + ok: false, + error: "unsupported-gc-remote-retention-action", + action: retentionAction, + supportedActions: ["plan", "run", "status"], + }; + } if (subaction === "policy") { const [policyAction = "plan", ...policyArgs] = args; const options = parseRemoteGcOptions(policyArgs); @@ -137,6 +172,9 @@ export async function runRemoteGcCommand(config: UniDeskConfig, providerId: stri }; } if (subaction === "memory") return await runRemoteGc(config, providerId, "memory", options); + if (subaction === "memory-distribution" || subaction === "memory-overview") { + return await runRemoteGc(config, providerId, "memory-distribution", options); + } if (subaction === "plan" || subaction === "dry-run") return await runRemoteGc(config, providerId, "plan", options); if (subaction === "snapshot" || subaction === "growth") return await runRemoteGc(config, providerId, "snapshot", options); if (subaction === "trend") return await runRemoteGc(config, providerId, "trend", options); @@ -159,7 +197,7 @@ export async function runRemoteGcCommand(config: UniDeskConfig, providerId: stri ok: false, error: "unsupported-gc-remote-action", action: subaction, - supportedActions: ["memory", "plan", "snapshot", "trend", "run", "status", "policy"], + supportedActions: ["memory", "memory-distribution", "retention", "plan", "snapshot", "trend", "run", "status", "policy"], }; } @@ -352,6 +390,12 @@ function remoteGcPython(configBase64: string): string { if (!template.includes(GC_REMOTE_WEB_OBSERVE_PLACEHOLDER)) { throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_WEB_OBSERVE_PLACEHOLDER}`); } + if (!template.includes(GC_REMOTE_MEMORY_DISTRIBUTION_PLACEHOLDER)) { + throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_MEMORY_DISTRIBUTION_PLACEHOLDER}`); + } + if (!template.includes(GC_REMOTE_KUBERNETES_RETENTION_PLACEHOLDER)) { + throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_KUBERNETES_RETENTION_PLACEHOLDER}`); + } if (!template.includes(GC_REMOTE_CONTAINERD_PLACEHOLDER)) { throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_CONTAINERD_PLACEHOLDER}`); } @@ -368,17 +412,28 @@ function remoteGcPython(configBase64: string): string { throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_POLICY_RUNNER_PLACEHOLDER}`); } const webObserveHelpers = readFileSync(rootPath(GC_REMOTE_WEB_OBSERVE_RELATIVE_PATH), "utf8"); + const memoryDistributionHelpers = readFileSync(rootPath(GC_REMOTE_MEMORY_DISTRIBUTION_RELATIVE_PATH), "utf8"); + const kubernetesRetentionHelpers = readFileSync(rootPath(GC_REMOTE_KUBERNETES_RETENTION_RELATIVE_PATH), "utf8"); const containerdHelpers = readFileSync(rootPath(GC_REMOTE_CONTAINERD_RELATIVE_PATH), "utf8"); const pvcHelpers = readFileSync(rootPath(GC_REMOTE_PVC_RELATIVE_PATH), "utf8"); const growthHelpers = readFileSync(rootPath(GC_REMOTE_GROWTH_RELATIVE_PATH), "utf8"); const registryHelpers = readFileSync(rootPath(GC_REMOTE_REGISTRY_RELATIVE_PATH), "utf8"); - const policyRunner = readFileSync(rootPath(GC_REMOTE_POLICY_RUNNER_RELATIVE_PATH), "utf8"); + const policyRunnerTemplate = readFileSync(rootPath(GC_REMOTE_POLICY_RUNNER_RELATIVE_PATH), "utf8"); + if (!policyRunnerTemplate.includes(GC_REMOTE_POLICY_KUBERNETES_RETENTION_PLACEHOLDER)) { + throw new Error(`${GC_REMOTE_POLICY_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_POLICY_KUBERNETES_RETENTION_PLACEHOLDER}`); + } + const policyRunner = policyRunnerTemplate.replace( + GC_REMOTE_POLICY_KUBERNETES_RETENTION_PLACEHOLDER, + () => kubernetesRetentionHelpers.trimEnd(), + ); return template - .replace(GC_REMOTE_WEB_OBSERVE_PLACEHOLDER, webObserveHelpers.trimEnd()) - .replace(GC_REMOTE_CONTAINERD_PLACEHOLDER, containerdHelpers.trimEnd()) - .replace(GC_REMOTE_PVC_PLACEHOLDER, pvcHelpers.trimEnd()) - .replace(GC_REMOTE_GROWTH_PLACEHOLDER, growthHelpers.trimEnd()) - .replace(GC_REMOTE_REGISTRY_PLACEHOLDER, registryHelpers.trimEnd()) + .replace(GC_REMOTE_MEMORY_DISTRIBUTION_PLACEHOLDER, () => memoryDistributionHelpers.trimEnd()) + .replace(GC_REMOTE_KUBERNETES_RETENTION_PLACEHOLDER, () => kubernetesRetentionHelpers.trimEnd()) + .replace(GC_REMOTE_WEB_OBSERVE_PLACEHOLDER, () => webObserveHelpers.trimEnd()) + .replace(GC_REMOTE_CONTAINERD_PLACEHOLDER, () => containerdHelpers.trimEnd()) + .replace(GC_REMOTE_PVC_PLACEHOLDER, () => pvcHelpers.trimEnd()) + .replace(GC_REMOTE_GROWTH_PLACEHOLDER, () => growthHelpers.trimEnd()) + .replace(GC_REMOTE_REGISTRY_PLACEHOLDER, () => registryHelpers.trimEnd()) .replace(GC_REMOTE_POLICY_RUNNER_PLACEHOLDER, Buffer.from(policyRunner, "utf8").toString("base64")) .replace(GC_REMOTE_RUNNER_CONFIG_PLACEHOLDER, configBase64); } diff --git a/scripts/src/help.ts b/scripts/src/help.ts index 62fc4b72..a6357545 100644 --- a/scripts/src/help.ts +++ b/scripts/src/help.ts @@ -519,6 +519,9 @@ function gcHelp(): unknown { "bun scripts/cli.ts gc policy install", "bun scripts/cli.ts gc remote G14 plan", "bun scripts/cli.ts gc remote NC01 memory", + "bun scripts/cli.ts gc remote NC01 memory-distribution", + "bun scripts/cli.ts gc remote NC01 retention plan --limit 500", + "bun scripts/cli.ts gc remote NC01 retention run --confirm --limit 500", "bun scripts/cli.ts gc remote NC01 plan --memory-pressure-only", "bun scripts/cli.ts gc remote NC01 run --confirm --memory-pressure-only", "bun scripts/cli.ts gc remote NC01 status --job-id ", @@ -569,6 +572,8 @@ function gcHelp(): unknown { "policy plan|install": "render or install journald caps and a daily low-risk gc systemd timer from config/unidesk-cli.yaml#gc.policyTimer", "remote plan|run": "run bounded GC through UniDesk SSH passthrough on a provider host; G14 protects HWLAB k3s/runtime/PVC/workspace paths, and HWLAB registry retention is explicit opt-in with workload-ref, digest-closure, recent-tag and per-repo tag protection", "remote memory": "只读输出目标节点 /proc/meminfo 的 MemAvailable；不合并 swap、cgroup 或 virtual memory，也不执行 GC mutation", + "remote memory-distribution": "只读输出目标节点物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布的紧凑 overview", + "remote retention plan|run|status": "按 owning YAML 对终态 Kubernetes 对象执行 fail-closed 计划、异步确认回收和状态查询；自动 policy 复用同一筛选器", "--no-file-logs|--no-docker-logs|--no-journal|--no-build-cache|--no-tmp|--no-db-summary": "disable one collector", }, reference: "docs/reference/gc.md", diff --git a/src/components/microservices/decision-center/Dockerfile b/src/components/microservices/decision-center/Dockerfile index e50345a7..82ac7eda 100644 --- a/src/components/microservices/decision-center/Dockerfile +++ b/src/components/microservices/decision-center/Dockerfile @@ -3,7 +3,7 @@ FROM ${DECISION_CENTER_BASE_IMAGE} WORKDIR /app/src/components/microservices/decision-center RUN apt-get update \ - && apt-get install -y --no-install-recommends git openssh-client ca-certificates \ + && apt-get install -y --no-install-recommends git openssh-client ca-certificates tini \ && apt-get clean \ && rm -rf /var/lib/apt/lists/* COPY src/components/microservices/decision-center/package.json ./package.json @@ -13,4 +13,5 @@ COPY src/components/shared /app/src/components/shared COPY src/components/microservices/decision-center/src ./src EXPOSE 4277 +ENTRYPOINT ["tini", "--"] CMD ["bun", "run", "src/index.ts"] diff --git a/src/components/microservices/decision-center/src/index.ts b/src/components/microservices/decision-center/src/index.ts index 4f7ed614..4e7c5198 100644 --- a/src/components/microservices/decision-center/src/index.ts +++ b/src/components/microservices/decision-center/src/index.ts @@ -38,6 +38,7 @@ interface RuntimeConfig { databaseUrl: string; logFile: string; databasePoolMax: number; + storageHealthRefreshMs: number; storage: StorageConfig; } @@ -196,6 +197,15 @@ function envNumber(name: string, fallback: number): number { return Number.isFinite(value) && value > 0 ? Math.floor(value) : fallback; } +function requiredEnvNumber(name: string): number { + const raw = process.env[name]; + const value = Number(raw); + if (raw === undefined || raw.length === 0 || !Number.isFinite(value) || value <= 0) { + throw new Error(`${name} must be a positive number`); + } + return Math.floor(value); +} + function configFromEnv(): RuntimeConfig { const databaseUrl = process.env.DATABASE_URL || (isCheckMode ? "postgres://syntax-check@127.0.0.1:1/syntax-check" : ""); if (!databaseUrl) throw new Error("DATABASE_URL is required"); @@ -207,6 +217,7 @@ function configFromEnv(): RuntimeConfig { databaseUrl, logFile: envString("LOG_FILE", "/var/log/unidesk/decision-center.jsonl"), databasePoolMax: Math.max(1, Math.min(8, envNumber("DATABASE_POOL_MAX", 2))), + storageHealthRefreshMs: requiredEnvNumber("DECISION_CENTER_STORAGE_HEALTH_REFRESH_MS"), storage: { primary: storagePrimary, repoSshUrl: envString("DECISION_CENTER_GIT_REPO_SSH_URL", ""), @@ -2188,8 +2199,15 @@ if (import.meta.main && !isCheckMode) { setInterval(() => { if (schemaReady) { void refreshDatabaseHealth(); - void updateStorageHealth(); } }, 15_000).unref(); - log("info", "service_started", { host: config.host, port: config.port, storage: storageConfigSummary() }); + setInterval(() => { + if (schemaReady) void updateStorageHealth(); + }, config.storageHealthRefreshMs).unref(); + log("info", "service_started", { + host: config.host, + port: config.port, + storage: storageConfigSummary(), + storageHealthRefreshMs: config.storageHealthRefreshMs, + }); }