fix(ops): alert on 110 process-family pressure

scripts/ops/host-sustained-load-controller.py

@@ -20,6 +20,7 @@ from __future__ import annotations
 import argparse
 import json
 import re
+import subprocess
 import time
 from pathlib import Path
 from typing import Any
@@ -50,7 +51,11 @@ def parse_args() -> argparse.Namespace:
     )
     parser.add_argument("--load5-per-core-threshold", type=float, default=1.5)
     parser.add_argument("--container-cpu-threshold", type=float, default=2.0)
+    parser.add_argument("--hot-container-cpu-threshold", type=float, default=1.0)
+    parser.add_argument("--process-family-cpu-threshold", type=float, default=50.0)
     parser.add_argument("--ci-stale-age-seconds", type=int, default=1800)
+    parser.add_argument("--ps-file", type=Path)
+    parser.add_argument("--top-n", type=int, default=8)
     parser.add_argument("--json", action="store_true", help="Print JSON only.")
     return parser.parse_args()
 
@@ -211,14 +216,171 @@ def _top_container_cpu(samples: list[dict[str, Any]], *, host: str) -> dict[str,
     return sorted(candidates, key=lambda item: (-item["cpu_cores"], item["container_name"]))[0]
 
 
+def _read_text(path: Path | None) -> str:
+    if path is None:
+        return ""
+    try:
+        return path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return ""
+
+
+def _collect_ps_text(ps_file: Path | None) -> str:
+    if ps_file is not None:
+        return _read_text(ps_file)
+    try:
+        result = subprocess.run(
+            ["ps", "-eo", "pid=,ppid=,pgid=,etimes=,pcpu=,pmem=,comm=,args="],
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
+        return ""
+    return result.stdout
+
+
+def _classify_process_family(comm: str, args: str) -> str:
+    text = f"{comm} {args}".lower()
+    if (
+        "act_runner" in text
+        or "gitea-actions-task" in text
+        or "/.cache/act/" in text
+        or "/opt/hostedtoolcache/" in text
+        or "pnpm install" in text
+        or "npm ci" in text
+        or "yarn install" in text
+    ):
+        return "gitea_actions_runner"
+    if "docker build" in text or "buildx" in text or "buildkit" in text:
+        return "docker_build"
+    if "next build" in text or "turbo build" in text or ("pnpm" in text and " build" in text):
+        return "web_build"
+    if "chrome" in text or "chromium" in text or "playwright" in text:
+        return "headless_browser"
+    if "gitea" in text:
+        return "gitea_service"
+    if "postgres" in text or "postmaster" in text:
+        return "postgres"
+    if "clickhouse" in text:
+        return "clickhouse"
+    if "kafka" in text:
+        return "kafka"
+    if "sentry" in text:
+        return "sentry"
+    if "systemctl" in text or "systemd" in text or "dbus" in text:
+        return "systemd_control_plane"
+    if "sshd" in text:
+        return "ssh_control_plane"
+    if "python" in text:
+        return "python_job"
+    if "node" in text:
+        return "node_service"
+    return "unknown"
+
+
+def _parse_ps_text(text: str) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for raw_line in text.splitlines():
+        line = raw_line.strip()
+        if not line:
+            continue
+        parts = line.split(None, 7)
+        if len(parts) < 7:
+            continue
+        pid, ppid, pgid, etimes, pcpu, pmem, comm = parts[:7]
+        args = parts[7] if len(parts) > 7 else comm
+        try:
+            rows.append(
+                {
+                    "pid": int(pid),
+                    "ppid": int(ppid),
+                    "pgid": int(pgid),
+                    "etimes": int(float(etimes)),
+                    "cpu_percent": float(pcpu),
+                    "mem_percent": float(pmem),
+                    "comm": Path(comm).name[:48],
+                    "family": _classify_process_family(comm, args),
+                }
+            )
+        except ValueError:
+            continue
+    return rows
+
+
+def _summarize_processes(rows: list[dict[str, Any]], *, top_n: int) -> dict[str, Any]:
+    top_rows = sorted(rows, key=lambda item: (-item["cpu_percent"], item["comm"], item["pid"]))[:top_n]
+    families: dict[str, dict[str, Any]] = {}
+    for row in rows:
+        family = row["family"]
+        current = families.setdefault(
+            family,
+            {
+                "family": family,
+                "process_count": 0,
+                "cpu_percent": 0.0,
+                "max_age_seconds": 0,
+                "sample_comm": "",
+            },
+        )
+        current["process_count"] += 1
+        current["cpu_percent"] += row["cpu_percent"]
+        current["max_age_seconds"] = max(current["max_age_seconds"], row["etimes"])
+        if not current["sample_comm"] or row["cpu_percent"] > current.get("_sample_cpu", -1):
+            current["sample_comm"] = row["comm"]
+            current["_sample_cpu"] = row["cpu_percent"]
+
+    family_rows = []
+    for item in families.values():
+        item.pop("_sample_cpu", None)
+        item["cpu_percent"] = round(float(item["cpu_percent"]), 3)
+        family_rows.append(item)
+
+    return {
+        "top_processes": [
+            {
+                "pid": row["pid"],
+                "ppid": row["ppid"],
+                "pgid": row["pgid"],
+                "cpu_percent": round(row["cpu_percent"], 3),
+                "mem_percent": round(row["mem_percent"], 3),
+                "age_seconds": row["etimes"],
+                "comm": row["comm"],
+                "family": row["family"],
+            }
+            for row in top_rows
+        ],
+        "families": sorted(family_rows, key=lambda item: (-item["cpu_percent"], item["family"]))[
+            :top_n
+        ],
+    }
+
+
+def _family_cpu(process_summary: dict[str, Any], *families: str) -> float:
+    return sum(
+        float(item.get("cpu_percent") or 0.0)
+        for item in process_summary.get("families", [])
+        if item.get("family") in families
+    )
+
+
+def _top_family(process_summary: dict[str, Any]) -> dict[str, Any] | None:
+    families = process_summary.get("families", [])
+    return families[0] if families else None
+
+
 def build_packet(
     *,
     host: str,
     samples: list[dict[str, Any]],
     docker_samples: list[dict[str, Any]],
     docker_stats_status: dict[str, Any],
+    process_summary: dict[str, Any],
     load5_per_core_threshold: float,
     container_cpu_threshold: float,
+    hot_container_cpu_threshold: float,
+    process_family_cpu_threshold: float,
     ci_stale_age_seconds: int,
 ) -> dict[str, Any]:
     monitor_up = int(
@@ -272,6 +434,10 @@ def build_packet(
     top_container = raw_top_container if docker_stats_status.get("fresh") is True else None
     top_container_name = str((top_container or {}).get("container_name") or "").lower()
     top_container_cpu = float((top_container or {}).get("cpu_cores") or 0.0)
+    stock_process_cpu = _family_cpu(process_summary, "postgres")
+    gitea_process_cpu = _family_cpu(process_summary, "gitea_service")
+    control_plane_cpu = _family_cpu(process_summary, "systemd_control_plane", "ssh_control_plane")
+    top_family = _top_family(process_summary)
 
     classification = "observing_load_within_threshold"
     severity = "info"
@@ -329,19 +495,58 @@ def build_packet(
             if controlled_apply_allowed
             else "keep_pressure_gate_fail_closed_until_ci_load_clears"
         )
-    elif top_container_name == "gitea" and top_container_cpu >= container_cpu_threshold:
+    elif (
+        (
+            "stockplatform-v2-postgres-1" in top_container_name
+            or top_container_name == "stockplatform-v2-api-1"
+        )
+        and top_container_cpu >= container_cpu_threshold
+    ) or stock_process_cpu >= process_family_cpu_threshold * 2:
+        classification = "blocked_stockplatform_hot_query_or_api_pressure_requires_playbook"
+        severity = (
+            "critical"
+            if load5_per_core > load5_per_core_threshold
+            or stock_process_cpu >= process_family_cpu_threshold * 2
+            else "warning"
+        )
+        dry_run_command = (
+            "scripts/ops/host-sustained-load-evidence.py "
+            f"--host {host} --metrics-file {DEFAULT_METRICS_FILE} "
+            f"--docker-stats-file {DEFAULT_DOCKER_STATS_FILE} --json"
+        )
+        next_action = "run_stockplatform_hot_query_or_api_pressure_playbook_check_mode"
+    elif (
+        top_container_name == "gitea" and top_container_cpu >= container_cpu_threshold
+    ) or gitea_process_cpu >= process_family_cpu_threshold * 2:
         classification = "blocked_gitea_queue_or_hook_backlog_requires_playbook"
-        severity = "critical" if load5_per_core > load5_per_core_threshold else "warning"
+        severity = (
+            "critical"
+            if load5_per_core > load5_per_core_threshold
+            or gitea_process_cpu >= process_family_cpu_threshold * 2
+            else "warning"
+        )
         dry_run_command = (
             "scripts/ops/host-sustained-load-evidence.py "
             f"--host {host} --metrics-file {DEFAULT_METRICS_FILE} "
             f"--docker-stats-file {DEFAULT_DOCKER_STATS_FILE} --json"
         )
         next_action = "run_gitea_queue_or_hook_backlog_playbook_check_mode"
+    elif control_plane_cpu >= process_family_cpu_threshold:
+        classification = "blocked_control_plane_saturation_requires_playbook"
+        severity = "critical" if load5_per_core > load5_per_core_threshold else "warning"
+        dry_run_command = (
+            "scripts/ops/host-sustained-load-evidence.py "
+            f"--host {host} --metrics-file {DEFAULT_METRICS_FILE} "
+            f"--docker-stats-file {DEFAULT_DOCKER_STATS_FILE} --json"
+        )
+        next_action = "run_control_plane_saturation_playbook_check_mode"
     elif (
-        top_container_name in {"stockplatform-v2-postgres-1", "stockplatform-v2-api-1"}
-        and top_container_cpu >= container_cpu_threshold
-    ):
+        "stockplatform-v2-postgres-1" in top_container_name
+        and top_container_cpu >= hot_container_cpu_threshold
+    ) or (
+        top_container_name == "stockplatform-v2-api-1"
+        and top_container_cpu >= hot_container_cpu_threshold
+    ) or stock_process_cpu >= process_family_cpu_threshold:
         classification = "blocked_stockplatform_hot_query_or_api_pressure_requires_playbook"
         severity = "critical" if load5_per_core > load5_per_core_threshold else "warning"
         dry_run_command = (
@@ -350,6 +555,17 @@ def build_packet(
             f"--docker-stats-file {DEFAULT_DOCKER_STATS_FILE} --json"
         )
         next_action = "run_stockplatform_hot_query_or_api_pressure_playbook_check_mode"
+    elif (
+        top_container_name == "gitea" and top_container_cpu >= hot_container_cpu_threshold
+    ) or gitea_process_cpu >= process_family_cpu_threshold:
+        classification = "blocked_gitea_queue_or_hook_backlog_requires_playbook"
+        severity = "critical" if load5_per_core > load5_per_core_threshold else "warning"
+        dry_run_command = (
+            "scripts/ops/host-sustained-load-evidence.py "
+            f"--host {host} --metrics-file {DEFAULT_METRICS_FILE} "
+            f"--docker-stats-file {DEFAULT_DOCKER_STATS_FILE} --json"
+        )
+        next_action = "run_gitea_queue_or_hook_backlog_playbook_check_mode"
     elif load5_per_core > load5_per_core_threshold and swap_used_ratio >= 0.85:
         classification = "blocked_memory_or_swap_pressure_requires_service_playbook"
         severity = "critical"
@@ -378,6 +594,8 @@ def build_packet(
             "load5_per_core": round(load5_per_core, 6),
             "load5_per_core_threshold": load5_per_core_threshold,
             "container_cpu_threshold": container_cpu_threshold,
+            "hot_container_cpu_threshold": hot_container_cpu_threshold,
+            "process_family_cpu_threshold": process_family_cpu_threshold,
             "swap_used_ratio": round(swap_used_ratio, 6),
             "remediation_authorized": remediation_authorized,
             "active_ci_container_count": active_ci_containers,
@@ -388,6 +606,10 @@ def build_packet(
             "top_container_cpu": top_container,
             "top_container_cpu_untrusted": raw_top_container,
             "docker_stats": docker_stats_status,
+            "top_process_family": top_family,
+            "stock_process_cpu_percent": round(stock_process_cpu, 3),
+            "gitea_process_cpu_percent": round(gitea_process_cpu, 3),
+            "control_plane_process_cpu_percent": round(control_plane_cpu, 3),
         },
         "commands": {
             "dry_run": dry_run_command,
@@ -440,8 +662,11 @@ def main() -> int:
             docker_stats_file=args.docker_stats_file,
             max_age_seconds=args.docker_stats_max_age_seconds,
         ),
+        process_summary=_summarize_processes(_parse_ps_text(_collect_ps_text(args.ps_file)), top_n=args.top_n),
         load5_per_core_threshold=args.load5_per_core_threshold,
         container_cpu_threshold=args.container_cpu_threshold,
+        hot_container_cpu_threshold=args.hot_container_cpu_threshold,
+        process_family_cpu_threshold=args.process_family_cpu_threshold,
         ci_stale_age_seconds=args.ci_stale_age_seconds,
     )
     if args.json: