-
-
Save wavedocs/3c726acad8aff6d4d8f6cde59e1686d6 to your computer and use it in GitHub Desktop.
Prints a PlantUML diagram that shows the DAG of the GitLab pipeline
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
"""Prints a PlantUML diagram that shows the DAG of the GitLab pipeline""" | |
import sys | |
import yaml | |
from pprint import pprint | |
def merge(user, default): | |
if isinstance(user,dict) and isinstance(default,dict): | |
for k,v in default.items(): | |
if k not in user: | |
user[k] = v | |
else: | |
user[k] = merge(user[k],v) | |
return user | |
merged = {} | |
with open('.gitlab-ci.yml') as file: | |
tmp = yaml.load(file) | |
filenames = tmp['include'] | |
filenames.append('.gitlab-ci.yml') | |
for filename in filenames: | |
with open(f'./{filename}') as file: | |
tmp = yaml.load(file) | |
merge(merged, tmp) | |
tmp = {} | |
print('@startuml') | |
print('left to right direction') | |
jobs = {k: v for k, v in merged.items() if k not in ['image', 'services', 'stages', 'include']} | |
branches = {} | |
all_changes = [v['only']['changes'] for k, v in jobs.items() if type(v.get('only')) is dict and v.get('only', {}).get('changes', {}) and not v.get('needs')] # gather all `only:changes` | |
for changes in all_changes: | |
branches[str(changes)] = [] # list of e.g. {"kibana/**/*": deployKibanaDev} | |
if len(sys.argv) > 1: | |
for stage in merged['stages']: # This groups by stage, which looks nice but may be misleading | |
matching_jobs = [k for k, v in jobs.items() if v['stage'] == stage] | |
print(f'partition "{stage}" {{') | |
for job in matching_jobs: | |
needs = merged[job].get('needs') | |
if needs: | |
for need in needs: | |
print(f' "{need}" --> "{job}" ') | |
else: | |
print(f' (*) --> "{job}" ') | |
print("}") | |
else: | |
for job in jobs: | |
changes = [] | |
needs = merged[job].get('needs') | |
try: | |
changes = merged[job]['only']['changes'] | |
except: | |
pass | |
if needs: # if `needs:` exists, `only:changes` is implied from transitive depenencies | |
for need in needs: | |
print(f'"{need}" --> "{job}" ') | |
else: | |
if changes: # first job, has `only:changes` | |
branches[str(changes)].append(job) | |
else: # first job, triggers off all commits | |
print(f'(*) --> "{job}" ') | |
if branches: | |
for branch in branches: | |
print(f'(*) --> if "{branch}" then') | |
for job in branches[branch]: | |
print(f' --> "{job}"') | |
print(f'else') # this is probably not how `else` is supposed to be used but it looks good. | |
print('}') | |
print('@enduml') |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment