ELK Stack — Filter Plugins

7 minute read

Filter Plugins

Filter plugins are the heart of Logstash — they parse unstructured data into structured fields, transform values, enrich events with external data, and route processing with conditionals.

grok — Pattern Matching

The grok filter parses unstructured text using predefined patterns. It’s the most important filter for log parsing.

Pattern syntax: %{PATTERN_NAME:field_name}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
}

Sample input:

83.149.9.216 - - [17/May/2025:10:05:03 +0000] "GET /presentations/logstash.html HTTP/1.1" 200 7697 "http://example.com/" "Mozilla/5.0"

Parsed output fields:

{
    "clientip" => "83.149.9.216",
       "ident" => "-",
        "auth" => "-",
   "timestamp" => "17/May/2025:10:05:03 +0000",
        "verb" => "GET",
     "request" => "/presentations/logstash.html",
  "httpversion" => "1.1",
    "response" => "200",
       "bytes" => "7697",
    "referrer" => "\"http://example.com/\"",
       "agent" => "\"Mozilla/5.0\""
}

Common grok patterns:

Pattern Matches Example
%{IP:ip} IP address 192.168.1.1
%{WORD:word} Single word ERROR
%{NUMBER:num} Number (int or float) 42, 3.14
%{GREEDYDATA:rest} Everything remaining Any text
%{TIMESTAMP_ISO8601:ts} ISO 8601 timestamp 2026-04-01T14:30:00Z
%{LOGLEVEL:level} Log level INFO, ERROR
%{COMBINEDAPACHELOG} Full Apache log line See example above
%{SYSLOGLINE} Syslog message Standard syslog format

Custom grok pattern:

filter {
  grok {
    match => {
      "message" => "%{TIMESTAMP_ISO8601:timestamp} \[%{LOGLEVEL:level}\] %{WORD:service} - %{GREEDYDATA:log_message}"
    }
  }
}

Sample input:

2026-04-01T14:30:00.123Z [ERROR] payment-service - Failed to process transaction ID=12345

Multiple match patterns (tries each until one succeeds):

filter {
  grok {
    match => {
      "message" => [
        "%{COMBINEDAPACHELOG}",
        "%{TIMESTAMP_ISO8601:timestamp} \[%{LOGLEVEL:level}\] %{GREEDYDATA:log_message}",
        "%{GREEDYDATA:log_message}"
      ]
    }
    tag_on_failure => ["_grokparsefailure"]
  }
}

mutate — Modify Fields

The mutate filter modifies event fields — rename, remove, add, convert types, and more:

filter {
  mutate {
    # Rename fields
    rename => {
      "clientip" => "client_ip"
      "httpversion" => "http_version"
    }

    # Remove unwanted fields
    remove_field => ["message", "host", "@version"]

    # Add new fields
    add_field => {
      "environment" => "production"
      "processed_by" => "logstash-01"
    }

    # Convert field types
    convert => {
      "response" => "integer"
      "bytes" => "integer"
      "duration" => "float"
    }

    # String replacement (gsub)
    gsub => [
      # field, regex, replacement
      "request", "\\?.*", "",           # Remove query strings
      "agent", "\"", ""                 # Remove quotes
    ]

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

    # Replace field value
    replace => { "message" => "Processed event from %{client_ip}" }

    # Strip whitespace
    strip => ["username", "email"]

    # Split string into array
    split => { "tags_string" => "," }

    # Merge arrays
    merge => { "all_tags" => "extra_tags" }
  }
}

Note: mutate operations within a single mutate block execute in a fixed order: rename, replace, convert, gsub, uppercase, lowercase, strip, remove_field, add_field, add_tag, remove_tag. If you need a different order, use multiple mutate blocks.

date — Parsing Timestamps

The date filter parses timestamp strings and sets @timestamp:

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

Common date formats:

Format String Example
ISO8601 2026-04-01T14:30:00.000Z
UNIX 1711964400 (epoch seconds)
UNIX_MS 1711964400000 (epoch millis)
dd/MMM/yyyy:HH:mm:ss Z 01/Apr/2026:14:30:00 +0530
yyyy-MM-dd HH:mm:ss 2026-04-01 14:30:00
MMM dd yyyy HH:mm:ss Apr 01 2026 14:30:00
MMM d HH:mm:ss Apr 1 14:30:00

Multiple date formats (tries each in order):

filter {
  date {
    match => [
      "timestamp",
      "ISO8601",
      "yyyy-MM-dd HH:mm:ss",
      "dd/MMM/yyyy:HH:mm:ss Z"
    ]
  }
}

geoip — Geographic Enrichment

The geoip filter adds geographic data based on IP addresses:

filter {
  geoip {
    source => "client_ip"
    target => "geo"
    # Optionally specify which fields to include
    fields => ["city_name", "country_name", "country_code2",
               "continent_code", "region_name", "location",
               "latitude", "longitude", "timezone"]
  }
}

Output fields:

{
    "geo" => {
        "city_name" => "San Francisco",
        "country_name" => "United States",
        "country_code2" => "US",
        "region_name" => "California",
        "location" => {
            "lat" => 37.7749,
            "lon" => -122.4194
        },
        "timezone" => "America/Los_Angeles"
    }
}

