用CEL输入接管API数据采集：从脚本到配置的Elastic Agent集成实践

前阵子有个同事问我：一个Elastic Agent集成里，最让人头疼的部分是哪个？我几乎没犹豫就回答——不是Agent本身，也不是日志文件的采集，而是那些为了对接HTTP API而写的自定义脚本。日志类采集只要定义好消息模式，剩下的就是按路径读文件；API类数据源才是真正的复杂度集中地：认证、分页、限流、增量游标、多层嵌套JSON清洗，每一项都够写一篇长文。后来团队把这类集成逐步切到CEL输入（Common Expression Language input），用表达式替代脚本，用配置描述数据收集逻辑，整个维护体验直接上了一个台阶。这篇文章就把我在这条路上看到的改进点、踩过的坑、现在日常使用的方式一起整理出来。

1. 从“采集器+脚本”到“一段配置”：API型数据源才是真正的战场

1.1 先聊清楚：Agent统一了采集器，但并没有解决数据源的“脾气”

早年间一个采集节点上可能同时跑着好几个采集器，每个采集器有各自的守护进程、权限模型、日志输出和升级周期。Elastic Agent这类统一代理出现之后，最直观的好处是：部署一个Agent，通过策略下发就能弹性伸缩采集能力。集成（Integration）机制又把数据源连接逻辑做成了可复用包，表面上看起来“装个集成就能采集”。

但这里有一个很容易被忽略的事实：统一Agent解决的是进程编排和配置管理问题，不等于每个数据源都是“开箱即吃”的。尤其是当数据源是一个暴露在公网上的REST API时，集成的真正工作量全都藏在API交互细节里。恰恰是这一层，过去最耗费人力，现在也最需要仔细设计。

1.2 API型数据源绕不开的六件套

我自己总结，接一个HTTP API做数据收集，至少要处理六件事：

• 认证：固定Token还算简单，OAuth流程就麻烦了，Token会过期，过期后要刷新，刷新失败还要触发重新登录。
• 分页：Offset分页、Cursor分页、时间滚动分页，不同API的翻页语义完全不同。
• 限流：每分钟多少次请求、429返回后要不要退避、退避多久，这些参数不写在代码里就会出事故。
• 增量游标：上一次同步到了哪个时间点、哪条记录ID，是增量采集的核心。
• 数据清洗：字段名映射、类型转换、字段删除、嵌套展平，做不好后面写索引映射全是坑。
• 错误恢复：网络超时、服务端5xx、数据半途断裂，采集进程要能自己站起来，而不是等人手动重启。

日志采集面对的是一个文件系统，有确定的位置和确定的读取进度；API采集面对的是一个外部系统的业务逻辑，上述六件事往往会扭在一起。以前处理这些问题的标准答案，就是写一个自定义脚本，然后用cron去跑。

1.3 自定义脚本真正的问题不是“不好写”，而是“养不起”

不是说脚本方案跑不通。一个Python脚本拉数据，按道理也能实现分页和游标。但把它放到生产环境里长期养，问题就来了：

• 依赖膨胀：requests、pandas、各种SDK，每个采集任务一套依赖，环境一换就抛异常。
• 权限失控：脚本要满足数据收集逻辑，往往要跑在较高权限的账户下，安全评估时很难说清这段代码到底做了什么。
• 监控断档：脚本自己打印日志，没人看它就是没日志；进程卡死、内存泄漏，也不会自动出现在统一监控面板里。
• 变更成本高：业务字段一变，要发一次代码版本，还要测试、走发布流程，迭代速度远跟不上数据源的变化。

所以当我看到CEL输入能够把“数据处理和同步决策”直接做成声明式表达式，放进Agent集成配置里，第一反应是：这条路终于对“API采集”这个场景下手了。它把最重的数据逻辑从“开发代码”降维成“写配置”，而且这个配置本身是可审计、可重试、可版本管理的。

2. CEL的执行模型，是我敢把逻辑写进YAML的底气

2.1 CEL不是脚本语言，它是一次性求值的表达式

通用表达式语言（Common Expression Language）最早是为了处理云平台中授权与策略描述而设计的，设计目标很明确：嵌入宿主程序，提供一种轻量、类型安全、无副作用的表达式求值能力。注意这里的用词——表达式，不是脚本。

一个CEL表达式是这样工作的：你给它一个输入对象，它做条件判断、字段投影、字符串处理、集合运算，最后产出一个新对象。整个过程没有变量赋值语句，没有for循环，没有副作用，不能修改外部状态。

举例来说，下面这段就是合法的CEL表达式：

ticket.status == "closed" && has(ticket.assignee) && timestamp(ticket.updated_at) > timestamp(state.since)

