weblib-ci/report_job_log.py

#!/usr/bin/env python3
"""Post the tail of a build log as a pull request comment.

Gitea 1.25.2 returns 500 from `actions/jobs/{id}/logs` for **every** id, so a
red job reports `Failing after 48s` and nothing else. Every hypothesis about
why then costs a push, a wait, and a one-bit answer. This puts the log where it
can be read, next to the change that caused it.

Standard library only, like `sync_blocked_label.py`, so it needs nothing but an
interpreter.

Lives in `weblib/weblib-ci` and is used by the other repos from there, so the
comment it posts must not name a path inside the repo it is reporting on --
there is no copy there to find.

Usage, from a workflow step guarded by `if: failure()`:

    report_job_log.py /tmp/build.log

The header says *failed* by default, which is what that guard means. A step
guarded by `if: always()` -- a probe, or a job whose output is the point rather
than its exit code -- has to say so, or the comment reports a failure that did
not happen:

    report_job_log.py /tmp/build.log --status "${{ job.status }}"

Everything else comes from the Actions environment: `GITHUB_REPOSITORY`,
`GITHUB_EVENT_PATH` (for the PR number), `GITEA_TOKEN`/`GITHUB_TOKEN`,
`GITHUB_SERVER_URL`, and `JOB_STATUS` if `--status` is not passed.

Remove this once the instance is updated and logs can be read directly.
"""

import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.request

DEFAULT_HOST = "https://git.chaosbit.de"

# Gitea renders a comment body wholesale; a full nix log is megabytes and the
# failure is always at the end.
DEFAULT_TAIL = 12000

# nix colours its output, and the raw escapes are noise in Markdown.
ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")

# The header used to hardcode "failed". Every caller guards the step with
# `if: failure()`, so that was true by construction -- until an `if: always()`
# probe in weblib-viewer#10 posted a failure report for a job that had passed.
# The default stays `failed` so those guarded callers are unchanged.
DEFAULT_STATUS = "failed"

# `${{ job.status }}` yields success/failure/cancelled/skipped; a human writing
# the flag by hand reaches for passed/failed. Accept both spellings so a call
# site can pass the expression straight through.
STATUS_PHRASES = {
    "failed": "failed",
    "failure": "failed",
    "passed": "passed",
    "success": "passed",
    "succeeded": "passed",
    "cancelled": "was cancelled",
    "canceled": "was cancelled",
    "skipped": "was skipped",
}


def status_phrase(status):
    """The verb for the comment header.

    An unrecognised status is reported verbatim rather than rejected. This
    script exists because a red job says nothing, so it must not itself become
    the thing that says nothing: `argparse`'s `choices=` would exit 2 on a
    status this table has not heard of, and the log would never be posted.
    """
    key = (status or "").strip().lower()
    if not key:
        key = DEFAULT_STATUS
    if key in STATUS_PHRASES:
        return STATUS_PHRASES[key]
    # Backticks removed, not escaped: the verbatim value goes inside a code
    # span in a **bold** header, and a backtick in it closes the span early --
    # the rest of the status then renders as markdown. Nothing hostile is
    # expected here (`${{ job.status }}` is written by whoever wrote the
    # workflow), but this repo is public and consumed by four others, and a
    # branch name or matrix value could reach this argument later.
    return "finished with status `" + status.strip().replace("`", "") + "`"


def missing_log_note(path, phrase):
    """What to say when the log file the step named is not there."""
    if phrase == "failed":
        return (f"The step produced no log file at `{path}` — it failed "
                "before the build started.")
    return (f"The step produced no log file at `{path}` — nothing was "
            "captured.")


def pull_request_number(event_path):
    """The PR this job is running for, from the event payload, or None."""
    if not event_path or not os.path.exists(event_path):
        return None
    with open(event_path) as fh:
        event = json.load(fh)
    pr = event.get("pull_request") or {}
    return pr.get("number")


