cel-programs

cel-programs

适用场景

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

• 创建或编辑 cel.yml.hbs Agent 流模板
• 为 cel 输入类型配置数据流清单
• 编写包含分页、游标管理或身份验证的 CEL 程序
• 使用 mito 在本地测试或调试 CEL 程序
• 为基于 CEL 的数据流设置带模拟 API 的系统测试
• 为新的基于 CEL 的数据流试编采集逻辑
• 任何 CEL 或 mito 相关问题，无论上下文如何

不适用场景

以下情况不要将此技能作为主要指南：

• Ingest 管道处理器设计（ingest-pipelines）
• ECS 字段映射（ecs-field-mappings）
• 软件包脚手架（create-integration）
• 使用 Elastic 栈执行系统测试（integration-testing → references/system-testing.md）

强制工作流 — mock → mito → 模板

这不是建议。每个 CEL 程序都必须按此顺序开发。 在使用 mito 针对运行中的 mock 验证 CEL 程序之前，子 Agent 不得编写 cel.yml.hbs。跳过步骤或调整顺序会导致难以调试的故障。

运行 mito 之前，新增 CEL 代码不要超过约 10–15 行。 分阶段增量构建程序（骨架 → 错误处理 → 事件映射 → 分页 → 游标保护），每个阶段后都使用 mito 验证。一次性编写大型程序会导致错误层层叠加，极难调试。请遵循 references/cel-incremental-build.md 中的分阶段方法。

步骤	操作	输出
1. 创建系统测试 mock	在 _dev/deploy/docker/files/config-<stream>.yml 中编写 elastic/stream 配置，其中的规则要匹配所有 API 端点。编写 test-default-config.yml。	Mock 配置文件、docker-compose 服务、测试配置
2. 在本地启动 mock	stream http-server --addr=:8090 --config=...	在 http://localhost:8090 运行的 mock
3. 创建纯 .cel 文件和 state.json	将 CEL 程序写为独立的 .cel 文件。创建 state.json，其中包含与未来 state: 块相同的键，但使用字面测试值，而不是 Handlebars。将 url 指向本地 mock。	/tmp 或工作目录中的 program.cel、state.json
4. 运行 mito 并迭代	按照 references/cel-incremental-build.md 增量构建：阶段 0 骨架 → 阶段 1 错误处理 → 阶段 2 事件 → 阶段 3 分页 → 阶段 4 游标。每个阶段后运行 mito -data state.json -log_requests program.cel。直到 mito 输出正确后才能继续。	已验证的 CEL 程序
5. 仅在此之后编写 cel.yml.hbs	将可用的 CEL 表达式复制到 Handlebars 模板的 program: | 中。将字面测试值替换为 {{var}} 引用。配置清单。	最终集成模板

步骤 3 详解 — 将模板变量转换为 mito 状态： 如果未来的 cel.yml.hbs 中会有类似 api_key: {{api_key}} 和 batch_size: {{batch_size}} 的 state: 块，那么用于 mito 测试的 state.json 应使用相同的键名和字面测试值：

{
  "url": "http://localhost:8090",
  "api_key": "test-key",
  "batch_size": 50,
  "initial_interval": "24h"
}

这与 CEL 输入提供的运行时状态一致。添加 cursor 以测试后续运行行为。

有关 mock 优先工作流、CLI 标志、执行模型和质量标准的完整详情：加载 references/mito-reference.md。

---

cel.yml.hbs 模板结构

data_stream/<stream>/agent/stream/cel.yml.hbs 中的 cel.yml.hbs 文件是一个 Handlebars 模板，用于渲染最终的 CEL 输入配置。它按以下顺序包含这些部分：

