tomcat 格式化输出到kafka

cat /data/tomcat/conf/server.xml

<Valve className="org.apache.catalina.valves.AccessLogValve" directory="logs"
prefix="localhost_access_log" suffix=".txt"
pattern="%h %l %u %t &quot;%r&quot; %s" />
<Context docBase="/data/webserver/" path="/" reloadable="false" />

输出格式如下：

 172.16.200.16 - - [21/Oct/2016:16:55:03 +0800] "GET /static/My97DatePicker/skin/WdatePicker.css HTTP/1.1" 304

input {
  file {
        path => "/data/tomcat/logs/localhost_access_log.2016-10-24.txt"
        start_position => "beginning"
        type => "tomcat_access"
        }
}

filter {
    if [type] == "tomcat_access" {
        grok{
           match => { "message" => "%{IPORHOST:clientip} %{USER:ident} %{USER:auth} [%{HTTPDATE:timestamp}] "(?:%{WORD:verb} %{URIPATHPARAM:request}(?: HTTP/%{NUMBER:httpversion})?|-)" %{NUMBER:response}"}
       }
    }
}

output {
    if [type] == "tomcat_access" {
      kafka {
          bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
          topic_id => "tomcat_access.log"
          compression_type => "snappy"
       }
    }
}

 logstash 服务器端

input {
    if [type] == "haproxy_http" {
        kafka {
                zk_connect => "zookeeper1:2181,zookeeper2:2181,zookeeper3:2181"
                topic_id => "haproxy_http.log"
                reset_beginning => false
                consumer_threads => 5
                decorate_events => true
                }
  } else if [type] == "haproxy_tcp" {
                kafka {
                zk_connect => "zookeeper1:2181,zookeeper2:2181,zookeeper3:2181"
                topic_id => "haproxy_tcp.log"
                reset_beginning => false
                consumer_threads => 5
                decorate_events => true
                }
 } else if [type] == "tomcat_access" {
                kafka {
                zk_connect => "zookeeper1:2181,zookeeper2:2181,zookeeper3:2181"
                topic_id => "tomcat_access.log"
                reset_beginning => false
                consumer_threads => 5
                decorate_events => true
                }
        }
}




output {
        if [type] == "haproxy_http" {
                elasticsearch {
                hosts => ["es1:9200","es2:9200","es3:9200"]
                manage_template => true
                index => "logstash-haproxy-http.log-%{+YYYY-MM-dd}"
                }
        }
        if [type] == "haproxy_tcp" {
                elasticsearch {
                hosts => ["es1:9200","es2:9200","es3:9200"]
                manage_template => true
                index => "logstash-haproxy-tcp.log-%{+YYYY-MM-dd}"
                }
        }
        if [type] == "tomcat_access" {
                elasticsearch {
                hosts => ["es1:9200","es2:9200","es3:9200"]
                manage_template => true
                index => "logstash-tomcat_access.log-%{+YYYY-MM-dd}"
                }
        }

}