本技能（ingest-pipelines）用于指导 Elasticsearch ingest pipeline 的设计与修改，属于 elastic/integration-skills 集合。技能明确说明：技能及其参考文件中定义的规则和模式是权威来源；在参考 elastic/integrations 仓库中现有集成时，可能遇到与此处规范冲突的模式，许多集成包含早于当前标准的遗留模式，应始终遵循本技能，而不要复制已弃用或禁止的模式。
何时使用
适用于以下任务：为某个 data stream 构建或修改 elasticsearch/ingest_pipeline/default.yml；选择解析与规范化处理器（grok、dissect、json、kv、date、convert）；使用 pipeline 处理器设计条件分支与子管道路由；通过顶层 on_failure 实现可靠的错误处理；为摄取性能与可维护性调优处理器顺序。
何时不使用
不应作为以下主题的主要指南：ECS 字段选择、分类值与字段映射策略（ecs-field-mappings）；elastic-package 命令与栈生命周期工作流（elastic-package-cli）；测试夹具编写与预期输出工作流（integration-testing → references/pipeline-testing.md）。
管道结构
在集成包中，摄取管道位于 data_stream/<stream>/elasticsearch/ingest_pipeline/ 目录下。每个 stream 通常有一个 default.yml，包含 description、processors 列表，以及可选的管道级 on_failure。应保持 default.yml 可读且聚焦；大型格式特定逻辑应移入子管道。
ECS 版本
应在 processors 顶部显式设置管道 ECS 引用版本，使用 9.3.0，不要固定更旧的 ECS 版本。例外：当编排器指定这是实体数据流（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'
rename 与 set 的选择
将值从自定义或厂商字段移入 ECS 字段时，优先使用 rename 处理器，以便删除源字段、避免重复数据。仅当必须保留源字段或 rename 不适用时，才使用带 copy_from 的 set。
处理器标签
管道中的每个处理器都应有 tag（不仅是可能失败的处理器）。标签使失败和遥测可归因到具体步骤。
仅 CEL 的起始处理器
仅对基于 CEL 的集成，在适用时于标准 message → event.original 处理之前加入：remove（当 organization、division、team 均为字符串时移除 Agentless 元数据字段，避免与 ECS 冲突，使用 ignore_missing: true 与条件 if）；terminate（当文档是采集器的错误占位符时停止处理，条件为 ctx.error?.message != null && ctx.message == null && ctx.event?.original == null）。非 CEL 集成（日志、syslog、filebeat 风格输入）不得盲目复制该块，因为这些字段和错误形态特定于 CEL/Agentless 路径。
标准开头顺序
ECS、可选 CEL 块、JSE00001，然后解析 event.original。所有解析（json、csv、grok 等）都作用于 event.original；后续处理器绝不覆盖或修改 event.original，而是把结构化字段派生到其他路径（如 json、_temp.*、ECS 字段）。技能给出了包含 set ecs.version、CEL 块、rename message→event.original、remove message、json 解析、条件 append preserve_original_event 及 on_failure 的完整示例代码。
单路径模式（线性管道）
当一种解析流程处理所有事件时使用。结合标准开头、带标签的中间处理器，以及管道级 on_failure 和条件 append preserve_original_event。示例中段包括 grok、date、convert、user_agent、geoip（含 GeoLite2-ASN.mmdb 及 source.as 重命名）、set event.kind、append event.category、remove temp 等，并强调每个处理器带 tag。
分支模式（路由器 + 子管道）
当事件格式或对象模型分叉时使用：基于格式的分支（如 JSON 与文本）、基于 class/category 的分支（如 OCSF class/category 路由）、基于对象存在性的分支（ctx.ocsf.user != null）。模式中使用 pipeline 处理器并配合 '{{ IngestPipeline "pipeline_branch_json" }}' 名称、if 条件、ignore_missing_pipeline: true 与 tag。在大型集成中，default.yml 作为路由器，分支逻辑放在 pipeline_object_<name>.yml、pipeline_category_<name>.yml 等文件。
多日志类型集成的子管道路由
当 data stream 接收多种不同日志类型（如同一防火墙发出 traffic、auth、DNS 日志）时，不要把所有解析放在单一巨型 default.yml 中。应以 default.yml 作为瘦路由器检测日志类型并委托给每种类型的专用子管道。文件布局为 data_stream 下 ingest_pipeline 目录中的 default.yml（仅路由）与 pipeline-<type>.yml（每种日志类型一个文件）。路由器使用标准开头的 ecs.version 与 JSE00001 rename/remove 对，以及完整管道级 on_failure，只分支子管道不解析负载。技能给出 route_traffic、route_auth、route_dns 的条件与 tag 示例。
规则：default.yml 只能包含路由逻辑与 on_failure 处理，不得进行字段解析；每个子管道处理自身日志类型的解析、ECS 映射与分类；每个子管道必须有自己的 on_failure；子管道文件命名为 pipeline-<type>.yml，其中 <type> 与路由条件中使用的日志类型标识一致；每种日志类型有各自的管道测试夹具文件，命名遵循 test-<package>-<datastream>-<type>-sample.log。
处理器顺序与性能
在昂贵操作前先做廉价存在性检查；若记录超出范围应尽早丢弃；对稳定分隔格式优先用 dissect 而非 grok；绝不要在内置处理器可完成工作时使用 script 处理器——set、rename、remove、append、convert、dissect、grok、gsub、lowercase、uppercase、trim 都比 Painless 更快且更易审查；仅在需要时使用 geoip、user_agent 等富化处理器；grok 模式务必用 ^ 与 $ 锚定，否则正则引擎会扫描整个输入字符串寻找部分匹配，速度慢且在噪声日志行上可能产生错误结果。
处理器值中的 Mustache 模板语法
在 value、message 等字符串参数中引用字段值时应使用三花括号 {{{field}}} 配合单引号，绝不用双花括号或双引号。原因：双花括号 {{...}} 会对值做 HTML 编码（如 & 变成 &），破坏摄取管道中的数据；三花括号 {{{...}}} 输出原始值；单引号防止 YAML 解释花括号。例外：pipeline.name 中的 {{ IngestPipeline "..." }} 是构建时处理的 Go 模板指令，不是 Mustache 模板，正确使用双花括号。技能给出正确与错误示例对比。
错误处理要点
使用管道级 on_failure 作为主要错误报告机制。推荐基线（顺序重要）：先用 ingest.on_failure* 变量 append 上下文 error.message；设置 event.kind: pipeline_error（set 处理器带 tag）；需要保留失败文档用于排查时把 preserve_original_event append 到 tags；给每个处理器加 tag。处理器级 on_failure 用于本地清理或回退解析，而非主要全局错误消息路径。
event.original 处理（JSE00001）
elastic-package 构建校验器强制要求管道正确处理 message 到 event.original 的重命名，该检查称为 JSE00001。新包必须遵守；部分遗留包通过 validation.yml 排除该检查。每个消费 message 字段的管道都必须包含 rename 与 remove 两个处理器（通常在 ecs.version 之后、适用的 CEL 专用 remove/terminate 步骤之后）。第一步 rename 仅在 event.original 尚未填充时把 message 移入 event.original（当上游管道或 Logstash 已设置时保持幂等）；第二步 remove 在 event.original 存在时移除冗余的 message 字段。不要在管道末尾添加删除 event.original 的处理器——部分现有集成含此模式（当 tags 中无 preserve_original_event 时删除 event.original），该模式已弃用，新管道不得使用；为存储优化而移除 event.original 现由后续机制处理（原文此处截断）。