input {

	beats {
		port => 5044
	}

	kafka {
		bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
		topics => ["msgTopic"]
		consumer_threads => 1
		decorate_events => true
		group_id => "logstash"
		codec => json {
			charset => "UTF-8"
		}
	}
}

filter {

	if [type] == "catalinalog" {
		grok {
			match => { "message" => "%{COMBINEDAPACHELOG}" }
		}
		date {
			match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
		}
	}

}

output {

	stdout {
		codec => rubydebug
	}

	elasticsearch {
		hosts => ["es-coordinating:9200"]
		index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
		document_type => "%{[@metadata][type]}"
	}

}