Logstash_过滤插件
1、Grok正则插件
Grok 是 Logstash 最重要的插件。你可以在 grok 里预定义好命名正则表达式，在稍后(grok参数或者其他正则表达式里)引用它。
Grok 支持把预定义的 grok 表达式 写入到文件中，官方提供的预定义 grok 表达式见：https://github.com/logstash/logstash/tree/v1.4.2/patterns。
正则表达式语法
运维工程师多多少少都会一点正则。你可以在 grok 里写标准的正则，像下面这样：
\s+(?<request_time>\d+(?:\.\d+)?)\s+
小贴士：这个正则表达式写法对于 Perl 或者 Ruby 程序员应该很熟悉了，Python 程序员可能更习惯写 (?P<name>pattern)，没办法，适应一下吧。
现在给我们的配置文件添加第一个过滤器区段配置。配置要添加在输入和输出区段之间(logstash 执行区段的时候并不依赖于次序，不过为了自己看得方便，还是按次序书写吧)：
input{ stdin{} }
filter{
grok {
match => {
"message" => "%{INT:int}"
}
}
}
output{ stdout{ codec => rubydebug } }
运行 logstash 进程然后输入 "begin 123"，你会看到类似下面这样的输出：
{
"message" => "begin 123",
"@version" => "1",
"@timestamp" => "2014-08-09T11:55:38.186Z",
"host" => "raochenlindeMacBook-Air.local",
"int" => "123"
}
这个时候int成功获取到了数值，但是会发现这是个字符串的，可以做如下修改改变类型
filter{
grok {
match => {
"message" => "%{INT:int:int}" # 改变之后的
#"message" => “%{INT:int}” # 改变之前的
}
}
}
我们可以清除的看到两者的区别
grok 表达式的打印复制格式的完整语法是下面这样的：
%{PATTERN_NAME:capture_name:data_type}
小贴士：data_type 目前只支持两个值：int 和 float。
重新运行进程然后可以得到如下结果：
{
"message" => "begin 123",
"@version" => "1",
"@timestamp" => "2014-08-09T11:55:38.186Z",
"host" => "raochenlindeMacBook-Air.local",
"int" => 123
}
这次获取到的结果是一个int类型的值
自定义正则
实际应用中，我们需要处理各种各样的数据，大部分都可以通过官方提供的grok表达式直接调用
# 调用的语法为
%{PATTERN_NAME:capture_name:data_type}
但是有些特殊情况 这些无法满足我们的需求，这个时候我们就需要自己编写正则了，正则的格式如下
(?<name>正则表达式) # 这只是定义正则的一种方式 还有其他方式
最佳实战
实际运用中，我们需要处理各种各样的日志文件，如果你都是在配置文件里各自写一行自己的表达式，就完全不可管理了。所以，我们建议是把所有的 grok 表达式统一写入到一个地方。然后用 filter/grok 的 patterns_dir选项来指明。
下面是一个实例
配置文件
input { stdin{} }
filter{
grok{
patterns_dir => "/root/patherns" # 只想自定义文件所在的目录
match => {
"message" => "%{JIAO:user} %{JIAOX:int}" # 进行过滤
}
remove_field => "message" # 过滤成功之后就删除message数据
}
}
output{ stdout{ codec=>rubydebug } }
/root/patherns/jiao 自定义文件
# 前面为字段名 后面为字段名对应的正则或者引用grok表达式
JIAO %{USERNAME} # 引用grok表达式
JIAOX \d+ # 自定义正则
在命令行输入 alice 123456 , 查看控制台输出结果
{
"int" => "123456",
"user" => "alice",
"@version" => "1",
"host" => "a656fc35ea31",
"message" => "alice 123456",
"@timestamp" => 2019-11-23T09:04:58.276Z
}
可以看到 文件内自定义的正则也可以正常使用
多项选择
有时候我们会碰上一个日志有多种可能格式的情况。这时候要写成单一正则就比较困难，或者全用 | 隔开又比较丑陋。这时候，logstash 的语法提供给我们一个有趣的解决方式。
文档中，都说明 logstash/filters/grok 插件的 match 参数应该接受的是一个 Hash 值。但是因为早期的 logstash 语法中 Hash 值也是用 [] 这种方式书写的，所以其实现在传递 Array 值给 match 参数也完全没问题。所以，我们这里其实可以传递多个正则来匹配同一个字段：
match => [
"message", "(?<request_time>\d+(?:\.\d+)?)",
"message", "%{SYSLOGBASE} %{DATA:message}",
"message", "(?m)%{WORD}"
]
logstash 会按照这个定义次序依次尝试匹配，到匹配成功为止。虽说效果跟用 | 分割写个大大的正则是一样的，但是可阅读性好了很多。
我强烈建议每个人都要使用 Grok Debugger 来调试自己的 grok 表达式。
时间处理
date过滤器用于解析字段中的日期，然后使用该日期或时间戳作为事件的logstash时间戳。
例如，syslog事件通常有这样的时间戳：
"Apr 17 09:32:01"
你应该使用MMM dd HH:mm:ss的日期格式来解析这个。
日期过滤器对于事件的排序和对旧数据的回填特别重要，如果在你的事件中没有得到正确的日期，那么以后搜索它们可能会出现顺序不对。
如果没有这个过滤器，logstash将根据第一次看到事件（在输入时）的时间（如果时间戳还没有在事件中设置）选择一个时间戳，例如，对于文件输入，时间戳被设置为每次读取的时间。
通过日期过滤器过滤之后，时间戳会按照日志文件内日志发生的时间，例如nginx访问日志，这样就可以知道什么时候，哪个ip访问了我们的服务器，具体访问了哪台服务器，以什么方式访问等等信息
日期实例
配置文件
input{ stdin{} }
filter{
grok {
match => {
"message" => "(?<time>.*)" # 定义字段名 time
}
}
date {
match => ["time","yyyy-MM-dd HH:mm:ss"] # 上面提供的字段名time
# 以及要在命令行的时间对应的时间格式
}
}
注意 日期格式一定要和日志内的时间格式或者自己输入的时间格式一致
执行命令，输入时间， 查看对应的结果
2000-11-11 10:10:10 # 这是输入的时间
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/awesome_print-1.7.0/lib/awesome_print/formatters/base_formatter.rb:31: warning: constant ::Fixnum is deprecated
{
"time" => "2000-11-11 10:10:10", # time字段
"host" => "a656fc35ea31",
"@version" => "1",
"@timestamp" => 2000-11-11T10:10:10.000Z, # 可以看到时间戳变成了命令行输入的时间
"message" => "2000-11-11 10:10:10"
}
-
下面是用来解析日期和时间的方法：
-
y 年
- yyyy => 全年数字，例如：
2015。 - yy => 两位数的年，例如：
15代表2015年。
- yyyy => 全年数字，例如：
-
M 一年中的月份
- M => 最小位数的月份，例如：
1为1月，12为12月。 - MM => 两位数的月份，如果需要将用零填充，例如：
01为1月，12为12月。 - MMM => 缩写月文本，例如：
Jan为1月，注意：使用的语言取决于你的语言环境，查看locale设置如何更改语言。 - MMMM => 全月的文本，例如：
January，注意：使用的语言取决于你的语言环境。
- M => 最小位数的月份，例如：
-
d 一月中的天
- d => 最小位数的天，例如：
1为一个月的第一天。 - dd => 两位数的天，如果需要将用零填充，例如：
01为一个月的第一天。
- d => 最小位数的天，例如：
-
H 一天中的小时（24小时时钟）
- H => 最小位数的小时，例如：
0为午夜。 - HH => 两位数的小时，如果需要，零填充，例如：
00为午夜。
- H => 最小位数的小时，例如：
-
m 一小时中的分钟（每小时60分钟）
- m => 最小位数的分钟，例如：
0。 - mm => 两位数的分钟，如果需要，零填充，例如：
00。
- m => 最小位数的分钟，例如：
-
s 一分钟中的秒（每分钟60秒）
- s => 最小位数的秒，例如：
0。 - ss => 两位数的秒，如果需要，零填充，例如：
00。
- s => 最小位数的秒，例如：
-
S 每秒最大精度的分数为毫秒(
SSS)，超过这个，追加零- S => 十分之一秒，例如：
0表示亚秒值012。 - SS => 百分之一秒，例如：
01为亚秒值01。 - SSS => 千分之一秒，例如：
012为亚秒值012。
- S => 十分之一秒，例如：
-
Z 时区偏移或标识
- Z => 时区偏移结构为HHmm（Zulu/UTC的小时和分钟偏移），例如：
-0700。 - ZZ => 时区偏移结构为HH:mm（在小时和分钟偏移量之间的冒号），例如：
-07:00。 - ZZZ => 时区的标识，例如：
America/Los_Angeles，注意：有效的id列在Joda.org可用时区页面上。
- Z => 时区偏移结构为HHmm（Zulu/UTC的小时和分钟偏移），例如：
-
z 时区名称，无法解析时区名称（z）。
-
w 一年中的周
- w => 最小位数的周，例如：
1。 - ww => 两位数的周，如果需要，零填充，例如：
01。
- w => 最小位数的周，例如：
-
D 一年中的天
-
e 一周中的星期（数字）
-
E 一周中的星期（文本）
- E, EE, EEE => 一周中缩短的星期，例如：
Mon，Tue，Wed，Thu，Fri，Sat，Sun，注意：它的实际语言取决于你的语言环境。 - EEEE => 一周的星期全称，例如：
Monday，Tuesday，...，注意：它的实际语言取决于你的语言环境。
查看更多点击此处
- E, EE, EEE => 一周中缩短的星期，例如：
GeoIP 地址查询归类
GeoIP 是最常见的免费 IP 地址归类查询库，同时也有收费版可以采购。GeoIP 库可以根据 IP 地址提供对应的地域信息，包括国别，省市，经纬度等，对于可视化地图和区域统计非常有用。
配置示例
input { stdin{ } }
filter {
geoip {
source => "message" # 对命令行输入的ip进行地址归类查询
# source是固定的，对应的字段可以是任一处理后的字段，
# 注意 只有公网ip才可以
}
}
output { stdout { codec => rubydebug } }
运行结果
183.60.92.253
{
"message" => "183.60.92.253",
"geoip" => {
"longitude" => 113.25,
"country_code3" => "CN",
"region_name" => "Guangdong",
"country_code2" => "CN",
"location" => {
"lon" => 113.25,
"lat" => 23.1167
},
"country_name" => "China",
"region_code" => "GD",
"timezone" => "Asia/Shanghai",
"latitude" => 23.1167,
"continent_code" => "AS",
"ip" => "183.60.92.253"
},
"host" => "a656fc35ea31",
"@timestamp" => 2019-11-23T10:39:59.990Z,
"@version" => "1"
}
配置说明
GeoIP 库数据较多，如果你不需要这么多内容，可以通过 fields 选项指定自己所需要的。下例为全部可选内容：
filter {
geoip {
fields => ["city_name", "continent_code", "country_code2", "country_code3", "country_name", "dma_code", "ip", "latitude", "longitude", "postal_code", "region_name", "timezone"]
}
}
需要注意的是：geoip.location 是 logstash 通过 latitude 和 longitude额外生成的数据。所以，如果你是想要经纬度又不想重复数据的话，应该像下面这样做：
filter {
geoip {
fields => ["city_name", "country_code2", "country_name", "latitude", "longitude", "region_name"] # 指定显示的字段
remove_field => ["[geoip][latitude]", "[geoip][longitude]"]
# 在显示字段的基础 上移除某些字段
}
}
小贴士
geoip 插件的 "source" 字段可以是任一处理后的字段，比如 "client_ip"，但是字段内容却需要小心！geoip 库内只存有公共网络上的 IP 信息，查询不到结果的，会直接返回 null，而 logstash 的 geoip 插件对 null 结果的处理是：不生成对应的 geoip.字段。
所以读者在测试时，如果使用了诸如 127.0.0.1, 172.16.0.1, 182.168.0.1, 10.0.0.1 等内网地址，会发现没有对应输出！
