ingest-pipelines — Skillopedia

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 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 for a data strea…

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n\n # Incorrect — unanchored, scans the whole string for a partial match\n patterns:\n - '%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}'\n ```\n- avoid unnecessary backtracking-heavy custom regex\n- add a `tag` to every grok (and every other) processor\n\nFor grok syntax (three expression forms, inline regex, type coercion, `pattern_definitions`), syslog header splitting recipes, and common mistakes, see `references/grok-recipes.md`.\n\n## Prohibited patterns\n\nThese patterns exist in many legacy integrations but **must not** be used in new or updated pipelines. Do not copy them from reference integrations.\n\n### Never set `event.ingested`\n\nThe `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:\n\n```yaml\n# PROHIBITED — do not use\n- set:\n field: event.ingested\n value: '{{{_ingest.timestamp}}}'\n```\n\nThe pipeline **should** set `@timestamp` from the original event's timestamp. When the source data contains multiple timestamps, map them as follows:\n\n- **`@timestamp`**: the primary event timestamp parsed from the source data. This is required.\n- **`event.created`**: when the event was first created or recorded by the source system (if different from `@timestamp`).\n- **`event.start`**: when an activity or period began (e.g., session start, connection start).\n- **`event.end`**: when an activity or period ended (e.g., session end, connection close).\n\nIf 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`.\n\n### Never use `preserve_duplicate_custom_fields`\n\nThe `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.\n\nInstead, follow these field mapping rules:\n- When a source field maps to an ECS field, use `rename` to move it directly. The source field is removed and no duplicate exists.\n- 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.\n- **Never design a pipeline that needs to preserve both the original vendor field and the ECS copy.** The ECS field is the canonical location.\n\nIf you encounter this pattern in a reference integration, ignore it — it is legacy.\n\n### Never add an `event.original` removal processor at the end\n\nAs 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.\n\n## References\n\n- `references/processor-cookbook.md` — processor selection, parsing/normalization/enrichment examples, ECS categorization mapping patterns (Pattern A/B/C + anti-patterns)\n- `references/branching-patterns.md`\n- `references/error-handling-patterns.md`\n- `references/grok-recipes.md` — grok syntax, type coercion, syslog header recipes, common mistakes, pattern library link\n- `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.\n- `references/cdr-pipeline-requirements.md` — cloud security / CDR integrations only (`cloudsecurity_cdr` category): event categorization, Must Have / Should Have fields, processor patterns, review checklist\n- `entity-mappings/references/entity-pipeline-patterns.md` — entity/inventory data streams only (`event.kind: asset`): categorization processors, `entity.id` mirroring, boolean coercion, relationship objects, anti-patterns, pipeline review checklist","attachment_filenames":["references/branching-patterns.md","references/builder-subagent-guidance.md","references/cdr-pipeline-requirements.md","references/error-handling-patterns.md","references/grok-recipes.md","references/painless-patterns.md","references/processor-cookbook.md"],"attachments":[{"filename":"references/branching-patterns.md","content":"# branching patterns for ingest pipelines\n\nUse this guide when one linear parser is not enough and you need conditional or staged routing.\n\n## When to branch\n\nBranch when at least one of these is true:\n- the same stream receives multiple formats (for example JSON and plain text)\n- different event classes need different object mapping logic\n- complex parsing becomes difficult to review in one large `default.yml`\n- array items require per-element pipeline processing (`foreach` + `pipeline`)\n\nKeep single-path pipelines for simple, uniform formats.\n\n## Branching primitives\n\n### `pipeline` processor\n\n```yaml\n- pipeline:\n name: '{{ IngestPipeline \"pipeline_branch_name\" }}'\n if: ctx.some?.field != null\n ignore_missing_pipeline: true\n tag: route_branch_name\n```\n\nUse:\n- `name` via `{{ IngestPipeline \"...\" }}` for package-aware pipeline naming\n- `if` guard to route selectively\n- `ignore_missing_pipeline: true` if branch presence may vary by package version\n- `tag` for failure diagnostics\n\n### `foreach` + `pipeline` for arrays\n\n```yaml\n- foreach:\n field: ocsf.resources\n ignore_missing: true\n processor:\n pipeline:\n name: '{{ IngestPipeline \"pipeline_resources_data_json\" }}'\n```\n\nUse this when each item in an array needs repeated parse/normalize logic.\n\n## Naming conventions\n\nRecommended conventions:\n- `default.yml` as orchestrator/router\n- `pipeline_parser_\u003cformat>.yml` for format parsers\n- `pipeline_object_\u003cobject>.yml` for object mapping\n- `pipeline_category_\u003ccategory>.yml` for category-level transforms\n- `pipeline_enrichment_\u003ctopic>.yml` for enrichment-only branches\n\nNaming should describe branch intent, not source implementation detail.\n\n## Common branching topologies\n\n### 1) Two-way format split\n\nUse when input may be JSON or text:\n\n```yaml\n- pipeline:\n name: '{{ IngestPipeline \"pipeline_parser_json\" }}'\n if: ctx.event?.original != null && ctx.event.original.startsWith('{')\n ignore_missing_pipeline: true\n tag: route_parser_json\n- pipeline:\n name: '{{ IngestPipeline \"pipeline_parser_text\" }}'\n if: ctx.event?.original != null && !ctx.event.original.startsWith('{')\n ignore_missing_pipeline: true\n tag: route_parser_text\n```\n\n### 2) Category fan-out\n\nUse when event category determines transform behavior:\n\n```yaml\n- pipeline:\n name: '{{ IngestPipeline \"pipeline_category_system_activity\" }}'\n if: ctx.ocsf?.category_uid == '1'\n ignore_missing_pipeline: true\n tag: route_category_system_activity\n- pipeline:\n name: '{{ IngestPipeline \"pipeline_category_network_activity\" }}'\n if: ctx.ocsf?.category_uid == '4'\n ignore_missing_pipeline: true\n tag: route_category_network_activity\n```\n\n### 3) Object-based branch with class guard\n\nUse when only some classes contain a specific object:\n\n```yaml\n- pipeline:\n name: '{{ IngestPipeline \"pipeline_object_user\" }}'\n if: ctx.ocsf?.class_uid != null && ['2005','3001','3002','3003'].contains(ctx.ocsf.class_uid) && ctx.ocsf.user != null\n ignore_missing_pipeline: true\n tag: route_object_user\n```\n\n### 4) Multi-branch graph (large integration pattern)\n\nThis is the pattern used by `amazon_security_lake`:\n- `default.yml` does common parse/normalization\n- object pipelines run conditionally by `class_uid` + object presence\n- category pipelines run by `category_uid`\n- some branches apply nested `foreach` processing\n\nThis yields maintainable sub-pipelines instead of one very large file.\n\n## Design rules\n\n- Keep `default.yml` focused on shared setup and routing.\n- Ensure each sub-pipeline is safe to run independently (with guards).\n- Prefer mutually intelligible branch conditions over deeply nested script logic.\n- If multiple branches can run on one event, make processor side effects explicit.\n- Add `description` in every pipeline file so intent is obvious in reviews.\n\n## Validation checklist\n\n- All routed sub-pipeline names resolve correctly through `IngestPipeline`.\n- Branch conditions are null-safe (`ctx.a?.b != null` style).\n- Branches do not overwrite each other unexpectedly.\n- Pipeline tests include one fixture per route and one route-miss fixture.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":4126,"content_sha256":"d1dbf1787cb8991dd5a6638884321e392d18ac1de99f14fe36a82532bd139002"},{"filename":"references/builder-subagent-guidance.md","content":"# Ingest pipeline builder subagent guidance\n\nOperating manual for a subagent building or fixing ingest pipelines on behalf of the\n`create-integration` or `maintain-integration` orchestrator.\n\nThe orchestrator dispatches you with a brief task prompt that points you at this\nfile by path. **Read this entire file end-to-end before doing any other work**,\nthen read the skills and reference files listed in the \"First steps\" section\nbelow — they are mandatory. The orchestrator does not paste this file's content\ninto your task prompt (to avoid burning context twice); you load it here in your\nown fresh context.\n\nThe orchestrator's task prompt tells you **what** to build or fix, **which data stream**\nto work on, **what sample data** is available, and **the package path**. This file tells\nyou **how to operate** as a pipeline builder subagent. Follow both.\n\n## Scope\n\nYour responsibility is strictly limited to:\n\n- Designing and implementing the ingest pipeline (`elasticsearch/ingest_pipeline/`)\n- Defining field mappings (`fields/fields.yml`, `fields/base-fields.yml`, `fields/ecs.yml`, `fields/beats.yml`)\n- Creating pipeline test fixtures (`_dev/test/pipeline/`) and running pipeline tests\n\n**You do NOT**:\n\n- Run system tests (`elastic-package test system`) — the orchestrator delegates those to\n a separate subagent; you do not run them yourself\n- Create or modify `sample_event.json` — this is only generated by\n `elastic-package test system`, run by that delegated test pass\n- Modify CEL programs or `cel.yml.hbs` templates — the CEL program builder handles this\n (see `cel-programs/references/builder-subagent-guidance.md`)\n- Modify system test config files or mock API definitions — the CEL program builder\n handles this for CEL streams; the data-collection setup builder handles this for\n non-CEL streams\n\n## Skill authority\n\nThe rules and patterns in the skills and their reference files are the **authoritative\nsource of truth**. See the skill-authority disclaimer in `ingest-pipelines/SKILL.md`.\nWhen examining existing integrations for reference, many contain legacy patterns that\npredate current standards — **always follow the skills over patterns observed in other\nintegrations**.\n\n## First steps — read the skills and their references\n\nBefore doing any work, read these skill files **and the specific reference files listed**\nto load the domain rules, patterns, and working code examples you must follow. Reading\nonly the SKILL.md files is not sufficient — the reference files contain the actual\nprocessor examples, test fixture formats, and field mapping patterns you need.\n\n1. **`ingest-pipelines` skill** (SKILL.md) — pipeline design patterns, processor\n ordering, error handling, JSE00001 compliance, syslog/KV strategies\n - **`references/processor-cookbook.md`** — MUST READ: processor examples for all\n common operations (JSON parsing, date handling, enrichment, type conversion, ECS\n categorization Patterns A/B/C and their anti-patterns)\n - **`references/error-handling-patterns.md`** — MUST READ: on_failure blocks,\n per-processor error handling, drop/tag strategies\n - **`references/branching-patterns.md`** — read when building pipelines with\n conditional logic or sub-pipelines\n - **`references/grok-recipes.md`** — read when writing grok patterns: syntax, type\n coercion, syslog header recipes, common mistakes\n - **`references/painless-patterns.md`** — read when writing Painless script\n processors: ctx access, params, HashMap for nested writes, foreach script context\n\n2. **`ecs-field-mappings` skill** (SKILL.md) — ECS categorization values, field nesting\n rules, field files (`ecs.yml`, `base-fields.yml`, `fields.yml`), custom field types\n - **`references/categorization-cheatsheet.md`** — MUST READ: valid\n `event.category` / `event.type` combinations\n - **`references/mapping-type-matrix.md`** — field type selection for custom fields\n - **`references/root-and-core-fields.md`** — ECS root and core field definitions\n - **`entity-mappings/references/entity-pipeline-patterns.md` and `entity-mappings/references/entity-field-catalog.md`** — read these **only if** the orchestrator's task prompt states this is an entity data stream, or the data is an inventory/snapshot of users, hosts, devices, groups, or apps rather than a timeline of events. If the entity pin is required (`[email protected]`), the orchestrator's prompt will say so explicitly — do not change the pin otherwise.\n\n3. **`integration-testing` skill** — then **read `references/pipeline-testing.md`\n fully**: test fixture format, config files, expected output generation,\n troubleshooting\n\n4. **`anonymize-logs` skill** — fixture sourcing, sanitization, scenario coverage\n\n5. **`elastic-package-cli` skill** — validation commands and pipeline test loop\n\nRead all skills and their MUST READ references before writing any pipeline code.\n\n## Input data sources\n\nYou receive sample data from one of two paths. **Do not research or re-investigate the\ndata format** when the data is already available from a prior step.\n\n### CEL-first path (data from the CEL program builder)\n\nWhen the orchestrating agent tells you that a CEL program builder subagent (see\n`cel-programs/references/builder-subagent-guidance.md`) has already run:\n\n1. **Mock API response data exists** in the system test config files (e.g.,\n `_dev/deploy/docker/` config files). Read these to understand the data structure the\n CEL program collects.\n2. The orchestrating agent's prompt will describe the data structure, key fields, and the\n mock API response format.\n3. The CEL program builder may have created initial field mappings in `fields/fields.yml`\n — read these to understand what raw fields exist.\n\nUse the mock API response data and the orchestrator's description as your primary input\nfor pipeline design. You will create your own pipeline test fixtures from this\ninformation.\n\n### Direct path (user-provided sample data)\n\nWhen no CEL builder has run (log-based, syslog, file inputs):\n\n1. Examine sample data provided by the user, research brief, or referenced files.\n2. Identify the format: JSON, syslog, CEF, key-value, custom delimited, multiline.\n\nIn both paths, catalog the fields present, their data types, and which map to ECS vs.\ncustom fields.\n\n## Workflow\n\n### 1. Analyze the input data\n\n- Read pipeline test fixtures if they already exist (CEL-first path), or examine\n user-provided samples\n- Catalog the fields present and their data types\n- Identify which fields map to ECS and which are custom\n- Identify the format: JSON (from CEL), syslog, CEF, key-value, custom delimited,\n multiline\n\n### 2. Design the pipeline\n\n- Choose the parsing strategy (grok, dissect, json, kv, or combination)\n- Plan the processor chain following the ordering from the `ingest-pipelines` skill:\n 1. `ecs.version` (`9.3.0`) and, **for CEL streams only**, Agentless `remove` /\n `terminate` when applicable\n 2. `event.original` handling (JSE00001 rename+remove pattern)\n 3. Parsing on `event.original` (grok/dissect/json/kv) — **never mutate\n `event.original`**\n 4. Date normalization (`@timestamp`)\n 5. Type conversions (convert processor)\n 6. Enrichment (geoip, user_agent) only when applicable\n 7. ECS categorization (`event.kind`, `event.category`, `event.type`,\n `event.outcome`) — use Pattern A (script with `params` lookup) for 5+ mappings,\n Pattern B (`set` with conditionals) for fewer than 5, Pattern C (sub-pipeline) for\n 100+. **Do NOT use bulk `append` processors or inline Painless `if`/`else` chains\n without `params`** — these are explicit anti-patterns.\n 8. Cleanup (remove temporary fields; conditional `append` for\n `preserve_original_event` when `ctx.error?.message != null`)\n- For complex formats, design branching with sub-pipelines\n- Prefer **`rename`** over **`set`** when moving values from custom/vendor fields into\n ECS fields\n- Add a **`tag`** to **every** processor\n\n### 3. Implement the pipeline\n\n- Write `data_stream/\u003cstream>/elasticsearch/ingest_pipeline/default.yml`\n- Write sub-pipeline files if branching is needed\n- Include pipeline-level `on_failure` with `event.kind: pipeline_error` and contextual\n `error.message`\n- Follow all patterns from the `ingest-pipelines` skill\n\n### 4. Define field mappings\n\n- Create or update `fields/ecs.yml` with the ECS fields the pipeline sets that\n dynamic mapping cannot infer — use `name` + `external: ecs` per entry, with\n type/value overrides only when needed. See the `ecs-field-mappings` skill\n (`ecs.yml` section) for the `ecs@mappings` coverage rule: on packages whose\n `conditions.kibana.version` floor is >= 8.13.0, only `geo_point` on\n non-standard parent prefixes, `geo_shape`, `nested`, `flattened`, or\n validation-failure cases need declarations; below that floor, declare every\n pipeline-set ECS field. Declaring more is harmless.\n- Create or update `fields/fields.yml` with custom integration-specific fields (non-ECS\n fields only)\n- Verify `fields/base-fields.yml` has the standard data stream routing constants\n using `external: ecs` (all six fields are ECS fields; override `type`/`value`\n only for `event.module` and `event.dataset`)\n- If the input emits Beats/Filebeat-specific fields (`input.type`, `log.offset`,\n `log.flags`) not present in `base-fields.yml`, add them to `fields/beats.yml`\n- Ensure `_dev/build/build.yml` exists at the package root with\n `dependencies.ecs.reference: \"[email protected]\"` for standard streams, or `\"[email protected]\"` for\n entity data streams (those with `event.kind: asset`). The orchestrator will specify which\n to use. Do not change an existing pin without being told to.\n- Follow the field mapping guidance from the `ecs-field-mappings` skill\n\n### 5. Create test fixtures\n\n- Write test input fixtures in `_dev/test/pipeline/` following the\n `integration-testing` skill → `references/pipeline-testing.md`, based on the data\n structure from the mock API responses (CEL path) or user-provided samples (direct path)\n- Use descriptive scenario names (`test-successful-login.log`, `test-malformed-json.log`)\n- Ensure `test-common-config.yml` exists with appropriate `dynamic_fields` for any\n non-deterministic fields the pipeline produces\n- Include `preserve_original_event` in config `fields.tags` when appropriate\n- Cover: happy path, format variants, edge cases, error paths\n\n### 6. Generate pipeline expected output\n\nRun pipeline tests with generation to produce or refresh expected files:\n\n```bash\nelastic-package test pipeline --data-streams \u003cstream> --generate\n```\n\n### 7. Review generated output before accepting it\n\n**Manually inspect** every file produced or updated by step 6 — primarily\n`*-expected.json` — and decide whether the pipeline output is **actually correct**, not\nmerely consistent with a buggy parser.\n\nCheck at minimum:\n\n- ECS fields are present and correctly populated\n- Dotted source fields are expanded into nested objects\n- geo_point fields appear under correct parent entities\n- `event.category` and `event.type` are arrays (not scalars)\n- `@timestamp` is parsed from source data (not only the ingest timestamp)\n- No unexpected fields, missing values, or wrong types\n- `event.original` is present when `preserve_original_event` is set in test config\n- Values match the intent of each fixture scenario (happy path vs error path)\n\nIf anything is wrong, fix the pipeline and regenerate — **never hand-edit expected\nJSON**.\n\n### 8. Run pipeline tests (no `--generate`)\n\nConfirm the reviewed expected files match the pipeline:\n\n```bash\nelastic-package test pipeline --data-streams \u003cstream>\n```\n\n### 9. Validate build\n\nRun the full check sequence:\n\n```bash\nelastic-package format\nelastic-package lint\nelastic-package check\n```\n\nFix any lint or build errors before reporting back.\n\n**Do not run system tests** — the orchestrator delegates them to a dedicated subagent\nafter pipeline work completes. Do not create or modify `sample_event.json` — it is\ngenerated exclusively by `elastic-package test system` in that delegated pass.\n\n## Data anonymization\n\n**All data committed to the repository must be fully anonymized.** No real production\ndata, customer data, or identifiable information may appear in any committed file:\n\n- **Pipeline test fixtures** (`test-*.log`, `test-*.json`): replace real IP addresses,\n hostnames, email addresses, usernames, organization names, account IDs, tokens, and any\n other traceable data with synthetic examples of the same format.\n- **Generated expected output** (`*-expected.json`): ensure source fixtures are\n anonymized first.\n\nUse RFC 5737 documentation IP ranges (`198.51.100.x`, `203.0.113.x`), `example.com`\ndomains, realistic placeholder names. Refer to the `anonymize-logs` skill for the full\nplaceholder convention list.\n\n## What to return\n\nWhen you finish, report:\n\n- Files created or modified (with paths)\n- Field files created or modified: list each field file and what was added/changed, so\n the orchestrating agent knows which field definitions are already handled\n- Pipeline architecture summary (single-path or branching, parser strategy)\n- ECS categorization choices made and rationale\n- Test coverage: scenarios covered, number of test events\n- Pipeline test results: `elastic-package test pipeline` pass/fail (after step 8)\n- Generated output review: confirm `*-expected.json` files were **read and validated** in\n step 7, not only that the command succeeded\n- Validation results from `elastic-package format/lint/check`\n- Any open issues or decisions that need user input\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":13622,"content_sha256":"02f6d0d2730123f882f3ec444debb173f9f03708f06ce195dd2694ecce25e7f9"},{"filename":"references/cdr-pipeline-requirements.md","content":"# CDR pipeline requirements\n\nCloud 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.\n\nAligned with: Elastic CDR 3P Developer Guide v1.0\n\n## Event categorization\n\nCorrect `event.*` values are critical -- the Kibana Findings page filters on them.\n\n| Finding type | `event.kind` | `event.category` | `event.type` |\n|-------------|-------------|------------------|-------------|\n| Misconfiguration | `state` | `configuration` | `info` |\n| Vulnerability | `state` | `vulnerability` | `info` |\n| Runtime detection | `alert` | varies | varies |\n\nUse `append` for `event.category` and `event.type` (they are arrays in ECS). Use `set` for `event.kind` (single value).\n\n## Must Have fields (pipeline must populate)\n\nMissing these causes critical issues in the Kibana Findings UI.\n\n| Field | Purpose |\n|-------|---------|\n| `resource.id` | Cloud resource ID (e.g., ARN). Transform uniqueness depends on it |\n| `resource.name` | Human-readable resource name. Default data grid column |\n| `result.evaluation` | `passed`, `failed`, or `unknown` (misconfiguration only) |\n| `rule.name` | Rule name for misconfiguration findings |\n| `rule.uuid` | Unique rule identifier for transform uniqueness |\n| `observer.vendor` | Vendor name (e.g., Wiz, Amazon) |\n| `event.id` | Unique event identifier for multi-value grouping |\n| `user.name` | For user-related findings (entity correlation) |\n| `host.name` | For host-related findings (entity correlation) |\n\nFor vulnerability findings additionally:\n\n| Field | Purpose |\n|-------|---------|\n| `vulnerability.id` | CVE ID |\n| `vulnerability.severity` | `Low`, `Medium`, `High`, `Critical`, or `None` |\n| `vulnerability.score.base` | CVSS base score |\n| `vulnerability.title` | Human-readable vulnerability title |\n| `package.name` | Affected package name |\n\n## Should Have fields\n\n| Field | Purpose |\n|-------|---------|\n| `cloud.provider` | Lowercase: `aws`, `gcp`, `azure` |\n| `cloud.account.id` | Account/project/subscription ID |\n| `cloud.region` | Region or location |\n| `cloud.service.name` | Service generating the finding |\n| `event.outcome` | `failure`, `success`, or `unknown` (mirrors `result.evaluation`) |\n| `resource.type` | Resource type identifier |\n| `resource.sub_type` | Resource sub-type |\n| `rule.description` | Rule description for flyout |\n| `rule.remediation` | Remediation steps |\n\n## Correlation fields\n\nCDR integrations MUST populate `related.*` fields for threat hunting:\n\n- `related.ip` -- resource IPs, actor IPs\n- `related.user` -- IAM users, service accounts\n- `related.hash` -- artifact hashes (for CWPP findings)\n\n## Value transformations\n\n- **Severity** -- must be lowercase: `CRITICAL` -> `critical`, `VERY_HIGH` -> `critical`, `MODERATE` -> `medium`\n- **Status** -- map vendor values: `ACTIVE` -> `failed`, `INACTIVE`/`ARCHIVED` -> `passed`/`resolved`\n- **Account IDs** -- extract from resource paths: `projects/my-project` -> `my-project`\n\n## Pipeline patterns\n\n### Event categorization for misconfiguration\n\n```yaml\n- set:\n tag: set_event_kind\n field: event.kind\n value: state\n- append:\n tag: append_event_category\n field: event.category\n value: configuration\n- append:\n tag: append_event_type\n field: event.type\n value: info\n```\n\n### Event categorization for vulnerability\n\n```yaml\n- set:\n tag: set_event_kind\n field: event.kind\n value: state\n- append:\n tag: append_event_category\n field: event.category\n value: vulnerability\n- append:\n tag: append_event_type\n field: event.type\n value: info\n```\n\n### Result evaluation mapping\n\n```yaml\n- set:\n tag: set_result_evaluation\n field: result.evaluation\n value: failed\n if: ctx.json?.compliance_status == \"NON_COMPLIANT\"\n- set:\n tag: set_result_evaluation\n field: result.evaluation\n value: passed\n if: ctx.json?.compliance_status == \"COMPLIANT\"\n```\n\n### Observer vendor (constant per integration)\n\n```yaml\n- set:\n tag: set_observer_vendor\n field: observer.vendor\n value: \"Wiz\"\n```\n\n### Cloud context\n\n```yaml\n- set:\n tag: set_cloud_provider\n field: cloud.provider\n value: aws\n- set:\n tag: set_cloud_account_id\n field: cloud.account.id\n value: '{{{json.account_id}}}'\n```\n\n### Vulnerability fields (conditional on finding type)\n\n```yaml\n- set:\n tag: set_vulnerability_id\n field: vulnerability.id\n copy_from: json.cve_id\n if: ctx.json?.finding_type == 'VULNERABILITY'\n- set:\n tag: set_vulnerability_severity\n field: vulnerability.severity\n value: '{{{json.severity}}}'\n```\n\n## Known CDR integrations\n\n`aws_security_hub`, `google_scc`, `azure_security_center`, `azure_defender`, `prisma_cloud`, `crowdstrike`, `wiz`, `orca`, `snyk`, `lacework`, `tenable`, `qualys`, `rapid7`, `sentinelone`, `sysdig`\n\nDetection 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`\n\n## CDR pipeline review checklist\n\n### What to flag\n\n- [ ] Missing `event.category`/`event.kind`/`event.type` or wrong values for the finding type -- **HIGH**\n- [ ] Missing `resource.id` or `resource.name` (breaks transform and UI) -- **HIGH**\n- [ ] Missing `result.evaluation` for misconfiguration findings -- **HIGH**\n- [ ] Missing `rule.uuid` (breaks transform uniqueness) -- **HIGH**\n- [ ] `event.kind` not set to `state` for posture findings -- **HIGH**\n- [ ] No `related.*` fields populated -- **MEDIUM**\n- [ ] Missing `observer.vendor` -- **HIGH**\n- [ ] Severity not lowercase -- **MEDIUM**\n- [ ] Vulnerability fields applied to non-vulnerability findings (must be conditional) -- **MEDIUM**\n- [ ] Using `cloud.detection.*` (NOT part of CDR spec -- use `rule.*`, `result.*`) -- **MEDIUM**\n\n### What NOT to flag\n\nCDR pipeline requirements are NOT applicable to:\n\n- General logging or metrics integrations\n- APM integrations\n- Non-security cloud integrations (e.g., billing, resource inventory without security posture)\n- Integrations that do not produce security findings (misconfiguration, vulnerability, or runtime detection)\n\nDo not flag missing CDR fields on these integration types.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":6356,"content_sha256":"13ea42088ecfc097fbfb51f0cecd71d85ff1bfcd525a5b2b49011df8cda76bb8"},{"filename":"references/error-handling-patterns.md","content":"# ingest pipeline error handling patterns\n\nThis guide focuses on resilient failure behavior and actionable debugging output.\n\n## Recommended top-level `on_failure`\n\nUse this pattern in `default.yml` to ensure all uncaught failures are visible:\n\n```yaml\non_failure:\n - append:\n field: error.message\n value: >-\n Processor '{{{ _ingest.on_failure_processor_type }}}'\n {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'\n {{{/_ingest.on_failure_processor_tag}}}in pipeline '{{{ _ingest.pipeline }}}'\n failed with message '{{{ _ingest.on_failure_message }}}'\n - set:\n field: event.kind\n tag: set_pipeline_error_to_event_kind\n value: pipeline_error\n - append:\n field: tags\n value: preserve_original_event\n allow_duplicates: false\n```\n\nWhy:\n- Appending `error.message` first preserves the full `_ingest.on_failure_*` context for triage.\n- `event.kind: pipeline_error` supports clean filtering and dashboards.\n- `preserve_original_event` on failure helps post-failure diagnostics.\n\n## Processor tag requirement\n\n**Every** processor in the pipeline should include a `tag` (not only processors that can fail).\n\n```yaml\n- grok:\n field: event.original\n patterns: ['^%{IP:source.ip} %{GREEDYDATA:message}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

]\n tag: parse_source_ip\n```\n\nWithout tags, error messages lose key context and triage becomes slower.\n\n## Processor-level `on_failure`: when and when not\n\nUse processor-level `on_failure` for:\n- cleanup (`remove` invalid temporary fields)\n- fallback operations on known parse failures\n- local annotations that complement top-level errors\n\nAvoid using processor-level `on_failure` as the only error-reporting path.\nIf a processor `if` expression itself fails, control may bypass that local handler and fall through to top-level `on_failure`.\n\n### Example: date parse fallback detail\n\n```yaml\n- date:\n field: nginx.access.time\n target_field: '@timestamp'\n formats:\n - dd/MMM/yyyy:H:m:s Z\n tag: parse_access_time\n on_failure:\n - append:\n field: error.message\n value: '{{{_ingest.on_failure_message}}}'\n```\n\n## `ignore_failure` usage\n\nUse `ignore_failure: true` only when failure should not block ingestion.\n\nGood candidates:\n- optional enrichment (`geoip`, `user_agent`)\n- best-effort cleanup/normalization\n- non-critical parsing of auxiliary fields\n\nExample:\n\n```yaml\n- geoip:\n field: source.ip\n target_field: source.geo\n ignore_failure: true\n ignore_missing: true\n tag: enrich_source_geo\n```\n\nAvoid `ignore_failure` on required parse steps that define event shape.\n\n## `fail` processor usage\n\nUse `fail` for invalid required input or unrecoverable branch conditions.\n\n### Input validation gate\n\n```yaml\n- fail:\n if: ctx.json == null || !(ctx.json instanceof Map)\n message: missing json object in input document\n tag: validate_json_input\n```\n\n### Critical parser escalation (inside `on_failure`)\n\n```yaml\n- kv:\n field: message\n field_split: ' '\n value_split: '='\n tag: parse_kv\n on_failure:\n - fail:\n message: 'unable to parse key-values: {{{ _ingest.on_failure_message }}}'\n tag: fail_parse_kv\n```\n\n## Pattern catalog\n\n### 1) Minimal pipeline-level error pattern (lightweight)\n\n```yaml\non_failure:\n - set:\n field: error.message\n value: '{{{_ingest.on_failure_message}}}'\n```\n\nUse this only for simple integrations where richer diagnostics are not yet needed.\n\n### 2) Full-context pipeline-level error pattern (preferred)\n\nUse the recommended top-level `on_failure` block from the section above. This is the standard pattern for `default.yml` in all integrations.\n\n### 3) Conditional preserve tag before successful pipeline end\n\nWhen the collector set an error but ingestion continued, tag the document so operators can find it:\n\n```yaml\nprocessors:\n - append:\n field: tags\n tag: append_preserve_on_collector_error\n value: preserve_original_event\n allow_duplicates: false\n if: ctx.error?.message != null\n```\n\n(This is separate from the `on_failure` block; use together with pattern 2.)\n\n### 4) Local cleanup in processor `on_failure`\n\n```yaml\n- json:\n field: event.original\n target_field: json\n tag: parse_json\n on_failure:\n - remove:\n field: json\n ignore_missing: true\n - append:\n field: error.message\n value: '{{{_ingest.on_failure_message}}}'\n```\n\n## Review checklist\n\n- Top-level `on_failure` exists in every primary ingest pipeline.\n- Every processor has a `tag`.\n- Required parse steps do not silently ignore failure.\n- `ignore_failure` is limited to optional/non-critical operations.\n- Error messages include enough context to locate the failed processor.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":4773,"content_sha256":"5b481d1e89c715febeca3d58e3289c2f65fbca305e0e214dfeb5c79a2572401d"},{"filename":"references/grok-recipes.md","content":"# Grok Pattern Recipes\n\n## Pattern library reference\n\nThe authoritative built-in pattern library for Elasticsearch's grok processor lives in the Elasticsearch source tree:\n\n- **Browsable directory (all pattern files):** https://github.com/elastic/elasticsearch/tree/master/libs/grok/src/main/resources/patterns/ecs-v1\n- **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\n\nAlways check those sources for the actual regex behind a pattern before assuming it matches a specific input shape.\n\n---\n\n## Core grok syntax\n\n### Three expression forms\n\n| Form | Syntax | Effect |\n|------|--------|--------|\n| Match only | `%{SYNTAX}` | Matches the pattern; no field created |\n| Named capture | `%{SYNTAX:field.name}` | Matches and stores result in `field.name` |\n| Typed capture | `%{SYNTAX:field.name:TYPE}` | Matches, stores, and coerces to `TYPE` |\n\n```\n# Match only — skip a token\n%{IP} - %{USER:user.name}\n\n# Named capture\n%{IP:source.ip} %{WORD:http.request.method} %{URIPATH:url.path}\n\n# Typed capture\n%{NUMBER:http.response.status_code:int} %{NUMBER:http.response.body.bytes:long}\n```\n\n### Inline regex captures\n\nWhen no built-in pattern fits, embed a raw regex directly:\n\n```\n(?\u003clog.level>DEBUG|INFO|WARN|ERROR|FATAL)\n(?\u003cevent.action>[a-zA-Z0-9_\\-]+)\n```\n\nInline regex and `%{SYNTAX}` references can be freely mixed in the same expression.\n\n---\n\n## Type coercion\n\nBy default every capture is a string. Append a type suffix to coerce:\n\n| Suffix | Result type | Common use |\n|--------|-------------|------------|\n| `int` | 32-bit integer | Response codes, small counts |\n| `long` | 64-bit integer | Bytes, large counters |\n| `double` | 64-bit float | Durations, rates |\n| `float` | 32-bit float | Rarely preferred over `double` |\n| `boolean` | Boolean | `true`/`false` (case-insensitive) |\n\n```\n%{NUMBER:http.response.status_code:int}\n%{NUMBER:http.response.body.bytes:long}\n%{NUMBER:event.duration:double}\n```\n\n---\n\n## Custom patterns with `pattern_definitions`\n\nDefine reusable inline patterns directly in the grok processor (no separate pattern file needed for Elasticsearch pipelines):\n\n```yaml\n- grok:\n field: event.original\n pattern_definitions:\n THREAD_ID: \"[A-Za-z0-9#]+\"\n JAVA_CLASS: \"[a-zA-Z$_][a-zA-Z$_0-9]*(?:\\\\.[a-zA-Z$_][a-zA-Z$_0-9]*)*\"\n patterns:\n - '^%{TIMESTAMP_ISO8601:timestamp} \\[%{THREAD_ID:thread}\\] %{WORD:log.level} %{JAVA_CLASS:logger} - %{GREEDYDATA:message}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n tag: parse_java_log\n```\n\nNaming convention: `SCREAMING_SNAKE_CASE`; prefix with a namespace to avoid collisions: `MYAPP_REQUEST_ID` instead of `REQUEST_ID`.\n\n---\n\n## Syslog header recipes\n\nSyslog 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.\n\n**Rule:** never overwrite `event.original`. Store extracted sub-fields in `_temp.*` when you need to pass them to sub-pipelines.\n\n### RFC 3164 — traditional syslog header\n\n```\nSample:\nJan 15 10:30:00 web-01 sshd[1234]: Accepted publickey for alice from 192.168.1.10 port 55234\n\nPattern:\n^%{SYSLOGTIMESTAMP:timestamp} %{IPORHOST:host.hostname} %{NOTSPACE:process.name}(?:\\[%{POSINT:process.pid:int}\\])?: %{GREEDYDATA:message}$\n\nFields: timestamp, host.hostname, process.name, process.pid\nPayload: message\n```\n\n### RFC 5424 — structured syslog header (no SD-ELEMENT)\n\n```\nSample:\n\u003c34>1 2024-01-15T10:30:00.000Z mymachine.example.com sshd 1234 ID47 - Accepted publickey for alice\n\nPattern:\n^\u003c%{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}$\n```\n\n### RFC 5424 — structured syslog header (with SD-ELEMENT block)\n\n```\nSample:\n\u003c34>1 2024-01-15T10:30:00.000Z mymachine.example.com su - ID47 [exampleSDID@32473 iut=\"3\"] BOM su root failed\n\nPattern:\n^\u003c%{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}$\n```\n\nCapture 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.\n\n### Split header from payload for sub-pipeline routing\n\nWhen `default.yml` is a thin router, extract the envelope but do not clobber `event.original`:\n\n```yaml\n- grok:\n field: event.original\n patterns:\n - '^%{SYSLOGTIMESTAMP:_temp.timestamp} %{IPORHOST:host.hostname} %{NOTSPACE:process.name}(?:\\[%{POSINT:process.pid:int}\\])?: %{GREEDYDATA:_temp.message}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n tag: parse_syslog_header\n```\n\nSub-pipelines then parse `_temp.message` for their specific event format. `_temp` fields are removed at the end of the pipeline.\n\n---\n\n## Syslog structured data strategies\n\nFirewall 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.\n\n### Strategy 1: Grok + KV with `trim_value` (simplest)\n\nWhen 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.\n\nThis pattern is used by integrations like juniper_srx, sophos, and sonicwall_firewall in the upstream `elastic/integrations` repository.\n\n```yaml\n- kv:\n field: _temp.kv_data\n field_split: ' (?=[a-zA-Z0-9_-]+=)'\n value_split: \"=\"\n prefix: \"vendor.product.\"\n trim_value: '\"'\n ignore_missing: true\n tag: kv_structured_data\n```\n\nKey settings:\n- `field_split: ' (?=[a-zA-Z0-9_-]+=)'` — splits on spaces only when followed by a key= pattern, preserving spaces in quoted values\n- `trim_value: '\"'` — strips surrounding quotes from values\n- `prefix` — namespaces all extracted keys under a vendor prefix\n\n### Strategy 2: Grok with `SYSLOG5424SD` + KV with regex splits\n\nWhen 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 `=`.\n\nThis pattern is based on the `system/auth` integration in the upstream `elastic/integrations` repository.\n\n```yaml\n- grok:\n field: event.original\n patterns:\n - '^\u003c%{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}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n tag: parse_rfc5424\n\n- kv:\n if: ctx.syslog5424_sd != null && ctx.syslog5424_sd != ''\n field: syslog5424_sd\n field_split: '(?\u003c=\") '\n value_split: '(?i)(?\u003c=[a-z])=(?=\")'\n trim_key: \" \"\n trim_value: \" \"\n prefix: parsed.\n strip_brackets: true\n tag: kv_sd_element\n```\n\n### Strategy 3: Painless script for complex quoted-value KV\n\nWhen 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.\n\nThis pattern is used by integrations like fortinet_fortigate and fortinet_fortimanager in the upstream `elastic/integrations` repository.\n\n```yaml\n- script:\n lang: painless\n if: ctx._temp?.kv_data != null\n tag: script_parse_quoted_kv\n description: Split KV pairs handling quoted values with embedded spaces/delimiters.\n source: |\n def splitUnquoted(String input, String sep) {\n def tokens = [];\n def startPosition = 0;\n def isInQuotes = false;\n char quote = (char)\"\\\"\";\n for (def i = 0; i \u003c input.length(); i++) {\n if (input.charAt(i) == quote) {\n isInQuotes = !isInQuotes;\n } else if (input.charAt(i) == (char)sep && !isInQuotes) {\n def token = input.substring(startPosition, i).trim();\n if (!token.equals(\"\")) { tokens.add(token); }\n startPosition = i + 1;\n }\n }\n def last = input.substring(startPosition).trim();\n if (!last.equals(\"\")) { tokens.add(last); }\n return tokens;\n }\n def arr = splitUnquoted(ctx._temp.kv_data, \" \");\n Map map = new HashMap();\n Pattern pattern = /^\\\"|\\\"$/;\n for (def i = 0; i \u003c arr?.length; i++) {\n def kv = splitUnquoted(arr[i], \"=\");\n if (kv.length == 2 && kv[0].length() > 0) {\n map[kv[0]] = pattern.matcher(kv[1]).replaceAll(\"\");\n }\n }\n ctx.vendor = new HashMap();\n ctx.vendor.product = map;\n```\n\nPrefer strategy 1 or 2 when possible; use the script approach only when KV edge cases demand it.\n\n---\n\n## Web server / HTTP access log\n\n```\nSample (Nginx combined):\n192.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\"\n\nPattern (ECS field names):\n^%{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}\"$\n```\n\nThe built-in `%{COMMONAPACHELOG}` and `%{COMBINEDAPACHELOG}` patterns exist but use legacy (non-ECS) field names. Prefer the explicit ECS-mapped pattern above.\n\n---\n\n## Common mistakes\n\n| Mistake | Symptom | Fix |\n|---------|---------|-----|\n| Unanchored pattern | Partial match produces wrong field values | Prepend `^` — fail fast on non-matching lines |\n| `DATA` or `GREEDYDATA` without a bounding delimiter | Catastrophic backtracking; high CPU | Use `NOTSPACE` or `WORD`, or bound with lookahead: `%{DATA:f}(?=\\s)` |\n| Unescaped literal special characters | Pattern silently fails or matches the wrong segment | Escape `[`, `]`, `(`, `)`, `.`, `{`, `}` with `\\` |\n| Multi-pattern array ordered worst-first | Every line tries the slow/uncommon pattern first | Put the most common format at index 0 |\n| No fallback pattern | Non-matching lines error or fail silently | Add `%{GREEDYDATA:message}` as the last entry |\n| Wrong type suffix (`integer`, `Integer`) | Field stays as string | Use `int`/`long`/`double`/`float`/`boolean` only |\n| Capturing the same ECS field twice | Second capture silently overwrites the first | Use distinct names; merge fields after parsing |\n\n---\n\n## Debugging\n\n- **Kibana Grok Debugger**: Stack Management → Grok Debugger — interactive pattern testing against sample input\n- **Elasticsearch Simulate API**: `POST _ingest/pipeline/_simulate` with `\"trace_match\": true` — see which pattern index matched and inspect intermediate fields\n- **regex101.com**: select Oniguruma flavor for regex-level step-by-step explanation\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":11133,"content_sha256":"f6859da2b6fca3fdcc24f06a43060acaa416e590367bcab6ce0e06a4cddcbee0"},{"filename":"references/painless-patterns.md","content":"# Painless Script Patterns\n\nReference 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.\n\n## ctx read patterns\n\nAll 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.\n\n```yaml\n- script:\n tag: extract_user_info\n description: Extract user info with safe nested access\n lang: painless\n source: |\n if (ctx.user?.identity != null) {\n def name = ctx.user.identity.name;\n if (name != null && !name.isEmpty()) {\n ctx.user.full_name = name;\n }\n }\n```\n\nBoth `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.\n\nBracket notation is useful when field names contain dots or special characters:\n\n```yaml\n- script:\n tag: read_dotted_field\n description: Read a field name that contains a literal dot\n lang: painless\n source: |\n if (ctx.containsKey('host.name')) {\n ctx.host_name = ctx['host.name'];\n }\n```\n\n`containsKey` checks for field presence without risk of a null pointer:\n\n```painless\nif (ctx.containsKey('source') && ctx.source.containsKey('ip')) {\n // safe to use ctx.source.ip\n}\n```\n\n## ctx write patterns\n\nDirect assignment creates or overwrites a field. `remove` deletes it.\n\n```painless\nctx.event.outcome = 'success';\nctx.put('event.outcome', 'success'); // equivalent map-style put\nctx.remove('_temp'); // delete a field\nctx.list_field.add(item); // append to an existing list\n```\n\n## params usage\n\n`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.\n\nUse `params` for:\n- Lookup/mapping tables\n- Regex patterns\n- Threshold values and configuration constants\n\n```yaml\n- script:\n tag: map_severity\n description: Map severity label string to numeric value using params lookup\n lang: painless\n params:\n severity_map:\n low: 1\n medium: 2\n high: 3\n critical: 4\n source: |\n def label = ctx.event?.severity_label;\n if (label != null && params.severity_map.containsKey(label)) {\n ctx.event = ctx.event ?: [:];\n ctx.event.severity = params.severity_map.get(label);\n }\n```\n\nFor 1-2 constant values, inline comparisons in `source` are fine. For 3+ values or lookup tables, move them into `params` for readability and maintainability:\n\n```yaml\n# Avoid for many values -- harder to read and maintain\n- script:\n source: |\n if (ctx.raw_status == \"ACTIVE\") { ctx.status = \"active\"; }\n else if (ctx.raw_status == \"INACTIVE\") { ctx.status = \"inactive\"; }\n else if (ctx.raw_status == \"DISABLED\") { ctx.status = \"disabled\"; }\n\n# Preferred -- params keep source generic and readable\n- script:\n tag: map_status\n description: Normalize raw status using params lookup table\n lang: painless\n params:\n status_map:\n ACTIVE: active\n INACTIVE: inactive\n DISABLED: disabled\n source: |\n if (ctx.raw_status != null) {\n ctx.status = params.status_map.getOrDefault(ctx.raw_status, ctx.raw_status);\n }\n```\n\n## Map initialization for nested writes\n\nBefore 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.\n\n```yaml\n- script:\n tag: init_event_outcome\n description: Set event outcome with safe parent initialization\n lang: painless\n source: |\n ctx.event = ctx.event ?: [:];\n ctx.event.outcome = 'success';\n```\n\nFor deeply nested paths, each level must be initialized:\n\n```yaml\n- script:\n tag: set_deep_nested_field\n description: Set a deeply nested field with full parent chain init\n lang: painless\n source: |\n ctx.organization = ctx.organization ?: [:];\n ctx.organization.department = ctx.organization.department ?: [:];\n ctx.organization.department.name = ctx._temp_dept;\n```\n\nWhen building a new nested object from scratch, initialize the root as a `HashMap` and populate it:\n\n```yaml\n- script:\n tag: build_related_object\n description: Build related.ip from multiple source fields\n lang: painless\n source: |\n def ips = new HashSet();\n if (ctx.source?.ip != null) {\n ips.add(ctx.source.ip);\n }\n if (ctx.destination?.ip != null) {\n ips.add(ctx.destination.ip);\n }\n if (!ips.isEmpty()) {\n ctx.related = ctx.related ?: [:];\n ctx.related.ip = new ArrayList(ips);\n }\n```\n\n## Field API (ES 9.2+)\n\nThe 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.\n\n```painless\n// Field API -- set a deeply nested field without manual HashMap init\nfield('system.cpu.total.norm.pct').set($('cpu.usage', 0.0) / 100.0)\n```\n\nWithout the Field API, the same operation requires explicit initialization of every parent:\n\n```painless\nctx.system = ctx.system ?: [:];\nctx.system.cpu = ctx.system.cpu ?: [:];\nctx.system.cpu.total = ctx.system.cpu.total ?: [:];\nctx.system.cpu.total.norm = ctx.system.cpu.total.norm ?: [:];\nctx.system.cpu.total.norm.pct = ctx.cpu.usage / 100.0;\n```\n\nReference: https://www.elastic.co/docs/manage-data/ingest/transform-enrich/readable-maintainable-ingest-pipelines\n\n## foreach processor context\n\nWhen a `script` runs inside a `foreach` processor, the current array element is accessed via `ctx._ingest._value`, not through the array directly.\n\n```yaml\n- foreach:\n tag: normalize_event_items\n description: Lowercase the name field on each item in event.items\n field: event.items\n processor:\n script:\n tag: lowercase_item_name\n description: Lowercase the current item name\n lang: painless\n source: |\n if (ctx._ingest._value.name != null) {\n ctx._ingest._value.name = ctx._ingest._value.name.toLowerCase();\n }\n```\n\nKey rules:\n- `ctx._ingest._value` refers to the current element of the array specified in `foreach.field`.\n- `ctx._ingest._value.name` with `foreach.field: event.items` resolves to `event.items[*].name`.\n- Writing to `ctx._ingest._value` or its sub-fields modifies the element in place.\n- The script still has access to the full document via `ctx` for reading other fields.\n\n## Conditional field removal\n\nUse `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:\n\n```yaml\n- script:\n tag: remove_temp_fields\n description: Remove all temporary fields after processing\n lang: painless\n source: |\n def to_remove = ['_temp', '_header', '_raw_message'];\n for (def field : to_remove) {\n if (ctx.containsKey(field)) {\n ctx.remove(field);\n }\n }\n```\n\nFor nested field removal:\n\n```yaml\n- script:\n tag: clean_empty_nested\n description: Remove nested object if all its children are null\n lang: painless\n source: |\n if (ctx.source != null && ctx.source.ip == null && ctx.source.port == null) {\n ctx.remove('source');\n }\n```\n\n## Common script patterns\n\n### Array deduplication\n\n```yaml\n- script:\n tag: dedup_tags\n description: Remove duplicate entries from tags array\n lang: painless\n source: |\n if (ctx.tags != null && ctx.tags instanceof List) {\n ctx.tags = new ArrayList(new LinkedHashSet(ctx.tags));\n }\n```\n\n`LinkedHashSet` preserves insertion order while removing duplicates.\n\n### IP normalization\n\n```yaml\n- script:\n tag: normalize_ipv6\n description: Expand compressed IPv6 addresses to full form\n lang: painless\n params:\n ipv4_mapped_prefix: '::ffff:'\n source: |\n if (ctx.source != null && ctx.source.ip != null) {\n def ip = ctx.source.ip;\n if (ip.startsWith(params.ipv4_mapped_prefix)) {\n ctx.source.ip = ip.substring(params.ipv4_mapped_prefix.length());\n }\n }\n```\n\n### String manipulation\n\n```yaml\n- script:\n tag: extract_domain_from_email\n description: Extract domain part from user email address\n lang: painless\n source: |\n if (ctx.user != null && ctx.user.email != null) {\n def email = ctx.user.email;\n int idx = email.indexOf('@');\n if (idx > 0) {\n ctx.user.domain = email.substring(idx + 1);\n }\n }\n```\n\n### Timestamp arithmetic\n\n```yaml\n- script:\n tag: compute_duration\n description: Compute event duration from start and end timestamps\n lang: painless\n source: |\n if (ctx.event != null && ctx.event.start != null && ctx.event.end != null) {\n def start = ZonedDateTime.parse(ctx.event.start);\n def end = ZonedDateTime.parse(ctx.event.end);\n ctx.event.duration = ChronoUnit.NANOS.between(start, end);\n }\n```\n\n## Anti-patterns\n\n**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:\n\n| Script doing | Use instead |\n|---|---|\n| `ctx.field = ctx.other_field` | `rename` or `set` with `copy_from` |\n| `ctx.field = value` (constant) | `set` |\n| `ctx.list.add(value)` | `append` |\n| Type conversion | `convert` |\n| String splitting or extraction | `dissect` or `grok` |\n| Regex replacement | `gsub` |\n| Regex matching / extraction | `grok` with `pattern_definitions` |\n| Case normalization | `lowercase` / `uppercase` |\n\n**Concrete example — extracting a domain from an email address:**\n\n```yaml\n# WRONG — script for a job dissect handles natively\n- script:\n tag: script_set_user_domain\n lang: painless\n description: Extract domain from email address.\n if: ctx.vendor?.owner instanceof String && ctx.vendor.owner.contains('@')\n source: |-\n String u = ctx.vendor.owner;\n int at = u.lastIndexOf('@');\n if (at > 0 && at \u003c u.length() - 1) {\n ctx.user = ctx.user ?: [:];\n ctx.user.domain = u.substring(at + 1);\n }\n\n# CORRECT — dissect is faster, shorter, and easier to review\n- dissect:\n tag: dissect_user_domain\n field: vendor.owner\n pattern: \"%{?_ignore}@%{user.domain}\"\n if: ctx.vendor?.owner != null && ctx.vendor.owner.contains('@')\n```\n\n**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.\n\n**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.\n\n## Review checklist\n\n- [ ] **Script could be replaced with a built-in processor** -- **MEDIUM** (see anti-patterns table above)\n- [ ] Script has `tag` and `description` -- **MEDIUM**\n- [ ] `params` used for constants instead of hardcoding in script body -- **LOW**\n- [ ] Null checks present before field access -- **MEDIUM**\n- [ ] No `ctx?` usage (ctx is always non-null) -- **LOW**\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":12131,"content_sha256":"7d4c45bef2b99384faeee5c63d6479f4ad34ae209ed4e838246441a75d824211"},{"filename":"references/processor-cookbook.md","content":"# ingest processor cookbook\n\nUse this cookbook to choose processors quickly while designing integration ingest pipelines.\n\n## Parsing processors\n\n| Processor | Best for | Key parameters | Notes |\n| --- | --- | --- | --- |\n| `grok` | Variable log formats | `field`, `patterns`, `pattern_definitions`, `ignore_missing`, `tag` | Use when token boundaries vary. Anchor patterns when possible. |\n| `dissect` | Stable delimited text | `field`, `pattern`, `ignore_missing`, `tag` | Usually faster than grok for fixed formats. |\n| `json` | JSON payload string parsing | `field`, `target_field`, `add_to_root`, `on_failure`, `tag` | Great for logs that embed raw JSON. |\n| `csv` | Delimited values | `field`, `target_fields`, `separator`, `quote`, `ignore_missing` | Useful for fixed CSV-like telemetry records. |\n| `kv` | `k=v` style logs | `field`, `field_split`, `value_split`, `target_field`, `trim_key`, `trim_value` | Common in firewall and audit-style logs. |\n\n### Example: `grok` + `dissect` fallback\n\n```yaml\n- dissect:\n field: event.original\n pattern: \"%{source.address} - %{user.name} [%{nginx.access.time}] \\\"%{http.request.method} %{url.original} HTTP/%{http.version}\\\" %{http.response.status_code} %{http.response.body.bytes}\"\n ignore_failure: true\n tag: dissect_access\n- grok:\n field: event.original\n patterns:\n - '^%{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}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n if: ctx.http?.request?.method == null\n tag: grok_access_fallback\n```\n\n### Example: `json` parser with local failure handling\n\n```yaml\n- json:\n field: event.original\n target_field: json\n tag: parse_json\n on_failure:\n - append:\n field: error.message\n value: '{{{_ingest.on_failure_message}}}'\n```\n\n## Normalization processors\n\n| Processor | Best for | Key parameters | Notes |\n| --- | --- | --- | --- |\n| `rename` | Move parsed fields | `field`, `target_field`, `ignore_missing`, `ignore_failure` | Common for `message -> event.original`. |\n| `set` | Add constants/derived values | `field`, `value`, `if`, `copy_from` | Use for ECS categorization and defaults. |\n| `remove` | Drop temporary/source fields | `field`, `ignore_missing`, `ignore_failure`, `if` | Keep documents clean after parsing. |\n| `convert` | Type coercion | `field`, `type`, `ignore_missing`, `ignore_failure` | Convert before comparisons or aggregation. |\n| `date` | Parse timestamps | `field`, `target_field`, `formats`, `timezone`, `on_failure` | Often has multiple format candidates. |\n| `lowercase` / `uppercase` / `trim` | String normalization | `field`, `ignore_missing`, `if` | Use after parse, before categorization. |\n| `split` | Turn strings into arrays | `field`, `separator`, `ignore_missing` | Useful for multi-IP and list fields. |\n| `gsub` | Regex replacement | `field`, `pattern`, `replacement` | Use sparingly for cleanup/transforms. |\n\n### Example: normalize timestamp + status\n\n```yaml\n- date:\n field: nginx.access.time\n target_field: '@timestamp'\n formats:\n - dd/MMM/yyyy:H:m:s Z\n tag: parse_timestamp\n- convert:\n field: http.response.status_code\n type: long\n ignore_missing: true\n tag: convert_status_code\n```\n\n## Enrichment processors\n\n| Processor | Best for | Key parameters | Notes |\n| --- | --- | --- | --- |\n| `geoip` | IP geolocation + ASN | `field`, `target_field`, `database_file`, `properties`, `ignore_missing`, `if` | Use on validated IP fields. |\n| `user_agent` | User agent parsing | `field`, `ignore_missing`, `if` | Adds browser/device metadata. |\n| `registered_domain` | Domain decomposition | `field`, `target_field`, `ignore_missing` | Splits FQDN into registered/subdomain pieces. |\n| `community_id` | Network flow hashing | `source_ip`, `source_port`, `destination_ip`, `destination_port`, `iana_number`, `target_field` | Useful for flow correlation. |\n| `uri_parts` | URL decomposition | `field`, `target_field`, `keep_original`, `ignore_failure` | Parse URL into scheme/host/path/query. |\n| `append` | Add related values/tags | `field`, `value`, `allow_duplicates`, `if` | Common for `related.ip` and `tags`. |\n\n### Example: common web enrichment path\n\n```yaml\n- user_agent:\n field: user_agent.original\n if: ctx.user_agent?.original != null\n ignore_missing: true\n tag: parse_user_agent\n- geoip:\n field: source.ip\n target_field: source.geo\n if: ctx.source?.ip != null\n ignore_missing: true\n tag: enrich_source_geo\n- append:\n field: related.ip\n value: '{{{source.ip}}}'\n if: ctx.source?.ip != null\n```\n\n### IP geolocation and ASN enrichment — full pattern\n\nWhen 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.).\n\n```yaml\n# IP Geolocation Lookup\n- geoip:\n field: source.ip\n target_field: source.geo\n if: ctx.source?.ip != null\n ignore_missing: true\n tag: enrich_source_geo\n- geoip:\n field: destination.ip\n target_field: destination.geo\n if: ctx.destination?.ip != null\n ignore_missing: true\n tag: enrich_destination_geo\n\n# IP Autonomous System (AS) Lookup\n- geoip:\n database_file: GeoLite2-ASN.mmdb\n field: source.ip\n target_field: source.as\n if: ctx.source?.ip != null\n properties:\n - asn\n - organization_name\n ignore_missing: true\n tag: enrich_source_asn\n- geoip:\n database_file: GeoLite2-ASN.mmdb\n field: destination.ip\n target_field: destination.as\n if: ctx.destination?.ip != null\n properties:\n - asn\n - organization_name\n ignore_missing: true\n tag: enrich_destination_asn\n\n# Rename ASN fields to ECS names\n- rename:\n field: source.as.asn\n target_field: source.as.number\n ignore_missing: true\n tag: rename_source_asn\n- rename:\n field: source.as.organization_name\n target_field: source.as.organization.name\n ignore_missing: true\n tag: rename_source_as_org\n- rename:\n field: destination.as.asn\n target_field: destination.as.number\n ignore_missing: true\n tag: rename_destination_asn\n- rename:\n field: destination.as.organization_name\n target_field: destination.as.organization.name\n ignore_missing: true\n tag: rename_destination_as_org\n```\n\nKey rules:\n- 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`.\n- Always include both geo and ASN lookups when enriching IP fields. Omitting ASN leaves `source.as` / `destination.as` empty.\n- Use `ignore_missing: true` on all geo/ASN processors — the IP field may not be present on every event.\n- 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.\n- Place these processors **after** all IP fields have been set (after parsing, renaming, and converting IP addresses) and **before** the `related.ip` append processors.\n\n## Flow control processors\n\n| Processor | Best for | Key parameters | Notes |\n| --- | --- | --- | --- |\n| `drop` | Early discard of unwanted docs | `if` | Put early for performance. |\n| `fail` | Stop processing with explicit error | `message`, `if`, `tag` | Use for invalid required input shape. |\n| `pipeline` | Sub-pipeline routing | `name`, `if`, `ignore_missing_pipeline`, `tag` | Core branching primitive. |\n| `foreach` | Iterate over array items | `field`, `processor`, `if`, `ignore_failure` | Useful for nested OCSF arrays. |\n| `script` | Custom logic in Painless | `source`, `lang`, `if`, `params`, `tag` | Use only when processors are insufficient. |\n\n### Example: route to sub-pipelines\n\n```yaml\n- pipeline:\n name: '{{ IngestPipeline \"pipeline_object_user\" }}'\n if: ctx.ocsf?.class_uid != null && ['2005','3001','3002'].contains(ctx.ocsf.class_uid) && ctx.ocsf.user != null\n ignore_missing_pipeline: true\n tag: route_object_user\n- pipeline:\n name: '{{ IngestPipeline \"pipeline_category_network_activity\" }}'\n if: ctx.ocsf?.category_uid == '4'\n ignore_missing_pipeline: true\n tag: route_category_network\n```\n\n## Utility processors\n\n| Processor | Best for | Key parameters | Notes |\n| --- | --- | --- | --- |\n| `dot_expander` | Expand dotted field names into objects | `field`, `path`, `ignore_failure` | Useful when source keys contain dots. |\n| `fingerprint` | Stable IDs or dedupe keys | `fields`, `target_field`, `method`, `ignore_missing` | Useful for TSDS and correlation identifiers. |\n| `bytes` | Human-readable size to numeric | `field`, `target_field`, `ignore_missing` | Converts values like `1kb` to bytes. |\n| `html_strip` | Remove HTML tags | `field`, `target_field`, `ignore_missing` | For message text sanitation. |\n| `sort` | Order array values deterministically | `field`, `order`, `target_field` | Useful for stable outputs and tests. |\n\n## ECS categorization mapping patterns\n\nMap 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.\n\n### Pattern A: Script with `params` lookup table (recommended for 5+ mappings)\n\nPut 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.\n\n**Single-key lookup** (mapping from one source field):\n\n```yaml\n - script:\n lang: painless\n tag: set_ecs_categorization\n description: Map event type to ECS categorization fields.\n if: ctx.json?.event_type != null\n params:\n process_start:\n category: [process]\n type: [start]\n process_end:\n category: [process]\n type: [end]\n network_connection:\n category: [network]\n type: [connection]\n file_creation:\n category: [file]\n type: [creation]\n file_modification:\n category: [file]\n type: [change]\n user_login:\n category: [authentication]\n type: [start]\n outcome: success\n user_login_failed:\n category: [authentication]\n type: [start]\n outcome: failure\n source: |-\n def mapping = params[ctx.json.event_type];\n if (mapping == null) {\n return;\n }\n ctx.event.category = mapping.category;\n ctx.event.type = mapping.type;\n if (mapping.containsKey('outcome')) {\n ctx.event.outcome = mapping.outcome;\n }\n```\n\n**Composite-key lookup** (mapping from two or more source fields combined):\n\n```yaml\n - script:\n lang: painless\n tag: set_ecs_categorization\n description: Map target type and action to ECS categorization fields.\n if: ctx.vendor?.target_type != null && ctx.event?.action != null\n params:\n \"device:create\":\n category: [host]\n type: [creation]\n outcome: success\n \"device:delete\":\n category: [host]\n type: [deletion]\n outcome: success\n \"device:update\":\n category: [host]\n type: [change]\n outcome: success\n \"api_token:create\":\n category: [iam, configuration]\n type: [creation]\n outcome: success\n \"api_token:delete\":\n category: [iam, configuration]\n type: [deletion]\n outcome: success\n \"blueprint:update\":\n category: [configuration]\n type: [change]\n outcome: success\n source: |-\n String key = ctx.vendor.target_type + ':' + ctx.event.action;\n def mapping = params[key];\n if (mapping == null) {\n return;\n }\n ctx.event.category = mapping.category;\n ctx.event.type = mapping.type;\n if (mapping.containsKey('outcome')) {\n ctx.event.outcome = mapping.outcome;\n }\n```\n\n**Merge variant** (when categorization fields may already have values from earlier processors and you need to add to them rather than overwrite):\n\n```yaml\n source: |-\n def addUnique(List dst, List src) {\n HashSet s = new HashSet(dst != null ? dst : []);\n s.addAll(src != null ? src : []);\n return new ArrayList(s);\n }\n def mapping = params[ctx.json.event_type];\n if (mapping == null) {\n return;\n }\n ctx.event.type = addUnique(ctx.event.type, mapping.type);\n ctx.event.category = addUnique(ctx.event.category, mapping.category);\n```\n\n**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.\n\nReference integrations using this pattern:\n- `carbon_black_cloud` (endpoint_event) — single-key lookup by `json.type`\n- `okta` (system) — merge variant with large params table in dedicated sub-pipeline\n- `thycotic_ss` (logs) — single-key lookup by `cef.name`\n- `zeronetworks` (audit) — composite fields with `action`, `outcome`, `type`, `category` per entry\n\n### Pattern B: Set processors with conditionals (fewer than 5 mappings)\n\nFor simple cases with only 2–4 distinct mappings, `set` processors with `if` conditions are clearer than a script:\n\n```yaml\n - set:\n field: event.category\n tag: set_category_auth\n value: [authentication]\n if: ctx.json?.event_type == 'login' || ctx.json?.event_type == 'logout'\n - set:\n field: event.type\n tag: set_type_start\n value: [start]\n if: ctx.json?.event_type == 'login'\n - set:\n field: event.type\n tag: set_type_end\n value: [end]\n if: ctx.json?.event_type == 'logout'\n - set:\n field: event.outcome\n tag: set_outcome_success\n value: success\n if: ctx.json?.result == 'success'\n - set:\n field: event.outcome\n tag: set_outcome_failure\n value: failure\n if: ctx.json?.result == 'failure'\n```\n\nUse 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.\n\n### Pattern C: Sub-pipeline for large mapping tables (100+ mappings)\n\nFor very large mapping tables (like Okta with 500+ event types), extract the categorization into a dedicated sub-pipeline file to keep `default.yml` readable:\n\n```yaml\n# In default.yml — route to categorization sub-pipeline:\n - pipeline:\n name: '{{ IngestPipeline \"ecs-categorization\" }}'\n tag: route_ecs_categorization\n```\n\n```yaml\n# In ecs-categorization.yml — single script processor with large params table:\n---\ndescription: Map event types to ECS categorization fields.\nprocessors:\n - script:\n lang: painless\n tag: set_ecs_categorization\n description: Map event type to ECS categorization fields.\n if: ctx.vendor?.event_type != null\n params:\n # ... hundreds of entries ...\n source: |-\n def mapping = params[ctx.vendor.event_type];\n if (mapping == null) {\n return;\n }\n ctx.event.category = mapping.category;\n ctx.event.type = mapping.type;\non_failure:\n - append:\n field: error.message\n value: >-\n Processor '{{{ _ingest.on_failure_processor_type }}}'\n {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'\n {{{/_ingest.on_failure_processor_tag}}}in pipeline '{{{ _ingest.pipeline }}}'\n failed with message '{{{ _ingest.on_failure_message }}}'\n```\n\nReference: `okta` (system) uses `ecs_category_type.yml` as a dedicated sub-pipeline.\n\n### Anti-patterns — do NOT use these\n\n**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.\n\n**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`.\n\n## Selection tips\n\n- **Do not use `script` when a built-in processor can do the job** — see `SKILL.md` → *Painless script best practices* for the mandatory checklist.\n- Add `tag` on **every** processor (not only processors that can fail).\n- For optional data paths, prefer guarded `if` over `ignore_missing`.\n- Keep parsing, normalization, and enrichment stages visually grouped.\n\n## Processor performance guide\n\n### Processor cost ordering\n\nProcessors vary dramatically in cost. Order pipelines so cheap operations run first and expensive ones run only when needed.\n\n| Tier | Processors | Notes |\n| --- | --- | --- |\n| Fastest | `set`, `rename`, `remove` | Simple field manipulation, near-zero overhead. |\n| Fast | `convert`, `lowercase`, `uppercase`, `trim` | String/type operations, still very cheap. |\n| Moderate | `date`, `grok`, `dissect` | Parsing involves format matching; grok is regex-heavy. |\n| Slow | `script` | Painless compilation + execution; avoid when a built-in suffices. |\n| Expensive | `geoip`, `user_agent` | Database/cache lookups on every invocation. |\n\n### Guard `geoip` with an `if` condition in new pipelines\n\nEven 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. In new pipelines, add an `if` condition to check field existence before the processor runs. For `user_agent` the guard is optional — the lookup is cheaper and bare `ignore_missing: true` is acceptable. In existing pipelines a missing guard is not worth churn on its own (see `review-integration/references/conflict-resolutions.md`):\n\n```yaml\n- geoip:\n tag: geoip_source_ip\n if: ctx.source?.ip != null\n field: source.ip\n target_field: source.geo\n ignore_missing: true\n```\n\nThe `if` guard prevents the expensive lookup from executing at all. Without it, every document pays the lookup cost regardless of whether the field exists.\n\n### Skip enrichment when results already exist\n\nBeyond 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.\n\n```yaml\n- geoip:\n tag: geoip_source_ip\n if: ctx.source?.ip != null && ctx.source.geo == null\n field: source.ip\n target_field: source.geo\n```\n\n### Batch remove operations\n\nUse a single `remove` with a field list instead of multiple separate processors. Each processor invocation has fixed overhead; batching eliminates it.\n\n```yaml\n- remove:\n field: [_tmp, json, message]\n ignore_missing: true\n tag: cleanup_temp_fields\n```\n\n### Use `equalsIgnoreCase` instead of `toLowerCase`\n\nIn Painless `if` conditions, `equalsIgnoreCase` avoids allocating a throwaway lowercase string on every document.\n\n```java\n// Preferred -- no allocation\nif (ctx.event?.action?.equalsIgnoreCase('login') == true)\n\n// Avoid -- allocates a new string per document\nif (ctx.event?.action?.toLowerCase() == 'login')\n```\n\n### Prefer `rename` over `set` + `remove`\n\n`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.\n\n```yaml\n# Preferred -- single operation\n- rename:\n tag: rename_json_user\n field: json.user\n target_field: user.name\n ignore_missing: true\n\n# Avoid -- two operations for the same result\n- set:\n tag: set_user_name\n field: user.name\n copy_from: json.user\n- remove:\n tag: remove_json_user\n field: json.user\n ignore_missing: true\n```\n\n## Foreach semantics and `_ingest._value`\n\n### How foreach works\n\nThe `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.\n\n```yaml\n- foreach:\n tag: foreach_event_items\n field: event.items\n processor:\n append:\n tag: append_related_user\n field: related.user\n value: '{{{_ingest._value.name}}}'\n```\n\n### Resolving `_ingest._value` references\n\nTo understand the actual data flow, resolve `_ingest._value` back to the `foreach` field:\n\n| foreach field | Processor reference | Resolved field |\n| --- | --- | --- |\n| `event.items` | `_ingest._value.name` | `event.items[*].name` |\n| `event.items` | `_ingest._value.id` | `event.items[*].id` |\n| `json.tags` | `_ingest._value` | `json.tags[*]` (scalar element) |\n\n### Subfield access within foreach\n\nWhen iterating over an array of objects, access subfields with dot notation after `_ingest._value`:\n\n```yaml\n- foreach:\n tag: foreach_json_network_connections\n field: json.network_connections\n processor:\n convert:\n tag: convert_port\n field: _ingest._value.port\n type: long\n ignore_missing: true\n```\n\nThe actual field being converted is `json.network_connections[*].port`.\n\n### Painless scripts inside foreach\n\nIn a `script` processor nested inside `foreach`, access the current element via `ctx._ingest._value` (not `ctx.field[i]`):\n\n```painless\ndef val = ctx._ingest._value;\nif (val.containsKey('ip')) {\n // val.ip refers to foreach_field[*].ip\n}\n```\n\nModifications to `_ingest._value` subfields mutate the original array element in place.\n\n### Common mistakes\n\n- Treating `_ingest._value` as a real document field path outside a foreach context.\n- Forgetting that writes to `_ingest._value` subfields modify the source array element in place.\n- Using `ctx._ingest._value` outside a `foreach` block (it does not exist there).\n\n## Condition patterns\n\n### If-clause fields as implicit inputs\n\nFields 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.\n\n```yaml\n- set:\n tag: set_event_category\n field: event.category\n value: [authentication]\n if: ctx.json?.event_type == 'login'\n```\n\nHere `json.event_type` is an input -- it determines whether `event.category` gets written.\n\n### Painless null-safe condition patterns\n\n| Pattern | Meaning |\n| --- | --- |\n| `ctx.field != null` | Field exists and is non-null |\n| `ctx?.field == 'value'` | Null-safe access, compare value |\n| `ctx.containsKey('field')` | Field key exists in the document map (even if null) |\n| `ctx.field instanceof List` | Field is an array |\n| `ctx.field?.size() != 0` | Non-empty collection (use `!=`, not `>`) |\n\nDo not use inequality operators (`\u003c`, `>`, `\u003c=`, `>=`) 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.\n\nFor nested fields, chain null-safe access: `ctx.source?.geo?.country_name != null`.\n\n### Use `.contains()` instead of chained OR conditions\n\nWhen 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:\n\n```yaml\n- script:\n tag: check_user_role\n params:\n privileged_names:\n - \"admin\"\n - \"system\"\n - \"root\"\n - \"service\"\n source: |\n if (params.privileged_names.contains(ctx.user?.name)) {\n ctx.user.privileged = true;\n }\n```\n\nFor 1-2 values, inline comparison is fine -- the allocation overhead is negligible:\n\n```painless\nif (ctx.event?.action == 'login' || ctx.event?.action == 'logout') { ... }\n```\n\nAvoid long chained OR conditions regardless of approach:\n\n```painless\n// Avoid -- verbose and error-prone\nctx.user?.name == 'admin' || ctx.user?.name == 'system' || ctx.user?.name == 'root' || ctx.user?.name == 'service'\n```\n\n### Mustache field references in conditions\n\nIn `set` and `append` processor `value` fields, Mustache `{{{field}}}` references are inputs that provide the value being written. Use triple braces to disable HTML escaping.\n\n```yaml\n- set:\n tag: set_host_id\n field: host.id\n value: '{{{crowdstrike.aid}}}'\n```\n\n`crowdstrike.aid` is the input, `host.id` is the output.\n\n### Nested condition and value references\n\nWhen a processor has both an `if` condition and a Mustache value reference, both are inputs:\n\n```yaml\n- append:\n tag: append_related_ip\n field: related.ip\n value: '{{{source.ip}}}'\n if: ctx.source?.ip != null\n```\n\nInputs: `source.ip` (from both the condition and the value reference). Output: `related.ip`.\n\n## Field transform pitfalls\n\n### `set` with `copy_from` vs Mustache value\n\nPrefer `copy_from` over a Mustache `value` when copying fields -- it avoids string coercion and preserves the original type (objects, arrays, numbers):\n\n```yaml\n- set:\n tag: set_event_original\n field: event.original\n copy_from: message\n if: ctx.tags?.contains('preserve_original_event') == true\n```\n\nGuard `copy_from` usage to avoid overwriting an existing value. The `event.original` preservation pattern should check whether the target is already set:\n\n```yaml\n- rename:\n tag: rename_message\n field: message\n target_field: event.original\n ignore_missing: true\n if: ctx.event?.original == null\n- remove:\n tag: remove_message\n field: message\n ignore_missing: true\n if: ctx.event?.original != null\n```\n\n### `convert` with `type: ip` must have a downstream consumer\n\nA `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.\n\n```yaml\n# Correct -- conversion feeds GeoIP enrichment downstream\n- convert:\n tag: convert_source_ip\n field: source.ip\n type: ip\n ignore_missing: true\n on_failure:\n - append:\n field: error.message\n value: >-\n Processor {{{_ingest.on_failure_processor_type}}}\n with tag '{{{_ingest.on_failure_processor_tag}}}'\n failed: {{{_ingest.on_failure_message}}}\n- geoip:\n tag: geoip_source_ip\n field: source.ip\n target_field: source.geo\n if: ctx.source?.ip != null\n```\n\n### Empty string guard before `convert`\n\nAn empty string fails `convert` with a confusing error. Always guard:\n\n```yaml\n- convert:\n field: json.severity\n type: long\n ignore_missing: true\n if: ctx.json?.severity != ''\n tag: convert_severity\n```\n\n### `append` for `related.*` fields\n\nAlways set `allow_duplicates: false` when appending to `related.*` arrays to avoid bloated documents:\n\n```yaml\n- append:\n tag: append_related_ip\n field: related.ip\n value: '{{{source.ip}}}'\n allow_duplicates: false\n if: ctx.source?.ip != null\n```\n\n## Non-grok parsing processors\n\n### `dissect` vs `grok` decision criteria\n\nUse `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.\n\n```yaml\n# Dissect -- fixed delimiters, faster\n- dissect:\n tag: dissect_message\n field: message\n pattern: \"%{ts} %{+ts} %{log_level} [%{thread}] %{class} - %{msg}\"\n\n# Grok -- variable format, regex needed\n- grok:\n tag: grok_message\n field: message\n patterns:\n - '^%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} \\[%{DATA:component}\\] %{GREEDYDATA:msg}'\n```\n\nDissect supports modifiers: `%{+field}` (append), `%{+field/order}` (ordered append), `%{?skip}` (discard), `%{*key}`/`%{&value}` (dynamic key-value).\n\n### `json` processor: `add_to_root` and conflict strategy\n\nWhen 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`:\n\n```yaml\n- json:\n field: message\n add_to_root: true\n add_to_root_conflict_strategy: replace\n tag: json_merge_to_root\n```\n\nPrefer `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.\n\n### `date` timezone patterns\n\nAvoid short timezone abbreviations (PST, EST) -- they are ambiguous across JDKs. Use full IANA names or UTC offsets.\n\n```yaml\n- date:\n tag: date_json_local_time\n field: json.local_time\n target_field: '@timestamp'\n formats:\n - \"dd/MM/yyyy HH:mm:ss\"\n timezone: \"Europe/Amsterdam\"\n if: ctx.json?.local_time != null\n```\n\nDynamic timezone from a document field:\n\n```yaml\n- date:\n tag: date_json_timestamp\n field: json.timestamp\n formats: [ISO8601]\n timezone: '{{{json.tz}}}'\n```\n\nMultiple format arrays let the processor try each format in order:\n\n```yaml\n- date:\n tag: date_json_timestamp\n field: json.timestamp\n target_field: '@timestamp'\n formats:\n - ISO8601\n - \"yyyy-MM-dd HH:mm:ss\"\n - UNIX\n if: ctx.json?.timestamp != null\n```\n\n### `date` with `on_failure`\n\nDate parsing failures should capture the error and remove the unparseable field to prevent downstream confusion:\n\n```yaml\n- date:\n field: json.eventTime\n tag: date_parse_eventTime\n formats: [ISO8601]\n on_failure:\n - remove:\n field: json.eventTime\n - append:\n field: error.message\n value: >-\n Processor {{{_ingest.on_failure_processor_type}}}\n with tag '{{{_ingest.on_failure_processor_tag}}}'\n failed: {{{_ingest.on_failure_message}}}\n```\n\n### `kv` processor patterns\n\nUse `field_split` and `value_split` to define the delimiters. Use `target_field` to namespace the output and `prefix` to avoid field name collisions:\n\n```yaml\n- kv:\n tag: kv_json_data\n if: ctx.json?.data != null && ctx.json.data != ''\n field: json.data\n target_field: parsed\n field_split: '&'\n value_split: '='\n trim_value: '\"'\n\n- kv:\n tag: kv_message\n field: message\n field_split: ' '\n value_split: '='\n prefix: 'vendor.product.'\n target_field: _temp.kv\n```\n\n### `uri_parts` usage\n\nDecomposes a URL string into scheme, host, port, path, query, and fragment:\n\n```yaml\n- uri_parts:\n tag: uri_parts_url_original\n field: url.original\n target_field: url\n keep_original: true\n if: ctx.url?.original != null\n```\n\n### `dot_expander` usage\n\nExpands 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):\n\n```yaml\n- dot_expander:\n tag: dot_expander_json\n field: '*'\n path: json\n if: ctx.json != null\n```\n\nUse `path` to scope expansion to a specific sub-tree and avoid unintended side effects on the root document.\n\n## Enrichment depth\n\n### `community_id` implicit inputs\n\nThe `community_id` processor reads several fields by convention without an explicit `field` parameter. All must be populated before the processor runs:\n\n- `source.ip`, `source.port`\n- `destination.ip`, `destination.port`\n- `network.transport` (protocol name) or `network.iana_number` (protocol number)\n\n```yaml\n- community_id:\n ignore_missing: true\n if: ctx.source?.ip != null && ctx.destination?.ip != null\n tag: add_community_id\n```\n\nMissing transport/iana_number fields cause the processor to fall back to a default protocol, which may produce incorrect hashes.\n\n### `registered_domain` targeting and output structure\n\nThe 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:\n\n```yaml\n- registered_domain:\n tag: registered_domain_dns_question_name\n field: dns.question.name\n target_field: dns.question\n if: ctx.dns?.question?.name != null\n```\n\nOutput fields written under `target_field`:\n- `registered_domain` -- the registered domain (e.g., `example.com`)\n- `top_level_domain` -- the TLD (e.g., `com`)\n- `subdomain` -- the subdomain portion (e.g., `www`)\n\n### Enrich policy lookup\n\nThe `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:\n\n```yaml\n- enrich:\n tag: enrich_host_ip\n policy_name: hosts-policy\n field: host.ip\n target_field: _temp.enrich\n if: ctx.host?.ip != null\n```\n\nAfter the enrich processor, selectively copy needed fields from `_temp.enrich` into their final locations and remove the temporary object.\n\n## Control flow semantics\n\n### `terminate` processor for failure paths\n\nUse `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.\n\n```yaml\n- set:\n tag: set_event_kind\n field: event.kind\n value: pipeline_error\n- terminate:\n tag: terminate_pipeline_error\n if: ctx.event?.kind == 'pipeline_error'\n```\n\nPlace `terminate` immediately after setting the failure marker in `on_failure` blocks.\n\n### `drop` conditional patterns\n\n`drop` silently discards the document. Always guard with an `if` condition -- an unconditional `drop` deletes everything.\n\n```yaml\n- drop:\n if: ctx.event?.action == 'heartbeat'\n tag: drop_heartbeat\n description: Discard periodic heartbeat events\n```\n\nCommon uses: filtering out health-check events, deduplication markers, or noise events that provide no analytical value.\n\n### `fail` for input validation\n\nUse `fail` to enforce preconditions early in the pipeline. This produces a clear error message rather than letting the document fail cryptically downstream.\n\n```yaml\n- fail:\n if: ctx.event?.kind == null\n message: 'event.kind is required but missing'\n tag: require_event_kind\n```\n\n### Foreach and pipeline chaining edge cases\n\nWhen a `foreach` contains a `pipeline` processor call, each array element is processed by the full sub-pipeline. Be aware that:\n\n- The sub-pipeline sees the entire document context, not just the array element.\n- `_ingest._value` is accessible in the sub-pipeline's processors.\n- Errors in the sub-pipeline for one element do not automatically skip remaining elements unless `ignore_failure: true` is set on the `foreach`.\n- Deeply nested foreach-pipeline chains are hard to debug; prefer flattening where possible.\n\n## Special processors\n\n### `fingerprint`\n\nGenerates a deterministic hash from one or more fields. Commonly used to set `_id` for deduplication.\n\n```yaml\n- fingerprint:\n tag: fingerprint_event_id\n fields:\n - event.id\n - '@timestamp'\n target_field: _id\n method: SHA-256\n```\n\nKey considerations:\n- `fields` is an array -- field order matters (changing order changes the hash).\n- If any field in `fields` is missing, the hash changes compared to when it is present. Guard with `if` or ensure fields always exist.\n- `target_field: _id` enables upsert-style deduplication in Elasticsearch.\n- Prefix `_id` with a timestamp for better index write performance when using time-series data.\n\n### `html_strip`\n\nRemoves HTML tags from a field value.\n\n```yaml\n- html_strip:\n tag: html_strip_message\n field: message\n target_field: message_clean\n if: ctx.message != null\n```\n\nWhen `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.\n\n### `url_decode`\n\nDecodes percent-encoded URL strings (e.g., `%20` becomes a space).\n\n```yaml\n- url_decode:\n tag: url_decode_url_query\n field: url.query\n if: ctx.url?.query != null\n```\n\nLike `html_strip`, writes back to `field` when no `target_field` is set.\n\n### `network_direction`\n\nDetermines whether network traffic is `inbound`, `outbound`, or `internal` based on source/destination IPs and configured internal networks.\n\nImplicit input fields (must be populated before the processor runs):\n- `source.ip`\n- `destination.ip`\n- `network.type` (optional -- IPv4 vs IPv6 hint)\n\n```yaml\n- network_direction:\n tag: network_direction\n internal_networks:\n - loopback\n - private\n if: ctx.source?.ip != null && ctx.destination?.ip != null\n```\n\nOutput defaults to `network.direction`. Use `internal_networks_field` to read the network list from a document field instead of hardcoding it.\n\n### `set_security_user`\n\nCopies 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.\n\n```yaml\n- set_security_user:\n tag: set_security_user\n field: user\n properties:\n - username\n - roles\n - full_name\n```\n\nTypically 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.\n","content_type":"text/markdown; charset=utf-8","language":"markdown","size":37867,"content_sha256":"069dc856461d5cd6b9f38669936e3edfc54d7b351d64e33ded530ba6ffd0510f"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"ingest-pipelines","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Skill authority","type":"text"}]},{"type":"paragraph","content":[{"text":"The rules and patterns defined in this skill and its reference files are the ","type":"text"},{"text":"authoritative source of truth","type":"text","marks":[{"type":"strong"}]},{"text":". When examining existing integrations in the ","type":"text"},{"text":"elastic/integrations","type":"text","marks":[{"type":"code_inline"}]},{"text":" repository for reference, you may encounter patterns that conflict with what is specified here — many integrations contain legacy patterns that predate current standards. ","type":"text"},{"text":"Always follow this skill over patterns observed in other integrations.","type":"text","marks":[{"type":"strong"}]},{"text":" If a reference integration uses a deprecated or prohibited pattern, do not copy it.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"When to use","type":"text"}]},{"type":"paragraph","content":[{"text":"Use this skill when tasks include:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"building or modifying ","type":"text"},{"text":"elasticsearch/ingest_pipeline/default.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" for a data stream","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"choosing parser and normalization processors (","type":"text"},{"text":"grok","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"dissect","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"json","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"kv","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"date","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"convert","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"designing conditional branches and sub-pipeline routing with ","type":"text"},{"text":"pipeline","type":"text","marks":[{"type":"code_inline"}]},{"text":" processors","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"implementing resilient error handling with top-level ","type":"text"},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"tuning processor order for ingest performance and maintainability","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"When not to use","type":"text"}]},{"type":"paragraph","content":[{"text":"Do not use this skill as the primary guide for:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"ECS field selection, categorization values, and field mapping strategy (","type":"text"},{"text":"ecs-field-mappings","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"elastic-package command and stack lifecycle workflows (","type":"text"},{"text":"elastic-package-cli","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"test fixture authoring and expected output workflows (","type":"text"},{"text":"integration-testing","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"references/pipeline-testing.md","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Pipeline anatomy","type":"text"}]},{"type":"paragraph","content":[{"text":"In integration packages, ingest pipelines live under:","type":"text"}]},{"type":"paragraph","content":[{"text":"data_stream/\u003cstream>/elasticsearch/ingest_pipeline/","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"paragraph","content":[{"text":"Every stream usually has a ","type":"text"},{"text":"default.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" with:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"description","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"processors","type":"text","marks":[{"type":"code_inline"}]},{"text":" list","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"optional pipeline-level ","type":"text"},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"paragraph","content":[{"text":"Keep ","type":"text"},{"text":"default.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" readable and focused. Move large format-specific logic into sub-pipelines where needed.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"ECS version","type":"text"}]},{"type":"paragraph","content":[{"text":"Set the pipeline ECS reference version explicitly at the top of ","type":"text"},{"text":"processors","type":"text","marks":[{"type":"code_inline"}]},{"text":" (after any introductory processors you already use). ","type":"text"},{"text":"Use ","type":"text","marks":[{"type":"strong"}]},{"text":"9.3.0","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" — do not pin an older ECS version. ","type":"text"},{"text":"Exception:","type":"text","marks":[{"type":"strong"}]},{"text":" when the orchestrator specifies this is an entity data stream (","type":"text"},{"text":"event.kind: asset","type":"text","marks":[{"type":"code_inline"}]},{"text":"), use ","type":"text"},{"text":"9.5.0","type":"text","marks":[{"type":"code_inline"}]},{"text":" instead — entity leaf fields (","type":"text"},{"text":"entity.attributes.*","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"entity.lifecycle.*","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"entity.relationships.*","type":"text","marks":[{"type":"code_inline"}]},{"text":") do not exist at ","type":"text"},{"text":"9.3.0","type":"text","marks":[{"type":"code_inline"}]},{"text":" and cause ","type":"text"},{"text":"field is undefined","type":"text","marks":[{"type":"code_inline"}]},{"text":" build failures at that pin.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":" - set:\n field: ecs.version\n tag: set_ecs_version\n value: '9.3.0'","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Rename vs set (mapping to ECS)","type":"text"}]},{"type":"paragraph","content":[{"text":"When moving a value from a ","type":"text"},{"text":"custom or vendor field","type":"text","marks":[{"type":"strong"}]},{"text":" into an ","type":"text"},{"text":"ECS field","type":"text","marks":[{"type":"strong"}]},{"text":", ","type":"text"},{"text":"prefer the ","type":"text","marks":[{"type":"strong"}]},{"text":"rename","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" processor","type":"text","marks":[{"type":"strong"}]},{"text":" so the source field is removed and you avoid duplicate data. Use ","type":"text"},{"text":"set","type":"text","marks":[{"type":"code_inline"}]},{"text":" with ","type":"text"},{"text":"copy_from","type":"text","marks":[{"type":"code_inline"}]},{"text":" only when you must keep the source field or when ","type":"text"},{"text":"rename","type":"text","marks":[{"type":"code_inline"}]},{"text":" is not applicable.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Processor tags","type":"text"}]},{"type":"paragraph","content":[{"text":"Every processor","type":"text","marks":[{"type":"strong"}]},{"text":" in the pipeline should have a ","type":"text"},{"text":"tag","type":"text","marks":[{"type":"code_inline"}]},{"text":" (not only processors that can fail). Tags make failures and telemetry attributable to a specific step.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"CEL-only opening processors (Agentless metadata and error-only documents)","type":"text"}]},{"type":"paragraph","content":[{"text":"For ","type":"text"},{"text":"CEL-based","type":"text","marks":[{"type":"strong"}]},{"text":" integrations only, include these ","type":"text"},{"text":"before","type":"text","marks":[{"type":"strong"}]},{"text":" the standard ","type":"text"},{"text":"message","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" handling when they apply:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"remove","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":": drop Agentless metadata fields (","type":"text"},{"text":"organization","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"division","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"team","type":"text","marks":[{"type":"code_inline"}]},{"text":") when all are strings, so they do not collide with ECS. Use ","type":"text"},{"text":"ignore_missing: true","type":"text","marks":[{"type":"code_inline"}]},{"text":" and a conditional ","type":"text"},{"text":"if","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"terminate","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":": stop processing when the document is an error placeholder from the collector (","type":"text"},{"text":"ctx.error?.message != null && ctx.message == null && ctx.event?.original == null","type":"text","marks":[{"type":"code_inline"}]},{"text":").","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Non-CEL","type":"text","marks":[{"type":"strong"}]},{"text":" integrations (logs, syslog, filebeat-style inputs) ","type":"text"},{"text":"must not","type":"text","marks":[{"type":"strong"}]},{"text":" copy this block blindly — those fields and error shapes are specific to the CEL/Agentless path. See the ","type":"text"},{"text":"create-integration","type":"text","marks":[{"type":"code_inline"}]},{"text":" skill: the orchestrator must only expect this block when the data stream uses CEL input.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Standard opening: ECS, optional CEL block, JSE00001, then parse ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"paragraph","content":[{"text":"After the optional CEL-only processors, the pipeline should follow this shape. ","type":"text"},{"text":"All parsing","type":"text","marks":[{"type":"strong"}]},{"text":" (","type":"text"},{"text":"json","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"csv","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"grok","type":"text","marks":[{"type":"code_inline"}]},{"text":", etc.) runs on ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":". ","type":"text"},{"text":"Never overwrite or mutate ","type":"text","marks":[{"type":"strong"}]},{"text":"event.original","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" in later processors — derive structured fields into other paths (for example ","type":"text"},{"text":"json","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"_temp.*","type":"text","marks":[{"type":"code_inline"}]},{"text":", ECS fields).","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"description: Parse \u003cdataset> events.\nprocessors:\n - set:\n field: ecs.version\n tag: set_ecs_version\n value: '9.3.0'\n\n # --- CEL input only (omit for log/syslog-only streams) ---\n - remove:\n field:\n - organization\n - division\n - team\n ignore_missing: true\n if: ctx.organization instanceof String && ctx.division instanceof String && ctx.team instanceof String\n tag: remove_agentless_tags\n description: >-\n Removes the fields added by Agentless as metadata,\n as they can collide with ECS fields.\n - terminate:\n tag: data_collection_error\n if: ctx.error?.message != null && ctx.message == null && ctx.event?.original == null\n description: error message set and no data to process.\n # --- end CEL-only ---\n\n - rename:\n field: message\n tag: rename_message_to_event_original\n target_field: event.original\n ignore_missing: true\n 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.\n if: ctx.event?.original == null\n - remove:\n field: message\n tag: remove_message\n ignore_missing: true\n description: The `message` field is no longer required if the document has an `event.original` field.\n if: ctx.event?.original != null\n\n # Parse (always read from event.original; do not modify event.original)\n - json:\n field: event.original\n target_field: json\n tag: parse_json\n if: ctx.event?.original != null\n\n # ... normalize, enrich, ECS categorization, cleanup ...\n\n - append:\n field: tags\n value: preserve_original_event\n allow_duplicates: false\n if: ctx.error?.message != null\n\non_failure:\n - append:\n field: error.message\n value: >-\n Processor '{{{ _ingest.on_failure_processor_type }}}'\n {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'\n {{{/_ingest.on_failure_processor_tag}}}in pipeline '{{{ _ingest.pipeline }}}'\n failed with message '{{{ _ingest.on_failure_message }}}'\n - set:\n field: event.kind\n tag: set_pipeline_error_to_event_kind\n value: pipeline_error\n - append:\n field: tags\n value: preserve_original_event\n allow_duplicates: false","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Single-path pattern (linear pipeline)","type":"text"}]},{"type":"paragraph","content":[{"text":"Use this pattern when one parser flow handles all events. Combine the ","type":"text"},{"text":"standard opening","type":"text","marks":[{"type":"strong"}]},{"text":" (ECS version, optional CEL-only block, JSE00001 rename/remove, parse from ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" without mutating it), middle processors with ","type":"text"},{"text":"tags on every step","type":"text","marks":[{"type":"strong"}]},{"text":", and the ","type":"text"},{"text":"pipeline-level ","type":"text","marks":[{"type":"strong"}]},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" and ","type":"text"},{"text":"conditional ","type":"text","marks":[{"type":"strong"}]},{"text":"append","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" for ","type":"text","marks":[{"type":"strong"}]},{"text":"preserve_original_event","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" shown above.","type":"text"}]},{"type":"paragraph","content":[{"text":"Example middle section (illustrative):","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":" - grok:\n field: event.original\n patterns:\n - '^...

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n tag: parse_main\n\n - date:\n field: some.time\n target_field: '@timestamp'\n formats: [ISO8601]\n tag: parse_timestamp\n - convert:\n field: http.response.status_code\n type: long\n ignore_missing: true\n tag: convert_status\n\n - user_agent:\n field: user_agent.original\n ignore_missing: true\n tag: enrich_user_agent\n - geoip:\n field: source.ip\n target_field: source.geo\n if: ctx.source?.ip != null\n ignore_missing: true\n tag: enrich_source_geo\n - geoip:\n database_file: GeoLite2-ASN.mmdb\n field: source.ip\n target_field: source.as\n if: ctx.source?.ip != null\n properties:\n - asn\n - organization_name\n ignore_missing: true\n tag: enrich_source_asn\n - rename:\n field: source.as.asn\n target_field: source.as.number\n ignore_missing: true\n tag: rename_source_asn\n - rename:\n field: source.as.organization_name\n target_field: source.as.organization.name\n ignore_missing: true\n tag: rename_source_as_org\n\n - set:\n field: event.kind\n tag: set_event_kind\n value: event\n - append:\n field: event.category\n tag: append_event_category_web\n value: web\n - remove:\n field: temp\n ignore_missing: true\n tag: remove_temp","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Branching pattern (router + sub-pipelines)","type":"text"}]},{"type":"paragraph","content":[{"text":"Use branching when event formats or object models diverge:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"format-based branching (for example JSON vs text)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"class/category-based branching (for example OCSF class/category routing)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"object-presence branching (","type":"text"},{"text":"ctx.ocsf.user != null","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Pattern:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"processors:\n - pipeline:\n name: '{{ IngestPipeline \"pipeline_branch_json\" }}'\n if: ctx.event?.original != null && ctx.event.original.startsWith('{')\n ignore_missing_pipeline: true\n tag: route_json\n - pipeline:\n name: '{{ IngestPipeline \"pipeline_branch_text\" }}'\n if: ctx.event?.original != null && !ctx.event.original.startsWith('{')\n ignore_missing_pipeline: true\n tag: route_text","type":"text"}]},{"type":"paragraph","content":[{"text":"In large integrations, keep ","type":"text"},{"text":"default.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" as the router and put branch logic in files like:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"pipeline_object_\u003cname>.yml","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"pipeline_category_\u003cname>.yml","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"paragraph","content":[{"text":"See ","type":"text"},{"text":"references/branching-patterns.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for full patterns from ","type":"text"},{"text":"amazon_security_lake","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Sub-pipeline routing for multi-log-type integrations","type":"text"}]},{"type":"paragraph","content":[{"text":"When a data stream receives multiple distinct log types (for example a firewall that emits traffic, auth, and DNS logs in the same stream), ","type":"text"},{"text":"do not implement all parsing in a single monolithic ","type":"text","marks":[{"type":"strong"}]},{"text":"default.yml","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":". Use ","type":"text"},{"text":"default.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" as a thin router that detects the log type and delegates to a dedicated sub-pipeline per type.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"File layout","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"text"},"content":[{"text":"elasticsearch/ingest_pipeline/\n default.yml # router only — detects log type, calls sub-pipelines\n pipeline-\u003ctype>.yml # one file per log type (e.g. pipeline-traffic.yml)","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Router pattern in ","type":"text"},{"text":"default.yml","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"paragraph","content":[{"text":"Use the same ","type":"text"},{"text":"ecs.version","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":", ","type":"text"},{"text":"JSE00001","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"rename","type":"text","marks":[{"type":"code_inline"}]},{"text":"/","type":"text"},{"text":"remove","type":"text","marks":[{"type":"code_inline"}]},{"text":" pair for ","type":"text"},{"text":"message","type":"text","marks":[{"type":"code_inline"}]},{"text":", and ","type":"text"},{"text":"full pipeline-level ","type":"text","marks":[{"type":"strong"}]},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" as in the standard opening. The router only branches sub-pipelines; it does not parse payloads.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"processors:\n - set:\n field: ecs.version\n tag: set_ecs_version\n value: '9.3.0'\n - rename:\n field: message\n tag: rename_message_to_event_original\n target_field: event.original\n ignore_missing: true\n if: ctx.event?.original == null\n - remove:\n field: message\n tag: remove_message\n ignore_missing: true\n if: ctx.event?.original != null\n - pipeline:\n name: '{{ IngestPipeline \"pipeline-traffic\" }}'\n if: 'ctx.event?.original != null && ctx.event.original.contains(\"TRAFFIC\")'\n tag: route_traffic\n - pipeline:\n name: '{{ IngestPipeline \"pipeline-auth\" }}'\n if: 'ctx.event?.original != null && ctx.event.original.contains(\"AUTH\")'\n tag: route_auth\n - pipeline:\n name: '{{ IngestPipeline \"pipeline-dns\" }}'\n if: 'ctx.event?.original != null && ctx.event.original.contains(\"DNS\")'\n tag: route_dns\non_failure:\n - append:\n field: error.message\n value: >-\n Processor '{{{ _ingest.on_failure_processor_type }}}'\n {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'\n {{{/_ingest.on_failure_processor_tag}}}in pipeline '{{{ _ingest.pipeline }}}'\n failed with message '{{{ _ingest.on_failure_message }}}'\n - set:\n field: event.kind\n tag: set_pipeline_error_to_event_kind\n value: pipeline_error\n - append:\n field: tags\n value: preserve_original_event\n allow_duplicates: false","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Rules","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"default.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" must contain ","type":"text"},{"text":"only","type":"text","marks":[{"type":"strong"}]},{"text":" routing logic and ","type":"text"},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"}]},{"text":" handling — no field parsing.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Each sub-pipeline handles parsing, ECS mapping, and categorization for its own log type.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Each sub-pipeline must have its own ","type":"text"},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"}]},{"text":" block.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Name sub-pipeline files ","type":"text"},{"text":"pipeline-\u003ctype>.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" where ","type":"text"},{"text":"\u003ctype>","type":"text","marks":[{"type":"code_inline"}]},{"text":" matches the log type identifier used in the routing condition.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Each log type gets its own pipeline test fixture file following the naming convention ","type":"text"},{"text":"test-\u003cpackage>-\u003cdatastream>-\u003ctype>-sample.log","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Processor ordering and performance","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"run cheap existence checks before expensive operations","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"drop early if records are out of scope","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"prefer ","type":"text"},{"text":"dissect","type":"text","marks":[{"type":"code_inline"}]},{"text":" over ","type":"text"},{"text":"grok","type":"text","marks":[{"type":"code_inline"}]},{"text":" for stable delimited formats","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"never use a ","type":"text","marks":[{"type":"strong"}]},{"text":"script","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" processor when a built-in processor can do the job","type":"text","marks":[{"type":"strong"}]},{"text":" — ","type":"text"},{"text":"set","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"rename","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"remove","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"append","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"convert","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"dissect","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"grok","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"gsub","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"lowercase","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"uppercase","type":"text","marks":[{"type":"code_inline"}]},{"text":", and ","type":"text"},{"text":"trim","type":"text","marks":[{"type":"code_inline"}]},{"text":" are all faster than Painless and easier to review. See the cost tiers in ","type":"text"},{"text":"references/processor-cookbook.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"Processor performance guide","type":"text","marks":[{"type":"strong"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"use enrichment processors (","type":"text"},{"text":"geoip","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"user_agent","type":"text","marks":[{"type":"code_inline"}]},{"text":") only when needed","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"always anchor ","type":"text"},{"text":"grok","type":"text","marks":[{"type":"code_inline"}]},{"text":" patterns with ","type":"text"},{"text":"^","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"$","type":"text","marks":[{"type":"code_inline"}]},{"text":" — 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","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Mustache template syntax in processor values","type":"text"}]},{"type":"paragraph","content":[{"text":"Ingest pipeline processors use Mustache templates to reference field values in ","type":"text"},{"text":"value","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"message","type":"text","marks":[{"type":"code_inline"}]},{"text":", and similar string parameters. Use ","type":"text"},{"text":"triple braces","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"{{{field}}}","type":"text","marks":[{"type":"code_inline"}]},{"text":" with ","type":"text"},{"text":"single quotes","type":"text","marks":[{"type":"strong"}]},{"text":" — never double braces or double quotes:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"# CORRECT — triple braces, single quotes\n- append:\n field: related.user\n value: '{{{user.target.email}}}'\n allow_duplicates: false\n if: ctx.user?.target?.email != null\n\n# WRONG — double braces HTML-escape the value; double quotes\n- append:\n field: related.user\n value: \"{{user.target.email}}\"\n allow_duplicates: false\n if: ctx.user?.target?.email != null","type":"text"}]},{"type":"paragraph","content":[{"text":"Why: Mustache double braces ","type":"text"},{"text":"{{...}}","type":"text","marks":[{"type":"code_inline"}]},{"text":" HTML-encode the value (e.g., ","type":"text"},{"text":"&","type":"text","marks":[{"type":"code_inline"}]},{"text":" becomes ","type":"text"},{"text":"&","type":"text","marks":[{"type":"code_inline"}]},{"text":"), which corrupts data in ingest pipelines. Triple braces ","type":"text"},{"text":"{{{...}}}","type":"text","marks":[{"type":"code_inline"}]},{"text":" emit the raw value. Single quotes prevent YAML from interpreting braces.","type":"text"}]},{"type":"paragraph","content":[{"text":"Exception:","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"{{ IngestPipeline \"...\" }}","type":"text","marks":[{"type":"code_inline"}]},{"text":" in ","type":"text"},{"text":"pipeline.name","type":"text","marks":[{"type":"code_inline"}]},{"text":" is a Go template directive processed at build time, not a Mustache template — it correctly uses double braces.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Error handling essentials","type":"text"}]},{"type":"paragraph","content":[{"text":"Use pipeline-level ","type":"text"},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"}]},{"text":" as the main error reporting mechanism.","type":"text"}]},{"type":"paragraph","content":[{"text":"Recommended baseline (order matters):","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"append","type":"text","marks":[{"type":"strong"}]},{"text":" contextual ","type":"text"},{"text":"error.message","type":"text","marks":[{"type":"code_inline"}]},{"text":" first using ","type":"text"},{"text":"_ingest.on_failure_*","type":"text","marks":[{"type":"code_inline"}]},{"text":" variables (full template in the standard opening example)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"set","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"event.kind: pipeline_error","type":"text","marks":[{"type":"code_inline"}]},{"text":" (with a ","type":"text"},{"text":"tag","type":"text","marks":[{"type":"code_inline"}]},{"text":" on the ","type":"text"},{"text":"set","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"append","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"preserve_original_event","type":"text","marks":[{"type":"code_inline"}]},{"text":" to ","type":"text"},{"text":"tags","type":"text","marks":[{"type":"code_inline"}]},{"text":" when you need to retain the failed document for triage","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"give ","type":"text"},{"text":"every","type":"text","marks":[{"type":"strong"}]},{"text":" processor a ","type":"text"},{"text":"tag","type":"text","marks":[{"type":"code_inline"}]},{"text":" (not only processors that can fail)","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Use processor-level ","type":"text"},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"}]},{"text":" for local cleanup or fallback parsing, not as the primary global error message path.","type":"text"}]},{"type":"paragraph","content":[{"text":"See ","type":"text"},{"text":"references/error-handling-patterns.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for full examples and tradeoffs (","type":"text"},{"text":"ignore_failure","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"fail","type":"text","marks":[{"type":"code_inline"}]},{"text":", processor-level ","type":"text"},{"text":"on_failure","type":"text","marks":[{"type":"code_inline"}]},{"text":").","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"event.original handling (JSE00001)","type":"text"}]},{"type":"paragraph","content":[{"text":"The ","type":"text"},{"text":"elastic-package build","type":"text","marks":[{"type":"code_inline"}]},{"text":" validator enforces that pipelines correctly handle the ","type":"text"},{"text":"message","type":"text","marks":[{"type":"code_inline"}]},{"text":" to ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" rename. This check is known as JSE00001. New packages must comply; some legacy packages exclude it via ","type":"text"},{"text":"validation.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Required two-processor pattern","type":"text"}]},{"type":"paragraph","content":[{"text":"Every pipeline that consumes a ","type":"text"},{"text":"message","type":"text","marks":[{"type":"code_inline"}]},{"text":" field must include both processors (typically ","type":"text"},{"text":"after","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"ecs.version","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"after","type":"text","marks":[{"type":"strong"}]},{"text":" any CEL-only ","type":"text"},{"text":"remove","type":"text","marks":[{"type":"code_inline"}]},{"text":"/","type":"text"},{"text":"terminate","type":"text","marks":[{"type":"code_inline"}]},{"text":" steps when applicable):","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"- rename:\n field: message\n tag: rename_message_to_event_original\n target_field: event.original\n ignore_missing: true\n 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.\n if: ctx.event?.original == null\n- remove:\n field: message\n tag: remove_message\n ignore_missing: true\n description: The `message` field is no longer required if the document has an `event.original` field.\n if: ctx.event?.original != null","type":"text"}]},{"type":"paragraph","content":[{"text":"Step 1 (","type":"text"},{"text":"rename","type":"text","marks":[{"type":"code_inline"}]},{"text":"): moves ","type":"text"},{"text":"message","type":"text","marks":[{"type":"code_inline"}]},{"text":" into ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":", but only when ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" is not already populated (idempotent when a prior pipeline or Logstash has already set it).","type":"text"}]},{"type":"paragraph","content":[{"text":"Step 2 (","type":"text"},{"text":"remove","type":"text","marks":[{"type":"code_inline"}]},{"text":"): removes the redundant ","type":"text"},{"text":"message","type":"text","marks":[{"type":"code_inline"}]},{"text":" field when ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" is present (after rename or from an upstream producer).","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Do NOT add an ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" removal processor at the end of the pipeline","type":"text"}]},{"type":"paragraph","content":[{"text":"Some existing integrations contain a ","type":"text"},{"text":"remove","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor that deletes ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" at the end of the pipeline when ","type":"text"},{"text":"preserve_original_event","type":"text","marks":[{"type":"code_inline"}]},{"text":" is not in ","type":"text"},{"text":"tags","type":"text","marks":[{"type":"code_inline"}]},{"text":". ","type":"text"},{"text":"This pattern is deprecated and must not be used in new pipelines.","type":"text","marks":[{"type":"strong"}]},{"text":" The removal of ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" 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.","type":"text"}]},{"type":"paragraph","content":[{"text":"That stack-side final pipeline honors the ","type":"text"},{"text":"preserve_original_event","type":"text","marks":[{"type":"code_inline"}]},{"text":" tag, so a package exposing the toggle needs NO in-pipeline remove for the toggle to work — the toggle is not a no-op without one. Never recommend adding this processor; when reviewing, its ","type":"text"},{"text":"presence","type":"text","marks":[{"type":"em"}]},{"text":" is the finding, never its absence.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Reference","type":"text"}]},{"type":"paragraph","content":[{"text":"The two-processor JSE00001 pattern (rename + remove of ","type":"text"},{"text":"message","type":"text","marks":[{"type":"code_inline"}]},{"text":") shown above is required and complete. Do not add any additional ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" processors beyond those two.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Timezone handling (","type":"text"},{"text":"tz_offset","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]},{"type":"paragraph","content":[{"text":"For data streams that include the ","type":"text"},{"text":"tz_offset","type":"text","marks":[{"type":"code_inline"}]},{"text":" manifest var (syslog streams where messages lack a timezone), set ","type":"text"},{"text":"event.timezone","type":"text","marks":[{"type":"code_inline"}]},{"text":" from ","type":"text"},{"text":"_conf.tz_offset","type":"text","marks":[{"type":"code_inline"}]},{"text":" early in the pipeline, before any date parsing:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"- set:\n field: event.timezone\n tag: set_event_timezone\n value: '{{{_conf.tz_offset}}}'\n if: ctx._conf?.tz_offset != null && ctx._conf.tz_offset != ''","type":"text"}]},{"type":"paragraph","content":[{"text":"This ensures date processors can apply the correct timezone when parsing timestamps that have no timezone component.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Syslog structured data (RFC 5424 SD-ELEMENT) parsing","type":"text"}]},{"type":"paragraph","content":[{"text":"For vendor ","type":"text"},{"text":"key=value","type":"text","marks":[{"type":"code_inline"}]},{"text":" payloads and RFC 5424 SD-ELEMENT blocks, three strategies are available: KV with ","type":"text"},{"text":"trim_value","type":"text","marks":[{"type":"code_inline"}]},{"text":" (simplest, Strategy 1), ","type":"text"},{"text":"SYSLOG5424SD","type":"text","marks":[{"type":"code_inline"}]},{"text":" grok + KV with regex splits (Strategy 2), and Painless for edge cases with embedded equals or mixed quoting (Strategy 3).","type":"text"}]},{"type":"paragraph","content":[{"text":"Prefer Strategy 1 or 2; use Painless only when KV edge cases demand it.","type":"text"}]},{"type":"paragraph","content":[{"text":"See ","type":"text"},{"text":"references/grok-recipes.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"Syslog structured data strategies","type":"text","marks":[{"type":"strong"}]},{"text":" for full code examples, key settings, and reference implementations.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Keyword fields delivered as numbers","type":"text"}]},{"type":"paragraph","content":[{"text":"Fields that carry identifiers, protocol codes, or other opaque values must be declared as ","type":"text"},{"text":"keyword","type":"text","marks":[{"type":"code_inline"}]},{"text":" in ","type":"text"},{"text":"fields.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" — even when the source data delivers them as numbers. Common examples:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"network protocol numbers (","type":"text"},{"text":"network.iana_number","type":"text","marks":[{"type":"code_inline"}]},{"text":")","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"port numbers used as identifiers","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"error codes, result codes, status codes","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"SNMP OIDs, event IDs, object class codes","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Do ","type":"text"},{"text":"not","type":"text","marks":[{"type":"strong"}]},{"text":" add a ","type":"text"},{"text":"convert","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor to stringify these values. Elasticsearch silently coerces numbers into ","type":"text"},{"text":"keyword","type":"text","marks":[{"type":"code_inline"}]},{"text":" strings at index time, so the pipeline can pass the raw numeric value through unchanged.","type":"text"}]},{"type":"paragraph","content":[{"text":"The field declaration in ","type":"text"},{"text":"fields.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"- name: network.iana_number\n type: keyword\n description: IANA protocol number.","type":"text"}]},{"type":"paragraph","content":[{"text":"Because the test runner compares raw value types against declared field types, it will flag ","type":"text"},{"text":"6","type":"text","marks":[{"type":"code_inline"}]},{"text":" (long) as a mismatch for ","type":"text"},{"text":"keyword","type":"text","marks":[{"type":"code_inline"}]},{"text":". Declare the field in ","type":"text"},{"text":"numeric_keyword_fields","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the pipeline test config so the runner accepts the numeric representation without requiring the fixture to artificially stringify the value. See ","type":"text"},{"text":"integration-testing/references/pipeline-testing.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the config syntax.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Vendor field naming","type":"text"}]},{"type":"paragraph","content":[{"text":"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 ","type":"text"},{"text":"src_ip","type":"text","marks":[{"type":"code_inline"}]},{"text":" to ","type":"text"},{"text":"source.ip","type":"text","marks":[{"type":"code_inline"}]},{"text":"). When a vendor field has no ECS equivalent, keep it under a vendor-namespaced prefix (e.g. ","type":"text"},{"text":"vendor.product.field_name","type":"text","marks":[{"type":"code_inline"}]},{"text":") using the original name from the source.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"related.ip population","type":"text"}]},{"type":"paragraph","content":[{"text":"Every IP address present in the document must be appended to ","type":"text","marks":[{"type":"strong"}]},{"text":"related.ip","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":".","type":"text","marks":[{"type":"strong"}]},{"text":" This includes source, destination, client, server, host, and any other IP fields — whatever applies to the event type.","type":"text"}]},{"type":"paragraph","content":[{"text":"Use one ","type":"text"},{"text":"append","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor per IP field, with ","type":"text"},{"text":"ignore_missing: true","type":"text","marks":[{"type":"code_inline"}]},{"text":" so it is a no-op when the field is absent. Place these processors after all IP fields have been set (for example after ","type":"text"},{"text":"geoip","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"convert","type":"text","marks":[{"type":"code_inline"}]},{"text":", and any ECS rename steps) and before the cleanup ","type":"text"},{"text":"remove","type":"text","marks":[{"type":"code_inline"}]},{"text":" processors.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":" - append:\n field: related.ip\n tag: append_source_ip_to_related\n value: '{{{source.ip}}}'\n allow_duplicates: false\n if: ctx.source?.ip != null\n - append:\n field: related.ip\n tag: append_destination_ip_to_related\n value: '{{{destination.ip}}}'\n allow_duplicates: false\n if: ctx.destination?.ip != null\n # repeat the same pattern for client.ip, server.ip, host.ip, and any other IP fields the pipeline sets","type":"text"}]},{"type":"paragraph","content":[{"text":"Rules:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use ","type":"text"},{"text":"allow_duplicates: false","type":"text","marks":[{"type":"code_inline"}]},{"text":" on every append to avoid repeated values.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Add an ","type":"text"},{"text":"if","type":"text","marks":[{"type":"code_inline"}]},{"text":" guard on every processor so it skips fields absent in the event.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Add one ","type":"text"},{"text":"append","type":"text","marks":[{"type":"code_inline"}]},{"text":" per IP field the pipeline actually writes — do not add processors for fields the pipeline never sets.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Painless script best practices","type":"text"}]},{"type":"paragraph","content":[{"text":"Before writing any ","type":"text","marks":[{"type":"strong"}]},{"text":"script","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" processor, you MUST check whether a built-in processor can do the same job.","type":"text","marks":[{"type":"strong"}]},{"text":" ","type":"text"},{"text":"script","type":"text","marks":[{"type":"code_inline"}]},{"text":" is the slowest general-purpose processor (Painless compilation + per-document execution). The following operations have dedicated processors that are cheaper and easier to review:","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"If you need to …","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Use this processor, not ","type":"text"},{"text":"script","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Copy, move, or rename a field","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"rename","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"set","type":"text","marks":[{"type":"code_inline"}]},{"text":" with ","type":"text"},{"text":"copy_from","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Set a constant or derived value","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"set","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Add a value to a list","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"append","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Change a field's type","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"convert","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Extract a substring from a delimited string","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"dissect","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Extract a substring with regex","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"grok","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Replace characters in a string","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"gsub","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Normalize case","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"lowercase","type":"text","marks":[{"type":"code_inline"}]},{"text":" / ","type":"text"},{"text":"uppercase","type":"text","marks":[{"type":"code_inline"}]}]}]}]}]},{"type":"paragraph","content":[{"text":"Only reach for ","type":"text"},{"text":"script","type":"text","marks":[{"type":"code_inline"}]},{"text":" 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 ","type":"text"},{"text":"dissect","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"grok","type":"text","marks":[{"type":"code_inline"}]},{"text":" genuinely cannot handle.","type":"text"}]},{"type":"paragraph","content":[{"text":"Case-insensitive comparisons — use ","type":"text","marks":[{"type":"strong"}]},{"text":"equalsIgnoreCase()","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" when casing is unpredictable","type":"text","marks":[{"type":"strong"}]}]},{"type":"paragraph","content":[{"text":"Syslog and vendor devices are often inconsistent about casing, so Painless scripts comparing vendor-specific free-text fields should use ","type":"text"},{"text":"equalsIgnoreCase()","type":"text","marks":[{"type":"code_inline"}]},{"text":" rather than ","type":"text"},{"text":"==","type":"text","marks":[{"type":"code_inline"}]},{"text":". However, ","type":"text"},{"text":"apply this judgement contextually, not blanket:","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use ","type":"text","marks":[{"type":"strong"}]},{"text":"equalsIgnoreCase()","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" when the vendor field value may vary in casing between devices, firmware versions, or log sources (e.g. action fields like ","type":"text"},{"text":"allow/Allow/ALLOW","type":"text","marks":[{"type":"code_inline"}]},{"text":", severity strings, free-text status fields).","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use ","type":"text","marks":[{"type":"strong"}]},{"text":"==","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" 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 ","type":"text"},{"text":"equalsIgnoreCase()","type":"text","marks":[{"type":"code_inline"}]},{"text":" to fixed-enum fields adds noise without value.","type":"text"}]}]}]},{"type":"code_block","attrs":{"wrap":false,"language":"painless"},"content":[{"text":"// Correct for unpredictable vendor casing\nif (ctx.vendor?.action?.equalsIgnoreCase('allow')) { ... }\n\n// Correct for a fixed lowercase API enum — == is appropriate here\nif (ctx.json?.event_type == 'login') { ... }\n\n// Incorrect for unpredictable casing — breaks on \"Allow\", \"ALLOW\"\nif (ctx.vendor?.action == 'allow') { ... }","type":"text"}]},{"type":"paragraph","content":[{"text":"Access ","type":"text","marks":[{"type":"strong"}]},{"text":"ctx","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" directly in script bodies — no null-safe operators","type":"text","marks":[{"type":"strong"}]}]},{"type":"paragraph","content":[{"text":"In ","type":"text"},{"text":"script","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor ","type":"text"},{"text":"source","type":"text","marks":[{"type":"code_inline"}]},{"text":" blocks, access ","type":"text"},{"text":"ctx","type":"text","marks":[{"type":"code_inline"}]},{"text":" fields directly. Use explicit null checks instead of the null-safe ","type":"text"},{"text":"?.","type":"text","marks":[{"type":"code_inline"}]},{"text":" operator.","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"painless"},"content":[{"text":"// Correct — direct access with explicit null check\nif (ctx.source != null && ctx.source.ip != null) { ... }\n\n// Incorrect — null-safe operator in a script body\nif (ctx.source?.ip != null) { ... }","type":"text"}]},{"type":"paragraph","content":[{"text":"Note: null-safe ","type":"text"},{"text":"?.","type":"text","marks":[{"type":"code_inline"}]},{"text":" is acceptable in processor ","type":"text"},{"text":"if","type":"text","marks":[{"type":"code_inline"}]},{"text":" conditions (YAML), which are a different Painless execution context:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"- append:\n field: related.ip\n value: '{{{source.ip}}}'\n if: ctx.source?.ip != null","type":"text"}]},{"type":"paragraph","content":[{"text":"Other rules","type":"text","marks":[{"type":"strong"}]}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Every ","type":"text"},{"text":"script","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor must have a ","type":"text"},{"text":"tag","type":"text","marks":[{"type":"code_inline"}]},{"text":" and a ","type":"text"},{"text":"description","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Keep scripts short and scoped — move complex logic into helper variables inside the script, not across multiple script processors.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Do not use ","type":"text","marks":[{"type":"strong"}]},{"text":"script","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":" when built-in processors suffice","type":"text","marks":[{"type":"strong"}]},{"text":" — see the mandatory checklist table at the top of this section.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"ECS categorization mapping","type":"text"}]},{"type":"paragraph","content":[{"text":"When mapping source event types or actions to ","type":"text"},{"text":"event.category","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"event.type","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"event.outcome","type":"text","marks":[{"type":"code_inline"}]},{"text":", and ","type":"text"},{"text":"event.action","type":"text","marks":[{"type":"code_inline"}]},{"text":", use the patterns in ","type":"text"},{"text":"references/processor-cookbook.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" → ","type":"text"},{"text":"ECS categorization mapping patterns","type":"text","marks":[{"type":"strong"}]},{"text":":","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Pattern A","type":"text","marks":[{"type":"strong"}]},{"text":" (script with ","type":"text"},{"text":"params","type":"text","marks":[{"type":"code_inline"}]},{"text":" lookup table): recommended for ","type":"text"},{"text":"5+ mappings","type":"text","marks":[{"type":"strong"}]},{"text":". Mapping data in ","type":"text"},{"text":"params","type":"text","marks":[{"type":"code_inline"}]},{"text":" enables Painless compilation caching and keeps the script body generic.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Pattern B","type":"text","marks":[{"type":"strong"}]},{"text":" (","type":"text"},{"text":"set","type":"text","marks":[{"type":"code_inline"}]},{"text":" processors with conditionals): for ","type":"text"},{"text":"fewer than 5 mappings","type":"text","marks":[{"type":"strong"}]},{"text":" where a script is overkill.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Pattern C","type":"text","marks":[{"type":"strong"}]},{"text":" (sub-pipeline): for ","type":"text"},{"text":"100+ mappings","type":"text","marks":[{"type":"strong"}]},{"text":", extract the categorization into a dedicated sub-pipeline file.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Do NOT","type":"text","marks":[{"type":"strong"}]},{"text":" use bulk ","type":"text"},{"text":"append","type":"text","marks":[{"type":"code_inline"}]},{"text":" processors (2 per event type = 50+ processors for 25 types) or inline Painless ","type":"text"},{"text":"if","type":"text","marks":[{"type":"code_inline"}]},{"text":"/","type":"text"},{"text":"else","type":"text","marks":[{"type":"code_inline"}]},{"text":" chains without ","type":"text"},{"text":"params","type":"text","marks":[{"type":"code_inline"}]},{"text":" (defeats compilation caching). These are explicit anti-patterns — see the cookbook for details.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Grok best practices","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"prefer ","type":"text"},{"text":"dissect","type":"text","marks":[{"type":"code_inline"}]},{"text":" when structure is fixed","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"use simpler grok patterns where possible","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"always anchor grok patterns with ","type":"text"},{"text":"^","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"$","type":"text","marks":[{"type":"code_inline"}]},{"text":":","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"# Correct — anchored, fails fast on non-matching lines\npatterns:\n - '^%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.

\n\n# Incorrect — unanchored, scans the whole string for a partial match\npatterns:\n - '%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}'","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"avoid unnecessary backtracking-heavy custom regex","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"add a ","type":"text"},{"text":"tag","type":"text","marks":[{"type":"code_inline"}]},{"text":" to every grok (and every other) processor","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"For grok syntax (three expression forms, inline regex, type coercion, ","type":"text"},{"text":"pattern_definitions","type":"text","marks":[{"type":"code_inline"}]},{"text":"), syslog header splitting recipes, and common mistakes, see ","type":"text"},{"text":"references/grok-recipes.md","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Prohibited patterns","type":"text"}]},{"type":"paragraph","content":[{"text":"These patterns exist in many legacy integrations but ","type":"text"},{"text":"must not","type":"text","marks":[{"type":"strong"}]},{"text":" be used in new or updated pipelines. Do not copy them from reference integrations.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Never set ","type":"text"},{"text":"event.ingested","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"paragraph","content":[{"text":"The ","type":"text"},{"text":"event.ingested","type":"text","marks":[{"type":"code_inline"}]},{"text":" field is managed by Elasticsearch outside the integration pipeline. ","type":"text"},{"text":"Do not","type":"text","marks":[{"type":"strong"}]},{"text":" add a ","type":"text"},{"text":"set","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor for ","type":"text"},{"text":"event.ingested","type":"text","marks":[{"type":"code_inline"}]},{"text":" in any integration pipeline. This includes patterns like:","type":"text"}]},{"type":"code_block","attrs":{"wrap":false,"language":"yaml"},"content":[{"text":"# PROHIBITED — do not use\n- set:\n field: event.ingested\n value: '{{{_ingest.timestamp}}}'","type":"text"}]},{"type":"paragraph","content":[{"text":"The pipeline ","type":"text"},{"text":"should","type":"text","marks":[{"type":"strong"}]},{"text":" set ","type":"text"},{"text":"@timestamp","type":"text","marks":[{"type":"code_inline"}]},{"text":" from the original event's timestamp. When the source data contains multiple timestamps, map them as follows:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"@timestamp","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":": the primary event timestamp parsed from the source data. This is required.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"event.created","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":": when the event was first created or recorded by the source system (if different from ","type":"text"},{"text":"@timestamp","type":"text","marks":[{"type":"code_inline"}]},{"text":").","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"event.start","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":": when an activity or period began (e.g., session start, connection start).","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"event.end","type":"text","marks":[{"type":"code_inline"},{"type":"strong"}]},{"text":": when an activity or period ended (e.g., session end, connection close).","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"If a source timestamp does not match the semantics of ","type":"text"},{"text":"event.created","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"event.start","type":"text","marks":[{"type":"code_inline"}]},{"text":", or ","type":"text"},{"text":"event.end","type":"text","marks":[{"type":"code_inline"}]},{"text":", map it to a custom field under the vendor namespace with ","type":"text"},{"text":"type: date","type":"text","marks":[{"type":"code_inline"}]},{"text":" in ","type":"text"},{"text":"fields.yml","type":"text","marks":[{"type":"code_inline"}]},{"text":" and use a ","type":"text"},{"text":"date","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor with the appropriate ","type":"text"},{"text":"target_field","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Never use ","type":"text"},{"text":"preserve_duplicate_custom_fields","type":"text","marks":[{"type":"code_inline"}]}]},{"type":"paragraph","content":[{"text":"The ","type":"text"},{"text":"preserve_duplicate_custom_fields","type":"text","marks":[{"type":"code_inline"}]},{"text":" tag pattern — where source fields are copied to ECS fields using ","type":"text"},{"text":"set","type":"text","marks":[{"type":"code_inline"}]},{"text":" with ","type":"text"},{"text":"copy_from","type":"text","marks":[{"type":"code_inline"}]},{"text":" and the originals are conditionally retained — is a legacy anti-pattern. ","type":"text"},{"text":"Do not use it in any new or updated pipeline.","type":"text","marks":[{"type":"strong"}]},{"text":" Do not add a ","type":"text"},{"text":"preserve_duplicate_custom_fields","type":"text","marks":[{"type":"code_inline"}]},{"text":" manifest variable, tag, or conditional logic.","type":"text"}]},{"type":"paragraph","content":[{"text":"Instead, follow these field mapping rules:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"When a source field maps to an ECS field, use ","type":"text"},{"text":"rename","type":"text","marks":[{"type":"code_inline"}]},{"text":" to move it directly. The source field is removed and no duplicate exists.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"When a type conversion is needed (e.g., string to date, string to long), use the appropriate processor (","type":"text"},{"text":"date","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"convert","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"set","type":"text","marks":[{"type":"code_inline"}]},{"text":" with ","type":"text"},{"text":"copy_from","type":"text","marks":[{"type":"code_inline"}]},{"text":") to populate the ECS target field, then ","type":"text"},{"text":"remove","type":"text","marks":[{"type":"code_inline"}]},{"text":" the source field in the cleanup section at the end of the pipeline.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Never design a pipeline that needs to preserve both the original vendor field and the ECS copy.","type":"text","marks":[{"type":"strong"}]},{"text":" The ECS field is the canonical location.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"If you encounter this pattern in a reference integration, ignore it — it is legacy.","type":"text"}]},{"type":"heading","attrs":{"level":3},"content":[{"text":"Never add an ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" removal processor at the end","type":"text"}]},{"type":"paragraph","content":[{"text":"As documented in the JSE00001 section above: do not add a ","type":"text"},{"text":"remove","type":"text","marks":[{"type":"code_inline"}]},{"text":" processor for ","type":"text"},{"text":"event.original","type":"text","marks":[{"type":"code_inline"}]},{"text":" at the end of the pipeline. This is handled by a separate final pipeline.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"References","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/processor-cookbook.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — processor selection, parsing/normalization/enrichment examples, ECS categorization mapping patterns (Pattern A/B/C + anti-patterns)","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/branching-patterns.md","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/error-handling-patterns.md","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/grok-recipes.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — grok syntax, type coercion, syslog header recipes, common mistakes, pattern library link","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/builder-subagent-guidance.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — 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 ","type":"text"},{"text":"path","type":"text","marks":[{"type":"strong"}]},{"text":" in the task prompt; the subagent reads it itself in its own fresh context. Do NOT embed/paste its contents into the task prompt.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"references/cdr-pipeline-requirements.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — cloud security / CDR integrations only (","type":"text"},{"text":"cloudsecurity_cdr","type":"text","marks":[{"type":"code_inline"}]},{"text":" category): event categorization, Must Have / Should Have fields, processor patterns, review checklist","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"entity-mappings/references/entity-pipeline-patterns.md","type":"text","marks":[{"type":"code_inline"}]},{"text":" — entity/inventory data streams only (","type":"text"},{"text":"event.kind: asset","type":"text","marks":[{"type":"code_inline"}]},{"text":"): categorization processors, ","type":"text"},{"text":"entity.id","type":"text","marks":[{"type":"code_inline"}]},{"text":" mirroring, boolean coercion, relationship objects, anti-patterns, pipeline review checklist","type":"text"}]}]}]}]},"metadata":{"date":"2026-11-02","name":"ingest-pipelines","author":"@skillopedia","source":{"stars":16,"repo_name":"integration-skills","origin_url":"https://github.com/elastic/integration-skills/blob/HEAD/skills/ingest-pipelines/SKILL.md","repo_owner":"elastic","body_sha256":"bcdf5c9297565cc7e5e336a0058965eed4cd725b52629c5fbf0f99b8fb254fe8","cluster_key":"c81cb6f05ff8ea3155f5be767de2566fc611da56bdc96d1cbdf5e174a76d3f34","clean_bundle":{"format":"clean-skill-bundle-v1","source":"elastic/integration-skills/skills/ingest-pipelines/SKILL.md","text_bytes":90008,"source_commit":"39792cf81750ee54cba5b653fbfb69706871d42b","skills_sh_rank":19225,"attachment_count":7,"text_attachments":7,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/ingest-pipelines/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"data-analytics","category_label":"Data"},"exact_dupes_collapsed_into_this":0},"license":"Apache-2.0","version":"v1","category":"data-analytics","metadata":{"author":"elastic","version":"1.0"},"import_tag":"skills-sh-top20k-2026-11","description":"Use when designing or modifying Elasticsearch ingest pipelines, including single-path parsing, branching logic, sub-pipelines, enrichment processors, and robust on_failure handling."}},"renderedAt":1793666277996}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.