1. Pipeline Structure

input { ... }
filter { ... }
output { ... }

Pipelines are processed top-to-bottom
Multiple pipelines via pipelines.yml
Each pipeline has its own input → filter → output

2. Input Plugins (Common)

Plugin	Use Case	Example
syslog	CF log-drains, network devices	syslog { port => 5514 }
beats	Filebeat, Metricbeat	beats { port => 5044 }
file	Local log files	file { path => "/var/log/*.log" }
s3	AWS S3 buckets	s3 { bucket => "logs" region => "eu-central-1" }
kafka	Kafka topics	kafka { topics => ["logs"] bootstrap_servers => "..." }
http	Webhook / HTTP input	http { port => 8080 }
stdin	Testing/debugging	stdin {}

3. Filter Plugins (Common)

Plugin	Purpose	Key Options
grok	Parse unstructured text	match, patterns_dir, tag_on_failure
json	Parse JSON strings	source, target
mutate	Rename, remove, add fields	rename, remove_field, add_field, convert, replace, gsub
date	Parse timestamps	match, target, timezone
kv	Key=value parsing	source, field_split, value_split
dissect	Fast structured parsing	mapping (faster than grok for simple patterns)
ruby	Custom Ruby code	code => "..."
drop	Drop events	if [level] == "DEBUG" { drop {} }
aggregate	Aggregate related events	task_id, code, map_action

4. Grok Syntax Quick Reference

Common Patterns

%{WORD}              Single word (no spaces)
%{DATA}              Any data (non-greedy)
%{GREEDYDATA}        Any data (greedy, use at end)
%{INT}               Integer
%{NUMBER}            Number (int or float)
%{IP}                IPv4 address
%{IPV6}              IPv6 address
%{HOSTNAME}          Hostname
%{TIMESTAMP_ISO8601} ISO 8601 timestamp
%{SYSLOG5424PRI}     Syslog priority
%{LOGLEVEL}          Log level (INFO, WARN, ERROR...)
%{JAVACLASS}         Java class name
%{UUID}              UUID
%{URI}               Full URI
%{URIPATH}           URI path only
%{QUOTEDSTRING}      Quoted string

Captures

Named capture

%{PATTERN:fieldname}

→ %{IP:client_ip}

Typed capture

%{NUMBER:duration:float}

Custom pattern

(?<fieldname>regex)

5. Conditionals Syntax

Comparison & Logic

# Equality
if [field] == "value" { ... }
if [field] != "value" { ... }

# Regex
if [field] =~ /regex/ { ... }
if [field] !~ /regex/ { ... }

# Membership
if [field] in ["a", "b", "c"] { ... }

# Existence
if [field] { ... }       # exists & truthy
if ![field] { ... }      # missing or falsy

# Nested fields
if [app][env] == "production" { ... }

Compound & Nesting

# Compound conditions
if [type] == "syslog" and [level] == "ERROR" {
  ...
}
if [status] >= 400 or [duration] > 5000 {
  ...
}
if [type] == "syslog" and !([app][name]) {
  ...
}

# Nested if/else
if [type] == "syslog" {
  if [level] == "ERROR" {
    # ...
  } else if [level] == "WARN" {
    # ...
  }
}

6. Mutate Operations

mutate {
  # Rename fields
  rename => { "old_name" => "new_name" }
  rename => { "src" => "[app][source]" }     # nested

  # Add fields
  add_field => { "env" => "production" }
  add_field => { "[app][version]" => "1.0" }  # nested

  # Remove fields
  remove_field => ["temp_field", "unwanted"]

  # Convert types
  convert => { "duration" => "integer" }
  convert => { "ratio" => "float" }

  # Replace value
  replace => { "message" => "%{source}: %{message}" }

  # Lowercase / Uppercase
  lowercase => ["level"]
  uppercase => ["env"]

  # String substitution (gsub)
  gsub => [
    "field_name", "/", "_",
    "message", "\r\n", "\n"
  ]

  # Strip whitespace
  strip => ["message"]

  # Split string to array
  split => { "tags" => "," }

  # Merge arrays
  merge => { "dest_array" => "source_array" }

  # Copy field
  copy => { "source" => "destination" }

  # Tags
  add_tag => ["processed", "cf-app"]
  remove_tag => ["_grokparsefailure"]
}

