Logstash configuration for ossec logs

ossec.conf

# Content of the file /etc/logstash/conf.d/ossec.log
# Logstash is on the same machine as OSSEC. 
# Logstash should run with OSSEC group privileges

input {
	 file {
		type => "OSSEC"
		path => "/var/ossec/logs/alerts/alerts.log"
		codec => multiline {
		pattern => "^\*\*"
		negate => true
		what => "previous"
		}
 	}
}

filter {
	if [type] == "OSSEC" {
	grok {
		match => ["message", "(?m)\*\* Alert %{DATA}:%{SPACE}%{WORD}?%{SPACE}\- %{DATA:ossec_group}\n%{YEAR} %{SYSLOGTIMESTAMP} \(%{DATA:reporting_host}\) %{IP:reporting_ip}\-\>%{DATA:reporting_source}\nRule: %{NONNEGINT:rule_number} \(level %{NONNEGINT:severity}\) \-\> '%{DATA:signature}'\n%{GREEDYDATA:remaining_message}"]
		match => ["message", "(?m)\*\* Alert %{DATA}:%{SPACE}%{WORD}?%{SPACE}\- %{DATA:ossec_group}\n%{YEAR} %{SYSLOGTIMESTAMP} %{DATA:reporting_host}\-\>%{DATA:reporting_source}\nRule: %{NONNEGINT:rule_number} \(level %{NONNEGINT:severity}\) \-\> '%{DATA:signature}'\n%{GREEDYDATA:remaining_message}"]
	}
	grok {
		match => ["remaining_message", "(?m)(Src IP: %{IP:src_ip}%{SPACE})?(Src Port: %{NONNEGINT:src_port}%{SPACE})?(Dst IP: %{IP:dst_ip}%{SPACE})?(Dst Port: %{NONNEGINT:dst_port}%{SPACE})?(User: %{USER:user}%{SPACE})?%{GREEDYDATA:real_message}"]
	}
	grok {
		match => ["real_message", "(?m)\S+\syum\S+\sInstalled:\s%{GREEDYDATA:package}" ]
		match => ["real_message", "(?m)\S+\syum\S+\sErased:\s%{GREEDYDATA:package}" ]
		match => ["real_message", "(?m)\S+\syum\S+\sUpdated:\s%{GREEDYDATA:package}" ]
		match => ["real_message", "(?m)Integrity checksum changed for: \'%{PATH:file}\'%{GREEDYDATA}" ]
		match => ["real_message", "(?m)File \'%{PATH:file}\' was deleted%{GREEDYDATA}" ]
		match => ["real_message", "(?m)\S+\sarpwatch: new station %{IP} %{COMMONMAC:src_mac} %{GREEDYDATA}" ]
		match => ["real_message", "(?m)\S+\sarpwatch: bogon %{IP} %{COMMONMAC:src_mac} %{GREEDYDATA}" ]
		match => ["real_message", "(?m)\S+\sarpwatch: ethernet mismatch %{IP} %{COMMONMAC:src_mac} %{GREEDYDATA}" ]
	}
	mutate {
		convert      => [ "severity", "integer"]
		remove_field => [ "syslog_program", "tags", "remaining_message", "path", "real_message" ]
	}
	
	if [src_ip] {
		geoip {
			source => "src_ip"
                	target => "src_geoip"
                	database => "/opt/logstash/vendor/geoip/GeoLiteCity.dat"
                	fields => [ "country_code2", "country_name" ]
                }
	}
                
        if [dst_ip] {
                geoip {
                	source => "dst_ip"
			target => "dst_geoip"
			database => "/opt/logstash/vendor/geoip/GeoLiteCity.dat"
			fields => [ "country_code2", "country_name" ]
                }
        }
	}
}

output {
        if [type] == "OSSEC" {
                elasticsearch {
                        flush_size => 1
                        protocol => "transport"
                        host => "xxx.xxx.xxx.xxx"
                        cluster => "xxxxx"
                        index => "logstash-ossec-%{+YYYY.MM.dd}"
                }
        }
}