rust/library/compiler-builtins/ci/ci-util.py

#!/usr/bin/env python3
"""Utilities for CI.

This dynamically prepares a list of routines that had a source file change based on
git history.
"""

import json
import os
import pprint
import re
import subprocess as sp
import sys
from dataclasses import dataclass
from functools import cache
from glob import glob
from inspect import cleandoc
from os import getenv
from pathlib import Path
from typing import TypedDict, Self

USAGE = cleandoc(
    """
    usage:

    ./ci/ci-util.py <COMMAND> [flags]

    COMMAND:
        generate-matrix
            Calculate a matrix of which functions had source change, print that as
            a JSON object.

        locate-baseline [--download] [--extract] [--tag TAG]
            Locate the most recent benchmark baseline available in CI and, if flags
            specify, download and extract it. Never exits with nonzero status if
            downloading fails.

            `--tag` can be specified to look for artifacts with a specific tag, such as
            for a specific architecture.

            Note that `--extract` will overwrite files in `iai-home`.

        handle-bench-regressions PR_NUMBER
            Exit with success if the pull request contains a line starting with
            `ci: allow-regressions`, indicating that regressions in benchmarks should
            be accepted. Otherwise, exit 1.
    """
)

REPO_ROOT = Path(__file__).parent.parent
GIT = ["git", "-C", REPO_ROOT]
DEFAULT_BRANCH = "master"
WORKFLOW_NAME = "CI"  # Workflow that generates the benchmark artifacts
ARTIFACT_PREFIX = "baseline-icount*"

# Don't run exhaustive tests if these files change, even if they contaiin a function
# definition.
IGNORE_FILES = [
    "libm/src/math/support/",
    "libm/src/libm_helper.rs",
    "libm/src/math/arch/intrinsics.rs",
]

# libm PR CI takes a long time and doesn't need to run unless relevant files have been
# changed. Anything matching this regex pattern will trigger a run.
TRIGGER_LIBM_CI_FILE_PAT = ".*(libm|musl).*"

TYPES = ["f16", "f32", "f64", "f128"]


def eprint(*args, **kwargs):
    """Print to stderr."""
    print(*args, file=sys.stderr, **kwargs)


@dataclass(init=False)
class PrCfg:
    """Directives that we allow in the commit body to control test behavior.

    These are of the form `ci: foo`, at the start of a line.
    """

    # Skip regression checks (must be at the start of a line).
    allow_regressions: bool = False
    # Don't run extensive tests
    skip_extensive: bool = False

    # Allow running a large number of extensive tests. If not set, this script
    # will error out if a threshold is exceeded in order to avoid accidentally
    # spending huge amounts of CI time.
    allow_many_extensive: bool = False

    # Max number of extensive tests to run by default
    MANY_EXTENSIVE_THRESHOLD: int = 20

    # Run tests for `libm` that may otherwise be skipped due to no changed files.
    always_test_libm: bool = False

    # String values of directive names
    DIR_ALLOW_REGRESSIONS: str = "allow-regressions"
    DIR_SKIP_EXTENSIVE: str = "skip-extensive"
    DIR_ALLOW_MANY_EXTENSIVE: str = "allow-many-extensive"
    DIR_TEST_LIBM: str = "test-libm"

    def __init__(self, body: str):
        directives = re.finditer(r"^\s*ci:\s*(?P<dir_name>\S*)", body, re.MULTILINE)
        for dir in directives:
            name = dir.group("dir_name")
            if name == self.DIR_ALLOW_REGRESSIONS:
                self.allow_regressions = True
            elif name == self.DIR_SKIP_EXTENSIVE:
                self.skip_extensive = True
            elif name == self.DIR_ALLOW_MANY_EXTENSIVE:
                self.allow_many_extensive = True
            elif name == self.DIR_TEST_LIBM:
                self.always_test_libm = True
            else:
                eprint(f"Found unexpected directive `{name}`")
                exit(1)

        pprint.pp(self)


@dataclass
class PrInfo:
    """GitHub response for PR query"""

    body: str
    commits: list[str]
    created_at: str
    number: int
    cfg: PrCfg

    @classmethod
    def from_env(cls) -> Self | None:
        """Create a PR object from the PR_NUMBER environment if set, `None` otherwise."""
        pr_env = os.environ.get("PR_NUMBER")
        if pr_env is not None and len(pr_env) > 0:
            return cls.from_pr(pr_env)

        return None

    @classmethod
    @cache  # Cache so we don't print info messages multiple times
    def from_pr(cls, pr_number: int | str) -> Self:
        """For a given PR number, query the body and commit list."""
        pr_info = sp.check_output(
            [
                "gh",
                "pr",
                "view",
                str(pr_number),
                "--json=number,commits,body,createdAt",
                # Flatten the commit list to only hashes, change a key to snake naming
                "--jq=.commits |= map(.oid) | .created_at = .createdAt | del(.createdAt)",
            ],
            text=True,
        )
        pr_json = json.loads(pr_info)
        eprint("PR info:", json.dumps(pr_json, indent=4))
        return cls(**json.loads(pr_info), cfg=PrCfg(pr_json["body"]))


