ingest-pipelines

ingest-pipelines

技能权威性

本技能及其参考文件中定义的规则和模式是权威事实来源。在 elastic/integrations 仓库中查看现有集成作为参考时，你可能会遇到与此处规定相冲突的模式——许多集成包含早于当前标准的旧模式。**始终以本技能为准，不要遵循其他集成中观察到的模式。**如果参考集成使用了已弃用或禁止的模式，请勿照搬。

适用场景

在以下任务中使用本技能：

• 为数据流构建或修改 elasticsearch/ingest_pipeline/default.yml
• 选择解析器和规范化处理器（grok、dissect、json、kv、date、convert）
• 使用 pipeline 处理器设计条件分支和子管道路由
• 使用顶层 on_failure 实现弹性错误处理
• 调整处理器顺序，以提升摄取性能和可维护性

不适用场景

以下任务不应以本技能作为主要指南：

• ECS 字段选择、分类值和字段映射策略（ecs-field-mappings）
• elastic-package 命令和 Stack 生命周期工作流（elastic-package-cli）
• 测试固件编写和预期输出工作流（integration-testing → references/pipeline-testing.md）

管道结构

在集成包中，摄取管道位于：

data_stream/<stream>/elasticsearch/ingest_pipeline/

每个数据流通常都有一个 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）

将值从自定义字段或供应商字段移至 ECS 字段时，优先使用 rename 处理器，以移除源字段并避免重复数据。只有在必须保留源字段或不适用 rename 时，才使用带有 copy_from 的 set。

处理器标签

管道中的每个处理器都应带有 tag（不仅限于可能失败的处理器）。标签便于将故障和遥测信息追溯到具体步骤。

CEL 专用的起始处理器（Agentless 元数据和仅含错误的文档）

仅对于基于 CEL 的集成，在适用时将以下处理器放在标准的 message → event.original 处理之前：

• remove：当 Agentless 元数据字段（organization、division、team）均为字符串时，将其移除，以免与 ECS 冲突。使用 ignore_missing: true 和条件 if。
• terminate：当文档是采集器提供的错误占位内容时，停止处理（ctx.error?.message != null && ctx.message == null && ctx.event?.original == null）。

非 CEL 集成（日志、syslog、filebeat 风格的输入）不得盲目复制此代码块——这些字段和错误形式仅适用于 CEL/Agentless 路径。请参阅 create-integration 技能：只有在数据流使用 CEL 输入时，编排器才应要求包含此代码块。

标准开头：ECS、可选的 CEL 块、JSE00001，然后解析 event.original

可选的仅限 CEL 处理器之后，管道应遵循以下结构。所有解析（json、csv、grok 等）都针对 event.original 运行。后续处理器中绝不要覆盖或修改 event.original——应将结构化字段派生到其他路径（例如 json、_temp.*、ECS 字段）。

description: Parse <dataset> events.
processors:
  - set:
      field: ecs.version
      tag: set_ecs_version
      value: '9.3.0'

  # --- CEL input only (omit for log/syslog-only streams) ---
  - remove:
      field:
        - organization
        - division
        - team
      ignore_missing: true
      if: ctx.organization instanceof String && ctx.division instanceof String && ctx.team instanceof String
      tag: remove_agentless_tags
      description: >-
        Removes the fields added by Agentless as metadata,
        as they can collide with ECS fields.
  - terminate:
      tag: data_collection_error
      if: ctx.error?.message != null && ctx.message == null && ctx.event?.original == null
      description: error message set and no data to process.
  # --- end CEL-only ---

  - rename:
      field: message
      tag: rename_message_to_event_original
      target_field: event.original
      ignore_missing: true
      description: Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.
      if: ctx.event?.original == null
  - remove:
      field: message
      tag: remove_message
      ignore_missing: true
      description: The `message` field is no longer required if the document has an `event.original` field.
      if: ctx.event?.original != null

  # Parse (always read from event.original; do not modify event.original)
  - json:
      field: event.original
      target_field: json
      tag: parse_json
      if: ctx.event?.original != null

  # ... normalize, enrich, ECS categorization, cleanup ...

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

