Using Ingest Pipelines

1. Understanding Ingest Node

Property	Description	Default Value	Notes
node.name	Name of the ingest node	elasticsearch	Used for identification in logs and monitoring
cluster.name	Name of the cluster this node belongs to	elasticsearch	Must match across all nodes in the cluster
network.host	Network interface to bind to	localhost	Can be set to 0.0.0.0 to bind to all interfaces
http.port	HTTP port for API access	9200	Can be changed to avoid conflicts with other services
transport.port	Port for inter-node communication	9300	Must be accessible from other nodes in the cluster

Example: Basic Ingest Node Configuration

node.name: ingest-node-1
cluster.name: elk-cluster
network.host: 0.0.0.0
http.port: 9200
transport.port: 9300

Note: Ingest nodes are designed to process and transform data before indexing it into Elasticsearch. They do not store data themselves.

Warning: Ensure that the network.host setting allows communication with other nodes in the cluster to prevent connectivity issues.

2. Creating Ingest Pipelines

Ingest pipelines in the ELK Stack allow you to process and transform data as it is ingested into Elasticsearch. They are particularly useful for enriching, parsing, or modifying documents before they are indexed.

Field	Description	Example
name	Name of the ingest pipeline	`my_pipeline`
description	Optional description of the pipeline	`Processes log data`
processors	List of processors to apply in order	`[{ "grok": { ... } }, { "set": { ... } }]`
on_failure	Optional array of processors to run if a processor fails	`[{ "set": { ... } }]`

Example: Creating a Basic Ingest Pipeline

{
  "name": "log-processing-pipeline",
  "description": "Processes log entries to extract timestamp and user info",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{TIMESTAMP_ISO8601:timestamp}"]
      }
    },
    {
      "set": {
        "field": "processed",
        "value": true
      }
    }
  ]
}

Note: The processors array defines the order in which transformations are applied. Each processor can modify or enrich the document before passing it to the next one.

Warning: If a processor in the pipeline fails, the entire document will be rejected unless on_failure is configured.

Key Points

Pipelines are defined using JSON.
Processors run in the order specified in the processors array.
You can define fallback processors using on_failure.
Pipelines are applied at ingest time, before documents are indexed.

3. Using Common Processors

Processor	Description	Configuration Example	Notes
grok	Extracts structured fields from unstructured text using regular expressions or predefined patterns.	`{ "grok": { "field": "message", "patterns": ["%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}"] } }`	Supports predefined patterns like `HTTPD1`, `JSON`, etc.
date	Parses date strings into a standardized date format.	`{ "date": { "field": "timestamp", "target_field": "parsed_timestamp", "formats": ["yyyy-MM-dd HH:mm:ss"] } }`	Supports multiple format strings and timezone handling.
gsub	Applies a regular expression replacement to a field value.	`{ "gsub": { "field": "message", "pattern": "ERROR", "replacement": "WARNING" } }`	Useful for normalizing log levels or cleaning up field values.
remove	Removes fields from the document.	`{ "remove": { "field": "temp_field" } }`	Can remove multiple fields by providing an array.
rename	Renames a field in the document.	`{ "rename": { "field": "old_name", "target_field": "new_name" } }`	Useful for standardizing field names across logs.
set	Sets a field to a specified value.	`{ "set": { "field": "source_type", "value": "application" } }`	Can set nested fields using dot notation.

Example: Grok Processor Usage

{
  "description": "Parse Apache logs using grok",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{COMBINED_LOG_FORMAT}"],
        "pattern_definitions": {
          "COMBINED_LOG_FORMAT": "%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \\[%{HTTPDATE:timestamp}\\] \"(?:%{METHOD:method} %{NOTSPACE:url} %{NOTSPACE:protocol}|%{NOTEMPTY:rawrequest})\" %{NUMBER:response} (?:%{NUMBER:bytes}|-)"
        }
      }
    },
    {
      "date": {
        "field": "timestamp",
        "target_field": "parsed_timestamp",
        "formats": ["dd/MMM/yyyy:HH:mm:ss Z"]
      }
    }
  ]
}

Example: Date Processor Usage

{
  "description": "Convert timestamp to ISO8601 format",
  "processors": [
    {
      "date": {
        "field": "log_timestamp",
        "target_field": "timestamp",
        "formats": ["yyyy-MM-dd HH:mm:ss.SSS"],
        "timezone": "UTC"
      }
    }
  ]
}

Note: Processors are executed in the order they appear in the pipeline. Ensure that field transformations occur before dependent operations.

4. Using Grok Processor

