Ingest Pipelines

ingest-pipelines

Skill authority

The rules and patterns defined in this skill and its reference files are the authoritative source of truth. When examining existing integrations in the elastic/integrations repository for reference, you may encounter patterns that conflict with what is specified here — many integrations contain legacy patterns that predate current standards. Always follow this skill over patterns observed in other integrations. If a reference integration uses a deprecated or prohibited pattern, do not copy it.

When to use

Use this skill when tasks include:

• building or modifying elasticsearch/ingest_pipeline/default.yml for a data stream
• choosing parser and normalization processors (grok, dissect, json, kv, date, convert)
• designing conditional branches and sub-pipeline routing with pipeline processors
• implementing resilient error handling with top-level on_failure
• tuning processor order for ingest performance and maintainability

When not to use

Do not use this skill as the primary guide for:

• ECS field selection, categorization values, and field mapping strategy (ecs-field-mappings)
• elastic-package command and stack lifecycle workflows (elastic-package-cli)
• test fixture authoring and expected output workflows (integration-testing → references/pipeline-testing.md)

Pipeline anatomy

In integration packages, ingest pipelines live under:

data_stream/<stream>/elasticsearch/ingest_pipeline/

Every stream usually has a default.yml with:

• description
• processors list
• optional pipeline-level on_failure

Keep default.yml readable and focused. Move large format-specific logic into sub-pipelines where needed.

ECS version

Set the pipeline ECS reference version explicitly at the top of processors (after any introductory processors you already use). Use 9.3.0 — do not pin an older ECS version. Exception: when the orchestrator specifies this is an entity data stream (event.kind: asset), use 9.5.0 instead — entity leaf fields (entity.attributes.*, entity.lifecycle.*, entity.relationships.*) do not exist at 9.3.0 and cause field is undefined build failures at that pin.

  - set:
      field: ecs.version
      tag: set_ecs_version
      value: '9.3.0'

Rename vs set (mapping to ECS)

When moving a value from a custom or vendor field into an ECS field, prefer the rename processor so the source field is removed and you avoid duplicate data. Use set with copy_from only when you must keep the source field or when rename is not applicable.

Processor tags

Every processor in the pipeline should have a tag (not only processors that can fail). Tags make failures and telemetry attributable to a specific step.

CEL-only opening processors (Agentless metadata and error-only documents)

For CEL-based integrations only, include these before the standard message → event.original handling when they apply:

• remove: drop Agentless metadata fields (organization, division, team) when all are strings, so they do not collide with ECS. Use ignore_missing: true and a conditional if.
• terminate: stop processing when the document is an error placeholder from the collector (ctx.error?.message != null && ctx.message == null && ctx.event?.original == null).

Non-CEL integrations (logs, syslog, filebeat-style inputs) must not copy this block blindly — those fields and error shapes are specific to the CEL/Agentless path. See the create-integration skill: the orchestrator must only expect this block when the data stream uses CEL input.

Standard opening: ECS, optional CEL block, JSE00001, then parse event.original

After the optional CEL-only processors, the pipeline should follow this shape. All parsing (json, csv, grok, etc.) runs on event.original. Never overwrite or mutate event.original in later processors — derive structured fields into other paths (for example json, _temp.*, ECS fields).

description: Parse <dataset> events.
processors:
  - set:
      field: ecs.version
      tag: set_ecs_version
      value: '9.3.0'

  # --- CEL input only (omit for log/syslog-only streams) ---
  - remove:
      field:
        - organization
        - division
        - team
      ignore_missing: true
      if: ctx.organization instanceof String && ctx.division instanceof String && ctx.team instanceof String
      tag: remove_agentless_tags
      description: >-
        Removes the fields added by Agentless as metadata,
        as they can collide with ECS fields.
  - terminate:
      tag: data_collection_error
      if: ctx.error?.message != null && ctx.message == null && ctx.event?.original == null
      description: error message set and no data to process.
  # --- end CEL-only ---

  - rename:
      field: message
      tag: rename_message_to_event_original
      target_field: event.original
      ignore_missing: true
      description: Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.
      if: ctx.event?.original == null
  - remove:
      field: message
      tag: remove_message
      ignore_missing: true
      description: The `message` field is no longer required if the document has an `event.original` field.
      if: ctx.event?.original != null

  # Parse (always read from event.original; do not modify event.original)
  - json:
      field: event.original
      target_field: json
      tag: parse_json
      if: ctx.event?.original != null

  # ... normalize, enrich, ECS categorization, cleanup ...

  - append:
      field: tags
      value: preserve_original_event
      allow_duplicates: false
      if: ctx.error?.message != null