on_failure:
  - append:
      field: error.message
      value: >-
        Processor '{{{ _ingest.on_failure_processor_type }}}'
        {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
        {{{/_ingest.on_failure_processor_tag}}}failed with message '{{{ _ingest.on_failure_message }}}'
  - set:
      field: event.kind
      tag: set_pipeline_error_to_event_kind
      value: pipeline_error
  - append:
      field: tags
      value: preserve_original_event
      allow_duplicates: false

单路径模式（线性管道）

当一个解析流程处理所有事件时，请使用此模式。将标准开头（ECS 版本、可选的仅限 CEL 块、JSE00001 重命名／移除操作，以及从 event.original 解析而不修改它）、每个步骤都带有标签的中间处理器，以及上文所示的管道级 on_failure 和**条件式 append（用于 preserve_original_event）**组合起来。

中间部分示例（仅供说明）：

  - grok:
      field: event.original
      patterns:
        - '^...$'
      tag: parse_main

  - date:
      field: some.time
      target_field: '@timestamp'
      formats: [ISO8601]
      tag: parse_timestamp
  - convert:
      field: http.response.status_code
      type: long
      ignore_missing: true
      tag: convert_status

  - user_agent:
      field: user_agent.original
      ignore_missing: true
      tag: enrich_user_agent
  - geoip:
      field: source.ip
      target_field: source.geo
      if: ctx.source?.ip != null
      ignore_missing: true
      tag: enrich_source_geo
  - geoip:
      database_file: GeoLite2-ASN.mmdb
      field: source.ip
      target_field: source.as
      if: ctx.source?.ip != null
      properties:
        - asn
        - organization_name
      ignore_missing: true
      tag: enrich_source_asn
  - rename:
      field: source.as.asn
      target_field: source.as.number
      ignore_missing: true
      tag: rename_source_asn
  - rename:
      field: source.as.organization_name
      target_field: source.as.organization.name
      ignore_missing: true
      tag: rename_source_as_org

  - set:
      field: event.kind
      tag: set_event_kind
      value: event
  - append:
      field: event.category
      tag: append_event_category_web
      value: web
  - remove:
      field: temp
      ignore_missing: true
      tag: remove_temp

分支模式（路由器 + 子管道）

当事件格式或对象模型不同时，请使用分支：

• 基于格式的分支（例如 JSON 与文本）
• 基于类别／类型的分支（例如 OCSF 类别／类型路由）
• 基于对象是否存在的分支（ctx.ocsf.user != null）

模式：

processors:
  - pipeline:
      name: '{{ IngestPipeline "pipeline_branch_json" }}'
      if: ctx.event?.original != null && ctx.event.original.startsWith('{')
      ignore_missing_pipeline: true
      tag: route_json
  - pipeline:
      name: '{{ IngestPipeline "pipeline_branch_text" }}'
      if: ctx.event?.original != null && !ctx.event.original.startsWith('{')
      ignore_missing_pipeline: true
      tag: route_text

对于大型集成，将 default.yml 保留为路由器，并将分支逻辑放入以下文件：

• pipeline_object_<name>.yml
• pipeline_category_<name>.yml

完整模式请参阅 references/branching-patterns.md，其中包含来自 amazon_security_lake 的示例。

多日志类型集成的子管道路由

当一个数据流接收多种不同的日志类型时（例如，同一个数据流中包含流量、身份验证和 DNS 日志的防火墙），不要在单个庞大的 default.yml 中实现所有解析逻辑。应将 default.yml 用作精简路由器，用于检测日志类型，并将处理委托给每种类型专属的子管道。

文件布局

elasticsearch/ingest_pipeline/
  default.yml              # router only — detects log type, calls sub-pipelines
  pipeline-<type>.yml      # one file per log type (e.g. pipeline-traffic.yml)

default.yml 中的路由器模式

使用与标准开头相同的 ecs.version、针对 message 的 JSE00001 rename／remove 操作对，以及完整的管道级 on_failure。路由器只负责分支调用子管道，不解析负载。

processors:
  - set:
      field: ecs.version
      tag: set_ecs_version
      value: '9.3.0'
  - rename:
      field: message
      tag: rename_message_to_event_original
      target_field: event.original
      ignore_missing: true
      if: ctx.event?.original == null
  - remove:
      field: message
      tag: remove_message
      ignore_missing: true
      if: ctx.event?.original != null
  - pipeline:
      name: '{{ IngestPipeline "pipeline-traffic" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("TRAFFIC")'
      tag: route_traffic
  - pipeline:
      name: '{{ IngestPipeline "pipeline-auth" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("AUTH")'
      tag: route_auth
  - pipeline:
      name: '{{ IngestPipeline "pipeline-dns" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("DNS")'
      tag: route_dns
on_failure:
  - append:
      field: error.message
      value: >-
        Processor '{{{ _ingest.on_failure_processor_type }}}'
        {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
        {{{/_ingest.on_failure_processor_tag}}}failed with message '{{{ _ingest.on_failure_message }}}'
  - set:
      field: event.kind
      tag: set_pipeline_error_to_event_kind
      value: pipeline_error
  - append:
      field: tags
      value: preserve_original_event
      allow_duplicates: false

规则

• 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 更快，也更易审查。请参阅 references/processor-cookbook.md → 处理器性能指南。
• 仅在需要时使用丰富处理器（geoip、user_agent）
• grok 模式始终应使用 ^ 和 $ 锚定——没有锚点时，正则表达式引擎会扫描整个输入字符串以查找部分匹配，这会拖慢速度，并可能在包含噪声的日志行中产生错误结果

处理器值中的 Mustache 模板语法

摄取管道处理器使用 Mustache 模板在 value、message 和类似的字符串参数中引用字段值。使用三重花括号 {{{field}}}，并使用单引号——绝不要使用双花括号或双引号：

# CORRECT — triple braces, single quotes
- append:
    field: related.user
    value: '{{{user.target.email}}}'
    allow_duplicates: false
    if: ctx.user?.target?.email != null

# WRONG — double braces HTML-escape the value; double quotes
- append:
    field: related.user
    value: "{{user.target.email}}"
    allow_duplicates: false
    if: ctx.user?.target?.email != null

原因：Mustache 双花括号 {{...}} 会对值进行 HTML 编码（例如，& 会变成 &），从而破坏摄取管道中的数据。三重花括号 {{{...}}} 会输出原始值。单引号可防止 YAML 将花括号解释为特殊字符。

例外：pipeline.name 中的 {{ IngestPipeline "..." }} 是在构建时处理的 Go 模板指令，而非 Mustache 模板，因此使用双花括号是正确的。

错误处理要点

使用管道级 on_failure 作为主要错误报告机制。

推荐的基线配置（顺序很重要）：

• 首先使用 _ingest.on_failure_* 变量将上下文信息追加到 error.message（完整模板见标准开头示例）
• 设置 event.kind: pipeline_error（并为 set 处理器添加 tag）
• 需要保留失败文档以供排查时，将 preserve_original_event 追加到 tags
• 为每个处理器添加 tag（不只是可能失败的处理器）

使用处理器级 on_failure 进行局部清理或回退解析，不要将其用作主要的全局错误消息处理路径。

完整示例和权衡取舍（ignore_failure、fail、处理器级 on_failure）请参阅 references/error-handling-patterns.md。

event.original 处理（JSE00001）

elastic-package build 验证器会强制检查管道是否正确处理将 message 重命名为 event.original。这项检查称为 JSE00001。新软件包必须遵守此要求；部分旧版软件包会通过 validation.yml 将其排除。

必需的双处理器模式

每个使用 message 字段的管道都必须包含这两个处理器（通常位于 ecs.version 之后，并在适用时位于任何仅限 CEL 的 remove/terminate 步骤之后）：

- rename:
    field: message
    tag: rename_message_to_event_original
    target_field: event.original
    ignore_missing: true
    description: Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.
    if: ctx.event?.original == null
- remove:
    field: message
    tag: remove_message
    ignore_missing: true
    description: The `message` field is no longer required if the document has an `event.original` field.
    if: ctx.event?.original != null

步骤 1（rename）：将 message 移到 event.original，但仅当 event.original 尚未填充时才执行（如果之前的管道或 Logstash 已设置该字段，则此操作仍具有幂等性）。

步骤 2（remove）：当 event.original 存在时，移除多余的 message 字段（无论是重命名后存在，还是由上游生成器提供）。

不要在管道末尾添加移除 event.original 的处理器

某些现有集成在管道末尾包含一个 remove 处理器，当 tags 中没有 preserve_original_event 时会删除 event.original。**此模式已弃用，不得用于新管道。**现在，集成之外的独立最终管道会负责移除 event.original 以优化存储。不要从仍保留此模式的参考集成中照搬此做法——这是旧版逻辑。

Stack 端的最终管道会遵循 preserve_original_event 标签，因此，暴露此开关的软件包无需在管道内添加 remove，该开关也不会因此失效。绝不要建议添加此处理器；在审查时，应将其存在视为问题，而不是将其缺失视为问题。

参考

上面展示的双处理器 JSE00001 模式（重命名 message 并将其移除）是必需且完整的。除了这两个处理器外，不要添加任何其他 event.original 处理器。

时区处理（tz_offset）

对于包含 tz_offset 清单变量的数据流（即消息不含时区的 syslog 数据流），应在管道早期、任何日期解析之前，将 event.timezone 设置为 _conf.tz_offset：

- set:
    field: event.timezone
    tag: set_event_timezone
    value: '{{{_conf.tz_offset}}}'
    if: ctx._conf?.tz_offset != null && ctx._conf.tz_offset != ''

这样可确保日期处理器在解析不含时区部分的时间戳时应用正确的时区。

Syslog 结构化数据（RFC 5424 SD-ELEMENT）解析

对于供应商的 key=value 负载和 RFC 5424 SD-ELEMENT 块，有三种策略可用：使用 trim_value 的 KV（最简单，策略 1）、SYSLOG5424SD grok + 使用正则拆分的 KV（策略 2），以及用于处理包含嵌入等号或混合引号等边缘情况的 Painless（策略 3）。

优先使用策略 1 或 2；只有在 KV 边缘情况有此需求时才使用 Painless。

完整代码示例、关键设置和参考实现，请参阅 references/grok-recipes.md → Syslog 结构化数据策略。

以数字形式传递的 keyword 字段

承载标识符、协议代码或其他不透明值的字段必须在 fields.yml 中声明为 keyword，即使源数据以数字形式传递这些值。常见示例：

• 网络协议编号（network.iana_number）
• 用作标识符的端口号
• 错误代码、结果代码、状态代码
• SNMP OID、事件 ID、对象类代码

不要添加 convert 处理器来将这些值转换为字符串。Elasticsearch 会在索引时自动将数字强制转换为 keyword 字符串，因此管道可以原样传递原始数值。

fields.yml 中的字段声明：

- name: network.iana_number
  type: keyword
  description: IANA protocol number.

由于测试运行器会将原始值类型与声明的字段类型进行比较，因此它会将 6（long）标记为与 keyword 不匹配。请在管道测试配置的 numeric_keyword_fields 中声明该字段，这样运行器就会接受数字表示形式，无需在测试夹具中人为地将值转换为字符串。配置语法请参阅 integration-testing/references/pipeline-testing.md。

供应商字段命名

按源数据中的原样保留供应商字段名称。不要重命名、重新格式化或规范化供应商专属字段名称——唯一允许的重命名是将供应商字段映射到 ECS 字段（例如，将 src_ip 重命名为 source.ip）。当供应商字段没有对应的 ECS 字段时，请使用源数据中的原始名称，将其保留在供应商命名空间前缀下（例如 vendor.product.field_name）。

填充 related.ip

文档中出现的每个 IP 地址都必须追加到 related.ip。 这包括源、目标、客户端、服务端、主机以及任何其他 IP 字段——具体取决于事件类型。

每个 IP 字段使用一个 append 处理器，并设置 ignore_missing: true，使其在字段缺失时不执行任何操作。请将这些处理器放在所有 IP 字段都已设置之后（例如，在 geoip、convert 和任何 ECS 重命名步骤之后），并放在清理用的 remove 处理器之前。

  - append:
      field: related.ip
      tag: append_source_ip_to_related
      value: '{{{source.ip}}}'
      allow_duplicates: false
      if: ctx.source?.ip != null
  - append:
      field: related.ip
      tag: append_destination_ip_to_related
      value: '{{{destination.ip}}}'
      allow_duplicates: false
      if: ctx.destination?.ip != null
  # repeat the same pattern for client.ip, server.ip, host.ip, and any other IP fields the pipeline sets

规则：

• 每个 append 都使用 allow_duplicates: false，以避免重复值。
• 每个处理器都添加 if 条件，确保字段在事件中缺失时跳过该处理器。
• 管道实际写入的每个 IP 字段都添加一个 append——不要为管道从未设置的字段添加处理器。

Painless 脚本最佳实践

编写任何 script 处理器之前，必须检查是否有内置处理器能完成相同任务。 script 是速度最慢的通用处理器（Painless 编译 + 逐文档执行）。以下操作有专用处理器，成本更低，也更容易审查：

如果需要……	请使用此处理器，而不是 script
复制、移动或重命名字段	rename 或带有 copy_from 的 set
设置常量或派生值	set
向列表中添加值	append
更改字段类型	convert
从分隔字符串中提取子字符串	dissect
使用正则表达式提取子字符串	grok
替换字符串中的字符	gsub
统一大小写	lowercase / uppercase

只有在内置处理器无法组合表达所需逻辑时，才使用 script——例如，包含 5 个以上条目的 ECS 分类查找表（模式 A）、复杂的条件算术，或 dissect 和 grok 确实无法处理的边缘字符串解析情况。

不区分大小写的比较——大小写不确定时使用 equalsIgnoreCase()

Syslog 和供应商设备的大小写通常不一致，因此 Painless 脚本在比较供应商特定的自由文本字段时，应使用 equalsIgnoreCase() 而不是 ==。不过，请根据具体情境判断，不要一概而论：

• 使用 equalsIgnoreCase()：当供应商字段值的大小写可能因设备、固件版本或日志来源而异时（例如 allow/Allow/ALLOW 之类的操作字段、严重性字符串、自由文本状态字段）。
• 使用 ==：当 API 或规范定义了固定的小写枚举，且值始终按规范提供时（例如 ECS 分类字段、文档中规定只能使用小写枚举值的 API 响应字段）。对固定枚举字段添加 equalsIgnoreCase() 只会增加冗余，不会带来好处。

// Correct for unpredictable vendor casing
if (ctx.vendor?.action?.equalsIgnoreCase('allow')) { ... }

// Correct for a fixed lowercase API enum — == is appropriate here
if (ctx.json?.event_type == 'login') { ... }

// Incorrect for unpredictable casing — breaks on "Allow", "ALLOW"
if (ctx.vendor?.action == 'allow') { ... }

在脚本正文中直接访问 ctx——不要使用 null 安全运算符

在 script 处理器的 source 块中，直接访问 ctx 字段。使用显式空值检查，而不是 null 安全运算符 ?.。

// Correct — direct access with explicit null check
if (ctx.source != null && ctx.source.ip != null) { ... }

// Incorrect — null-safe operator in a script body
if (ctx.source?.ip != null) { ... }

注意：在处理器的 if 条件（YAML）中可以使用 null 安全运算符 ?.，因为那是不同的 Painless 执行上下文：

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

其他规则

• 每个 script 处理器都必须有 tag 和 description。
• 脚本应简短且范围明确——将复杂逻辑放在脚本内部的辅助变量中，不要拆分到多个脚本处理器中。
• 内置处理器足以完成任务时，不要使用 script——参见本节顶部的强制检查清单表格。

ECS 分类映射

将来源事件类型或操作映射到 event.category、event.type、event.outcome 和 event.action 时，请使用 references/processor-cookbook.md → ECS 分类映射模式中的模式：

• 模式 A（使用 params 查找表的脚本）：推荐用于 5 个以上映射。将映射数据放在 params 中可启用 Painless 编译缓存，并使脚本正文保持通用。
• 模式 B（带条件的 set 处理器）：适用于少于 5 个映射且使用脚本显得过于复杂的情况。
• 模式 C（子流水线）：适用于 100 个以上映射，将分类逻辑提取到专用的子流水线文件中。

不要使用批量 append 处理器（每种事件类型使用 2 个，25 种类型就会有 50 多个处理器），也不要使用不带 params 的内联 Painless if / else 链（这会破坏编译缓存）。这些都是明确的反模式——详情请参阅操作手册。

Grok 最佳实践

• 结构固定时优先使用 dissect
• 尽可能使用更简单的 grok 模式
• grok 模式始终应以 ^ 和 $ 锚定：

  # Correct — anchored, fails fast on non-matching lines
  patterns:
    - '^%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}$'
  
  # Incorrect — unanchored, scans the whole string for a partial match
  patterns:
    - '%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}'
  

• 避免使用不必要且会导致大量回溯的自定义正则表达式
• 为每个 grok 处理器（以及其他所有处理器）添加 tag

有关 grok 语法（三种表达式形式、内联正则表达式、类型强制转换、pattern_definitions）、syslog 标头拆分方法和常见错误，请参阅 references/grok-recipes.md。

禁止的模式

许多旧版集成中都存在这些模式，但不得在新的或更新的管道中使用。请勿从参考集成中复制这些模式。

绝不要设置 event.ingested

event.ingested 字段由 Elasticsearch 在集成管道之外管理。不要在任何集成管道中为 event.ingested 添加 set 处理器。这包括以下模式：

# PROHIBITED — do not use
- set:
    field: event.ingested
    value: '{{{_ingest.timestamp}}}'

管道应根据原始事件的时间戳设置 @timestamp。当源数据包含多个时间戳时，请按以下方式映射：

• @timestamp：从源数据解析出的主要事件时间戳。此字段为必填项。
• event.created：事件首次由源系统创建或记录的时间（若与 @timestamp 不同）。
• event.start：活动或时段开始的时间（例如，会话开始、连接开始）。
• event.end：活动或时段结束的时间（例如，会话结束、连接关闭）。

如果源时间戳的语义与 event.created、event.start 或 event.end 不符，请将其映射到供应商命名空间下的自定义字段，在 fields.yml 中将其 type 设为 date，并使用带有适当 target_field 的 date 处理器。

绝不要使用 preserve_duplicate_custom_fields

preserve_duplicate_custom_fields 标签模式——使用带有 copy_from 的 set 将源字段复制到 ECS 字段，并有条件地保留原始字段——是一种旧式反模式。不要在任何新的或更新的管道中使用它。 不要添加 preserve_duplicate_custom_fields 清单变量、标签或条件逻辑。

请改为遵循以下字段映射规则：

• 当源字段映射到 ECS 字段时，使用 rename 直接移动该字段。源字段会被移除，因此不会产生重复字段。
• 当需要进行类型转换（例如，将字符串转换为日期或长整型）时，使用适当的处理器（date、convert、带有 copy_from 的 set）填充 ECS 目标字段，然后在管道末尾的清理部分移除源字段。
• 管道的设计绝不能需要同时保留原始供应商字段和 ECS 副本。 ECS 字段是规范位置。

如果在参考集成中遇到此模式，请忽略它——这是旧版模式。

绝不要在末尾添加移除 event.original 的处理器

如上文 JSE00001 部分所述：不要在管道末尾为 event.original 添加 remove 处理器。此操作由单独的最终管道处理。

参考资料

• references/processor-cookbook.md — 处理器选择、解析／规范化／丰富示例、ECS 分类映射模式（模式 A/B/C + 反模式）
• references/branching-patterns.md
• references/error-handling-patterns.md
• references/grok-recipes.md — grok 语法、类型强制转换、syslog 标头配方、常见错误、模式库链接
• references/builder-subagent-guidance.md — 子 Agent 操作手册：范围边界、技能加载顺序、输入数据路径（优先使用 CEL 与直接模式）、9 步管道构建工作流、“审查生成的输出，绝不手动编辑预期 JSON”、报告约定。协调器会在任务提示词中传递此文件的路径来分派子 Agent；子 Agent 会在自己的全新上下文中自行读取该文件。不要将其内容嵌入／粘贴到任务提示词中。
• references/cdr-pipeline-requirements.md — 仅适用于云安全／CDR 集成（cloudsecurity_cdr 类别）：事件分类、必需／应有字段、处理器模式、审查清单
• entity-mappings/references/entity-pipeline-patterns.md — 仅适用于实体／清单数据流（event.kind: asset）：分类处理器、entity.id 镜像、布尔值强制转换、关系对象、反模式、管道审查清单