diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index e884953480..2455be32a7 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -272,6 +272,10 @@ jobs: python3 scripts/workspace_architecture.py --self-test python3 scripts/workspace_architecture.py --check --print-summary + - name: Linux incident collector fixtures + if: ${{ !cancelled() }} + run: python3 scripts/test_capture_linux_incident.py + - name: Public benchmark evidence freshness if: ${{ !cancelled() }} run: | diff --git a/changelog.d/9947-linux-incident-capture.md b/changelog.d/9947-linux-incident-capture.md new file mode 100644 index 0000000000..31bafe3ffe --- /dev/null +++ b/changelog.d/9947-linux-incident-capture.md @@ -0,0 +1,4 @@ +Added an external Linux process incident collector for server hangs and memory +growth, preserving per-thread CPU, memory snapshots, and optional perf evidence +without depending on the affected event loop. This instruments issue #9942; +it does not claim to fix the reported leak or hang. diff --git a/docs/src/SUMMARY.md b/docs/src/SUMMARY.md index 335968a3cc..8400e3d587 100644 --- a/docs/src/SUMMARY.md +++ b/docs/src/SUMMARY.md @@ -157,6 +157,7 @@ - [CI Tiers (PR gate / sweep / full)](testing/ci-tiers.md) - [Claude Code Bundle Parity](testing/cc-parity.md) - [CI Gate Scheduling](testing/ci-gate-scheduling.md) +- [Linux Incident Capture](testing/linux-incident-capture.md) # CLI Reference diff --git a/docs/src/testing/linux-incident-capture.md b/docs/src/testing/linux-incident-capture.md new file mode 100644 index 0000000000..7f18faa58f --- /dev/null +++ b/docs/src/testing/linux-incident-capture.md @@ -0,0 +1,61 @@ +# Capturing a Linux server hang or memory slope + +Issue #9942 lost the spinning process before a user-space stack was captured. +`scripts/capture_linux_incident.py` runs outside the affected process, so it +works even when the JS event loop cannot run timers or signal callbacks. +It reads `/proc`; it does not restart or signal the server. + +Before restarting a wedged service, capture its current PID: + +```sh +python3 scripts/capture_linux_incident.py 3296296 \ + --output /tmp/perry-incident-20260907 --perf-seconds 10 +``` + +The output directory must not exist. It is created with owner-only access. +Use the service account (or an account permitted to inspect it). Missing +`smaps_rollup`, kernel stacks, or perf permissions are recorded explicitly; +available evidence is retained. `perf` is optional and is never installed by +the script. An unsuccessful requested perf recording makes the command fail +while preserving the `/proc` samples. + +For the steady growth reported in #9942, collect a healthy baseline and a later +sample under the same idle scheduler workload: + +```sh +python3 scripts/capture_linux_incident.py 3296296 \ + --output /tmp/perry-growth-20260907 --samples 61 --interval 1 +``` + +`summary.json` contains per-interval RSS growth and per-thread CPU percentages +(100% means one busy core), derived using the host clock tick and page sizes. +Thread IDs are compared with their start time so a recycled ID cannot produce +a false CPU delta. A replaced process ends the capture instead of mixing two +server lifetimes. Interrupting collection preserves the completed samples. + +Each sample retains `status`, `smaps_rollup`, `maps`, `io`, `limits`, and per-thread +`stat`, `wchan`, `stack`, and `schedstat`. `/proc/.../stack` is a **kernel** stack; +it cannot identify a spinning Rust/JS function. For that, inspect the optional +user-space profile alongside the exact deployed executable and matching symbols: + +```sh +perf report --stdio -i /tmp/perry-incident-20260907/perf.data +``` + +The default perf call chain uses frame pointers. Record whether the deployed +runtime and executable contain frame pointers/debug symbols if its stack is +incomplete. Keep the build commit, workload duration, and the last application +error beside the capture. Maps and profiles contain executable paths and +addresses; review the artifact before attaching it publicly. The collector does +not read the process environment, command line, request bodies, or application +heap. + +This is evidence collection, not a fix for #9942. RSS growth alone does not +distinguish retained live objects, allocator retention, or an off-heap leak. +The profile and memory breakdown are intended to choose the next reproducer. + +Collector tests (also runnable off Linux using synthetic `/proc` fixtures): + +```sh +python3 -m unittest discover -s scripts -p test_capture_linux_incident.py +``` diff --git a/scripts/capture_linux_incident.py b/scripts/capture_linux_incident.py new file mode 100644 index 0000000000..b83f99cc18 --- /dev/null +++ b/scripts/capture_linux_incident.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 +"""Capture a live Linux process without relying on its event loop (#9942).""" + +import argparse +import datetime +import json +import os +from pathlib import Path +import platform +import subprocess +import sys +import time + + +MAX_FILE_BYTES = 4 * 1024 * 1024 + + +def parse_stat(text): + # comm may contain spaces and closing parentheses; fields start after the + # final ')'. Index zero here is Linux stat field 3 (state). + if not isinstance(text, str): + raise ValueError("stat is unavailable") + end = text.rfind(")") + start = text.find("(") + if start < 0 or end <= start: + raise ValueError("malformed /proc stat comm") + fields = text[end + 1:].split() + return { + "comm": text[start + 1:end], + "state": fields[0], + "cpu_ticks": int(fields[11]) + int(fields[12]), + "start_ticks": int(fields[19]), + "virtual_bytes": int(fields[20]), + "rss_pages": int(fields[21]), + } + + +def read_file(path, errors): + try: + with path.open("rb") as source: + data = source.read(MAX_FILE_BYTES + 1) + if len(data) > MAX_FILE_BYTES: + errors.append(f"{path}: truncated at {MAX_FILE_BYTES} bytes") + return data[:MAX_FILE_BYTES].decode("utf-8", errors="replace") + except OSError as error: + errors.append(f"{path}: {error}") + return None + + +def identity(process): + return parse_stat((process / "stat").read_text())["start_ticks"] + + +def snapshot(process, expected_start, max_threads=256): + if identity(process) != expected_start: + raise RuntimeError("PID was reused; refusing to mix two processes") + result = {"monotonic_seconds": time.monotonic(), "files": {}, "threads": {}, "errors": []} + for name in ("stat", "status", "smaps_rollup", "io", "limits", "maps"): + result["files"][name] = read_file(process / name, result["errors"]) + tasks = sorted((process / "task").iterdir(), key=lambda path: int(path.name)) + result["threads_seen"] = len(tasks) + if len(tasks) > max_threads: + result["errors"].append(f"thread capture limited to {max_threads} of {len(tasks)} threads") + for task in tasks[:max_threads]: + result["threads"][task.name] = { + name: read_file(task / name, result["errors"]) + for name in ("stat", "wchan", "stack", "schedstat") + } + if identity(process) != expected_start: + raise RuntimeError("PID changed during capture; discarding this sample") + return result + + +def summarize(samples, ticks_per_second, page_size): + if len(samples) < 2: + return {"intervals": [], "note": "Need two samples to measure growth and CPU."} + intervals = [] + for before, after in zip(samples, samples[1:]): + elapsed = after["monotonic_seconds"] - before["monotonic_seconds"] + if elapsed <= 0: + continue + row = {"elapsed_seconds": elapsed, "threads": [], "errors": []} + try: + first = parse_stat(before["files"]["stat"]) + last = parse_stat(after["files"]["stat"]) + growth = (last["rss_pages"] - first["rss_pages"]) * page_size + row.update(rss_bytes=last["rss_pages"] * page_size, + rss_delta_bytes=growth, rss_bytes_per_second=growth / elapsed) + except (TypeError, ValueError, IndexError) as error: + row["errors"].append(f"process stat unavailable: {error}") + for tid, thread in after["threads"].items(): + previous = before["threads"].get(tid) + if previous is None: + continue + try: + first, last = parse_stat(previous["stat"]), parse_stat(thread["stat"]) + if first["start_ticks"] != last["start_ticks"]: + continue # A recycled thread ID is not a CPU delta. + cpu_seconds = (last["cpu_ticks"] - first["cpu_ticks"]) / ticks_per_second + if cpu_seconds < 0: + continue + row["threads"].append({"tid": int(tid), "comm": last["comm"], + "state": last["state"], "cpu_seconds": cpu_seconds, + "cpu_percent_one_core": 100 * cpu_seconds / elapsed, + "wchan": thread["wchan"]}) + except (TypeError, ValueError, IndexError) as error: + row["errors"].append(f"thread {tid} stat unavailable: {error}") + row["threads"].sort(key=lambda item: item["cpu_seconds"], reverse=True) + intervals.append(row) + return {"intervals": intervals} + + +def write_json(path, value): + path.write_text(json.dumps(value, indent=2) + "\n") + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("pid", type=int) + parser.add_argument("--output", required=True, type=Path, help="new directory; never overwrites a capture") + parser.add_argument("--samples", type=int, default=6) + parser.add_argument("--interval", type=float, default=2) + parser.add_argument("--max-threads", type=int, default=256) + parser.add_argument("--perf-seconds", type=int, default=0, + help="optionally record user-space call chains with perf for 1–60 seconds") + args = parser.parse_args(argv) + if platform.system() != "Linux": + parser.error("this collector requires Linux /proc") + if args.pid <= 0 or not 2 <= args.samples <= 3600 or not 0.1 <= args.interval <= 60: + parser.error("require positive PID, 2–3600 samples and interval 0.1–60 seconds") + if not 1 <= args.max_threads <= 4096 or not 0 <= args.perf_seconds <= 60: + parser.error("require max-threads 1–4096 and perf-seconds 0–60") + process = Path("/proc") / str(args.pid) + try: + expected_start = identity(process) + args.output.mkdir(mode=0o700, parents=False, exist_ok=False) + except (OSError, ValueError, IndexError) as error: + parser.error(str(error)) + metadata = {"pid": args.pid, "start_ticks": expected_start, + "captured_at_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "kernel": platform.release(), "machine": platform.machine(), + "ticks_per_second": os.sysconf("SC_CLK_TCK"), + "page_size": os.sysconf("SC_PAGE_SIZE"), "errors": []} + try: + metadata["executable"] = os.readlink(process / "exe") + except OSError as error: + metadata["errors"].append(str(error)) + samples = [] + perf = None + perf_log = None + try: + if args.perf_seconds: + command = ["perf", "record", "-F", "99", "-g", "-p", str(args.pid), + "-o", str(args.output / "perf.data"), "--", "sleep", str(args.perf_seconds)] + metadata["perf_command"] = command + perf_log = (args.output / "perf.log").open("w") + try: + perf = subprocess.Popen(command, stdout=perf_log, stderr=perf_log) + except OSError as error: + metadata["errors"].append(f"perf unavailable: {error}") + for index in range(args.samples): + try: + sample = snapshot(process, expected_start, args.max_threads) + except (OSError, RuntimeError, ValueError, IndexError) as error: + metadata["errors"].append(str(error)) + break + samples.append(sample) + write_json(args.output / f"sample-{index:04d}.json", sample) + if index + 1 < args.samples: + time.sleep(args.interval) + if perf is not None: + try: + metadata["perf_exit_code"] = perf.wait(timeout=args.perf_seconds + 5) + except subprocess.TimeoutExpired: + metadata["errors"].append("perf exceeded its deadline") + except KeyboardInterrupt: + metadata["errors"].append("capture interrupted; partial samples preserved") + finally: + if perf is not None and perf.poll() is None: + # Stop only the collector we launched, never the observed process. + perf.terminate() + try: + perf.wait(timeout=5) + except subprocess.TimeoutExpired: + perf.kill() + perf.wait() + if perf_log is not None: + perf_log.close() + metadata["samples_captured"] = len(samples) + metadata["samples_requested"] = args.samples + write_json(args.output / "metadata.json", metadata) + write_json(args.output / "summary.json", summarize( + samples, metadata["ticks_per_second"], metadata["page_size"])) + print(f"Captured {len(samples)}/{args.samples} samples in {args.output}") + if args.perf_seconds: + print("User-space stacks: inspect perf.log, then perf report --stdio -i /perf.data") + return 0 if len(samples) == args.samples and not metadata["errors"] and metadata.get("perf_exit_code", 0) == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_capture_linux_incident.py b/scripts/test_capture_linux_incident.py new file mode 100644 index 0000000000..632fa5374f --- /dev/null +++ b/scripts/test_capture_linux_incident.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +"""Deterministic /proc fixtures for the incident collector.""" +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import capture_linux_incident as capture + + +def stat(cpu=10, start=123, rss=100, comm="server (worker) name"): + fields = ["0"] * 22 + fields[0], fields[11], fields[12] = "R", str(cpu), "2" + fields[19], fields[20], fields[21] = str(start), "8192000", str(rss) + return "42 (" + comm + ") " + " ".join(fields) + + +def sample(at, cpu, rss, start=123): + return {"monotonic_seconds": at, "files": {"stat": stat(cpu, rss=rss)}, + "threads": {"42": {"stat": stat(cpu, start), "wchan": "0"}}} + + +class CaptureTests(unittest.TestCase): + def test_stat_with_parentheses_and_spaces(self): + parsed = capture.parse_stat(stat()) + self.assertEqual(parsed["comm"], "server (worker) name") + self.assertEqual(parsed["cpu_ticks"], 12) + self.assertEqual(parsed["start_ticks"], 123) + self.assertEqual(parsed["rss_pages"], 100) + + def test_cpu_and_growth_use_actual_time_ticks_and_page_size(self): + result = capture.summarize([sample(10, 0, 100), sample(12, 200, 110)], 100, 4096) + row = result["intervals"][0] + self.assertEqual(row["rss_delta_bytes"], 40960) + self.assertEqual(row["rss_bytes_per_second"], 20480) + self.assertEqual(row["threads"][0]["cpu_percent_one_core"], 100) + + def test_reused_tid_is_not_reported_as_cpu(self): + result = capture.summarize([sample(10, 0, 100), sample(12, 200, 110, start=456)], 100, 4096) + self.assertEqual(result["intervals"][0]["threads"], []) + + def test_missing_thread_stat_preserves_memory_summary(self): + before, after = sample(10, 0, 100), sample(12, 200, 110) + after["threads"]["42"]["stat"] = None + result = capture.summarize([before, after], 100, 4096)["intervals"][0] + self.assertEqual(result["rss_delta_bytes"], 40960) + self.assertTrue(result["errors"]) + + def test_partial_snapshot_records_missing_files(self): + with tempfile.TemporaryDirectory() as directory: + process = Path(directory) + (process / "stat").write_text(stat()) + task = process / "task" / "42" + task.mkdir(parents=True) + (task / "stat").write_text(stat()) + result = capture.snapshot(process, 123) + self.assertEqual(result["threads_seen"], 1) + self.assertIsNone(result["files"]["smaps_rollup"]) + self.assertTrue(result["errors"]) + + def test_pid_reuse_before_or_during_capture_is_rejected(self): + with patch.object(capture, "identity", return_value=456): + with self.assertRaisesRegex(RuntimeError, "reused"): + capture.snapshot(Path("/unused"), 123) + with tempfile.TemporaryDirectory() as directory: + process = Path(directory) + (process / "task").mkdir() + with patch.object(capture, "identity", side_effect=[123, 456]): + with self.assertRaisesRegex(RuntimeError, "changed"): + capture.snapshot(process, 123) + + def test_read_is_bounded(self): + with tempfile.TemporaryDirectory() as directory: + source = Path(directory) / "maps" + source.write_text("x" * 50) + errors = [] + with patch.object(capture, "MAX_FILE_BYTES", 10): + self.assertEqual(capture.read_file(source, errors), "x" * 10) + self.assertIn("truncated", errors[0]) + + +if __name__ == "__main__": + unittest.main()