Initial thoughts
As a GitLab user, you may be handling multiple projects at once, triggering pipelines. Wouldn't it be great if there was an easy way to monitor all these pipelines in real-time? Unfortunately, out-of-the-box solutions don't quite fit the bill.
That's why we've developed a Python script that leverages the power of GitLab API to display the latest pipeline runs for every projects in a group. As simple as :
watch --no-title --color python display-latest-pipelines.py --group-id 12345
1. Considered alternate solutions
Some alternate solutions have been explored before making a script from scratch.
Glab ci view
GLab is an open source GitLab CLI tool. It brings GitLab to your terminal: next to where you are already working with Git and your code, without switching between windows and browser tabs.
There is a particular command displaying a pipeline, glab ci view :
Multiple problems for our usage :
- Limited to one project
- The result is large and does not allow many pipelines on the same screen
- You do not get "the latest" pipeline; you have to choose a branch
Tabs Grid browser plugins
Tab grids browser plugins are easy to use, but does not update status in real time, you have to refresh the given tabs
2. The Python script
Here is the script, with some consideration :
- The width is not responsive, you can change the
stages_widthvariable to suit your needs - On MacOS, watch command disallows some emojis, that's why we are using π instead of π’
- Use
watchwith--no-titleoption, else your GitLab token will be displayed at all time - Projects pipelines are displayed one at a time to allow more real time display
- The least possible number of lines are used; you can obtain better output with some adjustment, if vertical space is not an issue for you
Pre-requisites
- Some Python packages installed
pip install pytz
- A token having access to all the projects in the group
Source code
#
### Display, in console, latest pipelines from each project in a given group
#
## one shot:
# python display-latest-pipelines.py --exclude='TPs Benoit C,whatever' --group-id=87844506
#
## continuous update:
# watch --no-title --color "python display-latest-pipelines.py --exclude='TPs Benoit C,whatever' --group-id=87844506"
#
import argparse
import requests
from datetime import datetime
import pytz
import os
from enum import Enum
class Color(Enum):
GREEN = "\033[92m"
GREY = "\033[90m"
CYAN = "\033[96m"
RED = "\033[91m"
YELLOW = "\033[93m"
BLUE = "\033[94m"
RESET = "\033[0m"
NO_CHANGE = ""
# Parse command-line arguments for access token, group ID, and GitLab hostname
parser = argparse.ArgumentParser(description='Retrieve GitLab pipeline data for projects in a group')
parser.add_argument('--host', type=str, default="gitlab.com", help='Hostname of the GitLab instance')
parser.add_argument('--token', type=str, default=None, help='GitLab API access token (default: $GITLAB_TOKEN (exported) environment variable)')
parser.add_argument('--group-id', type=int, help='ID of the group to retrieve projects from')
parser.add_argument('--exclude', type=str, default="", help='Comma-separated list of project names to exclude (default: none)')
args = parser.parse_args()
# Set maximum length for job names
stages_width = 36
# Check if token is set via environment variable and use it as default if not provided
if args.token is None:
args.token = os.getenv('GITLAB_TOKEN', 'NONE')
# Set headers and URLs for API requests
headers = {"Private-Token": args.token}
projects_url = f"https://{args.host}/api/v4/groups/{args.group_id}/projects?include_subgroups=true&simple=true"
response = requests.get(projects_url, headers=headers)
if response.status_code != 200:
print(f"\n{Color.RED.value}Failed to call GitLab instance: {response.json()}{Color.RESET.value}")
exit()
projects = response.json()
pipeline_data = {}
no_pipelines_projects = []
excluded_projects = set(args.exclude.split(','))
for project in projects:
if project["name"] in excluded_projects:
continue
# Retrieve the most recent pipeline for the current project
pipeline_url = f"https://gitlab.com/api/v4/projects/{project['id']}/pipelines?per_page=1&sort=desc&order_by=updated_at"
response = requests.get(pipeline_url, headers=headers)
if not response.json(): # check if there are any pipelines for this project
no_pipelines_projects.append(project['name'])
continue
pipeline = response.json()[0]
# Convert updated_at to a datetime object and convert it to Paris timezone
updated_time = datetime.strptime(pipeline["updated_at"], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=pytz.utc).astimezone(pytz.timezone('Europe/Paris'))
# Format the datetime object as a human-readable string
updated_at_human_readable = updated_time.strftime("%d %b %Y at %H:%M:%S")
time_diff = datetime.now(pytz.utc) - updated_time
delta = time_diff.total_seconds()
if delta < 120:
updated_ago = f'{int(delta)} seconds'
elif delta < 7200:
updated_ago = f'{int(delta / 60)} minutes'
elif delta < 172800: # 2 days in seconds (60*60*24*2)
updated_ago = f'{int(delta / 3600)} hours'
else:
updated_ago = f'{int(delta / 86400)} days' # 1 day in seconds (60*60*24)
# Determine color based on pipeline status
match pipeline["status"]:
case "success":
color = Color.GREEN
case "created" | "waiting_for_resource" | "preparing" | "pending" | "canceled" | "skipped" | "manual":
color = Color.GREY
case "running":
color = Color.CYAN
case "failed":
color = Color.RED
print(f"\nβ {color.value}{project['name']} for {pipeline['ref']} : {pipeline['status']} (since {updated_at_human_readable}, {updated_ago} ago){Color.RESET.value}")
job_data = {}
# Retrieve data about all jobs for the current pipeline
jobs_url = f"https://gitlab.com/api/v4/projects/{project['id']}/pipelines/{pipeline['id']}/jobs"
response = requests.get(jobs_url, headers=headers)
jobs = response.json()
for job in list(reversed(jobs)):
job_name = job["name"]
stage = job["stage"]
job_status = job["status"]
# Determine emoji based on job status and pipeline status
match (job_status, pipeline["status"]):
case ("success", _):
emoji = "π" # π’β
do not work within a "watch" on macos
job_color = Color.GREEN
case ("running", _):
emoji = "π΅"
job_color = Color.CYAN
case ("pending" | "created", _):
emoji = "π"
job_color = Color.NO_CHANGE
case ("skipped", _):
emoji = "π"
job_color = Color.GREY
case ("warning", _):
emoji = "π" # π does not work within a "watch" on macos
job_color = Color.YELLOW
case ("manual", _):
emoji = "βΆοΈ "
job_color = Color.NO_CHANGE
case ("failed", "success"):
emoji = "π" # π does not work within a "watch" on macos
job_color = Color.YELLOW
case ("failed", _):
emoji = "π΄"
job_color = Color.RED
case (_, _):
print(job_status)
if stage not in job_data:
job_data[stage] = []
job_data[stage].append((job_name, job_status, job_color, emoji))
# Sort jobs within each stage alphabetically by job name
for stage in job_data:
job_data[stage].sort(key=lambda x: x[0])
# Find the maximum number of jobs in a stage for this pipeline
max_jobs = max(len(jobs) for jobs in job_data.values())
lines = [" "] * (max_jobs + 1)
lines[0] = "" # stages start with a border character instead of a space
# Print out the job data for each stage, padding to make all stages content the same length (emojis make this exercise a bit more difficult)
for stage, jobs in job_data.items():
stage = "[ "+stage+" ]"
lines[0] = f"{lines[0]}β{stage.center(stages_width-3, 'β').upper()}β "
for i, (job_name, job_status, job_color, emoji) in enumerate(jobs, start=1):
lines[i] = f"{lines[i]}{emoji} {job_color.value}{job_name[:stages_width-5].ljust(stages_width-3)}{Color.RESET.value}"
for j in range(len(jobs) + 1, max_jobs + 1):
lines[j] = lines[j] + " ".ljust(stages_width)
for line in lines:
print(line)
if no_pipelines_projects:
print(f"\n\033[90mProjects without pipeline:", ', '.join(no_pipelines_projects))
Illustrations generated locally by Pinokio using Stable Cascade plugin
Further reading
This article was enhanced with the assistance of an AI language model to ensure clarity and accuracy in the content, as English is not my native language.
Top comments (0)