jenkins-利用Python API高效管理job与build状态监控
1. 为什么你需要用Python来“遥控”Jenkins?
如果你和我一样,每天都要和Jenkins打交道,手动点开几十个甚至上百个Job去看构建状态、检查日志,那感觉就像在玩一个永远也点不完的“找不同”游戏,既枯燥又低效。特别是当团队项目越来越多,CI/CD流水线越来越复杂时,这种重复劳动简直让人崩溃。你可能也想过,有没有一种方法,能让我在一个地方就看到所有Job的健康状况,或者自动处理一些任务?答案是肯定的,而且比你想象的要简单——那就是利用Jenkins自带的API,再配上我们熟悉的Python。
Jenkins的Web界面虽然直观,但它本质上是一个可以通过HTTP请求进行交互的服务。它提供了一套非常完善的REST API,这意味着你几乎可以通过代码完成所有在网页上能做的操作。而Python,凭借其简洁的语法和强大的requests库,就成了连接你和Jenkins自动化世界的最佳桥梁。我刚开始接触时也觉得有点神秘,但实际用下来发现,这就像是给Jenkins装了一个“遥控器”,你可以编程控制它,而不是每次都亲自跑到“机器”面前去按按钮。
那么,具体能做什么呢?想象一下这些场景:每天早上,你不需要打开浏览器,一个脚本就能把昨晚所有失败的构建报告发到你的钉钉或企业微信;你可以批量禁用那些已经不再使用的老旧Job,清理空间;你可以自动监控关键流水线的构建时长,一旦超时就触发告警;甚至,你可以根据Git仓库的提交情况,动态地创建或配置新的Job。这些都不是空想,而是我团队里每天都在运行的自动化操作。接下来,我就带你从零开始,一步步搭建起这个高效的“遥控”系统。
2. 环境准备:让你的Python能和Jenkins“对话”
工欲善其事,必先利其器。要让Python和Jenkins顺利沟通,我们需要先准备好两样东西:一个是Python环境及必要的库,另一个就是Jenkins上的访问凭证。别担心,这个过程就像给两个新朋友做介绍一样简单。
2.1 安装Python和必备的“通信员”库
首先,确保你的电脑上安装了Python 3.6或以上版本。这个应该大部分搞运维或开发的兄弟都已经有了。接下来,我们需要请出两位核心“通信员”:
requests库:这是Python里进行HTTP请求的“瑞士军刀”,我们和Jenkins API的所有交互都要靠它。python-jenkins库(可选但强烈推荐):这是一个对Jenkins API进行了封装的第三方库。它把很多底层的HTTP请求细节都包装成了更友好的函数。对于初学者,我建议从requests开始,理解原理;等熟练了,再用python-jenkins来提升开发效率,减少重复代码。
安装它们只需要一行命令。打开你的终端(Windows用CMD或PowerShell,Mac/Linux用Terminal),输入:
pip install requests python-jenkins
如果速度慢,可以加上国内的镜像源,比如:
pip install requests python-jenkins -i https://pypi.tuna.tsinghua.edu.cn/simple
看到“Successfully installed”的字样,就说明你的Python已经装备好了和Jenkins握手言和的工具。
2.2 在Jenkins上创建API访问令牌
现在,我们需要告诉Jenkins:“嗨,这是我的Python脚本,它以后会代表我来找你办事,请给它开个门。” 这个“门”就是API Token。比起直接使用你的登录密码,使用Token更安全,权限也更可控。
具体操作步骤如下:
- 登录你的Jenkins网页,点击右上角你的用户名,然后选择 “设置”(Configure)。
- 在设置页面中,找到 “API Token” 区域,点击 “添加新Token”。
- 给它起个名字,比如“python_script_token”,然后点击生成。千万注意! 生成的Token只会显示这一次,请务必立即把它复制并保存到安全的地方(比如密码管理器)。一旦关闭页面,你就再也看不到它了。
- 同时,记下你的Jenkins服务器地址(如
http://jenkins.your-company.com:8080)和用户名。
有了这三样东西——Jenkins地址、用户名、API Token——你的Python脚本就有了访问Jenkins的“身份证”。我这里踩过一个坑:早期我图省事直接用密码,后来公司安全审计差点出问题。所以,养成用Token的好习惯,安全又规范。
3. 核心实战:用Python API获取所有Job列表
基础打好了,我们来干第一件实事:获取Jenkins上所有的Job名称列表。这是所有自动化管理的基础,就像你要管理一个仓库,总得先知道里面有哪些货物吧。
3.1 理解Jenkins API的数据结构
Jenkins的API返回的数据格式主要是JSON,这是一种非常易于程序读写的结构。当你访问 {你的Jenkins地址}/api/json 这个URL时(需要带上认证信息),Jenkins会返回一个巨大的JSON对象,里面包含了关于它自身的几乎所有信息:系统状态、视图、插件,当然,还有我们最关心的 jobs 数组。
这个jobs数组里的每一个元素,就对应着一个Job。每个Job元素本身又是一个JSON对象,里面包含了这个Job的详细信息,比如name(名称)、url(地址)、color(构建状态对应的颜色,如blue代表成功,red代表失败)等。我们第一步,就是从这个结构中把每个Job的名字“挖”出来。
3.2 编写你的第一个“侦察兵”脚本
理论有点枯燥,我们直接上代码。下面这个函数,就是派出去的“侦察兵”,它的任务就是摸清Jenkins上所有Job的底细。
import requests
from requests.auth import HTTPBasicAuth
def get_all_jobs(jenkins_url, username, api_token):
"""
获取Jenkins服务器上所有的Job名称列表。
参数:
jenkins_url (str): Jenkins服务器地址,如 'http://jenkins.example.com:8080'
username (str): Jenkins用户名
api_token (str): 对应用户的API Token
返回:
list: 包含所有Job名称的列表,如果失败则返回空列表。
"""
# 构建API请求的URL
api_url = f"{jenkins_url.rstrip('/')}/api/json"
# 使用HTTP基本认证,注意密码处使用API Token
auth = HTTPBasicAuth(username, api_token)
try:
# 发送GET请求
response = requests.get(api_url, auth=auth, timeout=30)
# 检查HTTP状态码是否为200(成功)
if response.status_code == 200:
# 解析返回的JSON数据
data = response.json()
# 从数据中提取jobs数组
jobs = data.get("jobs", [])
# 使用列表推导式提取每个job的‘name’字段
job_names = [job["name"] for job in jobs]
print(f"成功获取到 {len(job_names)} 个Job。")
return job_names
else:
# 请求失败,打印错误信息
print(f"请求失败!状态码:{response.status_code}, 响应内容:{response.text[:200]}...")
return []
except requests.exceptions.ConnectionError:
print(f"错误:无法连接到Jenkins服务器,请检查地址 {jenkins_url} 是否正确或网络是否通畅。")
return []
except requests.exceptions.Timeout:
print("错误:请求超时,Jenkins服务器响应过慢。")
return []
except Exception as e:
print(f"获取Job列表时发生未知错误:{e}")
return []
# 使用示例
if __name__ == "__main__":
# 替换成你自己的信息
JENKINS_URL = "http://your-jenkins-server.com:8080"
USERNAME = "your_username"
API_TOKEN = "your_api_token_here" # 注意:这里填的是Token,不是密码!
all_jobs = get_all_jobs(JENKINS_URL, USERNAME, API_TOKEN)
if all_jobs:
print("所有Job列表:")
for idx, job_name in enumerate(all_jobs, 1):
print(f"{idx}. {job_name}")
else:
print("未能获取到Job列表。")
让我解释一下这段代码的几个关键点:
- 认证:
HTTPBasicAuth(username, api_token)是核心。Jenkins API使用HTTP基本认证,这里千万注意,第二个参数是api_token,不是你登录网页的密码。 - 错误处理:我用了
try...except包裹了请求,并专门处理了连接错误和超时错误。在实际生产环境,网络波动、服务重启都是常事,健壮的错误处理能让你的脚本更可靠。 - 数据提取:
data.get("jobs", [])比直接写data["jobs"]更安全。如果API返回的数据结构意外变化(比如没有jobs字段),.get()方法会返回一个空列表[],避免程序因KeyError而崩溃。 - 超时设置:
timeout=30给请求设置了一个30秒的超时。没有这个,如果Jenkins挂掉,你的脚本可能会一直卡住。
把上面的代码保存为jenkins_job_lister.py,替换里面的URL、用户名和Token,然后运行它。如果一切顺利,你应该能在终端里看到一长串Job名字。恭喜你,你已经成功迈出了Jenkins自动化管理的第一步!
4. 深度监控:获取每个Job的最新构建状态和时间
拿到Job列表只是第一步,我们更关心的是它们的最新状态:哪个Job刚失败了?哪个Job已经很久没构建了?这些才是运维监控的核心。接下来,我们就来编写一个更强大的“监控员”脚本。
4.1 解析“最后构建”API的丰富信息
每个Job都有一个 lastBuild 的链接,指向其最近一次构建。访问 {Job_URL}/lastBuild/api/json 可以获得关于这次构建的详尽信息。我们最关心的几个字段通常是:
result: 构建结果,如SUCCESS,FAILURE,UNSTABLE,ABORTED。timestamp: 构建开始的时间戳(毫秒级)。duration: 构建耗时(毫秒)。building: 布尔值,表示构建是否正在进行中。url: 本次构建的详细页面地址,方便我们快速跳转查看日志。
我们的目标是获取每个Job的最后构建时间和构建结果,并把它们清晰地展示出来。
4.2 构建一个健壮的状态监控脚本
直接循环调用API获取每个Job的状态虽然可行,但如果Job数量很多,串行请求会非常慢。这里我们可以引入Python的concurrent.futures模块进行简单的并发请求,大幅提升效率。同时,我们会把结果组织得更好看一些。
import requests
import concurrent.futures
from datetime import datetime
from requests.auth import HTTPBasicAuth
def get_job_last_build_info(jenkins_url, job_name, username, api_token):
"""
获取指定Job最后一次构建的详细信息。
参数:
jenkins_url (str): Jenkins服务器地址
job_name (str): Job名称
username (str): Jenkins用户名
api_token (str): API Token
返回:
dict: 包含构建状态、时间、耗时等信息的字典,如果失败或没有构建记录则返回None。
"""
# 构建查询最后一次构建信息的API URL
api_url = f"{jenkins_url.rstrip('/')}/job/{job_name}/lastBuild/api/json"
auth = HTTPBasicAuth(username, api_token)
try:
response = requests.get(api_url, auth=auth, timeout=15)
if response.status_code == 200:
data = response.json()
# 将时间戳(毫秒)转换为可读的日期时间字符串
timestamp_ms = data.get('timestamp', 0)
if timestamp_ms:
build_time = datetime.fromtimestamp(timestamp_ms / 1000).strftime('%Y-%m-%d %H:%M:%S')
else:
build_time = "N/A"
# 获取构建结果,如果正在构建中,则结果为None,我们用‘BUILDING’表示
build_result = data.get('result')
if data.get('building'):
build_result = "BUILDING"
# 计算构建耗时(秒)
duration_ms = data.get('duration', 0)
duration_sec = duration_ms / 1000 if duration_ms else 0
return {
'job_name': job_name,
'result': build_result,
'time': build_time,
'duration_seconds': round(duration_sec, 2),
'url': data.get('url', ''),
'building': data.get('building', False)
}
elif response.status_code == 404:
# 404通常意味着该Job从未被构建过
print(f"提示:Job '{job_name}' 暂无构建记录。")
return {
'job_name': job_name,
'result': 'NO_BUILD',
'time': 'N/A',
'duration_seconds': 0,
'url': '',
'building': False
}
else:
print(f"警告:获取Job '{job_name}' 信息失败,状态码:{response.status_code}")
return None
except requests.exceptions.RequestException as e:
print(f"请求Job '{job_name}' 时发生错误:{e}")
return None
def monitor_all_jobs(jenkins_url, username, api_token, max_workers=5):
"""
并发监控所有Job的最后构建状态。
参数:
jenkins_url (str): Jenkins服务器地址
username (str): 用户名
api_token (str): API Token
max_workers (int): 并发线程数,根据网络和服务器性能调整
返回:
list: 所有Job的构建信息字典列表
"""
print("开始获取所有Job列表...")
all_jobs = get_all_jobs(jenkins_url, username, api_token)
if not all_jobs:
print("未获取到任何Job,监控终止。")
return []
print(f"开始并发获取 {len(all_jobs)} 个Job的构建状态(最大并发数:{max_workers})...")
job_info_list = []
# 使用线程池进行并发请求
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# 创建一个将参数绑定到函数的lambda,方便提交任务
future_to_job = {
executor.submit(get_job_last_build_info, jenkins_url, job, username, api_token): job
for job in all_jobs
}
# 遍历完成的任务
for future in concurrent.futures.as_completed(future_to_job):
job_name = future_to_job[future]
try:
info = future.result(timeout=20) # 设置单个任务结果获取超时
if info:
job_info_list.append(info)
except concurrent.futures.TimeoutError:
print(f"获取Job '{job_name}' 状态超时。")
except Exception as exc:
print(f"处理Job '{job_name}' 时产生异常:{exc}")
# 按构建时间倒序排列,最新的排前面
job_info_list.sort(key=lambda x: x['time'] if x['time'] != 'N/A' else '1970-01-01', reverse=True)
return job_info_list
# 使用示例和结果展示
if __name__ == "__main__":
JENKINS_URL = "http://your-jenkins-server.com:8080"
USERNAME = "your_username"
API_TOKEN = "your_api_token_here"
all_build_info = monitor_all_jobs(JENKINS_URL, USERNAME, API_TOKEN, max_workers=10)
if all_build_info:
print("\n" + "="*120)
print(f"{'Job名称':<40} {'构建结果':<12} {'最后构建时间':<20} {'耗时(秒)':<10} {'状态':<10}")
print("-"*120)
success_count = 0
failure_count = 0
other_count = 0
for info in all_build_info:
result = info['result']
# 根据结果简单分类计数
if result == 'SUCCESS':
result_display = f"\033[92m{result}\033[0m" # 绿色
success_count += 1
elif result in ['FAILURE', 'UNSTABLE']:
result_display = f"\033[91m{result}\033[0m" # 红色
failure_count += 1
elif result == 'BUILDING':
result_display = f"\033[93m{result}\033[0m" # 黄色
other_count += 1
elif result == 'NO_BUILD':
result_display = f"\033[90m{result}\033[0m" # 灰色
other_count += 1
else:
result_display = result
other_count += 1
building_indicator = "▶ 运行中" if info['building'] else "✓ 已完成"
print(f"{info['job_name']:<40} {result_display:<20} {info['time']:<20} {info['duration_seconds']:<10.1f} {building_indicator:<10}")
print("="*120)
print(f"统计:成功({success_count}) | 失败/不稳定({failure_count}) | 其他/无构建({other_count}) | 总计({len(all_build_info)})")
print("="*120)
# 可以进一步处理,比如将失败的任务列表单独保存或发送通知
failed_jobs = [info for info in all_build_info if info['result'] in ['FAILURE', 'UNSTABLE']]
if failed_jobs:
print(f"\n\033[91m注意:以下 {len(failed_jobs)} 个Job构建失败或状态不稳定,请及时检查:\033[0m")
for job in failed_jobs:
print(f" - {job['job_name']} (最后构建于:{job['time']})")
这个脚本比第一个要复杂一些,但功能也强大得多:
- 并发处理:使用
ThreadPoolExecutor同时查询多个Job的状态,速度比一个个问快了好几倍。max_workers参数可以控制并发数,别设太高把Jenkins服务器压垮了。 - 完整的状态处理:它区分了“构建成功”、“构建失败”、“正在构建”和“从未构建”等多种状态,并用不同颜色在终端高亮显示(注意,颜色代码在部分终端可能不显示)。
- 数据聚合与统计:脚本最后会输出一个清晰的表格,并统计成功、失败的任务数量,还能单独列出所有失败的Job,方便你快速定位问题。
- 健壮性:对404(无构建记录)、网络超时、单个任务失败等情况都做了处理,不会因为一两个Job的问题导致整个脚本崩溃。
运行这个脚本,你就能得到一个实时的、全局的Jenkins构建状态仪表盘。你可以把它配置成定时任务(比如每5分钟运行一次),这样就能实现7x24小时的自动化监控了。
5. 进阶技巧与实战场景应用
掌握了基础的信息获取,我们就可以玩点更花的了。Jenkins API的能力远不止于此,结合Python,我们可以实现很多提升效率的自动化场景。下面分享几个我实际项目中常用的进阶技巧。
5.1 批量操作:禁用、启用与删除老旧Job
随着项目迭代,总会积累一批不再使用的Job。手动一个个去禁用或删除非常麻烦。我们可以写一个脚本,根据规则(比如最后构建时间超过180天)来批量处理它们。
def disable_old_jobs(jenkins_url, username, api_token, days_threshold=180):
"""
禁用超过指定天数未构建的Job。
注意:这是一个破坏性操作,请先在小范围测试!
"""
from datetime import datetime, timedelta
import time
auth = HTTPBasicAuth(username, api_token)
cutoff_time = datetime.now() - timedelta(days=days_threshold)
all_jobs = get_all_jobs(jenkins_url, username, api_token)
jobs_to_disable = []
# 第一步:先识别出需要禁用的Job
for job_name in all_jobs:
info = get_job_last_build_info(jenkins_url, job_name, username, api_token)
if info and info['time'] != 'N/A':
last_build_time = datetime.strptime(info['time'], '%Y-%m-%d %H:%M:%S')
if last_build_time < cutoff_time:
jobs_to_disable.append(job_name)
print(f"标记待禁用:{job_name},最后构建于 {info['time']}")
elif info and info['result'] == 'NO_BUILD':
# 从未构建过的Job也考虑禁用
jobs_to_disable.append(job_name)
print(f"标记待禁用(无构建记录):{job_name}")
# 第二步:确认并执行(安全起见,先注释掉执行部分,只打印)
if jobs_to_disable:
print(f"\n共找到 {len(jobs_to_disable)} 个超过{days_threshold}天未构建的Job。")
# user_input = input("确认要禁用这些Job吗?(输入 yes 继续): ")
# if user_input.lower() == 'yes':
# for job_name in jobs_to_disable:
# disable_url = f"{jenkins_url}/job/{job_name}/disable"
# resp = requests.post(disable_url, auth=auth)
# if resp.status_code in [200, 201]:
# print(f"已禁用:{job_name}")
# else:
# print(f"禁用失败:{job_name},状态码:{resp.status_code}")
# time.sleep(0.5) # 避免请求过快
# else:
# print("操作已取消。")
print("(实际禁用代码已注释,如需执行请取消注释并确认)")
else:
print("未找到符合条件的旧Job。")
重要提示:批量操作脚本一定要先“演习”(只打印不执行),确认目标无误后,再放开注释执行。我曾经不小心误操作过,幸好有备份。对于删除操作更要谨慎,建议先禁用,观察一段时间后再考虑删除。
5.2 集成告警:将失败构建通知到钉钉/企业微信
监控的最终目的是为了及时发现问题。我们可以把上面监控脚本发现的失败构建,自动发送到团队聊天工具里。
def send_dingtalk_alert(webhook_url, failed_jobs_list, jenkins_base_url):
"""
发送钉钉机器人告警。
"""
import json
if not failed_jobs_list:
return
# 构建Markdown格式的消息
job_links = "\n".join([f"- [{job['job_name']}]({job['url']}) (于 {job['time']})" for job in failed_jobs_list])
markdown_text = f"""## Jenkins 构建失败告警
发现 **{len(failed_jobs_list)}** 个构建失败或状态不稳定的Job,请相关同事及时检查!
**失败列表:**
{job_links}
[Jenkins控制台]({jenkins_base_url})"""
headers = {'Content-Type': 'application/json'}
data = {
"msgtype": "markdown",
"markdown": {
"title": "Jenkins构建告警",
"text": markdown_text
},
"at": {
"isAtAll": False, # 根据需要@所有人
# "atMobiles": ["138xxxxxxx"] # 可以指定@具体手机号同事
}
}
try:
resp = requests.post(webhook_url, data=json.dumps(data), headers=headers, timeout=10)
if resp.status_code == 200:
print("钉钉告警发送成功。")
else:
print(f"钉钉告警发送失败,状态码:{resp.status_code}")
except Exception as e:
print(f"发送钉钉告警时出错:{e}")
# 在主监控函数后调用
# failed_jobs = [info for info in all_build_info if info['result'] in ['FAILURE', 'UNSTABLE']]
# send_dingtalk_alert("你的钉钉机器人Webhook地址", failed_jobs, JENKINS_URL)
类似地,你也可以改成企业微信、Slack或者飞书的机器人。这样,一旦有构建失败,团队就能在第一时间收到通知,大大缩短问题响应时间。
5.3 使用python-jenkins库简化开发
之前我们一直用requests库直接调用API,这有助于理解底层原理。但在实际项目中,为了代码更简洁,我后期都转向了python-jenkins这个官方推荐的库。它封装了大部分常用操作。
import jenkins
# 连接服务器
server = jenkins.Jenkins('http://your-jenkins-server.com:8080', username='your_username', password='your_api_token') # 注意:这里的password也是填API Token
# 获取所有Job(返回的是包含详细信息的字典列表)
jobs = server.get_all_jobs()
for job in jobs:
print(f"Job: {job['fullname']}, 颜色(状态): {job['color']}")
# 获取某个Job的最后构建信息
job_name = 'your-job-name'
last_build_number = server.get_job_info(job_name)['lastBuild']['number']
build_info = server.get_build_info(job_name, last_build_number)
print(f"最后构建 #{last_build_number} 结果: {build_info['result']}")
# 触发一个无参数的构建
server.build_job(job_name)
# 获取构建的控制台输出
output = server.get_build_console_output(job_name, last_build_number)
print(output[:500]) # 打印前500个字符
可以看到,python-jenkins库的API更加直观和Pythonic。它帮你处理了URL拼接、JSON解析等琐事。我的建议是,在理解了原生API的基础上,对于常规的自动化管理任务,可以优先使用这个库来提高开发效率。
6. 避坑指南与最佳实践
在几年的使用过程中,我踩过不少坑,也总结了一些让脚本更稳定、更安全的心得,在这里分享给你。
1. 认证与安全是头等大事
- 永远使用API Token,不要用密码:这不仅是安全规范,也能避免因密码过期导致脚本失效。
- 为脚本创建专用账号并分配最小权限:不要在脚本里使用管理员账号的Token。在Jenkins中创建一个专门用于API调用的账号(如
api-bot),只给它执行脚本所需的最小权限(比如只读权限,或者特定文件夹的构建权限)。这样即使Token泄露,影响范围也有限。 - 妥善保管Token:不要把Token硬编码在脚本里然后上传到Git。应该使用环境变量、配置文件(
.ini,.yaml)或密钥管理服务来存储。例如:import os JENKINS_URL = os.environ.get('JENKINS_URL') API_TOKEN = os.environ.get('JENKINS_API_TOKEN')
2. 让你的脚本更健壮
- 增加重试机制:网络请求可能因瞬时波动失败。对于重要的操作,可以使用
tenacity或retrying库添加指数退避的重试逻辑。 - 设置合理的超时和限流:每个请求都要设置
timeout,避免脚本无限期挂起。对于并发请求,控制max_workers数量,别对Jenkins服务器造成DoS攻击。 - 做好日志记录:不要只用
print。使用Python的logging模块,将运行信息、错误记录到文件,方便后期排查问题。区分INFO、WARNING、ERROR等级别。 - 处理分页:如果你的Jenkins有成千上万个Job,API返回的列表可能会分页。Jenkins API支持
depth、tree参数和start、limit参数来优化查询,获取大量数据时需要留意。
3. 设计可维护的脚本结构
- 模块化:将不同的功能(如认证、获取Job、获取构建信息、发送通知)写成独立的函数或类,放在不同的模块文件中。这样主脚本清晰,也便于复用。
- 使用配置文件:将服务器地址、阈值天数、告警Webhook等配置项提取到配置文件(如
config.yaml)中,避免每次修改都要翻代码。 - 编写单元测试:对于核心的函数,比如解析API响应的函数,可以编写简单的单元测试,确保代码修改后基础功能正常。
把这些点都注意到,你的Jenkins自动化管理脚本就能从“勉强能用”进化到“稳定可靠”,真正成为你运维工作中的得力助手。从手动点击到自动化脚本,这种效率的提升感是非常实在的,你会发现自己有更多时间去处理更复杂、更有价值的问题。
更多推荐


所有评论(0)