The environment hasn't settled yet, so I wrote a little. The test class is underway.
In short, I want to be able to get information with ChatOps without releasing the Jenkins screen. The environment hasn't settled yet, so I hit the Jenkins API experimentally for the time being. I tried to get JOB information and start JOB.
If ChatOps is provided, it should be OK without writing the JOB startup code.
#!/opt/anaconda3/bin/python3
# -*- coding: utf-8 -*-
'''RocketChat Jenkins information acquisition
Providing Jenkins information to the Rocket Chat channel
CSRF measures are taken by Jenkins settings
So not only auth information but also Jenkins-Crumb information is needed.
1.admin TOKEN information
→ admin Generate a personal access token from the user console.
Don't forget to make a note as it is generated each time.
2.Jenkins-Crumb information acquisition method
→ You need to issue a command. Obtained by executing the following and returning
The admin TOKEN setting is set.
curl -u 'admin:ADMINS_TOKEN' 'http://xxxxxxx/crumbIssuer/api/xml?xpath=concat(//crumbRequestField,":",//crumb)'
3.Please use your own library to send the results. Not implemented here.
RocketChatChannelManager -> sendMessageToRocketChat(channel, msg)1
Todo:
'''
################################################
# library
################################################
import json
import requests
import pandas as pd
import sys
from datetime import datetime
from dateutil import parser
from pprint import pprint
from pytz import timezone
################################################
#Get environment variables
################################################
#
#
#HEADERS definition
# headers = {
# 'Jenkins-Crumb': '4d0e2b6e2f75aff392422caa2fb7b1f924a4c518c4561743273953b05c0cabdb'}
#Authentication definition
# AUTH = ('admin', '11062c796463d71a04e56e5c2cf49a26fa')
#
# URL = 'http://192.168.179.3:3000'
#
################################################
# RocketChatJenkinsManager
################################################
class RocketChatJenkinsManager(object):
def __init__(self, HEADERS, AUTH, URL):
#Argument check type
if not isinstance(HEADERS, dict):
print(f'Arguments: HEADERS type is incorrect dict<-> {type(HEADERS)}')
raise TypeError
#Argument check type
if not isinstance(AUTH, tuple):
print(f'Arguments: AUTH type is incorrect tuple<-> {type(AUTH)}')
raise TypeError
#Argument check type
if not isinstance(URL, str):
print(f'Arguments: URL type is incorrect str<-> {type(URL)}')
raise TypeError
#Instance generation
self.HEADERS = HEADERS
self.AUTH = AUTH
self.URL = URL
def exchangeUnixtimeToTimestamp(self, unixtime):
'''Convert unixtime to timestamp
Args:
unixtime: float
Returns:
timestamp: str
Raises:
TypeError
Examples:
>>> jenkins = RocketChatJenkinsManager(HEADERS, AUTH, URL)
>>> jenkins.exchangeUnixtimeToTimestamp(1610870939803)
-> '2021/01/17 17:08:59'
Note:
'''
#Argument check type
#if not isinstance(unixtime, int):
# print(f'Arguments: unixtime type is incorrect int<-> {type(unixtime)}')
# raise TypeError
timestamp_succsessful = float(unixtime)
return datetime.fromtimestamp(timestamp_succsessful/1000.).strftime('%Y/%m/%d %H:%M:%S')
def getJenkinsJobList(self):
'''Get a list of available jobs in Jenkins
Returns a list of available Jenkins JOBs in Pandas DataFrame.
Args:
None
Returns:
pd.DataFrame
Jobname, Params, JobDescription, ExecCount, URL
Raises:
API runtime error
Examples:
>>> jenkins = RocketChatJenkinsManager(HEADERS, AUTH, URL)
>>> df = jenkins.getJenkinsJobList()
Note:
'''
#Columns definition (data acquisition)
columns_in_df = ['JobName', 'JobDescription', 'ExecCount', 'URL','Params']
#Columns definition (data output)
columns_out_df = ['JobName', 'Params', 'JobDescription', 'ExecCount', 'URL']
#API definition
ENDPOINT = '/api/json'
QUERY = '?depth=1&tree=jobs[displayName,description,lastCompletedBuild[number,url],actions[parameterDefinitions[name]]]'
API = f'{self.URL}{ENDPOINT}{QUERY}'
#Acquisition process
try:
response = requests.post(
API,
auth=self.AUTH,)
except Exception as e:
print(f'API execution error: {API}')
print(f'Error: {e}')
return False
else:
#Prepare a temporary container
_list1 = []
_list2 = []
#Get information from get response
##Job basic information acquisition
for _ in response.json()['jobs']:
_list1.append([_['displayName'],
_['description'],
_['lastCompletedBuild']['number'],
_['lastCompletedBuild']['url']])
##Variable parameter acquisition
_list3 = []
for __ in _['actions']:
if (__ != {}) & (__ != {'_class': 'com.cloudbees.plugins.credentials.ViewCredentialsAction'}):
_key = ''
for ___ in __['parameterDefinitions']:
_key += f"param: {___['name']} "
_list3.append(_key)
_list2.append(_list3)
##Output format processing
df1 = pd.DataFrame(_list1)
df2 = pd.DataFrame(_list2)
df = pd.concat([df1, df2], axis=1)
df.columns = columns_in_df
#Output fine adjustment
return df[columns_out_df]
def execJenkinsJobListNoParameters(self, jobname):
'''Remotely execute the specified Jenkins JOB (without parameters)
The role here is only to run Jenkins Job remotely
The Job execution result is not handled.
In the first place, it is a mechanism of asynchronous execution.
Args:
jobname:str Jenkins Job name, but Job without parameter definition
Return:
response: <Response [201]>Passed to the execution schedule
Raises:
API runtime error
Examples:
>>> jenkins = RocketChatJenkinsManager(HEADERS, AUTH, URL)
>>> jenkins.execJenkinsJobListNoParameters('test_hubot')
Note:
'''
#Argument check type
if not isinstance(jobname, str):
print(f'Argument: jobname type is incorrect str<-> {type(jobname)}')
raise TypeError
#API definition
ENDPOINT = f'/job/{jobname}/build'
API = f'{self.URL}{ENDPOINT}'
#Job input
try:
response = requests.post(
API,
headers=self.HEADERS,
auth=self.AUTH,)
except Exception as e:
print(f'API execution error: {API}')
print(f'Error: {e}')
return False
else:
#Job tells the success of remote submission
print(f'{jobname}Was executed remotely')
print(response)
return '201'
def execJenkinsJobListWithParameters(self, jobname):
'''Remotely execute the specified JenkinsJOB (with parameter settings)
The role here is only to run Jenkins Job remotely
The Job execution result is not handled.
In the first place, it is a mechanism of asynchronous execution.
Args:
jobname:str Jenkins Job name, but Job without parameter definition
Return:
response: <Response [201]>Passed to the execution schedule
Raises:
API runtime error
Examples:
>>> jenkins = RocketChatJenkinsManager(HEADERS, AUTH, URL)
>>> jenkins.execJenkinsJobListWithParameters('test_hubot')
Note:
'''
#Argument check type
if not isinstance(jobname, str):
print(f'Argument: jobname type is incorrect str<-> {type(jobname)}')
raise TypeError
#API definition
ENDPOINT = f'/job/{jobname}/buildWithParameters'
API = f'{self.URL}{ENDPOINT}'
#Job input
try:
response = requests.post(
API,
headers=self.HEADERS,
auth=self.AUTH,)
except Exception as e:
print(f'API execution error: {API}')
print(f'Error: {e}')
return False
else:
#Job Remote input success is reported
print(f'{jobname}Was executed remotely')
print(response)
return '201'
def _lastBuildTimestamp(self, jobname):
'''Get the last execution time of the specified JenkinsJOB (with parameter setting)
Args:
jobname:str Jenkins Job name, but Job without parameter definition
Retur:
timestamp:str Last successful Build time YYYY/MM/DD HH:MM:SS
Raises:
API runtime error
Examples:
>>> jenkins = RocketChatJenkinsManager(HEADERS, AUTH, URL)
>>> jenkins.lastSuccessfulBuildTimestamp('test_hubot')
Note:
'''
#Argument check type
if not isinstance(jobname, str):
print(f'Argument: jobname type is incorrect str<-> {type(jobname)}')
raise TypeError
#API definition
ENDPOINT = f'/job/{jobname}/lastBuild/api/json'
API = f'{self.URL}{ENDPOINT}'
#JOB parameter definition
params = (
('pretty', 'true'),
)
#Job input
try:
response = requests.post(
API,
headers=self.HEADERS,
params=params,
auth=self.AUTH,)
except Exception as e:
print(f'API execution error: {API}')
print(f'Error: {e}')
return False
else:
#Convert unixtime to timestamp and back
return(self.exchangeUnixtimeToTimestamp(response.json()['timestamp']))
def _lastSuccessfulBuildTimestamp(self, jobname):
'''Get the last success time of the specified JenkinsJOB (with parameter setting)
Args:
jobname:str Jenkins Job name, but Job without parameter definition
Retur:
timestamp:str Last successful Build time YYYY/MM/DD HH:MM:SS
Raises:
API runtime error
Examples:
>>> jenkins = RocketChatJenkinsManager(HEADERS, AUTH, URL)
>>> jenkins.lastSuccessfulBuildTimestamp('test_hubot')
Note:
'''
#Argument check type
if not isinstance(jobname, str):
print(f'Argument: jobname type is incorrect str<-> {type(jobname)}')
raise TypeError
#API definition
ENDPOINT = f'/job/{jobname}/lastSuccessfulBuild/api/json'
API = f'{self.URL}{ENDPOINT}'
#JOB parameter definition
params = (
('pretty', 'true'),
)
#Job input
try:
response = requests.post(
API,
headers=self.HEADERS,
params=params,
auth=self.AUTH,)
except Exception as e:
print(f'API execution error: {API}')
print(f'Error: {e}')
return False
else:
#Convert unixtime to timestamp and back
return(self.exchangeUnixtimeToTimestamp(response.json()['timestamp']))
def _lastFailedBuildTimestamp(self, jobname):
'''Get the last failure time of the specified JenkinsJOB (with parameter setting)
Args:
jobname:str Jenkins Job name, but Job without parameter definition
Retur:
timestamp:str Last successful Build time YYYY/MM/DD HH:MM:SS
Raises:
API runtime error
Examples:
>>> jenkins = RocketChatJenkinsManager(HEADERS, AUTH, URL)
>>> jenkins.lastFailedBuildTimestamp('test_hubot')
Note:
'''
#Argument check type
if not isinstance(jobname, str):
print(f'Argument: jobname type is incorrect str<-> {type(jobname)}')
raise TypeError
#API definition
ENDPOINT = f'/job/{jobname}/lastFailedBuild/api/json'
API = f'{self.URL}{ENDPOINT}'
#JOB parameter definition
params = (
('pretty', 'true'),
)
#Job input
try:
response = requests.post(
API,
headers=self.HEADERS,
params=params,
auth=self.AUTH,)
except Exception as e:
print(f'API execution error: {API}')
print(f'Error: {e}')
return False
else:
#Convert unixtime to timestamp and back
return(self.exchangeUnixtimeToTimestamp(response.json()['timestamp']))
def getJobInformation(self, jobname):
'''Get execution information of specified JenkinsJOB (with parameter setting)
Args:
jobname:str Jenkins Job name, but Job without parameter definition
Return:
joburl: str
job information: DataFrame
Raises:
API runtime error
Examples:
>>> jenkins = RocketChatJenkinsManager(HEADERS, AUTH, URL)
>>> jenkins.getJobInformation('test_hubot')
Note:
'''
#Argument check type
if not isinstance(jobname, str):
print(f'Argument: jobname type is incorrect str<-> {type(jobname)}')
raise TypeError
#API definition
ENDPOINT = f'/job/{jobname}/api/json'
API = f'{self.URL}{ENDPOINT}'
#JOB parameter definition
params = (
('pretty', 'true'),
)
#Job input
try:
response = requests.post(
API,
headers=self.HEADERS,
params=params,
auth=self.AUTH,)
except Exception as e:
print(f'API execution error: {API}')
print(f'Error: {e}')
return False
else:
#JobLink Path generation
joburl = f"{response.json()['url']}"
#Summarize basic information
_list = []
_list.append([f"Job name", f"{response.json()['displayName']}"])
_list.append([f"Job details", f"{response.json()['description']}"])
_list.append([f"HealthReport", f"{response.json()['healthReport'][0]['description']}"])
_list.append([f"JobStatus Color", f"{response.json()['color']}"])
# _list.append([f"Job latest execution:Failure judgment", f"{response.json()['lastUnstableBuild']}"])
_list.append([f"Job Final Build No.", f"{response.json()['lastBuild']['number']}"])
_list.append([f"Job Last Build Time", f"{self._lastBuildTimestamp(jobname)}"])
_list.append([f"Job Final Success Build No.", f"{response.json()['lastSuccessfulBuild']['number']}"])
_list.append([f"Job Last Success Build URL", f"[link]({response.json()['lastSuccessfulBuild']['url']}consoleText)"])
_list.append([f"Job Last Success Build Time", f"{self._lastSuccessfulBuildTimestamp(jobname)}"])
_list.append([f"Job final failure BuildNo.", f"{response.json()['lastFailedBuild']['number']}"])
_list.append([f"Job Last Failure Build URL", f"[link]({response.json()['lastFailedBuild']['url']}consoleText)"])
_list.append([f"Job Last Failure Build Time", f"{self._lastFailedBuildTimestamp(jobname)}"])
#DataFrame generation
df = pd.DataFrame(_list)
df.columns = ['item', 'status']
return joburl, df
Recommended Posts