ELK Stack — Log Parsing Exercises

11 minute read

Practice Exercises

These exercises cover all log parsing techniques from this chapter. Each exercise increases in complexity — work through them in order.

Exercise 1 — Parse Apache Access Logs with Grok

Scenario:

Your web team needs visibility into Apache access logs. Parse the Combined Log Format and create a complete Logstash pipeline that extracts all fields, converts types, and parses timestamps.

Sample logs:

0.0.5 - admin [01/Apr/2026:10:15:30 +0000] "GET /api/v1/users?page=2 HTTP/1.1" 200 4523 "https://example.com/dashboard" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
0.0.8 - - [01/Apr/2026:10:15:31 +0000] "POST /api/v1/orders HTTP/1.1" 201 892 "-" "curl/7.68.0"
0.0.12 - john [01/Apr/2026:10:15:32 +0000] "GET /static/logo.png HTTP/1.1" 304 0 "https://example.com/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"

Requirements:

Write a Grok pattern to parse the Combined Apache Log Format
Convert response to integer and bytes to integer
Parse timestamp into @timestamp
Extract the request URI path and query string separately
Handle the - values for missing fields (ident, referrer)

Show Solution

Logstash Pipeline:

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    match => {
      "message" => "%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{NOTSPACE:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:response:int} %{NUMBER:bytes:int} %{QS:referrer} %{QS:agent}"
    }
    tag_on_failure => ["_grok_apache_fail"]
  }

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

  # Split request into path and query
  grok {
    match => {
      "request" => "%{URIPATH:url_path}(?:\?%{NOTSPACE:url_query})?"
    }
    tag_on_failure => []
  }

  # Clean up dash values
  if [ident] == "-" {
    mutate { remove_field => ["ident"] }
  }
  if [auth] == "-" {
    mutate { remove_field => ["auth"] }
  }

  # Remove quotes from referrer and agent
  mutate {
    gsub => [
      "referrer", "\"", "",
      "agent", "\"", ""
    ]
  }
  if [referrer] == "-" {
    mutate { remove_field => ["referrer"] }
  }
}

output {
  elasticsearch {
    hosts => ["https://es01:9200"]
    index => "logs-apache-%{+YYYY.MM.dd}"
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
  }
}

Parsed output for the first log line:

