beat integration with logstash

beat-logstash.config

# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {  
    beats {
    port => 5044
  }
}
filter 
{
	## Stage 1:
	## Parse XML Document convert into json
	xml 
	{
		source => "message" ## source property
		target => "parsed"  ## create new property with json formatted data
	}
	##Stage 2:
	split {
		field => "[parsed][System]"
		add_field => {
			log_type       				=> 		"%{[parsed][System][SubType][0][Name]}"
			timestamp     				=> 		"%{[parsed][System][TimeCreated][0][SystemTime]}"
			tempSource    				=> 		"%{[parsed][System][Source][0][Name]}"
			tempMessage	 				=>		"%{[parsed][ApplicationData][0]}"
			id	 						=>		"%{[parsed][System][Correlation][0][ActivityID]}"
			correlation_id				=>		"%{[parsed][System][Correlation][0][GlobalActivityId]}"
			event_type					=>		"log"
			"[execution][name]"			=>		"%{[parsed][System][Execution][0][ProcessName]}"
			"[execution][pid]"			=>		"%{[parsed][System][Execution][0][ProcessID]}"
			"[execution][thread_id]"	=>		"%{[parsed][System][Execution][0][ThreadID]}"
			machine						=>		"%{[parsed][System][Computer][0]}"
		}
	}
	##Stage 3:
	mutate {
		remove_field => ["parsed", "message","beat", "prospector", "log", "input", "tags", "source", "host", "@timestamp", "offset", "@version", "architecture" ]
	}
	##Stage 4:
	mutate {
		rename => {
			"tempMessage" => "message" 
			"tempSource" => "source"
		}
	}
	##Stage 5:
	mutate {		
		gsub => [
			"id","{","","id","}",""
		]
	}
}
output 
{
    elasticsearch 
    { 
        hosts =>  ["http://localhost:9200"]
        index => "log"
	document_id => "%{id}"
        document_type => "_doc"
    }
    stdout { codec => rubydebug}
}