ELK-使用nxlog+filebeat采集不同类型的日志输出到logstash
文章目录
ELK-使用nxlog+filebeat采集不同类型的日志输出到logstash前言正常的采集配置开启多个端口监听优化nxlog(关键字放在开头)再优化nxlog(关键字放在结尾)使用filebeat前言
网络上有很例子给出一示例是采集一种类型的日志输出到logstash,但一个系统上日志种类很多,同一个采集端能区分不同日志类型吗?
下面的结构是nxlog做客户端采集,通过tcp协议发送到logstash,然后logstash传输到elasticsearch。
前提条件必需:
安装好elasticsearch,logstash,nxlog,filebeatnxlog/filebeats安装在windows系统,elasticsearch/logstash安装在linux系统
正常的采集配置
nxlog配置nxlog.conf:
<Input in_donglilog>Module im_fileFile "D:\\jar\\dongli\\logs\\spring-boot.log"SavePos TRUE</Input><Output out_donglitcp>Module om_tcpHost 192.168.1.238Port 514</Output><Route 1>Path in_donglilog => out_donglitcp</Route>
采集D:\jar\dongli\logs\spring-boot.log日志,输出到192.168.1.238:514上
logstash配置:
input {tcp {port => 514type=>"plm"}}output{if [type] == "plm"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelian-%{+YYYY.MM.dd}"}}}
通过tcp协议监控514端口,这时候logstash工作mode是server(另一个工作mode是client,用于采集并发送数据),是监控514端口数据的。
网上例子多数止步于些,监听一个端口,然后接收数据,发送到elasticsearch。
假设,我们不仅仅监控dongli的日志D:\jar\dongli\logs\spring-boot.log日志,我还监听另一个系统日志,假设应用为kelian。这2个日志格式不一样。nxlog配置相对简单,主要是logstash怎么能区分接收的不同日志,在elasticsearch创建不同的index。总不能把2个应用日志输出到同一个index吧。
开启多个端口监听
这个方法最简单,不同应用开启不同端口监控
nxlog配置
<Input in_donglilog>Module im_fileFile "D:\\jar\\dongli\\logs\\spring-boot.log"SavePos TRUE</Input><Output out_donglitcp>Module om_tcpHost 192.168.1.238Port 514</Output><Route 1>Path in_donglilog => out_donglitcp</Route><Input in_kelianlog>Module im_fileFile "D:\\jar\\kelaien\\logs\\spring-boot.log"SavePos TRUE</Input><Output out_keliantcp>Module om_tcpHost 192.168.1.238Port 515</Output><Route 2>Path in_kelianlog => out_keliantcp</Route>
logstash配置:
input {tcp {port => 514type=>"dongli"}tcp {port => 515type=>"kelian"}}output{if [type] == "dongli"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "dongli-%{+YYYY.MM.dd}"}}if [type] == "kelian"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelian-%{+YYYY.MM.dd}"}}}
最简单,但我不希望这么做,因为每增加一个应用就会增加一个端口,而增加一个端口就要增加这个端口对外开放,如果是阿里云ECS,还要修改安全组规则。个人觉得麻烦,但是这也不失为一种可选方式
优化nxlog(关键字放在开头)
要是能携带一个数据区分发送日志类型就好了。遗憾的是,nxlog并没有提供这种选项,怎么办?
修改传输的数据。
nxlog每读一行发送到logstash,在每行日志前加个特殊的字符串,然后logstash截取这个字符串,根据这个特殊的字符串创建不同的index。
原理是logstash字符引用,只要是input输入的值,都可以引用
nxlog配置:
<Input in_donglilog>Module im_fileFile "D:\\jar\\dongli\\logs\\spring-boot.log"SavePos TRUE</Input><Input in_kelianlog>Module im_fileFile "D:\\jar\\kelaien\\logs\\spring-boot.log"SavePos TRUE</Input><Processor proc_donglilog>Modulepm_transformerExec $raw_event = "dongli " + $raw_event;</Processor><Processor proc_kelianlog>Modulepm_transformerExec $raw_event = "kelian " + $raw_event;</Processor><Output out_donglitcp>Module om_tcpHost 192.168.1.238Port 514</Output><Output out_keliantcp>Module om_tcpHost 192.168.1.238Port 514</Output><Route 1>Path in_donglilog => proc_donglilog => out_donglitcp</Route><Route 2>Path in_kelianlog => proc_kelianlog => out_keliantcp</Route>
通过Processor模块,在每行日志行都添加了应用名。
logstash配置:
input {tcp {port => 514type=>"plm"}}filter{if [type] == "plm" {grok{match=>{"message" => "%{WORD:key} %{WORD}"}}mutate{gsub=>["message","%{key}",""]}}}output{if [type] == "plm"{if [key] == "dongli" {elasticsearch {hosts => ["127.0.0.1:9200"]index => "dongli-%{+YYYY.MM.dd}"}}if [key] == "kelian" {elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelian-%{+YYYY.MM.dd}"}}}}
处理的关键在过滤器代码中
grok{match=>{#拿到应用名"message" => "%{WORD:key} %{WORD}"}}mutate{#将message里应用名替换为空gsub=>["message","%{key}",""]}
在output里就可以使用字段引用功能做判断了
if [type] == "plm"{if [key] == "dongli" {}if [key] == "kelian" {}}
缺点
有个缺点,只对单行日志有作用,如果有多行合并为一行的异常日志就不适合,因为将关键字添加到行头,破坏了数据完。
当使用multiline插件过滤输入数据时,行头没办法区分,当使用multiline插件,合并多行时会出问题。
codec => multiline{#以[开头视为一行pattern => "^["negate => truewhat => "previous"}
再优化nxlog(关键字放在结尾)
如果把关键字放在每行的结尾呢
nxlog配置:
<Input in_donglilog>Module im_fileFile "D:\\jar\\dongli\\logs\\spring-boot.log"SavePos TRUE</Input><Input in_kelianlog>Module im_fileFile "D:\\jar\\kelaien\\logs\\spring-boot.log"SavePos TRUE</Input><Processor proc_donglilog>Modulepm_transformerExec $raw_event = $raw_event + "(dongli)";</Processor><Processor proc_kelianlog>Modulepm_transformerExec $raw_event = $raw_event + "(kelian)";</Processor><Output out_donglitcp>Module om_tcpHost 192.168.1.238Port 514</Output><Output out_keliantcp>Module om_tcpHost 192.168.1.238Port 514</Output><Route 1>Path in_donglilog => proc_donglilog => out_donglitcp</Route><Route 2>Path in_kelianlog => proc_kelianlog => out_keliantcp</Route>
注意代码
<Processor proc_donglilog>Modulepm_transformerExec $raw_event = $raw_event + "(dongli)";</Processor><Processor proc_kelianlog>Modulepm_transformerExec $raw_event = $raw_event + "(kelian)";</Processor>
关键放在行尾并用小括号括起来。
logstash配置:
input {tcp {port => 514codec => multiline{pattern => "^\d{4}(\-|\/|.)\d{1,2}\1\d{1,2}"negate => truewhat => "previous"}type=>"plm"}}filter{if [type] == "plm" {grok{match=>{"message" => "(?<ckey>[(]\w+[)\\r])"}}mutate{gsub=>["message","[(]%{ckey}[)]",""]#gsub=>["ckey","\r",""]}}}output{if [type] == "plm"{if [ckey] == "(dongli)" {elasticsearch {hosts => ["127.0.0.1:9200"]index => "dongli-%{+YYYY.MM.dd}"}}if [ckey] == "(kelian)" {elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelian-%{+YYYY.MM.dd}"}}}}
主要的处理在过滤器里
if [type] == "plm" {grok{match=>{"message" => "(?<ckey>[(]\w+[)\\r])"}}mutate{gsub=>["message","[(]%{ckey}[)]",""]}}
拿到关键字,然后把message字段里关键删除。
这种方式也不失为一种解决方案,但不绝不是优雅的方案
使用filebeat
它天生可多携带关键字,而且在windows运行也很稳定,所以我推荐使用filebeat替代nxlog。
filebeat.yml配置:
filebeat.inputs:- type: logenabled: truepaths:- D:\jar\dongli\logs\spring-boot.loggfields:appname: dongli- type: logenabled: truepaths:- D:\jar\kelaien\logs\spring-boot.logfields:appname: kelaien
logstash配置
input{beats {port => 515type=>"beatss"}}output{if [fields][appname] == "dongli"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "dongli-%{+YYYY.MM.dd}"}}if [fields][appname] == "kelaien"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelaien-%{+YYYY.MM.dd}"}}}
上面对应的是单行日志,如果是多行日志,它的配置放在filebeats而不是logstash
filebeat.yml配置:
filebeat.inputs:- type: logenabled: truepaths:- D:\jar\dongli\logs\spring-boot.loggmultiline:pattern: '^\d{4}-\d{1,2}-\d{1,2}'negate: truematch: afterfields:appname: dongli- type: logenabled: truepaths:- D:\jar\kelaien\logs\spring-boot.logmultiline:pattern: '^\d{4}-\d{1,2}-\d{1,2}'negate: truematch: afterfields:appname: kelaien
多行的关键代码是
multiline:pattern: '^\d{4}-\d{1,2}-\d{1,2}'negate: truematch: after