logstash yml

input {
  beats {
    port => 5044
  }
}
filter {
    grok {
        match => {
                "message" => "%{TIMESTAMP_ISO8601:log_timestamp} %{IP:s_ip} %{WORD:cs_method} %{URIPATH:cs_uri_stem} %{NOTSPACE:cs_uri_query} %{NUMBER:s_port} %{NOTSPACE:cs_username} %{IP:c_ip} %{NOTSPACE:cs_useragent} %{NOTSPACE:cs_referer} %{NUMBER:sc_status} %{NUMBER:sc_substatus} %{NUMBER:sc_win32_status} %{NUMBER:time_taken}"
        }
        remove_field => ["message"]
        #存儲的結果中刪除message屬性
    }
}
output {
        if "_grokparsefailure" not in [tags]{
                #根據規則轉化成功的日誌才輸出到es
                elasticsearch {
                        # 導出到es，最常用的插件
                        codec => "json"
                        hosts => ["node1.larry.com:9200"]
            index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
            #document_type => "%{[@metadata][type]}"
                        #根據請求接口名動態生成動態
                        #document_type => "test"
                        #flush_size => 500
                        #idle_flush_time => 1
                }
                #這裡除了es，還可以把轉換好的內容輸出到其他路徑
        }else{
                #轉換失敗的日誌就不做處理
        }
}