Logstash Ingress Filter - Apache (for Filebeat Apache module)

logstash-filebeat_apache_filter.conf

# Based on https://github.com/elastic/beats/blob/7.2/filebeat/module/apache/error/ingest/pipeline.json
# Documentation at https://www.elastic.co/guide/en/logstash/current/logstash-config-for-filebeat-modules.html#parsing-apache2 is out of date

filter {
  if [event][module] == "apache" {
    if [fileset][name] == "access" {
      grok {
        match => { "message" => ["%{IPORHOST:[source][address]} - %{DATA:[user][name]} \[%{HTTPDATE:[apache][access][time]}\] \"(?:%{WORD:[http][request][method]} %{DATA:[url][original]} HTTP/%{NUMBER:[http][version]}|-)?\" %{NUMBER:[http][response][status_code]} (?:%{NUMBER:[http][response][body][bytes]}|-)( \"%{DATA:[http][request][referrer]}\")?( \"%{DATA:[user_agent][original]}\")?",
        "%{IPORHOST:[source][address]} - %{DATA:[user][name]} \\[%{HTTPDATE:[apache][access][time]}\\] \"-\" %{NUMBER:[http][response][status_code]} -",
        "\\[%{HTTPDATE:[apache][access][time]}\\] %{IPORHOST:[source][address]} %{DATA:[apache][access][ssl][protocol]} %{DATA:[apache][access][ssl][cipher]} \"%{WORD:[http][request][method} %{DATA:[url][original]} HTTP/%{NUMBER:[http][version]}\" %{NUMBER:[http][response][body][bytes]}"
        ] }
        remove_field => "message"
      }
      grok {
        match => { "[source][address]" => ["^(%{IP:[source][ip]}|%{HOSTNAME:[source][domain]})$"]}
      }
      mutate {
        rename => { "[event][created]" => "@timestamp" }
      }
      date {
        match => [ "[apache][access][time]", "dd/MMM/yyyy:H:m:s Z" ]
        remove_field => "[apache][access][time]"
      }
      useragent {
        source => "[user_agent][original]"
      }
      geoip {
        source => "[source][ip]"
        target => "[source][geo]"
      }
    }
    else if [fileset][name] == "error" {
      grok {
        match => {"message" => ["\[%{APACHE_TIME:[apache][error][timestamp]}\] \[%{LOGLEVEL:[log][level]}\] ( \[client %{IPORHOST:[source][address]}(:%{POSINT:[source][port]})?\])? %{GREEDYDATA:[message]}",
            "\[%{APACHE_TIME:[apache][error][timestamp]}\] \[%{DATA:[apache][error][module]}:%{LOGLEVEL:[log][level]}\] \[pid %{NUMBER:[process][pid]}(:tid %{NUMBER:[process][thread][id]})?\] ( \[client %{IPORHOST:[source][address]}(:%{POSINT:[source][port]})?\])? %{GREEDYDATA:[message]}"
        ]}
        pattern_definitions => {
          "APACHE_TIME" => "%{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{YEAR}"
        }
        remove_field => "message"
      }
      date {
        match => [ "[apache][error][timestamp]", "EEE MMM dd H:m:s YYYY", "EEE MMM dd H:m:s.SSSSSS YYYY" ]
        remove_field => "[apache][error][timestamp]"
      }
      grok {
        match => { "[source][address]" => ["^(%{IP:[source][ip]}|%{HOSTNAME:[source][domain]})$"]}
      }
      geoip {
        source => "[source][ip]"
        target => "[source][geo]"
      }
    }
  }
}