丰富的过滤器插件是 logstash威力如此强大的重要因素，过滤器插件主要处理流经当前Logstash的事件信息，可以添加字段、移除字段、转换字段类型，通过正则表达式切分数据等，也可以根据条件判断来进行不同的数据处理方式。 [TOC] ## A grok正则捕获插件 [Logstash简单介绍](https://blog.csdn.net/chenleiking/article/details/73563930) [grokdebug在线调试grok](http://grokdebug.herokuapp.com/) [grok预装正则表达式](https://www.cnblogs.com/stozen/p/5638369.html) [github上的预装正则表达式](https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns) grok插件是一个十分耗费资源的插件,是Logstash中将非结构化数据解析成结构化数据以便于查询的最好工具，非常适合解析syslog logs，apache log， mysql log，以及一些其他的web log ### **预定义表达式调用** Logstash提供120个常用正则表达式可供安装使用，安装之后你可以通过名称调用它们，语法如下： ```sh %{SYNTAX:SEMANTIC} SYNTAX：表示已经安装的正则表达式的名称 SEMANTIC：表示给Event中匹配到的内容指定什么名称 ``` >例如：Event的内容为`“[debug] 127.0.0.1 - test log content”`，匹配`%{IP:client}`将获得`“client: 127.0.0.1”`的结果； 如果想对捕获的数据进行数据类型转换,可以使用`%{NUMBER:num:int}`这种语法 默认返回结果都是string类型，且当前Logstash所支持的转换类型仅有“int”和“float”； ### **grok匹配事例：** ```sh #日志内容：55.3.244.1 GET /index.html 15824 0.043 filter { grok { match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"} } } ``` **​输出结果：** ~~~sh client: 55.3.244.1 method: GET request: /index.html bytes: 15824 duration: 0.043 ~~~ ### 自定义表达式 * 自定义表达式语法 语法：`(?<field_name>the_pattern)` 举例：捕获10或11和长度的十六进制`queue_id` 可以使用表达式`(?<queue_id>[0-9A-F]{10,11})` * 安装自定义表达式 可以将自定义的表达式配置到Logstash中，就可以像于定义的表达式一样使用； 1. 在Logstash根目录下创建文件夹`patterns` 2. 在`patterns`文件夹中创建文件`extra`(文件名自定义) 3. 在文件“extra”中添加表达式，格式：`patternName regexp` 4. 如:`POSTFIX_QUEUEID [0-9A-F]{10,11}` 5. 使用自定义的表达式时需要指定“patterns_dir”变量举例如下： ``` filter { grok { patterns_dir => "/path/to/your/own/patterns" match => { "message" => ........" } } } ``` ## B date时间处理插件 这里需要合前面的grok插件剥离出来的值logdate配合使用。可以格式化为需要的样子。 **为什什么要格式化？** 日志产生的时间肯定早于日志在logstash中处理的时间,如果不修改，保存进ES的时间(@timestamp)就是logstash当前处理数据的时间,时间就会不一致 格式化以后，可以通过target属性来指定到@timestamp，这样数据的时间就会是准确的，对以后图表的建设来说万分重要。 最后，顺手删除logdate这个字段已无用的字段。 ~~~ filter{ date{ match=>["logdate","dd/MMM/yyyy:HH:mm:ss Z"] target=>"@timestamp" remove_field => 'logdate' } } ~~~ >需要强调的是，@timestamp字段的值，不可以随便修改，最好就按照数据的某一个时间点来使用 如果是日志，就使用grok把时间抠出来，如果是数据库，就指定一个字段的值来格式化 如果没有这个字段的话，千万不要试着去修改它。 ## C mutate插件 mutate 插件是 Logstash另一个重要插件。提供了丰富的基础类型数据处理能力。可以重命名，删除，替换和修改事件中的字段。 ~~~ filter { mutate { #接收一个数组，其形式为value，type convert => [ #把request_time的值转换为浮点型，把costTime的值转换为整型 "request_time", "float"， "costTime", "integer" ] } } ~~~ * convert 字段转换 将指定字段转换为指定类型，字段内容是数组，则转换所有数组元素，如果字段内容是hash，则不做任何处理 目前支持的转换类型包括：integer,float, string, and boolean. 例如： convert=> { “fieldname” => “integer” } * rename 字段重命名 修改一个或者多个字段的名称。 例如： ` rename=> { “HOSTORIP” => “client_ip” }` * replace 值替换 使用新值完整的替换掉指定字段的原内容，支持变量引用。 例如： 使用字段“source_host”的内容拼接上字符串“: My new message”之后的结果替换“message”的值： `replace=> { “message” => “%{source_host}: My new message” }` * gsub 字符串替换 类似replace方法，但是只针对字符串类型有效 例如：`[ “fieldname”, “/”, “:“, “fieldname2”, “[\\?#-]”, “.”]`， 解释：使用`:`替换掉`fieldname`中的所有`/`，使用“.”替换掉“fieldname2”中的所有`\ ? # 和-` * update 更新字段 更新现有字段的内容， 例如： 将“sample”字段的内容更新为“Mynew message”： `update=> { “sample” => “My new message” }` * join 连接数组字段 使用指定的符号将array字段的每个元素连接起来，对非array字段无效。 例如： 使用`,`将array字段`fieldname`的每一个元素连接成一个字符串： `join=> { “fieldname” => “,” } ` * merge 合并数组 合并两个array或者hash，将一个array和一个hash合并。 例如： 将”added_field”合并到”dest_field”： `merge=> { “dest_field” => “added_field” }` * split 字段分割 按照自定的分隔符将字符串字段拆分成array字段，只能作用于string类型的字段。 例如： 将“fieldname”的内容按照`,`拆分成数组： `split=> { “fieldname” => “,” } ` * strip 去掉空格 去掉字段内容两头的空白字符。例如： `strip=> [“field1”, “field2”]` * uppercase 大写转换 将指定的字段值转换为大写 * lowercase 小写转换 将指定的字段值转换为小写 ## D ruby插件 ruby插件可以使用任何的ruby语法，无论是逻辑判断，条件语句，循环语句，对字符串的操作，或是对EVENT对象的操作 **ruby插件有两个属性，一个init 还有一个code** init属性是用来初始化字段的，这个字段只是在ruby{}作用域里面生效。 code属性使用两个冒号进行标识，所有ruby语法都可以在里面进行。 ~~~ filter { ruby { init => [field={}] code => " array=event.get('message').split('|') array.each do |value| if value.include? 'MD5_VALUE' then require 'digest/md5' md5=Digest::MD5.hexdigest(value) event.set('md5',md5) end if value.include? 'DEFAULT_VALUE' then event.set('value',value) end end remove_field=>"message" " } } #首先，把message字段里面的值拿到，并按照“|”分割为数组 #第二步，循环数组判断其值是否是我需要的数据（ruby条件语法、循环结构） #第三步，把需要的字段添加进入EVEVT对象。 ##event就是Logstash对象，可以在ruby插件的code属性里面操作，可以添加属性字段，删除，修改，数值运算。 #第四步，选取一个值，进行MD5加密或其他操作,最后把冗余的message字段去除。 ~~~ ## E json插件 JSON插件用于解码JSON格式的字符串，一般是一堆日志信息中，部分是JSON格式，部分不是的情况下,会将json格式部分的数据解析出来放到单独的字段中 和input中的codec插件的json区别在于,input中的插件,解析json格式后,会替代message中的数据,而filter中的json插件,解析出的字段是单独的,不会覆盖message的数据 ~~~ filter{ #source指定你的哪个值是json数据。 json { source => "value" } } ~~~ >如果你的json数据是多层的，那么解析出来的数据在多层结里是一个数组，你可以使用ruby语法对他进行操作，最终把所有数据都转换为平级的。 json插件还是需要注意一下使用的方法的，下图就是多层结构的弊端：  对应的解决方案为： ~~~ruby ruby{ code=>" kv=event.get('content')[0] kv.each do |k,v| event.set(k,v) end" remove_field => ['content','value','receiptNo','channelId','status'] } ~~~