Logstash过滤插件

filter初级

Logstash安装

### 设置YUM源
# rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
# tee /etc/yum.repos.d/elastic.repo << EOF
[logstash-5.x]
name=Elastic repository for 5.x packages
baseurl=https://artifacts.elastic.co/packages/5.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF
# yum install -y logstash

基本使用

# tee filter.conf << EOF
input {
stdin {
}
}
filter {
mutate {
split => ["message", "|"]
}
}
output {
stdout {
}
}
EOF

# /usr/share/logstash/bin/logstash -f filter.conf --path.settings /etc/logstash
Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties
The stdin plugin is now waiting for input:
12|fwd|343|dd
2017-09-18T01:35:03.342Z dnode [12, fwd, 343, dd]

ruby语法基本使用

# tee filter.conf << EOF
input {
stdin {
}
}
filter {
mutate {
split => ["message", "|"]
}
ruby {
code => '
msgs = event.get("message")
puts msgs.length
'
}
}
output {
stdout {
codec => "rubydebug"
}
}
EOF

# /usr/share/logstash/bin/logstash -f ruby.conf --path.settings /etc/logstash
Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties
The stdin plugin is now waiting for input:
r|g
2
{
"@version" => "1",
"host" => "dnode",
"@timestamp" => 2017-09-18T09:06:12.546Z,
"message" => [
[0] "r",
[1] "g"
]
}

filter高级用法

grok插件

• 自定义正则： 将需要提取的正则表达式用

  ()

  括起来，然后使用

  ?<tag_name>

  的固定语法格式给匹配项打上标签
• 内置正则： 使用

  %{WORD:tag_name}

  内置正则地址

如果想要给一串很长的字符的很多字段都打上标签，即多个自定义组合的情况，那么正则必须能完全匹配整个字符串（可以使用.*的方式跳过不关心的字段）

在线测试地址

ruby插件

### 1. 先实现rb脚本，输入从变量读取，输出也保存到变量
### 2. 脚本的输入由变量改成event.get("name")
### 3. 脚本的输出由变量改成event.set("name", $value)

举例

样例字符串一

• 使用grok内置正则

• 自定义正则

样例字符串二

将字符串转换成JSON

### 编写rb脚本实现所需功能
# vim ruby.rb
$result = Hash.new
$people = []
begin
msgs = "[MAN] name=fwd age=12#[WONMEN]name=xb age=10"
msgs.split("#").each { |msg|
ret = Hash.new
item = msg[/(?<=\[)MAN(?=\])|(?<=\[)WONMEN(?=\])/]
if item.empty?
raise "Invalid format"
end
ret["sex"] = item

beg = msg.index("name")
if beg == nil
raise "Invalid format"
end
msg[beg..-1].split().each { |item|
key, value = item.split("=")
ret[key] = value
}
$people.push(ret)
}
$result["peoples"] = $people
puts $result
end

# ruby ruby.rb
{"peoples"=>[{"sex"=>"MAN", "name"=>"fwd", "age"=>"12"}, {"sex"=>"WONMEN", "name"=>"xb", "age"=>"10"}]}

将ruby脚本放入Logstash的filter插件中

# vim ruby.conf
input {
stdin {
}
}
filter {
ruby {
code => '
$result = Hash.new
$people = []
begin
msgs = event.get("message")
msgs.split("#").each { |msg|
# 分割后的字符串样例 => [MAN] name=fwd age=12
ret = Hash.new
# 匹配头部的[MAN]或[WONMEN]
item = msg[/(?<=\[)MAN(?=\])|(?<=\[)WONMEN(?=\])/]
if item.empty?
raise "Invalid format"
end
ret["sex"] = item

# 获取从name到结束的字符串 => name=fwd age=12
beg = msg.index("name")
if beg == nil
raise "Invalid format"
end
msg[beg..-1].split().each { |item|
# 分割后的字符串样例 => name=fwd
key, value = item.split("=")
ret[key] = value
}
$people.push(ret)
}
$result["peoples"] = $people
event.set("message", $result)
event.set("[@metadata][drop]", false)
rescue
puts $!
event.set("[@metadata][drop]", true)
end
'
}
}
output {
if ![@metadata][drop] {
stdout {
codec => rubydebug
}
}
}

# /usr/share/logstash/bin/logstash -f ruby.conf --path.settings /etc/logstash
Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties
The stdin plugin is now waiting for input:
[MAN] name=fwd age=12#[WONMEN]name=xb age=10

{
"@version" => "1",
"host" => "dnode",
"@timestamp" => 2017-09-20T08:40:26.293Z,
"message" => {
"peoples" => [
[0] {
"name" => "fwd",
"age" => "12",
"sex" => "MAN"
},
[1] {
"name" => "xb",
"age" => "10",
"sex" => "WONMEN"
}
]
}
}

参考文档

Logstash实践
关于Logstash中grok插件的正则表达式例子
elastic文档
elastic插件文档