Logstash共通プラグインの説明

filter{
    mutate{
#     covert => ["response","integer","bytes","float"]  #       
        convert => {"message"=>"integer"}
    }
}
#  ------->
{
          "host" => "localhost",
       "message" => 123,    #  “”,int  
    "@timestamp" => 2018-06-26T02:51:08.651Z,
      "@version" => "1"
}

mutate{
    split => {"message"=>","}
}
#---------->
aaa,bbb
{
    "@timestamp" => 2018-06-26T02:40:19.678Z,
      "@version" => "1",
          "host" => "localhost",
       "message" => [
        [0] "aaa",
        [1] "bbb"
    ]}
192,128,1,100
{
        "host" => "localhost",
     "message" => [
      [0] "192",
      [1] "128",
      [2] "1",
      [3] "100"
 ],
  "@timestamp" => 2018-06-26T02:45:17.877Z,
    "@version" => "1"
}

filter{
    mutate{
        add_field => {"field1"=>"value1"}
    }
    mutate{ 
          split => {"message"=>"."}   # message    .  
    }
    mutate{
        merge => {"message"=>"field1"}   # filed1     message  
    }
}
#--------------->
abc
{
       "message" => [
        [0] "abc,"
        [1] "value1"
    ],
    "@timestamp" => 2018-06-26T03:38:57.114Z,
        "field1" => "value1",
      "@version" => "1",
          "host" => "localhost"
}

abc,.123
{
       "message" => [
        [0] "abc,",
        [1] "123",
        [2] "value1"
    ],
    "@timestamp" => 2018-06-26T03:38:57.114Z,
        "field1" => "value1",
      "@version" => "1",
          "host" => "localhost"
}

filter{
    mutate{
        rename => {"message"=>"info"}
    }
}
#-------->
{
    "@timestamp" => 2018-06-26T02:56:00.189Z,
          "info" => "123",
      "@version" => "1",
          "host" => "localhost"
}

mutate {
    remove_field => ["message","datetime"]
}

mutate{
        split => {"message"=>":"}
}
mutate{
        join => {"message"=>","}
}
------>
abc:123
{
    "@timestamp" => 2018-06-26T03:55:41.426Z,
       "message" => "abc,123",
          "host" => "localhost",
      "@version" => "1"
}
aa:cc
{
    "@timestamp" => 2018-06-26T03:55:47.501Z,
       "message" => "aa,cc",
          "host" => "localhost",
      "@version" => "1"
}

mutate{
        gsub => ["message","/","_"]   # _  /
    }

------>
a/b/c/
{
      "@version" => "1",
       "message" => "a_b_c_",
          "host" => "localhost",
    "@timestamp" => 2018-06-26T06:20:10.811Z
}

mutate{
        add_field => {"field1"=>"value1"}
    }
    mutate{
        update => {"field1"=>"v1"}
        update => {"field2"=>"v2"}    #field2        
    }
---------------->
{
    "@timestamp" => 2018-06-26T06:26:28.870Z,
        "field1" => "v1",
          "host" => "localhost",
      "@version" => "1",
       "message" => "a"
}

mutate{
        add_field => {"field1"=>"value1"}
    }
    mutate{
        replace => {"field1"=>"v1"}
        replace => {"field2"=>"v2"}
    }
---------------------->
{
       "message" => "1",
          "host" => "localhost",
    "@timestamp" => 2018-06-26T06:28:09.915Z,
        "field2" => "v2",        #field2   ，   
      "@version" => "1",
        "field1" => "v1"
}

geoip {
            source => "clientip"
            database =>"/tmp/GeoLiteCity.dat"
        }

filter{
    urldecode{
        field => "message"
    }
    ruby {
        init => "@kname = ['url_path','url_arg']"
        code => " 
            new_event = LogStash::Event.new(Hash[@kname.zip(event.get('message').split('?'))]) 
            event.append(new_event)"
    }
    if [url_arg]{
        kv{
            source => "url_arg"
            field_split => "&"
            target => "url_args"
            remove_field => ["url_arg","message"]
        }
    }
}
# ruby  
#  ？    ， request    url_path url_arg
-------------------->
www.test.com?test
{
       "url_arg" => "test",
          "host" => "localhost",
      "url_path" => "www.test.com",
       "message" => "www.test.com?test",  
      "@version" => "1",
    "@timestamp" =>  2018-06-26T07:31:04.887Z
}
www.test.com?title=elk&content=  elk
{
      "url_args" => {
          "title" => "elk",
        "content" => "  elk"
    },
          "host" => "localhost",
      "url_path" => "www.test.com",
      "@version" => "1",
    "@timestamp" =>  2018-06-26T07:33:54.507Z
}

    urldecode{
        field => "message"
    }

# field :  urldecode         ,    "message"
# charset(  ):           .  UTF-8

kv{
        prefix => "url_"   #     key   
        target => "url_ags"    #     key-value      
        source => "message"   #      
        field_split => "&"    #     
        remove_field => "message"
    }
-------------------------->
a=1&b=2&c=3
{
            "host" => "localhost",
       "url_ags" => {
          "url_c" => "3",
          "url_a" => "1",
          "url_b" => "2"
    },
      "@version" => "1",
    "@timestamp" => 2018-06-26T07:07:24.557Z

if [agent] != "-" {
  useragent {
    source => "agent"
    target => "ua"
    remove_field => "agent"
  }
}
# if  ，   agent             
#source      ,    
#target  useragent     ua   。             

output{
    stdout{
        codec => "rubydebug"
    }
}

 file {
       path => "/data/logstash/%{host}/{application}
       codec => line { format => "%{message}"} }
    }

 kafka{
        bootstrap_servers => "localhost:9092"
        topic_id => "test_topic"  #     。       
    }

  elasticsearch {
        hosts => "localhost:9200"
        index => "nginx-access-log-%{+YYYY.MM.dd}"  
    }
#index        。           ，                

input {
  stdin {
    codec => multiline {
      pattern => "pattern, a regexp"    #      ，                
      negate => "true" or "false"     #    false。            。   true，             。
      what => "previous" or "next"    #     。                 。
    }
  }
}
codec => multiline {
    pattern => "^\s"  
    what => "previous"  
}
#               

codec => multiline {
    # Grok pattern names are valid! :)
    pattern => "^%{TIMESTAMP_ISO8601} "
    negate => true
    what => "previous"
}
#                       

codec => multiline {
   pattern => "\\$"
   what => "next"
}
#