目的

处理多种不同的log格式，并打上标签

input log file

pid:9729 2015-03-25 10:47:44.302777 broker.go:97: resuming listening on [::]:8055
pid:9696 2015-03-25 10:47:44.303046 broker.go:119: quiting...
pid:9696 2015-03-25 10:47:44.303112 broker.go:135: detect uncover broker net error
pid:9696 2015-03-25 10:47:44.303126 broker.go:143: broker wait
pid:9696 2015-03-25 10:47:44.303130 broker.go:145: quit broker
pid:9696 2015-03-25 10:47:44.303136 broker.go:124: I am done.

logstash conf

input {
    file {
        path => ["/Users/duwei/go/hydra/logs/broker.log", "/Users/duwei/go/hydra/logs/err.log"]
        codec => "plain"
        start_position => "beginning"
    }
}
filter {
    grok {
        match => ["message", "(?<ts>%{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME}).*quit"]
        remove_field => [ "message" ]
        add_tag => ["ok", "quit"]
    }
    grok {
        match => ["message", "(?<ts>%{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME}).*done"]
        remove_field => [ "message" ]
        add_tag => ["ok", "done"]
    }
    if "ok"  not in [tags] {
        drop { }
    }
}
output {
    stdout { 
        codec => "rubydebug"
    }
}

output

{
      "@version" => "1",
    "@timestamp" => "2015-03-27T03:15:09.636Z",
          "host" => "duwei-rmbp.local",
          "path" => "/Users/duwei/go/hydra/logs/broker.log",
          "tags" => [
        [0] "_grokparsefailure",
        [1] "ok",
        [2] "done"
    ],
            "ts" => "2015/03/19 16:16:30.020656"
}
{
      "@version" => "1",
    "@timestamp" => "2015-03-27T03:15:09.645Z",
          "host" => "duwei-rmbp.local",
          "path" => "/Users/duwei/go/hydra/logs/broker.log",
            "ts" => "2015/03/19 16:16:44.435645",
          "tags" => [
        [0] "ok",
        [1] "quit",
        [2] "_grokparsefailure"
    ]
}

说明

因为只要有一次不命中就会打上_grokparsefailure的tag，所以增加另外一个tag: ok来做成功match的判断

参考

offical document