gitlab-pipeline-errors

#! /usr/bin/env python3

# List all errors in all failed jobs in the specified pipeline

import argparse
import gitlab

parser = argparse.ArgumentParser()
parser.add_argument("--server", nargs="?")
parser.add_argument("--project", "-p")
parser.add_argument("--pipeline", "-l", required=True, help="pipeline ID, or branch name for latest pipeline")
parser.add_argument("--use-config", "-c", action="store_true", help="Use configuration stored in the current git repository")
parser.add_argument("--retry", "-r", action="store_true", help="Retry jobs that failed with 'runner system failure'")
parser.add_argument("--all", "-a", action="store_true", help="Show full log not just last 10 lines'")
args = parser.parse_args()

# If we're using a configuration, load it into args and re-parse to override the settings
if args.use_config:
    import configparser, git
    repo = git.Repo(search_parent_directories=True)
    with repo.config_reader() as config:
        for key in ("project", "stage"):
            try:
                value = config.get("ross-tools", key)
                setattr(args, key, value)
            except configparser.NoOptionError:
                pass
    args = parser.parse_args(namespace=args)

# TODO check variables are set

gl = gitlab.Gitlab.from_config(args.server)
gl.auth()

project = gl.projects.get(args.project)
if args.pipeline.isdigit():
    pipeline = project.pipelines.get(args.pipeline)
else:
    pipeline = project.pipelines.list(ref=args.pipeline, page=1, per_page=1)[0]
    print(f"Found pipeline {pipeline.web_url}")
    print()

for job in pipeline.jobs.list(scope="failed", iterator=True):
    print(f"Job {job.name} failed due to {job.failure_reason}.")

    real_job = project.jobs.get(job.id, lazy=True)
    if job.failure_reason == "script_failure":
        print(f"Full log at {job.web_url}")
        log = real_job.trace().decode("utf-8").splitlines()
        lines = [l for l in log if l.startswith(("ERROR:", "WARNING:"))]
        line_start = 0 if args.all else -10
        print("\n".join(lines[line_start:]))
    elif job.failure_reason == "runner_system_failure":
        real_job.retry()
        print("Restarted job")
    print()