⚠️ Execution Order within a single mutate block:

rename → update → replace → convert → gsub → uppercase → lowercase → strip → remove → split → join → merge → copy

To guarantee order, use separate mutate blocks.

7. Common Patterns / Recipes

Parse Spring Boot JSON logs

if [message] =~ /^\{/ {
  json { source => "message" target => "log" }
  date { match => ["[log][timestamp]", "ISO8601"] }
}

Extract CF syslog metadata

grok {
  match => { "message" => "%{SYSLOG5424PRI}%{NONNEGINT:ver} +%{TIMESTAMP_ISO8601:ts} +%{DATA:host} +%{DATA:app} +%{DATA:proc} +%{DATA:msgid} +(?:%{SYSLOG5424SD:sd}|-) +%{GREEDYDATA:msg}" }
}

Merge metadata into prefix doc

mutate {
  rename => {
    "org_name"   => "[app][org]"
    "space_name" => "[app][space]"
    "app_name"   => "[app][name]"
  }
}

Drop debug logs in production

if [app][env] == "production" and [log][level] == "DEBUG" {
  drop {}
}

Multiline codec (stack traces)

input {
  beats {
    port => 5044
    codec => multiline {
      pattern => "^%{TIMESTAMP_ISO8601}"
      negate => true
      what => "previous"
    }
  }
}

GeoIP enrichment

geoip {
  source => "client_ip"
  target => "geo"
}

Throttle / rate limit

throttle {
  before_count => -1
  after_count => 5
  period => 60
  key => "%{client_ip}"
  add_tag => "throttled"
}

8. Change Management Best Practices

Testing Changes

# Validate config syntax
bin/logstash --config.test_and_exit -f pipeline.conf

# Test with stdin/stdout (dry run)
bin/logstash -f pipeline.conf --path.data /tmp/ls-test \
  -e 'input { stdin {} } output { stdout { codec => rubydebug } }'

# File input with sample log
input {
  file {
    path => "/tmp/test-logs.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}
output { stdout { codec => rubydebug } }

Rolling Out Changes

1. Config reload without restart

Enable config.reload.automatic: true in logstash.yml or start with --config.reload.automatic

Logstash watches for file changes, reloads pipeline
Zero downtime for filter/output changes
Input changes may require restart

2. Blue-green pipelines

# pipelines.yml
- pipeline.id: main-v1
  path.config: "/etc/logstash/conf.d/v1/*.conf"
- pipeline.id: main-v2
  path.config: "/etc/logstash/conf.d/v2/*.conf"

Route traffic to v2, then remove v1.

3. Dead letter queue (DLQ)

dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 1gb

Process DLQ with a separate pipeline.

4. Canary testing

if [app][name] == "canary-app" {
  # new parsing logic
} else {
  # existing logic
}

5. Version your configs

Treat pipeline configs like code. Git, PR reviews, CI validation.

9. Debugging Tips

# Enable debug logging
bin/logstash --log.level debug

# Check pipeline stats
curl -s localhost:9600/_node/stats/pipelines | jq .

# Check hot threads
curl -s localhost:9600/_node/hot_threads

# Monitoring API
curl -s localhost:9600/_node/stats | jq '.process.cpu, .jvm.mem'

# Tag failures for investigation
grok { tag_on_failure => ["_grokparsefailure_step1"] }

# Use stdout for debugging mid-pipeline
output {
  stdout { codec => rubydebug }    # Pretty print
  stdout { codec => json_lines }    # JSON output
  stdout { codec => dots }          # Just dots (throughput check)
}

10. Performance Tips

▸dissect > grok for simple patterns (10x faster)
▸Avoid ruby filter in hot paths
▸Use workers setting to parallelize filter execution
▸Monitor with /_node/stats/pipelines API

Batch size (throughput)

pipeline.batch.size: 125

Default — increase for throughput

Batch delay (latency)

pipeline.batch.delay: 50

In ms — decrease for latency

Persistent queues

queue.type: persisted

For durability