Filter是Logstash功能强大的主要原因,它可以对Logstash Event进行丰富的处理,比如说解析数据、删除字段、类型转换等等,常见的有如下几个:

date: 日志解析
grok:正则匹配解析
dissect:分割符解析
mutate:对字段做处理,比如重命名、删除、替换等
json:按照json解析字段内容到指定字段中
geoip:增加地理位置数据
ruby: 利用ruby代码来动态修改Logstash Event
date插件
从字段解析日期以用作事件的Logstash时间戳,以下配置解析名为logdate的字段以设置Logstash时间戳:

filter {
date {
match => [ “logdate”, “MMM dd yyyy HH:mm:ss” ]
}
}
1
2
3
4
5
返回结果:

{“logdate”:”Jan 01 2018 12:02:03″}
{
“@version” => “1”,
“host” => “Node2”,
“@timestamp” => 2018-01-01T04:02:03.000Z,
“logdate” => “Jan 01 2018 12:02:03”
}
1
2
3
4
5
6
7
8
说明:
match:类型为数组,用于指定日期匹配的格式,可以一次指定多种日志格式

match => [ “logdate”, “MMM dd yyyy HH:mm:ss” ,”MMM d yyyy HH:mm:ss”,”ISO8601″]
1
target:类型为字符串,用于指定赋值的字段名,默认是@timestamp
timezone:类型为字符串,用于指定时区

关于logstash时区的问题可以参考:logstash 时间戳时区问题

grok插件
将非结构化事件数据分析到字段中。 这个工具非常适用于系统日志,Apache和其他网络服务器日志,MySQL日志,以及通常为人类而不是计算机消耗的任何日志格式。

55.3.244.1 GET /index.html 15824 0.043
1
以下配置将消息解析为字段:

filter {
grok {
match => { “message” => “%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}” }
}
}
1
2
3
4
5
应用过滤器后,示例中的事件将具有以下字段:

client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043
1
2
3
4
5
Grok语法:

%{SYNTAX:SEMANTIC}   # SYNTAX为grok pattern的名称,SEMANTIC为赋值字段名称

%{NUMBER:duration}可以匹配数值类型,但是grok匹配出的内容都是字符串类型,可以通过在最后指定为int或者float来强制转换类型。
%{NUMBER:duration:float}
1
2
3
4
5
常见pattern可以查看:GitHub或者logstash家目录下的:

vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns
1
自定义匹配规则:
格式:(?<field_name>the pattern here)

pattern_definitions参数,以键值对的方式定义pattern名称和内容
pattern_dir参数,以文件的形式被读取
filter{
grok {
match => {“message”=>”%{SERVICE:service}”}
pattern_definitions => {“SERVICE” => “[a-z0-9]{10,11}”}
#patterns_dir => [“/opt/logstash/patterns”, “/opt/logstash/extra_patterns”]
}
}
1
2
3
4
5
6
7
8
tag_on_failure: 默认是_grokparsefailure,可以基于此做判断

调试:
正则表达式:
https://regexr.com/

grok:
http://grokdebug.herokuapp.com/
http://grok.elasticsearch.cn/
x-pack

dissect插件
基于分隔符原理解析数据,解决grok解析时消耗过多cpu资源的问题

使用分隔符将非结构化事件数据提取到字段中。 解剖过滤器不使用正则表达式,速度非常快。 但是,如果数据的结构因行而异,grok过滤器更合适。

dissect的应用有一定的局限性:主要适用于每行格式相似且分隔符明确简单的场景
dissect语法比较简单,有一系列字段(field)和分隔符(delimiter)组成

%{}字段
%{}之间是分隔符
1
2
例如,假设日志中包含以下消息:

Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool…
1
以下配置解析消息:

filter {
dissect {
mapping => { “message” => “%{ts} %{+ts} %{+ts} %{src} %{prog}[%{pid}]: %{msg}” }
}
}
1
2
3
4
5
解剖过滤器应用后,事件将被解剖到以下领域:

{
“msg”        => “Starting system activity accounting tool…”,
“@timestamp” => 2017-04-26T19:33:39.257Z,
“src”        => “localhost”,
“@version”   => “1”,
“host”       => “localhost.localdomain”,
“pid”        => “1”,
“message”    => “Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool…”,
“type”       => “stdin”,
“prog”       => “systemd”,
“ts”         => “Apr 26 12:20:02”
}
1
2
3
4
5
6
7
8
9
10
11
12
说明:

