ELK ⽇志采集和查询⽅法
我是个⽬录
回顾
前两篇⽂章简单的介绍了JAVA通过代码如何使⽤MDC进⾏⽇志打印，去查看，以及ELK平台的搭建，去查看。接下来将结合实际案例，简单介绍⽣产服务器的⽇志如何被logstash进⾏采集，并统⼀汇总，使得我们能够快速、⽅便、⾼效的查询⽇志，并且可以⽅便定位到该条⽇志是哪台服务器产⽣的，再也不⽤盲⽬的翻遍所有的⽣产服务器，只为找到出问题的那⼀台机器。
logstash ⽇志采集
因为我们打印的⽇志各种各样，为了使我们能够⽅便的通过kibana进⾏检索，我们需要在logstash配置对应的搜集规则，如果你啥也不做，只是简单的搜集，可能会出现⼀些意想不到的错误。
1. ⽇志收集是按⾏来收集的，当你的⽇志发⽣了换⾏，这时候换⾏的那⼏⾏就被认为是独⽴的⼀⾏，这样采集到的⽇志可读性很差
2. 如果你使⽤了MDC，配置了log4j的⽇志格式输出，不配规则的话⼀整⾏都会被采集到message字段，此时你要根据某个字段快速搜索的话基本也是不可能的事情了
基于以上可能出现的“错误”，我们需要针对当前项⽬的⽇志配置⽂件制定⼀套属于⾃⼰的收集规则。
其实logstash的配置⽂件很简单，基本就以下的套路，⽇志源从哪⾥来（input输⼊插件），通过什么样的规则（filter过滤插件），最终将⽇志输出到什么地⽅（output输出插件）
因为我们这⾥介绍的是项⽬⽇志的采集，所以input当然是从file⽂件中来，配置如下：
其中path是⽇志采集的地⽅，从⽇志⽂件的第⼀⾏开始采集，定义⼀个type（⼀般最后它就是kibana的索引）。
codec 插件# 输⼊input { ...}# 过滤器filter { ...}# 输出output { ...}
1
2
3
source是什么意思4
5
6
7
8
9
10
1112
13
14input { file { type => "wechat-log" path => ["/usr/local/tomcat/logs/wechat/*.log"] codec => multiline{ pattern => "^\[%{TIMESTAMP_ISO8601}\]" what => "previous" negate => true } start_position => "beginning" }}
1
2
3
4
勒克洛斯
5
6
幼儿英语学习
7
8
9
10
11
12
这⾥codec的出场能够解决诸如我们前⾯提到的⽇志换⾏的问题。在logstash读⼊的时候，通过codec编码解析⽇志为相应格式，从
safelogstash输出的时候，通过codec解码成相应格式。当我们的应⽤程序打印出具有换⾏的⽇志的时候，⽐如ERROR⽇志，⼀般有错误堆栈消息，各种at开头的⼀⾏，我们可以通过multiline来进⾏处理，让logstash认为这⼀⾏是属于上⼀⾏的内容，⽽不是把它作为新的⼀⾏进⾏处理。
⼀般我们的tomcat⽇志都是以时间开头的，对于像at那种堆栈的信息都是不存在时间的，所以我们可以配置正则表达式【^[%
{TIMESTAMP_ISO8601}]】，只有当以时间开头的⼀⾏才算新的⼀⾏，不是时间开头的就属于前⾯【what=>previous】或者后⾯的
【what=>next】，这⾥我们配置属于前⾯的。
posh上述的配置之后解决了换⾏问题，那么接下来我们还需要处理按字段进⾏⽇志切分的。
grok 插件
在grok中，⽀持以正则表达式的⽅式提取⽇志信息，其中，正则表达式⼜分两种，⼀种是内置的正则表达式，⼀种是⾃定义的正则表达式，当内置的正则表达式不能满⾜我们的需求的时候，我们就要上⾃定义的正则表达式了，但是内置的基本满⾜我们的需求了。具体细节可以查看
假设我们的⽇志配置⽂件是这样配置的：
我们设置了merchant、openid、queryType、orderId、wechatOrderId、input、source7个字段，⽣产服务器打印的⽇志将会是以下格式：
我们可以通过以下正则表达式将⽇志进⾏切割，并把内容赋值给对应的字段
(? < merchant >[\b\w\s]*) 代表使⽤[]⾥⾯的正则，把识别的结果放到merchant这个字段⾥⾯，其他类似。⾄于这⾥的match如何输写，跟你的⽇志配置⽂件及你想要的效果有很⼤的关联，所以只能慢慢调试，直到你写的match能够正确切割出你的⽇志⽂件。在线测试你的正则表达式是否可以匹配项⽬输出的⽇志
完整的配置⽂件<appender name="logic" class="org.apache.log4j.DailyRollingFileAppender"> <param name="File" value="${catalina.home}/logs/wechat/logic.log" /> <param name="DatePattern" value="'.'yyyy-MM-dd" /> <param name="threshold" value="info"/> <param name="append" value="true"/> <layout class="org.apache.log4j.PatternLayout"> <param name="ConversionPattern" value="[%d{yyyy-MM-dd HH:mm:ss.SSS}] - %X{merchant} - %X{openid} - %X{queryType} - %X{orderId} - %X{w
</layout></appender>1
antiy
2
3
gain什么意思4
5
6
7
8
9[2019-01-27 17:51:22.051] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - nior - aa4820190127175110 - 4200000283201901277968491[2019-01-27 17:51:22.230] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - nior - aa4820190127175110 - 42000002832019277968491434 - 35298[URL]: api./cgi-bin/message/template/nd [PARAMS]: {"tour":"olC5FwLnXjtCbQsW76lkevV57nH0","template_id":"Qt1zyzQs4R1uPrJylGQLSUTS6QcG6UyWB2zDzGt7QGY","url":"mp.weixin.q [RESPONSE]: {"errcode":40003,"errmsg":"invalid openid hint: [mLJNpa06824120]"}[2019-01-27 17:51:22.230] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - nior - aa4820190127175110 -
4200000283201901277968491434 - 352[2019-01-27 17:51:22.231] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - nior - aa4820190127175110 - 4200000283201901277968491434 - 352[2019-01-27 17:51:22.463] - - - - - - - - INFO com.apple.wechat.rvice.Worker:94 - 执⾏任务:QueryTask{msg='C39XQ4NFKPGN', fromUrId='oOEvt [2019-01-27 17:51:23.327] - iPhoneBaoXiu - oisb3smtzToo7jNA4abazKktnECQ - nior - aa4820190127175110 - 4200000283201901277968491434 - 352[2019-01-27 17:51:26.876] - - - - - - - - INFO com.apple.wechat.rvice.Worker:94 - 执⾏任务:QueryTask{msg='C39T81JEHG01', fromUrId='oOEvtjo [2019-01-27 17:51:28.003] - MLdress - oOEvtjsGdmAKrZx81zsACqBjjdsA - nior - e50b20190127175115 - 4200000269201901277039023012 - C39XQ41
2
3
4
你好啊英文5
67
8
9
10
11filter { grok { match => ["message", "\[%{TIMESTAMP_ISO8601:logdate}\] - (?<merchant>[\b\w\s]*) - (?<openid>[\u4e00-\u9fa5\b\w\s]*) - (?<queryType>[\b\w\s]*) - (?<
} date { match => ["logdate", "yyyy-MM-dd HH:mm:ss.SSS"] target => "@timestamp" }}1
2
3
4
5
6
7
8
9
保存后启动logstash即可。命令：
kibana 操作
1. ⾸先是创建索引，因为我们的output配置了index为logstash-%{type}，所以对于的索引为：logstash-wechat-log
2.
创建完索引，我们能发现kibana上⾯列出了我们前⾯定义的字段，如下图。
3. 接下来我们就可以通过各种条件来进⾏⽇志的搜索了。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}\] - (?<merchant>[\b\w\s]*) - (?<openid>[\u4e00-\u9fa5\b\w\s]*) - (?<queryType>[\b\w\s]*) - (?<
} date { match => ["logdate", "yyyy-MM-dd HH:mm:ss.SSS"] target => "@timestamp" }}output { elasticarch { hosts => "**************:9200" index => "logstash-%{type}" template_overwrite => true }}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
wood
18
19
20conductor
21
22
23
24
25
26
27
28nohup ./bin/logstash agent -f f &
1
假设我们要搜索商户是MLdress，⽤户输⼊内容为356110099676675的⽇志信息，那么我们只需要输⼊【merchant:MLdress AND input:356110099676675】即可搜索出对应的⽇志，如下图。
总结
通过在代码使⽤MDC进⾏规范的⽇志打印，在结合logstash提供的强⼤的⽇志采集插件，为了解决⽇
志的换⾏以及我们⾃定义字段等问题我们需要进⾏对logstash的配置⽂件进⾏配置，以将所有服务器的⽇志统⼀上报到es，并通过kibana进⾏放飞⾃我的查询操作，只有这样才能极⼤的提⾼⽇常开发效率。
程序员除了要会写代码之外，另外⼀项加分项那肯定是具有快速发现和定位问题的能⼒了，如果没有ELK这样的⼯具，那快速发现及定位其实是⽐较困难的。
所以，你懂的，赶紧⽤起来吧。
