ELK Stack — Logstash Exercises

10 minute read

Logstash Practice Exercises

These exercises progress from basic to advanced, building your skills with Logstash pipelines, filters, and multi-source processing.

Exercise 1: Basic stdin → stdout Pipeline

Scenario: You need to verify that Logstash is working correctly by creating the simplest possible pipeline that reads from the terminal and outputs parsed events.

Requirements:

  1. Create a pipeline config that reads from stdin
  2. Output to stdout using the rubydebug codec
  3. Add a mutate filter that:
    • Adds a field processed_by with value logstash
    • Removes the @version field
  4. Test by typing Hello Logstash and verify the output

Expected output after typing Hello Logstash:

{
    "@timestamp" => 2026-04-01T09:00:00.000Z,
       "message" => "Hello Logstash",
          "host" => { "hostname" => "your-host" },
         "event" => { "original" => "Hello Logstash" },
  "processed_by" => "logstash"
}
Show Solution

Create the file exercise1.conf:

input {
  stdin { }
}

filter {
  mutate {
    add_field => { "processed_by" => "logstash" }
    remove_field => ["@version"]
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

Run with:

bin/logstash -f exercise1.conf

Type Hello Logstash and press Enter. The output will show your message with the processed_by field added and @version removed.

To validate config without running:

bin/logstash --config.test_and_exit -f exercise1.conf

Exercise 2: Parsing Apache Access Logs with Grok

Scenario: Your web team needs Apache access logs parsed into structured fields for analysis in Kibana. You’ll build a pipeline that reads Apache log lines from stdin, parses them with grok, and outputs structured events.

Requirements:

  1. Create a pipeline that reads from stdin
  2. Use the grok filter with the %{COMBINEDAPACHELOG} pattern
  3. Parse the timestamp using the date filter
  4. Convert response and bytes fields to integers
  5. Add GeoIP data from the clientip field
  6. Remove the raw message field and @version
  7. Tag events with response >= 400 as client_error or >= 500 as server_error
  8. Output to stdout with rubydebug

Sample input data (paste these lines):

192.168.1.100 - frank [10/Oct/2025:13:55:36 -0700] "GET /api/users HTTP/1.1" 200 2326 "http://example.com/start" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0"
10.0.0.55 - - [10/Oct/2025:13:56:01 -0700] "POST /api/login HTTP/1.1" 401 187 "-" "curl/7.68.0"
172.16.0.1 - admin [10/Oct/2025:13:56:15 -0700] "DELETE /api/users/99 HTTP/1.1" 500 0 "http://example.com/admin" "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_2) Safari/605.1"
Show Solution

Create the file exercise2-apache.conf:

input {
  stdin { }
}

filter {
  # Parse Apache combined log format
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
    tag_on_failure => ["_grokparsefailure"]
  }

  # Parse the timestamp
  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }

  # Convert numeric fields
  mutate {
    convert => {
      "response" => "integer"
      "bytes" => "integer"
    }
    remove_field => ["message", "@version"]
  }

  # Add geographic data
  geoip {
    source => "clientip"
    target => "geo"
  }

  # Tag error responses
  if [response] >= 500 {
    mutate { add_tag => ["server_error"] }
  } else if [response] >= 400 {
    mutate { add_tag => ["client_error"] }
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

Run with:

bin/logstash -f exercise2-apache.conf

The 401 line should have the client_error tag, and the 500 line should have the server_error tag.

Exercise 3: Multi-Source Pipeline with Conditionals

Scenario: Your organization has three types of logs arriving on different TCP ports. You need a single pipeline that handles all three, applies source-specific parsing, and routes to different Elasticsearch indices.

Log types:

  • Port 5001 — JSON application logs: {"timestamp": "2026-04-01T10:30:00Z", "level": "ERROR", "service": "payment", "message": "Transaction failed", "tx_id": "TXN-12345"}
  • Port 5002 — Syslog messages: <34>Apr 1 10:30:00 webserver01 nginx: 192.168.1.1 - GET /api/health 200
  • Port 5003 — Key-value application metrics: timestamp=2026-04-01T10:30:00Z service=api endpoint=/users count=150 avg_duration_ms=45.2

Requirements:

  1. Three TCP inputs, each with a different type field
  2. Conditional filters based on type:
    • app-json: Parse JSON, set timestamp from timestamp field
    • syslog-raw: Parse with grok, extract severity
    • app-metrics: Parse key-value pairs, convert numeric fields
  3. All events get an environment field set to production
  4. Drop events where level is DEBUG
  5. Output to stdout AND to Elasticsearch (using conditional index names)
Show Solution

Create the file exercise3-multi-source.conf:

input {
  tcp {
    port => 5001
    codec => json_lines
    type => "app-json"
  }

  tcp {
    port => 5002
    codec => plain
    type => "syslog-raw"
  }

  tcp {
    port => 5003
    codec => plain
    type => "app-metrics"
  }
}

filter {
  # Add environment to all events
  mutate {
    add_field => { "environment" => "production" }
  }

  # JSON application logs
  if [type] == "app-json" {
    date {
      match => ["timestamp", "ISO8601"]
      target => "@timestamp"
      remove_field => ["timestamp"]
    }

    # Drop debug events
    if [level] == "DEBUG" {
      drop { }
    }
  }

  # Syslog messages
  else if [type] == "syslog-raw" {
    grok {
      match => {
        "message" => "<%{POSINT:priority}>%{SYSLOGTIMESTAMP:syslog_timestamp} %{HOSTNAME:hostname} %{WORD:program}: %{IP:client_ip} - %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:response:int}"
      }
      tag_on_failure => ["_grokparsefailure"]
    }

    date {
      match => ["syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
      target => "@timestamp"
      remove_field => ["syslog_timestamp"]
    }
  }

  # Key-value metrics
  else if [type] == "app-metrics" {
    kv {
      source => "message"
      field_split => " "
      value_split => "="
    }

    date {
      match => ["timestamp", "ISO8601"]
      target => "@timestamp"
      remove_field => ["timestamp"]
    }

    mutate {
      convert => {
        "count" => "integer"
        "avg_duration_ms" => "float"
      }
    }
  }

  # Clean up
  mutate {
    remove_field => ["@version"]
  }
}

output {
  # Console output for debugging
  stdout {
    codec => rubydebug
  }

  # Elasticsearch with conditional index names
  elasticsearch {
    hosts => ["https://localhost:9200"]
    user => "elastic"
    password => "changeme"
    index => "%{type}-%{+YYYY.MM.dd}"
  }
}

Test with netcat:

# Send JSON log
echo '{"timestamp": "2026-04-01T10:30:00Z", "level": "ERROR", "service": "payment", "message": "Transaction failed", "tx_id": "TXN-12345"}' | nc localhost 5001

# Send syslog
echo '<34>Apr  1 10:30:00 webserver01 nginx: 192.168.1.1 - GET /api/health 200' | nc localhost 5002

# Send metrics
echo 'timestamp=2026-04-01T10:30:00Z service=api endpoint=/users count=150 avg_duration_ms=45.2' | nc localhost 5003

Exercise 4: Beats → Logstash → Elasticsearch Pipeline

Scenario: Set up a production-ready pipeline that receives logs from Filebeat, parses mixed log formats (Nginx access logs and Java application logs), and sends structured data to Elasticsearch with proper index routing.

Requirements:

  1. Beats input on port 5044 with SSL/TLS enabled
  2. Handle two log formats based on Filebeat fields:
    • Nginx access logs (identified by [fields][log_type] == "nginx")
    • Java application logs with multiline stack traces (identified by [fields][log_type] == "java-app")
  3. Nginx logs: Parse with grok, add GeoIP, parse user agent
  4. Java logs: Parse log level, class, thread, and message with grok
  5. All events: Set @timestamp from parsed timestamp, add environment and team fields
  6. Send to Elasticsearch with index pattern {log_type}-{YYYY.MM.dd}
  7. Enable dead letter queue for failed events

Sample Nginx log:

10.0.1.50 - - [01/Apr/2026:14:30:00 +0530] "GET /api/v2/products?page=1 HTTP/1.1" 200 4523 "https://example.com/shop" "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X)"

Sample Java log:

2026-04-01 14:30:00.456 ERROR [http-nio-8080-exec-5] c.e.service.PaymentService - Payment processing failed
java.lang.NullPointerException: Order ID cannot be null
    at com.example.service.PaymentService.process(PaymentService.java:42)
    at com.example.controller.PaymentController.pay(PaymentController.java:28)
Show Solution

Create the file exercise4-beats-pipeline.conf:

input {
  beats {
    port => 5044
    ssl_enabled => true
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    ssl_client_authentication => "required"
  }
}

filter {
  # Add common fields
  mutate {
    add_field => {
      "environment" => "production"
      "team" => "platform-engineering"
    }
  }

  # Nginx access logs
  if [fields][log_type] == "nginx" {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
      tag_on_failure => ["_grokparsefailure"]
    }

    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
      remove_field => ["timestamp"]
    }

    mutate {
      convert => {
        "response" => "integer"
        "bytes" => "integer"
      }
    }

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

    useragent {
      source => "agent"
      target => "user_agent"
    }

    # Tag errors
    if [response] >= 500 {
      mutate { add_tag => ["server_error"] }
    }
  }

  # Java application logs
  else if [fields][log_type] == "java-app" {
    grok {
      match => {
        "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \[%{DATA:thread}\] %{JAVACLASS:class} - %{GREEDYDATA:log_message}"
      }
      tag_on_failure => ["_grokparsefailure"]
    }

    date {
      match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]
      target => "@timestamp"
      remove_field => ["timestamp"]
    }

    # Extract exception class if present
    if [log_message] =~ /Exception|Error/ {
      grok {
        match => {
          "log_message" => "%{JAVACLASS:exception_class}: %{GREEDYDATA:exception_message}"
        }
        tag_on_failure => []
      }
      mutate { add_tag => ["exception"] }
    }
  }

  # Clean up
  mutate {
    remove_field => ["@version", "message"]
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    user => "elastic"
    password => "changeme"
    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    index => "%{[fields][log_type]}-%{+YYYY.MM.dd}"
  }
}