Property	Values	Description
field	`message`	The field to process using the Grok pattern
patterns	`[%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}]`	Array of Grok patterns to match against the field
pattern_definitions	`{ "TIMESTAMP_ISO8601": "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}T%{HOUR}: %{MINUTE}: %{SECOND}(?:%{TZ})?" }`	Custom pattern definitions to use in the Grok expression
target_field	`parsed`	Field to store the parsed results in

Example: Parsing Log Messages with Grok

{
  "pipeline": {
    "description": "Parse log messages using Grok",
    "processors": [
      {
        "grok": {
          "field": "message",
          "patterns": [
            "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}"
          ],
          "pattern_definitions": {
            "TIMESTAMP_ISO8601": "%{YEAR}-%{MONTHNUM}-%{MONTHDAY}T%{HOUR}: %{MINUTE}: %{SECOND}(?:%{TZ})?"
          },
          "target_field": "parsed"
        }
      }
    ]
  }
}

Note: Grok patterns are powerful for parsing unstructured log data into structured fields. The target_field allows you to store parsed results in a nested object or flat field.

Warning: Complex Grok patterns can impact ingest performance. Use pattern_definitions to define reusable sub-patterns for better maintainability.

Grok Processor Summary

Used to parse unstructured log data into structured fields
Supports custom pattern definitions for reuse
Can target nested fields using dot notation
Performance impact increases with complex patterns

5. Using Date Processor

Property	Description	Default	Notes
field	Input field containing the date string to parse	-	Mandatory
target_field	Field to write the parsed date to	field	If not specified, overwrites input field
formats	List of date formats to try parsing	["ISO8601", "UNIX", "UNIX_MS"]	Supports Java date format patterns
timezone	Timezone to use for parsing	UTC	Can be specified as timezone ID or offset
locale	Locale to use for parsing (e.g., "en_US")	en_US	Used for parsing localized date strings

Example: Basic Date Parsing

{
  "description": "Parse a date field using default formats",
  "processors": [
    {
      "date": {
        "field": "timestamp",
        "target_field": "parsed_timestamp"
      }
    }
  ]
}

Example: Custom Date Formats

{
  "description": "Parse date using custom format",
  "processors": [
    {
      "date": {
        "field": "log_date",
        "target_field": "parsed_date",
        "formats": ["dd/MM/yyyy HH:mm:ss"],
        "timezone": "America/New_York"
      }
    }
  ]
}

Note: The date processor supports multiple format patterns and will try each one in order until a match is found.

6. Using GeoIP Processor

Property	Description	Default	Notes
field	Source field containing IP address to process	`ip`
target	Destination field to store results	`geoip`
database_file	Path to GeoIP2 database file	`/usr/share/elasticsearch/plugins/ingest-geoip/GeoLite2-City.mmdb`	Must be accessible to Elasticsearch node
ignore_missing	Whether to skip processing if source field is missing	`false`	Set to `true` to avoid errors on missing IPs
first_only	Whether to use only the first IP address if field is an array	`true`	Use `false` to process all IPs in array

Example: Basic GeoIP Processor Configuration

{
  "name": "geoip-pipeline",
  "description": "Add geoip information to log entries",
  "processors": [
    {
      "geoip": {
        "field": "client_ip",
        "target": "geo_location",
        "database_file": "/usr/share/elasticsearch/plugins/ingest-geoip/GeoLite2-City.mmdb"
      }
    }
  ]
}

Example: Using GeoIP with Custom Target Field

{
  "geoip": {
    "field": "source.ip",
    "target": "location",
    "database_file": "/opt/elasticsearch/GeoLite2-City.mmdb",
    "ignore_missing": true
  }
}

Note: The GeoIP processor requires the ingest-geoip plugin to be installed in Elasticsearch.

Warning: Ensure the GeoIP database file is updated regularly to maintain accuracy of location data.

7. Using User Agent Processor

Field	Description	Default
field	Input field containing the user agent string	`user_agent.original`
target	Output field to store parsed user agent data	`user_agent`
ignore_missing	If true, skip processing if the input field is missing	`false`

Example: Parsing User Agent String

{
  "user_agent": {
    "original": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
  }
}

Example: Ingest Pipeline Configuration

{
  "description": "Parse user agent strings",
  "processors": [
    {
      "user_agent": {
        "field": "user_agent.original",
        "target": "user_agent"
      }
    }
  ]
}

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

Parsed User Agent Fields

name: Browser name
major: Major version number
minor: Minor version number
patch: Patch version number
os.name: Operating system name
os.major: OS major version
os.minor: OS minor version
device.name: Device type (e.g., "Desktop", "Mobile")

8. Using Script Processor