{
  "@timestamp": "2026-04-01T10:15:30.000Z",
  "client_ip": "10.0.0.5",
  "auth": "admin",
  "method": "GET",
  "request": "/api/v1/users?page=2",
  "url_path": "/api/v1/users",
  "url_query": "page=2",
  "http_version": "1.1",
  "response": 200,
  "bytes": 4523,
  "referrer": "https://example.com/dashboard",
  "agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}

Exercise 2 — Create Custom Grok Patterns for a Custom Log Format

Scenario:

Your microservices produce a custom log format. Create reusable custom Grok patterns to parse these logs and deploy them as a shared pattern library.

Sample logs:

[2026-04-01T10:15:30.456Z] [trace-id=abc12def-3456-7890-abcd-ef1234567890] [PaymentService/v2.1.0] WARN  processPayment - Payment retry attempt 3/5 for order_id=ORD-98765, amount=149.99, currency=USD, gateway=stripe
[2026-04-01T10:15:31.789Z] [trace-id=fed21cba-6543-0987-dcba-fe9876543210] [UserService/v1.5.2] ERROR authenticateUser - Authentication failed for user_id=USR-12345, method=oauth2, provider=google, ip=192.168.1.100
[2026-04-01T10:15:32.012Z] [trace-id=11223344-5566-7788-9900-aabbccddeeff] [OrderService/v3.0.1] INFO  createOrder - Order created successfully for user_id=USR-67890, items=3, total=299.97

Requirements:

Create a custom pattern file with reusable patterns for: trace ID, service name with version, log level, method name
Create a composite pattern that parses the entire log line
Extract key-value pairs from the message body
Write the complete Logstash pipeline

Show Solution

Custom Pattern File — /etc/logstash/patterns/microservice:

# Trace ID (UUID with prefix)
MS_TRACE_ID trace-id=[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}

# Service name with version
MS_SERVICE [A-Za-z][A-Za-z0-9]*/v[0-9]+\.[0-9]+\.[0-9]+

# Log level (with optional trailing spaces)
MS_LEVEL (TRACE|DEBUG|INFO|WARN|ERROR|FATAL)\s*

# Method name
MS_METHOD [a-zA-Z][a-zA-Z0-9_]*

# Full log pattern
MS_LOG \[%{TIMESTAMP_ISO8601:timestamp}\] \[%{MS_TRACE_ID:trace_id}\] \[%{MS_SERVICE:service_info}\] %{MS_LEVEL:level}%{MS_METHOD:method_name} - %{GREEDYDATA:log_message}

Logstash Pipeline:

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    patterns_dir => ["/etc/logstash/patterns"]
    match => {
      "message" => "%{MS_LOG}"
    }
    tag_on_failure => ["_grok_microservice_fail"]
  }

  # Strip "trace-id=" prefix from trace_id
  mutate {
    gsub => ["trace_id", "^trace-id=", ""]
  }

  # Split service_info into service name and version
  grok {
    match => {
      "service_info" => "(?<service_name>[^/]+)/v(?<service_version>[0-9]+\.[0-9]+\.[0-9]+)"
    }
    remove_field => ["service_info"]
  }

  # Trim whitespace from level
  mutate {
    strip => ["level"]
  }

  # Extract key-value pairs from the log message
  kv {
    source => "log_message"
    field_split => ", "
    value_split => "="
    target => "fields"
    trim_key => " "
  }

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

output {
  elasticsearch {
    hosts => ["https://es01:9200"]
    index => "logs-microservices-%{+YYYY.MM.dd}"
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
  }
}

Parsed output for the first log line:

{
  "@timestamp": "2026-04-01T10:15:30.456Z",
  "trace_id": "abc12def-3456-7890-abcd-ef1234567890",
  "service_name": "PaymentService",
  "service_version": "2.1.0",
  "level": "WARN",
  "method_name": "processPayment",
  "log_message": "Payment retry attempt 3/5 for order_id=ORD-98765, amount=149.99, currency=USD, gateway=stripe",
  "fields": {
    "order_id": "ORD-98765",
    "amount": "149.99",
    "currency": "USD",
    "gateway": "stripe"
  }
}

Exercise 3 — High-Throughput Parsing with Dissect

Scenario:

Your load balancer generates millions of access logs per minute in a fixed format. Grok is too slow — switch to Dissect for high-throughput parsing. Also implement a hybrid approach for the small percentage of error logs that need Grok.

Sample access logs (99% of traffic):

2026-04-01T10:15:30.456Z | 10.0.0.5 | GET | /api/users | 200 | 45ms | 1234 | req-abc123
2026-04-01T10:15:30.457Z | 10.0.0.8 | POST | /api/orders | 201 | 120ms | 892 | req-def456
2026-04-01T10:15:30.458Z | 10.0.0.12 | GET | /health | 200 | 2ms | 15 | req-ghi789

Sample error logs (1% of traffic):

2026-04-01T10:15:30.459Z | 10.0.0.5 | POST | /api/payments | 500 | 5023ms | 0 | req-jkl012 | error=TimeoutException: upstream_host=payment-gw.internal:8443 retry_count=3 circuit_breaker=open

Requirements:

Use Dissect to parse the standard access log format
Use Grok only for the error detail portion of error logs
Convert numeric fields to integers
Tag access logs vs error logs

Show Solution

Logstash Pipeline:

input {
  beats {
    port => 5044
  }
}

