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 :
python display-latest-pipelines.py --group-id 12345 --watch
1. Considered alternate solutions
Some alternate solutions have been explored before making a script from scratch.
GitLab's operations dashboard
For premium and ultimate GitLab users, there is the Operations Dashboard:
This is a nice start, but only the overall pipeline status is available, which is too light for pipeline-focused usages.
GitLab CI pipelines exporter
Mentioned in official documentation, the GitLab CI Pipelines Exporter for Prometheus fetches metrics from the API and pipeline events. It can check branches in projects automatically and get the pipeline status and duration. In combination with a Grafana dashboard, this helps build an actionable view for your operations team. Metric graphs can also be embedded into incidents making problem resolving easier. Additionally, it can also export metrics about jobs and environments.
Very interesting, but the scope is way larger than our usage and does not follow pipelines in progress.
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 :
- GitLab host, token, group-id, excluded projects list, stages width and watch mode are configurable
- For one shot mode (versus watch mode), projects pipelines are displayed one at a time to allow more real time display
- The least possible vertical space is used as a specific goal; you could obtain nicer (but more verbose) output with some minor script adjustment
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
#
# python display-latest-pipelines.py --group-id=8784450 [--watch] [--token=$GITLAB_TOKEN] [--host=gitlab.com] [--exclude='TPs Benoit C,whatever'] [--stages-width=30]
#
import argparse
from datetime import datetime
from enum import Enum
import os
import pytz
import requests
import sys
import unicodedata
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 = ""
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)')
parser.add_argument('--watch', action='store_true', help='Run indefinitely while refreshing output')
parser.add_argument('--stages-width', type=int, default=42, help='Width for stages display (default: 42)')
args = parser.parse_args()
if args.token is None:
args.token = os.getenv('GITLAB_TOKEN', 'NONE')
headers = {"Private-Token": args.token}
projects_url = f"https://{args.host}/api/v4/groups/{args.group_id}/projects?include_subgroups=true&simple=true"
def print_or_gather(output, text):
if args.watch:
output.append(text)
else:
print(text)
def count_emoji(text):
"""Count the number of emojis in the input text."""
custom_lengths = {
"\U0001F3D7": 0, # Construction sign 🏗️
# Add more special characters as needed.
}
count = 0
for char in text:
if unicodedata.category(char).startswith('So'):
if char in custom_lengths:
count += custom_lengths[char]
else:
count += 1
return count
def fetch_pipelines():
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}")
return
projects = response.json()
pipeline_data = {}
no_pipelines_projects = []
excluded_projects = set(args.exclude.split(','))
output = []
for project in projects:
if project["name"] in excluded_projects:
continue
pipeline_url = f"https://gitlab.com/api/v4/projects/{project['id']}/pipelines?per_page=1&sort=desc&order_by=id"
response = requests.get(pipeline_url, headers=headers)
if not response.json():
no_pipelines_projects.append(project['name'])
continue
pipeline = response.json()[0]
updated_time = datetime.strptime(pipeline["updated_at"], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=pytz.utc).astimezone(pytz.timezone('Europe/Paris'))
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: # 2 hours in seconds
updated_ago = f'{int(delta / 60)} minutes'
elif delta < 172800: # 2 days in seconds
updated_ago = f'{int(delta / 3600)} hours'
else:
updated_ago = f'{int(delta / 86400)} days'
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.BLUE
case "failed":
color = Color.RED
print_or_gather(output,f"\n↓ {color.value}{project['name']} for {pipeline['ref']} : {pipeline['status']} (since {updated_at_human_readable}, {updated_ago} ago){Color.RESET.value}")
job_data = {}
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"]
match (job_status, pipeline["status"]):
case ("success", _):
emoji = "🟢"
job_color = Color.GREEN
case ("running", _):
emoji = "🔵"
job_color = Color.BLUE
case ("pending" | "created", _):
emoji = "🔘"
job_color = Color.NO_CHANGE
case ("skipped" | "canceled", _):
emoji = "🔘"
job_color = Color.GREY
case ("warning", _):
emoji = "🟠"
job_color = Color.YELLOW
case ("manual", _):
emoji = "▶️"
job_color = Color.NO_CHANGE
case ("failed", "success"):
emoji = "🟠"
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 any 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
for stage, jobs in job_data.items():
stage = "[ "+stage+" ]"
lines[0] = f"{lines[0]}╔{stage.center(args.stages_width - 3 - count_emoji(stage), '═').upper()}╗ "
for i, (job_name, job_status, job_color, emoji) in enumerate(jobs, start=1):
# emojis in job names make this exercise a bit more difficult: ljust make them expand the size, so we compensate
lines[i] = f"{lines[i]}{emoji} {job_color.value}{job_name[:args.stages_width - 6 - count_emoji(job_name)].ljust(args.stages_width - 3 - count_emoji(job_name))}{Color.RESET.value}"
for j in range(len(jobs) + 1, max_jobs + 1):
lines[j] = lines[j] + " ".ljust(args.stages_width)
for line in lines:
print_or_gather(output,line)
if no_pipelines_projects:
print_or_gather(output,f"\n\033[90mProjects without pipeline: {', '.join(no_pipelines_projects)}\033[0m")
return "\n".join(output)
try:
if args.watch:
while True:
output = fetch_pipelines()
sys.stdout.write("\x1b[2J\x1b[H") # Clear the screen
sys.stdout.flush()
print(output)
else:
fetch_pipelines()
except KeyboardInterrupt:
pass
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)