使用版本：python3.9，centos7.8
python里提供了os模块，可以让我在代码中对linux发送命令，并获取返回值。
这里会讲到两种方法去使用os模块----os.popen和os.system
一. os.system
os.system('命令字符串')。这个命令适用于一些你只需要知道结果是否成功的命令，因为它只会返回给你一个int类型的数字，0代表成功，非0代表失败。而且当你的命令字符串是一个子进程命令的话，他的返回值并不准确，他不会等待子进程在后台运行结束，只要子进程能够进入运行状态，他就会返回0。
二. os.popen
在我看来，这个方法在使用子进程做并行或者在获取返回结果上，是很好用的，首先它一共有3个参数，
cmd--命令
mode--方式（默认值为‘r’,可选值为‘w’,一般场景中，r的使用比较普遍）
buffering--缓存值，默认为-1
该方法调用了subprocess里的方法，并最后帮助你关闭了资源，一条龙服务还是很人性化的。所以一般使用时，传一个命令即可，然后它提供了read(),readlines()给你去拿到返回值
即os.popen('cmd').read()
而且笔者认为它强大之处在于，当cmd是一个后台进程，即在命令最后加入&符号的进程，它依旧会等待进程结束并拿到返回值，我也用他帮我将原本需要5分钟+的程序中，加上使用子进程并行将时间缩短到20秒左右。
话不多说，分享实战经验，一个读取jenkins信息并写入时序数据库的流程：
if action == 'jenkins':
# 获取主机地址
host = request.get_host()
response_data = {'kind': 'JenkinsInfo.get', 'msg': 'jenkins任务构建信息入库', 'status': 200, 'param': ''}
data = request.GET
jenkins_name = data.get('jenkins_name', '')
if jenkins_name in JENKINS.keys():
jenkins_info = JENKINS[jenkins_name]
else:
response_data.update({
'param': '参数无效',
'status': 403
})
return JsonResponse(response_data, status=response_data['status'])
# 连接jenkins
server = self.to_jenkins(jenkins_info)
all_jobs = server.get_jobs()
job_name_list = [x.get('name') for x in all_jobs]
cmdb_job = {}
# 获取cmdb中的app_name
cmdb_apps = AppBasisNew.objects.filter(**{'status': '已上线'}).values('app_name')
app_name_list = [x.get('app_name') for x in cmdb_apps]
# 判断cmdb中app是否在jenkins中
env_list = ['archive-PRE-', 'PRE-', 'bak-PRE-', 'ARCHIVE-PRE-', 'BAK-PRE-']
if app_name_list and job_name_list:
for app_name in app_name_list:
for env in env_list:
if env+app_name in job_name_list:
cmdb_job[env+app_name] = app_name
if cmdb_job:
job_str = str(cmdb_job).replace("{'", "").replace("'}", "").replace("', '", " ").replace("': '", "\|")
res = os.popen(f"for job in {job_str}; do curl -G http://{host}/jenkins/pre/to_db?"f"jenkins_name={jenkins_name}\&job_name=$job & done").read()
res_str = res.replace('}{', '}, {').join('[]')
res_json = json.loads(res_str)
response_data.update({
'param': res_json
})
else:
response_data.update({
'param': 'jenkins与cmdb无匹配数据'
})
return JsonResponse(response_data, status=response_data['status'])接受请求，返回相应：
if action == 'to_db':
data = request.GET
jenkins_name = data['jenkins_name']
job_name = data['job_name'].split('|')[0]
project_name = data['job_name'].split('|')[1]
server = self.to_jenkins(JENKINS[jenkins_name])
build_info_list = []
# 最新jenkins项目的构建时间
last_timestamp = 0
# 连接influxdb
db_client = self.to_influxdb()
try:
job_info = server.get_job_info(job_name)
except jenkins.JenkinsException:
db_client.close()
return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})
if job_info.get('builds'):
# 获取该工程名每次构筑的时间：
res = db_client.query(f"select max(build_time) from pre_job_info "
f"where project_name::tag='{project_name}' and from_jenkins='{jenkins_name}'")
last_time = 0
if res:
last_time = [x[u'max'] for x in res.get_points()][0]
for job_build in range(1, len(job_info['builds'])+1):
job_time = job_info['builds'][-job_build]['number']
try:
build_info = server.get_build_info(job_name, job_time)
except jenkins.JenkinsException:
db_client.close()
return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})
if build_info.get('timestamp') // 1000 <= last_time:
break
if job_build == 1:
last_timestamp = build_info.get('timestamp')//1000
build_info_dict = ToDict().return_dic(build_info)
# 控制台输出
build_output = server.get_build_console_output(job_name, job_time)
build_out_list = build_output.split('\n')
build_agent_name = 'null'
project_path = 'null'
if jenkins_name == 'JENKINS_PRE1' or jenkins_name == 'JENKINS_PRE3':
res_info = ''
for i in build_out_list:
if 'Running on' in i:
res_info = i
if res_info:
build_agent_name = res_info.split()[-3]
project_path = res_info.split()[-1]
if jenkins_name == 'JENKINS_PRE2':
project_path = build_out_list[1].split()[-1]
build_agent_name = build_info_dict.get('builtOn')
if job_info.get('lastStableBuild'):
last_stable_build = job_info['lastStableBuild']['number']
else:
last_stable_build = 0
if job_info.get('lastSuccessfulBuild'):
last_successful_build = job_info['lastSuccessfulBuild']['number']
else:
last_successful_build = 0
if build_info_dict.get('actions_lastBuiltRevision_branch_name'):
build_branch_name = build_info_dict.get(
'actions_lastBuiltRevision_branch_name')
code_version = build_info_dict.get(
'actions_lastBuiltRevision_branch_SHA1')
else:
build_branch_name = build_info_dict.get('changeSet_revisions_module', 'null')
code_version = str(build_info_dict.get('changeSet_revisions_revision', 'null'))
if build_info_dict.get('result') == 'SUCCESS':
build_successful = True
else:
build_successful = False
if build_info_dict.get('actions_parameters_name1') == 'Fabu_Type':
fabu_type = build_info_dict.get('actions_parameters_value1')
else:
fabu_type = build_info_dict.get('actions_parameters_value2', 'all-in-one')
res = {'measurement': 'pre_job_info',
'tags': {
'project_name': project_name,
'build_result': build_info_dict.get('result'),
'project_path': project_path
},
'time': datetime.fromtimestamp(build_info_dict.get('timestamp') // 1000),
'fields': {
'build_agent_name': build_agent_name,
'build_branch_name': build_branch_name,
'build_causer': build_info_dict.get('actions_causes_userName'),
'build_exec_time': build_info_dict.get('duration') // 1000,
'build_measured_time': int(datetime.Now().timestamp()),
'build_number': build_info_dict.get('id'),
'build_result': build_info_dict.get('result'),
'build_result_ordinal': build_info_dict.get('queueId'),
'build_scheduled_time': build_info_dict.get('estimatedDuration') // 1000,
'build_status_message': build_info_dict.get('building'),
'build_successful': build_successful,
'build_time': build_info_dict.get('timestamp') // 1000,
'last_stable_build': last_stable_build,
'last_successful_build': last_successful_build,
'project_build_health': job_info['healthReport'][0]['score'],
'project_name': project_name,
'project_path': project_path,
'Code_Version': code_version,
'Fabu_Type': fabu_type,
'from_jenkins': jenkins_name
},
}
build_info_list.append(res)
if build_info_list:
db_client.write_points(build_info_list)
else:
db_client.close()
return JsonResponse({f'{jenkins_name}': f'{job_name} has no new data'})
# 数据入库验证
new_res = db_client.query(f"select build_number from pre_job_info "
f"where project_name::tag='{project_name}' and build_time={last_timestamp}")
if new_res:
db_client.close()
return JsonResponse({f'{jenkins_name}': f'{job_name} success'})
db_client.close()
return JsonResponse({f'{jenkins_name}': f'{job_name} fail'})版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
