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