做运维的小伙伴应该都知道nginx日志的重要性,一般出现访问问题,我们可能第一时间要去看日志去分析问题,但除了协助我们排查问题外,如果对nginx日志进一步分析可以得到更有用的数据,例如可以监控某站点的http状态码、PV,UV情况,request_time和response_time等,如果辅助其它工具进一步分析可以预防一些安全问题,比如同一个IP的访问某页面超出了限制,我们可以设置策略发现后可以自动拒绝有危险的IP访问,这也是我们经常说的防刷功能,今天要跟大家分享的是如何将nginx日志存储到es中,存储的目的当时是为了更好的分析它,大家都知道,如果要存储到es中,数据必须是json格式的,但如果当前日志不是json格式的就需要进行一个转换,要把每行数据转成json格式的这里有几个问题要明确:
首先,access.log是不断增长的,这就需要有个程序能实时读取新的日志进行转换,这是第一个要解决的问题。
其次,对转换完的数据我们能不能直接存储到es中,答案是不建议,因为如果转一次存一次,在程序中我们势必要判断每次存储是否成功,在高并发的网站中日志的产生量是非常巨大的,这样一来会影响整体的效率,那怎么办?在这种情况下,最好的解决办法就是用消息队列的方式去解决,这里我们选择kafka,这个后续会说。
根据上面的内容大家可以看到其实在分析nginx日志的过程就是一个数据流的处理,这也是我们标题所说的data-pipline,pipline故名意思就是管道的意思,加一个data就是一个数据管道,我们就是要建立一个这样的管道将数据源源不断的产生出来,就像水管的水一样。
因为要解决的二个问题,今天我们先解决第一个,先实现access.log实时读取和转换成json格式,先看我们的access.log的文件内容:
1 |
111.20.21.22 - - [19/Dec/2016:12:00:30 +0800] "GET / HTTP/1.1" 502 602 "-" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36" |
划重点了,请看代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
import time import datetime import json import socket def parse_log_line(line): strptime = datetime.datetime.strptime hostname = socket.gethostname() time = line.split(' ')[3][1::] entry = {} entry['datetime'] = strptime( time, "%d/%b/%Y:%H:%M:%S").strftime("%Y-%m-%d %H:%M") entry['source'] = "{}".format(hostname) entry['type'] = "www_access" entry['log'] = "'{}'".format(line.rstrip()) return entry def show_entry(entry): temp = ",".join([ entry['datetime'], entry['source'], entry['type'], entry['log'] ]) log_entry = {'log': entry} temp = json.dumps(log_entry) print("{}".format(temp)) return temp def follow(syslog_file): syslog_file.seek(0, 2) while True: line = syslog_file.readline() if not line: time.sleep(0.1) continue else: entry = parse_log_line(line) if not entry: continue json_entry = show_entry(entry) print(json_entry) f = open("access.log", "rt") follow(f) |
运行结果:
1 2 3 4 5 6 7 8 |
{ "log": { "datetime": "2016-12-19 12:00", "source": "iZ258ml0cx5Z", "type": "www_access", "log": "'111.20.21.22 - - [19/Dec/2016:12:00:30 +0800] \"GET / HTTP/1.1\" 502 602 \"-\" \"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36\"'" } } |
脚本整体逻辑是先将行日志转换成一个字典(parse_log_line()函数),然后将字典转成json格式(show_entry()函数),最后follow函数是开始从尾部读取日志文件注意seek(0,2)参数是2从文件底部读取。
写到这我们第一个问题就算是解决了,接下来的工作就是如果把数据传给kafka了,因为kafka本身就是一块比较大的内容限于篇幅我们下次继续,多谢各位观看,如果觉得还不错,还请帮转发。