用Data-pipeline模式将nginx日志存储到elasticsearch中(1)

做运维的小伙伴应该都知道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的文件内容:

 

划重点了,请看代码:

运行结果:

 

 

脚本整体逻辑是先将行日志转换成一个字典(parse_log_line()函数),然后将字典转成json格式(show_entry()函数),最后follow函数是开始从尾部读取日志文件注意seek(0,2)参数是2从文件底部读取。

写到这我们第一个问题就算是解决了,接下来的工作就是如果把数据传给kafka了,因为kafka本身就是一块比较大的内容限于篇幅我们下次继续,多谢各位观看,如果觉得还不错,还请帮转发。