ELK Stack — Ingest Pipelines

Ingest Pipelines

An ingest pipeline is a sequence of processors that transform a document before it is indexed. Instead of preprocessing data in your application or in Logstash, you can define transformations directly in Elasticsearch.

---

Why Ingest Pipelines

• Transform data at index time without external tools.
• Parse unstructured log lines with grok or dissect.
• Enrich documents with geoip or user_agent data.
• Rename, remove, or convert fields to fit your mappings.
• Test transformations with the simulate API before going live.

---

Creating a Pipeline

Use the _ingest/pipeline API:

curl -s -X PUT "localhost:9200/_ingest/pipeline/my-pipeline" \
  -H "Content-Type: application/json" \
  -d '{
  "description": "Demo pipeline with basic processors",
  "processors": [
    {
      "set": {
        "field": "ingested_at",
        "value": "{{_ingest.timestamp}}"
      }
    },
    {
      "lowercase": {
        "field": "status"
      }
    }
  ]
}'

{
  "acknowledged": true
}

---

Processors Reference

set

Sets a field to a static or dynamic value:

curl -s -X PUT "localhost:9200/_ingest/pipeline/set-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "set": {
        "field": "env",
        "value": "production"
      }
    },
    {
      "set": {
        "field": "processed_at",
        "value": "{{_ingest.timestamp}}"
      }
    }
  ]
}'

{
  "acknowledged": true
}

rename

Renames a field:

curl -s -X PUT "localhost:9200/_ingest/pipeline/rename-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "rename": {
        "field": "hostname",
        "target_field": "host.name"
      }
    }
  ]
}'

{
  "acknowledged": true
}

remove

Removes one or more fields:

curl -s -X PUT "localhost:9200/_ingest/pipeline/remove-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "remove": {
        "field": ["temp_field", "debug_info"]
      }
    }
  ]
}'

{
  "acknowledged": true
}

convert

Converts a field to a different data type:

curl -s -X PUT "localhost:9200/_ingest/pipeline/convert-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "convert": {
        "field": "response_code",
        "type": "integer"
      }
    },
    {
      "convert": {
        "field": "bytes_sent",
        "type": "long"
      }
    }
  ]
}'

{
  "acknowledged": true
}

grok

Parses unstructured text using named capture patterns (like regex on steroids):

curl -s -X PUT "localhost:9200/_ingest/pipeline/grok-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IP:client_ip} - %{USER:user} \\[%{HTTPDATE:timestamp}\\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:bytes}"
        ]
      }
    }
  ]
}'

{
  "acknowledged": true
}

date

Parses a date from a field and sets @timestamp:

curl -s -X PUT "localhost:9200/_ingest/pipeline/date-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "date": {
        "field": "timestamp_str",
        "formats": ["dd/MMM/yyyy:HH:mm:ss Z", "ISO8601"],
        "target_field": "@timestamp"
      }
    }
  ]
}'

{
  "acknowledged": true
}

gsub

Replaces occurrences of a pattern in a string field:

curl -s -X PUT "localhost:9200/_ingest/pipeline/gsub-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "gsub": {
        "field": "message",
        "pattern": "\\n",
        "replacement": " "
      }
    }
  ]
}'

{
  "acknowledged": true
}

json

Parses a JSON string into an object:

curl -s -X PUT "localhost:9200/_ingest/pipeline/json-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "json": {
        "field": "payload",
        "target_field": "parsed_payload"
      }
    }
  ]
}'

{
  "acknowledged": true
}

dissect

Splits a string using a fixed pattern (faster than grok for structured formats):

curl -s -X PUT "localhost:9200/_ingest/pipeline/dissect-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "dissect": {
        "field": "message",
        "pattern": "%{client_ip} %{ident} %{auth} [%{timestamp}] \"%{verb} %{request} HTTP/%{httpversion}\" %{status} %{size}"
      }
    }
  ]
}'

{
  "acknowledged": true
}

script

Executes a Painless script:

curl -s -X PUT "localhost:9200/_ingest/pipeline/script-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "script": {
        "lang": "painless",
        "source": "ctx.full_name = ctx.first_name + \" \" + ctx.last_name"
      }
    }
  ]
}'

{
  "acknowledged": true
}

user_agent

Parses a user agent string into structured fields:

curl -s -X PUT "localhost:9200/_ingest/pipeline/ua-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "user_agent": {
        "field": "agent"
      }
    }
  ]
}'

{
  "acknowledged": true
}

geoip

Looks up geographic information from an IP address:

