[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}