它表达的是“这张工单已关闭、有处理人、更新时间比游标新”。逻辑清楚，边界确定。

相比之下，脚本语言给你的是流程控制、随机访问、环境IO——这些能力非常强大，但一旦放进配置里，意味着每一个跑脚本的节点都在执行一段几乎无法静态审查的代码。而CEL表达式的执行范围是受限的，强类型检查在解析阶段就能拦掉一批低级错误，运行时则是确定的、不依赖外部环境的纯计算。

2.2 无副作用的真正价值：可重试、可检查、可审计

数据收集这种场景，最怕的就是“这次拉了一半，下次不知道从哪里续上”。脚本方案里为什么游标经常出问题？因为脚本在一次运行里混进了太多职责：请求外部API是副作用，解析响应是计算，写入本地游标文件又是副作用。三条线搅在一起，一旦中途崩溃，你很难判断哪一步完成了。

CEL输入把这一坨拆开了：外部请求由Agent的运行时扩展处理，请求一旦返回，响应体就是一块“只读数据”；接下来的过滤、清洗、游标更新都是纯表达式计算。纯计算意味着同一份数据你就算一百遍结果也一样，也意味着失败后可以带着同样的上下文安全地重新执行。

这一点对生产环境的意义很大。表达式执行的每一步都有迹可循：输入是什么、过滤条件是什么、输出是什么、游标从多少变成了多少，全都可以描述出来。数据出问题时，你不是面对一个黑盒脚本，而是面对一段可直接审查的表达式。

2.3 它和JQ、和Python脚本的边界在哪里

我刚接触时也想过：这不就是JQ吗？后来在项目里把三者的边界捋清楚了。

实践中我的用法是：JQ用于人工排查数据时快速看结构；CEL用于集成里需要反复执行的同步决策；自定义脚本只在逻辑复杂到“单轮表达式描述不了”时才考虑，而且会尽量把脚本包裹成Agent可监控的子进程，而不是裸跑在cron里。

3. 一次真实的数据流重构：工单API从请求到索引的CEL实现

3.1 业务场景与原始API结构

为了讲清楚CEL输入在实际集成里怎么起作用，我拿一个典型场景做拆解：一个项目工单系统的REST API，地址是https://api.ticket-demo.example/v1/tickets。需求是每5分钟增量拉取一次工单，入库前要完成三件事：

• 只保留更新时间晚于上次采集游标的工单；
• 把嵌套字段展平，URL里的对象拆成平铺字段；
• 移除邮箱、手机号之类的敏感字段，并统一时间格式。

接口返回结构大概是这样的：

{ "pagination": { "has_more": true, "next_token": "page-token-2" }, "tickets": [ { "id": 123, "title": "订单同步失败", "updated_at": "2026-03-18T12:30:11Z", "user": { "name": "张三", "email": "zhangsan@example.com" }, "meta": { "priority": "high", "source": "web" } } ] }

这类结构在真实API里非常常见，拉了数据不等于能用数据，真正的麻烦在后半程。

3.2 集成配置骨架长什么样

在CEL输入方案里，这个任务的配置视图大致如下。注意：我把HTTP请求原语简化命名为http_get，是为了方便说明数据流；不同版本的CEL集成运行时对请求功能的封装命名会略有差异，但整体结构是一致的。

- type: cel name: ticket-sync interval: 5m vars: endpoint: "https://api.ticket-demo.example/v1/tickets" token: "${TICKET_API_TOKEN}" program: | http_get(state.endpoint) .withQuery({ "since": state.since, "limit": 200, "page_token": state.next_token }) .withHeader("Authorization", "Bearer " + state.token) .doRequest() .as(resp, { "events": clean_tickets(resp.body.tickets), "state": update_state(state, resp), "want_more": resp.body.pagination.has_more })

这个配置里的核心是program字段。CEL表达式负责三件事：拼请求参数、把响应体里的工单数组清洗成目标文档、根据响应内容更新下一轮要用的状态。events是交给Agent后续处理器的数据数组，state是Agent会持久化并在下一轮传入的新状态。

很多集成模板的CEL脚本看起来比我这个复杂，是因为它们会把认证Token刷新、错误码判断、特殊字段处理全部揉进一个程序里。基本结构就是上面这个样子：请求、转换、状态更新，三个职责，用表达式组织在一起。

3.3 关键表达式一步一步长出来的过程

我写CEL采集表达式很少一步到位，通常是先在本地把逻辑分层，再逐段拼起来。拿上面的clean_tickets来说，第一步只做过滤：

resp.body.tickets .filter(t, timestamp(t.updated_at) > timestamp(state.since))