Property	Description	Values
id	Unique identifier for the processor	`"script_processor"`
description	Human-readable description of the processor	`"Process documents with custom script"`
if	Condition to evaluate before running the processor	`"ctx.status == 'active'"`
script	Script to execute on each document	`{ "lang": "painless", "source": "ctx.processed = true;" }`
on_error	Action to take if script fails	`"continue"` or `"stop"`

Example: Applying a Script Processor

{
  "ingest": {
    "pipeline": {
      "id": "my_pipeline",
      "processors": [
        {
          "script": {
            "lang": "painless",
            "source": "ctx.timestamp = new Date().toISOString();"
          }
        }
      ]
    }
  }
}

Note: The painless language is a secure scripting language built into Elasticsearch.

Warning: Scripts can impact performance. Avoid heavy computations or loops in scripts.

9. Handling Processor Failures

Note: This section covers how to configure and manage failures in Ingest Pipeline processors within the ELK Stack.

Processor Type	Failure Behavior	Configuration Option	Description
grok	Skips event if pattern doesn't match	`ignore_missing`	If true, skips event if field is missing
grok	Throws exception if pattern fails	`on_failure`	Defines action when processor fails (e.g., `retry`, `drop`)
date	Skips event if date parsing fails	`on_failure`	Allows defining fallback behavior for date parsing errors
script	Stops pipeline execution on error	`on_failure`	Can be used to define custom error handling logic

Example: Configuring `on_failure` in a Pipeline

{
  "description": "Pipeline to process logs with error handling",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{GREEDYDATA:message}"],
        "on_failure": [
          {
            "set": {
              "field": "error",
              "value": "Failed to parse log line"
            }
          },
          {
            "drop": {
              "tag": "failed_log_line"
            }
          }
        ]
      }
    }
  ]
}

Failure Handling Strategies

Drop: Remove the event entirely from the pipeline
Set: Add metadata or error tags to the event
Retry: Re-process the event (if supported by processor)
Log: Log error to Elasticsearch or external system

Warning: Improperly configured on_failure actions can lead to infinite loops or unexpected data loss. Always test pipelines with sample data before deploying.

10. Testing Pipelines

Example: Test Pipeline with Sample Document

{
  "pipeline": {
    "id": "my-test-pipeline",
    "description": "Test pipeline for log processing",
    "processors": [
      {
        "grok": {
          "field": "message",
          "patterns": ["%{TIMESTAMP_ISO8601:timestamp}"]
        }
      },
      {
        "set": {
          "field": "processed",
          "value": true
        }
      }
    ]
  },
  "docs": [
    {
      "_id": "test-doc-1",
      "_source": {
        "message": "2025-04-05T12:34:56.789Z INFO This is a test log"
      }
    }
  ]
}

Field	Description	Example Value
pipeline.id	Unique identifier for the pipeline	`my-test-pipeline`
pipeline.description	Human-readable description of the pipeline	`Test pipeline for log processing`
pipeline.processors	List of processors to apply	`[{"grok": {...}}, {"set": {...}}]`
docs	Array of documents to process	`[{"_id": "test-doc-1", "_source": {...}}]`

Note: The test endpoint returns the processed documents with their updated fields, allowing you to verify pipeline logic before applying it to real data.

Warning: Ensure that the pipeline does not modify sensitive fields or introduce security vulnerabilities during testing.

Testing Pipeline Steps

Create a test pipeline configuration with sample documents
Use the /_ingest/pipeline/_simulate API to test
Review the output to verify transformations
Iterate on pipeline logic as needed

11. Attaching Pipelines to Indices

Attach ingest pipelines to indices using the index.lifecycle.name setting or directly on the index creation.

Setting	Value	Description
index.lifecycle.name	`pipeline-name`	Assigns a lifecycle policy that includes the pipeline
index.ingest.pipeline	`pipeline-name`	Directly assigns an ingest pipeline to the index

Example: Attaching Pipeline to Index

{
  "settings": {
    "index": {
      "ingest": {
        "pipeline": "my-pipeline"
      }
    }
  }
}

Note: The pipeline must exist before attaching it to an index.

12. Managing Default Pipeline

Property	Value	Description
pipeline.id	`default`	Identifier for the default ingest pipeline
pipeline.description	`Default pipeline for processing logs`	Description of the default pipeline's purpose
pipeline.processors	`[{ ... }]`	Array of processors to apply to each document

Example: Setting Default Pipeline in Logstash

{
  "pipeline": {
    "id": "default",
    "description": "Default pipeline for processing logs",
    "processors": [
      {
        "grok": {
          "match": {
            "message": "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{MESSAGE:message}"
          }
        }
      }
    ]
  }
}