input {
 	beats { port => 5046 type => hdfs }
 }

 filter {
 	if [type] == "hdfs" and "hdfs_On_Premise" in [tags]{
 		if "callerContext=" in [message] and "trackingId=" in [message] {
 			dissect {
 				mapping => {
 					"message" => "%{timestamp} INFO FSNamesystem.audit: allowed=%{allowed}	ugi=%{ugi}	ip=%{hostname}/%{ip}	cmd=%{cmd}	src=%{src}	dst=%{dst}	perm=%{perm}	trackingId=%{tracking_id}	proto=%{proto}	callerContext=%{call_ctx}"
 				}
 			}
 		} else if "trackingId=" in [message] {
 			dissect {
 				mapping => {
 					"message" => "%{timestamp} INFO FSNamesystem.audit: allowed=%{allowed}	ugi=%{ugi}	ip=%{hostname}/%{ip}	cmd=%{cmd}	src=%{src}	dst=%{dst}	perm=%{perm}	trackingId=%{tracking_id}	proto=%{proto}"
 				}
 			}
 		} else if "callerContext=" in [message] {
 			dissect {
 				mapping => {
 					"message" => "%{timestamp} INFO FSNamesystem.audit: allowed=%{allowed}	ugi=%{ugi}	ip=%{hostname}/%{ip}	cmd=%{cmd}	src=%{src}	dst=%{dst}	perm=%{perm}	proto=%{proto}	callerContext=%{call_ctx}"
 				}
 			}
 		} else {
 			dissect {
 				mapping => {
 					"message" => "%{timestamp} INFO FSNamesystem.audit: allowed=%{allowed}	ugi=%{ugi}	ip=%{hostname}/%{ip}	cmd=%{cmd}	src=%{src}	dst=%{dst}	perm=%{perm}	proto=%{proto}"
 				}
 			}
 		}
 		if "_dissectfailure" not in [tags] {
 			hdfs_guardium_filter {}
 		}
 	}
 }

 output {
 	stdout { codec => rubydebug }
 }