A Logstash grok filter for RFC 5424 syslog

The syslog filter most Logstash tutorials give you, mine included, expects the old BSD format: Jan 5 10:01:02 host program[123]: message. Applications on Cloud Foundry send their logs to a syslog drain in the newer RFC 5424 format instead, and that filter matches none of it. Everything lands in Kibana as one long message field tagged _grokparsefailure.

This is the filter I ended up with on Logstash 7. It is written for Cloud Foundry, but the first half works for any RFC 5424 source.

What an RFC 5424 line looks like

<14>1 2022-01-18T10:15:02.123456+00:00 my-org.my-space.my-app <app-guid> [APP/PROC/WEB/0] - [tags@47450 app_name="my-app" space_name="my-space"] {"level":"info","msg":"started"}

In order, separated by single spaces:

Over TCP the sender may also put the message length in front of the whole thing (RFC 6587 octet counting), so a line can start with something like 187 <14>1 ....

Input

Use a plain tcp input, not the syslog input. The syslog input only understands the old format and will tag every one of these as a failure.

input {
  tcp {
    port => 5514
    type => "cf"
  }
}

The filter

filter {
  grok {
    match => {
      "message" => "^(?:%{NONNEGINT:syslog_length} )?<%{NONNEGINT:syslog_pri}>%{NONNEGINT:syslog_version} +(?:%{TIMESTAMP_ISO8601:syslog_timestamp}|-) +%{NOTSPACE:syslog_host} +%{NOTSPACE:syslog_app} +%{NOTSPACE:syslog_procid} +%{NOTSPACE:syslog_msgid} +(?:-|\[%{NOTSPACE:syslog_sd_id} %{DATA:syslog_sd}\]) ?%{GREEDYDATA:syslog_msg}"
    }
    add_field => { "received_at" => "%{@timestamp}" }
    tag_on_failure => ["_syslog5424_failure"]
  }

  if !("_syslog5424_failure" in [tags]) {
    syslog_pri {
      syslog_pri_field_name => "syslog_pri"
    }

    date {
      match => ["syslog_timestamp", "ISO8601"]
    }

    if [syslog_sd] {
      kv {
        source => "syslog_sd"
        target => "sd"
        field_split => " "
        value_split => "="
        trim_value => '"'
      }
    }

    mutate {
      rename => { "syslog_msg" => "message" }
      remove_field => ["syslog_length", "syslog_sd"]
    }
  }
}

What each part does:

• grok splits the header into fields. NOTSPACE is used for the header parts because RFC 5424 does not allow spaces in them. The structured data is taken as one bracketed block, which is all Cloud Foundry sends; a source that sends several blocks, or escaped ] inside a value, would need a more careful pattern.
• syslog_pri turns 14 into facility user-level and severity informational. It defaults to a field called syslog_pri already; I name it so the link to the grok field is obvious.
• date replaces @timestamp, which is when Logstash received the event, with the time in the log line. Without it, a backlog that arrives late is plotted at the wrong time in Kibana. The original arrival time is kept in received_at.
• kv turns app_name="my-app" space_name="my-space" into sd.app_name and sd.space_name.
• mutate puts the actual log text back in message, so Kibana's default view shows the line and not the whole syslog wrapper.

Logstash has a built-in SYSLOG5424LINE pattern. I wrote my own because the built-in one does not allow for the length prefix, and I wanted control over the field names.

Splitting the Cloud Foundry host name

The HOSTNAME field is <org>.<space>.<app>. Breaking it up means you can filter Kibana by space (all of staging, say) without a wildcard search:

filter {
  if [syslog_host] {
    dissect {
      mapping => { "syslog_host" => "%{[cf][org]}.%{[cf][space]}.%{[cf][app]}" }
      tag_on_failure => ["_cf_host_failure"]
    }
  }
}

dissect is faster than grok for a fixed layout like this. It does assume none of the names contain a dot.

Router logs and application logs

The PROCID tells you which component wrote the line. The two I care about are the router, which writes an access log line per request, and the app itself.

filter {
  if [syslog_procid] =~ /^\[RTR/ {
    mutate { replace => { "type" => "router" } }
  }

  if [syslog_procid] =~ /^\[APP/ and [message] =~ /^\{/ {
    json {
      source => "message"
      target => "app"
      tag_on_failure => ["_app_json_failure"]
    }
  }
}

The router's access line is a combined-log-style request followed by a long run of key:"value" pairs (response time, app instance, x_forwarded_for and so on). Grok the request part with the usual HTTP patterns, then hand the tail to kv with value_split => ":". Convert response_time to a float with mutate, or Kibana cannot graph it.

For application logs, the json filter only runs when the message looks like a JSON object. Apps that log structured JSON get every key as a searchable field under app; plain text lines are left alone instead of collecting a parse-failure tag.

Testing it

Before restarting the service, check the config parses:

/usr/share/logstash/bin/logstash --path.settings /etc/logstash -t

Then paste a sample line into Kibana's Grok Debugger (under Dev Tools) with the pattern, and fix anything that does not match there rather than by restarting Logstash over and over.

Today

Logstash 8 turns on ECS compatibility by default, which renames several fields the inputs and syslog_pri create (host becomes an object, for one). Either set ecs_compatibility => disabled on the pipeline or move your field names to their ECS equivalents.