項(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)化。

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 timeimport threadingimport subprocessfrom collections import dequedef 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 += lineif 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.1while task_status != ’req_log_got’: # 請(qǐng)求觸發(fā)的nginx日志收集未完成 time.sleep(time_to_wait) time_to_wait += 0.01else:# 獲取到用例請(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)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！