设计或修改 Elasticsearch ingest…

适用于在 Elastic 集成包中构建或调整 data stream 的 Elasticsearch ingest pipeline，包括单路径解析、条件分支、子管道、enrichment 和错误处理。

适用场景

解决思路

1. 把本技能及其参考文件中的规则和模式视为权威来源；检查 `elastic/integrations` 中已有集成作参考时，如果发现与技能冲突，总是遵循本技能，不要复制已废弃或禁止的旧模式。
2. 在集成包中把 ingest pipeline 放在 `data_stream/<stream>/elasticsearch/ingest_pipeline/`；每个 stream 通常有 `default.yml`，包含 `description`、`processors` 列表和可选的 pipeline-level `on_failure`；保持 `default.yml` 可读且聚焦，把大型格式特定逻辑移到子管道。
3. 在 `processors` 顶部（已有介绍性处理器之后）显式设置 pipeline 的 ECS 参考版本；使用 `9.3.0`，不要 pin 更旧 ECS 版本；例外：当 orchestrator 指明这是 entity data stream（`event.kind: asset`）时改用 `9.5.0`，因为 `entity.attributes.*`、`entity.lifecycle.*`、`entity.relationships.*` 在 `9.3.0` 不存在并导致 `field is undefined` 构建失败；示例：`- set: {field: ecs.version, tag: set_ecs_version, value: '9.3.0'}`。
4. 把值从 custom/vendor 字段移动到 ECS 字段时，优先使用 `rename` processor，使源字段被移除并避免重复数据；仅在必须保留源字段或 `rename` 不适用时使用 `set` 加 `copy_from`。
5. 给 pipeline 中每个 processor 都设置 `tag`，不仅给可能失败的 processor；这样失败和遥测能归因到具体步骤。
6. 仅对 CEL-based 集成，在标准 `message` → `event.original` 处理前、且适用时加入 CEL-only opening processors：用 `remove` 删除 Agentless metadata 字段 `organization`、`division`、`team`（当三者都是字符串时），避免与 ECS 冲突，使用 `ignore_missing: true` 和条件 `if`；用 `terminate` 在文档是 collector 的错误占位文档时停止处理，条件为 `ctx.error?.message != null && ctx.message == null && ctx.event?.original == null`。
7. 非 CEL 集成（logs、syslog、filebeat-style inputs）不要盲目复制 CEL-only 块；那些字段和错误形状是 CEL/Agentless 路径特有的。
8. 按标准 opening 组织 pipeline：设置 ECS 版本；可选 CEL-only 块；JSE00001 的 `message` → `event.original` rename/remove；然后解析 `event.original`；所有解析（`json`、`csv`、`grok` 等）都读取 `event.original`；绝不在后续 processor 中覆盖或变更 `event.original`，要把结构化字段派生到其他路径（如 `json`、`_temp.*`、ECS 字段）。
9. 标准 opening 的具体处理：`rename` 字段 `message` 到 `event.original`，`tag: rename_message_to_event_original`，`ignore_missing: true`，条件 `if: ctx.event?.original == null`；再 `remove` 字段 `message`，`tag: remove_message`，`ignore_missing: true`，条件 `if: ctx.event?.original != null`；解析示例：`json` 的 `field: event.original`、`target_field: json`、`tag: parse_json`、条件 `if: ctx.event?.original != null`；最后按需做 normalize、enrich、ECS categorization、cleanup；当 `ctx.error?.message != null` 时用 `append` 把 `preserve_original_event` 加到 `tags`，`allow_duplicates: false`。
10. 配置 pipeline-level `on_failure`：先 `append` 上下文 `error.message`，使用 `_ingest.on_failure_processor_type`、`_ingest.on_failure_processor_tag`、`_ingest.on_failure_message` 变量；再 `set` `event.kind: pipeline_error`，并给该 `set` 加 `tag: set_pipeline_error_to_event_kind`；再 `append` `preserve_original_event` 到 `tags`，`allow_duplicates: false`。
11. 单路径线性 pipeline 模式：当一个 parser flow 处理所有事件时，组合标准 opening、每步带 `tag` 的中间 processors、pipeline-level `on_failure`、以及条件 append `preserve_original_event`；中间示例做法包括：`grok` 解析 `event.original` 并锚定 pattern；`date` 解析 `some.time` 到 `@timestamp`，格式 `[ISO8601]`；`convert` 转换 `http.response.status_code` 为 `long`；`user_agent` enrichment 处理 `user_agent.original`；`geoip` 处理 `source.ip` 到 `source.geo`；`geoip` 用 `database_file: GeoLite2-ASN.mmdb` 处理 `source.ip` 到 `source.as` 并取 `asn`、`organization_name`；`rename` `source.as.asn` 到 `source.as.number`；`rename` `source.as.organization_name` 到 `source.as.organization.name`；`set` `event.kind: event`；`append` `event.category: web`；`remove` `temp`。
12. 分支模式：当事件格式或对象模型分歧时使用；适用格式分支（如 JSON vs text）、类别分支（如 OCSF class/category 路由）、对象存在性分支（如 `ctx.ocsf.user != null`）。
13. 分支模式实现：用 `pipeline` processor 按条件路由，例如 name 为 `{{ IngestPipeline "pipeline_branch_json" }}`，条件 `ctx.event?.original != null && ctx.event.original.startsWith('{')`，`ignore_missing_pipeline: true`，`tag: route_json`；另一个路由到 `pipeline_branch_text`，条件为非 `startsWith('{')`；大型集成中让 `default.yml` 作为 router，把分支逻辑放在 `pipeline_object_<name>.yml` 或 `pipeline_category_<name>.yml`，并参考 `references/branching-patterns.md`。
14. 多日志类型集成使用子管道路由：当 data stream 接收多个不同日志类型（例如同一流中的 traffic、auth、DNS 日志）时，不要把全部解析放进单体的 `default.yml`；用 `default.yml` 作为 thin router，检测日志类型并委托给每种类型的专用子管道。
15. 多日志类型集成的文件布局：`elasticsearch/ingest_pipeline/default.yml` 仅做 router；每种日志类型一个 `pipeline-<type>.yml`（例如 `pipeline-traffic.yml`）。
16. 多日志类型 router 的 `default.yml`：使用与标准 opening 相同的 `ecs.version`、JSE00001 的 `rename`/`remove` `message` 对、完整 pipeline-level `on_failure`；router 只分支子管道，不解析 payload；用 `pipeline` processor 按条件路由，例如 `contains("TRAFFIC")`、`contains("AUTH")`、`contains("DNS")`，并加 `tag: route_traffic` 等。
17. 多日志类型规则：`default.yml` 必须只包含 routing logic 和 `on_failure` 处理，不包含字段解析；每个子管道处理其日志类型的 parsing、ECS mapping、categorization；每个子管道必须有自己的 `on_failure`；子管道文件命名为 `pipeline-<type>.yml`，其中 `<type>` 与路由条件中使用的日志类型标识匹配；每种日志类型有自己的 pipeline test fixture 文件，命名 `test-<package>-<datastream>-<type>-sample.log`。
18. 处理器排序和性能：在昂贵操作前先做便宜的存在检查；对超出范围的记录尽早 drop；对稳定分隔格式优先 `dissect` 而不是 `grok`；绝不在内置 processor 能完成工作时使用 `script` processor；`set`、`rename`、`remove`、`append`、`convert`、`dissect`、`grok`、`gsub`、`lowercase`、`uppercase`、`trim` 都比 Painless 更快、更易审查；仅在需要时使用 enrichment processors（`geoip`、`user_agent`）；总是用 `^` 和 `$` 锚定 `grok` pattern，否则 regex engine 会扫描整个输入字符串寻找部分匹配，慢且可能在嘈杂日志行产生错误结果。
19. 在 processor 的 `value`、`message` 等字符串参数中使用 Mustache 模板时，使用 triple braces `{{{field}}}` 和 single quotes，绝不用 double braces 或 double quotes；正确示例：`- append: {field: related.user, value: '{{{user.target.email}}}', allow_duplicates: false, if: ctx.user?.target?.email != null}`；错误示例使用 double braces 和 double quotes 包裹值。
20. 使用 triple braces 的原因：Mustache double braces `{{...}}` 会 HTML-encode 值（如 `&` 变成 `&amp;`），在 ingest pipeline 中损坏数据；triple braces `{{{...}}}` 输出原始值；single quotes 防止 YAML 解释 braces。例外：`pipeline.name` 中的 `{{ IngestPipeline "..." }}` 是构建时处理的 Go template directive，不是 Mustache，正确使用 double braces。
21. 错误处理：把 pipeline-level `on_failure` 作为主要错误报告机制；推荐 baseline 顺序为：先用 `_ingest.on_failure_*` 变量 append 上下文 `error.message`；再 set `event.kind: pipeline_error`（带 tag）；再在需要保留失败文档用于 triage 时 append `preserve_original_event` 到 `tags`；给每个 processor 加 `tag`，不只可能失败的。
22. processor-level `on_failure` 用于局部 cleanup 或 fallback parsing，不作为主要全局错误消息路径；完整示例和权衡（`ignore_failure`、`fail`、processor-level `on_failure`）见 `references/error-handling-patterns.md`。
23. JSE00001 的 `message` 到 `event.original` 处理：`elastic-package build` validator 强制 pipeline 正确完成该 rename；新 package 必须合规，一些 legacy package 通过 `validation.yml` 排除。
24. 每个消费 `message` 字段的 pipeline 都必须包含两个 processor（通常在 `ecs.version` 后、CEL-only `remove`/`terminate` 后）：`rename` 把 `message` 移到 `event.original`，`tag: rename_message_to_event_original`，`ignore_missing: true`，描述说明 `event.original` 已存在时不动它（例如 Logstash 已发送），条件 `if: ctx.event?.original == null`；`remove` 删除 `message`，`tag: remove_message`，`ignore_missing: true`，描述说明有 `event.original` 后不再需要 `message`，条件 `if: ctx.event?.original != null`。
25. JSE00001 第一步 `rename` 只在 `event.original` 未填充时把 `message` 移入 `event.original`，在 prior pipeline 或 Logstash 已设置时保持幂等；第二步 `remove` 在 `event.original` 存在时删除冗余 `message`。
26. 不要在 pipeline 末尾添加删除 `event.original` 的 `remove` processor。有些已有集成在 `preserve_original_event` 不在 `tags` 时删除 `event.original`，该模式已废弃，不得用于新 pipeline；为存储优化而删除 `event.original` 现在由 integration 外部的 separate final pipeline 处理；不要从仍有该模式的参考集成复制。
27. stack-side final pipeline 会尊重 `preserve_original_event` tag，因此 package 暴露该开关时不需要 in-pipeline remove 也能生效；永远不要推荐添加这个 processor；审查时如果它存在才是 finding，缺失不是 finding。
28. JSE00001 参考：上面的 two-processor 模式（`message` 的 rename + remove）是必需且完整的；不要在这两个之外添加任何额外的 `event.original` processor。
29. 时区处理 `tz_offset`：对于包含 `tz_offset` manifest var 的 data stream（消息缺少时区的 syslog stream），在任何 date parsing 之前从 `_conf.tz_offset` 设置 `event.timezone`：`- set: {field: event.timezone, tag: set_event_timezone, value: '{{{_conf.tz_offset}}}', if: ctx._conf?.tz_offset != null && ctx._conf.tz_offset != ''}`；这确保 date processor 解析无时区组件的时间戳时应用正确时区。
30. Syslog structured data（RFC 5424 SD-ELEMENT）解析：对 vendor `key=value` payload 和 RFC 5424 SD-ELEMENT 块可用三种策略：Strategy 1 用 KV 加 `trim_value`（最简单）；Strategy 2 用 `SYSLOG5424SD` grok 加 KV 与 regex splits；Strategy 3 对含 embedded equals 或 mixed quoting 的边界情况用 Painless；优先 Strategy 1 或 2，仅在 KV 边界情况需要时用 Painless；完整代码示例、设置和参考实现见 `references/grok-recipes.md`。
31. Keyword fields delivered as numbers：携带 identifiers、protocol codes 或其他 opaque values 的字段必须在 `fields.yml` 声明为 `keyword`，即使源数据以数字交付；常见例子包括 network protocol numbers（`network.iana_number`）、用作标识的 port numbers、error codes、result codes、status codes、SNMP OIDs、event IDs、object class codes；不要添加 `convert` processor 把这些值 stringify；Elasticsearch 在 index time 会静默把数字强制成 `keyword` 字符串，所以 pipeline 可原样传递 numeric value。
32. Keyword fields delivered as numbers 的测试配置：在 `fields.yml` 写 `- name: network.iana_number, type: keyword, description: IANA protocol number.`；因为 test runner 比较 raw value types 与 declared field types，会把 `6`（long）标记为与 `keyword` 不匹配；在 pipeline test config 的 `numeric_keyword_fields` 中声明该字段，让 runner 接受 numeric representation，而不要求 fixture 人为 stringify；语法见 `integration-testing/references/pipeline-testing.md`。
33. Vendor field naming：精确保留源中出现的 vendor field names；不要 rename、reformat、normalize vendor-specific field names；唯一允许的 renaming 是把 vendor field 映射到 ECS field（如 `src_ip` 到 `source.ip`）；当 vendor field 没有 ECS 等价字段时，保留在 vendor-namespaced prefix 下（如 `vendor.product.field_name`），使用源中的原始名称。
34. related.ip population：文档中出现的每个 IP 地址都必须 append 到 `related.ip`；包括 source、destination、client、server、host 以及适用于该事件类型的其他 IP 字段。
35. related.ip population 实现：每个 IP 字段用一个 `append` processor，并设 `ignore_missing: true`，字段缺失时 no-op；放在所有 IP 字段已设置之后（例如 `geoip`、`convert`、ECS rename 后）、cleanup `remove` processors 之前；示例：`- append: {field: related.ip, tag: append_source_ip_to_related, value: '{{{source.ip}}}', allow_duplicates: false, if: ctx.source?.ip != null}`，destination、client、server、host 等重复同样模式。
36. related.ip 规则：每个 append 都使用 `allow_duplicates: false` 避免重复值；每个 processor 加 `if` guard 跳过 absent fields；只对 pipeline 实际写入的每个 IP 字段加一个 `append`，不要为 pipeline 从不设置的字段加 processor。
37. Painless script best practices：在写任何 `script` processor 前，必须检查 built-in processor 是否能完成同样工作；`script` 是最慢的通用 processor（Painless compilation + per-document execution）。
38. Painless 替代表：复制/移动/重命名字段用 `rename` 或 `set` with `copy_from`；设置常量或派生值用 `set`；向列表添加值用 `append`；改变字段类型用 `convert`；从分隔字符串提取子串用 `dissect`；用 regex 提取子串用 `grok`；替换字符串字符用 `gsub`；规范化大小写用 `lowercase`/`uppercase`；仅当没有 built-in processors 组合能表达逻辑时才用 `script`，例如 ECS categorization lookup tables with 5+ entries（Pattern A）、复杂条件算术、`dissect` 和 `grok` 真正无法处理的边界字符串解析。
39. Painless 大小写不敏感比较：当 casing 不可预测时使用 `equalsIgnoreCase()`；syslog 和 vendor devices 经常大小写不一致，所以比较 vendor-specific free-text fields 应使用 `equalsIgnoreCase()` 而不是 `==`；但要按上下文判断，不 blanket：当 vendor field value 可能因设备、固件版本或日志源变化大小写时（如 action 字段 `allow/Allow/ALLOW`、severity strings、free-text status fields）使用 `equalsIgnoreCase()`；当 API 或 spec 定义固定 lowercase enum 且总按指定交付时（如 ECS categorization fields、文档为 lowercase-only enums 的 API response fields）使用 `==`，给 fixed-enum fields 加 `equalsIgnoreCase()` 是噪音。
40. Painless 示例：不可预测 vendor casing 正确写法 `if (ctx.vendor?.action?.equalsIgnoreCase('allow')) { ... }`；固定 lowercase API enum 正确写法 `if (ctx.json?.event_type == 'login') { ... }`；不可预测 casing 的错误写法 `if (ctx.vendor?.action == 'allow') { ... }`。
41. Painless 访问 `ctx`：在 `script` processor 的 `source` block 中直接访问 `ctx` 字段，用显式 null checks，不用 null-safe `?.` 操作符；正确：`if (ctx.source != null && ctx.source.ip != null) { ... }`；错误：`if (ctx.source?.ip != null) { ... }`；注意 null-safe `?.` 在 processor `if` 条件（YAML）中可以接受，那是不同的 Painless 执行上下文。
42. Painless 其他规则：每个 `script` processor 必须有 `tag` 和 `description`；保持 scripts 短且 scoped，把复杂逻辑移到 script 内的 helper variables，不要跨多个 script processors；不要在内置 processors 足够时使用 `script`。
43. ECS categorization mapping：把 source event types 或 actions 映射到 `event.category`、`event.type`、`event.outcome`、`event.action` 时，使用 `references/processor-cookbook.md` 的 ECS categorization mapping patterns：Pattern A（带 `params` lookup table 的 script）推荐用于 5+ mappings，mapping 数据放在 `params` 可启用 Painless compilation caching 并保持 script body generic；Pattern B（带条件判断的 `set` processors）用于少于 5 个 mappings 且 script 过度；Pattern C（sub-pipeline）用于 100+ mappings，把 categorization 提取到专用子管道文件；不要使用 bulk `append` processors（每个 event type 2 个，25 种类型就是 50+ processors）或不带 `params` 的 inline Painless `if`/`else` 链（破坏 compilation caching），这些是明确 anti-patterns。
44. Grok best practices：结构固定时优先 `dissect`；尽可能使用更简单的 grok patterns；总是用 `^` 和 `$` 锚定 grok patterns；正确示例 `patterns: - '^%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}$'`，错误示例 `patterns: - '%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}'`，因为未锚定会扫描整个字符串寻找部分匹配；避免不必要 backtracking-heavy custom regex；给每个 grok（和其他每个）processor 加 `tag`；grok 语法（三种 expression forms、inline regex、type coercion、`pattern_definitions`）、syslog header splitting recipes、常见错误见 `references/grok-recipes.md`。
45. 禁止模式：绝不在 pipeline 中设置 `event.ingested`；`event.ingested` 由 Elasticsearch 在 integration pipeline 外管理；不要添加 `set` processor for `event.ingested`，包括 `- set: {field: event.ingested, value: '{{{_ingest.timestamp}}}'}`。
46. 时间戳映射规则：pipeline 应从原始事件时间戳设置 `@timestamp`；当源数据有多个时间戳时：`@timestamp` 是从源数据解析出的主事件时间戳，必需；`event.created` 是事件首次由源系统创建或记录的时间（如果不同于 `@timestamp`）；`event.start` 是活动或周期开始时间（如 session start、connection start）；`event.end` 是活动或周期结束时间（如 session end、connection close）；如果某个源时间戳不匹配 `event.created`、`event.start`、`event.end` 的语义，将其映射到 vendor namespace 下的 custom field，在 `fields.yml` 中声明 `type: date`，并使用带适当 `target_field` 的 `date` processor。
47. 禁止模式：绝不使用 `preserve_duplicate_custom_fields`；该 tag pattern（用 `set` with `copy_from` 把 source fields 复制到 ECS fields，并有条件保留 originals）是 legacy anti-pattern；不得在新或更新 pipeline 中使用；不要添加 `preserve_duplicate_custom_fields` manifest variable、tag 或条件逻辑。
48. 字段映射替代 `preserve_duplicate_custom_fields` 的规则：当 source field 映射到 ECS field 时，用 `rename` 直接移动，源字段被移除且无重复；当需要类型转换（如 string to date、string to long）时，用适当 processor（`date`、`convert`、`set` with `copy_from`）填充 ECS target field，然后在 pipeline 末尾 cleanup section `remove` 源字段；绝不要设计需要同时保留原始 vendor field 和 ECS copy 的 pipeline，ECS field 是 canonical location；如果参考集成中看到该模式，忽略它，它是 legacy。
49. 禁止模式：绝不添加 pipeline 末尾的 `event.original` removal processor；如 JSE00001 节所述，不要为 `event.original` 添加 `remove` processor，存储优化由 separate final pipeline 处理。
50. 查阅参考文件：`references/processor-cookbook.md` 用于 processor 选择、parsing/normalization/enrichment 示例、ECS categorization mapping patterns（Pattern A/B/C + anti-patterns）；`references/branching-patterns.md` 用于分支模式；`references/error-handling-patterns.md` 用于错误处理；`references/grok-recipes.md` 用于 grok syntax、type coercion、syslog header recipes、common mistakes、pattern library link。
51. 查阅参考文件：`references/builder-subagent-guidance.md` 是 subagent 操作手册，涵盖 scope boundaries、skill-load sequence、输入数据路径（CEL-first vs Direct）、9-step pipeline build workflow、审查生成输出而非手改 expected JSON、reporting contract；orchestrator 通过在 task prompt 中传入该文件的 path 来派发 subagent，subagent 在自己的 fresh context 中读取它；不要把其内容嵌入或粘贴到 task prompt。
52. 查阅参考文件：`references/cdr-pipeline-requirements.md` 仅用于 cloud security / CDR integrations（`cloudsecurity_cdr` category）：event categorization、Must Have / Should Have fields、processor patterns、review checklist。
53. 查阅参考文件：`entity-mappings/references/entity-pipeline-patterns.md` 仅用于 entity/inventory data streams（`event.kind: asset`）：categorization processors、`entity.id` mirroring、boolean coercion、relationship objects、anti-patterns、pipeline review checklist。

适用边界 · 注意事项

• 不要把它作为 ECS 字段选择、categorization values 和字段映射策略的主要指南；这些应使用 `ecs-field-mappings`。
• 不要把它作为 `elastic-package` 命令和 stack 生命周期工作流的主要指南；这些应使用 `elastic-package-cli`。
• 不要把它作为测试 fixture 编写和期望输出工作流的主要指南；这些应使用 `integration-testing` → `references/pipeline-testing.md`。