[elk]Mutate filter plugin增删改查字段

Mutate filter plugin参考: https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html

在线匹配:
http://grokdebug.herokuapp.com/

grok github正则:
https://github.com/kkos/oniguruma/blob/master/doc/RE

logstash grok目录:
/usr/local/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-4.1.2/patterns

主要研究下这个插件的这些功能

增加字段
删除字段
拆分字段
聚合

add_field: 增加字段

input { stdin { codec => "json" } }

filter {
    mutate {
        add_field => { "status_true" => "1" }
    }
}

output {
    stdout { codec => rubydebug }
}

remove_field: 删除字段

input { stdin { codec => "json" } }

filter {
    mutate {
        remove_field => [isp]
    }
}

output {
    stdout { codec => rubydebug }
}

rename: 重命名字段名

input { stdin { codec => "json" } }

filter {
    mutate {
        rename => { "isp" => "province_isp" }
    }
}

output {
    stdout { codec => rubydebug }
}

replace: 修改字段的值(可调用其他字段值)

input { stdin { codec => "json" } }

filter {
    mutate {
        replace => { "isp" => "阿里飞飞" }
    }
}

output {
    stdout { codec => rubydebug }
}

// 相对update多了个调用其他字段的能力

input { stdin { codec => "json" } }

filter {
    mutate {
        replace => { "isp" => "%{isp}: My new message" }
    }
}

output {
    stdout { codec => rubydebug }
}

update: 更新某字段的值(不能调用其他字段)

input { stdin { codec => "json" } }

filter {
    mutate {
        update => { "isp" => "My new message" }
    }
}

output {
    stdout { codec => rubydebug }
}

convert: 转换字段的值的类型

input { stdin { codec => "json" } }

filter {
    mutate {
        convert => { "success" => "string" }
    }
}

output {
    stdout { codec => rubydebug }
}

mutate {  
    convert => { "dest_Port" => "integer" }  
    convert => { "source_Port" => "integer" }
}  

{"mobile" : "15812345606", "province": "上海", "isp": "中国移动","time" : "2017-12-06T09:30:51.244Z", "success" : false}

####################################################

copy: 复制一个字段(重命名字段名/复制字段值)

input { stdin { codec => "json" } }

filter {
    mutate {
        copy => { "isp" => "isps" }
    }
}

output {
    stdout { codec => rubydebug }
}

合并2个字段为1个

input { stdin { codec => "json" } }

filter {
    mutate {
        replace => { "isp_province" => "%{isp} - %{province}" }
        remove_field => [isp, province]
    }
}

output {
    stdout { codec => rubydebug }
}

拆分2个字段为1个

filter {
  mutate {
     copy => { "source_field" => "dest_field" }
  }
}

拆分值

input { stdin { codec => "json" } }

filter {
    mutate {
        replace => { "isp_province" => "%{isp} - %{province}" }
        remove_field => [isp, province]
    }
}

output {
    stdout { codec => rubydebug }
}

lowercase: 值大小写转换

input { stdin { codec => "json" } }

filter {
    mutate {
        lowercase => [ "isp" ]
    }
}

output {
    stdout { codec => rubydebug }
}

{"mobile" : "15812345606", "province": "上海", "isp": "ZGYD","time" : "2017-12-06T09:30:51.244Z", "success" : false}

uppercase: 值大小写转换

input { stdin { codec => "json" } }

filter {
    mutate {
        uppercase => [ "isp" ]
    }
}

output {
    stdout { codec => rubydebug }
}

{"mobile" : "15812345606", "province": "上海", "isp": "zgyd","time" : "2017-12-06T09:30:51.244Z", "success" : false}

split: 值的分割

input { stdin { codec => "json" } }

filter {
    mutate {
        split => { "isp" => ", " }
    }
}

output {
    stdout { codec => rubydebug }
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

{"mobile" : "15812345606", "province": "上海", "isp": "移动, 联通, 电信","time" : "2017-12-06T09:30:51.244Z", "success" : false}

{
    "@timestamp" => 2017-12-08T01:47:37.860Z,
      "province" => "上海",
       "success" => false,
           "isp" => [
        [0] "移动",
        [1] "联通",
        [2] "电信"
    ],
        "mobile" => "15812345606",
      "@version" => "1",
          "host" => "no1.ma.com",
          "time" => "2017-12-06T09:30:51.244Z"
}

kibana效果

strip: 去掉字段值的收尾空格

Strip whitespace from field. NOTE: this only works on leading and trailing whitespace.

input { stdin { codec => "json" } }

filter {
    mutate {
        strip => ["field1", "field2"]
    }
}

output {
    stdout { codec => rubydebug }
}

type&add_tag设type,打tag

打tag为了过滤

input { 
    stdin {
            type => "isp"
            codec => "json"
        }
}

filter {
    mutate {
        add_tag => [ "foo_%{isp}" ]
    }
}

// 根据type分流到不同的index
output {
    stdout { codec => rubydebug }
    
    if [type] == "isp"{
        elasticsearch {
            hosts => [ "localhost:9200" ]
        }
    }
}

{
    "@timestamp" => 2017-12-08T02:14:12.042Z,
      "province" => "上海",
       "success" => false,
           "isp" => "ZGYD",
        "mobile" => "15812345606",
      "@version" => "1",
          "host" => "lb-212-222.above.com",
          "time" => "2017-12-06T09:40:51.244Z",
          "type" => "isp",
          "tags" => [
        [0] "foo_ZGYD"
    ]
}

参考: https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html#plugins-filters-mutate-common-options
http://www.cnblogs.com/qq27271609/p/4762562.html

id字段

这里没帮我改id,不知道为什么

input { stdin { codec => "json" } }

filter {
    mutate {
        id => "ABC"
    }
}

output {
    stdout { codec => rubydebug }
    elasticsearch {
        hosts => [ "localhost:9200" ]
    }
}

{"mobile" : "15812345606", "province": "上海", "isp": "ZGYD","time" : "2017-12-06T10:18:51.244Z", "success" : false}