on_failure:
  - append:
      field: error.message
      value: >-
        Processor '{{{ _ingest.on_failure_processor_type }}}'
        {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
        {{{/_ingest.on_failure_processor_tag}}}failed with message '{{{ _ingest.on_failure_message }}}'
  - set:
      field: event.kind
      tag: set_pipeline_error_to_event_kind
      value: pipeline_error
  - append:
      field: tags
      value: preserve_original_event
      allow_duplicates: false

Single-path pattern (linear pipeline)

Use this pattern when one parser flow handles all events. Combine the standard opening (ECS version, optional CEL-only block, JSE00001 rename/remove, parse from event.original without mutating it), middle processors with tags on every step, and the pipeline-level on_failure and conditional append for preserve_original_event shown above.

Example middle section (illustrative):

  - grok:
      field: event.original
      patterns:
        - '^...$'
      tag: parse_main

  - date:
      field: some.time
      target_field: '@timestamp'
      formats: [ISO8601]
      tag: parse_timestamp
  - convert:
      field: http.response.status_code
      type: long
      ignore_missing: true
      tag: convert_status

  - user_agent:
      field: user_agent.original
      ignore_missing: true
      tag: enrich_user_agent
  - geoip:
      field: source.ip
      target_field: source.geo
      if: ctx.source?.ip != null
      ignore_missing: true
      tag: enrich_source_geo
  - geoip:
      database_file: GeoLite2-ASN.mmdb
      field: source.ip
      target_field: source.as
      if: ctx.source?.ip != null
      properties:
        - asn
        - organization_name
      ignore_missing: true
      tag: enrich_source_asn
  - rename:
      field: source.as.asn
      target_field: source.as.number
      ignore_missing: true
      tag: rename_source_asn
  - rename:
      field: source.as.organization_name
      target_field: source.as.organization.name
      ignore_missing: true
      tag: rename_source_as_org

  - set:
      field: event.kind
      tag: set_event_kind
      value: event
  - append:
      field: event.category
      tag: append_event_category_web
      value: web
  - remove:
      field: temp
      ignore_missing: true
      tag: remove_temp

Branching pattern (router + sub-pipelines)

Use branching when event formats or object models diverge:

• format-based branching (for example JSON vs text)
• class/category-based branching (for example OCSF class/category routing)
• object-presence branching (ctx.ocsf.user != null)

Pattern:

processors:
  - pipeline:
      name: '{{ IngestPipeline "pipeline_branch_json" }}'
      if: ctx.event?.original != null && ctx.event.original.startsWith('{')
      ignore_missing_pipeline: true
      tag: route_json
  - pipeline:
      name: '{{ IngestPipeline "pipeline_branch_text" }}'
      if: ctx.event?.original != null && !ctx.event.original.startsWith('{')
      ignore_missing_pipeline: true
      tag: route_text

In large integrations, keep default.yml as the router and put branch logic in files like:

• pipeline_object_<name>.yml
• pipeline_category_<name>.yml

See references/branching-patterns.md for full patterns from amazon_security_lake.

Sub-pipeline routing for multi-log-type integrations

When a data stream receives multiple distinct log types (for example a firewall that emits traffic, auth, and DNS logs in the same stream), do not implement all parsing in a single monolithic default.yml. Use default.yml as a thin router that detects the log type and delegates to a dedicated sub-pipeline per type.

File layout

elasticsearch/ingest_pipeline/
  default.yml              # router only — detects log type, calls sub-pipelines
  pipeline-<type>.yml      # one file per log type (e.g. pipeline-traffic.yml)

Router pattern in default.yml

Use the same ecs.version, JSE00001 rename/remove pair for message, and full pipeline-level on_failure as in the standard opening. The router only branches sub-pipelines; it does not parse payloads.

processors:
  - set:
      field: ecs.version
      tag: set_ecs_version
      value: '9.3.0'
  - rename:
      field: message
      tag: rename_message_to_event_original
      target_field: event.original
      ignore_missing: true
      if: ctx.event?.original == null
  - remove:
      field: message
      tag: remove_message
      ignore_missing: true
      if: ctx.event?.original != null
  - pipeline:
      name: '{{ IngestPipeline "pipeline-traffic" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("TRAFFIC")'
      tag: route_traffic
  - pipeline:
      name: '{{ IngestPipeline "pipeline-auth" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("AUTH")'
      tag: route_auth
  - pipeline:
      name: '{{ IngestPipeline "pipeline-dns" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("DNS")'
      tag: route_dns
on_failure:
  - append:
      field: error.message
      value: >-
        Processor '{{{ _ingest.on_failure_processor_type }}}'
        {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
        {{{/_ingest.on_failure_processor_tag}}}failed with message '{{{ _ingest.on_failure_message }}}'
  - set:
      field: event.kind
      tag: set_pipeline_error_to_event_kind
      value: pipeline_error
  - append:
      field: tags
      value: preserve_original_event
      allow_duplicates: false

Rules

• default.yml must contain only routing logic and on_failure handling — no field parsing.
• Each sub-pipeline handles parsing, ECS mapping, and categorization for its own log type.
• Each sub-pipeline must have its own on_failure block.
• Name sub-pipeline files pipeline-<type>.yml where <type> matches the log type identifier used in the routing condition.
• Each log type gets its own pipeline test fixture file following the naming convention test-<package>-<datastream>-<type>-sample.log.

Processor ordering and performance

• run cheap existence checks before expensive operations
• drop early if records are out of scope
• prefer dissect over grok for stable delimited formats
• never use a script processor when a built-in processor can do the job — set, rename, remove, append, convert, dissect, grok, gsub, lowercase, uppercase, and trim are all faster than Painless and easier to review. See the cost tiers in references/processor-cookbook.md → Processor performance guide.
• use enrichment processors (geoip, user_agent) only when needed
• always anchor grok patterns with ^ and $ — without anchors the regex engine scans the entire input string looking for a partial match, which is slow and can produce incorrect results on noisy log lines

Mustache template syntax in processor values

Ingest pipeline processors use Mustache templates to reference field values in value, message, and similar string parameters. Use triple braces {{{field}}} with single quotes — never double braces or double quotes:

# CORRECT — triple braces, single quotes
- append:
    field: related.user
    value: '{{{user.target.email}}}'
    allow_duplicates: false
    if: ctx.user?.target?.email != null

# WRONG — double braces HTML-escape the value; double quotes
- append:
    field: related.user
    value: "{{user.target.email}}"
    allow_duplicates: false
    if: ctx.user?.target?.email != null

Why: Mustache double braces {{...}} HTML-encode the value (e.g., & becomes &), which corrupts data in ingest pipelines. Triple braces {{{...}}} emit the raw value. Single quotes prevent YAML from interpreting braces.

Exception: {{ IngestPipeline "..." }} in pipeline.name is a Go template directive processed at build time, not a Mustache template — it correctly uses double braces.

Error handling essentials

Use pipeline-level on_failure as the main error reporting mechanism.

Recommended baseline (order matters):

• append contextual error.message first using _ingest.on_failure_* variables (full template in the standard opening example)
• set event.kind: pipeline_error (with a tag on the set processor)
• append preserve_original_event to tags when you need to retain the failed document for triage
• give every processor a tag (not only processors that can fail)

Use processor-level on_failure for local cleanup or fallback parsing, not as the primary global error message path.

See references/error-handling-patterns.md for full examples and tradeoffs (ignore_failure, fail, processor-level on_failure).

event.original handling (JSE00001)

The elastic-package build validator enforces that pipelines correctly handle the message to event.original rename. This check is known as JSE00001. New packages must comply; some legacy packages exclude it via validation.yml.

Required two-processor pattern

Every pipeline that consumes a message field must include both processors (typically after ecs.version and after any CEL-only remove/terminate steps when applicable):

- rename:
    field: message
    tag: rename_message_to_event_original
    target_field: event.original
    ignore_missing: true
    description: Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.
    if: ctx.event?.original == null
- remove:
    field: message
    tag: remove_message
    ignore_missing: true
    description: The `message` field is no longer required if the document has an `event.original` field.
    if: ctx.event?.original != null

Step 1 (rename): moves message into event.original, but only when event.original is not already populated (idempotent when a prior pipeline or Logstash has already set it).

Step 2 (remove): removes the redundant message field when event.original is present (after rename or from an upstream producer).

Do NOT add an event.original removal processor at the end of the pipeline

Some existing integrations contain a remove processor that deletes event.original at the end of the pipeline when preserve_original_event is not in tags. This pattern is deprecated and must not be used in new pipelines. The removal of event.original for storage optimization is now handled by a separate final pipeline outside the integration. Do not copy this pattern from reference integrations that still have it — it is legacy.

That stack-side final pipeline honors the preserve_original_event tag, so a package exposing the toggle needs NO in-pipeline remove for the toggle to work — the toggle is not a no-op without one. Never recommend adding this processor; when reviewing, its presence is the finding, never its absence.

Reference

The two-processor JSE00001 pattern (rename + remove of message) shown above is required and complete. Do not add any additional event.original processors beyond those two.

Timezone handling (tz_offset)

For data streams that include the tz_offset manifest var (syslog streams where messages lack a timezone), set event.timezone from _conf.tz_offset early in the pipeline, before any date parsing:

- set:
    field: event.timezone
    tag: set_event_timezone
    value: '{{{_conf.tz_offset}}}'
    if: ctx._conf?.tz_offset != null && ctx._conf.tz_offset != ''

This ensures date processors can apply the correct timezone when parsing timestamps that have no timezone component.

Syslog structured data (RFC 5424 SD-ELEMENT) parsing

For vendor key=value payloads and RFC 5424 SD-ELEMENT blocks, three strategies are available: KV with trim_value (simplest, Strategy 1), SYSLOG5424SD grok + KV with regex splits (Strategy 2), and Painless for edge cases with embedded equals or mixed quoting (Strategy 3).

Prefer Strategy 1 or 2; use Painless only when KV edge cases demand it.

See references/grok-recipes.md → Syslog structured data strategies for full code examples, key settings, and reference implementations.

Keyword fields delivered as numbers

Fields that carry identifiers, protocol codes, or other opaque values must be declared as keyword in fields.yml — even when the source data delivers them as numbers. Common examples:

• network protocol numbers (network.iana_number)
• port numbers used as identifiers
• error codes, result codes, status codes
• SNMP OIDs, event IDs, object class codes

Do not add a convert processor to stringify these values. Elasticsearch silently coerces numbers into keyword strings at index time, so the pipeline can pass the raw numeric value through unchanged.

The field declaration in fields.yml:

- name: network.iana_number
  type: keyword
  description: IANA protocol number.

Because the test runner compares raw value types against declared field types, it will flag 6 (long) as a mismatch for keyword. Declare the field in numeric_keyword_fields in the pipeline test config so the runner accepts the numeric representation without requiring the fixture to artificially stringify the value. See integration-testing/references/pipeline-testing.md for the config syntax.

Vendor field naming

Preserve vendor field names exactly as they appear in the source. Do not rename, reformat, or normalize vendor-specific field names — the only permitted renaming is mapping a vendor field to an ECS field (e.g. renaming src_ip to source.ip). When a vendor field has no ECS equivalent, keep it under a vendor-namespaced prefix (e.g. vendor.product.field_name) using the original name from the source.

related.ip population

Every IP address present in the document must be appended to related.ip. This includes source, destination, client, server, host, and any other IP fields — whatever applies to the event type.

Use one append processor per IP field, with ignore_missing: true so it is a no-op when the field is absent. Place these processors after all IP fields have been set (for example after geoip, convert, and any ECS rename steps) and before the cleanup remove processors.

  - append:
      field: related.ip
      tag: append_source_ip_to_related
      value: '{{{source.ip}}}'
      allow_duplicates: false
      if: ctx.source?.ip != null
  - append:
      field: related.ip
      tag: append_destination_ip_to_related
      value: '{{{destination.ip}}}'
      allow_duplicates: false
      if: ctx.destination?.ip != null
  # repeat the same pattern for client.ip, server.ip, host.ip, and any other IP fields the pipeline sets

Rules:

• Use allow_duplicates: false on every append to avoid repeated values.
• Add an if guard on every processor so it skips fields absent in the event.
• Add one append per IP field the pipeline actually writes — do not add processors for fields the pipeline never sets.

Painless script best practices

Before writing any script processor, you MUST check whether a built-in processor can do the same job. script is the slowest general-purpose processor (Painless compilation + per-document execution). The following operations have dedicated processors that are cheaper and easier to review:

Only reach for script when no combination of built-in processors can express the logic — for example, ECS categorization lookup tables with 5+ entries (Pattern A), complex conditional arithmetic, or edge-case string parsing that dissect and grok genuinely cannot handle.

Case-insensitive comparisons — use equalsIgnoreCase() when casing is unpredictable

Syslog and vendor devices are often inconsistent about casing, so Painless scripts comparing vendor-specific free-text fields should use equalsIgnoreCase() rather than ==. However, apply this judgement contextually, not blanket:

• Use equalsIgnoreCase() when the vendor field value may vary in casing between devices, firmware versions, or log sources (e.g. action fields like allow/Allow/ALLOW, severity strings, free-text status fields).
• Use == when the API or spec defines a fixed lowercase enum and the values are always delivered as-specified (e.g. ECS categorization fields, API response fields documented as lowercase-only enums). Adding equalsIgnoreCase() to fixed-enum fields adds noise without value.

// Correct for unpredictable vendor casing
if (ctx.vendor?.action?.equalsIgnoreCase('allow')) { ... }

// Correct for a fixed lowercase API enum — == is appropriate here
if (ctx.json?.event_type == 'login') { ... }

// Incorrect for unpredictable casing — breaks on "Allow", "ALLOW"
if (ctx.vendor?.action == 'allow') { ... }

Access ctx directly in script bodies — no null-safe operators

In script processor source blocks, access ctx fields directly. Use explicit null checks instead of the null-safe ?. operator.

// Correct — direct access with explicit null check
if (ctx.source != null && ctx.source.ip != null) { ... }

// Incorrect — null-safe operator in a script body
if (ctx.source?.ip != null) { ... }

Note: null-safe ?. is acceptable in processor if conditions (YAML), which are a different Painless execution context:

- append:
    field: related.ip
    value: '{{{source.ip}}}'
    if: ctx.source?.ip != null

Other rules

• Every script processor must have a tag and a description.
• Keep scripts short and scoped — move complex logic into helper variables inside the script, not across multiple script processors.
• Do not use script when built-in processors suffice — see the mandatory checklist table at the top of this section.

ECS categorization mapping

When mapping source event types or actions to event.category, event.type, event.outcome, and event.action, use the patterns in references/processor-cookbook.md → ECS categorization mapping patterns:

• Pattern A (script with params lookup table): recommended for 5+ mappings. Mapping data in params enables Painless compilation caching and keeps the script body generic.
• Pattern B (set processors with conditionals): for fewer than 5 mappings where a script is overkill.
• Pattern C (sub-pipeline): for 100+ mappings, extract the categorization into a dedicated sub-pipeline file.

Do NOT use bulk append processors (2 per event type = 50+ processors for 25 types) or inline Painless if/else chains without params (defeats compilation caching). These are explicit anti-patterns — see the cookbook for details.

Grok best practices

• prefer dissect when structure is fixed
• use simpler grok patterns where possible
• always anchor grok patterns with ^ and $:

  # Correct — anchored, fails fast on non-matching lines
  patterns:
    - '^%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}$'
  
  # Incorrect — unanchored, scans the whole string for a partial match
  patterns:
    - '%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}'
  

• avoid unnecessary backtracking-heavy custom regex
• add a tag to every grok (and every other) processor

For grok syntax (three expression forms, inline regex, type coercion, pattern_definitions), syslog header splitting recipes, and common mistakes, see references/grok-recipes.md.

Prohibited patterns

These patterns exist in many legacy integrations but must not be used in new or updated pipelines. Do not copy them from reference integrations.

Never set event.ingested

The event.ingested field is managed by Elasticsearch outside the integration pipeline. Do not add a set processor for event.ingested in any integration pipeline. This includes patterns like:

# PROHIBITED — do not use
- set:
    field: event.ingested
    value: '{{{_ingest.timestamp}}}'

The pipeline should set @timestamp from the original event's timestamp. When the source data contains multiple timestamps, map them as follows:

• @timestamp: the primary event timestamp parsed from the source data. This is required.
• event.created: when the event was first created or recorded by the source system (if different from @timestamp).
• event.start: when an activity or period began (e.g., session start, connection start).
• event.end: when an activity or period ended (e.g., session end, connection close).

If a source timestamp does not match the semantics of event.created, event.start, or event.end, map it to a custom field under the vendor namespace with type: date in fields.yml and use a date processor with the appropriate target_field.

Never use preserve_duplicate_custom_fields

The preserve_duplicate_custom_fields tag pattern — where source fields are copied to ECS fields using set with copy_from and the originals are conditionally retained — is a legacy anti-pattern. Do not use it in any new or updated pipeline. Do not add a preserve_duplicate_custom_fields manifest variable, tag, or conditional logic.

Instead, follow these field mapping rules:

• When a source field maps to an ECS field, use rename to move it directly. The source field is removed and no duplicate exists.
• When a type conversion is needed (e.g., string to date, string to long), use the appropriate processor (date, convert, set with copy_from) to populate the ECS target field, then remove the source field in the cleanup section at the end of the pipeline.
• Never design a pipeline that needs to preserve both the original vendor field and the ECS copy. The ECS field is the canonical location.

If you encounter this pattern in a reference integration, ignore it — it is legacy.

Never add an event.original removal processor at the end

As documented in the JSE00001 section above: do not add a remove processor for event.original at the end of the pipeline. This is handled by a separate final pipeline.

References

• references/processor-cookbook.md — processor selection, parsing/normalization/enrichment examples, ECS categorization mapping patterns (Pattern A/B/C + anti-patterns)
• references/branching-patterns.md
• references/error-handling-patterns.md
• references/grok-recipes.md — grok syntax, type coercion, syslog header recipes, common mistakes, pattern library link
• references/builder-subagent-guidance.md — subagent operating manual: scope boundaries, skill-load sequence, input data paths (CEL-first vs Direct), 9-step pipeline build workflow, "review generated output, never hand-edit expected JSON", reporting contract. The orchestrator dispatches subagents by passing this file's path in the task prompt; the subagent reads it itself in its own fresh context. Do NOT embed/paste its contents into the task prompt.
• references/cdr-pipeline-requirements.md — cloud security / CDR integrations only (cloudsecurity_cdr category): event categorization, Must Have / Should Have fields, processor patterns, review checklist
• entity-mappings/references/entity-pipeline-patterns.md — entity/inventory data streams only (event.kind: asset): categorization processors, entity.id mirroring, boolean coercion, relationship objects, anti-patterns, pipeline review checklist