Logstash Notes

Debugging

This topic is very open-ended, but note the existence of a command-line option to launch Logstash, force it to read its configuration and report whether any of it was bad:

# logstash -f logstash.yml --config.test_and_exit

Other debugging will be covered in this document as relevant and timely.

Inputs

Input can come from many places, but in particular, from stdin and from the more productive and expected origin of an Elastic Beat:

input
{
  stdin { }
  beats
  {
    port => "5044"
  }
}

Outputs

The most typical output is Elasticsearch, configured as a list of hosts running this application. It's possible, however, to output to stdout and make that output useful to Ruby for debugging.

output
{
  elasticsearch { hosts => [ "localhost:9200", "remotehost:9200" ] }
  stdout { codec => rubydebug }
}

Filter grok (and general filter notes)

grok is the oldest filtering. It's based on regular expressions and is correspondingly expensive, slow and as well as suffers from difficulty in maintenance just as any solutions involving regular expressions.

I try not to use grok because of expense and because, for the things that I've done, dissect works better. There are "libraries" of grok (regular) expressions available, just Google for them.

If we were parsing /var/log/syslog, for instance, we might have the following:

filter
{
  grok
  {
    match =>
    {
      "message" => ...
    }
  }
}

This instructs Logstash to match (what's not expressed here except by "...") and assign the data it matches to new field (to be created) in the data named message. For the format of this regular expression, we could use:

"%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}"

1. SYSLOGTIMESTAMP: look for a timestamp in the syslog format, assign it to (will show up in Kibana as) field syslog_timestamp
2. SYSLOGHOST: look for a hostname in the syslog format, assign it to field syslog_hostname
3. DATA: look for program data in the syslog format, assign it to field syslog_program
4. POSINT: look for the process identity, if present (because optional), assign it to field syslog_pid
5. GREEDYDATA: the "greedy" data pattern takes everything else on the log line and assigns it to syslog_message field (basically, just the text part of the message)

This also puts the entire syslog message line into field message.

Filter dissect

Extracts structured fields from (more or less) unstructured event data, but without grok, which uses regular expressions. In Logstash processing, the slowest component will slow down the entire pipeline.

First, what data does the filter operate on?

The operation is mapping because you're mapping constituent parts of an existing field, let's say message in a syslog entry, to new fields (or tags). Here's the syntax:

filter
{
  dissect
  {
    mapping =>
    {
      "message" => ...
    }

    remove_field => [ "message" ]
  }
}

This means that this filter (in the Logstash pipeline of tasks to accomplish) will intercept each entry that contains a field, message, and treat it according to what's written. It won't touch any other field. When it has done this, it will also simply remove field message too. Now, you can rig any number of existing fields to be dissected into constituent fields if you want—even those you create new in your filter:

filter
{
  dissect
  {
    mapping =>
    {
      "message" => "%{field1} %{field2} %{description}"
      "description" => ...
    }

    remove_field => [ "message" ]
  }
}

After the above, message is discarded as unwanted, but description, which was create from message in the first place, persists even though it too is subjected to dissection.

But, we're writing this to discuss what kinds of operations we can run on message to produce other, new fields. Read on...

Delimiters

dissect is like a split operation, yet more succinct and powerful. Imagine in Java sort of being able to write:

void split( String string )
{
  String[] values;
  values  = string.split( "<", string );
  values += string.split( ">", string );
  values += string.split( " ", string );
  .
  .
  .

...except that these splits are to happen in just the right places. And as you go. And in a self-documenting way. Etc. You'll understand immediately when you read the examples here. Image a log entry that looks like this:

<9> Nov 14 08:09:23,169 127.0.0.1 WARN Core: An irregular situation has occurred.

dissect pattern-writing asks you to specify the patterns of delimiters between the fields of interest, queues up the delimiters and search the source text once from left to right, finding each successive delimiter in the queue. The format of the patterns of fields and delimiters is similar to grok as follows.

Yellow indicates delimiters; green indicates fields:

"<%{priority}> %{syslog_timestamp} %{+syslog_timestamp} %{+syslog_timestamp} %{severity} %{component}: %{message_text}"

What this means:

The delimiter exists physically in data you're parsing. For instance, above, it expects that there will be some value, priority, between opening- and closing angle brackets immediately followed upon by a syslog timestamp value (not complete) and additional space-separated syslog timestamp values all of which are to be concatenated together into a single syslog_timestamp field. Then, separated from the preceding by a space, is a value to call severity, then another space, and, finally, something called rest_url.

The delimiters on which the message is split are these (in order):

1. <       (left angle bracket)
2. >⎵     (right angle plus space)
3. ⎵       (space)
4. ⎵       (space)
5. ⎵       (space)
6. ⎵       (space)
7. :⎵     (colon plus space)

The new fields that will be created are:

1. priority
2. syslog_timestamp
3. severity
4. component
5. message_text

...in addition to what might have been the original field (perhaps message).

Field types

Also, above, and used in the explanation, are field types:

1. normal field type, simply adds the value found to the even using the matched text/data as the value and the pronoun or specifier as the name of the field. (The pronoun or specifier is really called key by Elastic Logstash documentation.) For example, <9> is matched by expression <%{priority}> and, in Kibana, it yields priority: 9 or, in the Table tab:

   t priority  ...  9
   

2. skip field type, simple skips any value—and makes no field %{} or %{?priority}. For example, from these three pieces of data:

   9 14-Nov-2018 08:09:23,169
   

   ...no priority field will be created, though it's named for self-documentary purposes, in the output from this pattern:

   <?priority> %{timestamp} %{+timestamp}
   

   Only the (entire) timestamp will be retained (in field timestamp).
3. append field type, allows you to rebuild values split by any delimiter you happen to be using. There was an example of this just above when the plus sign (+) was used with timestamp to gather both 14-Nov-2018 and 08:09:23,169 into a single timestamp value.
4. suffix field type, is the variation of those field types above to which are added numeric suffixes for ordering purposes. For example, let's imagine this rule pattern:

   <?priority> %{timestamp/2}-%{+timestamp/3}-%{+timestamp/1}
   

   ...which would produce (from the data we used already above):

   timestamp: 2018-Nov-14
   

   Why are there spaces in the result here? This is because (as Elastic Logstash documentation says), the delimiter found before the field is appended with the value.

It's very useful to control the application of dissect using conditionals to reduce useless or confusing overhead. Below, unless the in-coming log entry is of type syslog, Logstash won't waste its time:

filter
{
  if [type] == "syslog" or "syslog" in [tags]
  {
    dissect
    {
      mapping =>
      {
        "message" => "%{ts} %{+ts} %{+ts} %{src} %{} %{prog}[%{pid}]: %{msg}"
      }
    }
  }
}

Filter convert

It's possible to rig the filter to convert some field from whatever they are (usually strings) to something else:

filter
{
  dissect
  {
    mapping =>
    {
      "message" => "%{time} %{+time} %{level} %{pid} %{thread} %{loggername} %{function} %{line} %{msg}"
    }
    convert_datatype =>
    {
      line   => "int"
      pid    => "int"
      thread => "int"
    }
  }
}

Appendix 1

Here are some helps on Filebeat and Logstash configuration.

Dockerfile:

This is for the container Filebeat is running on. The certificates are offered by Sébastien Pujadas. Here are some useful links to sebp/elk, the best Docker container for ELK:

filebeat.yml:

Filebeat will deliver its host's messages, syslog and secure log files to Logstash on host elk-host.

filebeat.modules:
  - module: system

filebeat.inputs:
  - type: log
  enabled: true
  paths:
    - /var/log/messages
    - /var/log/secure
    - /var/log/syslog
  - type: docker
    containers.ids:    '*'
    containers.path:   '/var/lib/docker/containers'
    containers.stream: 'stdout'
    include_lines:     [ '^ERR', '^WARN' ]

output.logstash:
  hosts:           [ "elk-host:5044" ]
  ssl.enabled:     true
  ssl.certificate: "/etc/pki/tls/certs/logstash-beats.crt"
  ssl.key:         "/etc/pki/tls/private/logstash-beats.key"

Dockerfile:

This is for the ELK stack container. It creates a new container image from Sébastien Pujadas' ELK container and adds one filter to the Logstash pipeline configuration.

Dockerfile:

FROM sebp/elk:642
COPY 07-debug.conf /etc/logstash/conf.d/
COPY 08-cef.conf  /etc/logstash/conf.d/

08-cef.conf:

This filter is an example of fixing up Common Event Format (CEF) log entries^* to give more information to a Kibana user. It sort of completes these notes by showing a real-world implementation. It also should inform you that the tightness of your logging format can have a very big effect on what you can get Logstash to present to Elasticsearch (and thence to Kibana). The utility of your log files and Kibana to you, to your support staff or even to customers can be dramatically enhanced by paying attention what you put into your log statements (from your application code).

I added some very a-syntactic line-wrapping to make it fit (so, don't do that in your real file):

filter
{
  dissect
  {
    mapping =>
    {
      "message" => "%{acme.date} %{acme.time} CEF:%{acme.version}
                                      |%{acme.device_vendor}
                                      |%{acme.device_product}
                                      |%{acme.device_version}
                                      |%{acme.device_event_class_id}
                                      |%{acme.name}
                                      |%{acme.severity}
                                      |%{acme.extensions}"
    }
  }
  mutate
  {
    convert => { "acme.date" => "string" }
  }
}

docker-compose.yml:

If deploying using Docker Compose, here are the guts for the Filebeat and Logstash containers (well, the whole ELK container in Logstash' case).

version: '3.5'
services:
  .
  .
  .
  elk:
    image: acme-elk
    ports:
      - "5601:5601"
      - "9200:9200"
      - "9300:9300"
      - "5044:5044"
    ulimits:
      nofile:
        soft: 65536
        hard: 65536
    networks:
      acme:
    deploy:
      mode: global
      placement:
        constraints:
        - node.labels.acme.elk==true

  filebeat:
    image: acme-filebeat
    networks:
      acme:
    volumes:
      - "logs:/opt/logs"
    deploy:
      mode: global

networks:
  acme:
    name:     'acme'
    external: true
    driver:   bridge

Appendix 2: on writing a Logstash filter...

Appendix 3: splitting up filters

I split the filter code above out into two, separate pipeline configuration files. It makes it simpler to maintain, still works and probably presents a number of other advantages.

07-debug.conf:

filter
{
  if [source] =~ "debug"
  {
    # send debug.log entries through here:
    dissect
    {
      mapping =>
      {
        "message" => "[%{ignore}] %{acme.date} %{acme.time} - REST: Path: %{acme.restpath}"
      }
      # Now that the new, split-out fields are created, we don't need to keep
      # 'message' any longer:
      remove_field => [ "message" ]
    }
  }
}

08-cef.conf:

filter
{
  if [source] =~ "audit"
  {
    # send audit.log (purely CEF) entries through here:
    # CEF format:
    # 2018-11-28 15:11:09,635 CEF:Version|Device Vendor|Device Product|Device Version\
    #   |Device Event Class ID|Name|Severity|Extension
    # (line-wrapping below is illegal and won't work)
    dissect
    {
      mapping =>
      {
        "message" => "%{acme.date} %{acme.time} CEF:%{acme.version}|%{acme.device_vendor}\
                  |%{acme.device_product}|%{acme.device_version}|%{acme.device_event_class_id}\
                  |%{acme.name}|%{acme.severity}|%{acme.extensions}"
      }
      # Now that the new, split-out fields are created, we don't need to keep
      # 'message' any longer:
      remove_field => [ "message" ]
    }

    # Parse 'acme.extensions' for its key-value pairs and put the keys out as
    # separate fields (with values). Values parsed out are impoverished beyond
    # the first space they contain without 'field_split_pattern' and
    # 'whitespace' below. The results are prefixed for easy recognition.
    kv
    {
      source              => "acme.extensions"
      field_split_pattern => " (?=[A-Za-z0-9]+=)"
      whitespace          => "strict"
      prefix              => "acme_kv."
    }
  }
}

Appendix 4: debugging filters

Debugging filter health at start-up

When I start my containers (in particular, my ELK and Filebeat containers), I like to get ELK's id, then get into it:

$ docker stack deploy -c docker-compose.yaml acme
$ docker ps --format '{{.ID}}\t{{.Image}}\t{{.Status}}' -a
(Let's say the ELK container's id is 6597c035a965)
$ docker exec --user root -it  6597c035a965  bash

Then, I checkout these files in /var/log/logstash:

root@6597c035a965:/# cd /var/log/logstash
root@6597c035a965:/var/log/logstash# ll
total 19036
drwxr-xr-x 1 logstash logstash     4096 Dec 15 20:29 ./
drwxr-xr-x 1 root     root         4096 Oct  5 19:12 ../
-rw-r--r-- 1 logstash logstash 18899721 Dec 15 20:29 debug.log
-rw-r--r-- 1 root     root            0 Dec 15 20:28 logstash.err
-rw-r--r-- 1 logstash logstash   286689 Dec 15 20:29 logstash-plain.log
-rw-r--r-- 1 logstash logstash        0 Dec 15 20:28 logstash-slowlog-plain.log
-rw-r--r-- 1 root     root       286778 Dec 15 20:29 logstash.stdout

• debug.log —holds the output from applying filter
• logstash-plain.log —holds the errors, including filter-fatal ones

If I tail logstash-plain.log (immediately or edit it and look near the top), I can see if there's an error that's broken my filter (not shown here).

root@6597c035a965:/var/log/logstash# tail -f logstash-plain.log
[2018-12-15T20:28:48,285][INFO ][logstash.setting.writabledirectory] Creating directory {:setting=>"path.queue",...
[2018-12-15T20:28:48,293][INFO ][logstash.setting.writabledirectory] Creating directory {:setting=>"path.dead_le...
[2018-12-15T20:28:48,669][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modul...
[2018-12-15T20:28:48,702][INFO ][logstash.agent           ] No persistent UUID file found. Generating new UUID {:uu...
[2018-12-15T20:28:49,377][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.4.2"}
.
.
.

When lines begin streaming into this file, you can be sure that something is working. Turn to your browser to check out Kibana.

Debugging filter workings

Create a pipeline-configuration file thus. This will drop a file, debug.log, on the path, /var/log/lostash that will contain output in the Ruby-debug format.

output
{
  file
  {
    path  => "/var/log/logstash/debug.log"
    codec => "rubydebug"
  }
}

The output debug file will contain stuff like this. Caution: it will get very big—on the order of one or more of these structures per log entry processed. Note that field tags contains "_dissectfailure".

{
        "source" => "/opt/acme-2/logs/debug.log",
      "@version" => "1",
         "input" => {
        "type" => "log"
    },
        "offset" => 2762812,
          "host" => {
        "name" => "93dfeb0f3845"
     },
    "prospector" => {
        "type" => "log"
     },
          "beat" => {
            "name" => "93dfeb0f3845",
         "version" => "6.4.2",
        "hostname" => "93dfeb0f3845"
     },
    "@timestamp" => 2018-12-07T17:45:56.620Z,
          "tags" => [
        [0] "beats_input_codec_plain_applied",
        [1] "_dissectfailure"
     ],
       "message" => "[http-bio-8181-exec-105] 2018-11-28 17:14:03,970 - REST: Request length: 72"
}

When you look at message, it's clear that our "debug.log" filter above will fail, hence the existence of "_dissectfailure" in tags. As there is no coverage for this, it doesn't concern me except to point out the need to handle such failure for those log entries I anticipate handling (and this is one of them).

Appendix 5: more filters

Yeah, we'll get this right pretty soon.