Master Logstash: Essential Commands and Top Log Collection Plugins

Logstash Basic Operations and Common Log‑Collection Plugins

Running a Minimal Logstash Pipeline

Logstash pipelines require at least input and output sections; filter is optional. The input plugin reads data, filters can transform it, and the output plugin writes it to a destination.

Use the

-e

option to specify a pipeline directly on the command line:

<code>logstash -e 'input { stdin {} } output { stdout { codec => rubydebug } }'</code>

Or place the same configuration in a file (e.g.,

test.conf

) and run:

<code>input {
  stdin {}
}
output {
  stdout { codec => rubydebug }
}</code>

Execute with

logstash -f test.conf

. The

-e

or

-f

options bypass

pipelines.yml

and generate a warning.

Logstash Data Types

Array

Boolean – true or false (e.g.,

ssl_enable => true

)

Bytes – supports SI (k, M, G…) and binary (KiB, MiB…) units (e.g.,

my_bytes => "10MiB"

)

Codec – e.g.,

codec => "json"

Hash – key/value pairs (e.g.,

match => { "field1" => "value1" }

)

Number – integer or float (e.g.,

port => 33

)

Password – stored as a plain string (e.g.,

my_password => "password"

)

URI – e.g.,

my_uri => "http://foo:[email protected]"

Path – file system path (e.g.,

my_path => "/tmp/logstash"

)

Escape Sequence – enable with

config.support_escapes: true

in

logstash.yml

Conditional Statements

Conditions work like programming language if/else statements and can be nested. Syntax:

<code>if EXPRESSION { ... } else if EXPRESSION { ... } else { ... }</code>

Supported comparison operators:

==

,

!=

,

<

,

>

,

<=

,

>=

. Regular‑expression operators:

=~

,

!~

. Inclusion operators:

in

,

not in

. Boolean operators:

and

,

or

,

nand

,

xor

. Unary operator:

!

.

Glob Pattern Support

Logstash accepts standard glob patterns such as

*

(any characters),

**

(recursive directories),

?

(single character),

[set]

(character set), and

{p,q}

(alternatives).

Grok Filter Plugin

Grok parses unstructured log lines into structured fields. A pattern follows the syntax

%{PATTERN_NAME:field_name[:data_type]}

. Example for a typical web request line:

<code>grok {
  match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
}</code>

Resulting fields include

client

,

method

,

request

,

bytes

, and

duration

. Custom patterns can be defined in external files or using Oniguruma named captures.

Mutate Filter Plugin

The mutate filter performs field transformations such as rename, remove, replace, convert, copy, gsub, split, join, lowercase, uppercase, merge, update, coerce, and strip. Example converting several fields to integers:

<code>mutate {
  convert => ["reqTime","integer","statusCode","integer","bytes","integer"]
  convert => { "port" => "integer" }
}</code>

Execution order (when the corresponding option is present): rename → update → replace → convert → gsub → uppercase → lowercase → strip → remove → split → join → merge.

Date Filter Plugin

The date plugin parses a field containing a timestamp and sets the Logstash

@timestamp

. Key options:

locale – language tag for month/day names (e.g.,

en

)

match – array of

[field, pattern]

pairs (e.g.,

match => ["createtime", "yyyyMMdd", "yyyy-MM-dd"]

)

target – field to store the parsed timestamp (default

@timestamp

)

timezone – canonical time‑zone ID; final stored time is always UTC

GeoIP Filter Plugin

GeoIP enriches events with geographic information based on an IP address. Required option source specifies the field containing the IP. Optional settings include database (path to MaxMind DB), fields (list of desired attributes), and default_database_type ("city" or "ASN").

Multiline Codec Plugin

Combines multiple lines from a file into a single event (e.g., Java stack traces). Important options:

negate – boolean; when true the pattern result is inverted

pattern – regular expression to match lines

what – "previous" or "next" to indicate which event the matching line belongs to

pattern_dir – array of files containing additional patterns

Example:

<code>codec => multiline {
  pattern => "^\\["
  negate => true
  what => "previous"
}</code>

Collecting NGINX Logs with Logstash

<code>input {
  file {
    path => "/var/log/nginx/*.log"
    start_position => "beginning"
  }
}
filter {
  if [path] =~ "access" {
    mutate { replace => { "type" => "nginx_access" } }
    grok { match => { "message" => "%{COMBINEDAPACHELOG}" } }
    date { match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] }
  } else if [path] =~ "error" {
    mutate { replace => { "type" => "nginx_error" } }
    grok { match => { "message" => "(?<datetime>\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2})\[(?<errtype>\w+)\] \S+: \*\d+ (?<errmsg>[^,]+),(?<errinfo>.*$)" } }
    date { match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] }
  }
}
output {
  if [type] =~ "access" {
    elasticsearch { hosts => ["192.168.179.134:9200"] index => "nginx_access-%{+YYYY.MM.dd}" }
  } else if [type] =~ "error" {
    elasticsearch { hosts => ["192.168.179.134:9200"] index => "nginx_error-%{+YYYY.MM.dd}" }
  }
}</code>

Using rsyslog with Logstash

Configure rsyslog to forward all logs:

<code>*.* @@192.168.179.134:5514</code>

Logstash input:

<code>input {
  syslog { host => "192.168.179.134" port => 5514 }
}
output { stdout { codec => rubydebug } }</code>

Redis as a Message Queue for Logstash

Redis is a recommended broker. Producer writes to a list, consumer reads from it.

Consumer configuration (e.g.,

redis-consumer.conf

):

<code>input {
  redis {
    data_type => "list"
    key => "redis_logstash"
    host => "192.168.179.134"
    port => 6379
    db => 1
  }
}
output {
  elasticsearch { hosts => "192.168.179.134" index => "logstash_redis-%{+YYYY.MM.dd}" }
  stdout { codec => rubydebug }
}</code>

Producer configuration (e.g.,

redis-producer.conf

):

<code>input { stdin { } }
output {
  redis {
    host => "192.168.179.134"
    data_type => "list"
    db => 1
    port => 6379
    key => "logstash_redis"
  }
}</code>

Collecting Docker Logs with Logspout and ELK

Install Docker, then pull the Logspout image:

<code>docker pull gliderlabs/logspout</code>

Logstash configuration for Logspout (e.g.,

logspout.conf

):

<code>input {
  tcp { port => 5140 }
  udp { port => 5140 }
}
output {
  stdout { codec => rubydebug }
  elasticsearch { hosts => "192.168.179.134" index => "logspout" }
}</code>

Run Logspout container to forward logs to Logstash:

<code>docker run --name="logspout" \
  --volume=/var/run/docker.sock:/var/run/docker.sock \
  -e ROUTE_URIS=logstash://192.168.179.134:5140 \
  gliderlabs/logspout</code>

Logspout can filter containers, ignore specific ones via the

LOGSPOUT=ignore

env variable, or include only containers matching a pattern using URI query parameters.