class FunctionDef(TypedDict):
    """Type for an entry in `function-definitions.json`"""

    sources: list[str]
    type: str


class Context:
    gh_ref: str | None
    changed: list[Path]
    defs: dict[str, FunctionDef]

    def __init__(self) -> None:
        self.gh_ref = getenv("GITHUB_REF")
        self.changed = []
        self._init_change_list()

        with open(REPO_ROOT.joinpath("etc/function-definitions.json")) as f:
            defs = json.load(f)

        defs.pop("__comment", None)
        self.defs = defs

    def _init_change_list(self):
        """Create a list of files that have been changed. This uses GITHUB_REF if
        available, otherwise a diff between `HEAD` and `master`.
        """

        # For pull requests, GitHub creates a ref `refs/pull/1234/merge` (1234 being
        # the PR number), and sets this as `GITHUB_REF`.
        ref = self.gh_ref
        eprint(f"using ref `{ref}`")
        if not self.is_pr():
            # If the ref is not for `merge` then we are not in PR CI
            eprint("No diff available for ref")
            return

        # The ref is for a dummy merge commit. We can extract the merge base by
        # inspecting all parents (`^@`).
        merge_sha = sp.check_output(
            GIT + ["show-ref", "--hash", ref], text=True
        ).strip()
        merge_log = sp.check_output(GIT + ["log", "-1", merge_sha], text=True)
        eprint(f"Merge:\n{merge_log}\n")

        parents = (
            sp.check_output(GIT + ["rev-parse", f"{merge_sha}^@"], text=True)
            .strip()
            .splitlines()
        )
        assert len(parents) == 2, f"expected two-parent merge but got:\n{parents}"
        base = parents[0].strip()
        incoming = parents[1].strip()

        eprint(f"base: {base}, incoming: {incoming}")
        textlist = sp.check_output(
            GIT + ["diff", base, incoming, "--name-only"], text=True
        )
        self.changed = [Path(p) for p in textlist.splitlines()]

    def is_pr(self) -> bool:
        """Check if we are looking at a PR rather than a push."""
        return self.gh_ref is not None and "merge" in self.gh_ref

    @staticmethod
    def _ignore_file(fname: str) -> bool:
        return any(fname.startswith(pfx) for pfx in IGNORE_FILES)

    def changed_routines(self) -> dict[str, list[str]]:
        """Create a list of routines for which one or more files have been updated,
        separated by type.
        """
        routines = set()
        for name, meta in self.defs.items():
            # Don't update if changes to the file should be ignored
            sources = (f for f in meta["sources"] if not self._ignore_file(f))

            # Select changed files
            changed = [f for f in sources if Path(f) in self.changed]

            if len(changed) > 0:
                eprint(f"changed files for {name}: {changed}")
                routines.add(name)

        ret: dict[str, list[str]] = {}
        for r in sorted(routines):
            ret.setdefault(self.defs[r]["type"], []).append(r)

        return ret

    def may_skip_libm_ci(self) -> bool:
        """If this is a PR and no libm files were changed, allow skipping libm
        jobs."""

        # Always run on merge CI
        if not self.is_pr():
            return False

        pr = PrInfo.from_env()
        assert pr is not None, "Is a PR but couldn't load PrInfo"

        # Allow opting in to libm tests
        if pr.cfg.always_test_libm:
            return False

        # By default, run if there are any changed files matching the pattern
        return all(not re.match(TRIGGER_LIBM_CI_FILE_PAT, str(f)) for f in self.changed)

    def emit_workflow_output(self):
        """Create a JSON object a list items for each type's changed files, if any
        did change, and the routines that were affected by the change.
        """

        skip_tests = False
        error_on_many_tests = False

        pr = PrInfo.from_env()
        if pr is not None:
            skip_tests = pr.cfg.skip_extensive
            error_on_many_tests = not pr.cfg.allow_many_extensive

            if skip_tests:
                eprint("Skipping all extensive tests")

        changed = self.changed_routines()
        matrix = []
        total_to_test = 0

        # Figure out which extensive tests need to run
        for ty in TYPES:
            ty_changed = changed.get(ty, [])
            ty_to_test = [] if skip_tests else ty_changed
            total_to_test += len(ty_to_test)

            item = {
                "ty": ty,
                "changed": ",".join(ty_changed),
                "to_test": ",".join(ty_to_test),
            }

            matrix.append(item)

        ext_matrix = json.dumps({"extensive_matrix": matrix}, separators=(",", ":"))
        may_skip = str(self.may_skip_libm_ci()).lower()
        print(f"extensive_matrix={ext_matrix}")
        print(f"may_skip_libm_ci={may_skip}")
        eprint(f"total extensive tests: {total_to_test}")

        if error_on_many_tests and total_to_test > PrCfg.MANY_EXTENSIVE_THRESHOLD:
            eprint(
                f"More than {PrCfg.MANY_EXTENSIVE_THRESHOLD} tests would be run; add"
                f" `{PrCfg.DIR_ALLOW_MANY_EXTENSIVE}` to the PR body if this is"
                " intentional. If this is refactoring that happens to touch a lot of"
                f" files, `{PrCfg.DIR_SKIP_EXTENSIVE}` can be used instead."
            )
            exit(1)


