diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml index 0d9aac66e7a5..83bcd6fadbd1 100644 --- a/.github/workflows/backend.yml +++ b/.github/workflows/backend.yml @@ -203,7 +203,7 @@ jobs: needs.files-changed.outputs.backend == 'true' && needs.calculate-shards.outputs.shard-count != '0' needs: [files-changed, select-tests, calculate-shards] name: backend test - runs-on: ubuntu-24.04 + runs-on: ubuntu-26.04 timeout-minutes: 60 permissions: contents: read @@ -224,11 +224,16 @@ jobs: TEST_GROUP_STRATEGY: roundrobin PYTHONHASHSEED: '0' XDIST_PER_WORKER_SNUBA: '1' - XDIST_WORKERS: ${{ github.event_name == 'pull_request' && '2' || '3' }} + XDIST_WORKERS: '4' + # Set to '1' to sample pytest and its xdist workers with py-spy (adds ~0.4 cores of overhead). + PROFILE_BACKEND_TESTS: '0' steps: - uses: actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332 # v4.1.7 + - name: Start resource monitor + run: python3 .github/workflows/scripts/monitor-resources.py --start + # Overlaps the per-worker ClickHouse bootstrap with devservices up. - name: Start per-worker Snuba bootstrap in the background if: env.XDIST_PER_WORKER_SNUBA == '1' @@ -258,7 +263,8 @@ jobs: -v /tmp/pg-sock:/var/run/postgresql \ -e POSTGRES_HOST_AUTH_METHOD=trust -e POSTGRES_DB=sentry \ "$PG_IMAGE" postgres \ - -c wal_level=logical -c max_replication_slots=1 -c max_wal_senders=1 + -c wal_level=logical -c max_replication_slots=1 -c max_wal_senders=1 \ + -c fsync=off -c synchronous_commit=off -c full_page_writes=off # docker run -d returns 0 even if postgres dies right after — bound the # wait so a crashing postgres surfaces fast instead of hitting the job # timeout silently. @@ -284,6 +290,27 @@ jobs: if: env.XDIST_PER_WORKER_SNUBA == '1' run: python3 .github/workflows/scripts/bootstrap-snuba.py --wait + # sentry-shared-kafka's healthcheck runs `kafka-topics --list` every 5s, which starts + # a JVM each time: about 0.25 of the runner's 4 cores during tests. Everything that + # waits on Kafka's health is done by now, so answer that exact command with a TCP + # probe instead (kafka-topics-healthcheck.sh). + - name: Make the Kafka healthcheck cheap + continue-on-error: true + run: | + real=$(docker exec kafka-kafka-1 sh -c 'command -v kafka-topics') + docker exec -u 0 kafka-kafka-1 mv "$real" "$real.real" + docker cp .github/workflows/scripts/kafka-topics-healthcheck.sh "kafka-kafka-1:$real" + time docker exec kafka-kafka-1 kafka-topics --bootstrap-server 127.0.0.1:9092 --list + + - name: Install py-spy + if: env.PROFILE_BACKEND_TESTS == '1' + run: | + uv tool install py-spy==0.4.1 + echo "PY_SPY=$(uv tool dir --bin)/py-spy" >> "$GITHUB_ENV" + echo "PY_SPY_OUTPUT=$PWD/.artifacts/py-spy.txt" >> "$GITHUB_ENV" + # py-spy attaches to the running pytest controller, which isn't its child. + sudo sysctl -w kernel.yama.ptrace_scope=0 + - name: Download selected tests artifact if: needs.select-tests.outputs.has-selected-tests == 'true' uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0 @@ -297,7 +324,11 @@ jobs: run: | if [ -n "${XDIST_WORKERS}" ]; then export PYTEST_ADDOPTS="$PYTEST_ADDOPTS -n ${XDIST_WORKERS} --dist=loadfile" - timeout -k 30 1200 make test-python-ci || { + profile=() + if [ -n "${PY_SPY:-}" ]; then + profile=(.github/workflows/scripts/profile-backend-tests.sh) + fi + "${profile[@]}" timeout -k 30 1200 make test-python-ci || { rc=$? if [ "$rc" -eq 124 ]; then echo "::error::Test run timed out after 20 minutes (possible xdist hang)" @@ -308,6 +339,11 @@ jobs: make test-python-ci fi + - name: Stop resource monitor + if: always() + continue-on-error: true + run: python3 .github/workflows/scripts/monitor-resources.py --stop + # pytest-selected.py balances full-run shards with these. See # .github/workflows/scripts/backend-test-durations.py to refresh them. - name: Extract test durations @@ -327,6 +363,18 @@ jobs: path: .artifacts/test-durations.json retention-days: 30 + - name: Upload resource usage + if: always() + continue-on-error: true + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: backend-resource-usage-${{ github.run_id }}-${{ matrix.instance }} + path: | + .artifacts/resource-usage.jsonl + .artifacts/resource-usage-summary.json + .artifacts/py-spy.txt + retention-days: 30 + - name: Report failures if: ${{ !cancelled() && github.event_name == 'pull_request' }} continue-on-error: true diff --git a/.github/workflows/scripts/kafka-topics-healthcheck.sh b/.github/workflows/scripts/kafka-topics-healthcheck.sh new file mode 100755 index 000000000000..15013d589dd0 --- /dev/null +++ b/.github/workflows/scripts/kafka-topics-healthcheck.sh @@ -0,0 +1,9 @@ +#!/bin/bash +# Stands in for kafka-topics inside the CI Kafka container (see backend.yml). +# The healthcheck's exact command gets a TCP probe instead of a JVM start; +# anything else runs the real kafka-topics. +if [ "$*" = "--bootstrap-server 127.0.0.1:9092 --list" ] && + (exec 3<>/dev/tcp/127.0.0.1/9092) 2>/dev/null; then + exit 0 +fi +exec "$0.real" "$@" diff --git a/.github/workflows/scripts/monitor-resources.py b/.github/workflows/scripts/monitor-resources.py new file mode 100644 index 000000000000..423e08051d57 --- /dev/null +++ b/.github/workflows/scripts/monitor-resources.py @@ -0,0 +1,542 @@ +#!/usr/bin/env python3 +"""Sample CPU/memory during backend CI shards to judge xdist headroom. + +Starts a background sampler before the test run, stops it after, and uploads +per-shard usage as an artifact. To check whether one more xdist worker fits, +download the artifacts and compare peak CPU and memory against runner limits: + + gh run download RUN_ID --repo getsentry/sentry \\ + --pattern 'backend-resource-usage-*' --dir /tmp/usage + python3 -c "import json,glob; ..." + +Each shard writes `.artifacts/resource-usage.jsonl` (one sample per line) and +`.artifacts/resource-usage-summary.json` (aggregates plus a headroom verdict). + +Each sample also attributes CPU to process groups: docker containers by name, +xdist workers, the pytest controller, and other host processes by command name. +CPU that exited between samples shows up as `unattributed`. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import statistics +import subprocess +import sys +import time +from pathlib import Path + +INTERVAL_S = 5.0 +STOP_TIMEOUT_S = 30 + +STATE_DIR = Path(os.environ.get("RUNNER_TEMP", "/tmp")) / "resource-monitor" +PID_FILE = STATE_DIR / "pid" +STOP_FILE = STATE_DIR / "stop" +LOG_FILE = STATE_DIR / "log" + +CPU_HEADROOM_P95 = 75.0 +CPU_HEADROOM_MAX = 90.0 +CPU_MARGINAL_P95 = 85.0 +MEM_HEADROOM_USED_PCT = 90.0 +MEM_HEADROOM_MIN_AVAIL_GB = 1.0 + +CPU_MODES = ("user", "nice", "system", "idle", "iowait", "irq", "softirq", "steal") +CLK_TCK = os.sysconf("SC_CLK_TCK") +DOCKER_ID_RE = re.compile(r"docker[-/]([0-9a-f]{64})") +# execnet starts xdist workers as `python -c "import sys;exec(eval(sys.stdin.readline()))"`. +XDIST_WORKER_MARKER = "exec(eval(sys.stdin.readline()))" +CONTAINER_REFRESH_S = 5.0 + + +def read_cpu_times() -> dict[str, int] | None: + try: + with open("/proc/stat") as f: + parts = f.readline().split() + if not parts or parts[0] != "cpu": + return None + return dict(zip(CPU_MODES, (int(v) for v in parts[1:]))) + except (OSError, ValueError): + return None + + +def cpu_percent(prev: dict[str, int] | None, curr: dict[str, int] | None) -> float | None: + if prev is None or curr is None: + return None + total_delta = sum(curr.values()) - sum(prev.values()) + if total_delta <= 0: + return None + idle_delta = curr["idle"] + curr["iowait"] - prev["idle"] - prev["iowait"] + return round(100.0 * (1.0 - idle_delta / total_delta), 1) + + +def cpu_mode_percents(prev: dict[str, int], curr: dict[str, int]) -> dict[str, float]: + total_delta = sum(curr.values()) - sum(prev.values()) + if total_delta <= 0: + return {} + return { + mode: round(100.0 * (curr[mode] - prev[mode]) / total_delta, 1) + for mode in ("user", "system", "iowait", "softirq", "steal") + } + + +def busy_ticks(prev: dict[str, int], curr: dict[str, int]) -> int: + return sum(curr[m] - prev[m] for m in CPU_MODES if m not in ("idle", "iowait")) + + +def read_text(path: str) -> str: + try: + with open(path, "rb") as f: + return f.read().decode(errors="replace") + except OSError: + return "" + + +def cpu_model(cpuinfo: str = "/proc/cpuinfo") -> str | None: + """Runner CPUs vary a lot in speed, so record which one a shard got.""" + for line in read_text(cpuinfo).splitlines(): + key, _, value = line.partition(":") + if key.strip() == "model name": + return value.strip() + return None + + +def read_proc_stat(pid: str) -> tuple[str, int, int, int] | None: + """Returns (comm, cpu ticks, reaped children's cpu ticks, start time) for a pid.""" + data = read_text(f"/proc/{pid}/stat") + if ")" not in data: + return None + comm = data[data.index("(") + 1 : data.rindex(")")] + fields = data[data.rindex(")") + 2 :].split() + try: + return ( + comm, + int(fields[11]) + int(fields[12]), + int(fields[13]) + int(fields[14]), + int(fields[19]), + ) + except (IndexError, ValueError): + return None + + +def read_cgroup_cpu_s(cgroup: str) -> float | None: + for line in read_text(f"/sys/fs/cgroup{cgroup}/cpu.stat").splitlines(): + key, _, value = line.partition(" ") + if key == "usage_usec": + return int(value) / 1e6 + return None + + +def docker_container_names() -> dict[str, str]: + try: + r = subprocess.run( + ["docker", "ps", "-a", "--no-trunc", "--format", "{{.ID}} {{.Names}}"], + capture_output=True, + text=True, + timeout=10, + ) + except (OSError, subprocess.TimeoutExpired): + return {} + return dict(line.split(" ", 1) for line in r.stdout.splitlines() if " " in line) + + +def classify_process(cmdline: str, comm: str) -> str: + if not cmdline: + return "kernel" + if XDIST_WORKER_MARKER in cmdline: + return "pytest:worker" + if "pytest" in cmdline: + return "pytest:main" + if "monitor-resources.py" in cmdline: + return "monitor" + return comm + + +class ProcessCpuTracker: + """Attributes CPU between samples to process groups. + + Live processes are attributed by their own counters. CPU of processes that + exited between samples is recovered from their cgroup's cpu.stat and + reported as `exited:`. + """ + + def __init__(self) -> None: + self.prev: dict[tuple[str, int], tuple[int, int]] = {} + self.labels: dict[tuple[str, int], str] = {} + self.cgroups: dict[tuple[str, int], str] = {} + self.cgroup_cpu_s: dict[str, float] = {} + self.containers: dict[str, str] = {} + self.containers_refreshed = float("-inf") + self.primed = False + + def container_name(self, container_id: str) -> str | None: + name = self.containers.get(container_id) + if name is None and time.monotonic() - self.containers_refreshed > CONTAINER_REFRESH_S: + self.containers = docker_container_names() + self.containers_refreshed = time.monotonic() + name = self.containers.get(container_id) + return name + + def cgroup_label(self, cgroup: str) -> tuple[str, bool]: + """Returns (label, final). Unresolved container ids are retried later.""" + if m := DOCKER_ID_RE.search(cgroup): + name = self.container_name(m.group(1)) + return f"docker:{name or m.group(1)[:12]}", name is not None + return cgroup, True + + def label(self, key: tuple[str, int], comm: str) -> str: + if (label := self.labels.get(key)) is not None: + return label + label, final = self.cgroup_label(self.cgroups[key]) + if not label.startswith("docker:"): + cmdline = read_text(f"/proc/{key[0]}/cmdline").replace("\0", " ").strip() + label = classify_process(cmdline, comm) + if final: + self.labels[key] = label + return label + + def sample(self) -> tuple[dict[str, float], dict[str, float]]: + """Returns CPU-seconds since the previous call, by group and by the + group of the parent that reaped the children.""" + current: dict[tuple[str, int], tuple[int, int]] = {} + by_group: dict[str, float] = {} + children: dict[str, float] = {} + live_by_cgroup: dict[str, float] = {} + for pid in os.listdir("/proc"): + if not pid.isdigit() or (stat := read_proc_stat(pid)) is None: + continue + comm, ticks, child_ticks, start = stat + key = (pid, start) + current[key] = (ticks, child_ticks) + if key not in self.cgroups: + self.cgroups[key] = read_text(f"/proc/{pid}/cgroup").partition("::")[2].strip() + if not self.primed: + continue + prev_ticks, prev_child_ticks = self.prev.get(key, (0, child_ticks)) + delta = (ticks - prev_ticks) / CLK_TCK + child_delta = (child_ticks - prev_child_ticks) / CLK_TCK + if delta <= 0 and child_delta <= 0: + continue + label = self.label(key, comm) + if delta > 0: + by_group[label] = by_group.get(label, 0.0) + delta + cgroup = self.cgroups[key] + live_by_cgroup[cgroup] = live_by_cgroup.get(cgroup, 0.0) + delta + if child_delta > 0: + children[label] = children.get(label, 0.0) + child_delta + + cgroups = {cg for k, cg in self.cgroups.items() if k in current} | set(self.cgroup_cpu_s) + # A parent's cpu.stat includes its children; only leaves are exact. + leaves = { + cg + for cg in cgroups + if cg not in ("", "/") and not any(o.startswith(cg + "/") for o in cgroups) + } + cgroup_cpu_s: dict[str, float] = {} + for cg in leaves: + if (usage := read_cgroup_cpu_s(cg)) is None: + continue + cgroup_cpu_s[cg] = usage + if (prev_usage := self.cgroup_cpu_s.get(cg)) is not None: + exited = usage - prev_usage - live_by_cgroup.get(cg, 0.0) + if exited > 0.01: + group = f"exited:{self.cgroup_label(cg)[0]}" + by_group[group] = by_group.get(group, 0.0) + exited + + self.labels = {k: v for k, v in self.labels.items() if k in current} + self.cgroups = {k: v for k, v in self.cgroups.items() if k in current} + self.prev = current + self.cgroup_cpu_s = cgroup_cpu_s + self.primed = True + return by_group, children + + +def read_mem() -> dict[str, float] | None: + try: + info: dict[str, int] = {} + with open("/proc/meminfo") as f: + for line in f: + key, _, rest = line.partition(":") + info[key.strip()] = int(rest.strip().split()[0]) + total = info.get("MemTotal", 0) + avail = info.get("MemAvailable", info.get("MemFree", 0)) + if not total: + return None + swap_total = info.get("SwapTotal", 0) + swap_free = info.get("SwapFree", 0) + return { + "total_kb": float(total), + "available_kb": float(avail), + "used_pct": round(100.0 * (total - avail) / total, 1), + "swap_used_kb": float(swap_total - swap_free), + } + except (OSError, ValueError): + return None + + +def read_load() -> list[float] | None: + try: + with open("/proc/loadavg") as f: + return [float(v) for v in f.read().split()[:3]] + except (OSError, ValueError): + return None + + +def read_pressure(path: str) -> dict[str, float] | None: + """Parses the `some` line of a PSI file. The `full` line is ignored.""" + try: + with open(path) as f: + for line in f: + kind, *tokens = line.split() + if kind == "some": + return {k: float(v) for k, v in (t.split("=", 1) for t in tokens)} + except (OSError, ValueError): + pass + return None + + +def sample( + prev_cpu: dict[str, int] | None, tracker: ProcessCpuTracker | None = None +) -> tuple[dict, dict[str, int] | None]: + curr_cpu = read_cpu_times() + mem = read_mem() + load = read_load() + entry: dict = {"ts": round(time.time(), 1)} + pct = cpu_percent(prev_cpu, curr_cpu) + if pct is not None: + entry["cpu_pct"] = pct + if (cores := os.cpu_count()) is not None: + entry["cpu_count"] = cores + if mem is not None: + entry["mem_used_pct"] = mem["used_pct"] + entry["mem_available_gb"] = round(mem["available_kb"] / 1_048_576, 2) + entry["swap_used_mb"] = round(mem["swap_used_kb"] / 1024, 1) + if load is not None: + entry["load_1m"], entry["load_5m"], entry["load_15m"] = (round(v, 2) for v in load) + for name in ("cpu", "memory"): + if (p := read_pressure(f"/proc/pressure/{name}")) is not None: + entry[f"pressure_{name}_some_avg10"] = p.get("avg10") + if prev_cpu is not None and curr_cpu is not None: + entry["cpu_modes_pct"] = cpu_mode_percents(prev_cpu, curr_cpu) + if tracker is not None: + groups, children = tracker.sample() + unattributed = busy_ticks(prev_cpu, curr_cpu) / CLK_TCK - sum(groups.values()) + if unattributed > 0: + groups["unattributed"] = unattributed + # CPU-seconds per group since the previous sample. + entry["cpu_s_by_group"] = rounded_by_value(groups) + # CPU-seconds of exited children, by the group of the parent that reaped them. + # Overlaps cpu_s_by_group for children that were alive at an earlier sample. + entry["children_cpu_s_by_parent"] = rounded_by_value(children) + return entry, curr_cpu + + +def percentile(values: list[float], pct: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + idx = min(len(ordered) - 1, int(pct / 100.0 * len(ordered))) + return ordered[idx] + + +def rounded_by_value(d: dict[str, float]) -> dict[str, float]: + return {k: round(v, 2) for k, v in sorted(d.items(), key=lambda kv: -kv[1]) if v >= 0.005} + + +def cpu_s_by_group(samples: list[dict]) -> dict[str, float]: + totals: dict[str, float] = {} + for s in samples: + for group, cpu_s in s.get("cpu_s_by_group", {}).items(): + totals[group] = totals.get(group, 0.0) + cpu_s + return {k: round(v, 1) for k, v in sorted(totals.items(), key=lambda kv: -kv[1])} + + +def summarize(samples: list[dict]) -> dict: + cpus = [s["cpu_pct"] for s in samples if "cpu_pct" in s] + mems = [s["mem_used_pct"] for s in samples if "mem_used_pct" in s] + avails = [s["mem_available_gb"] for s in samples if "mem_available_gb" in s] + swaps = [s["swap_used_mb"] for s in samples if "swap_used_mb" in s] + loads = [s["load_1m"] for s in samples if "load_1m" in s] + counts = {s["cpu_count"] for s in samples if "cpu_count" in s} + + cpu_p95 = round(percentile(cpus, 95), 1) + cpu_max = round(max(cpus), 1) if cpus else 0.0 + cpu_avg = round(statistics.fmean(cpus), 1) if cpus else 0.0 + mem_max = round(max(mems), 1) if mems else 0.0 + avail_min = round(min(avails), 2) if avails else 0.0 + swap_max = round(max(swaps), 1) if swaps else 0.0 + + if not cpus: + cpu_verdict: str = "unknown" + elif cpu_p95 < CPU_HEADROOM_P95 and cpu_max < CPU_HEADROOM_MAX: + cpu_verdict = "headroom" + elif cpu_p95 < CPU_MARGINAL_P95: + cpu_verdict = "marginal" + else: + cpu_verdict = "saturated" + + if not mems: + mem_verdict = "unknown" + elif mem_max < MEM_HEADROOM_USED_PCT and avail_min > MEM_HEADROOM_MIN_AVAIL_GB: + mem_verdict = "headroom" + else: + mem_verdict = "saturated" + + if cpu_verdict == "headroom" and mem_verdict == "headroom": + overall = "likely headroom for +1 worker" + elif "saturated" in (cpu_verdict, mem_verdict): + overall = "no headroom for +1 worker" + elif "unknown" in (cpu_verdict, mem_verdict): + overall = "inconclusive (no samples)" + else: + overall = "marginal for +1 worker" + + summary = { + "samples": len(samples), + "duration_s": round(samples[-1]["ts"] - samples[0]["ts"], 1) if len(samples) > 1 else 0.0, + "cpu_count": next(iter(counts), None), + "cpu": {"avg_pct": cpu_avg, "p95_pct": cpu_p95, "max_pct": cpu_max}, + "memory": { + "max_used_pct": mem_max, + "min_available_gb": avail_min, + "max_swap_used_mb": swap_max, + }, + "load_1m_max": round(max(loads), 2) if loads else None, + "cpu_s_by_group": cpu_s_by_group(samples), + # Samples where xdist workers used CPU, i.e. the test run itself. + "cpu_s_by_group_during_tests": cpu_s_by_group( + [s for s in samples if "pytest:worker" in s.get("cpu_s_by_group", {})] + ), + "headroom": { + "cpu": cpu_verdict, + "memory": mem_verdict, + "one_more_worker": overall, + }, + } + return summary + + +def run_sampler(output: Path, interval: float) -> None: + output.parent.mkdir(parents=True, exist_ok=True) + tracker = ProcessCpuTracker() + tracker.sample() + prev_cpu = read_cpu_times() + with open(output, "a") as f: + while not STOP_FILE.exists(): + time.sleep(interval) + entry, prev_cpu = sample(prev_cpu, tracker) + if len(entry) > 1: + f.write(json.dumps(entry) + "\n") + f.flush() + + +def start_background(output: Path, interval: float) -> None: + STATE_DIR.mkdir(parents=True, exist_ok=True) + for stale in (PID_FILE, STOP_FILE): + try: + stale.unlink() + except OSError: + pass + output.parent.mkdir(parents=True, exist_ok=True) + with open(LOG_FILE, "w") as log_file: + proc = subprocess.Popen( + [ + sys.executable, + __file__, + "--run-background", + "--output", + str(output), + "--interval", + str(interval), + ], + stdin=subprocess.DEVNULL, + stdout=log_file, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + PID_FILE.write_text(str(proc.pid)) + print(f"Started resource monitor (pid {proc.pid}), output: {output}") + + +def run_background(output: Path, interval: float) -> None: + rc = 1 + try: + run_sampler(output, interval) + rc = 0 + finally: + PID_FILE.write_text(f"{rc}\n") + + +def stop_background(output: Path, summary_path: Path | None) -> int: + STOP_FILE.touch(exist_ok=True) + deadline = time.monotonic() + STOP_TIMEOUT_S + while time.monotonic() < deadline: + if PID_FILE.exists(): + try: + if int(PID_FILE.read_text().strip()) == 0: + break + except ValueError: + pass + time.sleep(1) + + samples: list[dict] = [] + if output.exists(): + for line in output.read_text().splitlines(): + try: + samples.append(json.loads(line)) + except json.JSONDecodeError: + continue + summary = summarize(samples) + summary["cpu_model"] = cpu_model() + if summary_path is not None: + summary_path.parent.mkdir(parents=True, exist_ok=True) + summary_path.write_text(json.dumps(summary, indent=2) + "\n") + + line = ( + f"samples={summary['samples']} duration_s={summary['duration_s']} " + f"cpu_avg={summary['cpu']['avg_pct']}% cpu_p95={summary['cpu']['p95_pct']}% " + f"cpu_max={summary['cpu']['max_pct']}% " + f"mem_max={summary['memory']['max_used_pct']}% " + f"mem_avail_min={summary['memory']['min_available_gb']}GB " + f"swap_max={summary['memory']['max_swap_used_mb']}MB " + f"verdict={summary['headroom']['one_more_worker']} " + f"cpu_model={summary['cpu_model']}" + ) + print(line) + top = list(summary["cpu_s_by_group_during_tests"].items())[:10] + print("cpu-seconds during tests: " + ", ".join(f"{k}={v}" for k, v in top)) + if step_summary := os.environ.get("GITHUB_STEP_SUMMARY"): + with open(step_summary, "a") as f: + f.write(f"### Resource usage ({summary['samples']} samples)\n\n") + f.write(f"{line}\n\n") + f.write(f"```json\n{json.dumps(summary, indent=2)}\n```\n") + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--output", type=Path, default=Path(".artifacts/resource-usage.jsonl")) + parser.add_argument( + "--summary", type=Path, default=Path(".artifacts/resource-usage-summary.json") + ) + parser.add_argument("--interval", type=float, default=INTERVAL_S) + parser.add_argument("--start", action="store_true") + parser.add_argument("--stop", action="store_true") + parser.add_argument("--run-background", action="store_true") + args = parser.parse_args() + + if args.start: + start_background(args.output, args.interval) + elif args.run_background: + run_background(args.output, args.interval) + elif args.stop: + stop_background(args.output, args.summary) + else: + parser.error("expected one of --start, --stop, --run-background") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.github/workflows/scripts/prewarm_assert_rewrite.py b/.github/workflows/scripts/prewarm_assert_rewrite.py new file mode 100644 index 000000000000..672ce84a228e --- /dev/null +++ b/.github/workflows/scripts/prewarm_assert_rewrite.py @@ -0,0 +1,75 @@ +"""Write pytest's assertion-rewritten pycs for a shard's test files. + +Every xdist worker imports every test file in its shard, and they all start +together, so without this each worker rewrites and compiles the same files. +pytest-selected.py starts this in the background just before pytest, while the +workers are still starting up. Workers that import a file after it is done +load the cached pyc; pytest writes pycs atomically and validates them against +the source, so a worker that gets there first just rewrites it itself. + +Usage: prewarm_assert_rewrite.py FILE_LIST +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +from _pytest.assertion.rewrite import PYC_TAIL, _read_pyc, _rewrite_test, _write_pyc + + +class _State: + def trace(self, message: str) -> None: + pass + + +# pyproject.toml's python_files makes pytest rewrite these too, and test modules import them. +TESTUTILS = Path("src/sentry/testutils") + + +def rewrite_targets(files: list[str]) -> list[Path]: + """Test modules in `files` plus the conftests and testutils pytest rewrites + for them, roughly in import order.""" + targets: dict[Path, None] = dict.fromkeys(sorted(TESTUTILS.rglob("*.py"))) + for name in files: + path = Path(name) + # Conftests along the path load before the module itself. + for parent in reversed(path.parents): + if (conftest := parent / "conftest.py").is_file(): + targets[conftest] = None + if path.name.startswith("test_") and path.suffix == ".py": + targets[path] = None + return list(targets) + + +def prewarm(path: Path) -> None: + if sys.dont_write_bytecode: + return + # pytest compiles with the module's spec origin, an absolute (not resolved) + # path, and execs the cached code without fixing co_filename, so this has + # to match exactly or tracebacks show a different filename. + path = path.absolute() + cache_dir = path.parent / "__pycache__" + cache_dir.mkdir(exist_ok=True) + pyc = cache_dir / (path.name[:-3] + PYC_TAIL) + state = _State() + if _read_pyc(path, pyc, state.trace) is not None: + return + # A config is only needed for the enable_assertion_pass_hook ini option, + # which Sentry doesn't set. + source_stat, co = _rewrite_test(path, None) # type: ignore[arg-type] + _write_pyc(state, co, source_stat, pyc) # type: ignore[arg-type] + + +def main(file_list: str) -> int: + files = Path(file_list).read_text().split() + for path in rewrite_targets(files): + try: + prewarm(path) + except (OSError, SyntaxError): + pass + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1])) diff --git a/.github/workflows/scripts/profile-backend-tests.sh b/.github/workflows/scripts/profile-backend-tests.sh new file mode 100755 index 000000000000..e7718d4ddfb5 --- /dev/null +++ b/.github/workflows/scripts/profile-backend-tests.sh @@ -0,0 +1,54 @@ +#!/usr/bin/env bash +# Runs a command and samples its pytest processes (controller and xdist workers) +# with py-spy. Writes collapsed stacks, one root frame per process, to +# $PY_SPY_OUTPUT. Exits with the command's status, not py-spy's. +# +# PY_SPY=/path/to/py-spy PY_SPY_OUTPUT=.artifacts/py-spy.txt \ +# profile-backend-tests.sh make test-python-ci +set -u + +descends_from() { + local pid=$1 ancestor=$2 + while [ "$pid" -gt 1 ]; do + [ "$pid" -eq "$ancestor" ] && return 0 + pid=$(ps -o ppid= -p "$pid" 2>/dev/null | tr -d ' ') || return 1 + [ -n "$pid" ] || return 1 + done + return 1 +} + +"$@" & +cmd_pid=$! + +# py-spy attaches to the controller and follows the workers it spawns. +pytest_pid="" +for _ in $(seq 1 240); do + for pid in $(pgrep -f 'pytest-selected\.py' || true); do + if descends_from "$pid" "$cmd_pid"; then + pytest_pid=$pid + break + fi + done + [ -n "$pytest_pid" ] && break + kill -0 "$cmd_pid" 2>/dev/null || break + sleep 0.5 +done + +spy_pid="" +if [ -n "$pytest_pid" ]; then + "$PY_SPY" record --pid "$pytest_pid" --subprocesses --nonblocking --full-filenames \ + --rate "${PY_SPY_RATE:-20}" --format raw --output "$PY_SPY_OUTPUT" \ + >"${RUNNER_TEMP:-/tmp}/py-spy.log" 2>&1 & + spy_pid=$! +else + echo "::warning::py-spy: pytest-selected.py never started; not profiling" +fi + +rc=0 +wait "$cmd_pid" || rc=$? +if [ -n "$spy_pid" ]; then + # py-spy stops and writes its output once the controller exits. + wait "$spy_pid" || echo "::warning::py-spy exited non-zero; see py-spy.log" + tail -5 "${RUNNER_TEMP:-/tmp}/py-spy.log" +fi +exit "$rc" diff --git a/.github/workflows/scripts/pytest-selected.py b/.github/workflows/scripts/pytest-selected.py index 6fa86e75d215..6dc015324444 100644 --- a/.github/workflows/scripts/pytest-selected.py +++ b/.github/workflows/scripts/pytest-selected.py @@ -23,6 +23,7 @@ import math import os import shlex +import subprocess import sys from dataclasses import dataclass, field from pathlib import Path @@ -176,6 +177,17 @@ def per_file_plan_applies(args: list[str], files: list[str]) -> bool: return not any(os.path.isabs(f) or f.startswith(os.pardir + os.sep) for f in files) +def start_assert_rewrite_prewarm(shard_file: Path) -> None: + """Rewrite the shard's test files once in the background instead of in every + xdist worker. See prewarm_assert_rewrite.py.""" + # Without xdist the one pytest process would just race it for the same work. + if "-n" in shlex.split(os.environ.get("PYTEST_ADDOPTS", "")): + subprocess.Popen( + [sys.executable, str(Path(__file__).parent / "prewarm_assert_rewrite.py"), shard_file], + stdin=subprocess.DEVNULL, + ) + + def run_full_shard(args: list[str], total: int, group: int) -> int: files = list_test_files(args) if not per_file_plan_applies(args, files): @@ -216,6 +228,7 @@ def run_full_shard(args: list[str], total: int, group: int) -> int: os.environ["PYTEST_PLUGINS"] = ",".join( filter(None, [os.environ.get("PYTEST_PLUGINS"), "pytest_shard_plugin"]) ) + start_assert_rewrite_prewarm(shard_file) return int(pytest.main(args)) @@ -251,6 +264,7 @@ def main(args: list[str]) -> int: os.environ["TOTAL_TEST_GROUPS"] = str(shard.item_groups) os.environ["TEST_GROUP"] = str(shard.item_group) + start_assert_rewrite_prewarm(shard_file) exit_code = pytest.main(args) # Counts are estimates, so a shard can end up with no tests. if exit_code == pytest.ExitCode.NO_TESTS_COLLECTED: diff --git a/.github/workflows/scripts/test_monitor_resources.py b/.github/workflows/scripts/test_monitor_resources.py new file mode 100644 index 000000000000..e7f3b0cc07a8 --- /dev/null +++ b/.github/workflows/scripts/test_monitor_resources.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import importlib +import subprocess +import sys +import time + +_mod = importlib.import_module("monitor-resources") + + +def _sample(cpu: float, mem: float, avail: float, ts: float) -> dict: + return { + "ts": ts, + "cpu_pct": cpu, + "cpu_count": 4, + "mem_used_pct": mem, + "mem_available_gb": avail, + "swap_used_mb": 0.0, + "load_1m": 2.0, + } + + +def test_summarize_headroom() -> None: + samples = [_sample(50.0, 60.0, 5.0, float(i)) for i in range(20)] + summary = _mod.summarize(samples) + assert summary["samples"] == 20 + assert summary["cpu"]["avg_pct"] == 50.0 + assert summary["headroom"] == { + "cpu": "headroom", + "memory": "headroom", + "one_more_worker": "likely headroom for +1 worker", + } + + +def test_summarize_saturated_cpu() -> None: + samples = [_sample(95.0, 60.0, 5.0, float(i)) for i in range(20)] + summary = _mod.summarize(samples) + assert summary["headroom"]["cpu"] == "saturated" + assert summary["headroom"]["one_more_worker"] == "no headroom for +1 worker" + + +def test_summarize_saturated_memory() -> None: + samples = [_sample(50.0, 95.0, 0.2, float(i)) for i in range(20)] + summary = _mod.summarize(samples) + assert summary["headroom"]["memory"] == "saturated" + assert summary["headroom"]["one_more_worker"] == "no headroom for +1 worker" + + +def test_summarize_no_samples_is_inconclusive() -> None: + assert _mod.summarize([])["headroom"]["one_more_worker"] == "inconclusive (no samples)" + + +def test_summarize_cpu_by_group_splits_out_test_run() -> None: + samples = [ + {"ts": 0.0, "cpu_s_by_group": {"docker:postgres-postgres-1": 3.0}}, + {"ts": 5.0, "cpu_s_by_group": {"pytest:worker": 12.0, "docker:postgres-postgres-1": 4.0}}, + {"ts": 10.0, "cpu_s_by_group": {"pytest:worker": 10.0, "docker:snuba-gw0": 2.5}}, + ] + summary = _mod.summarize(samples) + assert summary["cpu_s_by_group"] == { + "pytest:worker": 22.0, + "docker:postgres-postgres-1": 7.0, + "docker:snuba-gw0": 2.5, + } + assert summary["cpu_s_by_group_during_tests"] == { + "pytest:worker": 22.0, + "docker:postgres-postgres-1": 4.0, + "docker:snuba-gw0": 2.5, + } + + +def test_read_pressure_uses_some_line(tmp_path) -> None: + psi = tmp_path / "cpu" + psi.write_text( + "some avg10=42.50 avg60=30.00 avg300=10.00 total=123\n" + "full avg10=0.00 avg60=0.00 avg300=0.00 total=0\n" + ) + assert _mod.read_pressure(str(psi))["avg10"] == 42.5 + + +def test_classify_process() -> None: + worker = "python -c import sys;exec(eval(sys.stdin.readline()))" + assert _mod.classify_process(worker, "python") == "pytest:worker" + assert ( + _mod.classify_process( + "python3 -b .github/workflows/scripts/pytest-selected.py tests", "python3" + ) + == "pytest:main" + ) + assert _mod.classify_process("", "kworker/0:1") == "kernel" + assert ( + _mod.classify_process("/usr/bin/docker-proxy -proto tcp", "docker-proxy") == "docker-proxy" + ) + + +def test_tracker_attributes_cpu_on_this_host() -> None: + tracker = _mod.ProcessCpuTracker() + assert tracker.sample() == ({}, {}) + deadline = time.monotonic() + 0.3 + while time.monotonic() < deadline: + pass + groups, _ = tracker.sample() + assert groups.get("pytest:main", 0) > 0 + + +def test_tracker_attributes_reaped_children() -> None: + tracker = _mod.ProcessCpuTracker() + tracker.sample() + subprocess.run( + [ + sys.executable, + "-c", + "import time\nt = time.monotonic()\nwhile time.monotonic() - t < 0.3: pass", + ], + check=True, + ) + _, children = tracker.sample() + assert children.get("pytest:main", 0) > 0 + + +def test_cpu_model(tmp_path) -> None: + cpuinfo = tmp_path / "cpuinfo" + cpuinfo.write_text("processor\t: 0\nmodel name\t: AMD EPYC 7763 64-Core Processor\n") + assert _mod.cpu_model(str(cpuinfo)) == "AMD EPYC 7763 64-Core Processor" + assert _mod.cpu_model(str(tmp_path / "missing")) is None diff --git a/.github/workflows/scripts/test_prewarm_assert_rewrite.py b/.github/workflows/scripts/test_prewarm_assert_rewrite.py new file mode 100644 index 000000000000..388e81f0418e --- /dev/null +++ b/.github/workflows/scripts/test_prewarm_assert_rewrite.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import importlib +import subprocess +import sys +from pathlib import Path + +import pytest +from _pytest.assertion.rewrite import PYC_TAIL, _read_pyc + +_mod = importlib.import_module("prewarm_assert_rewrite") + + +def test_prewarm_writes_pycs_pytest_reads(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(tmp_path) + (tmp_path / "src/sentry/testutils").mkdir(parents=True) + (tmp_path / "src/sentry/testutils/cases.py").write_text("X = 1\n") + (tmp_path / "tests/a").mkdir(parents=True) + (tmp_path / "tests/conftest.py").write_text("") + (tmp_path / "tests/a/test_one.py").write_text("def test_x():\n assert [1] == [2]\n") + (tmp_path / "tests/a/helper.py").write_text("") + file_list = tmp_path / "files.txt" + file_list.write_text("tests/a/test_one.py\ntests/a/helper.py\n") + + targets = _mod.rewrite_targets(file_list.read_text().split()) + assert targets == [ + Path("src/sentry/testutils/cases.py"), + Path("tests/conftest.py"), + Path("tests/a/test_one.py"), + ] + + assert _mod.main(str(file_list)) == 0 + for path in targets: + pyc = path.parent / "__pycache__" / (path.name[:-3] + PYC_TAIL) + assert _read_pyc(path, pyc, lambda msg: None) is not None + + +def test_prewarmed_pycs_keep_pytest_filenames(tmp_path: Path) -> None: + test_file = tmp_path / "tests/test_tb.py" + test_file.parent.mkdir() + test_file.write_text( + "import os\n" + "def test_filename():\n" + " assert test_filename.__code__.co_filename == os.path.abspath(__file__)\n" + ) + (tmp_path / "files.txt").write_text("tests/test_tb.py\n") + subprocess.run([sys.executable, _mod.__file__, "files.txt"], cwd=tmp_path, check=True) + assert list((tmp_path / "tests/__pycache__").glob("test_tb.*-pytest-*.pyc")) + result = subprocess.run( + [sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider", "tests/test_tb.py"], + cwd=tmp_path, + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stdout + result.stderr