9998-logstash-parse-failures.conf

filter {
    # In case [tags] are being used for something else and or geoip lookup failures, then we do NOT want to just assume [tags] means something we don't want to index.
    # However, we also do NOT want to perform the expensive multi or statement on everything that does not have [tags]
    if [tags] {
        if "_parsefailure" in [tags] or "_jsonparsefailure" in [tags] or "_grokparsefailure" in [tags] or "_dissectfailure" in [tags] or "_groktimeout" in [tags] or "_rubyexception" in [tags] or "_dateparsefailure" in [tags] or "_jdbcstreamingfailure" in [tags] or "_elasticsearch_lookup_failure" in [tags] or "_urldecodefailure" in [tags] or "_csvparsefailure" in [tags] or "_xmlparsefailure" in [tags] {
            # After setting the event as a parse failure, then we need to remove all events except for what was set in 00*-.conf and above
            prune {
                whitelist_names => [ "$@timestamp^", "@timestamp", "$_original_message^", "_original_message", "^tags$", "$%{[@meta][event_type]}^", "${[@meta][log][timestamp]}^", "$[@meta][log][level]^", "$[@meta][log][timestamp]^" ]
            }
            mutate {
                add_field => {
                    "[@meta][event_type]" => "_parsefailure"
                    "[@meta][log][timestamp]" => "%{@timestamp}"
                }
            }
        }
    }
    # Remove the original message that we copied in 101-*.conf if no parse failure
    if [@meta][event_type] != "_parsefailure" {
        mutate { remove_field => "_original_message"}
    }
}