《ELK Stack权威指南(第2版)》一2.3 过滤器配置-阿里云开发者社区

开发者社区> 华章出版社> 正文
登录阅读全文

《ELK Stack权威指南(第2版)》一2.3 过滤器配置

简介:

本节书摘来自华章出版社《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中这样计算:




版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享: