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.

  - 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
      ignore_missing: true
      tag: enrich_source_geo
  - geoip:
      database_file: GeoLite2-ASN.mmdb
      field: source.ip
      target_field: source.as
      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.

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.

branching patterns for ingest pipelines

Use this guide when one linear parser is not enough and you need conditional or staged routing.

When to branch

Branch when at least one of these is true:

• the same stream receives multiple formats (for example JSON and plain text)
• different event classes need different object mapping logic
• complex parsing becomes difficult to review in one large default.yml
• array items require per-element pipeline processing (foreach + pipeline)

Keep single-path pipelines for simple, uniform formats.

Branching primitives

pipeline processor

- pipeline:
    name: '{{ IngestPipeline "pipeline_branch_name" }}'
    if: ctx.some?.field != null
    ignore_missing_pipeline: true
    tag: route_branch_name

Use:

• name via {{ IngestPipeline "..." }} for package-aware pipeline naming
• if guard to route selectively
• ignore_missing_pipeline: true if branch presence may vary by package version
• tag for failure diagnostics

foreach + pipeline for arrays

- foreach:
    field: ocsf.resources
    ignore_missing: true
    processor:
      pipeline:
        name: '{{ IngestPipeline "pipeline_resources_data_json" }}'

Use this when each item in an array needs repeated parse/normalize logic.

Naming conventions

Recommended conventions:

• default.yml as orchestrator/router
• pipeline_parser_<format>.yml for format parsers
• pipeline_object_<object>.yml for object mapping
• pipeline_category_<category>.yml for category-level transforms
• pipeline_enrichment_<topic>.yml for enrichment-only branches

Naming should describe branch intent, not source implementation detail.

Common branching topologies

1) Two-way format split

Use when input may be JSON or text:

- pipeline:
    name: '{{ IngestPipeline "pipeline_parser_json" }}'
    if: ctx.event?.original != null && ctx.event.original.startsWith('{')
    ignore_missing_pipeline: true
    tag: route_parser_json
- pipeline:
    name: '{{ IngestPipeline "pipeline_parser_text" }}'
    if: ctx.event?.original != null && !ctx.event.original.startsWith('{')
    ignore_missing_pipeline: true
    tag: route_parser_text

2) Category fan-out

Use when event category determines transform behavior:

- pipeline:
    name: '{{ IngestPipeline "pipeline_category_system_activity" }}'
    if: ctx.ocsf?.category_uid == '1'
    ignore_missing_pipeline: true
    tag: route_category_system_activity
- pipeline:
    name: '{{ IngestPipeline "pipeline_category_network_activity" }}'
    if: ctx.ocsf?.category_uid == '4'
    ignore_missing_pipeline: true
    tag: route_category_network_activity

3) Object-based branch with class guard

Use when only some classes contain a specific object:

- pipeline:
    name: '{{ IngestPipeline "pipeline_object_user" }}'
    if: ctx.ocsf?.class_uid != null && ['2005','3001','3002','3003'].contains(ctx.ocsf.class_uid) && ctx.ocsf.user != null
    ignore_missing_pipeline: true
    tag: route_object_user

4) Multi-branch graph (large integration pattern)

This is the pattern used by amazon_security_lake:

• default.yml does common parse/normalization
• object pipelines run conditionally by class_uid + object presence
• category pipelines run by category_uid
• some branches apply nested foreach processing

This yields maintainable sub-pipelines instead of one very large file.

Design rules

• Keep default.yml focused on shared setup and routing.
• Ensure each sub-pipeline is safe to run independently (with guards).
• Prefer mutually intelligible branch conditions over deeply nested script logic.
• If multiple branches can run on one event, make processor side effects explicit.
• Add description in every pipeline file so intent is obvious in reviews.

Validation checklist

• All routed sub-pipeline names resolve correctly through IngestPipeline.
• Branch conditions are null-safe (ctx.a?.b != null style).
• Branches do not overwrite each other unexpectedly.
• Pipeline tests include one fixture per route and one route-miss fixture.

Ingest pipeline builder subagent guidance

Operating manual for a subagent building or fixing ingest pipelines on behalf of the create-integration or maintain-integration orchestrator.

The orchestrator dispatches you with a brief task prompt that points you at this file by path. Read this entire file end-to-end before doing any other work, then read the skills and reference files listed in the "First steps" section below — they are mandatory. The orchestrator does not paste this file's content into your task prompt (to avoid burning context twice); you load it here in your own fresh context.

The orchestrator's task prompt tells you what to build or fix, which data stream to work on, what sample data is available, and the package path. This file tells you how to operate as a pipeline builder subagent. Follow both.

Scope

Your responsibility is strictly limited to:

• Designing and implementing the ingest pipeline (elasticsearch/ingest_pipeline/)
• Defining field mappings (fields/fields.yml, fields/base-fields.yml, fields/ecs.yml, fields/beats.yml)
• Creating pipeline test fixtures (_dev/test/pipeline/) and running pipeline tests

You do NOT:

• Run system tests (elastic-package test system) — the orchestrator delegates those to

a separate subagent; you do not run them yourself

• Create or modify sample_event.json — this is only generated by

elastic-package test system, run by that delegated test pass

• Modify CEL programs or cel.yml.hbs templates — the CEL program builder handles this

(see cel-programs/references/builder-subagent-guidance.md)

• Modify system test config files or mock API definitions — the CEL program builder

handles this for CEL streams; the data-collection setup builder handles this for non-CEL streams

Skill authority

The rules and patterns in the skills and their reference files are the authoritative source of truth. See the skill-authority disclaimer in ingest-pipelines/SKILL.md. When examining existing integrations for reference, many contain legacy patterns that predate current standards — always follow the skills over patterns observed in other integrations.

First steps — read the skills and their references

Before doing any work, read these skill files and the specific reference files listed to load the domain rules, patterns, and working code examples you must follow. Reading only the SKILL.md files is not sufficient — the reference files contain the actual processor examples, test fixture formats, and field mapping patterns you need.

1. `ingest-pipelines` skill (SKILL.md) — pipeline design patterns, processor ordering, error handling, JSE00001 compliance, syslog/KV strategies

• `references/processor-cookbook.md` — MUST READ: processor examples for all

common operations (JSON parsing, date handling, enrichment, type conversion, ECS categorization Patterns A/B/C and their anti-patterns)

• `references/error-handling-patterns.md` — MUST READ: on_failure blocks,

per-processor error handling, drop/tag strategies

• `references/branching-patterns.md` — read when building pipelines with

conditional logic or sub-pipelines

• `references/grok-recipes.md` — read when writing grok patterns: syntax, type

coercion, syslog header recipes, common mistakes

• `references/painless-patterns.md` — read when writing Painless script

processors: ctx access, params, HashMap for nested writes, foreach script context

2. `ecs-field-mappings` skill (SKILL.md) — ECS categorization values, field nesting rules, field files (ecs.yml, base-fields.yml, fields.yml), custom field types

• `references/categorization-cheatsheet.md` — MUST READ: valid

event.category / event.type combinations

• `references/mapping-type-matrix.md` — field type selection for custom fields
• `references/root-and-core-fields.md` — ECS root and core field definitions

3. `integration-testing` skill — then read `references/pipeline-testing.md` fully: test fixture format, config files, expected output generation, troubleshooting

4. `anonymize-logs` skill — fixture sourcing, sanitization, scenario coverage

5. `elastic-package-cli` skill — validation commands and pipeline test loop

Read all skills and their MUST READ references before writing any pipeline code.

Input data sources

You receive sample data from one of two paths. Do not research or re-investigate the data format when the data is already available from a prior step.

CEL-first path (data from the CEL program builder)

When the orchestrating agent tells you that a CEL program builder subagent (see cel-programs/references/builder-subagent-guidance.md) has already run:

1. Mock API response data exists in the system test config files (e.g., _dev/deploy/docker/ config files). Read these to understand the data structure the CEL program collects. 2. The orchestrating agent's prompt will describe the data structure, key fields, and the mock API response format. 3. The CEL program builder may have created initial field mappings in fields/fields.yml — read these to understand what raw fields exist.

Use the mock API response data and the orchestrator's description as your primary input for pipeline design. You will create your own pipeline test fixtures from this information.

Direct path (user-provided sample data)

When no CEL builder has run (log-based, syslog, file inputs):

1. Examine sample data provided by the user, research brief, or referenced files. 2. Identify the format: JSON, syslog, CEF, key-value, custom delimited, multiline.

In both paths, catalog the fields present, their data types, and which map to ECS vs. custom fields.

Workflow

