input {
	kafka {
        	bootstrap_servers => ["xxx:9092"]
        	group_id => "logstash_xxxx"
        	topics => ["xxx"]
        	auto_offset_reset => ["latest"]
        	consumer_threads => 6
        	decorate_events => true
        	codec => "json"
        }
}


filter {
      if "exception.Exception" in [message] or "exception.ApiException" in [message] {
          drop { }
      }
      grok {
           match => [ "message", "(?<time>^\d+-\d+-\d+\s\d+:\d+:\d+.\d+)" ]
         }
      grok {
           match => [ "message", "(?<level>[I|E]\w+)" ]
         }
      grok {
           match => [ "message", "(?<tid>[nio|o|io].*-\d+)" ]
         }
      grok {
           match => [ "message", "(?<class>[c]\w+.\w+.\w+.\w+.\w+.\w+.\w+.)" ]
         }
      grok {
           match => [ "message", "(?<method>\[\s\s+.*\d\]{1}\s)" ]
         }
      grok {
           match => [ "message", "(?<ms>[0-9]+[m][s])" ]
           #match => [ "message", "(?<ms>\d+)+ms" ]
         }
      grok {
           match => [ "message", "(?<msg>\s:+.*)" ]
         }

      date {
            match => [ "time" , "yyyy-MM-dd HH:mm:ss.SSS" ]
            target => "@timestamp"
      }
     ## 解析嵌套json格式数据
      mutate {
         add_field => { "@fields" => "%{fields}" }
      }
      
      json {
         source => "@fields"
         ## 删除filebeat 自带的不需要的元数据
         remove_field => [ "beat","@fields","fields","index_name","offset","source","message","time","tags"]
      }
      mutate {
        add_field => { "message" => "%{msg}" }
        remove_field => [ "msg" ]
      }
}
#

output {
        elasticsearch {
        ## 使用Search guard
        hosts => ["https://es:9200"]
        index => "xxxxx_%{+YYYYMMdd}"
        user => admin
        password => "xxxxxxxxx"
        ssl => true
        ssl_certificate_verification => true
        truststore => "/opt/logstash/ssl/truststore.jks"
        truststore_password => "xxxxxxxxxx"
         }
       
#stdout { codec => rubydebug }
}