当前位置:   article > 正文

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

nxlog output elasticsearch

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

前言

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

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

正常的采集配置

nxlog配置nxlog.conf:

<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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

采集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}"
		}
  }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

通过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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

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}"
		}
  }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24

最简单,但我不希望这么做,因为每增加一个应用就会增加一个端口,而增加一个端口就要增加这个端口对外开放,如果是阿里云ECS,还要修改安全组规则。个人觉得麻烦,但是这也不失为一种可选方式

优化nxlog(关键字放在开头)

要是能携带一个数据区分发送日志类型就好了。遗憾的是,nxlog并没有提供这种选项,怎么办?
修改传输的数据。
nxlog每读一行发送到logstash,在每行日志前加个特殊的字符串,然后logstash截取这个字符串,根据这个特殊的字符串创建不同的index。

原理是logstash字符引用,只要是input输入的值,都可以引用
  • 1

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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

通过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}"
			}
		}
	}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

处理的关键在过滤器代码中

grok{
			match=>{
			 #拿到应用名
				"message" => "%{WORD:key} %{WORD}"
			}
		}
		mutate{
		#将message里应用名替换为空
			gsub=>["message","%{key}",""]
	    }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

在output里就可以使用字段引用功能做判断了

if [type] == "plm"{
		if [key] == "dongli" {	
		}
		if [key] == "kelian" {	
		}
	}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

缺点
有个缺点,只对单行日志有作用,如果有多行合并为一行的异常日志就不适合,因为将关键字添加到行头,破坏了数据完。
当使用multiline插件过滤输入数据时,行头没办法区分,当使用multiline插件,合并多行时会出问题。

codec => multiline{
			#以[开头视为一行
			pattern => "^["
			negate => true
			what => "previous"
	}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

再优化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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41

注意代码

<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>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

关键放在行尾并用小括号括起来。
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}"
			}
		}
	}
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41

主要的处理在过滤器里

if [type] == "plm" {
		grok{
			match=>{
				"message" => "(?<ckey>[(]\w+[)\\r])"
			}
		}
		mutate{
			gsub=>["message","[(]%{ckey}[)]",""]
	    }
	}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

拿到关键字,然后把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
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

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}"
		}
	}
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21

上面对应的是单行日志,如果是多行日志,它的配置放在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
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

多行的关键代码是

 multiline:
    pattern: '^\d{4}-\d{1,2}-\d{1,2}'
    negate: true
    match: after
  • 1
  • 2
  • 3
  • 4
声明:本文内容由网友自发贡献,转载请注明出处:【wpsshop】
推荐阅读
相关标签
  

闽ICP备14008679号