python分析nginx日志并推送到open-falcon

tanggang1740

不解释，直接撸代码　　

　　
#!/usr/bin/python
　　
# --*-- coding: utf-8 --*--
　　
# NGINX日志分析
　　

　　
import time
　　
import datetime
　　
import sys
　　
import os
　　
import os.path
　　
import re
　　
import json
　　
import socket
　　
import requests
　　
import subprocess
　　

　　
class NginxLog(object):
　　"""初始化数据"""
　　def __init__(self, log_file, interface_list, seek_file):
　　self.log_file = log_file
　　self.interface_list = interface_list
　　self.seek_file = seek_file
　　

　　def jsonFormat(self, python_data):
　　"""格式化成json格式"""
　　json_data = json.dumps(python_data, indent=2)
　　return json_data
　　

　　def hostname(self):
　　"""host_name: 主机名
　　host_ip: 主机ip
　　"""
　　host_name = socket.getfqdn(socket.gethostname( ))
　　host_ip = socket.gethostbyname(host_name)
　　return host_name
　　

　　def writeSeek(self, seek):
　　"""读过的文件游标写入临时文件"""
　　with open(self.seek_file,'w') as f:
　　f.write(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())) + '\n')
　　f.write(str(seek) + "\n")
　　

　　def LogRead(self):
　　"""读出新生成的日志
　　# 如果第一次运行，或是删除临时文件，从头运行，否则，从上次读取之后运行
　　# 0代表从头开始，1代表当前位置，2代表文件最末尾位置
　　"""
　　if os.path.exists(self.seek_file):
　　with open(self.seek_file) as f:
　　seek_tmp = f.readlines()
　　seek_old = int(seek_tmp[1].strip())
　　else:
　　seek_old = 0
　　with open(self.log_file) as f:
　　#记录当前最新文件游标
　　f.seek(0,2)#最新游标位置
　　seek_now = f.tell()
　　#读取上次读完之后的日志
　　if seek_now >= seek_old:
　　f.seek(seek_old,0)#从文件开头为止偏移
　　chunk = f.read(seek_now - seek_old)
　　#如果seek_now-seek_old小于0说明日志轮训
　　else:
　　f.seek(0,0)
　　chunk = f.read(seek_now)
　　# 将这次的游标写入临时文件
　　self.writeSeek(seek_now)
　　return chunk
　　

　　def LogStatistics(self):
　　"""分析NGINX日志的正则表达示，如果日志格式更改，则需要相应作更改
　　"""
　　

　　#log_examp = '127.0.0.1 - - [02/Mar/2018:11:01:09 +0800] "HEAD /aaa HTTP/1.1" 404 0 "-" "curl/7.29.0"'
　　#输出结果
　　#100.64.40.7 - - [05/Mar/2018:09:15:19 +0800] "GET /wpsad.php HTTP/1.0" 200 0 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.98 Safari/537.36 LBBROWSER" "114.255.44.143" 0.369
　　result_list = []
　　time_ns =  datetime.datetime.now().microsecond
　　time_stamp = int(str(time.time())[0:10])
　　host_name = self.hostname()
　　# url列表循环
　　for interface_item in self.interface_list:
　　interface_item_dict_count = {}
　　interface_item_dict_avg_request_time = {}
　　interface_item_dict_2xx = {}
　　interface_item_dict_4xx = {}
　　interface_item_dict_5xx = {}
　　interface_item_dict_count['ns']=interface_item_dict_avg_request_time['ns']=interface_item_dict_2xx['ns']=interface_item_dict_4xx['ns']=interface_item_dict_5xx['ns']=time_ns
　　interface_item_dict_count['clock']=interface_item_dict_avg_request_time['clock']=interface_item_dict_2xx['clock']=interface_item_dict_4xx['clock']=interface_item_dict_5xx['clock']=time_stamp
　　interface_item_dict_count['host']=interface_item_dict_avg_request_time['host']=interface_item_dict_2xx['host']=interface_item_dict_4xx['host']=interface_item_dict_5xx['host']=host_name
　　interface_item_dict_count['key'] = interface_item + '_count'
　　interface_item_dict_count['value'] = 0
　　interface_item_dict_avg_request_time['key'] = interface_item + '_avg_request_time'
　　interface_item_dict_avg_request_time['value'] = 0
　　interface_item_dict_2xx['key'] = interface_item + '_2xx'
　　interface_item_dict_2xx['value'] = 0
　　interface_item_dict_4xx['key'] = interface_item + '_4xx'
　　interface_item_dict_4xx['value'] = 0
　　interface_item_dict_5xx['key'] = interface_item + '_5xx'
　　interface_item_dict_5xx['value'] = 0
　　hit_url_count = 0
　　

