-
Notifications
You must be signed in to change notification settings - Fork 0
/
gitlab-pipeline-errors
executable file
·54 lines (45 loc) · 1.97 KB
/
gitlab-pipeline-errors
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
#! /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'")
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}")
for line in real_job.trace().decode("utf-8").splitlines():
if line.startswith(("ERROR:", "WARNING:")):
print(line)
elif job.failure_reason == "runner_system_failure":
real_job.retry()
print("Restarted job")
print()