Applying Logstash Filters

1. Understanding Filter Plugins

Plugin Type	Description	Usage Example	Notes
grok	Extracts structured data from unstructured logs using regular expressions.	`{<br> "grok": {<br> "match": {<br> "message": "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}"<br> }<br> }<br>}`	Supports predefined patterns like `TIMESTAMP_ISO8601`, `LOGLEVEL`.
mutate	Modifies fields in the event, such as renaming or converting types.	`{<br> "mutate": {<br> "convert": {<br> "status_code": "integer"<br> },<br> "rename": {<br> "old_field": "new_field"<br> }<br> }<br>}`	Can convert types, add/remove fields, and modify field values.
date	Parses date strings and sets the `@timestamp` field.	`{<br> "date": {<br> "match": [<br> "timestamp", "yyyy-MM-dd HH:mm:ss"<br> ]<br> }<br>}`	Supports multiple date formats and timezone handling.
json	Parses JSON logs into structured fields.	`{<br> "json": {<br> "source": "message",<br> "target": "parsed_json"<br> }<br>}`	Can parse nested JSON and set fields from parsed data.
clone	Duplicates events to process them multiple times with different filters.	`{<br> "clone": {<br> "streams": [<br> { "if": "ctx.status == 'error'" },<br> { "if": "ctx.status == 'warning'" }<br> ]<br> }<br>}`	Useful for routing events to different outputs or applying different transformations.

Example: Using grok to parse Apache logs

{
  "grok": {
    "match": {
      "message": "%{COMBINED_LOG_FORMAT}"
    }
  }
}

Note: Filter plugins are processed in the order they appear in the Logstash configuration file.

Warning: Misconfigured grok patterns can cause performance issues or failed parsing.

2. Using Grok Filter

Field	Description	Example
match	Pattern to match against the log message	`"message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}"`
target	Field to apply the grok pattern to	`"message"`
add_field	Adds new fields to the event	`"[user][id]" => "%{USER_ID}"`
remove_field	Removes specified fields from the event	`"user_agent"`

Example: Parsing Apache Logs

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

Example: Custom Pattern

filter {
  grok {
    match => { "message" => "%{IP:client_ip} %{WORD:method} %{URIPATH:uri}" }
  }
}

Note: Grok patterns are based on regular expressions and can be reused across multiple filters.

3. Using Mutate Filter

Field	Description	Example
convert	Converts field values to a specified type	`{ "convert": { "status_code": "integer" } }`
copy	Copies field values to a new field	`{ "copy": { "message": "log_message" } }`
remove	Removes fields from the event	`{ "remove": [ "temp_field" ] }`
rename	Renames a field	`{ "rename": { "old_name": "new_name" } }`
gsub	Applies a regex substitution to field values	`{ "gsub": [ "message", "error", "ERROR" ] }`

Example: Converting field types

{
  "filter": [
    {
      "mutate": {
        "convert": {
          "status_code": "integer",
          "response_time": "float"
        }
      }
    }
  ]
}

Example: Copying and renaming fields

{
  "filter": [
    {
      "mutate": {
        "copy": {
          "message": "log_message"
        },
        "rename": {
          "timestamp": "@timestamp"
        }
      }
    }
  ]
}

Note: The mutate filter is powerful for transforming log data in-place. It supports various data type conversions and field manipulations.

4. Using Date Filter

Field	Description	Example
match	Array of format patterns to match against the date field	`["YYYY-MM-dd HH:mm:ss", "dd/MM/yyyy:HH:mm:ss"]`
target	Field name to write the parsed date to (defaults to `@timestamp`)	`event_time`
timezone	Timezone to use for parsing (default is UTC)	`America/New_York`
locale	Locale to use for parsing (default is `en`)	`fr`

Example: Parsing Log Timestamps

{<br>  "filter": {<br>    "date": {<br>      "match": ["YYYY-MM-dd HH:mm:ss"],<br>      "source": "timestamp"<br>    }<br>  }<br>}

Note: The match field supports multiple patterns to handle various timestamp formats in your logs.

5. Using JSON Filter

Field	Description	Default	Notes
source	Field to parse as JSON	`message`	Can be any field containing a JSON string
target	Field to store parsed JSON data	`nil`	If not set, parsed data is merged into the root event
skip_on_error	Whether to skip filter on parse errors	`false`	Set to `true` to prevent errors from halting pipeline
remove_field	Fields to remove after parsing	`nil`	Accepts array of field names to delete

Example: Parsing JSON from a field

filter {
  json {
    source => "message"
    target => "parsed_json"
  }
}