1. Analyze the input data

• Read pipeline test fixtures if they already exist (CEL-first path), or examine

user-provided samples

• Catalog the fields present and their data types
• Identify which fields map to ECS and which are custom
• Identify the format: JSON (from CEL), syslog, CEF, key-value, custom delimited,

multiline

2. Design the pipeline

• Choose the parsing strategy (grok, dissect, json, kv, or combination)
• Plan the processor chain following the ordering from the ingest-pipelines skill:

1. ecs.version (9.3.0) and, for CEL streams only, Agentless remove / terminate when applicable 2. event.original handling (JSE00001 rename+remove pattern) 3. Parsing on event.original (grok/dissect/json/kv) — never mutate `event.original` 4. Date normalization (@timestamp) 5. Type conversions (convert processor) 6. Enrichment (geoip, user_agent) only when applicable 7. ECS categorization (event.kind, event.category, event.type, event.outcome) — use Pattern A (script with params lookup) for 5+ mappings, Pattern B (set with conditionals) for fewer than 5, Pattern C (sub-pipeline) for 100+. Do NOT use bulk `append` processors or inline Painless `if`/`else` chains without `params` — these are explicit anti-patterns. 8. Cleanup (remove temporary fields; conditional append for preserve_original_event when ctx.error?.message != null)

• For complex formats, design branching with sub-pipelines
• Prefer `rename` over `set` when moving values from custom/vendor fields into

ECS fields

• Add a `tag` to every processor

3. Implement the pipeline

• Write data_stream/<stream>/elasticsearch/ingest_pipeline/default.yml
• Write sub-pipeline files if branching is needed
• Include pipeline-level on_failure with event.kind: pipeline_error and contextual

error.message

• Follow all patterns from the ingest-pipelines skill

4. Define field mappings

• Create or update fields/ecs.yml with every ECS field the pipeline sets — use

name + external: ecs per entry, with type/value overrides only when needed.

• Create or update fields/fields.yml with custom integration-specific fields (non-ECS

fields only)

• Verify fields/base-fields.yml has the standard data stream routing constants

using external: ecs (all six fields are ECS fields; override type/value only for event.module and event.dataset)

• If the input emits Beats/Filebeat-specific fields (input.type, log.offset,

log.flags) not present in base-fields.yml, add them to fields/beats.yml

• Ensure _dev/build/build.yml exists at the package root with

dependencies.ecs.reference: "git@v9.3.0"

• Follow the field mapping guidance from the ecs-field-mappings skill

5. Create test fixtures

• Write test input fixtures in _dev/test/pipeline/ following the

integration-testing skill → references/pipeline-testing.md, based on the data structure from the mock API responses (CEL path) or user-provided samples (direct path)

• Use descriptive scenario names (test-successful-login.log, test-malformed-json.log)
• Ensure test-common-config.yml exists with appropriate dynamic_fields for any

non-deterministic fields the pipeline produces

• Include preserve_original_event in config fields.tags when appropriate
• Cover: happy path, format variants, edge cases, error paths

6. Generate pipeline expected output

Run pipeline tests with generation to produce or refresh expected files:

elastic-package test pipeline --data-streams <stream> --generate

7. Review generated output before accepting it

Manually inspect every file produced or updated by step 6 — primarily *-expected.json — and decide whether the pipeline output is actually correct, not merely consistent with a buggy parser.

Check at minimum:

• ECS fields are present and correctly populated
• Dotted source fields are expanded into nested objects
• geo_point fields appear under correct parent entities
• event.category and event.type are arrays (not scalars)
• @timestamp is parsed from source data (not only the ingest timestamp)
• No unexpected fields, missing values, or wrong types
• event.original is present when preserve_original_event is set in test config
• Values match the intent of each fixture scenario (happy path vs error path)

If anything is wrong, fix the pipeline and regenerate — never hand-edit expected JSON.

8. Run pipeline tests (no --generate)

Confirm the reviewed expected files match the pipeline:

elastic-package test pipeline --data-streams <stream>

9. Validate build

Run the full check sequence:

elastic-package format
elastic-package lint
elastic-package check

Fix any lint or build errors before reporting back.

Do not run system tests — the orchestrator delegates them to a dedicated subagent after pipeline work completes. Do not create or modify sample_event.json — it is generated exclusively by elastic-package test system in that delegated pass.

Data anonymization

All data committed to the repository must be fully anonymized. No real production data, customer data, or identifiable information may appear in any committed file:

• Pipeline test fixtures (test-*.log, test-*.json): replace real IP addresses,

hostnames, email addresses, usernames, organization names, account IDs, tokens, and any other traceable data with synthetic examples of the same format.

• Generated expected output (*-expected.json): ensure source fixtures are

anonymized first.

Use RFC 5737 documentation IP ranges (198.51.100.x, 203.0.113.x), example.com domains, realistic placeholder names. Refer to the anonymize-logs skill for the full placeholder convention list.

What to return

When you finish, report:

• Files created or modified (with paths)
• Field files created or modified: list each field file and what was added/changed, so

the orchestrating agent knows which field definitions are already handled

• Pipeline architecture summary (single-path or branching, parser strategy)
• ECS categorization choices made and rationale
• Test coverage: scenarios covered, number of test events
• Pipeline test results: elastic-package test pipeline pass/fail (after step 8)
• Generated output review: confirm *-expected.json files were read and validated in

step 7, not only that the command succeeded

• Validation results from elastic-package format/lint/check
• Any open issues or decisions that need user input

CDR pipeline requirements

Cloud Detection & Response (CDR) integrations handle findings from cloud security posture management (CSPM), cloud workload protection (CWPP), and vulnerability management tools. This reference covers the pipeline-side requirements for CDR compliance.

Aligned with: Elastic CDR 3P Developer Guide v1.0

Event categorization

Correct event.* values are critical -- the Kibana Findings page filters on them.

Use append for event.category and event.type (they are arrays in ECS). Use set for event.kind (single value).

Must Have fields (pipeline must populate)

Missing these causes critical issues in the Kibana Findings UI.

For vulnerability findings additionally:

Should Have fields

Correlation fields

CDR integrations MUST populate related.* fields for threat hunting:

• related.ip -- resource IPs, actor IPs
• related.user -- IAM users, service accounts
• related.hash -- artifact hashes (for CWPP findings)

Value transformations

• Severity -- must be lowercase: CRITICAL -> critical, VERY_HIGH -> critical, MODERATE -> medium
• Status -- map vendor values: ACTIVE -> failed, INACTIVE/ARCHIVED -> passed/resolved
• Account IDs -- extract from resource paths: projects/my-project -> my-project

Pipeline patterns

Event categorization for misconfiguration

- set:
    tag: set_event_kind
    field: event.kind
    value: state
- append:
    tag: append_event_category
    field: event.category
    value: configuration
- append:
    tag: append_event_type
    field: event.type
    value: info

Event categorization for vulnerability

- set:
    tag: set_event_kind
    field: event.kind
    value: state
- append:
    tag: append_event_category
    field: event.category
    value: vulnerability
- append:
    tag: append_event_type
    field: event.type
    value: info

Result evaluation mapping

- set:
    tag: set_result_evaluation
    field: result.evaluation
    value: failed
    if: ctx.json?.compliance_status == "NON_COMPLIANT"
- set:
    tag: set_result_evaluation
    field: result.evaluation
    value: passed
    if: ctx.json?.compliance_status == "COMPLIANT"

Observer vendor (constant per integration)

- set:
    tag: set_observer_vendor
    field: observer.vendor
    value: "Wiz"

Cloud context

- set:
    tag: set_cloud_provider
    field: cloud.provider
    value: aws
- set:
    tag: set_cloud_account_id
    field: cloud.account.id
    value: '{{{json.account_id}}}'

Vulnerability fields (conditional on finding type)

- set:
    tag: set_vulnerability_id
    field: vulnerability.id
    copy_from: json.cve_id
    if: ctx.json?.finding_type == 'VULNERABILITY'
- set:
    tag: set_vulnerability_severity
    field: vulnerability.severity
    value: '{{{json.severity}}}'

Known CDR integrations

aws_security_hub, google_scc, azure_security_center, azure_defender, prisma_cloud, crowdstrike, wiz, orca, snyk, lacework, tenable, qualys, rapid7, sentinelone, sysdig

Detection indicators in file paths or package names: security_hub, securityhub, security_center, scc, defender, cloud_security, cspm, cnvm, cwpp, cdr, finding, findings, vulnerability, compliance, posture

CDR pipeline review checklist

What to flag

