Log Transformation & Enrichment

Advanced Usage

The previous example doesn't show how to parse the correct timestamp or how to ship multi-line logs. These are important issues that can easily be addressed.

To get started, let's assume our logs contain a timestamp (UTC), severity and message. If your logs have different structure you'll need to make small adjustments to the configs below.

2015-10-03 12:01:58,345 ERROR Processing request failed.

Multi-line messages require a slight changed input section.

input {
  file {
    path => "/var/log/test.log"
    start_position => "beginning"
    codec => multiline {
      pattern => "^%{TIMESTAMP_ISO8601}"
      negate => true
      what => "previous"
    }
  }
}
The pattern looks for log lines starting with a timestamp and, until a new match is found, all lines are considered part of the event. This is done by setting the negate parameter to true.
Parsing logs means adding a new section to the config file:
filter {
  grok {
    match => [ "message", "(?m)%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:severity} %{GREEDYDATA:message}" ]
    overwrite => [ "message" ]
  }
  date {
    match => [ "timestamp" , "yyyy-MM-dd HH:mm:ss,SSS" ]
  }
}

The grok filter splits the log content into 3 variable. The (?m) in the beginning of the regexp is used for multi-line matching. Otherwise, only the first line would be read. The patterns used in the regexp used in the regexp are provided with LogStash and should be used when possible to simplify regexps. By default, the timestamp of the log line is considered the moment when the log line is read from the file. the date filter changes the timestamp to the one matched earlier by the grok filter.