需求描述
項(xiàng)目需求測(cè)試過(guò)程中,需要向Nginx服務(wù)器發(fā)送一些用例請(qǐng)求,然后查看對(duì)應(yīng)的Nginx日志,判斷是否存在特征內(nèi)容,來(lái)判斷任務(wù)是否執(zhí)行成功。為了提升效率,需要將這一過(guò)程實(shí)現(xiàn)自動(dòng)化。
實(shí)踐環(huán)境
Python 3.6.5
代碼設(shè)計(jì)與實(shí)現(xiàn)
#!/usr/bin/env python
# -*- coding:utf-8 -*-
'''
@CreateTime: 2021/06/26 9:05
@Author : shouke
'''
import time
import threading
import subprocess
from collections import deque
def collect_nginx_log():
global nginx_log_queue
global is_tasks_compete
global task_status
args = 'tail -0f /usr/local/openresty/nginx/logs/access.log'
while task_status != 'req_log_got':
with subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, universal_newlines = True) as proc:
log_for_req = ''
outs, errs = '', ''
try:
outs, errs = proc.communicate(timeout=2)
except subprocess.TimeoutExpired:
print('獲取nginx日志超時(shí),正在重試')
proc.kill()
try:
outs, errs = proc.communicate(timeout=5)
except subprocess.TimeoutExpired:
print('獲取nginx日志超時(shí),再次超時(shí),停止重試')
break
finally:
for line in outs.split('\n'):
flag = '\"client_ip\":\"10.118.0.77\"' # 特征
if flag in line: # 查找包含特征內(nèi)容的日志
log_for_req += line
if task_status == 'req_finished':
nginx_log_queue.append(log_for_req)
task_status = 'req_log_got'
def run_tasks(task_list):
'''
運(yùn)行任務(wù)
:param task_list 任務(wù)列表
'''
global nginx_log_queue
global is_tasks_compete
global task_status
for task in task_list:
thread = threading.Thread(target=collect_nginx_log,
name="collect_nginx_log")
thread.start()
time.sleep(1) # 執(zhí)行任務(wù)前,讓收集日志線程先做好準(zhǔn)備
print('正在執(zhí)行任務(wù):%s' % task.get('name'))
# 執(zhí)行Nginx任務(wù)請(qǐng)求
# ...
task_status = 'req_finished'
time_to_wait = 0.1
while task_status != 'req_log_got': # 請(qǐng)求觸發(fā)的nginx日志收集未完成
time.sleep(time_to_wait)
time_to_wait += 0.01
else:# 獲取到用例請(qǐng)求觸發(fā)的nginx日志
if nginx_log_queue:
nginx_log = nginx_log_queue.popleft()
task_status = 'req_ready'
# 解析日志
# do something here
# ...
else:
print('存儲(chǔ)請(qǐng)求日志的隊(duì)列為空')
# do something here
# ...
if __name__ == '__main__':
nginx_log_queue = deque()
is_tasks_compete = False # 所有任務(wù)是否執(zhí)行完成
task_status = 'req_ready' # req_ready,req_finished,req_log_got # 存放執(zhí)行次任務(wù)任務(wù)的一些狀態(tài)
print('###########################任務(wù)開(kāi)始###########################')
tast_list = [{'name':'test_task', 'other':'...'}]
run_tasks(tast_list)
is_tasks_compete = True
current_active_thread_num = len(threading.enumerate())
while current_active_thread_num != 1:
time.sleep(2)
current_active_thread_num = len(threading.enumerate())
print('###########################任務(wù)完成###########################')
注意:
1、上述代碼為啥不一步到位,直接 tail -0f /usr/local/openresty/nginx/logs/access.log | grep "特征內(nèi)容"
呢?這是因?yàn)檫@樣做無(wú)法獲取到Nginx的日志
2、實(shí)踐時(shí)發(fā)現(xiàn),第一次執(zhí)行proc.communicate(timeout=2)
獲取日志時(shí),總是無(wú)法獲取,會(huì)超時(shí),需要二次獲取,并且timeout
設(shè)置太小時(shí)(實(shí)踐時(shí)嘗試過(guò)設(shè)置為1秒
),也會(huì)導(dǎo)致第二次執(zhí)行時(shí)無(wú)法獲取Nginx日志。
到此這篇關(guān)于Python 實(shí)時(shí)獲取任務(wù)請(qǐng)求對(duì)應(yīng)的Nginx日志的文章就介紹到這了,更多相關(guān)Python獲取Nginx日志內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
您可能感興趣的文章:- python實(shí)現(xiàn)的分析并統(tǒng)計(jì)nginx日志數(shù)據(jù)功能示例
- python+pandas分析nginx日志的實(shí)例
- Python解析nginx日志文件
- python實(shí)現(xiàn)分析apache和nginx日志文件并輸出訪客ip列表的方法
- Python 分析Nginx訪問(wèn)日志并保存到MySQL數(shù)據(jù)庫(kù)實(shí)例