Logstash syslog parser

logstash-syslog-filter.conf

# syslog/relp

grok {
    match => { "@message" => "(?:%{INT:syslog6587_msglen} )?<%{POSINT:syslog_pri}>(?:%{NONNEGINT:syslog5424_ver} )?(?:%{SYSLOGTIMESTAMP:syslog_timestamp}|%{TIMESTAMP_ISO8601:syslog_timestamp}) %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?(:)? %{GREEDYDATA:syslog_message}" }
    add_field => [ "received_at", "%{@timestamp}" ]
    add_field => [ "received_from", "%{host}" ]
    add_tag => [ "syslog_standard" ]
    tag_on_failure => ["_grokparsefailure-syslog_standard"]
}

if !("_grokparsefailure-syslog_standard" in [tags]) {
    syslog_pri { }

    date {
        match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss", "ISO8601" ]
        timezone => "UTC"
    }

    # hostname: handle syslog configurations where hostname is localhost
    if ([syslog_hostname] == "localhost" ) {
        grok {
            match => { "received_from" => "%{IPORHOST:syslog_hostname}(?::%{POSINT:syslog_port})?" }
            overwrite => [ "syslog_hostname", "syslog_port" ]
            tag_on_failure => [ "_grokparsefailure-syslog_standard-hostname"]
        }
    }

    mutate {
        replace => [ "@source.host", "%{syslog_hostname}" ]
    }

    mutate {
        convert => [ "syslog5424_ver", "integer" ]
        convert => [ "syslog6587_msglen", "integer" ]
        remove_field => [
            #"syslog_pri",
            "syslog_hostname",
            "syslog_port",
            "syslog_timestamp"
        ]
    }

    if [syslog5424_ver] == 1 {
        grok {
            # I don't think this is rfc5424-legal because it says SD *must* exist and message *may* exist.
            # However, this makes parsing compatible with common syslog implementations.
            match => [ "syslog_message", "(?:%{DATA:syslog_procid}|\-) (?:%{DATA:syslog_msgid}|\-)(?: %{SYSLOG5424SD:syslog_sd}| \-)? %{GREEDYDATA:syslog_message}" ]
            overwrite => [
                "syslog_message"
            ]
            tag_on_failure => [ "_grokparsefailure-syslog_standard-5424" ]
        }

        # structured-data
        if [syslog_sd] {
            grok {
                match => [ "syslog_sd", "\[%{DATA:syslog_sd_id} (?<syslog_sd_params_raw]>[^\]]+)\]" ]
                remove_field => [
                    "syslog_sd"
                ]
                tag_on_failure => [ "_grokparsefailure-syslog_standard-5424/sds" ]
            }

            if !("_grokparsefailure-syslog_standard-5424/sd" in [tags]) {
                # convert the the key-value pairs
                kv {
                    source => "syslog_sd_params_raw"
                    target => "syslog_sd_params"
                    remove_field => [
                        "syslog_sd_params_raw"
                    ]
                }

                if [syslog_sd_params][type] {
                    # establish a convention that a structured data key of "type" will be the log type
                    mutate {
                        replace => { "@type" => "%{syslog_sd_params[type]}" }
                        remove_field => [ "syslog_sd_params[type]" ]
                    }
                }
            }
        }
    }
}