input{
	kafka{+
		bootstrap_servers => ["192.168.234.*:9092,192.168.234.*:9092"]
		group_id => "loan-log-consumer"
		auto_offset_reset => "latest"
		consumer_threads => 5
		topics => ["loan-log"]
		type => "loan"
	}
}


filter{
	if[type] == "loan"{
        mutate{
        split=>["message","|"]
			add_field => {
					"application" => "%{[message][0]}"
			}
			add_field => {
					"ip" => "%{[message][1]}"
			}
			add_field => {
					"traceId" => "%{[message][2]}"
			}
			add_field => {
					"parentSpanId" => "%{[message][3]}"
			}
			add_field => {
					"spanId" => "%{[message][4]}"
			}
			add_field => {
					"tmpTime" => "%{[message][5]}"
			}
			add_field => {
					"thread" => "%{[message][6]}"
			}
			add_field => {
					"level" => "%{[message][7]}"
			}
			add_field => {
					"class" => "%{[message][8]}"
			}
			add_field => {
					"msg" => "%{[message][9]}"
			}
			remove_field => ["message"]
        }
	date {
        	match => [ "tmpTime", "yyyy-MM-dd HH:mm:ss.SSS" ]
        	target => "time"
		remove_field => ["tmpTime"]
      	     }
	}
}

output {
	if[type] == "loan"{
	  elasticsearch{
		   hosts => ["192.168.234.*:9200","192.168.234.*:9200"]
		   index => "loan-index-%{+YYYY.MM.dd}"
		   timeout => 300
		   manage_template => false
		   template_name => "loan_template"
	  }
	}
}