def pull_request_for_branch(host, repo, token, branch):
    """The open PR whose head is `branch`, or None.

    Only a `pull_request` event carries the PR in its payload. A
    `workflow_dispatch` or `schedule` run does not, and the first version of
    this script simply printed "not a pull request build" and exited 0 -- so a
    dispatched run that failed stayed exactly as silent as the one this script
    exists to fix. Looking the branch up covers those.

    A push to a branch with no PR still has nowhere to comment, which is
    honest: there is no thread for it.
    """
    if not branch:
        return None
    url = (f"{host.rstrip('/')}/api/v1/repos/{repo}/pulls"
           f"?state=open&limit=50")
    req = urllib.request.Request(url)
    req.add_header("Authorization", f"token {token}")
    try:
        with urllib.request.urlopen(req, timeout=30) as res:
            pulls = json.loads(res.read() or b"[]")
    except (urllib.error.URLError, OSError, ValueError):
        # A reporter that raises reports nothing. HTTPError is a URLError, and
        # ValueError covers a body that is not JSON.
        return None
    for pull in pulls:
        if ((pull.get("head") or {}).get("ref")) == branch:
            return pull.get("number")
    return None


def tail(path, limit):
    with open(path, "rb") as fh:
        try:
            fh.seek(0, os.SEEK_END)
            size = fh.tell()
            fh.seek(max(0, size - limit * 4))
        except OSError:                     # not seekable; read it all
            size = None
        raw = fh.read()
    text = ANSI.sub("", raw.decode("utf-8", "replace"))
    if len(text) > limit:
        text = text[-limit:]
        # Do not start mid-line; it reads as corruption rather than truncation.
        text = text.split("\n", 1)[-1]
        text = "[…truncated…]\n" + text
    return text


def comment(host, repo, number, token, body):
    url = f"{host.rstrip('/')}/api/v1/repos/{repo}/issues/{number}/comments"
    req = urllib.request.Request(url, data=json.dumps({"body": body}).encode(),
                                 method="POST")
    req.add_header("Content-Type", "application/json")
    req.add_header("Authorization", f"token {token}")
    with urllib.request.urlopen(req, timeout=30) as res:
        return json.loads(res.read() or b"null")


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("log", help="path to the captured build log")
    ap.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY"))
    ap.add_argument("--host", default=os.environ.get("GITEA_HOST")
                    or os.environ.get("GITHUB_SERVER_URL") or DEFAULT_HOST)
    ap.add_argument("--token", default=os.environ.get("GITEA_TOKEN")
                    or os.environ.get("GITHUB_TOKEN"))
    ap.add_argument("--pr", type=int, default=None)
    ap.add_argument("--job", default=os.environ.get("GITHUB_JOB") or "job")
    ap.add_argument("--tail", type=int, default=DEFAULT_TAIL)
    ap.add_argument("--status",
                    default=os.environ.get("JOB_STATUS") or DEFAULT_STATUS,
                    help="outcome of the job being reported: failed (the "
                         "default, and what `if: failure()` means), passed, "
                         "cancelled, skipped, or ${{ job.status }} verbatim")
    args = ap.parse_args(argv)

    if not args.repo or not args.token:
        print("no repo or no token; set GITHUB_REPOSITORY and GITEA_TOKEN")
        return 0

    number = args.pr or pull_request_number(os.environ.get("GITHUB_EVENT_PATH"))
    if number is None:
        branch = (os.environ.get("GITHUB_HEAD_REF")
                  or os.environ.get("GITHUB_REF_NAME"))
        number = pull_request_for_branch(args.host, args.repo, args.token, branch)
    if number is None:
        print("no pull request for this run: nothing to comment on")
        return 0

    phrase = status_phrase(args.status)

    if not os.path.exists(args.log):
        text = missing_log_note(args.log, phrase)
    else:
        text = "```\n" + tail(args.log, args.tail).rstrip() + "\n```"

    sha = (os.environ.get("GITHUB_SHA") or "")[:8]
    body = (f"**`{args.job}` {phrase}**{f' on `{sha}`' if sha else ''}.\n\n"
            "Job logs return 500 on this Gitea, so here is the tail of the "
            "build output, posted by [`report_job_log.py`]"
            "(https://git.chaosbit.de/weblib/weblib-ci).\n\n"
            f"{text}\n")

    # A failure to report a failure must not itself be silent, but it also must
    # not mask the real one: the step is already `if: failure()`.
    try:
        result = comment(args.host, args.repo, number, args.token, body)
    except urllib.error.HTTPError as e:
        print(f"could not comment: {e.code} "
              f"{e.read()[:300].decode('utf-8', 'replace')}", file=sys.stderr)
        return 0
    print(f"reported to {args.repo}#{number}: {(result or {}).get('html_url')}")
    return 0


if __name__ == "__main__":
    sys.exit(main())