def locate_baseline(flags: list[str]) -> None:
    """Find the most recent baseline from CI, download it if specified.

    This returns rather than erroring, even if the `gh` commands fail. This is to avoid
    erroring in CI if the baseline is unavailable (artifact time limit exceeded, first
    run on the branch, etc).
    """

    download = False
    extract = False
    tag = ""

    while len(flags) > 0:
        match flags[0]:
            case "--download":
                download = True
            case "--extract":
                extract = True
            case "--tag":
                tag = flags[1]
                flags = flags[1:]
            case _:
                eprint(USAGE)
                exit(1)
        flags = flags[1:]

    if extract and not download:
        eprint("cannot extract without downloading")
        exit(1)

    try:
        # Locate the most recent job to complete with success on our branch
        latest_job = sp.check_output(
            [
                "gh",
                "run",
                "list",
                "--status=success",
                f"--branch={DEFAULT_BRANCH}",
                "--json=databaseId,url,headSha,conclusion,createdAt,"
                "status,workflowDatabaseId,workflowName",
                # Return the first array element matching our workflow name. NB: cannot
                # just use `--limit=1`, jq filtering happens after limiting. We also
                # cannot just use `--workflow` because GH gets confused from
                # different file names in history.
                f'--jq=[.[] | select(.workflowName == "{WORKFLOW_NAME}")][0]',
            ],
            text=True,
        )
    except sp.CalledProcessError as e:
        eprint(f"failed to run github command: {e}")
        return

    try:
        latest = json.loads(latest_job)
        eprint("latest job: ", json.dumps(latest, indent=4))
    except json.JSONDecodeError as e:
        eprint(f"failed to decode json '{latest_job}', {e}")
        return

    if not download:
        eprint("--download not specified, returning")
        return

    job_id = latest.get("databaseId")
    if job_id is None:
        eprint("skipping download step")
        return

    artifact_glob = f"{ARTIFACT_PREFIX}{f"-{tag}" if tag else ""}*"

    sp.run(
        ["gh", "run", "download", str(job_id), f"--pattern={artifact_glob}"],
        check=False,
    )

    if not extract:
        eprint("skipping extraction step")
        return

    # Find the baseline with the most recent timestamp. GH downloads the files to e.g.
    # `some-dirname/some-dirname.tar.xz`, so just glob the whole thing together.
    candidate_baselines = glob(f"{artifact_glob}/{artifact_glob}")
    if len(candidate_baselines) == 0:
        eprint("no possible baseline directories found")
        return

    candidate_baselines.sort(reverse=True)
    baseline_archive = candidate_baselines[0]
    eprint(f"extracting {baseline_archive}")
    sp.run(["tar", "xJvf", baseline_archive], check=True)
    eprint("baseline extracted successfully")


def handle_bench_regressions(args: list[str]):
    """Exit with error unless the PR message contains an ignore directive."""

    match args:
        case [pr_number]:
            pr_number = pr_number
        case _:
            eprint(USAGE)
            exit(1)

    pr = PrInfo.from_pr(pr_number)
    if pr.cfg.allow_regressions:
        eprint("PR allows regressions")
        return

    eprint("Regressions were found; benchmark failed")
    exit(1)


def main():
    match sys.argv[1:]:
        case ["generate-matrix"]:
            ctx = Context()
            ctx.emit_workflow_output()
        case ["locate-baseline", *flags]:
            locate_baseline(flags)
        case ["handle-bench-regressions", *args]:
            handle_bench_regressions(args)
        case ["--help" | "-h"]:
            print(USAGE)
            exit()
        case _:
            eprint(USAGE)
            exit(1)


if __name__ == "__main__":
    main()