• [ ] Missing event.category/event.kind/event.type or wrong values for the finding type -- HIGH
• [ ] Missing resource.id or resource.name (breaks transform and UI) -- HIGH
• [ ] Missing result.evaluation for misconfiguration findings -- HIGH
• [ ] Missing rule.uuid (breaks transform uniqueness) -- HIGH
• [ ] event.kind not set to state for posture findings -- HIGH
• [ ] No related.* fields populated -- MEDIUM
• [ ] Missing observer.vendor -- HIGH
• [ ] Severity not lowercase -- MEDIUM
• [ ] Vulnerability fields applied to non-vulnerability findings (must be conditional) -- MEDIUM
• [ ] Using cloud.detection.* (NOT part of CDR spec -- use rule.*, result.*) -- MEDIUM

What NOT to flag

CDR pipeline requirements are NOT applicable to:

• General logging or metrics integrations
• APM integrations
• Non-security cloud integrations (e.g., billing, resource inventory without security posture)
• Integrations that do not produce security findings (misconfiguration, vulnerability, or runtime detection)

Do not flag missing CDR fields on these integration types.

ingest pipeline error handling patterns

This guide focuses on resilient failure behavior and actionable debugging output.

Recommended top-level on_failure

Use this pattern in default.yml to ensure all uncaught failures are visible:

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

Why:

• Appending error.message first preserves the full _ingest.on_failure_* context for triage.
• event.kind: pipeline_error supports clean filtering and dashboards.
• preserve_original_event on failure helps post-failure diagnostics.

Processor tag requirement

Every processor in the pipeline should include a tag (not only processors that can fail).

- grok:
    field: event.original
    patterns: ['^%{IP:source.ip} %{GREEDYDATA:message}$']
    tag: parse_source_ip

Without tags, error messages lose key context and triage becomes slower.

Processor-level on_failure: when and when not

Use processor-level on_failure for:

• cleanup (remove invalid temporary fields)
• fallback operations on known parse failures
• local annotations that complement top-level errors

Avoid using processor-level on_failure as the only error-reporting path. If a processor if expression itself fails, control may bypass that local handler and fall through to top-level on_failure.

Example: date parse fallback detail

- date:
    field: nginx.access.time
    target_field: '@timestamp'
    formats:
      - dd/MMM/yyyy:H:m:s Z
    tag: parse_access_time
    on_failure:
      - append:
          field: error.message
          value: '{{{_ingest.on_failure_message}}}'

ignore_failure usage

Use ignore_failure: true only when failure should not block ingestion.

Good candidates:

• optional enrichment (geoip, user_agent)
• best-effort cleanup/normalization
• non-critical parsing of auxiliary fields

Example:

- geoip:
    field: source.ip
    target_field: source.geo
    ignore_failure: true
    ignore_missing: true
    tag: enrich_source_geo

Avoid ignore_failure on required parse steps that define event shape.

fail processor usage

Use fail for invalid required input or unrecoverable branch conditions.

Input validation gate

- fail:
    if: ctx.json == null || !(ctx.json instanceof Map)
    message: missing json object in input document
    tag: validate_json_input

Critical parser escalation (inside on_failure)

- kv:
    field: message
    field_split: ' '
    value_split: '='
    tag: parse_kv
    on_failure:
      - fail:
          message: 'unable to parse key-values: {{{ _ingest.on_failure_message }}}'
          tag: fail_parse_kv

Pattern catalog

1) Minimal pipeline-level error pattern (lightweight)

on_failure:
  - set:
      field: error.message
      value: '{{{_ingest.on_failure_message}}}'

Use this only for simple integrations where richer diagnostics are not yet needed.

2) Full-context pipeline-level error pattern (preferred)

Use the recommended top-level on_failure block from the section above. This is the standard pattern for default.yml in all integrations.

3) Conditional preserve tag before successful pipeline end

When the collector set an error but ingestion continued, tag the document so operators can find it:

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

(This is separate from the on_failure block; use together with pattern 2.)

4) Local cleanup in processor on_failure

- json:
    field: event.original
    target_field: json
    tag: parse_json
    on_failure:
      - remove:
          field: json
          ignore_missing: true
      - append:
          field: error.message
          value: '{{{_ingest.on_failure_message}}}'

Review checklist

• Top-level on_failure exists in every primary ingest pipeline.
• Every processor has a tag.
• Required parse steps do not silently ignore failure.
• ignore_failure is limited to optional/non-critical operations.
• Error messages include enough context to locate the failed processor.

Grok Pattern Recipes

Pattern library reference

The authoritative built-in pattern library for Elasticsearch's grok processor lives in the Elasticsearch source tree:

• Browsable directory (all pattern files): https://github.com/elastic/elasticsearch/tree/master/libs/grok/src/main/resources/patterns/ecs-v1
• Core `grok-patterns` file (IP, hostname, timestamps, numbers, paths, syslog, HTTP): https://github.com/elastic/elasticsearch/blob/master/libs/grok/src/main/resources/patterns/ecs-v1/grok-patterns

Always check those sources for the actual regex behind a pattern before assuming it matches a specific input shape.

---

Core grok syntax

Three expression forms

# Match only — skip a token
%{IP} - %{USER:user.name}

# Named capture
%{IP:source.ip} %{WORD:http.request.method} %{URIPATH:url.path}

# Typed capture
%{NUMBER:http.response.status_code:int} %{NUMBER:http.response.body.bytes:long}

Inline regex captures

When no built-in pattern fits, embed a raw regex directly:

(?<log.level>DEBUG|INFO|WARN|ERROR|FATAL)
(?<event.action>[a-zA-Z0-9_\-]+)

Inline regex and %{SYNTAX} references can be freely mixed in the same expression.

---

Type coercion

By default every capture is a string. Append a type suffix to coerce:

%{NUMBER:http.response.status_code:int}
%{NUMBER:http.response.body.bytes:long}
%{NUMBER:event.duration:double}

---

Custom patterns with pattern_definitions

Define reusable inline patterns directly in the grok processor (no separate pattern file needed for Elasticsearch pipelines):

- grok:
    field: event.original
    pattern_definitions:
      THREAD_ID: "[A-Za-z0-9#]+"
      JAVA_CLASS: "[a-zA-Z$_][a-zA-Z$_0-9]*(?:\\.[a-zA-Z$_][a-zA-Z$_0-9]*)*"
    patterns:
      - '^%{TIMESTAMP_ISO8601:timestamp} \[%{THREAD_ID:thread}\] %{WORD:log.level} %{JAVA_CLASS:logger} - %{GREEDYDATA:message}$'
    tag: parse_java_log

Naming convention: SCREAMING_SNAKE_CASE; prefix with a namespace to avoid collisions: MYAPP_REQUEST_ID instead of REQUEST_ID.

---

Syslog header recipes

Syslog integrations receive a full syslog line and need to split the header from the payload. Parse the header in default.yml or a shared sub-pipeline, then route the extracted message to a format-specific sub-pipeline.

Rule: never overwrite event.original. Store extracted sub-fields in _temp.* when you need to pass them to sub-pipelines.

RFC 3164 — traditional syslog header

Sample:
Jan 15 10:30:00 web-01 sshd[1234]: Accepted publickey for alice from 192.168.1.10 port 55234

Pattern:
^%{SYSLOGTIMESTAMP:timestamp} %{IPORHOST:host.hostname} %{NOTSPACE:process.name}(?:\[%{POSINT:process.pid:int}\])?: %{GREEDYDATA:message}$

Fields: timestamp, host.hostname, process.name, process.pid
Payload: message

RFC 5424 — structured syslog header (no SD-ELEMENT)

Sample:
<34>1 2024-01-15T10:30:00.000Z mymachine.example.com sshd 1234 ID47 - Accepted publickey for alice

Pattern:
^<%{NONNEGINT:syslog.priority:int}>%{POSINT:syslog.version:int} %{TIMESTAMP_ISO8601:timestamp} %{IPORHOST:host.hostname} %{NOTSPACE:process.name} %{NOTSPACE:process.pid} %{NOTSPACE:syslog.msgid} - %{GREEDYDATA:message}$

RFC 5424 — structured syslog header (with SD-ELEMENT block)

Sample:
<34>1 2024-01-15T10:30:00.000Z mymachine.example.com su - ID47 [exampleSDID@32473 iut="3"] BOM su root failed

Pattern:
^<%{NONNEGINT:syslog.priority:int}>%{POSINT:syslog.version:int} %{TIMESTAMP_ISO8601:timestamp} %{IPORHOST:host.hostname} %{NOTSPACE:process.name} %{NOTSPACE:process.pid} %{NOTSPACE:syslog.msgid} (?:\[%{DATA:syslog.structured_data}\]|-) %{GREEDYDATA:message}$