Corresponding Filebeat config for Nginx:

# filebeat.yml
filebeat.inputs:
  - type: log
    paths:
      - /var/log/nginx/access.log
    fields:
      log_type: nginx
    fields_under_root: false

  - type: log
    paths:
      - /var/log/app/application.log
    fields:
      log_type: java-app
    fields_under_root: false
    multiline.pattern: '^\d{4}-\d{2}-\d{2}'
    multiline.negate: true
    multiline.match: after

output.logstash:
  hosts: ["logstash-server:5044"]
  ssl.certificate: "/etc/filebeat/certs/filebeat.crt"
  ssl.key: "/etc/filebeat/certs/filebeat.key"
  ssl.certificate_authorities: ["/etc/filebeat/certs/ca.crt"]

Enable DLQ in logstash.yml:

dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 1024mb

Exercise 5: Multiple Pipelines Setup

Scenario: Design a multi-pipeline Logstash deployment for a microservices platform. You need separate pipelines for different data sources with a distributor pattern for efficient routing.

Requirements:

  1. Create a pipelines.yml with four pipelines:
    • intake — Receives all data from Beats on port 5044, forwards to distributor
    • distributor — Routes events to the appropriate downstream pipeline based on [fields][service]
    • web-pipeline — Processes web server logs (Nginx/Apache) with grok, GeoIP
    • api-pipeline — Processes JSON API logs, enriches with request duration classification
  2. Intake pipeline uses persistent queue (512 MB)
  3. Web pipeline: 2 workers, batch size 250
  4. API pipeline: 4 workers, batch size 125
  5. Each downstream pipeline outputs to its own Elasticsearch index
  6. Enable DLQ on the web pipeline