curl -s -X PUT "localhost:9200/_ingest/pipeline/geoip-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "geoip": {
        "field": "client_ip",
        "target_field": "geo"
      }
    }
  ]
}'

{
  "acknowledged": true
}

---

Simulate API

Test a pipeline without actually indexing any data:

curl -s -X POST "localhost:9200/_ingest/pipeline/grok-demo/_simulate?pretty" \
  -H "Content-Type: application/json" \
  -d '{
  "docs": [
    {
      "_source": {
        "message": "192.168.1.10 - frank [10/Oct/2025:13:55:36 -0700] \"GET /api/users HTTP/1.1\" 200 1234"
      }
    }
  ]
}'

{
  "docs": [
    {
      "doc": {
        "_source": {
          "message": "192.168.1.10 - frank [10/Oct/2025:13:55:36 -0700] \"GET /api/users HTTP/1.1\" 200 1234",
          "client_ip": "192.168.1.10",
          "user": "frank",
          "timestamp": "10/Oct/2025:13:55:36 -0700",
          "method": "GET",
          "request": "/api/users",
          "http_version": "1.1",
          "response_code": "200",
          "bytes": "1234"
        },
        "_ingest": {
          "timestamp": "2026-04-01T09:00:00.000Z"
        }
      }
    }
  ]
}

Simulate with an Inline Pipeline

You do not need to save a pipeline first — you can simulate inline:

curl -s -X POST "localhost:9200/_ingest/pipeline/_simulate?pretty" \
  -H "Content-Type: application/json" \
  -d '{
  "pipeline": {
    "processors": [
      { "uppercase": { "field": "name" } },
      { "set": { "field": "processed", "value": true } }
    ]
  },
  "docs": [
    { "_source": { "name": "alice" } },
    { "_source": { "name": "bob" } }
  ]
}'

{
  "docs": [
    {
      "doc": {
        "_source": { "name": "ALICE", "processed": true },
        "_ingest": { "timestamp": "2026-04-01T09:00:00.000Z" }
      }
    },
    {
      "doc": {
        "_source": { "name": "BOB", "processed": true },
        "_ingest": { "timestamp": "2026-04-01T09:00:00.000Z" }
      }
    }
  ]
}

---

Using Pipelines in Index Requests

Specify the pipeline in the query parameter:

curl -s -X POST "localhost:9200/my-index/_doc?pipeline=my-pipeline&pretty" \
  -H "Content-Type: application/json" \
  -d '{
  "status": "ERROR",
  "message": "Something went wrong"
}'

{
  "_index": "my-index",
  "_id": "xyz789",
  "_version": 1,
  "result": "created",
  "_shards": { "total": 2, "successful": 1, "failed": 0 }
}

---

Default Pipeline in Index Settings

Set a default pipeline so every document indexed into that index is automatically processed:

curl -s -X PUT "localhost:9200/server-logs" \
  -H "Content-Type: application/json" \
  -d '{
  "settings": {
    "default_pipeline": "grok-demo"
  }
}'

{
  "acknowledged": true,
  "shards_acknowledged": true,
  "index": "server-logs"
}

---

Pipeline in Index Templates

Apply a pipeline to all indices matching a pattern:

curl -s -X PUT "localhost:9200/_index_template/server-logs-template" \
  -H "Content-Type: application/json" \
  -d '{
  "index_patterns": ["server-logs-*"],
  "template": {
    "settings": {
      "default_pipeline": "grok-demo"
    }
  },
  "priority": 200
}'

{
  "acknowledged": true
}

---

Handling Failures with on_failure

When a processor fails, the document is rejected by default. Use on_failure to handle errors gracefully:

curl -s -X PUT "localhost:9200/_ingest/pipeline/safe-grok" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{IP:client_ip} %{WORD:method} %{URIPATHPARAM:path}"],
        "on_failure": [
          {
            "set": {
              "field": "grok_failed",
              "value": true
            }
          },
          {
            "set": {
              "field": "error_message",
              "value": "{{_ingest.on_failure_message}}"
            }
          }
        ]
      }
    }
  ]
}'

{
  "acknowledged": true
}

Now documents that fail grok parsing will still be indexed — with grok_failed: true and the error message.

---

Conditional Execution

Use the if parameter with a Painless condition to run a processor only when certain criteria are met:

