#!/usr/bin/env python3 # Copyright (C) 2020 Simon Quigley # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import datetime import requests_cache import time from jenkinsapi.custom_exceptions import NoBuildData from jenkinsapi.jenkins import Jenkins from jinja2 import Template from os import getenv, makedirs, path requests_cache.install_cache("jenkins", backend="sqlite", expire_after=300) class JenkinsModule: """Jenkins module for the Metrics program""" def __init__(self): self.name = "Jenkins" def _auth_jenkins_server(self): """Authenticate to the Jenkins server This uses the API_SITE, API_USER, and API_KEY env vars. """ # Load the API values from the environment variables api_site = getenv("API_SITE") api_user = getenv("API_USER") api_key = getenv("API_KEY") for envvar in [api_site, api_user, api_key]: if not envvar: raise ValueError("API_SITE, API_USER, and API_KEY must be", "defined") # Authenticate to the server server = Jenkins(api_site, username=api_user, password=api_key) return server def _get_data(self): """Get the data from the Jenkins server This function returns three distinct values as one list: [nonpassing, failing, total] """ # Authenticate to the server server = self._auth_jenkins_server() # Initialize the data, and get the total jobs on the server data = [0, 0, len(server.jobs.keys())] # jenkinsapi has a built-in method for iterating on jobs # val will always be a jenkins Job class for val in server.jobs.itervalues(): # If we come across a job that has no build, make it a SUCCESS # The goal of this is to identify problematic jobs, and jobs with # no existing builds aren't necessarily problematic (yet) try: status = val.get_last_build().get_status() except NoBuildData: status = "SUCCESS" # If it's not successful, add it to nonpassing, since failing is # reserved for jobs with the specific status of FAILURE if status != "SUCCESS": data[0] += 1 if status == "FAILURE": data[1] += 1 return data def sqlite_setup(self): """Initially set up the table for usage in SQLite This returns a str which will then be executed in our SQLite db Here is the "jenkins" table layout: - date is the primary key, and it is the Unix timestamp as an int - nonpassing is the number of !(SUCCESS) jobs as an int - failing is the number of FAILURE jobs as an int - total is the total number of jobs on the Jenkins server as an int """ command = "CREATE TABLE IF NOT EXISTS jenkins (date INTEGER PRIMARY " command += "KEY, nonpassing INTEGER, failing INTEGER, total INTEGER);" return command def sqlite_add(self): """Add data to the SQLite db This retrieves the current data from the Jenkins server, and returns a str which will then be executed in our SQLite db """ # Match the variable names with the column names in the db nonpassing, failing, total = self._get_data() date = "strftime('%s', 'now')" # Craft the str command = "INSERT INTO jenkins VALUES ({}, {}, {}, {});".format( date, nonpassing, failing, total) return command def sqlite_time_range(self, days): """Get the rows which have been inserted given days e.g. if days is 180, it gets all of the values which have been inserted in the past 180 days. Note: this just returns the command to be ran, it doesn't actually run """ now = datetime.datetime.now() timedelta = datetime.timedelta(days=days) unix_time = int(time.mktime((now - timedelta).timetuple())) command = "SELECT * FROM jenkins WHERE date > %s;" % unix_time return command def render_template(self, days, data): """Render a template with days in the filename, given the data The above function sqlite_time_range() is ran on the database with some predetermined date ranges. This function actually interprets that data, uses Jinja2 to magically render a template, and voila. """ # Initialize a (softly) ephemeral dict to store data jenkins = {} _data = {"date": [], "nonpassing": [], "failing": [], "total": []} # Put the data from the DB query into _data for row in data: _data["date"].append(row[0]) _data["nonpassing"].append(row[1]) _data["failing"].append(row[2]) _data["total"].append(row[3]) # Get human-readable averages and throw it in a dict _nonpassing = sum(_data["nonpassing"]) / len(_data["nonpassing"]) _nonpassing = format(_nonpassing, ".1f") _failing = sum(_data["failing"]) / len(_data["failing"]) _failing = format(_failing, ".1f") _total = sum(_data["total"]) / len(_data["total"]) _total = format(_total, ".1f") average = {"nonpassing": _nonpassing, "failing": _failing, "total": _total} # Assign data to the dict Jinja2 is actually going to use jenkins = {"nonpassing": zip(_data["date"], _data["nonpassing"]), "failing": zip(_data["date"], _data["failing"]), "total": zip(_data["date"], _data["total"])} # Grab our template from templates/ and store it as a Template t_path = path.join("templates", "jenkins.html") with open(t_path) as templatef: template = "" for text in templatef.readlines(): template += text template = Template(template) # Render the template template = template.render(jenkins=jenkins, average=average, days=days) # Make the output dir if it doesn't already exist if not path.exists("output"): makedirs("output") # Write it back to the filename in the output dir with open(path.join("output", "jenkins_%sdays.html" % days), "w+") as f: f.write(template) # Return the averages for use in the summary return (average["nonpassing"], average["failing"], average["total"])