logstash kippo filter

filter {
    if [type] == "kippo" {
        mutate {
            add_tag => [ "ssh" ]
        }

        grok {
            match => [ "message", "%{TIMESTAMP_ISO8601:logdate} \[%{DATA:service}\] %{DATA:message}$"]
            overwrite => [ "message" ]
        }

        date {
            match => [ "logdate" , "YYYY-MM-dd HH:mm:ssZ" ]
        }

        if [service] != "-" and [service] =~ /[^,]+,[^,]+,[^,]+/ {
            grok {
                match => [ "service", "%{DATA:service},%{POSINT:src_port},%{IP:src_ip}"]
                overwrite => [ "service" ]
            }

            if [message] =~ /login attempt \[.*\] (failed|succeeded)/ {
                mutate {
                    add_tag => [ "credentials" ]
                }
            }
        } else {
            mutate {
                drop { }
            }
        }
    }

    if [src_ip] {
        geoip {
            source => src_ip
        }

        geoip {
            source => src_ip
            database => "/opt/logstash/vendor/geoip/GeoIPASNum.dat"
        }
    }
}