filter {
  # Step 1: Dissect the common fixed-format portion (fast)
  dissect {
    mapping => {
      "message" => "%{timestamp} | %{client_ip} | %{method} | %{path} | %{status} | %{duration}ms | %{bytes} | %{request_id}%{->}%{error_detail}"
    }
    convert_datatype => {
      "status" => "int"
      "bytes" => "int"
    }
  }

  # Strip "ms" and convert duration
  mutate {
    strip => ["duration", "error_detail"]
  }
  mutate {
    convert => {
      "duration" => "integer"
    }
  }

  # Step 2: Tag and parse error logs
  if [error_detail] and [error_detail] != "" {
    mutate {
      add_tag => ["error_log"]
    }

    # Strip the leading "| " from error_detail
    mutate {
      gsub => ["error_detail", "^\|\s*", ""]
    }

    # Use Grok for the complex error detail
    grok {
      match => {
        "error_detail" => "error=%{NOTSPACE:error_type}: %{GREEDYDATA:error_context}"
      }
      tag_on_failure => ["_grok_error_detail_fail"]
    }

    # Extract key=value pairs from error context
    if [error_context] {
      kv {
        source => "error_context"
        field_split => " "
        value_split => "="
        target => "error_fields"
      }
    }
  } else {
    mutate {
      add_tag => ["access_log"]
    }
  }

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

  # Cleanup
  mutate {
    remove_field => ["error_detail", "error_context"]
  }
}

output {
  elasticsearch {
    hosts => ["https://es01:9200"]
    index => "logs-loadbalancer-%{+YYYY.MM.dd}"
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
  }
}

Parsed access log:

{
  "@timestamp": "2026-04-01T10:15:30.456Z",
  "client_ip": "10.0.0.5",
  "method": "GET",
  "path": "/api/users",
  "status": 200,
  "duration": 45,
  "bytes": 1234,
  "request_id": "req-abc123",
  "tags": ["access_log"]
}

Parsed error log:

{
  "@timestamp": "2026-04-01T10:15:30.459Z",
  "client_ip": "10.0.0.5",
  "method": "POST",
  "path": "/api/payments",
  "status": 500,
  "duration": 5023,
  "bytes": 0,
  "request_id": "req-jkl012",
  "error_type": "TimeoutException",
  "error_fields": {
    "upstream_host": "payment-gw.internal:8443",
    "retry_count": "3",
    "circuit_breaker": "open"
  },
  "tags": ["error_log"]
}

Exercise 4 — Handle Java Multiline Stack Traces

Scenario:

Your Spring Boot application writes logs to a file. Some entries include full Java stack traces that span many lines. Configure end-to-end multiline handling from Filebeat through Logstash to Elasticsearch.

Sample log file:

2026-04-01 10:15:30.456 [http-nio-8080-exec-1] INFO  c.e.api.UserController - GET /api/users/42 completed in 23ms
2026-04-01 10:15:31.789 [http-nio-8080-exec-3] ERROR c.e.api.OrderController - Failed to create order
org.springframework.dao.DataAccessException: Could not execute statement
	at org.springframework.orm.jpa.vendor.HibernateJpaDialect.convertHibernateAccessException(HibernateJpaDialect.java:259)
	at org.springframework.orm.jpa.AbstractEntityManagerFactoryBean.translateExceptionIfPossible(AbstractEntityManagerFactoryBean.java:528)
	at org.springframework.dao.support.ChainedPersistenceExceptionTranslator.translateExceptionIfPossible(ChainedPersistenceExceptionTranslator.java:61)
Caused by: org.hibernate.exception.ConstraintViolationException: could not execute statement
	at org.hibernate.exception.internal.SQLExceptionTypeDelegate.convert(SQLExceptionTypeDelegate.java:59)
	at org.hibernate.internal.SessionImpl.doFlush(SessionImpl.java:1407)
	... 42 more
2026-04-01 10:15:32.012 [http-nio-8080-exec-2] INFO  c.e.api.HealthController - Health check OK

