1#!/usr/bin/env python3 2# Copyright © 2023 Collabora Ltd. 3# Authors: 4# Helen Koike <[email protected]> 5# 6# For the dependencies, see the requirements.txt 7# SPDX-License-Identifier: MIT 8 9 10import argparse 11import gitlab 12import plotly.express as px 13from gitlab_common import pretty_duration 14from datetime import datetime, timedelta 15from gitlab_common import read_token, GITLAB_URL, get_gitlab_pipeline_from_url 16 17 18def calculate_queued_at(job): 19 # we can have queued_duration without started_at when a job is canceled 20 if not job.queued_duration or not job.started_at: 21 return None 22 started_at = job.started_at.replace("Z", "+00:00") 23 return datetime.fromisoformat(started_at) - timedelta(seconds=job.queued_duration) 24 25 26def calculate_time_difference(time1, time2): 27 if not time1 or not time2: 28 return None 29 if type(time1) is str: 30 time1 = datetime.fromisoformat(time1.replace("Z", "+00:00")) 31 if type(time2) is str: 32 time2 = datetime.fromisoformat(time2.replace("Z", "+00:00")) 33 34 diff = time2 - time1 35 return pretty_duration(diff.seconds) 36 37 38def create_task_name(job): 39 status_color = {"success": "green", "failed": "red"}.get(job.status, "grey") 40 return f"{job.name}\t(<span style='color: {status_color}'>{job.status}</span>,<a href='{job.web_url}'>{job.id}</a>)" 41 42 43def add_gantt_bar(job, tasks): 44 queued_at = calculate_queued_at(job) 45 task_name = create_task_name(job) 46 47 tasks.append( 48 { 49 "Job": task_name, 50 "Start": job.created_at, 51 "Finish": queued_at, 52 "Duration": calculate_time_difference(job.created_at, queued_at), 53 "Phase": "Waiting dependencies", 54 } 55 ) 56 tasks.append( 57 { 58 "Job": task_name, 59 "Start": queued_at, 60 "Finish": job.started_at, 61 "Duration": calculate_time_difference(queued_at, job.started_at), 62 "Phase": "Queued", 63 } 64 ) 65 tasks.append( 66 { 67 "Job": task_name, 68 "Start": job.started_at, 69 "Finish": job.finished_at, 70 "Duration": calculate_time_difference(job.started_at, job.finished_at), 71 "Phase": "Running", 72 } 73 ) 74 75 76def generate_gantt_chart(pipeline): 77 if pipeline.yaml_errors: 78 raise ValueError("Pipeline YAML errors detected") 79 80 # Convert the data into a list of dictionaries for plotly 81 tasks = [] 82 83 for job in pipeline.jobs.list(all=True, include_retried=True): 84 add_gantt_bar(job, tasks) 85 86 # Make it easier to see retried jobs 87 tasks.sort(key=lambda x: x["Job"]) 88 89 title = f"Gantt chart of jobs in pipeline <a href='{pipeline.web_url}'>{pipeline.web_url}</a>." 90 title += ( 91 f" Total duration {str(timedelta(seconds=pipeline.duration))}" 92 if pipeline.duration 93 else "" 94 ) 95 96 # Create a Gantt chart 97 fig = px.timeline( 98 tasks, 99 x_start="Start", 100 x_end="Finish", 101 y="Job", 102 color="Phase", 103 title=title, 104 hover_data=["Duration"], 105 ) 106 107 # Calculate the height dynamically 108 fig.update_layout(height=len(tasks) * 10, yaxis_tickfont_size=14) 109 110 # Add a deadline line to the chart 111 created_at = datetime.fromisoformat(pipeline.created_at.replace("Z", "+00:00")) 112 timeout_at = created_at + timedelta(hours=1) 113 fig.add_vrect( 114 x0=timeout_at, 115 x1=timeout_at, 116 annotation_text="1h Timeout", 117 fillcolor="gray", 118 line_width=2, 119 line_color="gray", 120 line_dash="dash", 121 annotation_position="top left", 122 annotation_textangle=90, 123 ) 124 125 return fig 126 127 128def parse_args() -> None: 129 parser = argparse.ArgumentParser( 130 description="Generate the Gantt chart from a given pipeline." 131 ) 132 parser.add_argument("pipeline_url", type=str, help="URLs to the pipeline.") 133 parser.add_argument( 134 "-o", 135 "--output", 136 type=str, 137 help="Output file name. Use html ou image suffixes to choose the format.", 138 ) 139 parser.add_argument( 140 "--token", 141 metavar="token", 142 help="force GitLab token, otherwise it's read from ~/.config/gitlab-token", 143 ) 144 return parser.parse_args() 145 146 147if __name__ == "__main__": 148 args = parse_args() 149 150 token = read_token(args.token) 151 152 gl = gitlab.Gitlab(url=GITLAB_URL, private_token=token, retry_transient_errors=True) 153 154 pipeline, _ = get_gitlab_pipeline_from_url(gl, args.pipeline_url) 155 fig = generate_gantt_chart(pipeline) 156 if args.output and "htm" in args.output: 157 fig.write_html(args.output) 158 elif args.output: 159 fig.update_layout(width=1000) 160 fig.write_image(args.output) 161 else: 162 fig.show() 163