useragent — Parse User Agent Strings

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

Output:

{
    "user_agent" => {
        "name" => "Chrome",
        "version" => "120.0.0",
        "os" => "Mac OS X",
        "os_version" => "14.2",
        "device" => "Mac"
    }
}

dissect — Fixed-Format Parsing

Faster than grok for logs with a consistent, delimiter-based format:

filter {
  dissect {
    mapping => {
      "message" => "%{timestamp} %{+timestamp} %{level} [%{thread}] %{logger} - %{log_message}"
    }
  }
}

Sample input:

2026-04-01 14:30:00 ERROR [main] com.example.App - Connection refused

Output:

{
    "timestamp" => "2026-04-01 14:30:00",
       "level" => "ERROR",
      "thread" => "main",
      "logger" => "com.example.App",
 "log_message" => "Connection refused"
}

The %{+field} notation appends to an existing field (here, joining date and time into timestamp).

json — Parse JSON

filter {
  json {
    source => "message"
    target => "parsed"          # Omit target to merge into root
    remove_field => ["message"]
  }
}

Sample input:

{"user": "alice", "action": "login", "ip": "10.0.1.50"}

Output (with target):

{
    "parsed" => {
        "user" => "alice",
        "action" => "login",
        "ip" => "10.0.1.50"
    }
}

kv — Key-Value Pairs

Parse key=value formatted data:

filter {
  kv {
    source => "message"
    field_split => " "
    value_split => "="
    trim_key => "\""
    trim_value => "\""
    target => "params"
  }
}

Sample input:

user="alice" action="login" status="success" duration="1.23"

Output:

{
    "params" => {
        "user" => "alice",
        "action" => "login",
        "status" => "success",
        "duration" => "1.23"
    }
}

ruby — Inline Ruby Code

For logic that no existing plugin covers, use the ruby filter:

filter {
  ruby {
    code => '
      # Calculate request duration in seconds
      if event.get("duration_ms")
        event.set("duration_sec", event.get("duration_ms").to_f / 1000.0)
      end

      # Classify response time
      duration = event.get("duration_sec") || 0
      if duration < 0.5
        event.set("performance", "fast")
      elsif duration < 2.0
        event.set("performance", "normal")
      else
        event.set("performance", "slow")
      end
    '
  }
}

Conditionals — if / else if / else

Use conditionals to apply filters selectively based on field values:

filter {
  # Only parse Apache logs
  if [type] == "apache-access" {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }

    geoip {
      source => "clientip"
    }
  }

  # Parse syslog messages
  else if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGLINE}" }
    }
  }

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

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

  # Check field existence
  if [user_agent] {
    useragent {
      source => "user_agent"
      target => "ua"
    }
  }

  # Check for specific tag
  if "server_error" in [tags] {
    mutate {
      add_field => { "notify" => "true" }
    }
  }

  # Negation
  if [status] not in ["success", "ok"] {
    mutate { add_tag => ["needs_review"] }
  }

  # Regex matching
  if [message] =~ /Exception|Error|FATAL/ {
    mutate { add_tag => ["exception"] }
  }
}

Conditional operators:

Operator Description Example
== Equality [status] == 200
!= Not equal [env] != "test"
>, >=, <, <= Comparison [bytes] > 1000
=~ Regex match [path] =~ /\.php$/
!~ Regex not match [path] !~ /\.(css\|js)$/
in Contains "error" in [tags]
not in Not contains [env] not in ["dev", "test"]
and Logical AND [status] == 200 and [bytes] > 0
or Logical OR [level] == "ERROR" or [level] == "FATAL"
! or not Negation !( [status] == 200 )

Complete Pipeline Example — Apache Access Logs

Bringing it all together:

input {
  beats {
    port => 5044
  }
}

filter {
  if [type] == "apache-access" {
    # 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 types
    mutate {
      convert => {
        "response" => "integer"
        "bytes" => "integer"
      }
      remove_field => ["message", "@version"]
    }

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

    # Parse user agent
    useragent {
      source => "agent"
      target => "user_agent"
    }

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

    # Drop health check requests
    if [request] =~ /^\/health/ {
      drop { }
    }
  }
}

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

Quick Reference

Filter Purpose Key Option
grok Parse unstructured text with patterns match => { "field" => "pattern" }
mutate Modify fields (rename, remove, convert) Multiple operations in fixed order
date Parse timestamps into @timestamp match => ["field", "format"]
geoip Enrich IP addresses with geo data source => "ip_field"
useragent Parse browser user agent strings source => "ua_field"
dissect Parse fixed-format delimited data mapping => { "field" => "pattern" }
json Parse JSON strings into fields source => "json_field"
kv Parse key=value pairs field_split, value_split
ruby Run custom Ruby code code => 'ruby code'
drop Discard events Used inside conditionals

Tip: Use dissect instead of grok when your log format has a fixed structure. It’s significantly faster because it doesn’t use regular expressions.

👉 Next: Output Plugins

You may also enjoy