Logstash configuration for TCP input, JSON filter and ElasticSearch output

logstash.conf

input { 
	tcp{
		port => 5000
	}
}

filter {
	# Only process messages that have the keywords Audit or System.
	if ([message] =~ "Audit|System" ) {
		json {
			# Parses the incoming JSON message into fields.
			source => "message"
		}
		
		# Replace logstash timestamp with log's timestamp
		date {
			match => [ "Timestamp", "ISO8601" ]
		}
		
		mutate {
			# Revert special chars in the JSON fields.
			gsub => [ "Description", "\\n", "\n"]
			gsub => [ "ErrorMessage", "\\n", "\n"]
			gsub => [ "DebugInformation", "\\n", "\n"]
			gsub => [ "StackTrace", "\\n", "\n"]
			gsub => [ "ProbableCause", "\\n", "\n"]
			gsub => [ "ProposedAction", "\\n", "\n"]
			
			gsub => [ "Description", "\\t", "\t"]
			gsub => [ "ErrorMessage", "\\t", "\t"]
			gsub => [ "DebugInformation", "\\t", "\t"]
			gsub => [ "StackTrace", "\\t", "\t"]
			gsub => [ "ProbableCause", "\\t", "\t"]
			gsub => [ "ProposedAction", "\\t", "\t"]
			
			# Replace logstash client host with log's host
			replace => [ "host", "%{Hostname}" ]
		}
	}
	
	# If we succeed we remove the original message from
	#	the doc so it won't be stored as a field 
	#	which would be pointless since we parsed and
	#	stored it's fields separately.
	if ("_jsonparsefailure" not in [tags] and [Type] =~ "Audit|System") {
		mutate {
			remove_field => [ "message" ]
		}
	}
}

output {
	# If the doc is ok, push to ES, otherwise push to file for later analysis.
	if ("_jsonparsefailure" not in [tags] and [Type] =~ "Audit|System") {
		elasticsearch {
			host => localhost 
		}
		
		# output to console for debugging purposes
		stdout { 
			codec => rubydebug
		}
	} else {
		file {
			path => ["logs/log-%{+YYYY-MM-dd}.txt"]
		}
	}
}