Logstash configuration for reading OSSEC alerts files and send to Elasticsearch (credits to https://mig5.net)

logstash-ossec-alerts.conf

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

filter {
  # Parse the header of the alert
  grok {
    # Matches  2014 Mar 08 00:57:49 (some.server.com) 10.1.2.3->ossec
    # (?m) fixes issues with multi-lines see https://logstash.jira.com/browse/LOGSTASH-509
    match => ["message", "(?m)\*\* Alert %{DATA:timestamp_seconds}:%{SPACE}%{WORD}?%{SPACE}\- %{DATA:ossec_group}\n%{YEAR} %{SYSLOGTIMESTAMP:syslog_timestamp} \(%{DATA:reporting_host}\) %{IP:reporting_ip}\-\>%{DATA:reporting_source}\nRule: %{NONNEGINT:rule_number} \(level %{NONNEGINT:severity}\) \-\> '%{DATA:signature}'\n%{GREEDYDATA:remaining_message}"]
      
    # Matches  2014 Mar 08 00:00:00 ossec-server01->/var/log/auth.log
    match => ["message", "(?m)\*\* Alert %{DATA:timestamp_seconds}:%{SPACE}%{WORD}?%{SPACE}\- %{DATA:ossec_group}\n%{YEAR} %{SYSLOGTIMESTAMP:syslog_timestamp} %{DATA:reporting_host}\-\>%{DATA:reporting_source}\nRule: %{NONNEGINT:rule_number} \(level %{NONNEGINT:severity}\) \-\> '%{DATA:signature}'\n%{GREEDYDATA:remaining_message}"]
  }

  # Attempt to parse additional data from the alert
  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:acct}%{SPACE})?%{GREEDYDATA:real_message}"]
  }

  geoip {
    source => "src_ip"
  }

  mutate {
    convert      => [ "severity", "integer"]
    replace      => [ "@message", "%{real_message}" ]
    replace      => [ "@fields.hostname", "%{reporting_host}"]
    add_field    => [ "@fields.product", "ossec"]
    add_field    => [ "raw_message", "%{message}"]
    add_field    => [ "ossec_server", "%{host}"]
    remove_field => [ "type", "syslog_program", "syslog_timestamp", "reporting_host", "message", "timestamp_seconds", "real_message", "remaining_message", "path", "host", "tags"]
  }
}

output {
   elasticsearch_http {
     host => "127.0.0.1"
   }
}