这一行表达的意思很清楚：“只要更新时间比上次游标晚的”。timestamp()是把API返回的时间字符串解析成可比较的时间对象。

第二步做字段展平和映射：

resp.body.tickets .filter(t, has(t.id) && has(t.updated_at)) .map(t, { "id": string(t.id), "title": t.title, "updated": t.updated_at, "reporter": t.user.name, "priority": t.meta.priority, "source": "ticket-api" })

注意CEL的强类型特性：id在JSON里可能是数字也可能被解析成字符串格式，这里显式用string(t.id)做一次归一化，后面索引映射就不会因为类型漂移报错。has(t.id)则是一个很实用的保护性判断，避免某条记录缺字段导致整个转换表达式抛异常。

第三步加上脱敏：

resp.body.tickets .filter(t, has(t.id) && has(t.updated_at)) .map(t, { "id": string(t.id), "title": t.title, "updated": t.updated_at, "reporter": t.user.name, "priority": t.meta.priority, "source": "ticket-api" }) .map(d, d.with({ "safe_title": d.title.size() > 0 ? d.title : "untitled" }))

真实落地时脱敏逻辑会复杂一点，比如判断email字段是否满足格式、打码后是否仍能用于分析。这里想表达的重点是：CEL支持方法链和匿名函数，你在脚本里习惯写的map/filter思路，在表达式里同样成立，只是它没有for和while，所有循环语义都收敛在集合操作里。

3.4 数据从CEL到可检索索引的完整路径

很多人第一次看CEL输入，会误以为表达式的结果会直接写入目标数据存储。实际上不是，events只是“交给Agent后续处理器”的入口，后面通常还要经过解码、字段映射、标准事件处理，最后才落到索引里。

也就是说，CEL输入负责的是“把不可控的外部API形态，收敛成内部标准事件形态”。这一步做好了，后续处理器的工作就会非常轻松。事实也证明，这个“入口标准化”的价值比想象中更大：同一个内部标准结构可以被日志、指标、安全告警等多条消费链路复用，CEl输入实际上是站在了数据管道的最前端。

4. state里的“记忆”：增量采集的核心机制和三个大坑

4.1 为什么API增量采集必须背着“记忆”

全量拉取谁都会：每5分钟把整个API数据拖一遍，清洗后覆盖写入。问题是绝大多数真实API都扛不住高频全量拉取，数据量一大，限流和资源开销都会爆炸。增量采集的核心在于——你要让采集程序记住“上次已经处理到哪了”。

这个“上次的位置”就是游标。在CEL输入里，游标以及其他需要在轮询之间保留的上下文，统一放在state字段里。

4.2 state在CEL输入里的生命周期

state的生命周期分成四步：

• 初始化：节点首次运行时，从配置里的初始值构造state，比如把since设为很久以前的零点，保证首次能拉到足够多的数据补数。
• 传入：每一轮定时任务执行时，Agent把上一轮持久化的state塞进表达式的state变量，让表达式在最新上下文里工作。
• 更新：表达式执行过程中读取响应体，计算出一份全新的state，比如把最新的更新时间写入since。
• 存储：运行时的收尾工作就是持久化新state，下一轮开始前再加载。

这个流程和“人看书时夹书签”完全是一个道理：每读一页就更新书签位置，下次从书签接着读。

4.3 多页拉取与want_more：一轮任务里的内部循环

一个常见的误解是：CEL输入的一个定时周期只能发出一次HTTP请求。实际上，分页这种“一轮任务内部要连续请求多页”的场景，是靠want_more这类信号解决的。

当表达式返回的want_more为真，Agent会立刻用更新后的state重新执行当前轮的任务，而不是干等到下一个定时周期。这相当于在每5分钟的窗口里，Agent内部可以连续吞掉多个分页，直到最后一页返回空或has_more为假，再进入休眠等待下一轮。

这个设计把两个时间维度分得非常清楚：

• 批内突发：一次定时触发后，为追平游标连续请求十几页，属于“补数据”，走want_more快速循环；
• 周期增量：批内数据追平后，等下一个5分钟再做新一轮检查，属于“正常增量”，走interval调度。

在我曾经用脚本实现同样逻辑时，这两件事混合在一个进程里，代码里到处是while has_more和time.sleep混着写，维护起来相当痛苦。

4.4 我踩过的三个状态坑

第一个坑：把响应体塞进state。刚开始图省事，想把原始响应体直接存进state，下一轮判断用。结果state越来越大，Agent持久化负担明显上升，而且响应体里经常有大段不需要的文本，白白占资源。正确做法是state只保留能重建请求的最小信息量：游标、下一次分页Token、Token过期时间，最多加一个上次处理的时间戳。

