ELK日志采集和查询方式

　　我是个目录

　　回顾

　　前两篇文章简单的介绍了JAVA通过代码怎样使用MDC进行日志复印，去查看log4j MDC用户操作日志追踪配置，以及ELK平台的搭建，去查看log4j MDC用户操作日志追踪配置。接下来将结合实际案例，简单介绍生产服务器的日志怎样被logstash进行采集，并统一汇总，使得我们就能快速、方便、高效的查询日志，并且可以便捷定位到该条日志是哪台服务器形成的，再也不用盲目的翻遍所有的生产服务器，只为找到出问题的那一台机器。

　　logstash日志采集

　　因为我们复印的日志各种各样，为了让我们能否便捷的通过kibana进行检索，我们须要在logstash配置对应的采集规则，如果你啥也不做，只是简单的采集，可能会出现一些意想不到的错误。

　　日志搜集是按行来搜集的，当你的日志发生了换行，这时候换行的那几行就被觉得是独立的一行，这样采集到的日志可读性太差假如你使用了MDC，配置了log4j的日志格式输出，不配规则的话一整行就会被采集到message数组，此时你要按照某个数组快速搜索的话基本也是不可能的事情了

　　基于以上可能出现的“错误”，我们须要针对当前项目的日志配置文件制订一套属于自己的搜集规则。

　　其实logstash的配置文件很简单，基本就以下的套路，日志源从哪儿来（input输入插件），通过什么样的规则（filter过滤插件），最终将日志输出到哪些地方（output输出插件）

　　#　输入

input {

...

}

# 过滤器

filter {

...

}

# 输出

output {

...

}

　　因为我们这儿介绍的是项目日志的采集，所以input其实是从file文件中来，配置如下：

　　input {

file {

type => "wechat-log"

path => ["/usr/local/tomcat/logs/wechat/*.log"]

codec => multiline{

pattern => "^\[%{TIMESTAMP_ISO8601}\]"

what => "previous"

negate => true

}

start_position => "beginning"

}

}

　　其中path是日志采集的地方，从日志文件的第一行开始采集，定义一个type（一般最后它就是kibana的索引）。

　　codec插件

　　这里codec的上场才能解决例如我们上面提及的日志换行的问题。在logstash读入的时侯，通过codec编码解析日志为相应格式，从logstash输出的时侯，通过codec解码成相应格式。当我们的应用程序复印出具有换行的日志的时侯，比如ERROR日志，一般有错误堆栈消息，各种at开头的一行，我们可以通过multiline来进行处理，让logstash觉得这一行是属于上一行的内容，而不是把它作为新的一行进行处理。

　　一般我们的tomcat日志都是以时间开头的，对于象at那个堆栈的信息都是不存在时间的，所以我们可以配置正则表达式【^[%{TIMESTAMP_ISO8601}]】，只有当以时间开头的一行才算新的一行，不是时间开头的就属于后面【what=>previous】或者前面的【what=>next】，这里我们配置属于后面的。

　　上述的配置以后解决了换行问题，那么接下来我们还须要处理按数组进行日志切分的。

　　grok插件

　　在grok中，支持以正则表达式的形式提取日志信息，其中，正则表达式又分两种，一种是外置的正则表达式，一种是自定义的正则表达式，当外置的正则表达式不能满足我们的需求的时侯，我们就要上自定义的正则表达式了，但是外置的基本满足我们的需求了。具体细节可以查看grok介绍

　　假设我们的日志配置文件是这样配置的：

　　我们设置了merchant、openid、queryType、orderId、wechatOrderId、input、source7个数组，生产服务器复印的日志将会是以下格式：

　　[2019-01-27 17:51:22.051] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - senior - aa4820190127175110 - 4200000283201901277968491434 - 352982093855677 - 1 - INFO com.apple.wechat.service.Worker:401 - 发送模板消息,查询结果为:[email protected]

[2019-01-27 17:51:22.230] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - senior - aa4820190127175110 - 42000002832019277968491434 - 352982093855677 - 1 - ERROR me.chanjar.weixin.mp.api.impl.WxMpServiceImpl:403 -

[URL]: https://api.weixin.qq.com/cgi-bin/message/template/send

