本节书摘来自华章出版社《ELK Stack权威指南（第2版）》一书中的第2章，第2.3节，作者 饶琛琳 更多章节内容可以访问云栖社区“华章计算机”公众号查看。
2.3 过滤器配置
有丰富的过滤器插件，是Logstash威力如此强大的重要因素。名为过滤器，其实提供的不单单是过滤的功能。下面我们就会重点介绍几个插件，它们扩展了进入过滤器的原始数据，进行复杂的逻辑处理，甚至可以无中生有地添加新的Logstash事件到后续的流程中去！
2.3.1 date时间处理
之前章节已经提过，logstash-filter-date插件可以用来转换你的日志记录中的时间字符串，变成LogStash::Timestamp对象，然后转存到@timestamp字段里。
因为在稍后的logstash-output-elasticsearch中常用的%{+YYYY.MM.dd}这种写法必须读取@timestamp数据，所以一定不要直接删掉这个字段保留自己的字段，而是应该用logstash-filter-date转换后删除自己的字段！
这在导入旧数据的时候固然非常有用，而在实时数据处理的时候同样有效，因为一般情况下数据流程中我们都会有缓冲区，导致最终的实际处理时间跟事件产生时间略有偏差。
强烈建议打开Nginx的access_log配置项的buffer参数，对极限响应性能有极大提升！
1.配置示例
logstash-filter-date插件支持五种时间格式：
ISO8601：类似“2011-04-19T03:44:01.103Z”这样的格式。具体Z后面可以有“08:00”也可以没有，“.103”这个也可以没有。常用场景里来说，Nginx的log_format配置里就可以使用$time_iso8601变量来记录请求时间成这种格式。
UNIX：UNIX时间戳格式，记录的是从1970年起始至今的总秒数。Squid默认日志格式中就使用了这种格式。
UNIX_MS：这个时间戳则是从1970年起始至今的总毫秒数。据我所知，JavaScript里经常使用这个时间格式。
TAI64N：TAI64N格式比较少见，是这个样子的：@4000000052f88ea32489532c。我目前只知道常见应用中，qmail会用这个格式。
Joda-Time库：Logstash内部使用了Java的Joda时间库来作时间处理。所以我们可以使用Joda库所支持的时间格式来作具体定义。Joda时间格式定义见表2-1。
表2-1 Joda时间库格式
格式符 含 义 描 述 示 例
2.时区问题的解释
很多中国用户经常提一个问题：为什么@timestamp比我们晚了8个小时？怎么修改成北京时间？
其实，Elasticsearch内部，对时间类型字段，是统一采用UTC时间，存成long长整形数据的！对日志统一采用UTC时间存储，是国际安全/运维界的一个通识—欧美公司的服务器普遍广泛分布在多个时区里—不像中国，地域横跨五个时区却只用北京时间。
对于页面查看，ELK的解决方案是在Kibana上，读取浏览器的当前时区，然后在页面上转换时间内容的显示。
所以，建议大家接受这种设定。否则，即便你用.getLocalTime修改，也还要面临在Kibana过去修改，以及Elasticsearch原有的["now-1h" TO "now"]这种方便的搜索语句无法正常使用的尴尬。
以上，请读者自行斟酌。
2.3.2 grok正则捕获
grok是Logstash最重要的插件。你可以在grok里预定义好命名正则表达式，在稍后（grok参数或者其他正则表达式里）引用它。
1.正则表达式语法
运维工程师多多少少都会一点正则。你可以在grok里写标准的正则，像下面这样：
\s+(?<request_time>\d+(?:\.\d+)?)\s+
这个正则表达式写法对于Perl或者Ruby程序员应该很熟悉了，Python程序员可能更习惯写（?P<name>pattern），没办法，适应一下吧。
现在给我们的配置文件添加第一个过滤器区段配置。配置要添加在输入和输出区段之间（Logstash执行区段的时候并不依赖于次序，不过为了自己看得方便，还是按次序书写吧）：
2. grok表达式语法
grok支持把预定义的grok表达式写入到文件中，官方提供的预定义grok表达式见：https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns。
下面是从官方文件中摘抄的最简单但是足够说明用法的示例：
USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}
第一行，用普通的正则表达式来定义一个grok表达式；第二行，通过打印赋值格式（sprintf format），用前面定义好的grok表达式来定义另一个grok表达式。
grok表达式的打印复制格式的完整语法见下行示例。其中data_type目前只支持两个值：int和float。
%{PATTERN_NAME:capture_name:data_type}
所以我们可以改进我们的配置成下面这样：
3.最佳实践
实际运用中，我们需要处理各种各样的日志文件，如果你都是在配置文件里各自写一行自己的表达式，就完全不可管理了。所以，我们建议是把所有的grok表达式统一写入到一个地方。然后用filter/grok的patterns_dir选项来指明。
如果你把“message”里所有的信息都grok到不同的字段了，数据实质上就相当于是重复存储了两份。所以你可以用remove_field参数来删除掉message字段，或者用overwrite参数来重写默认的message字段，只保留最重要的部分。
重写参数的示例如下：
filter {
grok {
patterns_dir =>["/path/to/your/own/patterns"]
match => {
"message" =>"%{SYSLOGBASE} %{DATA:message}"
}
overwrite => ["message"]
}
}
更多有关grok正则性能的最佳实践（比如timeout_millis等配置参数），请参考：https://www.elastic.co/blog/do-you-grok-grok。
4.高级用法
多行匹配 在和codec/multiline搭配使用的时候，需要注意一个问题，grok正则和普通正则一样，默认是不支持匹配回车换行的。就像你需要=~ // m一样也需要单独指定，具体写法是在表达式开始位置加（?m）标记。如下所示：
match => {
"message" =>"(?m)\s+(?<request_time>\d+(?:\.\d+)?)\s+"
}
多项选择 有时候我们会碰上一个日志有多种可能格式的情况。这时候要写成单一正则就比较困难，或者全用|隔开又比较丑陋。这时候，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 (http://grokdebug.herokuapp.com/)来调试自己的grok表达式。
2.3.3 dissect解析
grok作为Logstash最广为人知的插件，在性能和资源损耗方面同样也广为诟病。为了应对这个情况，同时也考虑到大多数情况下日志格式并没有那么复杂，Logstash开发团队在5.0版新添加了另一个解析字段的插件：dissect。当日志格式有比较简明的分隔标志位且重复性较大的时候，可以使用dissect插件更快地完成解析工作。下面是解析syslog的示例。
filter {
dissect {
mapping => {
"message" => "%{ts} %{+ts} %{+ts} %{src} %{} %{prog}[%{pid}]: %{msg}"
}
convert_datatype => {
pid => "int"
}
}
}
我们看到上面使用了和Grok很类似的%{}语法来表示字段，这显然是基于习惯延续的考虑。不过示例中%{+ts}的加号就不一般了。dissect除了字段外面的字符串定位功能以外，还通过几个特殊符号来处理字段提取的规则：
%{+key} 这个+表示前面已经捕获到一个key字段了，而这次捕获的内容自动添补到之前key字段内容的后面。
%{+key/2} 这个/2表示在有多次捕获内容都填到key字段里的时候，拼接字符串的顺序谁前谁后。/2表示排第2位。
%{?string} 这个?表示这块只是一个占位，并不会实际生成捕获字段存到Event里面。
%{?string} %{&string} 当同样捕获名称都是string，但是一个?和一个&在一起的时候，表示这是一个键值对。
比如对http://rizhiyi.com/index.do?id=123写这么一段配置：
http://%{domain}/%{?url}?%{?arg1}=%{&arg1}
则最终生成的 Event 内容是这样的：
{
domain => "rizhiyi.com",
id => "123"
}
2.3.4 GeoIP地址查询
GeoIP是最常见的免费IP地址归类查询库，同时也有收费版可以采购。GeoIP库可以根据IP地址提供对应的地域信息，包括国别、省市、经纬度等，对于可视化地图和区域统计非常有用。
配置示例如下：
2.3.5 JSON编解码
在上一章，已经讲过在Codec中使用JSON编码。但是，有些日志可能是一种复合的数据结构，其中只有一部分记录是JSON格式的。这时候，我们依然需要在filter阶段，单独启用JSON解码插件。
配置示例如下：
2.3.6 key-value切分
在很多情况下，日志内容本身都是一个类似于key-value的格式，但是格式具体的样式却是多种多样的。Logstash提供logstash-filter-kv插件，帮助处理不同样式的key-value日志，变成实际的LogStash::Event数据。
配置示例如下：
2.3.7 metrics数值统计
logstash-filter-metrics插件是使用Ruby的Metriks模块来实现在内存里实时地计数和采样分析。该模块支持两个类型的数值分析：meter和timer。下面分别举例说明。
1. Meter示例（速率阈值检测）
Web访问日志的异常状态码频率是运维人员会非常关心的一个数据。通常我们的做法是通过Logstash或者其他日志分析脚本，把计数发送到rrdtool或者graphite里面，然后再通过check_graphite脚本之类的东西来检查异常并报警。
事实上这个事情可以直接在Logstash内部就完成。比如如果最近一分钟504请求的个数超过100个就报警，如下所示：
2. Timer示例（box and whisker异常检测）
官版的logstash-filter-metrics插件只适用于metric事件的检查。由插件生成的新事件内部不存有来自input区段的实际数据信息。所以，要完成我们的百分比分布箱体检测，需要首先对代码稍微做几行变动，即在metric的timer事件里加一个属性，存储最近一个实际事件的数值：
2.3.8 mutate数据修改
logstash-filter-mutate插件是Logstash另一个重要插件，它提供了丰富的基础类型数据处理能力，包括类型转换、字符串处理和字段处理等。
1.类型转换
类型转换是logstash-filter-mutate插件最初诞生时的唯一功能，其应用场景在之前的2.3.5节“JSON编解码”中已经提到。
可以设置的转换类型包括：“integer”、“float”和“string”。示例如下：
filter {
mutate {
convert => ["request_time", "float"]
}
}
mutate除了转换简单的字符值，还支持对数组类型的字段进行转换，即将[“1”,“2”]转换成[1,2]。但不支持对哈希类型的字段做类似处理。有这方面需求的可以采用稍后讲述的logstash-filter-ruby插件完成。
2. 字符串处理
有如下字符串处理的插件：
3.字段处理
字段处理的插件有：
rename：重命名某个字段，如果目的字段已经存在，会被覆盖掉，如下所示：
filter {
mutate {
rename => ["syslog_host", "host"]
}
}
update：更新某个字段的内容。如果字段不存在，不会新建。
replace：作用和update类似，但是当字段不存在的时候，它会起到add_field参数一样的效果，自动添加新的字段。
4.执行次序
需要注意的是，filter/mutate内部是有执行次序的。其次序如下：
2.3.9 随心所欲的Ruby处理
如果你稍微懂那么一点点Ruby语法的话，logstash-filter-ruby插件将会是一个非常有用的工具。比如你需要稍微修改一下LogStash::Event对象，但是又不打算为此写一个完整的插件，用logstash-filter-ruby插件绝对感觉良好。
配置示例如下：
filter {
ruby {
init =>"@kname = ['client','servername','url','status','time','size','upstream',
'upstreamstatus','upstreamtime','referer','xff','useragent']"
code => "
ew_event = LogStash::Event.new(Hash[@kname.zip(event.get('message').split('|'))])
new_event.remove('@timestamp')
event.append(new_event)"
}
}
官网示例是一个比较有趣但是没啥大用的做法—随机取消90%的事件。
所以上面我们给出了一个有用而且强大的实例。
通常我们都是用logstash-filter-grok插件来捕获字段的，但是正则耗费大量的CPU资源，很容易成为Logstash进程的瓶颈。
而实际上，很多流经Logstash的数据都是有自己预定义的特殊分隔符的，我们可以很简单的直接切割成多个字段。
从Logstash2.3开始，LogStash::event.append不再直接接受Hash对象，而必须是LogStash:: Event对象。所以示例变成要先初始化一个新的event，再把无用的@timestamp移除，再append进去。否则会把@timestamp变成有两个时间的数组了！
从Logstash 5.0开始，LogStash::Event改为Java实现，直接使用event["parent"]["child"]形式获取的，不是原事件的引用而是复制品，需要改用event.get('[parent][child]')和event.set('[parent][child]', 'value')的方法。
logstash-filter-mutate插件里的“split”选项只能切成数组，后续很不方便使用和识别。而在logstash-filter-ruby里，我们可以通过“init”参数预定义好由每个新字段的名字组成的数组，然后在“code”参数指定的Ruby语句里通过两个数组的zip操作生成一个哈希并添加进数组里。短短一行Ruby代码，可以减少50%以上的CPU使用率。
logstash-filter-ruby插件用途远不止这一点，下一节你还会继续见到它的身影。
更多实例如下：
filter{
date {
match => ["datetime" , "UNIX"]
}
ruby {
code =>"event.cancel if 5 * 24 * 3600 < (event.get( '@timestamp')-::Time.now).abs"
}
}
在实际运用中，我们几乎肯定会碰到出乎意料的输入数据。这都有可能导致Elasticsearch集群出现问题。
当数据格式发生变化，比如UNIX时间格式变成UNIX_MS时间格式，会导致Logstash疯狂创建新索引，集群崩溃。
或者误输入过老的数据时，因为一般我们会close几天之前的索引以节省内存，必要时再打开。而直接尝试把数据写入被关闭的索引会导致内存问题。
这时候我们就需要提前校验数据的合法性。上面配置，就是用于过滤掉时间范围与当前时间差距太大的非法数据的。
2.3.10 split拆分事件
上一章我们通过multiline插件将多行数据合并进一个事件里，那么反过来，也可以把一行数据，拆分成多个事件。这就是split插件。
配置示例如下：
filter {
split {
field =>"message"
terminator =>"#"
}
}
这个测试中，我们在intputs/stdin的终端中输入一行数据：“test1#test2”，结果看到输出两个事件：
2.3.11 交叉日志合并
Splunk有一项非常有用的功能，叫做transaction。可以在错乱的多行日志中，根据connected字段、maxspan窗口、startswith/endwith标签等信息计算出事件的duration和count结果。其文档见：http://docs.splunk.com/Documentation/Splunk/latest/SearchReference/Transaction
ELK中，承载计算功能的Elasticsearch并不支持这种跨行计算，所以，变通的处理方式是：在Logstash中，提前做好事件的归并，直接计算出来transaction的duration数据。
比如一个transaction task_id startswith=START endwith=END的查询，可以在Logstash中这样计算：