第二个坑：游标用本地时钟计算。如果直接在表达式里写“当前时间减去一分钟作为since”，就会埋下一个隐患：Agent的重试、节点时间漂移、时区不一致，都会让游标倒着走，导致同一条数据被重复采集。稳妥的做法是以API返回的数据时间戳为准，让游标始终跟着业务数据走，而不是跟着机器时钟走。

第三个坑：空列表时不更新游标。API按时间过滤时，如果这一轮没有新数据，返回的tickets列表是空的。如果表达式在空列表时直接返回原样state，下一次还是同一个since，就会反复拉取同一段空窗口。虽然不会造成数据重复，但会白白浪费请求次数。正确逻辑是：即便列表为空，也要把当前请求时间或最大时间戳写进state，表示“已经追上这个点了”。

5. 出问题时我是怎么排查的，以及不建议用CEL的场景

5.1 让表达式先在本地跑起来，别直接在集成里盲调

CEL表达式的调试，最难的不是语法，而是你得知道“表达式在这个输入下到底求值成了什么”。我的习惯是：先构造一份最小化的JSON样例，模拟API返回体，在本地把表达式跑通，再丢进Agent集成里验证。

做法是把你写的核心表达式拆出来，给它一个假的输入变量，比如：

vars: fake_resp: '{"tickets":[...]}' program: | clean_tickets(fake_resp.tickets)

这样你能快速看到输出结构是否符合预期。表达式的纯计算特性在这里很占便宜——不用真的发起HTTP请求，也能验证整个转换逻辑。等这一步稳定了，再接上运行时请求原语，问题就能被快速定位在“请求层”还是“转换层”。

5.2 需要调试输出时，把它临时塞进events

CEL表达式本身不是日志系统，纯计算函数里你想“打印一行”并不直观。我调试时的土办法是：临时把一个调试字段塞进events，让后续处理器把它当成一条日志事件输出，看完再把这个字段删掉。

比如我怀疑某个字段解析结果不对，就先在表达式里加一句：

"debug_candidates": resp.body.tickets.map(t, t.updated_at)

把它混进返回结构里，跟着数据流走到调试日志里观察。体量不大时这个方法很好用，确认没问题后删掉即可。比一遍遍改配置、看采集状态要快得多。

5.3 限流、重试和认证过期：这些“旁边问题”同样会反噬

表达式写得再对，HTTP请求本身出问题，采集一样会断。按经验，有三类“旁边问题”容易忽略：

• 429限流：CEL输入天然适合声明式重试与退避，但你要在表达式里把退避逻辑写清楚。无脑在同一时刻重试，只会加重服务端压力。
• Token过期：状态里最好显式保存Token的获取时间和有效期。表达式判断接近过期时，先走刷新流程，再发起业务请求，而不是等请求返回401再补救。
• interval设置：定时周期不能短于API的限流窗口。比如API限制每分钟60次请求，而你设置每10秒跑一轮加内部多页循环，必然触发限流。我的经验是周期起步设5分钟，等确认稳定再适度缩短。

5.4 什么情况我不建议硬上CEL

CEL输入确实解决了大部分API采集问题，但它不是银弹。下面几类场景，我还是会回归自定义程序：

• 多阶段状态机：需要根据历史状态在不同处理流程之间切换，而且状态分支多、转移条件复杂；
• 依赖本地库或系统命令：比如要做OCR、图片处理、机器学习推理，表达式语言无法承载；
• 跨节点协调：两个采集节点不能重复处理同一条数据，需要分布式锁或共享游标，这也不是单节点CEL表达式能解决的。

我对团队定的原则是：凡是一次请求加有限分页加确定性清洗能描述的，优先用CEL输入；超出这个边界，才启动自定义采集器，而且自定义采集器必须纳入Agent统一监控，不能因为“是自定义的”就游离在体系之外。

我在实际使用中最明显的一点体会

项目跑久了，最直观的变化不是“少写了几百行代码”，而是排查问题的路径变了。以前API采集出问题，要翻脚本日志、查游标文件、看cron执行记录，定位链路很长。现在打开配置，读一遍表达式，基本就知道数据处理逻辑在哪个环节出了问题。表达式本身成了最可靠的文档，这句话是真真切切感受到的效率提升。

如果让我给一个最实用的建议：写CEL表达式时不要追求“一行流”。把分页逻辑、清洗逻辑、状态更新逻辑拆成语义明确的函数，哪怕代码长一点也没关系。后面排障时，你会感谢自己当初没有把所有逻辑拧巴成一团。