1. 本地 ES 9.x 里 Filebeat CEL Input 到底解决什么问题如果你在本地跑了一套 Elasticsearch 9.x想把一个自研 HTTP 接口、一个定时导出的 CSV、或者一个内部监控系统的返回结果接进来传统做法通常是写 Logstash 插件、写 Python 脚本定时推、或者干脆用 curl 加 cron 拼。这些方案都能跑但维护起来各有各的麻烦脚本要自己管重试、管断点、管认证Logstash 又偏重。Filebeat 的 CEL Input 就是为这类场景准备的。CEL 全称 Common Expression Language是一种非图灵完备的表达式语言没有循环、没有递归所以单次执行时间可控不会因为写错逻辑把本地机器 CPU 吃满。Filebeat 内置了一个轻量 CEL 运行时按你配置的 interval 周期性执行你写的 CEL 程序程序里可以通过扩展库去读本地文件、发 HTTP 请求、解析 JSON/XML/CSV然后把结果转成 Filebeat 事件发到 Elasticsearch。它适合谁我总结下来是三类人一是在本地做 ES 9.x 验证、需要快速接入自定义数据源的开发和运维二是内网环境里有一堆零散 HTTP 接口、不想为每个接口写采集脚本的人三是想把采集逻辑收敛到 Filebeat 一套配置里、少维护一个组件的团队。ES 9.x 在 8.6 的 CEL 能力上又补了动态令牌文件、时间截断函数、失败请求 dump 这些本地部署很实用的东西所以现在用 9.x 本地跑 CEL Input体验比早几年顺不少。这篇就按“本地部署 能直接复制跑通”的思路来先讲清楚 CEL Input 在 ES 9.x 下的关键配置项再给一份完整的 filebeat.yml然后写 CEL 程序、启动、用 curl 验证采集结果最后把几个高频报错挨个拆开。你跟着做本地环境基本能一次跑通。2. TaoToken 前置准备本地联调大模型接口时的 Key 与地址CEL Input 本身不依赖任何大模型服务它采集的是你自己的数据源。但本地做 ES 9.x 验证时很多人会顺手把采集到的日志或结构化数据丢给模型做摘要、分类、异常判断这时候就需要一个稳定的模型调用入口。我本地联调时用的是 TaoToken它的 API 地址是https://taotoken.net/api官网在https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content模型对话、Coding Plan、控制台、API Keys、接入文档这些入口都在官网能找到。先把 Key 拿到手后面 CEL 程序里如果要调模型接口直接引用环境变量就行。操作路径是进控制台创建 API Key然后到接入文档确认 Base URL 和 Model ID 的写法。本地部署场景下我建议把 Key 放在环境变量或者一个权限 0600 的文件里不要写死在 filebeat.yml 里原因后面排障章节会讲。具体三件套这样配Base URLhttps://taotoken.net/apiAPI Key控制台生成的 Key本地放环境变量TAOTOKEN_API_KEYModel ID按接入文档里当前可用的模型名填比如对话类模型如果你只是验证 CEL Input 采集链路不调模型这一步可以跳过直接进第 3 节。但如果你打算在 CEL 程序里对采集到的数据做二次处理再上报建议先把 Key 准备好后面配置里我会留出引用位置。API Keys 页面在https://taotoken.net/api-keys接入文档在https://taotoken.net/doc模型对话入口在https://taotoken.net/chat长期跑编码或 Agent 任务可以看 Coding Planhttps://taotoken.net/coding-plan。3. 可复制配置filebeat.yml 与 CEL 程序完整片段这一节是核心给一份能在本地 ES 9.x 下直接跑的配置。场景设定为本地有一个 HTTP 接口http://127.0.0.1:8080/api/v1/logs返回 JSON 数组需要每分钟采集一次带 Bearer Token 认证采集结果写入本地 ES 9.x。先看 filebeat.yml 的完整片段filebeat.inputs: - type: cel enabled: true interval: 1m resource.url: http://127.0.0.1:8080/api/v1/logs resource.timeout: 10s resource.retry: max_attempts: 3 wait_min: 500ms wait_max: 5s resource.keep_alive: disable: false max_idle_connections_per_host: 5 auth.token: enabled: true type: Bearer value: ${LOCAL_API_TOKEN} program: | get(state.url).Body.as(body, { events: body.decode_json(), cursor: {last_id: body.decode_json()[0].id}, want_more: false }) allowed_environment: [LOCAL_API_TOKEN] redact: fields: [token, password] delete: true failure_dump: enabled: false filename: /var/log/filebeat/cel-failures/ tags: [local-api, cel] fields: service: local-app fields_under_root: true keep_null: false output.elasticsearch: hosts: [http://localhost:9200] username: ${ES_USER} password: ${ES_PASS} index: local-api-logs-%{yyyy.MM.dd} logging.level: info logging.to_files: true logging.files: path: /var/log/filebeat name: filebeat rotateeverybytes: 10485760 keepfiles: 7几个关键点解释一下。resource.url用127.0.0.1而不是localhost避免本地 DNS 解析带来的额外延迟。auth.token.value引用环境变量${LOCAL_API_TOKEN}配合allowed_environment只放行这一个变量防止 Filebeat 读到系统里其他敏感环境变量。redact里delete: true是直接删字段比替换成星号更干净。failure_dump本地调试时可以临时开成true把失败请求和响应落到/var/log/filebeat/cel-failures/定位完问题再关掉。CEL 程序部分get(state.url)调用 HTTP 扩展库发请求Body拿响应体decode_json()把 JSON 字符串转成 CEL 对象。events字段就是最终上报的事件列表cursor用来记录断点位置want_more: false表示单次 interval 内只执行一轮。如果你要采集的是本地 CSV 文件把resource.url换成file:///var/log/app/app.csvCEL 程序改成program: | get(state.url).Body.decode_mime(text/csv; headerpresent).as(rows, { events: rows, cursor: {last_row: len(rows)}, want_more: false })decode_mime(text/csv; headerpresent)会按 CSV 解析第一行当表头每行转成一个对象。这里 cursor 记的是行数实际生产里要结合文件偏移量做更精确的断点本地验证够用。如果你要在 CEL 程序里调模型接口做二次处理可以在 program 里再发一次请求把https://taotoken.net/api作为目标地址Key 从环境变量读。不过要注意CEL 是非图灵完备的复杂逻辑建议拆成多步或者放到 Filebeat 的 processors 里做。4. 验证请求启动 Filebeat 并用 curl 确认采集结果配置写好后先别急着启动用 mito 工具预校验 CEL 程序。mito 是 Elastic 提供的 CEL 测试工具本地装一下go install github.com/elastic/mito/cmd/mitolatest然后写一个测试文件test.cel把 program 内容贴进去用 mito 跑mito -data {url: http://127.0.0.1:8080/api/v1/logs} test.cel如果 CEL 语法有问题这一步就会报出来不用等 Filebeat 启动。确认没问题后启动 Filebeatsudo filebeat -e -c /etc/filebeat/filebeat.yml-e表示日志输出到 stderr方便本地看。启动后观察日志正常会看到celinput 启动、Harvester started之类的信息。等一个 interval这里配的 1m然后去 ES 里查curl -X GET http://localhost:9200/local-api-logs-*/_search?pretty -H Content-Type: application/json -d { query: {match_all: {}}, size: 3 }如果返回的hits.hits里有你接口的数据说明采集链路通了。再确认一下 cursor 是否生效重启 Filebeat观察第二次采集是不是从上次位置继续而不是全量重采。本地验证时可以把接口返回的数据量调小方便观察。另外如果你在 CEL 程序里调了模型接口可以用 curl 单独验证模型侧是否通curl -X POST https://taotoken.net/api \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d {model: 你的模型ID, messages: [{role: user, content: ping}]}模型对话入口在https://taotoken.net/chat接入文档在https://taotoken.net/doc具体请求体格式以文档为准。这一步通了再回到 Filebeat 里集成。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth本地跑 CEL Input报错集中在几个地方我按真实遇到的顺序拆。401 Unauthorized最常见。先确认auth.token.value引用的环境变量在 Filebeat 进程里能读到。Filebeat 以 systemd 启动时环境变量不会自动继承 shell 的需要在 unit 文件里加EnvironmentLOCAL_API_TOKENxxx或者用EnvironmentFile指向一个文件。另外确认type: Bearer和接口要求的认证方式一致有的接口要Authorization: Token xxx那就把 type 改成Token。如果 Key 放在文件里检查文件权限是不是 0600relaxed_permissions: false时权限不对会直接拒绝加载。local proxy failed这个报错通常出现在resource.url指向本地地址但 Filebeat 试图走代理时。本地部署场景下如果系统设了HTTP_PROXY环境变量Filebeat 会默认走代理导致访问127.0.0.1失败。解决办法是在配置里显式禁用代理或者在启动 Filebeat 时清掉代理环境变量。本地内网接口不需要代理禁用后请求直接走回环速度也更快。reading choices这个报错一般出现在 CEL 程序解析响应时decode_json()拿到的结构和你预期的不一样。比如接口返回的是{data: [...]}你直接body.decode_json()当成数组用就会在取字段时报错。排查方法是开failure_dump把失败响应落到本地文件看一眼实际返回结构再调整 CEL 程序里的字段路径。另外reading choices有时也和want_more配合有关如果程序里want_more: true但 cursor 没正确更新下一轮会重复读同一批数据。OAuth 相关报错ES 9.3 支持 OAuth2本地接内网身份提供商时常见问题是 token 过期没刷新、或者refresh_interval设得太长导致用了旧 token。检查auth.oauth2配置里的 token URL、client ID、client secret 是否和本地身份提供商一致。如果用的是文件动态令牌确认auth.file的path、prefix、refresh_interval三个字段都配了prefix要和接口要求的令牌类型匹配比如Bearer后面有个空格。还有一个容易忽略的点ES 9.x 里 cursor 是唯一跨重启持久化的状态其他 state 字段重启后会丢。如果你在 program 里依赖了非 cursor 的 state 字段做断点重启后会从头开始。所有需要跨重启的采集位置都必须放进 cursor。6. 语义一致 CTA本地跑通后往哪走本地把 CEL Input 跑通只是第一步。接下来你大概率会碰到两类需求一是采集量上来了需要更稳的调度和重试策略二是采集到的数据要做二次处理比如调模型做分类或摘要。前者建议把resource.retry和resource.keep_alive调优本地服务波动大就把max_attempts提到 5wait_max放到 10s。后者如果要在 CEL 程序里调模型Key 和 Base URL 按第 2 节准备好模型对话入口在https://taotoken.net/chat接入文档在https://taotoken.net/docAPI Keys 在https://taotoken.net/api-keys。长期跑编码或 Agent 类任务可以看 Coding Planhttps://taotoken.net/coding-plan。本地 ES 9.x 的 CEL Input 能力还在迭代9.3 的时间截断函数和失败 dump 对调试帮助很大建议本地验证时就把failure_dump开起来定位完再关。采集链路稳定后把配置里的明文密钥全部换成环境变量或权限 0600 的文件这是本地部署到内网长期运行之间必须补的一步。