Requirements:

Configure Filebeat to merge Java stack traces into single events
Write a Logstash pipeline to parse the merged events
Extract the exception class and message from stack traces
Store the full stack trace in a separate field
Index INFO and ERROR logs to different data streams

Show Solution

Filebeat Configuration — filebeat.yml:

filebeat.inputs:
  - type: filestream
    id: spring-boot-logs
    paths:
      - /var/log/spring-app/*.log
    multiline:
      type: pattern
      pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}'
      negate: true
      match: after
      max_lines: 500
      timeout: 5s

output.logstash:
  hosts: ["logstash:5044"]

processors:
  - add_host_metadata: ~
  - add_fields:
      target: service
      fields:
        name: order-service
        version: "2.3.1"
        environment: production

Logstash Pipeline:

input {
  beats {
    port => 5044
  }
}

filter {
  # Parse the first line (always present)
  grok {
    match => {
      "message" => "^%{TIMESTAMP_ISO8601:timestamp} \[%{DATA:thread}\] %{LOGLEVEL:level}\s+%{NOTSPACE:logger} - %{GREEDYDATA:log_body}"
    }
    tag_on_failure => ["_grok_firstline_fail"]
  }

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

  # For error logs with stack traces, extract exception details
  if [level] == "ERROR" and [log_body] =~ /\n/ {
    grok {
      match => {
        "log_body" => "(?m)(?<error_summary>[^\n]+)\n(?<exception_class>[a-zA-Z0-9_.]+(?:Exception|Error)): (?<exception_message>[^\n]+)\n(?<stack_trace>[\s\S]*)"
      }
      tag_on_failure => ["_grok_stacktrace_fail"]
    }

    # Extract the root cause
    if [stack_trace] =~ /Caused by:/ {
      grok {
        match => {
          "stack_trace" => "(?m)Caused by: (?<root_cause_class>[a-zA-Z0-9_.]+(?:Exception|Error)): (?<root_cause_message>[^\n]+)"
        }
        tag_on_failure => []
      }
    }
  }

  # Map to ECS fields
  mutate {
    rename => {
      "exception_class"    => "[error][type]"
      "exception_message"  => "[error][message]"
      "stack_trace"        => "[error][stack_trace]"
      "level"              => "[log][level]"
      "logger"             => "[log][logger]"
    }
  }
}

output {
  if [log][level] == "ERROR" {
    elasticsearch {
      hosts => ["https://es01:9200"]
      index => "logs-errors-%{+YYYY.MM.dd}"
      ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    }
  } else {
    elasticsearch {
      hosts => ["https://es01:9200"]
      index => "logs-app-%{+YYYY.MM.dd}"
      ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    }
  }
}

Parsed ERROR event:

{
  "@timestamp": "2026-04-01T10:15:31.789Z",
  "thread": "http-nio-8080-exec-3",
  "log": {
    "level": "ERROR",
    "logger": "c.e.api.OrderController"
  },
  "error_summary": "Failed to create order",
  "error": {
    "type": "org.springframework.dao.DataAccessException",
    "message": "Could not execute statement",
    "stack_trace": "\tat org.springframework.orm.jpa.vendor.HibernateJpaDialect...\nCaused by: org.hibernate.exception.ConstraintViolationException: could not execute statement\n\tat ..."
  },
  "root_cause_class": "org.hibernate.exception.ConstraintViolationException",
  "root_cause_message": "could not execute statement",
  "service": {
    "name": "order-service",
    "version": "2.3.1",
    "environment": "production"
  }
}

Parsed INFO event:

{
  "@timestamp": "2026-04-01T10:15:30.456Z",
  "thread": "http-nio-8080-exec-1",
  "log": {
    "level": "INFO",
    "logger": "c.e.api.UserController"
  },
  "log_body": "GET /api/users/42 completed in 23ms"
}

Exercise 5 — Implement a JSON Structured Logging Pipeline

Scenario:

Your team is migrating from unstructured text logs to JSON structured logging. Build a complete pipeline that ingests JSON logs from multiple services, maps them to ECS, and handles both old (text) and new (JSON) formats during the migration period.

New JSON log format (from migrated services):

{"timestamp":"2026-04-01T10:15:30.456Z","level":"info","service":"user-service","version":"2.1.0","trace_id":"abc12def-3456-7890-abcd-ef1234567890","message":"User login successful","user_id":"USR-12345","method":"POST","path":"/api/auth/login","status":200,"duration_ms":87,"client_ip":"10.0.0.5"}
{"timestamp":"2026-04-01T10:15:31.789Z","level":"error","service":"payment-service","version":"1.8.3","trace_id":"fed21cba-6543-0987-dcba-fe9876543210","message":"Payment processing failed","order_id":"ORD-98765","amount":149.99,"currency":"USD","error":"InsufficientFunds","duration_ms":5023,"client_ip":"10.0.0.8"}

Old text log format (from legacy services):

2026-04-01 10:15:32.012 [main] INFO  c.e.LegacyService - Processing request for user=admin path=/api/legacy status=200 duration=150ms

Requirements:

Configure Filebeat to handle both JSON and text log files
Write a Logstash pipeline that detects and parses both formats
Map all fields to ECS
Ensure both formats produce consistent output fields
Add a log.format field to track which format was used

Show Solution

Filebeat Configuration — filebeat.yml:

filebeat.inputs:
  # JSON logs from migrated services
  - type: filestream
    id: json-service-logs
    paths:
      - /var/log/services/*/*.json
    parsers:
      - ndjson:
          target: ""
          overwrite_keys: true
          add_error_key: true
          message_key: message
    fields:
      log_format: json
    fields_under_root: true

  # Text logs from legacy services
  - type: filestream
    id: legacy-text-logs
    paths:
      - /var/log/services/legacy/*.log
    multiline:
      type: pattern
      pattern: '^\d{4}-\d{2}-\d{2}'
      negate: true
      match: after
      max_lines: 200
      timeout: 5s
    fields:
      log_format: text
    fields_under_root: true

output.logstash:
  hosts: ["logstash:5044"]

Logstash Pipeline:

input {
  beats {
    port => 5044
  }
}

filter {
  if [log_format] == "json" {
    # JSON logs — fields are already extracted by Filebeat ndjson parser

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

    # Map to ECS
    mutate {
      rename => {
        "level"       => "[log][level]"
        "service"     => "[service][name]"
        "version"     => "[service][version]"
        "trace_id"    => "[trace][id]"
        "client_ip"   => "[source][ip]"
        "method"      => "[http][request][method]"
        "path"        => "[url][path]"
        "status"      => "[http][response][status_code]"
        "duration_ms" => "[event][duration]"
        "error"       => "[error][message]"
        "user_id"     => "[user][id]"
        "order_id"    => "[labels][order_id]"
        "amount"      => "[labels][amount]"
        "currency"    => "[labels][currency]"
      }
      add_field => {
        "[log][format]" => "json"
      }
    }

  } else if [log_format] == "text" {
    # Legacy text logs — need Grok parsing

    grok {
      match => {
        "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{DATA:thread}\] %{LOGLEVEL:level}\s+%{NOTSPACE:logger} - %{GREEDYDATA:log_body}"
      }
      tag_on_failure => ["_grok_legacy_fail"]
    }

    # Extract key=value pairs from log body
    kv {
      source => "log_body"
      field_split => " "
      value_split => "="
      target => "kv_fields"
    }

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

    # Map to ECS
    mutate {
      rename => {
        "level"               => "[log][level]"
        "logger"              => "[log][logger]"
        "[kv_fields][user]"   => "[user][name]"
        "[kv_fields][path]"   => "[url][path]"
        "[kv_fields][status]" => "[http][response][status_code]"
      }
      add_field => {
        "[log][format]"    => "text"
        "[service][name]"  => "legacy-service"
      }
    }

    # Convert status code to integer
    if [http][response][status_code] {
      mutate {
        convert => {
          "[http][response][status_code]" => "integer"
        }
      }
    }

    # Parse duration (remove "ms" suffix)
    if [kv_fields][duration] {
      mutate {
        copy => { "[kv_fields][duration]" => "[event][duration]" }
        gsub => ["[event][duration]", "ms$", ""]
      }
      mutate {
        convert => {
          "[event][duration]" => "integer"
        }
      }
    }

    # Cleanup temporary fields
    mutate {
      remove_field => ["log_body", "kv_fields", "thread"]
    }
  }

  # Common processing for both formats
  # Normalize log level to lowercase
  if [log][level] {
    mutate {
      lowercase => ["[log][level]"]
    }
  }

  # Clean up internal fields
  mutate {
    remove_field => ["log_format"]
  }
}

output {
  elasticsearch {
    hosts => ["https://es01:9200"]
    index => "logs-services-%{+YYYY.MM.dd}"
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
  }
}

Parsed JSON log (migrated service):

{
  "@timestamp": "2026-04-01T10:15:30.456Z",
  "message": "User login successful",
  "log": {
    "level": "info",
    "format": "json"
  },
  "service": {
    "name": "user-service",
    "version": "2.1.0"
  },
  "trace": {
    "id": "abc12def-3456-7890-abcd-ef1234567890"
  },
  "source": {
    "ip": "10.0.0.5"
  },
  "http": {
    "request": { "method": "POST" },
    "response": { "status_code": 200 }
  },
  "url": { "path": "/api/auth/login" },
  "event": { "duration": 87 },
  "user": { "id": "USR-12345" }
}

Parsed text log (legacy service):

{
  "@timestamp": "2026-04-01T10:15:32.012Z",
  "message": "2026-04-01 10:15:32.012 [main] INFO  c.e.LegacyService - Processing request for user=admin path=/api/legacy status=200 duration=150ms",
  "log": {
    "level": "info",
    "logger": "c.e.LegacyService",
    "format": "text"
  },
  "service": {
    "name": "legacy-service"
  },
  "user": { "name": "admin" },
  "url": { "path": "/api/legacy" },
  "http": {
    "response": { "status_code": 200 }
  },
  "event": { "duration": 150 }
}

Both formats now produce consistent ECS-mapped output that can share the same Kibana dashboards.

Summary

Exercise	Technique	Key Takeaway
1	Grok (built-in)	Use `COMBINEDAPACHELOG` for standard log formats
2	Custom Grok	Create reusable pattern libraries for team-wide use
3	Dissect + Grok	Use dissect for speed, grok only where regex is needed
4	Multiline	Handle at Filebeat level with `negate: true` + `match: after`
5	Structured JSON	JSON + ECS = best approach for new services

Previous: Structured Logging ← · Up: Log Parsing ↑

Share on

Twitter Facebook LinkedIn

ELK Stack — Log Parsing Exercises

Practice Exercises

Exercise 1 — Parse Apache Access Logs with Grok

Exercise 2 — Create Custom Grok Patterns for a Custom Log Format

Exercise 3 — High-Throughput Parsing with Dissect

Exercise 4 — Handle Java Multiline Stack Traces

Exercise 5 — Implement a JSON Structured Logging Pipeline

Summary

Share on

You may also enjoy

AWS EKS Karpenter: Intelligent Kubernetes Node Autoscaling and Provisioning

Terraform Provider Development: Building Custom Providers with the Plugin Framework

EKS IRSA and Pod Identity: Fine-Grained AWS Permissions Without Static Keys

Writing Resume: Quantifying Infrastructure Impact with Metrics and Cost Savings