Capture the SD-ELEMENT block into syslog.structured_data, then pass it to a kv processor. See the Syslog structured data strategies section below for the full KV, SYSLOG5424SD, and Painless approaches.

Split header from payload for sub-pipeline routing

When default.yml is a thin router, extract the envelope but do not clobber event.original:

- grok:
    field: event.original
    patterns:
      - '^%{SYSLOGTIMESTAMP:_temp.timestamp} %{IPORHOST:host.hostname} %{NOTSPACE:process.name}(?:\[%{POSINT:process.pid:int}\])?: %{GREEDYDATA:_temp.message}$'
    tag: parse_syslog_header

Sub-pipelines then parse _temp.message for their specific event format. _temp fields are removed at the end of the pipeline.

---

Syslog structured data strategies

Firewall and network integrations frequently receive syslog with RFC 5424 structured data elements — the [sdId key1="value1" key2="value2"] format, or vendor-specific key=value key2="quoted value" payloads embedded in syslog messages.

Strategy 1: Grok + KV with trim_value (simplest)

When values use consistent quoting and keys contain no special characters, the built-in kv processor handles this well. Use a lookahead-based field_split to handle spaces inside quoted values.

This pattern is used by integrations like juniper_srx, sophos, and sonicwall_firewall in the upstream elastic/integrations repository.

- kv:
    field: _temp.kv_data
    field_split: ' (?=[a-zA-Z0-9_-]+=)'
    value_split: "="
    prefix: "vendor.product."
    trim_value: '"'
    ignore_missing: true
    tag: kv_structured_data

Key settings:

• field_split: ' (?=[a-zA-Z0-9_-]+=)' — splits on spaces only when followed by a key= pattern, preserving spaces in quoted values
• trim_value: '"' — strips surrounding quotes from values
• prefix — namespaces all extracted keys under a vendor prefix

Strategy 2: Grok with SYSLOG5424SD + KV with regex splits

When the syslog header follows RFC 5424 strictly, use the built-in SYSLOG5424SD grok pattern to capture the structured data block, then parse it with kv. Some vendors use : or :: as the value separator instead of =.

This pattern is based on the system/auth integration in the upstream elastic/integrations repository.

- grok:
    field: event.original
    patterns:
      - '^<%{NONNEGINT:log.syslog.priority:int}>%{NONNEGINT} %{TIMESTAMP} %{IPORHOST:host.hostname} %{DATA:process.name} %{POSINT:process.pid:long} %{DATA:event.code} (?:-|%{SYSLOG5424SD:syslog5424_sd}) %{GREEDYDATA:message}$'
    tag: parse_rfc5424

- kv:
    if: ctx.syslog5424_sd != null && ctx.syslog5424_sd != ''
    field: syslog5424_sd
    field_split: '(?<=") '
    value_split: '(?i)(?<=[a-z])=(?=")'
    trim_key: " "
    trim_value: " "
    prefix: parsed.
    strip_brackets: true
    tag: kv_sd_element

Strategy 3: Painless script for complex quoted-value KV

When values contain embedded equals signs, mixed quoting, or irregular delimiters that defeat the kv processor, use a Painless script. This is heavier but handles edge cases reliably.

This pattern is used by integrations like fortinet_fortigate and fortinet_fortimanager in the upstream elastic/integrations repository.

- script:
    lang: painless
    if: ctx._temp?.kv_data != null
    tag: script_parse_quoted_kv
    description: Split KV pairs handling quoted values with embedded spaces/delimiters.
    source: |
      def splitUnquoted(String input, String sep) {
        def tokens = [];
        def startPosition = 0;
        def isInQuotes = false;
        char quote = (char)"\"";
        for (def i = 0; i < input.length(); i++) {
          if (input.charAt(i) == quote) {
            isInQuotes = !isInQuotes;
          } else if (input.charAt(i) == (char)sep && !isInQuotes) {
            def token = input.substring(startPosition, i).trim();
            if (!token.equals("")) { tokens.add(token); }
            startPosition = i + 1;
          }
        }
        def last = input.substring(startPosition).trim();
        if (!last.equals("")) { tokens.add(last); }
        return tokens;
      }
      def arr = splitUnquoted(ctx._temp.kv_data, " ");
      Map map = new HashMap();
      Pattern pattern = /^\"|\"$/;
      for (def i = 0; i < arr?.length; i++) {
        def kv = splitUnquoted(arr[i], "=");
        if (kv.length == 2 && kv[0].length() > 0) {
          map[kv[0]] = pattern.matcher(kv[1]).replaceAll("");
        }
      }
      ctx.vendor = new HashMap();
      ctx.vendor.product = map;

Prefer strategy 1 or 2 when possible; use the script approach only when KV edge cases demand it.

---

Web server / HTTP access log

Sample (Nginx combined):
192.168.1.1 - alice [15/Jan/2024:10:30:00 +0000] "GET /api/v1/health HTTP/1.1" 200 512 "https://example.com" "Mozilla/5.0"

Pattern (ECS field names):
^%{IPORHOST:source.ip} - %{DATA:user.name} \[%{HTTPDATE:timestamp}\] "%{WORD:http.request.method} %{NOTSPACE:url.original}(?: HTTP/%{NUMBER:http.version})?" %{NUMBER:http.response.status_code:int} (?:%{NUMBER:http.response.body.bytes:long}|-) "(?:%{URI:http.request.referrer}|-)" "%{DATA:user_agent.original}"$

The built-in %{COMMONAPACHELOG} and %{COMBINEDAPACHELOG} patterns exist but use legacy (non-ECS) field names. Prefer the explicit ECS-mapped pattern above.

---

Common mistakes

---

Debugging

• Kibana Grok Debugger: Stack Management → Grok Debugger — interactive pattern testing against sample input
• Elasticsearch Simulate API: POST _ingest/pipeline/_simulate with "trace_match": true — see which pattern index matched and inspect intermediate fields
• regex101.com: select Oniguruma flavor for regex-level step-by-step explanation

Painless Script Patterns

Reference for Painless usage inside script ingest processors. For foundational rules (no ?. in script bodies, equalsIgnoreCase guidance, tag/description requirement, prefer built-in processors), see SKILL.md -- Painless script best practices. This document covers deeper patterns and examples.

ctx read patterns

All document fields are accessed through ctx. Use the null-safe ?. operator for nested paths to avoid NullPointerException. Note: ctx itself is always non-null in ingest pipelines, so ctx?.field is unnecessary -- use ctx.field directly. The ?. operator is useful on nested paths where a parent may be absent.

- script:
    tag: extract_user_info
    description: Extract user info with safe nested access
    lang: painless
    source: |
      if (ctx.user?.identity != null) {
        def name = ctx.user.identity.name;
        if (name != null && !name.isEmpty()) {
          ctx.user.full_name = name;
        }
      }

Both ctx.user?.identity != null and the explicit chained form ctx.user != null && ctx.user.identity != null are valid. The null-safe ?. form is preferred for conciseness. The verbose form may be clearer in complex conditions with side effects.

Bracket notation is useful when field names contain dots or special characters:

- script:
    tag: read_dotted_field
    description: Read a field name that contains a literal dot
    lang: painless
    source: |
      if (ctx.containsKey('host.name')) {
        ctx.host_name = ctx['host.name'];
      }

containsKey checks for field presence without risk of a null pointer:

if (ctx.containsKey('source') && ctx.source.containsKey('ip')) {
  // safe to use ctx.source.ip
}

ctx write patterns

Direct assignment creates or overwrites a field. remove deletes it.

ctx.event.outcome = 'success';
ctx.put('event.outcome', 'success');   // equivalent map-style put
ctx.remove('_temp');                   // delete a field
ctx.list_field.add(item);             // append to an existing list

params usage

params holds immutable constants declared in the processor config. Values in params are set once and shared across all documents processed by that script. Elasticsearch compiles the script once and caches it; changing only params values does not trigger recompilation.

Use params for:

• Lookup/mapping tables
• Regex patterns
• Threshold values and configuration constants

- script:
    tag: map_severity
    description: Map severity label string to numeric value using params lookup
    lang: painless
    params:
      severity_map:
        low: 1
        medium: 2
        high: 3
        critical: 4
    source: |
      def label = ctx.event?.severity_label;
      if (label != null && params.severity_map.containsKey(label)) {
        ctx.event = ctx.event ?: [:];
        ctx.event.severity = params.severity_map.get(label);
      }

