--- title: "A Logstash grok filter for RFC 5424 syslog" date: "2022-01-18" path: "/212-a-logstash-grok-filter-for-rfc-5424-syslog.html" section: "/ramblings" description: "A Logstash filter that parses RFC 5424 syslog from a Cloud Foundry log drain into useful fields, with the router access logs and JSON application logs split out." --- 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 ```text <14>1 2022-01-18T10:15:02.123456+00:00 my-org.my-space.my-app [APP/PROC/WEB/0] - [tags@47450 app_name="my-app" space_name="my-space"] {"level":"info","msg":"started"} ``` In order, separated by single spaces: | Part | Example | Meaning | |---|---|---| | PRI | `<14>` | facility and severity packed into one number | | VERSION | `1` | always 1 | | TIMESTAMP | ISO 8601 | or `-` if unknown | | HOSTNAME | `my-org.my-space.my-app` | Cloud Foundry puts org, space and app here | | APP-NAME | a GUID | the app's GUID on Cloud Foundry | | PROCID | `[APP/PROC/WEB/0]` | which component: app instance, router (`RTR`), staging and so on | | MSGID | `-` | usually empty | | STRUCTURED-DATA | `[tags@47450 ...]` | key="value" pairs in brackets, or `-` | | MSG | the rest | the log line itself | 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. ```text input { tcp { port => 5514 type => "cf" } } ``` ## The filter ```text 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 `..`. Breaking it up means you can filter Kibana by space (all of staging, say) without a wildcard search: ```text 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. ```text 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: ```bash /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.