1#!/usr/bin/env python3 2# Copyright © 2023 Collabora Ltd. 3# Authors: 4# Helen Koike <helen.koike@collabora.com> 5# 6# For the dependencies, see the requirements.txt 7# SPDX-License-Identifier: MIT 8 9 10import argparse 11from datetime import datetime, timedelta, timezone 12from typing import Dict, List 13 14import plotly.express as px 15import plotly.graph_objs as go 16from gitlab import Gitlab, base 17from gitlab.v4.objects import ProjectPipeline 18from gitlab_common import (GITLAB_URL, get_gitlab_pipeline_from_url, 19 get_token_from_default_dir, pretty_duration, 20 read_token) 21 22 23def calculate_queued_at(job) -> datetime: 24 started_at = job.started_at.replace("Z", "+00:00") 25 return datetime.fromisoformat(started_at) - timedelta(seconds=job.queued_duration) 26 27 28def calculate_time_difference(time1, time2) -> str: 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) -> str: 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( 44 job: base.RESTObject, tasks: List[Dict[str, str | datetime | timedelta]] 45) -> None: 46 queued_at = calculate_queued_at(job) 47 task_name = create_task_name(job) 48 49 tasks.append( 50 { 51 "Job": task_name, 52 "Start": job.created_at, 53 "Finish": queued_at, 54 "Duration": calculate_time_difference(job.created_at, queued_at), 55 "Phase": "Waiting dependencies", 56 } 57 ) 58 tasks.append( 59 { 60 "Job": task_name, 61 "Start": queued_at, 62 "Finish": job.started_at, 63 "Duration": calculate_time_difference(queued_at, job.started_at), 64 "Phase": "Queued", 65 } 66 ) 67 68 if job.finished_at: 69 tasks.append( 70 { 71 "Job": task_name, 72 "Start": job.started_at, 73 "Finish": job.finished_at, 74 "Duration": calculate_time_difference(job.started_at, job.finished_at), 75 "Phase": "Time spent running", 76 } 77 ) 78 else: 79 current_time = datetime.now(timezone.utc).isoformat() 80 tasks.append( 81 { 82 "Job": task_name, 83 "Start": job.started_at, 84 "Finish": current_time, 85 "Duration": calculate_time_difference(job.started_at, current_time), 86 "Phase": "In-Progress", 87 } 88 ) 89 90 91def generate_gantt_chart( 92 pipeline: ProjectPipeline, ci_timeout: float = 60 93) -> go.Figure: 94 if pipeline.yaml_errors: 95 raise ValueError("Pipeline YAML errors detected") 96 97 # Convert the data into a list of dictionaries for plotly 98 tasks: List[Dict[str, str | datetime | timedelta]] = [] 99 100 for job in pipeline.jobs.list(all=True, include_retried=True): 101 # we can have queued_duration without started_at when a job is canceled 102 if not job.queued_duration or not job.started_at: 103 continue 104 add_gantt_bar(job, tasks) 105 106 # Make it easier to see retried jobs 107 tasks.sort(key=lambda x: x["Job"]) 108 109 title = f"Gantt chart of jobs in pipeline <a href='{pipeline.web_url}'>{pipeline.web_url}</a>." 110 title += ( 111 f" Total duration {str(timedelta(seconds=pipeline.duration))}" 112 if pipeline.duration 113 else "" 114 ) 115 116 # Create a Gantt chart 117 default_colors = px.colors.qualitative.Plotly 118 fig: go.Figure = px.timeline( 119 tasks, 120 x_start="Start", 121 x_end="Finish", 122 y="Job", 123 color="Phase", 124 title=title, 125 hover_data=["Duration"], 126 color_discrete_map={ 127 "In-Progress": default_colors[3], # purple 128 "Waiting dependencies": default_colors[0], # blue 129 "Queued": default_colors[1], # red 130 "Time spent running": default_colors[2], # green 131 }, 132 ) 133 134 # Calculate the height dynamically 135 fig.update_layout(height=len(tasks) * 10, yaxis_tickfont_size=14) 136 137 # Add a deadline line to the chart 138 created_at = datetime.fromisoformat(pipeline.created_at.replace("Z", "+00:00")) 139 timeout_at = created_at + timedelta(minutes=ci_timeout) 140 fig.add_vrect( 141 x0=timeout_at, 142 x1=timeout_at, 143 annotation_text=f"{int(ci_timeout)} min Timeout", 144 fillcolor="gray", 145 line_width=2, 146 line_color="gray", 147 line_dash="dash", 148 annotation_position="top left", 149 annotation_textangle=90, 150 ) 151 152 return fig 153 154 155def main( 156 token: str | None, 157 pipeline_url: str, 158 output: str | None, 159 ci_timeout: float = 60, 160): 161 if token is None: 162 token = get_token_from_default_dir() 163 164 token = read_token(token) 165 gl = Gitlab(url=GITLAB_URL, private_token=token, retry_transient_errors=True) 166 167 pipeline, _ = get_gitlab_pipeline_from_url(gl, pipeline_url) 168 fig: go.Figure = generate_gantt_chart(pipeline, ci_timeout) 169 if output and "htm" in output: 170 fig.write_html(output) 171 elif output: 172 fig.update_layout(width=1000) 173 fig.write_image(output) 174 else: 175 fig.show() 176 177 178if __name__ == "__main__": 179 parser = argparse.ArgumentParser( 180 description="Generate the Gantt chart from a given pipeline." 181 ) 182 parser.add_argument("pipeline_url", type=str, help="URLs to the pipeline.") 183 parser.add_argument( 184 "-o", 185 "--output", 186 type=str, 187 help="Output file name. Use html or image suffixes to choose the format.", 188 ) 189 parser.add_argument( 190 "--token", 191 metavar="token", 192 help="force GitLab token, otherwise it's read from ~/.config/gitlab-token", 193 ) 194 parser.add_argument( 195 "--ci-timeout", 196 metavar="ci_timeout", 197 type=float, 198 default=60, 199 help="Time that marge-bot will wait for ci to finish. Defaults to one hour.", 200 ) 201 args = parser.parse_args() 202 main(args.token, args.pipeline_url, args.output, args.ci_timeout) 203