　　##实时输出日志
　　for line in self.LogRead().split('\n'):
　　#print(line.split()[0])
　　if line != None and len(line.split()) != 0:
　　#匹配字段
　　remote_addr = line.split()[0]
　　#切割请求的url
　　request_url = line.split()[6]
　　status_code = line.split()[8]
　　request_time = line.split()[-1]#保留等会用
　　# 匹配之后数据结构操作
　　if interface_item == request_url:
　　hit_url_count += 1
　　interface_item_dict_count['value'] += 1
　　#响应时间
　　interface_item_dict_avg_request_time['value'] += float(request_time)
　　if status_code.strip('\"').startswith('2'):
　　interface_item_dict_2xx['value'] += 1
　　if status_code.strip('\"').startswith('4'):
　　interface_item_dict_4xx['value'] += 1
　　if status_code.strip('\"').startswith('5'):
　　interface_item_dict_5xx['value'] += 1
　　# 求平均请求反应时间
　　if interface_item_dict_avg_request_time['value'] != 0:
　　interface_item_dict_avg_request_time['value'] = interface_item_dict_avg_request_time['value'] / hit_url_count
　　# 结果加入列表
　　result_list.append(interface_item_dict_count)
　　result_list.append(interface_item_dict_avg_request_time)
　　result_list.append(interface_item_dict_2xx)
　　result_list.append(interface_item_dict_4xx)
　　result_list.append(interface_item_dict_5xx)
　　return result_list
　　#return self.jsonFormat(result_list)
　　

　　

　　
def pushFalcon(return_data):
　　"""数据推到openfalcon"""
　　all_data = []
　　all_request_dic = {}
　　wpsad_2xx_request_dic = {}
　　wpsad_4xx_request_dic = {}
　　wpsad_5xx_request_dic = {}
　　wpsad_response_time_dic = {}
　　for i in return_data:
　　if 'wpsad.php_count' in i['key']:
　　all_request_dic['value'] = i['value']
　　all_request_dic['key'] = 'wpsad.php_count'
　　all_request_dic['host_name'] = i['host']
　　all_data.append(all_request_dic)
　　if 'wpsad.php_2xx' in i['key']:
　　wpsad_2xx_request_dic['value'] = i['value']
　　wpsad_2xx_request_dic['key'] = 'wpsad.php_2xx_count'
　　wpsad_2xx_request_dic['host_name'] = i['host']
　　all_data.append(wpsad_2xx_request_dic)
　　if 'wpsad.php_4xx' in i['key']:
　　wpsad_4xx_request_dic['value'] = i['value']
　　wpsad_4xx_request_dic['key'] = 'wpsad.php_4xx_count'
　　wpsad_4xx_request_dic['host_name'] = i['host']
　　all_data.append(wpsad_4xx_request_dic)
　　if 'wpsad.php_5xx' in i['key']:
　　wpsad_5xx_request_dic['value'] = i['value']
　　wpsad_5xx_request_dic['key'] = 'wpsad.php_5xx_count'
　　wpsad_5xx_request_dic['host_name'] = i['host']
　　all_data.append(wpsad_5xx_request_dic)
　　if 'wpsad.php_avg_request_time' in i['key']:
　　wpsad_response_time_dic['key'] = 'wpsad_response_time'
　　wpsad_response_time_dic['value'] = i['value']
　　#all_data.append(wpsad_response_time_dic)
　　# nginx请求状态数据
　　ts = int(time.time())
　　payload = []
　　for i in all_data:
　　temp_dic = {
　　#"endpoint": i['host_name'],
　　"endpoint": "vm172-31-32-13.ksc.com",
　　"metric": i['key'],
　　"timestamp": ts,
　　"step": 60,
　　"value": i['value'],
　　"counterType": "GAUGE",
　　"tags": "url="+i['key']
　　}
　　payload.append(temp_dic)
　　#print(payload)
　　

