example.conf

input {
	http {
		host => "127.0.0.1"
		port => "5543"
		type => "mautic-lead"
	}

}
filter {

if [type] == "mautic-lead" {
		mautic {
			source => "message"
			remove_field => ["headers", "message", "host"]
		}
		if [type] == "lead"{
			mutate {
				replace => { "@timestamp" => "%{dateAdded}"}
			}
		}
		if [type] == "form_submission"{
			mutate {
				replace => { "@timestamp" => "%{dateSubmmited}"}
			}
		}
		if [type] == "email"{
			mutate {
				replace => { "@timestamp" => "%{dateSent}"}
			}
		}
		if [type] == "page_hit"{
			mutate {
				replace => { "@timestamp" => "%{dateHit}"}
			}
		}
		
	}

}
output {
	if [type] == "lead" {
		elasticsearch { 
			hosts => ["localhost:9200"] 
			index => ["mautic-leads"]
			#document_id => "%{[leadid]}"
		}
	}
	else if [type] == "form_submission" {
		elasticsearch { 
			hosts => ["localhost:9200"] 
			index => ["mautic-leads"]
			document_id => "F%{[leadid]}-%{[form][id]}-%{[submissionid]}"
			routing => "%{[leadid]}"
		}
	}
	else if [type] == "page_hit" {
		elasticsearch { 
			hosts => ["localhost:9200"] 
			index => ["mautic-leads"]
			routing => "%{[leadid]}"
		}
	}
	else if [type] == "email" {
		elasticsearch { 
			hosts => ["localhost:9200"] 
			index => ["mautic-leads"]
			document_id => "E%{[leadid]}-%{[email][id]}-%{[emailopenid]}"
			routing => "%{[leadid]}"
		}
	}
}