logstash config and filter to fully parse a syslog message (PRI, timestamp, host)

logstash.conf

filter {
  # strip the syslog PRI part and create facility and severity fields.
  # the original syslog message is saved in field %{syslog_raw_message}.
  # the extracted PRI is available in the %{syslog_pri} field.
  #
  # You get %{syslog_facility_code} and %{syslog_severity_code} fields.
  # You also get %{syslog_facility} and %{syslog_severity} fields if the
  # use_labels option is set True (the default) on syslog_pri filter.
  grok {
    type => "syslog-relay"
    pattern => [ "<%{POSINT:syslog_pri}>%{SPACE}%{GREEDYDATA:message_remainder}" ]
    add_tag => "got_syslog_pri"
    add_field => [ "syslog_raw_message", "%{@message}" ]
  }
  syslog_pri {
    type => "syslog-relay"
    tags => [ "got_syslog_pri" ]
  }
  mutate {
    type => "syslog-relay"
    tags => [ "got_syslog_pri" ]
    replace => [ "@message", "%{message_remainder}" ]
  }
  mutate {
    # XXX must not be combined with replacement which uses same field
    type => "syslog-relay"
    tags => [ "got_syslog_pri" ]
    remove => [ "message_remainder" ]
  }

  # strip the syslog timestamp and force event timestamp to be the same.
  # the original string is saved in field %{syslog_timestamp}.
  # the original logstash input timestamp is saved in field %{received_at}.
  grok {
    type => "syslog-relay"
    pattern => [ "%{SYSLOGTIMESTAMP:syslog_timestamp}%{SPACE}%{GREEDYDATA:message_remainder}" ]
    add_tag => "got_syslog_timestamp"
    add_field => [ "received_at", "%{@timestamp}" ]
  }
  mutate {
    type => "syslog-relay"
    tags => [ "got_syslog_timestamp" ]
    replace => [ "@message", "%{message_remainder}" ]
  }
  mutate {
    # XXX must not be combined with replacement which uses same field
    type => "syslog-relay"
    tags => [ "got_syslog_timestamp" ]
    remove => [ "message_remainder" ]
  }
  date {
    type => "syslog-relay"
    tags => [ "got_syslog_timestamp" ]
    # season to taste for your own syslog format(s)
    syslog_timestamp => [ "MMM  d HH:mm:ss", "MMM dd HH:mm:ss", "ISO8601" ]
  }

  # strip the host field from the syslog line.
  # the extracted host field becomes the logstash %{@source_host} metadata
  # and is also available in the filed %{syslog_hostname}.
  # the original logstash source_host is saved in field %{logstash_source}.
  grok {
    type => "syslog-relay"
    pattern => [ "%{SYSLOGHOST:syslog_hostname}%{SPACE}%{GREEDYDATA:message_remainder}" ]
    add_tag => "got_syslog_host"
    add_field => [ "logstash_source", "%{@source_host}" ]
  }
  mutate {
    type => "syslog-relay"
    tags => [ "got_syslog_host" ]
    replace => [ "@source_host", "%{syslog_hostname}" ]
    replace => [ "@message", "%{message_remainder}" ]
  }
  mutate {
    # XXX must not be combined with replacement which uses same field
    type => "syslog-relay"
    tags => [ "got_syslog_host" ]
    remove => [ "message_remainder" ]
  }
}

syslog_pri.rb

require "logstash/filters/base"
require "logstash/namespace"

# Filter plugin for logstash to parse the PRI field from the front
# of a Syslog (RFC3164) message.
#
# This filter is based on the original syslog.rb code shipped
# with logstash.
#
class LogStash::Filters::SyslogPRI < LogStash::Filters::Base
  config_name "syslog_pri"

  # set the status to experimental/beta/stable
  plugin_status "experimental"

  # Add human-readable names after parsing severity and facility from PRI
  config :use_labels, :validate => :boolean, :default => true

  # Name of field which passes in the extracted PRI part of the syslog message
  config :syslog_pri_field_name, :validate => :string, :default => "syslog_pri"

  # Labels for facility levels. This comes from RFC3164.
  config :facility_labels, :validate => :array, :default => [
    "kernel",
    "user-level",
    "mail",
    "daemon",
    "security/authorization",
    "syslogd",
    "line printer",
    "network news",
    "uucp",
    "clock",
    "security/authorization",
    "ftp",
    "ntp",
    "log audit",
    "log alert",
    "clock",
    "local0",
    "local1",
    "local2",
    "local3",
    "local4",
    "local5",
    "local6",
    "local7"
  ]

  # Labels for severity levels. This comes from RFC3164.
  config :severity_labels, :validate => :array, :default => [
    "emergency",
    "alert",
    "critical",
    "error",
    "warning",
    "notice",
    "informational",
    "debug"
  ]

  public
  def register
    # Nothing
  end # def register

  public
  def filter(event)
    return unless filter?(event)
    parse_pri(event)
    filter_matched(event)
  end # def filter

  private
  def parse_pri(event)
    # Per RFC3164, priority = (facility * 8) + severity
    #                       = (facility << 3) & (severity)
    priority = event.fields[@syslog_pri_field_name].first.to_i rescue 13
    severity = priority & 7   # 7 is 111 (3 bits)
    facility = priority >> 3
    event.fields["syslog_severity_code"] = severity
    event.fields["syslog_facility_code"] = facility

    # Add human-readable names after parsing severity and facility from PRI
    if @use_labels
      facility_number = event.fields["syslog_facility_code"]
      severity_number = event.fields["syslog_severity_code"]

      if @facility_labels[facility_number]
        event.fields["syslog_facility"] = @facility_labels[facility_number]
      end

      if @severity_labels[severity_number]
        event.fields["syslog_severity"] = @severity_labels[severity_number]
      end
    end
  end # def parse_pri
end # class LogStash::Filters::SyslogPRI