Apr 26 12:20:02
%{ts} %{+ts} %{+ts}     #+代表该匹配值追加到ts字段下
{
“ts”:”Apr 26 12:20:02″
}

two three one go
%{+order/2} %{+order/3} %{+order/1} %{+order/4}     #/后面的数字代表拼接的次序
{
“order”: “one two three go”
}

a=1&b=2
%{?key1}=%{&key1}&%{?key2}=%{&key2}  #%{?}代表忽略匹配值,但是富裕字段名,用于后续匹配用;%{&}代表将匹配值赋予key1的匹配值
{
“a”:”1″,
“b”:”2″
}

#dissect可以自动处理空的匹配值
John Smith,Big Oaks,Wood Lane,Hambledown,Canterbury,CB34RY
%{name},%{addr1},%{addr2},%{addr3},%{city},%{zip}

Jane Doe,4321 Fifth Avenue,,,New York,87432
{
“name”:”Jane Doe”,
“addr1″:”4321 Fifth Avenue”,
“addr2″:””,
“addr3″:””,
“city”:”New York”,
“zip”:”87432″
}

#dissect分割后的字段值都是字符串,可以使用convert_datatype属性进行类型转换
filter{
dissect{
convert_datatype => {age => “int”}
}
}

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
mutate插件
使用最频繁的操作,可以对字段进行各种操作,比如重命名、删除、替换、更新等,主要操作如下:

convert   #类型转换
gsub      #字符串替换
split/join/merge    #字符串切割、数组合并为字符串、数组合并为数组
rename    #字段重命名
update/replace   #字段内容更新或替换
remove_field     #删除字段
1
2
3
4
5
6
7
convert:实现字段类型的转换,类型为hash,仅支持转换为integer、float、string和Boolean

filter{
mutate{
convert => {“age” => “integer”}
}
}
1
2
3
4
5
6
gsub:对字段内容进行替换,类型为数组,每3项为一个替换配置

filter {
mutate {
gsub => [
# replace all forward slashes with underscore
“fieldname”, “/”, “_”,
# replace backslashes, question marks, hashes, and minuses
# with a dot “.”
“fieldname2”, “[\\?#-]”, “.”
]
}
}
1
2
3
4
5
6
7
8
9
10
11
12
split: 将字符串切割为数组

filter {
mutate {
split => { “fieldname” => “,” }
}
}
1
2
3
4
5
6
join:将数组拼接为字符串
merge:将两个数组合并为1个数组,字符串会被转为1个元素的数组进行操作
rename:字段重命名
update/replace:更新字段内容,区别在于update只在字段存在时生效,而replace在字段不存在时会执行新增字段的操作

filter {
mutate {
update => { “sample” => “My new message” }
update => { “message” => “source from c:%{source_host}” }   #%{source_host}可以引用logstash Event中的字段值
}
}
1
2
3
4
5
6
7
input {
stdin{type=>stdin}
}
filter{
dissect{ mapping => {“message” => “%{a}-%{b}-%{c}”} }
mutate{ replace => {“d” =>”source from c:%{c}”} }
}
output{
stdout{codec=>rubydebug}
}

hi-hello-123
{
“a” => “hi”,
“b” => “hello”,
“@timestamp” => 2018-06-29T02:01:24.473Z,
“c” => “123”,
“d” => “source from c:123”,
“@version” => “1”,
“host” => “Node2”,
“message” => “hi-hello-123”,
“type” => “stdin”
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
json插件
将字段内容为json格式的数据进行解析

filter {
json {
source => “message”     #要解析的字段名
target => “msg_json”    #解析后的存储字段,默认和message同级别
}
}
1
2
3
4
5
6
7
geoip插件
常用的插件,根据ip地址提供对应的地域信息,比如经纬度、城市名等,方便进行地理数据分析

filter {
geoip {
source => “clientip”
}
}
1
2
3
4
5
ruby插件
最灵活的插件,可以以ruby语言来随心所欲的修改Logstash Event对象

filter{
ruby{
code => ‘size = event.get(“message”).size;
event.set(“message_size”,size)’
}
}

ruby {
code => “event.set(‘@read_timestamp’,event.get(‘@timestamp’))”
}
————————————————
版权声明:本文为CSDN博主「wfs1994」的原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接及本声明。
原文链接:https://blog.csdn.net/wfs1994/article/details/80862952

发表评论

邮箱地址不会被公开。 必填项已用*标注