For 1-2 constant values, inline comparisons in source are fine. For 3+ values or lookup tables, move them into params for readability and maintainability:

# Avoid for many values -- harder to read and maintain
- script:
    source: |
      if (ctx.raw_status == "ACTIVE") { ctx.status = "active"; }
      else if (ctx.raw_status == "INACTIVE") { ctx.status = "inactive"; }
      else if (ctx.raw_status == "DISABLED") { ctx.status = "disabled"; }

# Preferred -- params keep source generic and readable
- script:
    tag: map_status
    description: Normalize raw status using params lookup table
    lang: painless
    params:
      status_map:
        ACTIVE: active
        INACTIVE: inactive
        DISABLED: disabled
    source: |
      if (ctx.raw_status != null) {
        ctx.status = params.status_map.getOrDefault(ctx.raw_status, ctx.raw_status);
      }

Map initialization for nested writes

Before writing to a nested path, ensure every parent map exists. A write to ctx.event.outcome fails if ctx.event is null. Use the ?: (Elvis) operator with [:] (empty map literal) for concise null-coalescing initialization.

- script:
    tag: init_event_outcome
    description: Set event outcome with safe parent initialization
    lang: painless
    source: |
      ctx.event = ctx.event ?: [:];
      ctx.event.outcome = 'success';

For deeply nested paths, each level must be initialized:

- script:
    tag: set_deep_nested_field
    description: Set a deeply nested field with full parent chain init
    lang: painless
    source: |
      ctx.organization = ctx.organization ?: [:];
      ctx.organization.department = ctx.organization.department ?: [:];
      ctx.organization.department.name = ctx._temp_dept;

When building a new nested object from scratch, initialize the root as a HashMap and populate it:

- script:
    tag: build_related_object
    description: Build related.ip from multiple source fields
    lang: painless
    source: |
      def ips = new HashSet();
      if (ctx.source?.ip != null) {
        ips.add(ctx.source.ip);
      }
      if (ctx.destination?.ip != null) {
        ips.add(ctx.destination.ip);
      }
      if (!ips.isEmpty()) {
        ctx.related = ctx.related ?: [:];
        ctx.related.ip = new ArrayList(ips);
      }

Field API (ES 9.2+)

The Field API provides cleaner syntax for deeply nested field access. It handles null parent maps automatically, eliminating manual HashMap initialization chains. Available in Elasticsearch 9.2+ for conditionals.

// Field API -- set a deeply nested field without manual HashMap init
field('system.cpu.total.norm.pct').set($('cpu.usage', 0.0) / 100.0)

Without the Field API, the same operation requires explicit initialization of every parent:

ctx.system = ctx.system ?: [:];
ctx.system.cpu = ctx.system.cpu ?: [:];
ctx.system.cpu.total = ctx.system.cpu.total ?: [:];
ctx.system.cpu.total.norm = ctx.system.cpu.total.norm ?: [:];
ctx.system.cpu.total.norm.pct = ctx.cpu.usage / 100.0;

Reference: https://www.elastic.co/docs/manage-data/ingest/transform-enrich/readable-maintainable-ingest-pipelines

foreach processor context

When a script runs inside a foreach processor, the current array element is accessed via ctx._ingest._value, not through the array directly.

- foreach:
    tag: normalize_event_items
    description: Lowercase the name field on each item in event.items
    field: event.items
    processor:
      script:
        tag: lowercase_item_name
        description: Lowercase the current item name
        lang: painless
        source: |
          if (ctx._ingest._value.name != null) {
            ctx._ingest._value.name = ctx._ingest._value.name.toLowerCase();
          }

Key rules:

• ctx._ingest._value refers to the current element of the array specified in foreach.field.
• ctx._ingest._value.name with foreach.field: event.items resolves to event.items[*].name.
• Writing to ctx._ingest._value or its sub-fields modifies the element in place.
• The script still has access to the full document via ctx for reading other fields.

Conditional field removal

Use containsKey to check before removing. Removing a non-existent key from ctx does not throw, but checking first is idiomatic when the removal is conditional on other logic:

- script:
    tag: remove_temp_fields
    description: Remove all temporary fields after processing
    lang: painless
    source: |
      def to_remove = ['_temp', '_header', '_raw_message'];
      for (def field : to_remove) {
        if (ctx.containsKey(field)) {
          ctx.remove(field);
        }
      }

For nested field removal:

- script:
    tag: clean_empty_nested
    description: Remove nested object if all its children are null
    lang: painless
    source: |
      if (ctx.source != null && ctx.source.ip == null && ctx.source.port == null) {
        ctx.remove('source');
      }

Common script patterns

Array deduplication

- script:
    tag: dedup_tags
    description: Remove duplicate entries from tags array
    lang: painless
    source: |
      if (ctx.tags != null && ctx.tags instanceof List) {
        ctx.tags = new ArrayList(new LinkedHashSet(ctx.tags));
      }

LinkedHashSet preserves insertion order while removing duplicates.

IP normalization

- script:
    tag: normalize_ipv6
    description: Expand compressed IPv6 addresses to full form
    lang: painless
    params:
      ipv4_mapped_prefix: '::ffff:'
    source: |
      if (ctx.source != null && ctx.source.ip != null) {
        def ip = ctx.source.ip;
        if (ip.startsWith(params.ipv4_mapped_prefix)) {
          ctx.source.ip = ip.substring(params.ipv4_mapped_prefix.length());
        }
      }

String manipulation

- script:
    tag: extract_domain_from_email
    description: Extract domain part from user email address
    lang: painless
    source: |
      if (ctx.user != null && ctx.user.email != null) {
        def email = ctx.user.email;
        int idx = email.indexOf('@');
        if (idx > 0) {
          ctx.user.domain = email.substring(idx + 1);
        }
      }

Timestamp arithmetic

- script:
    tag: compute_duration
    description: Compute event duration from start and end timestamps
    lang: painless
    source: |
      if (ctx.event != null && ctx.event.start != null && ctx.event.end != null) {
        def start = ZonedDateTime.parse(ctx.event.start);
        def end = ZonedDateTime.parse(ctx.event.end);
        ctx.event.duration = ChronoUnit.NANOS.between(start, end);
      }

Anti-patterns

Overuse of scripts when processors suffice. Before writing any script processor, you must verify that no built-in processor can do the job. script is slower than every built-in processor except geoip/user_agent — it carries Painless compilation cost and per-document execution overhead. Common replacements:

Concrete example — extracting a domain from an email address:

# WRONG — script for a job dissect handles natively
- script:
    tag: script_set_user_domain
    lang: painless
    description: Extract domain from email address.
    if: ctx.vendor?.owner instanceof String && ctx.vendor.owner.contains('@')
    source: |-
      String u = ctx.vendor.owner;
      int at = u.lastIndexOf('@');
      if (at > 0 && at < u.length() - 1) {
        ctx.user = ctx.user ?: [:];
        ctx.user.domain = u.substring(at + 1);
      }

# CORRECT — dissect is faster, shorter, and easier to review
- dissect:
    tag: dissect_user_domain
    field: vendor.owner
    pattern: "%{?_ignore}@%{user.domain}"
    if: ctx.vendor?.owner != null && ctx.vendor.owner.contains('@')

Missing null checks. Every field access in a script body must be guarded. A missing null check on a field that is absent in some documents causes NullPointerException at ingest time, which triggers on_failure for that document.

Hardcoded values in script bodies. Constants belong in params, not inline in the source string. Inline values prevent Elasticsearch from caching the compiled script across different configurations and make the mapping logic harder to review.

Review checklist

• [ ] Script could be replaced with a built-in processor -- MEDIUM (see anti-patterns table above)
• [ ] Script has tag and description -- MEDIUM
• [ ] params used for constants instead of hardcoding in script body -- LOW
• [ ] Null checks present before field access -- MEDIUM
• [ ] No ctx? usage (ctx is always non-null) -- LOW

ingest processor cookbook

Use this cookbook to choose processors quickly while designing integration ingest pipelines.

Parsing processors

Example: grok + dissect fallback

- dissect:
    field: event.original
    pattern: "%{source.address} - %{user.name} [%{nginx.access.time}] \"%{http.request.method} %{url.original} HTTP/%{http.version}\" %{http.response.status_code} %{http.response.body.bytes}"
    ignore_failure: true
    tag: dissect_access
- grok:
    field: event.original
    patterns:
      - '^%{IPORHOST:source.address} - %{DATA:user.name} \[%{HTTPDATE:nginx.access.time}\] "%{WORD:http.request.method} %{DATA:url.original} HTTP/%{NUMBER:http.version}" %{NUMBER:http.response.status_code:long} %{NUMBER:http.response.body.bytes:long}$'
    if: ctx.http?.request?.method == null
    tag: grok_access_fallback

