通用表达式语言（CEL）：CEL 输入如何改进 Elastic Agent 集成中的数据收集

Elastic Agent 集成支持用户从多种来源将数据摄取至 Elasticsearch。这些集成会将采集逻辑、摄取管道、仪表板以及其他构件打包在一起，形成一个可通过 Kibana Web 界面安装和管理的包。

集成通过配置 Filebeat 输入来执行数据采集。为了从 HTTP API 采集数据，我们通常使用 HTTP JSON 输入。然而，即便是最基础的列表 API，在具体细节上也可能千差万别，而 HTTP JSON 输入采用 YAML 配置进行数据转换的模式，往往使所需的采集逻辑难以自然表达，有时甚至无法实现。

为此，我们引入了通用表达式语言 (CEL) 输入，让与 HTTP API 的交互更加灵活。CEL 是一种专为嵌入到应用中而设计的语言，适用于需要以快速、安全且可扩展的方式来表达条件和数据转换的场景。通过 CEL 输入，集成构建者只需编写一个表达式，即可读取设置、跟踪自身状态、发起请求、处理响应，并最终返回可直接摄取的事件。

本文将探讨 CEL 与其他编程语言的区别、我们针对 Filebeat 的 CEL 输入所进行的扩展，以及由此带来的数据采集逻辑表达灵活性与能力提升。

CEL 及其在输入中的工作方式

CEL 是一种表达式语言，它没有“语句”这一概念。在编写 CEL 时，您不是通过编写语句来指示它执行操作，而是通过编写表达式来定义要生成的值。每个 CEL 表达式都会计算出一个值，较小的表达式可以组合成更大的表达式，按照更复杂的规则生成结果。稍后，我们将了解如何通过表达式来实现其他语言中需要用语句来完成的逻辑。

CEL 有意设计为一种非图灵完备的语言，因此不支持无界循环。稍后，我们将介绍如何使用宏来处理列表和映射；正是通过禁止无界循环，CEL 能够保证每个表达式的执行时间可预测且有限。

CEL 输入需要配置一个 CEL 程序（即一个表达式）以及一些初始状态。这个初始状态会作为输入传递给程序。程序执行后会产生一个新的输出状态。如果输出状态中包含事件列表，这些事件会被提取出来并发布。输出状态的其余部分则作为下一次执行的输入。如果输出状态包含一个或多个事件，并且带有 want_more: true 标志，程序会立即再次执行；否则，它会在剩余的配置时间间隔内等待，然后才继续执行。下面是输入控制流的简化示意图：

只要输入还在运行，每次执行的输出都会作为下一次执行的输入。存储在 “cursor” 键下的数据会被保存到磁盘，并在输入重启后重新加载，但其余状态在重启后不会保留。

CEL 语言本身功能有限，且不会产生副作用，但它支持扩展。cel-go 实现为其添加了一些功能，例如可选语法和类型。Mito 库在 cel-go 的基础上进一步扩展，增加了发起 HTTP 请求等功能。CEL 输入使用的正是 Mito 提供的 CEL 版本。

与 Mito 合作

要使用 CEL 输入构建或调试集成，最关键的是理解：针对给定的输入状态，您的 CEL 程序会输出什么。在开发过程中，如果每次都要在完整的 Elastic Stack 环境中通过 CEL 输入来运行程序，会非常繁琐。为了更快地获得反馈，可以使用 Mito 的命令行工具，它允许您直接运行 CEL 程序，并查看它对给定输入产生的输出。

Mito 是用 Go 语言编写的，您可以通过以下命令安装：

go install github.com/elastic/mito/cmd/mito@latest

使用 Mito 运行 CEL 程序时，通常需要提供两个文件：一个是 JSON 文件，包含初始输入状态；另一个是 CEL 源文件，包含程序代码。

mito -data state.json src.cel

为了方便复制粘贴，本文示例采用单条命令的形式，利用 <(echo '...content...') 语法将每个文件的内容动态生成临时文件。在实际开发中，直接使用真实的文件会更方便。

从 GitHub 获取 issue 数据

以下示例是一个完整的 CEL 程序，用于从 GitHub API 获取 issue 数据。它的初始输入状态包含 API 端点的 URL 以及分页处理的相关信息。CEL 程序利用这些输入数据生成请求，然后解码响应，从中生成事件，并将这些事件作为输出状态的一部分返回。

