filebeat-kafka-logstash-elasticsearch日志收集[中文时区问题、自己创建type类index]

logstash配置文件如下：

input{
    kafka{
        bootstrap_servers => "xxxx:9092,xxxx:9092,dw72.xxxx.:9092,...."
        group_id => "xxxx_service_server_logstash"
        topics => ["xxxx_service_server_error", "xxxx_service_server_runtime"]
        auto_offset_reset => latest
            codec => "json"
        consumer_threads => 10
        auto_commit_interval_ms => 500
    }
}

filter {
        grok {
                #patterns_dir => ["./patterns"]
                match => {
                        "message" => "%{TIMESTAMP_ISO8601:logdatetime}"
                }
        }
        date {
                match => [ "logdatetime", "yyyy-MM-dd HH:mm:ss"]
                target => "@timestamp"
#               timezone => "Asia/Shanghai"
                timezone => "+00:00"
                locale => "en"
        }
}

output{
        elasticsearch{
            hosts => ["192.168.1.89:9200"]
                action => "index"
            index => "%{[type]}-%{+YYYY.MM.dd}"
            flush_size => 8000
        }
}

主要关注两个问题：
1.event的timestamp问题，中国区会显示比实际的时间早8个小时（一般不建议修改时间，国际标准，在kibana中做处理。），这里只有国内的服务器，显示为区域时间比较好看，所以我就改了。
2.es 的index根据filebeat中的document_type自动创建index。