25-ufw.conf

filter {
  if [fields][beats_output] == "ufw" {
    grok {
      patterns_dir  => "/etc/logstash/patterns.d"
      match         => { "message" => "%{IPTABLES}" }
      tag_on_failure => [ "_grok_ufw_nomatch" ]
      add_tag        => [ "_grok_ufw_success" ]
    }
    date {
      match           => [ "[ufw][timestamp]", "MMM dd HH:mm:ss" ]
      target          => "timestamp"
      remove_field    => ["[ufw][timestamp]"]
    }
    if [ufw][source][ip] {
      geoip {
        source => "[ufw][source][ip]"
        target => "[ufw][source][geoip]"
      }

      mutate {
        add_field => {"[ufw][source][hostname]" => "%{[ufw][source][ip]}"}
      }
      dns {
        reverse => [ "[ufw][source][hostname]" ]
        action => "replace"
        hit_cache_size => 8000
        hit_cache_ttl => 300
        failed_cache_size => 1000
        failed_cache_ttl => 300
      }
      mutate {
        update    => { "message" => "[%{[ufw][status]}] IP: %{[ufw][source][ip]} (%{[ufw][source][hostname]}) dst_port: %{[ufw][destination][port]}" }
      }
    }
  }
}