[PARAMS]: {"touser":"olC5FwLnXjtCbQsW76lkevV57nH0","template_id":"Qt1zyzQs4R1uPrJylGQLSUTS6QcG6UyWB2zDzGt7QGY","url":"http://mp.weixin.qq.com/bizmall/mallshelf?id=&t=mall/list&biz=MjM5OTAxMzk4MQ==&shelf_id=7&showwxpaytitle=1#wechat_redirect","data":{"first":{"value":"查询结果","color":"#B452CD"},"keyword1":{"value":"aa4820190127175110","color":"#FF4040"},"keyword2":{"value":"352982093855677","color":"#FF4040"},"keyword3":{"value":"1.00元","color":"#FF4040"},"remark":********************}

[RESPONSE]: {"errcode":40003,"errmsg":"invalid openid hint: [mLJNpa06824120]"}

[2019-01-27 17:51:22.230] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - senior - aa4820190127175110 - 4200000283201901277968491434 - 352982093855677 - 1 - ERROR com.apple.wechat.service.Worker:405 - 发送模板消息失败,{"errcode":40003,"errmsg":"invalid openid hint: [mLJNpa06824120]"}

[2019-01-27 17:51:22.231] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - senior - aa4820190127175110 - 4200000283201901277968491434 - 352982093855677 - 1 - INFO com.apple.wechat.service.RefundService:57 - 开始发起退款,退款订单id:2056653,微信订单号:4200000283201901277968491434

[2019-01-27 17:51:22.463] - - - - - - - - INFO com.apple.wechat.service.Worker:94 - 执行任务:QueryTask{msg='C39XQ4NFKPGN', fromUserId='oOEvtjsGdmAKrZx81zsACqBjjdsA', merchant='MLdress', type='senior', authUserId='olC5FwH40UpZakKBZRls_t_HR9Ew', price='1.00', tradeNo='e50b20190127175115', model='', orderId=2056654, needRefund=false, needRedo=false, sendMsg=false, msgType='1', lat='', lon='', token='e50ba187b2f84297b60fc14699748679', wechatOrderNo='4200000269201901277039023012'}

[2019-01-27 17:51:23.327] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - senior - aa4820190127175110 - 4200000283201901277968491434 - 352982093855677 - 1 - INFO com.apple.wechat.service.RefundService:97 - 退款结果:success, 实付金额：1.00,发起退款金额:1.00

[2019-01-27 17:51:26.876] - - - - - - - - INFO com.apple.wechat.service.Worker:94 - 执行任务:QueryTask{msg='C39T81JEHG01', fromUserId='oOEvtjotDEF8doO3xVxyJ0-dCqFM', merchant='MLdress', type='normal', authUserId='', price='', tradeNo='', model='', orderId=0, needRefund=false, needRedo=false, sendMsg=false, msgType='1', lat='', lon='', token='', wechatOrderNo=''}

[2019-01-27 17:51:28.003] - MLdress - oOEvtjsGdmAKrZx81zsACqBjjdsA - senior - e50b20190127175115 - 4200000269201901277039023012 - C39XQ4NFKPGN - 1 - INFO com.apple.wechat.util.HttpUtils:125 - http execute cost total seconds 5540

　　我们可以通过以下正则表达式将日志进行切割，并把内容形参给对应的数组

　　filter {

grok {

match => ["message", "\[%{TIMESTAMP_ISO8601:logdate}\] - (?[\b\w\s]*) - (?[\u4e00-\u9fa5\b\w\s]*) - (?[\b\w\s]*) - (?[\b\w\s]*) - (?[\b\w\s]*) - (?[\b\w\s]*) - (?[\b\w\s]*) - %{WORD:level}\s*%{JAVACLASS:class}:%{NUMBER:lineNumber} - (?[\W\w\S\s]*)"]

}

date {

match => ["logdate", "yyyy-MM-dd HH:mm:ss.SSS"]

target => "@timestamp"

}

}

　　(? [\b\w\s]*) 代表使用[]里面的正则，把辨识的结果放在merchant这个数组上面，其他类似。至于这儿的match怎么输写，跟你的日志配置文件及你想要的疗效有很大的关联，所以只能渐渐调试，直到你写的match才能正确切割出你的日志文件。在线测试你的正则表达式是否可以匹配项目输出的日志测试工具传送门

　　完整的配置文件

　　input {

file {

type => "wechat-log"

path => ["/usr/local/tomcat/logs/wechat/*.log"]

codec => multiline{

pattern => "^\[%{TIMESTAMP_ISO8601}\]"

what => "previous"

negate => true

}

start_position => "beginning"

}

}

filter {

grok {

match => ["message", "\[%{TIMESTAMP_ISO8601:logdate}\] - (?[\b\w\s]*) - (?[\u4e00-\u9fa5\b\w\s]*) - (?[\b\w\s]*) - (?[\b\w\s]*) - (?[\b\w\s]*) - (?[\b\w\s]*) - (?[\b\w\s]*) - %{WORD:level}\s*%{JAVACLASS:class}:%{NUMBER:lineNumber} - (?[\W\w\S\s]*)"]

}

date {

match => ["logdate", "yyyy-MM-dd HH:mm:ss.SSS"]

target => "@timestamp"

}

}

output {

elasticsearch {

hosts => "**************:9200"

index => "logstash-%{type}"

template_overwrite => true

}

}

　　保存后启动logstash即可。命令：

　　nohup ./bin/logstash agent -f config/log.conf &

　　kibana操作首先是创建索引，因为我们的output配置了index为logstash-%{type}，所以对于的索引为：logstash-wechat-log创建完索引，我们能发觉kibana里面列举了我们上面定义的数组，如下图。

　　

　　接下来我们就可以通过各类条件来进行日志的搜索了。

　　假设我们要搜索商户是MLdress，用户输入内容为3565的日志信息，那么我们只须要输入【merchant:MLdress AND input:3565】即可搜索出对应的日志，如下图。

　　

　　总结

　　通过在代码使用MDC进行规范的日志复印，在结合logstash提供的强悍的日志采集插件，我们可以将所有服务器的日志统一上报到es，并通过kibana进行放飞自我的查询操作，只有这样就能极大的提升日常开发效率。程序员不仅要会写代码之外，另外一项加分项那肯定是具有快速发觉和定位问题的能力了，如果没有ELK这样的工具，那快速发觉及定位虽然是比较困难的。

　　所以，你懂的，赶紧用上去吧。