Example: Merging parsed JSON into root event

filter {
  json {
    source => "log_data"
  }
}

Note: If the source field contains invalid JSON, the event will be dropped unless skip_on_error is set to true.

6. Using Dissect Filter

Field	Description	Example
match	Defines the pattern to match against the input field.	`"message" => "%{timestamp} %{level} %{logger} %{message}"`
target	Specifies the field to parse. Defaults to `message`.	`"target" => "log_message"`
break_on_match	Stops processing after the first match. Default is `false`.	`"break_on_match" => true`

Example: Parsing Log Messages

{
  "filter": {
    "dissect": {
      "match": "%{timestamp} %{level} %{logger} %{message}",
      "target": "message"
    }
  }
}

Note: The dissect filter is ideal for structured logs where fields are separated by spaces or other delimiters.

7. Using GeoIP Filter

Field	Description	Example
source	Input field to extract IP address from	`clientip`
target	Target field to store the result	`geoip`
database	Path to the GeoIP database file	`/usr/share/GeoIP/GeoLite2-City.mmdb`
add_field	Additional fields to add to the event	`{ "country": "%{[geoip][country_name]}" }`

Example: Basic GeoIP Filter Configuration

{
  "filter": [
    {
      "geoip": {
        "source": "clientip",
        "target": "geoip",
        "database": "/usr/share/GeoIP/GeoLite2-City.mmdb"
      }
    }
  ]
}

Example: Adding Custom Fields with GeoIP Data

{
  "filter": [
    {
      "geoip": {
        "source": "clientip",
        "target": "geoip",
        "database": "/usr/share/GeoIP/GeoLite2-City.mmdb",
        "add_field": {
          "country_name": "%{[geoip][country_name]}",
          "city_name": "%{[geoip][city_name]}"
        }
      }
    }
  ]
}

Note: Ensure the GeoIP database file is accessible and up-to-date for accurate geolocation data.

8. Using User Agent Filter

Field	Description	Example
user_agent	Input field containing the user agent string to parse	`"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"`
target	Target field to store parsed user agent data (default: `user_agent`)	`"target": "parsed_ua"`

Example: Parsing User Agent String

{
  "filter": {
    "user_agent": {
      "source": "user_agent",
      "target": "parsed_ua"
    }
  }
}

Note: The User Agent filter uses the ua-parser library under the hood to parse user agent strings into structured data.

Output Fields

name: Browser name (e.g., "Chrome")
version: Browser version (e.g., "91.0.4472.124")
os.name: Operating system name (e.g., "Windows")
os.version: Operating system version (e.g., "10")
device.type: Device type (e.g., "desktop")

9. Using Drop Filter

Property	Values	Description
drop	`{}`	Drops events based on a condition

Example: Basic Drop Filter Usage

{<br>  "filter": [<br>    {<br>      "drop": {<br>        "if": "[status] == 200"<br>      }<br>    }<br>  ]<br>}

Note: The drop filter removes events entirely from the pipeline. Use with caution.

10. Using Clone Filter

Option	Values	Notes
clone	`array`	Array of clone configuration objects
clone { count }	`integer`	Number of times to clone the event (default: 1)
clone { add_field }	`hash`	Fields to add to cloned events

Example: Clone a Log Event

{
  "clone": {
    "count": 2,
    "add_field": {
      "cloned": "true"
    }
  }
}

Note: The original event is not modified. Each cloned event will have the new field added.

11. Using Split Filter

Option	Values	Notes
field	`message`	Field to split on
separator		Separator to split on (default: space)
target	`message_parts`	Target field to store split results (optional)

Example: Splitting a message field

{<br>  "filter": {<br>    "split": {<br>      "field": "message",<br>      "separator": " "<br>    }<br>  }<br>}

Note: The split filter is useful for parsing structured log messages where fields are separated by a known delimiter.

12. Using Aggregate Filter

Property	Values	Description
aggregate	`filter`	Filter type for aggregating events
code	`string`	JavaScript code to execute for aggregation logic
field	`string`	Field name to use as key for grouping events
timeout	`integer`	Timeout in seconds to flush the aggregate

Example: Aggregating Log Events by Session ID

filter {
  aggregate {
    code => "
      map['session_events'] ||= []
      map['session_events'].push(event.get('message'))
      if (event.get('is_end') == true) {
        event.set('session_logs', map['session_events'])
        map['session_events'] = []
      }
    "
    push_map_as_event => true
    timeout => 300
    field => "session_id"
  }
}

Note: The aggregate filter is useful for combining multiple log events into a single event based on a common field like session ID or transaction ID.