mito -data <(echo '
  {
    "url": "https://api.github.com/repos/elastic/integrations/issues",
    "per_page": 3,
    "max_pages": 3
  }
') <(echo '
  int(state.?cursor.page.orValue(1)).as(page,
    (
      state.url + "?" + {
        "state": ["all"],
        "sort": ["created"],
        "direction": ["asc"],
        "per_page": [string(state.per_page)],
        "page": [string(page)],
      }.format_query()
    ).as(full_url,
      request("GET", full_url).with({
        "Header": {
          "Accept": ["application/vnd.github+json"],
          "X-GitHub-Api-Version": ["2022-11-28"],
        }
      }).do_request().as(resp,
        resp.Body.decode_json().as(data,
          state.with({
            "events": data.map(i, {
              "html_url": i.html_url,
              "title": i.title,
              "created_at": i.created_at,
            }),
            "cursor": { "page": page + 1 },
            "want_more": size(data) == state.per_page && page < state.max_pages,
          })
        )
      )
    )
  )
')

程序第一次执行会产生以下输出：

{
  "cursor": {
    "page": 2
  },
  "events": [
    {
      "created_at": "2018-09-14T09:47:35Z",
      "html_url": "https://github.com/elastic/integrations/issues/3250",
      "title": "Increase support of log formats in haproxy filebeat module"
    },
    {
      "created_at": "2019-02-06T12:37:37Z",
      "html_url": "https://github.com/elastic/integrations/issues/487",
      "title": "ETCD Metricbeat module needs polishing and grooming"
    },
    {
      "created_at": "2019-08-13T11:33:11Z",
      "html_url": "https://github.com/elastic/integrations/pull/1",
      "title": "Initial structure"
    }
  ],
  "max_pages": 3,
  "per_page": 3,
  "url": "https://api.github.com/repos/elastic/integrations/issues",
  "want_more": true
}

这些事件会被提取出来，并在 CEL 输入中发布，以便摄取。剩余的输出数据将作为输入状态传递给下一次 CEL 程序执行。

为了帮助理解这个 CEL 程序的工作原理，我们先看一些更简单的 CEL 示例，并深入讨论 CEL 输入的运行细节。

CEL 基础知识

CEL 语言中只有表达式，没有语句。每个 CEL 表达式成功执行后都会得到一个最终值。以下是一个最简单的 CEL 表达式示例及其输出：

mito <(echo '
  "hello" + " " + "world"
')

"hello world"

许多简单表达式都很直观。数学运算要求操作数类型相同（例如 int 与 int），因此请根据需要进行类型转换（此处是从 int 转换为 double）：

mito <(echo '
  double((1 + 2) * (3 + 4)) / 2.0
')

10.5

CEL 语言中没有变量，但您可以为表达式的结果命名，并借助 Mito 的 as 宏在更大的表达式中使用它。在此示例中，表达式 (1 + 1) 的值为 2，而 .as(n, ...) 会将该值命名为 n，以便在表达式 "one plus one is "+string(n) 中使用：

mito <(echo '
  (1 + 1).as(n, "one plus one is "+string(n))
')

"one plus one is 2"

还可以在映射中累积信息，并在表达式中稍后使用；下面的示例用 with 演示了这一点：

mito <(echo '
  { "key": "value" }.with({ "key2": "value2" }).as(data,
    {
      "data": data,
      "size": size(data),
    }
  )
')

{
  "data": {
    "key": "value",
    "key2": "value2"
  },
  "size": 2
}

我们再来看这个例子。注意嵌套部分 ({ "data": data, "size": size(data), })，它定义了最终值的结构。这是一个映射，包含 "data" 和 "size" 两个键。这些键的值依赖于 data，而它是由外层表达式定义的。从内向外阅读 CEL 表达式，有助于快速理解它们会返回什么。

CEL 没有 if 这样的控制流语句，但可以用三元运算符实现条件分支：

mito <(echo '
  1 + 1 < 12 ? "few" : "many"
')

"few"

由于 CEL 不是图灵完备语言，因此不支持无限循环和递归。这保证了执行时间可预测，并且与输入数据的大小和表达式的复杂度成正比。

虽然单个 CEL 表达式不能使用无限循环，但您可以用 map 这样的宏来处理列表和映射：

mito <(echo '
  [1, 2, 3].map(x, x * 2)
')

[2, 4, 6]

本节介绍了以下内容：

• 字符串、数字、列表和映射。

• 字符串连接。

• 数学运算。

• 类型转换。

• 条件语句。

• 命名子表达式。

• 处理集合。

接下来，我们将学习如何发出 HTTP 请求。

请求

Mito 为 CEL 扩展了发起 HTTP 请求的能力：

mito <(echo '
  get("https://example.com").as(resp, string(resp.Body))
')

"<!doctype html><html lang=\"en\"><head><title>Example Domain</title>..."

请求可以在执行前明确构造，这样就可以使用不同的 HTTP 方法，并添加请求头和请求体。

在这个示例中，我们借助 format_query 构建 URL，向请求添加一个请求头，并使用 decode_json 解析响应体。当传入 -log_requests 选项时，Mito 会以 JSON 格式记录每个请求和响应的详细信息。

mito -log_requests <(echo '
  request("GET",
    "https://postman-echo.com/get?" + {
        "q": ["query value"]
     }.format_query()
  ).with({
    "Header": { "Accept": ["application/json"] }
  }).do_request().as(resp, {
    "status": resp.StatusCode,
    "data": resp.Body.decode_json(),
  })
')

{"time":"...","level":"INFO","msg":"HTTP request",...}
{"time":"...","level":"INFO","msg":"HTTP response",...}
{
  "data": {
    "args": {
      "q": "query value"
    },
    "headers": {
      "accept": "application/json",
      "accept-encoding": "gzip, br",
      "host": "postman-echo.com",
      "user-agent": "Go-http-client/2.0",
      "x-forwarded-proto": "https"
    },
    "url": "https://postman-echo.com/get?q=query+value"
  },
  "status": 200
}

管理状态和评估

我们已经介绍了如何发出请求，以及生成期望输出状态所需的 CEL 基础知识，接下来就来仔细看看输出状态应该包含哪些内容，以及这些内容如何帮助我们引导后续的处理流程。

集成的 CEL 程序需要确保其输出状态能够作为下一次执行的输入。配置设定了初始状态，输出中应保留这些状态值，并根据需要更新。一个简单的做法是使用 state.with({ ... })，在原有状态映射的基础上合并覆盖。小型程序常见的一种模式是将整个程序包裹在 state.with() 中，这样在成功、错误等输出数据的分支中，就无需重复处理状态传递逻辑。

如果某些状态值不是在初始输入状态中硬编码，而是在执行过程中动态初始化，那么程序在设置初始值之前需要先检查是否已存在值。这正是可选语法和类型支持能够发挥作用的场景。在映射键的字段名前面加上问号，访问就变为可选：可能得到值，也可能得不到，但后续仍可以进行可选访问，并且在无值时可以方便地提供默认值：

mito -data <(echo '{}') <(echo '
  int(state.?counter.orValue(0)).as(counter,
    state.with({
      "counter": counter + 1,
      "want_more": counter + 1 < 3,
    })
  )
')

{ "counter": 1, "want_more": true }
{ "counter": 2, "want_more": true }
{ "counter": 3, "want_more": false }

在这个例子中，从 state 读取的计数器值需要转换为 int，因为状态中的所有数字都按照 JSON 和 JavaScript Number 类型的惯例序列化为浮点数。另外需要注意的是，Mito 会响应 "want_more": true，但在 CEL 输入中运行时，只有当输出中同时包含事件时，才会重复执行。

由 CEL 输入运行的 CEL 程序必须在输出映射中包含一个 "events" 键。它的值可以是事件映射的列表、空列表，或单个事件映射。单个事件映射通常用于表示错误。该事件会被输入发布，其值也会被记录到日志；如果设置了 error.message，该值还会用于更新集成在 Fleet 中的健康状态。如果程序生成的是单个非错误事件，最好将其包装在列表中。

回顾一下我们之前的 GitHub issues 程序的输出：

{
  "url": "https://api.github.com/repos/elastic/integrations/issues",
  "per_page": 3,
  "max_pages": 3,
  "cursor": {
    "page": 2
  },
  "events": [
    { ... },
    { ... },
    { ... }
  ],
  "want_more": true
}

该程序通过以下方式有效地管理了其状态：

• 在 url、per_page 和 max_pages 字段中沿用了初始状态的值。

• 在 cursor.page 中添加了需要在重启后持久化的状态。

• 返回准备发布在 events 列表中的事件。

• 请求立即与 want_more: true 进行重新评估。

现在您已经理解了可选访问和状态管理，以及 CEL 基础知识和 HTTP 请求，完整的 GitHub issues 示例程序应该已经比较容易阅读了。不妨用 Mito 运行一下，并尝试做些修改。

回顾与资源

本文介绍了 CEL 语言，以及它如何在 Mito 库中得到扩展以用于 CEL 输入。通过一个从 GitHub API 获取 issue 数据的示例程序，我们展示了 CEL 的灵活性，并逐一解析了理解该程序所需的全部细节：访问初始状态中的配置、与 HTTP API 交互、返回待导入的事件，以及为后续执行管理状态。

要深入学习并使用 CEL 输入构建集成，以下资源值得探索：

• CEL 输入 － Filebeat 文档

• Mito 文档

• 通用表达式语言 － cel.dev 网站

• 创建集成 － Elastic 文档

对于使用 CEL 输入构建集成，最有价值的资源或许是现有 Elastic 集成中的 CEL 代码，这些代码可以在 GitHub 上找到：

cel.yml.hbs Elastic 集成存储库中的 cel.yml.hbs 文件 － GitHub。