Example: json parser with local failure handling

- json:
    field: event.original
    target_field: json
    tag: parse_json
    on_failure:
      - append:
          field: error.message
          value: '{{{_ingest.on_failure_message}}}'

Normalization processors

Example: normalize timestamp + status

- date:
    field: nginx.access.time
    target_field: '@timestamp'
    formats:
      - dd/MMM/yyyy:H:m:s Z
    tag: parse_timestamp
- convert:
    field: http.response.status_code
    type: long
    ignore_missing: true
    tag: convert_status_code

Enrichment processors

Example: common web enrichment path

- user_agent:
    field: user_agent.original
    if: ctx.user_agent?.original != null
    ignore_missing: true
    tag: parse_user_agent
- geoip:
    field: source.ip
    target_field: source.geo
    if: ctx.source?.ip != null
    ignore_missing: true
    tag: enrich_source_geo
- append:
    field: related.ip
    value: '{{{source.ip}}}'
    if: ctx.source?.ip != null

IP geolocation and ASN enrichment — full pattern

When the pipeline has IP address fields, always apply both the geo lookup and the ASN lookup, followed by the renames to map the raw geoip output into ECS field names. This pattern applies to any IP entity (source, destination, client, server, etc.).

# IP Geolocation Lookup
- geoip:
    field: source.ip
    target_field: source.geo
    ignore_missing: true
    tag: enrich_source_geo
- geoip:
    field: destination.ip
    target_field: destination.geo
    ignore_missing: true
    tag: enrich_destination_geo

# IP Autonomous System (AS) Lookup
- geoip:
    database_file: GeoLite2-ASN.mmdb
    field: source.ip
    target_field: source.as
    properties:
      - asn
      - organization_name
    ignore_missing: true
    tag: enrich_source_asn
- geoip:
    database_file: GeoLite2-ASN.mmdb
    field: destination.ip
    target_field: destination.as
    properties:
      - asn
      - organization_name
    ignore_missing: true
    tag: enrich_destination_asn

# Rename ASN fields to ECS names
- 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
- rename:
    field: destination.as.asn
    target_field: destination.as.number
    ignore_missing: true
    tag: rename_destination_asn
- rename:
    field: destination.as.organization_name
    target_field: destination.as.organization.name
    ignore_missing: true
    tag: rename_destination_as_org

Key rules:

• The geoip processor with GeoLite2-ASN.mmdb outputs asn (number) and organization_name (string). These must be renamed to ECS names: as.number and as.organization.name.
• Always include both geo and ASN lookups when enriching IP fields. Omitting ASN leaves source.as / destination.as empty.
• Use ignore_missing: true on all geo/ASN processors — the IP field may not be present on every event.
• When only one IP entity is present (e.g., only source.ip), include only the source block. Add the destination block only when destination.ip exists in the pipeline.
• Place these processors after all IP fields have been set (after parsing, renaming, and converting IP addresses) and before the related.ip append processors.

Flow control processors

Example: route to sub-pipelines

- pipeline:
    name: '{{ IngestPipeline "pipeline_object_user" }}'
    if: ctx.ocsf?.class_uid != null && ['2005','3001','3002'].contains(ctx.ocsf.class_uid) && ctx.ocsf.user != null
    ignore_missing_pipeline: true
    tag: route_object_user
- pipeline:
    name: '{{ IngestPipeline "pipeline_category_network_activity" }}'
    if: ctx.ocsf?.category_uid == '4'
    ignore_missing_pipeline: true
    tag: route_category_network

Utility processors

ECS categorization mapping patterns

Map source event types/actions to event.category, event.type, event.outcome, and event.action using these patterns. Choose the pattern that fits the number of mappings.

Pattern A: Script with params lookup table (recommended for 5+ mappings)

Put all mapping data in params so the Painless script body stays generic and benefits from compilation caching. The script source is short and identical regardless of how many mappings exist.

Single-key lookup (mapping from one source field):

  - script:
      lang: painless
      tag: set_ecs_categorization
      description: Map event type to ECS categorization fields.
      if: ctx.json?.event_type != null
      params:
        process_start:
          category: [process]
          type: [start]
        process_end:
          category: [process]
          type: [end]
        network_connection:
          category: [network]
          type: [connection]
        file_creation:
          category: [file]
          type: [creation]
        file_modification:
          category: [file]
          type: [change]
        user_login:
          category: [authentication]
          type: [start]
          outcome: success
        user_login_failed:
          category: [authentication]
          type: [start]
          outcome: failure
      source: |-
        def mapping = params[ctx.json.event_type];
        if (mapping == null) {
          return;
        }
        ctx.event.category = mapping.category;
        ctx.event.type = mapping.type;
        if (mapping.containsKey('outcome')) {
          ctx.event.outcome = mapping.outcome;
        }

Composite-key lookup (mapping from two or more source fields combined):

  - script:
      lang: painless
      tag: set_ecs_categorization
      description: Map target type and action to ECS categorization fields.
      if: ctx.vendor?.target_type != null && ctx.event?.action != null
      params:
        "device:create":
          category: [host]
          type: [creation]
          outcome: success
        "device:delete":
          category: [host]
          type: [deletion]
          outcome: success
        "device:update":
          category: [host]
          type: [change]
          outcome: success
        "api_token:create":
          category: [iam, configuration]
          type: [creation]
          outcome: success
        "api_token:delete":
          category: [iam, configuration]
          type: [deletion]
          outcome: success
        "blueprint:update":
          category: [configuration]
          type: [change]
          outcome: success
      source: |-
        String key = ctx.vendor.target_type + ':' + ctx.event.action;
        def mapping = params[key];
        if (mapping == null) {
          return;
        }
        ctx.event.category = mapping.category;
        ctx.event.type = mapping.type;
        if (mapping.containsKey('outcome')) {
          ctx.event.outcome = mapping.outcome;
        }

Merge variant (when categorization fields may already have values from earlier processors and you need to add to them rather than overwrite):

      source: |-
        def addUnique(List dst, List src) {
          HashSet s = new HashSet(dst != null ? dst : []);
          s.addAll(src != null ? src : []);
          return new ArrayList(s);
        }
        def mapping = params[ctx.json.event_type];
        if (mapping == null) {
          return;
        }
        ctx.event.type = addUnique(ctx.event.type, mapping.type);
        ctx.event.category = addUnique(ctx.event.category, mapping.category);

Why `params`? Elasticsearch compiles and caches Painless scripts by their source text. When the mapping data is in params rather than inlined in the script body, the same compiled script handles all event types. Inline if/else chains produce a unique script body that cannot be shared, defeating the cache.

Reference integrations using this pattern:

• carbon_black_cloud (endpoint_event) — single-key lookup by json.type
• okta (system) — merge variant with large params table in dedicated sub-pipeline
• thycotic_ss (logs) — single-key lookup by cef.name
• zeronetworks (audit) — composite fields with action, outcome, type, category per entry

Pattern B: Set processors with conditionals (fewer than 5 mappings)

For simple cases with only 2–4 distinct mappings, set processors with if conditions are clearer than a script:

  - set:
      field: event.category
      tag: set_category_auth
      value: [authentication]
      if: ctx.json?.event_type == 'login' || ctx.json?.event_type == 'logout'
  - set:
      field: event.type
      tag: set_type_start
      value: [start]
      if: ctx.json?.event_type == 'login'
  - set:
      field: event.type
      tag: set_type_end
      value: [end]
      if: ctx.json?.event_type == 'logout'
  - set:
      field: event.outcome
      tag: set_outcome_success
      value: success
      if: ctx.json?.result == 'success'
  - set:
      field: event.outcome
      tag: set_outcome_failure
      value: failure
      if: ctx.json?.result == 'failure'

Use this pattern when the mapping is straightforward and adding a script processor would be overkill. Once the number of conditions exceeds ~4 distinct event types, switch to Pattern A.

Pattern C: Sub-pipeline for large mapping tables (100+ mappings)

For very large mapping tables (like Okta with 500+ event types), extract the categorization into a dedicated sub-pipeline file to keep default.yml readable:

# In default.yml — route to categorization sub-pipeline:
  - pipeline:
      name: '{{ IngestPipeline "ecs-categorization" }}'
      tag: route_ecs_categorization