Show Solution

pipelines.yml:

# /etc/logstash/pipelines.yml

- pipeline.id: intake
  path.config: "/etc/logstash/conf.d/intake.conf"
  pipeline.workers: 2
  queue.type: persisted
  queue.max_bytes: 512mb

- pipeline.id: distributor
  path.config: "/etc/logstash/conf.d/distributor.conf"
  pipeline.workers: 1

- pipeline.id: web-pipeline
  path.config: "/etc/logstash/conf.d/web-pipeline.conf"
  pipeline.workers: 2
  pipeline.batch.size: 250
  dead_letter_queue.enable: true
  dead_letter_queue.max_bytes: 256mb

- pipeline.id: api-pipeline
  path.config: "/etc/logstash/conf.d/api-pipeline.conf"
  pipeline.workers: 4
  pipeline.batch.size: 125

intake.conf:

input {
  beats {
    port => 5044
  }
}

output {
  pipeline {
    send_to => ["distributor"]
  }
}

distributor.conf:

input {
  pipeline {
    address => "distributor"
  }
}

output {
  if [fields][service] == "nginx" or [fields][service] == "apache" {
    pipeline {
      send_to => ["web-sink"]
    }
  } else if [fields][service] =~ /^api-/ {
    pipeline {
      send_to => ["api-sink"]
    }
  } else {
    # Fallback: send unrecognized events to Elasticsearch directly
    elasticsearch {
      hosts => ["https://localhost:9200"]
      index => "unclassified-%{+YYYY.MM.dd}"
      user => "elastic"
      password => "changeme"
    }
  }
}

