ELK-使用nxlog+filebeat采集不同类型的日志输出到logstash

布满荆棘的人生 2022-03-12 07:21 558阅读 0赞

## ELK-使用nxlog+filebeat采集不同类型的日志输出到logstash ##

### 文章目录 ###

*   *  ELK-使用nxlog+filebeat采集不同类型的日志输出到logstash
 *  前言
 *  正常的采集配置
 *  开启多个端口监听
 *  优化nxlog（关键字放在开头）
 *  再优化nxlog（关键字放在结尾）
 *  使用filebeat

# 前言 #

网络上有很例子给出一示例是采集一种类型的日志输出到logstash，但一个系统上日志种类很多，同一个采集端能区分不同日志类型吗？  
下面的结构是nxlog做客户端采集，通过tcp协议发送到logstash，然后logstash传输到elasticsearch。  
前提条件必需：

1.  安装好elasticsearch,logstash,nxlog,filebeat
2.  nxlog/filebeats安装在windows系统，elasticsearch/logstash安装在linux系统

# 正常的采集配置 #

nxlog配置nxlog.conf:

采集D:\\jar\\dongli\\logs\\spring-boot.log日志，输出到192.168.1.238:514上

logstash配置：

input {
    	tcp {
    		port => 514
    		type=>"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_file
    	File "D:\\jar\\dongli\\logs\\spring-boot.log"
    	SavePos TRUE
    </Input>
    
    <Output out_donglitcp>
    	Module om_tcp
    	Host 192.168.1.238
    	Port 514
    </Output>
    
    <Route 1>
    	Path in_donglilog => out_donglitcp
    </Route>
    
    <Input in_kelianlog>
    	Module im_file
    	File "D:\\jar\\kelaien\\logs\\spring-boot.log"
    	SavePos TRUE
    </Input>
     <Output out_keliantcp>
    	Module om_tcp
    	Host 192.168.1.238
    	Port 515
    </Output>
    <Route 2>
    	Path in_kelianlog => out_keliantcp
    </Route>

logstash配置：

input {
    	tcp {
    		port => 514
    		type=>"dongli"
       	}
       	tcp {
    		port => 515
    		type=>"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_file
    	File "D:\\jar\\dongli\\logs\\spring-boot.log"
    	SavePos TRUE
    </Input>
    <Input in_kelianlog>
    	Module im_file
    	File "D:\\jar\\kelaien\\logs\\spring-boot.log"
    	SavePos TRUE
    </Input>
    
    
    <Processor proc_donglilog>
    	Module      pm_transformer
    	Exec $raw_event = "dongli " + $raw_event;
    </Processor>
    <Processor proc_kelianlog>
    	Module      pm_transformer
    	Exec $raw_event = "kelian " + $raw_event;
    </Processor>
    
    <Output out_donglitcp>
    	Module om_tcp
    	Host 192.168.1.238
    	Port 514
    </Output>
     <Output out_keliantcp>
    	Module om_tcp
    	Host 192.168.1.238
    	Port 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 => 514
    		type=>"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 => true
    			what => "previous"
    	}

# 再优化nxlog（关键字放在结尾） #

如果把关键字放在每行的结尾呢  
nxlog配置：

<Input in_donglilog>
    	Module im_file
    	File "D:\\jar\\dongli\\logs\\spring-boot.log"
    	SavePos TRUE
    </Input>
    <Input in_kelianlog>
    	Module im_file
    	File "D:\\jar\\kelaien\\logs\\spring-boot.log"
    	SavePos TRUE
    </Input>
    
    
    <Processor proc_donglilog>
    	Module      pm_transformer
    	Exec $raw_event = $raw_event + "(dongli)";
    </Processor>
    <Processor proc_kelianlog>
    	Module      pm_transformer
    	Exec $raw_event = $raw_event + "(kelian)";
    </Processor>
    
    <Output out_donglitcp>
    	Module om_tcp
    	Host 192.168.1.238
    	Port 514
    </Output>
     <Output out_keliantcp>
    	Module om_tcp
    	Host 192.168.1.238
    	Port 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>
    	Module      pm_transformer
    	Exec $raw_event = $raw_event + "(dongli)";
    </Processor>
    <Processor proc_kelianlog>
    	Module      pm_transformer
    	Exec $raw_event = $raw_event + "(kelian)";
    </Processor>

关键放在行尾并用小括号括起来。  
logstash配置：

input {
    	tcp {
    		port => 514
    		codec => multiline{
    			pattern => "^\d{4}(\-|\/|.)\d{1,2}\1\d{1,2}"
    			negate => true
    			what => "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: log
      enabled: true
    
      paths:
        - D:\jar\dongli\logs\spring-boot.logg
      fields:
        appname: dongli
    - type: log
      enabled: true
      paths:
        - D:\jar\kelaien\logs\spring-boot.log
      fields:
        appname: kelaien

logstash配置

input{
    	beats  {
    		port => 515
    		type=>"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: log
      enabled: true
    
      paths:
        - D:\jar\dongli\logs\spring-boot.logg
       multiline:
        pattern: '^\d{4}-\d{1,2}-\d{1,2}'
        negate: true
        match: after
      fields:
        appname: dongli
    - type: log
      enabled: true
      paths:
        - D:\jar\kelaien\logs\spring-boot.log
       multiline:
        pattern: '^\d{4}-\d{1,2}-\d{1,2}'
        negate: true
        match: after
      fields:
        appname: kelaien

多行的关键代码是

multiline:
        pattern: '^\d{4}-\d{1,2}-\d{1,2}'
        negate: true
        match: after