# In ecs-categorization.yml — single script processor with large params table:
---
description: Map event types to ECS categorization fields.
processors:
  - script:
      lang: painless
      tag: set_ecs_categorization
      description: Map event type to ECS categorization fields.
      if: ctx.vendor?.event_type != null
      params:
        # ... hundreds of entries ...
      source: |-
        def mapping = params[ctx.vendor.event_type];
        if (mapping == null) {
          return;
        }
        ctx.event.category = mapping.category;
        ctx.event.type = mapping.type;
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 }}}'

Reference: okta (system) uses ecs_category_type.yml as a dedicated sub-pipeline.

Anti-patterns — do NOT use these

Bulk `append` processors: Using 2 append processors per event type (one for event.category, one for event.type) creates 50+ processors for 25 event types. The pipeline becomes hard to read, review, and maintain. Each processor re-evaluates its if condition against the same field. Use Pattern A instead.

Inline Painless `if`/`else` chains without `params`: Writing hardcoded string comparisons directly in the script source (e.g., if ('login'.equals(act)) { ctx.event.category = ['authentication']; }) defeats Painless compilation caching because the script body is unique to the integration. It is also harder to maintain than a data-driven params table. Always put lookup data in params.

Selection tips

• Do not use `script` when a built-in processor can do the job — see SKILL.md → Painless script best practices for the mandatory checklist.
• Add tag on every processor (not only processors that can fail).
• For optional data paths, prefer guarded if over ignore_missing.
• Keep parsing, normalization, and enrichment stages visually grouped.

Processor performance guide

Processor cost ordering

Processors vary dramatically in cost. Order pipelines so cheap operations run first and expensive ones run only when needed.

Always guard geoip and user_agent with an if condition

Even with ignore_missing: true, the geoip processor performs expensive database setup and lookup before checking whether the source field exists. This is an Elasticsearch performance issue where the missing-field check happens too late in the execution path. Always add an if condition to check field existence before the processor runs:

- geoip:
    tag: geoip_source_ip
    if: ctx.source?.ip != null
    field: source.ip
    target_field: source.geo
    ignore_missing: true

The if guard prevents the expensive lookup from executing at all. Without it, every document pays the lookup cost regardless of whether the field exists.

Skip enrichment when results already exist

Beyond guarding with a null check on the source field, skip enrichment entirely when the target is already populated. This avoids redundant lookups on documents that were pre-enriched upstream.

- geoip:
    tag: geoip_source_ip
    if: ctx.source?.ip != null && ctx.source.geo == null
    field: source.ip
    target_field: source.geo

Batch remove operations

Use a single remove with a field list instead of multiple separate processors. Each processor invocation has fixed overhead; batching eliminates it.

- remove:
    field: [_tmp, json, message]
    ignore_missing: true
    tag: cleanup_temp_fields

Use equalsIgnoreCase instead of toLowerCase

In Painless if conditions, equalsIgnoreCase avoids allocating a throwaway lowercase string on every document.

// Preferred -- no allocation
if (ctx.event?.action?.equalsIgnoreCase('login') == true)

// Avoid -- allocates a new string per document
if (ctx.event?.action?.toLowerCase() == 'login')

Prefer rename over set + remove

rename is a single atomic operation. Using set with copy_from followed by remove requires two processors and introduces a window where both source and target fields coexist.

# Preferred -- single operation
- rename:
    tag: rename_json_user
    field: json.user
    target_field: user.name
    ignore_missing: true

# Avoid -- two operations for the same result
- set:
    tag: set_user_name
    field: user.name
    copy_from: json.user
- remove:
    tag: remove_json_user
    field: json.user
    ignore_missing: true

Foreach semantics and _ingest._value

How foreach works

The foreach processor iterates over elements of an array field. Inside the inner processor, _ingest._value is a loop variable representing the current element -- it is not a real document field.

- foreach:
    tag: foreach_event_items
    field: event.items
    processor:
      append:
        tag: append_related_user
        field: related.user
        value: '{{{_ingest._value.name}}}'

Resolving _ingest._value references

To understand the actual data flow, resolve _ingest._value back to the foreach field:

Subfield access within foreach

When iterating over an array of objects, access subfields with dot notation after _ingest._value:

- foreach:
    tag: foreach_json_network_connections
    field: json.network_connections
    processor:
      convert:
        tag: convert_port
        field: _ingest._value.port
        type: long
        ignore_missing: true

The actual field being converted is json.network_connections[*].port.

Painless scripts inside foreach

In a script processor nested inside foreach, access the current element via ctx._ingest._value (not ctx.field[i]):

def val = ctx._ingest._value;
if (val.containsKey('ip')) {
    // val.ip refers to foreach_field[*].ip
}

Modifications to _ingest._value subfields mutate the original array element in place.

Common mistakes

• Treating _ingest._value as a real document field path outside a foreach context.
• Forgetting that writes to _ingest._value subfields modify the source array element in place.
• Using ctx._ingest._value outside a foreach block (it does not exist there).

Condition patterns

If-clause fields as implicit inputs

Fields referenced in processor if conditions are implicit inputs that influence whether a transformation runs. They must be tracked alongside explicit source/target fields for data lineage.

- set:
    tag: set_event_category
    field: event.category
    value: [authentication]
    if: ctx.json?.event_type == 'login'

Here json.event_type is an input -- it determines whether event.category gets written.

Painless null-safe condition patterns

Do not use inequality operators (<, >, <=, >=) with null-safe ?. results. The ?. operator returns a def type when the path is missing, and def is not an orderable type -- inequality comparisons fail. Equality checks (==, !=) work because all types support equality. Use != or add an explicit null guard before any inequality.

For nested fields, chain null-safe access: ctx.source?.geo?.country_name != null.

Use .contains() instead of chained OR conditions

When checking a field against multiple values, use .contains() instead of chained ||. For 3+ values, define the list in params to avoid allocating a new array on every document:

- script:
    tag: check_user_role
    params:
      privileged_names:
        - "admin"
        - "system"
        - "root"
        - "service"
    source: |
      if (params.privileged_names.contains(ctx.user?.name)) {
        ctx.user.privileged = true;
      }

For 1-2 values, inline comparison is fine -- the allocation overhead is negligible:

if (ctx.event?.action == 'login' || ctx.event?.action == 'logout') { ... }

Avoid long chained OR conditions regardless of approach:

// Avoid -- verbose and error-prone
ctx.user?.name == 'admin' || ctx.user?.name == 'system' || ctx.user?.name == 'root' || ctx.user?.name == 'service'

Mustache field references in conditions

In set and append processor value fields, Mustache {{{field}}} references are inputs that provide the value being written. Use triple braces to disable HTML escaping.

- set:
    tag: set_host_id
    field: host.id
    value: '{{{crowdstrike.aid}}}'

crowdstrike.aid is the input, host.id is the output.

Nested condition and value references

When a processor has both an if condition and a Mustache value reference, both are inputs:

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

Inputs: source.ip (from both the condition and the value reference). Output: related.ip.

Field transform pitfalls

set with copy_from vs Mustache value

Prefer copy_from over a Mustache value when copying fields -- it avoids string coercion and preserves the original type (objects, arrays, numbers):

- set:
    tag: set_event_original
    field: event.original
    copy_from: message
    if: ctx.tags?.contains('preserve_original_event') == true

Guard copy_from usage to avoid overwriting an existing value. The event.original preservation pattern should check whether the target is already set:

- rename:
    tag: rename_message
    field: message
    target_field: event.original
    ignore_missing: true
    if: ctx.event?.original == null
- remove:
    tag: remove_message
    field: message
    ignore_missing: true
    if: ctx.event?.original != null

convert with type: ip must have a downstream consumer

A convert to IP type is not useful unless something downstream consumes the result (GeoIP enrichment, community_id, or an IP-typed mapping). A bare in-place conversion with no consumer is dead code -- or is missing a target_field that feeds enrichment.

# Correct -- conversion feeds GeoIP enrichment downstream
- convert:
    tag: convert_source_ip
    field: source.ip
    type: ip
    ignore_missing: true
    on_failure:
      - append:
          field: error.message
          value: >-
            Processor {{{_ingest.on_failure_processor_type}}}
            with tag '{{{_ingest.on_failure_processor_tag}}}'
            failed: {{{_ingest.on_failure_message}}}
- geoip:
    tag: geoip_source_ip
    field: source.ip
    target_field: source.geo
    if: ctx.source?.ip != null

Empty string guard before convert

An empty string fails convert with a confusing error. Always guard:

- convert:
    field: json.severity
    type: long
    ignore_missing: true
    if: ctx.json?.severity != ''
    tag: convert_severity

append for related.* fields

Always set allow_duplicates: false when appending to related.* arrays to avoid bloated documents:

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