interval: {{interval}}
resource.tracer:
  enabled: {{enable_request_tracer}}
  filename: "../../logs/cel/http-request-trace-*.ndjson"
  maxbackups: 5
{{#if proxy_url}}
resource.proxy_url: {{proxy_url}}
{{/if}}
{{#if ssl}}
resource.ssl: {{ssl}}
{{/if}}
{{#if http_client_timeout}}
resource.timeout: {{http_client_timeout}}
{{/if}}
resource.url: <constructed from vars>
state:
  <credentials and pagination config from vars>
redact:
  fields:
    - <sensitive state keys>
max_executions: <number, for heavy pagination>
program: |
  <CEL expression>
tags:
{{#if preserve_original_event}}
  - preserve_original_event
{{/if}}
{{#each tags as |tag|}}
  - {{tag}}
{{/each}}
{{#contains "forwarded" tags}}
publisher_pipeline.disable_host: true
{{/contains}}
{{#if processors}}
processors:
{{processors}}
{{/if}}

Handlebars 模式

模式	用途
{{var_name}}	直接变量替换
{{#if var_name}}...{{/if}}	可选配置的条件块
{{#each tags as |tag|}}	遍历列表变量
{{#contains "forwarded" tags}}	检查列表是否包含指定值

关键模板字段

• resource.url — 基础 URL，通常由多个变量构造（例如，{{url}}/api/v1/endpoint）
• resource.headers（ga 8.18.1）— 每个请求都相同的静态请求头（Content-Type、Accept、API 版本请求头）。如果请求头始终不变，请在此处设置，而不要在程序中设置。此处设置的请求头会在认证请求头之前应用。
• state: — 用于将清单变量注入为 CEL 状态的代码块；凭据和分页设置应放在这里
• redact.fields — 包含敏感信息、需要从调试日志中脱敏的状态键列表
• max_executions — 覆盖默认值 1000，适用于分页量大的集成（例如，5000）
• program: | — CEL 表达式；必须是 YAML 字面块标量

不要在集成包中设置 data_stream.dataset

集成包（type: integration）绝不能在 cel.yml.hbs 中包含 data_stream.dataset，也不能定义 data_stream.dataset 清单变量。框架会自动将文档路由到正确的数据流。设置 data_stream.dataset 会覆盖此路由，导致文档进入错误的索引，通常会在系统测试中出现“0 hits”。

只有输入类型的包（type: input）才会使用 data_stream.dataset，因为它们没有预定义的数据流。

数据流清单配置

数据流的 manifest.yml 定义 CEL 输入流及其变量。

每个 CEL 流都应包含的标准变量

变量	类型	用途
url	text	API 基础 URL
interval	text	轮询间隔（例如，5m）
initial_interval	text	首次运行时的回溯时间范围（例如，24h）
enable_request_tracer	bool	启用 HTTP 请求跟踪
http_client_timeout	text	请求超时时间（例如，30s）
proxy_url	text	HTTP 代理 URL
ssl	yaml	TLS 配置
tags	text (multi)	事件标签
preserve_original_event	bool	保留原始事件
processors	yaml	Beat 处理器

具体的认证变量取决于 API（API 密钥、OAuth client_id/secret/token_url、Bearer 令牌等）。

请在数据流清单中声明 enable_request_tracer，而不是在输入级别声明。在输入级别启用跟踪会为策略中的所有数据流启用日志记录。

包级变量与数据流级变量

• 根目录 manifest.yml 中 policy_templates[].inputs[].vars 下的包级变量：在各个流之间共享（例如，url、认证凭据）
• data_stream/<stream>/manifest.yml 中 streams[].vars 下的数据流级变量：仅适用于特定流（例如，interval、batch_size、initial_interval）

CEL 程序的职责范围

CEL 程序的职责仅限于数据收集：

1. 获取数据：从 API 端点获取数据
2. 处理分页：在单个轮询周期内遍历所有页面
3. 管理游标状态：将时间戳或页面令牌存储在 cursor 中，使下一次轮询能够从上次停止的位置继续，避免重新收集已获取的事件
4. 输出原始事件：为每条记录输出 {"message": e.encode_json()}

CEL 程序不负责处理：

• Elasticsearch 级别的去重 — 如果重叠的时间范围导致收集到少量重复事件，这是可以接受的。去重由摄取管道或 Elasticsearch _id 路由在索引时处理，而不是由 CEL 程序处理。
• 字段映射或转换 — 摄取管道负责解析、ECS 映射和丰富数据。
• 按内容筛选 — 除非 API 支持服务端筛选参数，否则不要在 CEL 程序中筛选事件。应输出所有事件，由管道决定如何处理。

不要在代码库中搜索 _id、document_id 或去重模式。这些不属于 CEL 的职责范围。

CEL 程序结构模式

分页策略选择

API 行为	模式	关键指标
返回总数 + 支持偏移量	偏移分页	请求/响应中的 total_count、offset、limit
返回指定时间戳之后的记录	时间戳游标	时间范围参数，没有明确的页面令牌
返回包含下一个 URL 的 Link 标头	Link 标头	响应标头中的 Link: <url>; rel="next"
在响应正文中返回下一页 URL	下一页 URL	JSON 中的 next、nextLink、@odata.nextLink 字段
带有 pageInfo 的 GraphQL	GraphQL 游标	pageInfo 对象中的 hasNextPage、endCursor
多阶段订阅/内容流程	多步状态机	状态中包含多个 API 调用和工作队列

游标时间戳选择——当 API 按升序排序时，使用最后一条记录的时间戳；按降序排序时，使用第一条；无法保证排序顺序时，使用 max() 并添加回退保护。

每种模式的完整代码、包引用和 YAML 片段：references/cel-pagination-patterns.md。

身份验证模式

三种策略：标头（凭据放在 state: 中，通过 Header 映射传递）、查询参数（通过 .format_query() 将凭据追加到 URL）、签名查询（在 CEL 中计算 HMAC 签名）。配置级别的 auth.oauth2/auth.digest/auth.aws 会应用于所有请求，包括 .do_request()；auth.basic/auth.token 仅应用于直接调用（get()、post()）。优先使用输入级别的身份验证，而不是在程序中获取令牌。

完整代码示例、可选标头语法以及配置级别身份验证的作用范围详情：加载 references/cel-auth-patterns.md。

状态管理规则

1. state.url 从 resource.url 配置填充；必须在输出中保留，或硬编码
2. cursor 是输入重启后唯一会持久化的状态；在此处存储分页位置和时间戳
3. events 数组会在每次评估后移除；切勿在后续运行中依赖它
4. want_more: true 会触发立即重新评估，但前提是 events 非空。**分页续传防护：**只要存在下一页游标／令牌，就始终设置 want_more: true，无论当前页收集到多少事件。将 want_more 与 size(events) > 0 绑定会导致分页悄然停滞——下一个游标有效，输出空的 events 数组也是安全的。正确写法是 "want_more": next_cursor != ""。
5. 所有其他状态键会在会话期间保留，但重启后会丢失——使用 state.with() 自动传递它们
6. 数字在状态 JSON 中会序列化为浮点数；用 int() 转换为整数后再使用
7. 使用 state.?cursor.last_timestamp.orValue(default) 进行可选访问，可避免游标缺失时出错
8. 密钥——state 中的每个敏感字段都必须有对应的 redact 条目。state.secret 始终会自动脱敏。若可使用 secret_state（v8.19.14 / v9.2.8 / v9.3.3 / v9.4.0），请优先使用它。
9. 更新游标必须发布事件——只有至少发布一个事件时，输入才会持久化游标更新。如果程序更新了游标，但返回零个事件，游标更改会丢失。
10. 不要在不同分支中重复处理请求／响应——当初始化分支（创建游标、订阅、令牌交换）和稳定运行分支都需要相同的获取逻辑时，应将其整合。有两种方法：通过 want_more: true 将初始化拆分到单独的一次评估中（技巧 6 变体 A），或使用中间结果映射在一次评估中统一各分支（变体 B）。两种方法都有效——参见 references/cel-code-style.md 中的技巧 6，以及 references/cel-pagination-patterns.md 中的先初始化后稳定运行模式。
11. 嵌套深度——任何执行路径上的 .as() 链深度不得超过 5 层。HTTP 程序在 state.with() 内的嵌套层级必须控制在 2 层（resp + body）。游标默认值、窗口边界和页面令牌必须在 state.with() 之前提取为预绑定变量。仅使用一次的值（例如 int(state.batch_size)）必须内联在调用位置，不要用 .as() 包裹。请加载 references/cel-code-style.md，了解扁平化技巧及修改前后的示例。

映射合并与字段移除

with()、with_replace()、with_update() 和 drop() 是通用映射操作——可用于任何映射，而不仅是状态或游标。with() 执行浅合并：嵌套对象会被整体替换。因此，它既可用于游标状态转换（通过覆盖移除子对象），也可用于构建请求标头、转换响应数据和构造中间映射。完整语义和示例：references/cel-code-style.md。

事件输出格式

事件只能包含 "message"——{"message": e.encode_json()}。不要设置 @timestamp 或任何其他字段；框架会添加 @timestamp，重复字段会导致 ES 9.x 悄然拒收文档。正确和错误示例请参见 references/cel-idioms.md。

响应处理

resp.Body.decode_json()——旧版运行时需要使用 bytes(resp.Body) 包装，但现在已不再需要。直接使用 resp.Body.decode_json()。

错误处理

每个 CEL 程序都必须处理 HTTP 错误。有两种形式：

• 单对象错误（重试）： "events": {"error": {...}} — 记录 ERROR 级别日志，设置降级状态，删除游标，以便下次评估时重试。数据未能收集时使用。
• 数组错误（推进）： "events": [{"error": {...}}] — 会更新游标。程序应越过错误继续推进时使用。摄取管道中需要有 terminate 处理器（ES 8.16.0+）。

错误消息格式："METHOD path: body-or-status"。代码示例：references/cel-idioms.md。

占位事件

在没有真实事件但需要推进游标时，发出占位事件（[{"retry": true}]），并在 processors: 部分添加 - drop_event.when.equals.retry: true 条目，以便在索引前将其丢弃。完整模式和替代方案：references/cel-idioms.md。

速率限制与重试

不要在 CEL 程序中实现速率限制或重试逻辑。 不要调用 rate_limit()，不要传播 "rate_limit" 状态，不要编写针对 429 的分支，也不要添加重试循环。这些做法会大幅增加嵌套层级和复杂度，收益却微乎其微。

当 API 有记录在案的速率限制时，请在 cel.yml.hbs 中仅使用 YAML 配置：

resource.rate_limit.limit: 10   # max requests per second
resource.rate_limit.burst: 5    # max burst above sustained rate

需要自定义重试行为时，请仅使用 YAML 配置：

resource.retry.max_attempts: 5    # default: 5
resource.retry.wait_min: 1s       # default: 1s
resource.retry.wait_max: 60s      # default: 60s

输入框架会透明地处理这两种情况。有关何时添加这些设置的指导，请参阅 references/cel-rate-limiting.md。

类型安全

避免使用 dyn() — 它会破坏类型检查。实际中很少需要使用。

所有数字都是 float64 — CEL 输入会将所有数字以 float64 传输。在 Elasticsearch 中，大于或等于 1e7 的数字会以科学计数法呈现。请在 CEL 程序中或通过摄取管道将预期为整数的字段转换为字符串。安全整数范围：[-(2^53 - 1), 2^53 - 1]。

调试辅助工具

• debug(tag, value) — 以 DEBUG 级别记录到 cel_debug。
• try(expr) / is_error(value) — 以结构化方式处理错误，避免程序崩溃。
• failure_dump（ga 8.18.0）— 失败时转储完整评估状态。注意：转储内容可能包含机密。
• remaining_executions（ga 9.2）— max_executions 预算中还剩余的评估次数。

---

Mito CLI

Mito（github.com/elastic/mito）是本地 CEL 评估 CLI。未经 mito 测试的 CEL 程序不可接受。 请遵循本技能顶部的强制工作流：mock → mito → template。程序通过 mito 验证之前，不要编写 cel.yml.hbs。

有关安装、CLI 标志、输入状态结构、执行模型、完整的 mock 优先工作流步骤、mito→集成映射和质量标准，请加载 references/mito-reference.md。

---

数据匿名化

提交到仓库的所有数据都必须经过完全匿名化处理。 这适用于清单 vars 中的默认值（使用 https://api.example.com）、CEL 状态中的示例值、模拟 API 响应、CEL 输出的管道测试夹具，以及 mito 原型开发期间捕获的示例负载。

完整的匿名化政策和占位符约定，请参阅 anonymize-logs 技能。

移交给其他技能

• integration-testing → references/system-testing.md，用于运行系统测试（mock API 已在 CEL 开发过程中准备就绪）
• integration-testing → references/pipeline-testing.md，用于验证 CEL 生成事件的摄取管道行为
• create-integration 技能，用于了解软件包整体布局

参考文件

重要： 这些参考文件包含实际可用的代码示例和模式。以上摘要不足以编写正确的 CEL 程序——编写代码前必须加载相关参考文件。

编写 CEL 程序时，始终按以下顺序加载这五个文件（先 mock/mito，再模板）：

File	Contains	Load order
references/cel-system-tests.md	使用 elastic/stream 的模拟 API 设置、docker-compose 配置、规则格式、变量捕获模式、GraphQL 模拟示例、hit_count 计算，以及 0 次命中故障的调试方法	第 1 项——编写任何 CEL 之前都需要阅读
references/cel-incremental-build.md	强制分阶段构建流程（骨架 → 错误处理 → 事件 → 分页 → 游标）、会导致编译失败的语法反模式（bytes()、parse_time()、元组、括号不匹配），以及调试指南	第 2 项——必须遵循此分阶段方法；在验证骨架之前不要编写完整程序
references/mito-reference.md	Mito CLI 标志、输入状态结构、模拟优先工作流、将模板变量转换为 state.json、扩展库速查、语法陷阱、testscript 测试框架	第 3 项——开发和验证程序时需要阅读
references/cel-template-examples.md	完整可用的 cel.yml.hbs 示例（最简 GET、带时间戳游标的分页、OAuth、GraphQL 游标）及对应的清单配置——这些是最终输出；mito 通过之前不要编写模板	第 4 项——仅在工作流第 5 步需要
references/cel-code-style.md	嵌套规范：HTTP 核心的三级规则、六种扁平化/结构化技巧（包括用于共享逻辑的中间结果映射）、浅合并语义、使用 clobber 的游标命名空间、合并策略（with/with_replace/with_update）、drop()，以及结构良好的参考集成链接——编写任何多行 CEL 之前必须阅读	第 5 项——编写 CEL 程序前阅读，确保从一开始就采用正确的结构

根据任务加载以下内容：

文件	加载时机
references/cel-pagination-patterns.md	编写任何分页逻辑时——包含全部 6 种模式及代码
references/cel-auth-patterns.md	实现身份验证时——涵盖标头、查询参数、签名和配置级身份验证模式
references/cel-rate-limiting.md	速率限制策略——仅使用配置的方法，以及何时添加 resource.rate_limit.* 和 resource.retry.* 设置
references/cel-idioms.md	常见惯用写法、HTTP 请求模式和结构约定的快速参考
references/cel-polymorphic-patterns.md	为身份验证、标头和速率限制选择纯 CEL、mito 库或配置方案——标注版本
references/cel-expression.md	表达式专用参考：接口契约、转换框架（Python→CEL）、增量构建阶段、核心结构、事件输出、错误处理、分页、状态管理、语法规则、质量检查清单
references/cel-taxonomy.md	分类法：分页和状态管理类别、最低复杂度原则、与技能词汇的映射方式，以及如何根据 test-api.py 分类
references/cel-complexity-baselines.md	基于对 316 个程序的 ceplx 调查得出的各模式类别复杂度基线、跳过阈值、审查者质疑示例及诊断解读
references/expression-builder-subagent-guidance.md	cel-expression-builder 子 Agent 操作手册：将 test-api.py 转换为经过验证的 .cel 文件并进行分类。不处理模板、清单或模拟对象。
references/reviewer-subagent-guidance.md	cel-expression-reviewer 子 Agent 操作手册：根据复杂度基线和源代码保真度检查生成的 CEL，提出具体质疑或予以接受
references/cel-function-reference.md	查询各扩展中可用的 CEL 函数及其首次出现的 mito 版本
references/builder-subagent-guidance.md	cel-program-builder 编排器的子 Agent 操作手册：范围边界、技能加载顺序、包含模拟完整性门槛的 9 步“模拟优先 / mito 增量式”工作流、委派给 cel-expression-builder 的方式及报告约定。编排器通过在任务提示词中传递此文件的路径来调度子 Agent；子 Agent 会在自己的全新上下文中自行读取该文件。请勿将其内容嵌入或粘贴到任务提示词中。

另请参阅：CEL 输入文档 · Mito 库文档 · Mito 仓库 · CEL 语言规范