Extract Jobs/Build Metadata From Jenkins

""" This script retrieves data from the Jenkins server and saves it to a CSV file. IMPORTANT: IMPORTANT: This script takes forever to run. It took me a couple of hours to create and test. I use it now and then but for sure there are many things to improve. I have never needed to make it run faster or solve its issues. If you fix it, please let me know to check what I mess up. """ import xml.etree.ElementTree as ET from time import strftime, localtime import jenkins import pandas as pd # Jenkins server information jenkins_url = 'https://ae-jenkins.prod.analytics.aws.kixeye.com/' # CHANGE ME <------------------------------------------ username = 'cgomez' # CHANGE ME <------------------------------------------ password = 'S0lr4c123$' # CHANGE ME <------------------------------------------ # Initialize the Jenkins server connection server = jenkins.Jenkins(jenkins_url, username=username, password=password) # Initialize global variables job_name = "No name" job_url = "No URL" job_fullname = "No fullname" job_color = "No color" cron_schedule = "No cron schedule" warehouse = "No warehouse" last_duration = -1 last_execution_date = -1 last_build_number = -1 def get_last_build_number(): """ Get the last build number of a job :return: Last build number """ # Get job information job_info = server.get_job_info(job_name) # Initialize the build number build_num = -1 try: # Get details of the last build build_num = job_info['lastBuild']['number'] except: print(f"- Job {job_name} does not exist") return build_num def get_build_info(build_number): build_info = -1 try: # Get build information build_info = server.get_build_info(job_name, build_number) except: print(f"- Job {job_name} with build number {build_number} does not exist") # Get details of the last build return build_info def get_data_from_configuration(xml_root, tag): """ Function to retrieve data from the XML configuration file Parameters: xml_root (xml.etree.ElementTree.Element): XML root element tag (str): XML tag job_name (str): Job name """ tagvalue = "No value" try: tagvalue = xml_root.find(f".//{tag}").text except: print(f"- Job {job_name} does not a {tag} attribute") return tagvalue def get_data_from_job(job, tag): """ Function to retrieve data from the job object :param job: Job object :param tag: Tag to find in the job object :return: Value of the tag """ tag_value = "No value" try: tag_value = job[tag] except: print(f"- Job {job_name} does not have a {tag} attribute") return tag_value def get_duration(build_info): # Initialize the duration duration = -1 # Get the last duration and last execution date try: duration = build_info['actions'][2]['executingTimeMillis'] except: print(f"- Job {job_name} does not have a last duration") return duration def get_execution_date(build_info): # Initialize the duration execution_date = -1 try: execution_date = build_info['timestamp'] except: print(f"- Job {job_name} does not have a last execution date") return execution_date def get_jobs_data(job): """ Retrieve data from the job object :param job: Job object """ global job_name global job_url global job_fullname global job_color # Extract the job name, URL, and XML configuration job_name = get_data_from_job(job, 'name') job_url = get_data_from_job(job, 'url') job_fullname = get_data_from_job(job, 'fullname') job_color = get_data_from_job(job, 'color') def get_configuration_data(): """ Retrieve data from the job's configuration XML file """ global cron_schedule global warehouse try: job_config = server.get_job_config(job_name) # Parse the XML content root = ET.fromstring(job_config) # Extract the cron schedule and warehouse name tag = "hudson.triggers.TimerTrigger/spec" cron_schedule = get_data_from_configuration(root, tag) # Extract the warehouse name tag = "hudson.model.StringParameterDefinition[name='ETL_SNOWFLAKE_WAREHOUSE']/defaultValue" warehouse = get_data_from_configuration(root, tag) except: print(f"- Job {job_name} does not have a configuration file") def get_build_data(): """ Retrieve data from the last build of a job """ global last_duration global last_execution_date global last_build_number # Extract the last duration (in milliseconds) and last execution date last_build_number = get_last_build_number() # If the job has been executed at least once if last_build_number > 0: build_info = get_build_info(last_build_number) # Check if the build_info is not empty if bool(build_info): last_duration = get_duration(build_info) last_execution_date = get_execution_date(build_info) / 1000 def get_jenkins_data(): """ Function to retrieve data from the Jenkins server :return: data (list): List of lists with the data """ jobs = server.get_jobs() data = [] for job in jobs: get_jobs_data(job) get_configuration_data() get_build_data() print(f"- Job {job_name} has been processed") # Append the data to the list data.append([job_name, job_url, job_fullname, job_color, cron_schedule, last_duration / 60000, warehouse, last_duration, strftime('%Y-%m-%d %H:%M:%S', localtime(last_execution_date)), last_execution_date, last_build_number]) return data def array_to_csv(data): """ Convert an array to a CSV file """ # Create a Pandas DataFrame and save it to a CSV file df = pd.DataFrame(data, columns=['Name', 'URL', 'Fullname', 'Color', 'Cron Schedule', 'Last Duration (min)', 'Warehouse', 'Last Duration (ms)', 'Last Execution DT', 'Last Execution Epoch', 'Last Build Number']) # Save the DataFrame to a CSV file df.to_csv('jenkins_data.csv', index=False) if __name__ == "__main__": array_to_csv(get_jenkins_data())