web-pipeline.conf:

input {
  pipeline {
    address => "web-sink"
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
    tag_on_failure => ["_grokparsefailure"]
  }

  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }

  mutate {
    convert => {
      "response" => "integer"
      "bytes" => "integer"
    }
  }

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

  useragent {
    source => "agent"
    target => "user_agent"
  }

  mutate {
    remove_field => ["message", "@version"]
    add_field => { "pipeline" => "web-pipeline" }
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "web-logs-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "changeme"
  }
}

api-pipeline.conf:

input {
  pipeline {
    address => "api-sink"
  }
}

filter {
  json {
    source => "message"
  }

  date {
    match => ["timestamp", "ISO8601"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }

  # Classify request duration
  if [duration_ms] {
    ruby {
      code => '
        duration = event.get("duration_ms").to_f
        if duration < 100
          event.set("duration_class", "fast")
        elsif duration < 500
          event.set("duration_class", "normal")
        elsif duration < 2000
          event.set("duration_class", "slow")
        else
          event.set("duration_class", "critical")
        end
      '
    }

    mutate {
      convert => { "duration_ms" => "float" }
    }
  }

  if [level] == "DEBUG" {
    drop { }
  }

  mutate {
    remove_field => ["message", "@version"]
    add_field => { "pipeline" => "api-pipeline" }
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "api-logs-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "changeme"
  }
}

Start Logstash — it will automatically load all pipelines from pipelines.yml:

bin/logstash

Monitor pipeline performance:

# Check all pipelines
curl -s localhost:9600/_node/stats/pipelines | jq '.pipelines | keys'

# Check specific pipeline stats
curl -s localhost:9600/_node/stats/pipelines/web-pipeline | jq '.pipelines["web-pipeline"].events'

# Verify events flow through the distributor
curl -s localhost:9600/_node/stats/pipelines/distributor | jq '.pipelines["distributor"].events'

Summary

Exercise Concepts Practiced
1 — stdin → stdout Basic pipeline, mutate filter, testing workflow
2 — Apache log parsing grok, date, mutate convert, GeoIP, conditionals
3 — Multi-source Multiple TCP inputs, JSON/grok/KV filters, conditional routing
4 — Beats → ES Beats input with SSL, mixed log parsing, Filebeat config, DLQ
5 — Multiple pipelines pipelines.yml, pipeline-to-pipeline, distributor pattern, tuning

Tip: After completing these exercises, try combining concepts — for example, add Kafka as an intermediate buffer between Beats and Logstash, or set up a DLQ reprocessing pipeline for Exercise 4.

👉 Back to: Logstash

You may also enjoy