#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import sys from pathlib import Path DEFAULT_BUILD = { "tsc", "vite", "esbuild", "rollup", "webpack", "turbo", "pnpm", "npm", "yarn", "node", "cargo", "rustc", "cc", "clang", "gcc", "ld", "make", "ninja", "go", "bun", } METRIC_FILE = "cpu_nonbuild.prom" STATE_FILE = "cpu_nonbuild.state.json" def total_ticks(proc_root: Path) -> int: try: parts = (proc_root / "stat").read_text().splitlines()[0].split()[1:] return sum(int(p) for p in parts) except (OSError, ValueError, IndexError): return 0 def parse_proc_stat(text: str) -> tuple[str, int]: end = text.rfind(")") name = text[text.find("(") + 1:end] parts = text[end + 2:].split() utime = int(parts[11]) if len(parts) > 11 else 0 stime = int(parts[12]) if len(parts) > 12 else 0 return name, utime + stime def exe_name(pid_dir: Path, fallback: str) -> str: try: return os.path.basename(os.readlink(pid_dir / "exe")) except OSError: return fallback def is_build_group(name: str, cmdline: str, cgroup: str, build_allowlist: list[str]) -> bool: base = name.split("/")[-1] if "builds.slice" in cgroup: return True if base not in set(build_allowlist): return False if base == "node": return any(word in cmdline for word in (" build", " dev", "vite", "webpack", "rollup", "turbo")) if base in {"go", "bun"}: return any(word in cmdline for word in (" build", " test", " run")) return True def group_ticks(proc_root: Path, build_allowlist: list[str]) -> dict[str, dict[str, int | bool]]: groups: dict[str, dict[str, int | bool]] = {} try: entries = list(proc_root.iterdir()) except OSError: entries = [] for entry in entries: if not entry.name.isdigit(): continue try: fallback, ticks = parse_proc_stat((entry / "stat").read_text()) name = exe_name(entry, fallback) cmdline = (entry / "cmdline").read_text().replace("\0", " ") cgroup = (entry / "cgroup").read_text() except (OSError, ValueError): continue group = groups.setdefault(name, {"ticks": 0, "build": False}) group["ticks"] = int(group["ticks"]) + ticks group["build"] = bool(group["build"]) or is_build_group(name, cmdline, cgroup, build_allowlist) return groups def load_allowlist() -> list[str]: raw = os.environ.get("SM_BUILD_ALLOWLIST", "") if raw.strip(): return [item.strip() for item in raw.split(",") if item.strip()] return sorted(DEFAULT_BUILD) def snapshot(proc_root: Path, build_allowlist: list[str]) -> dict[str, object]: groups = group_ticks(proc_root, build_allowlist) return { "total": total_ticks(proc_root), "groups": {name: int(data["ticks"]) for name, data in groups.items()}, "build": {name: bool(data["build"]) for name, data in groups.items()}, } def load_previous(path: Path) -> dict[str, object] | None: try: data = json.loads(path.read_text()) if isinstance(data, dict): return data except (OSError, json.JSONDecodeError): pass return None def save_previous(path: Path, current: dict[str, object]) -> None: try: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp") tmp.write_text(json.dumps(current, separators=(",", ":")) + "\n") os.replace(tmp, path) except OSError: pass def collect(proc_root: Path, previous: dict[str, object] | None, build_allowlist: list[str]) -> tuple[dict[str, float], dict[str, object]]: current = snapshot(proc_root, build_allowlist) if not previous: return {"node_cpu_nonbuild_busy_percent": 0.0, "node_cpu_build_busy_percent": 0.0}, current total_delta = max(1, int(current["total"]) - int(previous.get("total", 0))) prev_groups = previous.get("groups", {}) if not isinstance(prev_groups, dict): prev_groups = {} build = 0 nonbuild = 0 current_groups = current["groups"] current_build = current["build"] if isinstance(current_groups, dict) and isinstance(current_build, dict): for name, ticks in current_groups.items(): delta = max(0, int(ticks) - int(prev_groups.get(name, 0))) if bool(current_build.get(name, False)): build = max(build, delta) else: nonbuild = max(nonbuild, delta) scale = 100.0 / total_delta return { "node_cpu_nonbuild_busy_percent": round(nonbuild * scale, 2), "node_cpu_build_busy_percent": round(build * scale, 2), }, current def render(metrics: dict[str, float]) -> str: lines = [ "# HELP node_cpu_nonbuild_busy_percent Busiest non-build process group CPU percentage using cpu_runaway.plugin classification.", "# TYPE node_cpu_nonbuild_busy_percent gauge", f"node_cpu_nonbuild_busy_percent {metrics['node_cpu_nonbuild_busy_percent']}", "# HELP node_cpu_build_busy_percent Busiest build process group CPU percentage using cpu_runaway.plugin classification.", "# TYPE node_cpu_build_busy_percent gauge", f"node_cpu_build_busy_percent {metrics['node_cpu_build_busy_percent']}", ] return "\n".join(lines) + "\n" def write_atomic(out_dir: Path, name: str, content: str) -> None: out_dir.mkdir(parents=True, exist_ok=True) tmp = out_dir / f"{name}.{os.getpid()}.tmp" tmp.write_text(content) os.replace(tmp, out_dir / name) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--output-dir", default=os.environ.get("NODE_TEXTFILE_DIR", "/var/lib/node_exporter/textfile")) parser.add_argument("--state-dir", default=os.environ.get("SM_STATE_DIR")) parser.add_argument("--proc-root", default=os.environ.get("SM_PROC_ROOT", "/proc")) parser.add_argument("--stdout", action="store_true") args = parser.parse_args() out_dir = Path(args.output_dir) state_dir = Path(args.state_dir) if args.state_dir else out_dir state_path = state_dir / STATE_FILE metrics, current = collect(Path(args.proc_root), load_previous(state_path), load_allowlist()) content = render(metrics) if args.stdout: sys.stdout.write(content) else: write_atomic(out_dir, METRIC_FILE, content) save_previous(state_path, current) return 0 if __name__ == "__main__": raise SystemExit(main())