20-fail2ban.conf

filter {

  if [fields][beats_output] == "fail2ban" {
    grok {
      patterns_dir  => "/etc/logstash/patterns.d"
      match         => { "message" => "%{FAIL2BAN}" }
      overwrite     => ["message"]
    }
    date {
      match           => [ "[fail2ban][timestamp]", "ISO8601" ]
      timezone        => "Europe/Berlin"
      locale          => "de"
      target          => "timestamp"
      remove_field    => ["[fail2ban][timestamp]"]
    }

    if [fail2ban][source][ip] {
      geoip {
        source => "[fail2ban][source][ip]"
        target => "[fail2ban][geoip]"
      }
      mutate {
        add_field => {"[fail2ban][source][hostname]" => "%{[fail2ban][source][ip]}"}
      }
      dns {
        reverse => [ "[fail2ban][source][hostname]" ]
        action => "replace"
        hit_cache_size => 8000
        hit_cache_ttl => 300
        failed_cache_size => 1000
        failed_cache_ttl => 300
      }
      mutate {
        update    => { "message" => "[%{[fail2ban][service]}] %{[fail2ban][action]} %{[fail2ban][source][ip]} (%{[fail2ban][source][hostname]})" }
      }

    }
    mutate {
      convert => [
          "[fail2ban][pid]", "integer"
      ]
    }
  }
}