logstash的filter模块

丰富的过滤器插件是 logstash威力如此强大的重要因素，过滤器插件主要处理流经当前Logstash的事件信息，可以添加字段、移除字段、转换字段类型，通过正则表达式切分数据等，也可以根据条件判断来进行不同的数据处理方式。

A grok正则捕获插件

• 预定义表达式调用
• grok匹配事例：
• 自定义表达式

B date时间处理插件
C mutate插件
D ruby插件
E json插件

A grok正则捕获插件

Logstash简单介绍
grokdebug在线调试grok
grok预装正则表达式
github上的预装正则表达式
grok插件是一个十分耗费资源的插件,是Logstash中将非结构化数据解析成结构化数据以便于查询的最好工具，非常适合解析syslog logs，apache log， mysql log，以及一些其他的web log

预定义表达式调用
Logstash提供120个常用正则表达式可供安装使用，安装之后你可以通过名称调用它们，语法如下：

%{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匹配事例：**

#日志内容：55.3.244.1 GET /index.html 15824 0.043
filter {
grok {
match => {“message” => “%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}”}
}
}
​输出结果：

client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043

自定义表达式
自定义表达式语法
语法：(?the_pattern)
举例：捕获10或11和长度的十六进制queue_id
可以使用表达式(?[0-9A-F]{10,11})
安装自定义表达式
可以将自定义的表达式配置到Logstash中，就可以像于定义的表达式一样使用；
在Logstash根目录下创建文件夹patterns
在patterns文件夹中创建文件extra(文件名自定义)
在文件“extra”中添加表达式，格式：patternName regexp
如:POSTFIX_QUEUEID [0-9A-F]{10,11}
使用自定义的表达式时需要指定“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插件还是需要注意一下使用的方法的，下图就是多层结构的弊端：
UTOOLS1574316439843.png
对应的解决方案为：

ruby{
    code=>"
      kv=event.get('content')[0]
      kv.each do |k,v|
      event.set(k,v)
      end"
      remove_field => ['content','value','receiptNo','channelId','status']
}