curl -s -X PUT "localhost:9200/_ingest/pipeline/conditional-demo" \
  -H "Content-Type: application/json" \
  -d '{
  "processors": [
    {
      "uppercase": {
        "field": "severity",
        "if": "ctx.severity != null"
      }
    },
    {
      "set": {
        "field": "is_error",
        "value": true,
        "if": "ctx.response_code != null && ctx.response_code >= 500"
      }
    },
    {
      "remove": {
        "field": "debug_data",
        "if": "ctx.env == \"production\""
      }
    }
  ]
}'

{
  "acknowledged": true
}

Test Conditional Pipeline

curl -s -X POST "localhost:9200/_ingest/pipeline/conditional-demo/_simulate?pretty" \
  -H "Content-Type: application/json" \
  -d '{
  "docs": [
    {
      "_source": {
        "severity": "warning",
        "response_code": 503,
        "env": "production",
        "debug_data": "stack trace here"
      }
    }
  ]
}'

{
  "docs": [
    {
      "doc": {
        "_source": {
          "severity": "WARNING",
          "response_code": 503,
          "env": "production",
          "is_error": true
        },
        "_ingest": {
          "timestamp": "2026-04-01T09:00:00.000Z"
        }
      }
    }
  ]
}

severity was uppercased, is_error was set because response_code >= 500, and debug_data was removed because env is "production".

---

A Real-World Pipeline: Web Access Logs

This pipeline combines multiple processors to parse, enrich, and clean web access logs:

curl -s -X PUT "localhost:9200/_ingest/pipeline/web-access-logs" \
  -H "Content-Type: application/json" \
  -d '{
  "description": "Parse and enrich web access logs",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{IPORHOST:source.ip} - %{DATA:user.name} \\[%{HTTPDATE:event.timestamp}\\] \"%{WORD:http.method} %{DATA:url.path} HTTP/%{NUMBER:http.version}\" %{NUMBER:http.response.status_code} %{NUMBER:http.response.body.bytes} \"%{DATA:http.request.referrer}\" \"%{DATA:user_agent.original}\""
        ]
      }
    },
    {
      "date": {
        "field": "event.timestamp",
        "formats": ["dd/MMM/yyyy:HH:mm:ss Z"],
        "target_field": "@timestamp"
      }
    },
    {
      "convert": {
        "field": "http.response.status_code",
        "type": "integer"
      }
    },
    {
      "convert": {
        "field": "http.response.body.bytes",
        "type": "long"
      }
    },
    {
      "user_agent": {
        "field": "user_agent.original"
      }
    },
    {
      "geoip": {
        "field": "source.ip",
        "target_field": "source.geo",
        "if": "ctx.source?.ip != null && !ctx.source.ip.startsWith(\"192.168.\") && !ctx.source.ip.startsWith(\"10.\")"
      }
    },
    {
      "set": {
        "field": "event.kind",
        "value": "event"
      }
    },
    {
      "remove": {
        "field": ["message", "event.timestamp"],
        "if": "ctx.source?.ip != null"
      }
    }
  ],
  "on_failure": [
    {
      "set": {
        "field": "event.kind",
        "value": "pipeline_error"
      }
    },
    {
      "set": {
        "field": "error.message",
        "value": "{{_ingest.on_failure_message}}"
      }
    }
  ]
}'

{
  "acknowledged": true
}

---

Managing Pipelines

List All Pipelines

curl -s -X GET "localhost:9200/_ingest/pipeline?pretty"

Get a Specific Pipeline

curl -s -X GET "localhost:9200/_ingest/pipeline/web-access-logs?pretty"

Delete a Pipeline

curl -s -X DELETE "localhost:9200/_ingest/pipeline/my-pipeline"

{
  "acknowledged": true
}

---

Quick Reference

Task	Command
Create / update pipeline	PUT _ingest/pipeline/<name>
Get pipeline	GET _ingest/pipeline/<name>
Delete pipeline	DELETE _ingest/pipeline/<name>
List all pipelines	GET _ingest/pipeline
Simulate saved pipeline	POST _ingest/pipeline/<name>/_simulate
Simulate inline pipeline	POST _ingest/pipeline/_simulate
Index with pipeline	POST <index>/_doc?pipeline=<name>
Set default pipeline	Index setting default_pipeline
Handle failures	on_failure array on processor or pipeline
Conditional execution	if parameter with Painless expression

Processor	Purpose
set	Set a field value
rename	Rename a field
remove	Remove fields
convert	Change field type
grok	Parse with named regex patterns
dissect	Parse with fixed delimiter patterns
date	Parse dates
gsub	Regex replace in strings
json	Parse JSON strings
script	Run Painless scripts
user_agent	Parse UA strings
geoip	IP geolocation lookup

---

Next → Practice Exercises