　　# 响应时间数据
　　response_time_dic = {
　　"endpoint": "vm172-31-32-13.ksc.com",
　　"metric": wpsad_response_time_dic['key'],
　　"timestamp": ts,
　　"step": 60,
　　"value": wpsad_response_time_dic['value'],
　　"counterType": "GAUGE",
　　"tags": "",
　　}
　　payload.append(response_time_dic)
　　# nginx并发请求数统计
　　estab_data = {
　　"endpoint": "vm172-31-32-13.ksc.com",
　　"metric": "nginx_estab_num",
　　"timestamp": ts,
　　"step": 60,
　　"value": 0,
　　"counterType": "GAUGE",
　　"tags": "",
　　}
　　time_wait = {
　　"endpoint": "vm172-31-32-13.ksc.com",
　　
                "metric": "nginx_timewait_num",
　　
                "timestamp": ts,
　　
                "step": 60,
　　
                "value": 0,
　　
                "counterType": "GAUGE",
　　
                "tags": "",
　　}
　　# TIME_WAIT
　　time_wait_cmd = "netstat -ant|grep -i '80'|grep 'TIME_WAIT'|wc -l"
　　time_wait_p = subprocess.Popen(time_wait_cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
　　timewait_out = time_wait_p.stdout.read()
　　timewait_err = time_wait_p.stderr.read()
　　if not timewait_err:
　　time_wait['value'] = int(timewait_out.strip())
　　payload.append(time_wait)
　　

　　# ESTABLISHED
　　estab_cmd = "netstat -ant|grep -i '80'|grep 'ESTABLISHED'|wc -l"
　　estab_p = subprocess.Popen(estab_cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
　　estab_out = estab_p.stdout.read()
　　estab_err = estab_p.stderr.read()
　　if not estab_err:
　　estab_data['value'] = int(estab_out.strip())
　　payload.append(estab_data)
　　# nginx进程占用的内存监控
　　mem_dic = {
　　"endpoint": "vm172-31-32-13.ksc.com",
　　"metric": "nginx_mem",
　　"timestamp":ts,
　　"step": 60,
　　"value": 0,
　　"counterType": "GAUGE",
　　"tags": "",
　　}
　　#mem_cmd = "top -b -n1|grep nginx|gawk '{if($6~/m$/) {sum+=$6*1024} else {sum+=$6} }; END {print int(sum/1024)}'"
　　#mem_p = subprocess.Popen(time_wait_cmd, shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
　　#mem_out = mem_p.stdout.read()
　　#print(mem_out)
　　#mem_err = mem_p.stderr.read()
　　#if not mem_err:
　　#mem_dic['value'] = int(mem_out.strip())
　　#print(mem_dic['value'])
　　#
　　

　　nginxpid = subprocess.Popen(["pidof", "nginx"], stdout=subprocess.PIPE)
　　nginxpid = nginxpid.stdout.read().split()
　　memsum = 0
　　for i in nginxpid:
　　pidfile = os.path.join("/proc/", str(i), "status")
　　with open(pidfile) as f:
　　for mem in f:
　　if mem.startswith("VmRSS"):
　　pidmem = int(mem.split()[1])
　　memsum += pidmem
　　memsum = int(memsum)//1024
　　#print("%d %s" %(memsum,"M"))
　　mem_dic['value'] = memsum
　　payload.append(mem_dic)
　　# 推送到falcon-agent
　　#print(payload)
　　r = requests.post("http://127.0.0.1:1988/v1/push", data=json.dumps(payload))
　　
def main():
　　# 需要分析的url列表
　　interface_list = ['/wpsad.php']
　　

　　# 日志文件位置
　　log_file = "/data/logs/nginx/ads.access.log"
　　# 临时文件位置
　　seek_file = "/data/logs/nginx/ads_log_check_seek.tmp"
　　nginx_log = NginxLog(log_file, interface_list, seek_file)
　　return_data = nginx_log.LogStatistics()
　　#print return_json_data
　　pushFalcon(return_data)
　　

　　
if __name__ == '__main__':
　　main()