Non-grok parsing processors

dissect vs grok decision criteria

Use dissect when the delimiter structure is fixed and predictable -- it is faster because it does not use regex. Use grok when token boundaries vary or when you need regex-based extraction.

# Dissect -- fixed delimiters, faster
- dissect:
    tag: dissect_message
    field: message
    pattern: "%{ts} %{+ts} %{log_level} [%{thread}] %{class} - %{msg}"

# Grok -- variable format, regex needed
- grok:
    tag: grok_message
    field: message
    patterns:
      - '^%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} \[%{DATA:component}\] %{GREEDYDATA:msg}'

Dissect supports modifiers: %{+field} (append), %{+field/order} (ordered append), %{?skip} (discard), %{*key}/%{&value} (dynamic key-value).

json processor: add_to_root and conflict strategy

When the parsed JSON should merge into the document root rather than a namespace, use add_to_root. Control collision behavior with add_to_root_conflict_strategy:

- json:
    field: message
    add_to_root: true
    add_to_root_conflict_strategy: replace
    tag: json_merge_to_root

Prefer target_field over add_to_root in most integrations to avoid polluting the root namespace and accidentally overwriting fields. Use add_to_root only when the JSON payload IS the event structure.

date timezone patterns

Avoid short timezone abbreviations (PST, EST) -- they are ambiguous across JDKs. Use full IANA names or UTC offsets.

- date:
    tag: date_json_local_time
    field: json.local_time
    target_field: '@timestamp'
    formats:
      - "dd/MM/yyyy HH:mm:ss"
    timezone: "Europe/Amsterdam"
    if: ctx.json?.local_time != null

Dynamic timezone from a document field:

- date:
    tag: date_json_timestamp
    field: json.timestamp
    formats: [ISO8601]
    timezone: '{{{json.tz}}}'

Multiple format arrays let the processor try each format in order:

- date:
    tag: date_json_timestamp
    field: json.timestamp
    target_field: '@timestamp'
    formats:
      - ISO8601
      - "yyyy-MM-dd HH:mm:ss"
      - UNIX
    if: ctx.json?.timestamp != null

date with on_failure

Date parsing failures should capture the error and remove the unparseable field to prevent downstream confusion:

- date:
    field: json.eventTime
    tag: date_parse_eventTime
    formats: [ISO8601]
    on_failure:
      - remove:
          field: json.eventTime
      - append:
          field: error.message
          value: >-
            Processor {{{_ingest.on_failure_processor_type}}}
            with tag '{{{_ingest.on_failure_processor_tag}}}'
            failed: {{{_ingest.on_failure_message}}}

kv processor patterns

Use field_split and value_split to define the delimiters. Use target_field to namespace the output and prefix to avoid field name collisions:

- kv:
    tag: kv_json_data
    if: ctx.json?.data != null && ctx.json.data != ''
    field: json.data
    target_field: parsed
    field_split: '&'
    value_split: '='
    trim_value: '"'

- kv:
    tag: kv_message
    field: message
    field_split: ' '
    value_split: '='
    prefix: 'vendor.product.'
    target_field: _temp.kv

uri_parts usage

Decomposes a URL string into scheme, host, port, path, query, and fragment:

- uri_parts:
    tag: uri_parts_url_original
    field: url.original
    target_field: url
    keep_original: true
    if: ctx.url?.original != null

dot_expander usage

Expands field names containing literal dots into nested objects. Required when source data uses dotted keys (e.g., host.name as a flat key rather than a nested object):

- dot_expander:
    tag: dot_expander_json
    field: '*'
    path: json
    if: ctx.json != null

Use path to scope expansion to a specific sub-tree and avoid unintended side effects on the root document.

Enrichment depth

community_id implicit inputs

The community_id processor reads several fields by convention without an explicit field parameter. All must be populated before the processor runs:

• source.ip, source.port
• destination.ip, destination.port
• network.transport (protocol name) or network.iana_number (protocol number)

- community_id:
    ignore_missing: true
    if: ctx.source?.ip != null && ctx.destination?.ip != null
    tag: add_community_id

Missing transport/iana_number fields cause the processor to fall back to a default protocol, which may produce incorrect hashes.

registered_domain targeting and output structure

The processor splits an FQDN into its registered domain, top-level domain, and subdomain. Point field at the FQDN source and target_field at the parent object that should receive the decomposed fields:

- registered_domain:
    tag: registered_domain_dns_question_name
    field: dns.question.name
    target_field: dns.question
    if: ctx.dns?.question?.name != null

Output fields written under target_field:

• registered_domain -- the registered domain (e.g., example.com)
• top_level_domain -- the TLD (e.g., com)
• subdomain -- the subdomain portion (e.g., www)

Enrich policy lookup

The enrich processor joins external data into the document via a pre-built enrich policy. Always guard with an if condition and use a temporary target to control which fields are promoted:

- enrich:
    tag: enrich_host_ip
    policy_name: hosts-policy
    field: host.ip
    target_field: _temp.enrich
    if: ctx.host?.ip != null

After the enrich processor, selectively copy needed fields from _temp.enrich into their final locations and remove the temporary object.

Control flow semantics

terminate processor for failure paths

Use terminate to stop the pipeline chain early for documents marked as failures. Without it, failed documents continue through expensive enrichment processors and produce confusing partial output.

- set:
    tag: set_event_kind
    field: event.kind
    value: pipeline_error
- terminate:
    tag: terminate_pipeline_error
    if: ctx.event?.kind == 'pipeline_error'

Place terminate immediately after setting the failure marker in on_failure blocks.

drop conditional patterns

drop silently discards the document. Always guard with an if condition -- an unconditional drop deletes everything.

- drop:
    if: ctx.event?.action == 'heartbeat'
    tag: drop_heartbeat
    description: Discard periodic heartbeat events

Common uses: filtering out health-check events, deduplication markers, or noise events that provide no analytical value.

fail for input validation

Use fail to enforce preconditions early in the pipeline. This produces a clear error message rather than letting the document fail cryptically downstream.

- fail:
    if: ctx.event?.kind == null
    message: 'event.kind is required but missing'
    tag: require_event_kind

Foreach and pipeline chaining edge cases

When a foreach contains a pipeline processor call, each array element is processed by the full sub-pipeline. Be aware that:

• The sub-pipeline sees the entire document context, not just the array element.
• _ingest._value is accessible in the sub-pipeline's processors.
• Errors in the sub-pipeline for one element do not automatically skip remaining elements unless ignore_failure: true is set on the foreach.
• Deeply nested foreach-pipeline chains are hard to debug; prefer flattening where possible.

Special processors

fingerprint

Generates a deterministic hash from one or more fields. Commonly used to set _id for deduplication.

- fingerprint:
    tag: fingerprint_event_id
    fields:
      - event.id
      - '@timestamp'
    target_field: _id
    method: SHA-256

Key considerations:

• fields is an array -- field order matters (changing order changes the hash).
• If any field in fields is missing, the hash changes compared to when it is present. Guard with if or ensure fields always exist.
• target_field: _id enables upsert-style deduplication in Elasticsearch.
• Prefix _id with a timestamp for better index write performance when using time-series data.

html_strip

Removes HTML tags from a field value.

- html_strip:
    tag: html_strip_message
    field: message
    target_field: message_clean
    if: ctx.message != null

When target_field is absent, the processor writes back to field, destroying the original HTML. Use a separate target_field if the raw value must be preserved.

url_decode

Decodes percent-encoded URL strings (e.g., %20 becomes a space).

- url_decode:
    tag: url_decode_url_query
    field: url.query
    if: ctx.url?.query != null

Like html_strip, writes back to field when no target_field is set.

network_direction

Determines whether network traffic is inbound, outbound, or internal based on source/destination IPs and configured internal networks.

Implicit input fields (must be populated before the processor runs):

• source.ip
• destination.ip
• network.type (optional -- IPv4 vs IPv6 hint)

- network_direction:
    tag: network_direction
    internal_networks:
      - loopback
      - private
    if: ctx.source?.ip != null && ctx.destination?.ip != null

Output defaults to network.direction. Use internal_networks_field to read the network list from a document field instead of hardcoding it.

set_security_user

Copies the authenticated user's information from the Elasticsearch security context into a document field. There is no explicit input field -- the processor reads from the indexing user's auth context.

- set_security_user:
    tag: set_security_user
    field: user
    properties:
      - username
      - roles
      - full_name

Typically used in monitoring or audit pipelines where the indexing user's identity must be recorded. The properties list controls which attributes are copied; omitting it copies all available properties.