Filter, reshape, and redact logs before export with Transformers

在导出前使用Transformers过滤、重塑和擦除日志 | Cloudflare 博客

post

开发者

2026年10月12日

在导出前使用Transformers过滤、重塑和擦除日志

Rob Taylor 和 Sahidya Devadoss

8分钟阅读时间

复制链接

Logpush以尽可能快的批量方式传输日志，使Cloudflare能够几乎实时提供信息。在最近的生日周活动期间，我们让Logpush在所有自助服务计划中可用，并使Transformers正式上线。

这些变更属于Cloudflare Observability的八项重大更新之一，体现了我们致力于让每个人都能使用Cloudflare所有功能的承诺。本文将深入探讨Transformers如何在Logpush的导出流程中工作。Observability团队使用Logpush将Cloudflare日志导出到安全信息和事件管理（SIEM）系统、分析平台和对象存储中。在这些日志到达目标之前，团队可能需要移除健康检查、省略敏感字段或重塑记录以适应仪表板和调查需求。目前，这通常需要团队部署、扩展、监控和维护单独的处理服务。

例如，一个从多个CDN收集日志的团队可以在日志到达其分析平台前添加静态提供商字段以标准化记录。团队还需要为导出、摄入和存储最终被丢弃的记录付费，因此在传输前过滤日志可以减少下游数据量和成本。

通过Transformers，这些处理流程被整合到Logpush的导出工作流中。团队通过定义SQL查询来过滤记录、擦除敏感数据并重塑输出（通过添加、重命名、删除或嵌套字段）。Logpush在导出前应用该查询，因此每个目标都能以所需格式接收数据。我们通过在Logpush的高吞吐量导出流程中嵌入SQL转换引擎，同时保持性能和资源使用可预测，使这一流程对客户更加简单。

Logpush Transformers的工作原理

Transformer是为特定Logpush数据集编写的SQL查询。该查询引用数据集中的字段，并定义Logpush应导出的记录和字段。Logpush在导出前将查询应用于每个批次。不满足WHERE子句的记录将被移除。对于剩余记录，SELECT语句的输出将成为导出的记录。Transformers可以更改每条记录的结构和内容，转换后的结果以NDJSON格式导出。

客户可以通过Cloudflare仪表板的Transformer Studio或通过API以编程方式创建、预览和管理Transformers。保存Transformer后，可以将其附加到使用相同数据集的兼容NDJSON Logpush任务。每次更新都会生成新版本，因此客户可以查看Transformer随时间的变化情况。

以下示例将Cloudflare HTTP请求日志重塑为团队定义的跨CDN提供商使用的模式。它不会在序列化格式之间进行转换：输入和输出均使用NDJSON。

{
"ClientIP"
:
"203.0.113.42"
,
"ClientRequestHost"
:
"example.com"
,
"ClientRequestMethod"
:
"POST"
,
"ClientRequestPath"
:
"/api/checkout"
,
"ClientRequestUserAgent"
:
"Mozilla/5.0"
,
"EdgeResponseStatus"
:
502
,
"EdgeStartTimestamp"
:
"2026-05-21T15:00:00Z"
,
"RayID"
:
"8e2a1c60ef9e1c9a"
}

Transformer 会重命名并嵌套字段、部分隐藏客户端 IP 地址、添加提供商和错误信息、移除未使用的字段，并过滤掉内部流量和健康检查流量：

SELECT
extract(epoch
FROM
to_timestamp(EdgeStartTimestamp))
AS
unix_ts,
regexp_replace(ClientIP,
'\.\d+\.\d+$'
,
'.XXX.XXX'
)
AS
redacted_ip,
UPPER
(ClientRequestMethod)
AS
method,
EdgeResponseStatus
AS
status
,
RayID
AS
ray_id,
'cloudflare'
AS
provider
,
named_struct(
'host'
, ClientRequestHost,
'path'
, ClientRequestPath
)
AS
request,
EdgeResponseStatus
>=
500
AS
is_server_error
FROM
http_requests
WHERE
ClientRequestHost
NOT
IN
(
'internal.example.com'
,
'health.example.com'
)
AND
ClientRequestPath
NOT
LIKE
'/healthz%'
AND
ClientRequestPath
NOT
LIKE
'/metrics%'

生成的 NDJSON 记录使用标准化的模式：

{
"unix_ts"
:
1779375600
,
"redacted_ip"
:
"203.0.XXX.XXX"
,
"method"
:
"POST"
,
"status"
:
502
,
"ray_id"
:
"8e2a1c60ef9e1c9a"
,
"provider"
:
"cloudflare"
,
"request"
: {
"host"
:
"example.com"
,
"path"
:
"/api/checkout"
},
"is_server_error"
:
true
}

Transformer 如何通过 Apache DataFusion 和 Arrow 执行 SQL

Cloudflare 的数据管道每秒处理超过十亿个事件。其称为 Bufstream 的日志推送服务每秒向客户目的地传输超过 1.1 亿条日志记录。该服务运行在数百个 Kubernetes 容器中，每个容器从 Cloudflare 的分布式基于日志的内部消息代理 Buftee 中获取特定时间片的数据。这种模式在日志推送任务之间水平分配工作负载。任务数量遵循类似帕累托分布：大多数任务处于较低流量水平，而少数高流量任务每秒处理数百万条日志。Transformer 需要满足与日志推送其他部分相同的运营要求：可靠传输、高吞吐量以及在不增加导出热路径故障场景的前提下实现可预测的资源使用。

我们最初的原型在 Cloudflare Workers 中运行客户定义的 WebAssembly Transformer，该平台提供全球分布式、隔离的计算环境。然而，调用 Worker 会在日志推送导出路径中插入网络调用。这种远程依赖可能引入瞬时超时和重试，降低传输吞吐量并增加传输延迟。

为了优先保证可靠性和传输吞吐量，我们决定在交付服务内部进程执行 Transformer。我们评估了多个 SQL 执行引擎，最终选择了 Apache DataFusion，该引擎已用于 Cloudflare Basin。

Apache DataFusion 是基于 Apache Arrow 的 Rust 查询引擎，包含 SQL 计划器和向量化列式执行引擎。在基准测试中，通过单个 Go 协程对合成的 5000 行 http_requests 批次执行预处理 Transformer，耗时约 3 毫秒，相当于每秒处理约 160 万行数据。

将 Transformer 关联到任务会增加其处理路径的 CPU 工作量。Bufstream 为每个任务使用独立的转换并行处理，使高流量任务能够并发执行转换批次。将 DataFusion 嵌入 Bufstream 使这些批次能够利用服务现有的交付基础设施，包括调度、批处理和目标控制，而无需引入独立的转换层级或网络跳转。

选择 DataFusion 意味着需要将使用 Go 编写的 Bufstream 连接到使用 Rust 编写的引擎。下一个挑战是高效地在 Go 和 Rust 之间传递类型化批次。

下图展示了进程内 Transformer 管道的最终设计。

从记录到列

Bufstream 接收来自 Buftee 的压缩 Logpush 批次，其中包含 Cap'n Proto 记录。对于附加了 Transformer 的任务，Bufstream 会解压批次、解析符合条件的日志，并构建一个类型化的 Apache Arrow RecordBatch，然后执行转换。

Apache Arrow 是一种语言无关的列式内存格式，专为现代 CPU 和 GPU 上的高效分析而设计。它将平面和嵌套数据表示为类型化、连续的列。Arrow IPC 为 Bufstream 和 DataFusion 提供了一种标准方式，通过 Go 和 Rust 之间的外部函数接口（FFI）边界交换批次，该接口通过 C 应用程序二进制接口（ABI）暴露，并通过 Go 的 cgo 机制从 Go 调用。Go 和 Rust 运行时不会共享转换数据的所有权。Go 仅在 cgo 调用期间将输入 Arrow IPC 缓冲区的指针传递给 Rust，调用返回后 Rust 不会保留该指针。Rust 拥有转换后的输出缓冲区；Go 将其复制到 Go 管理的内存中，然后显式释放 Rust 的分配。这在语言边界上明确表达了所有权和生命周期，避免了跨运行时内存管理错误。

使用 Arrow IPC 意味着 Rust 的 DataFusion 封装器可以使用 Arrow StreamReader 读取类型化的 RecordBatch 值，而不是从中间格式（如 JSON）解析单个行并重新构建类型化的 Arrow 列以执行。字段类型、可空性和嵌套值在边界两侧表示一致，因此 DataFusion 可以以原生列式模型接收数据。

跨越 Go/Rust 边界并非没有代价：批次仍然需要序列化、解码和在内存中管理。Arrow IPC 使这些工作保持批次导向，避免逐条记录的解析和转换。这使边界更适配 Logpush 的高吞吐量传输管道，同时保持对 Logpush 数据集定义的模式验证。

每个 Arrow 批次都需要一个描述其字段和类型的模式。 Bufstream 不会从特定批次的日志中推断该模式，而是从每个 Logpush 数据集规范中提供生成的模式。这为 DataFusion 提供了对可用字段和类型的稳定视图，从而实现查询验证。Transformer 的 FROM 子句中命名的表选择查询使用的数据集模式。例如，此 Transformer 查询 http_requests 数据集：

SELECT
ClientIP, EdgeResponseStatus, RayID
FROM
http_requests
WHERE
EdgeResponseStatus
>=
500

执行转换

当客户通过 API 或 Transformer Studio 保存 Transformer 时，Cloudflare 会将其 SQL 查询存储在 R2 中。Bufstream 通过多层缓存检索查询，并将其转换为可重复使用的 DataFusion 逻辑计划，该计划定义了扫描、过滤、投影和表达式。Bufstream 为任务数据的每个时间片准备一次该计划，并在该时间片的每个 Arrow 批次中重复使用。对于每个新的 Arrow 批次，Rust 封装器将批次绑定为计划的表输入，执行查询，并返回包含所选记录和字段的 Arrow 批次。

Transformers 被设计为在失败时关闭。如果转换无法执行，Bufstream 不会推进任务的投递偏移量，也不会发送未转换的数据批次。用原始数据替代转换后的数据可能会违反下游流水线的模式、过滤或数据擦除假设，因此 Bufstream 会选择重新尝试失败的转换。

为防止持续失败的任务占用健康任务所需的资源，Bufstream 会逐步延迟重试并报告任务错误。长时间处于失败状态的任务将被自动禁用。客户可以在 Logpush 的“健康”标签页中“转换器健康视图”监控转换成功与失败情况、转换前后数据量以及转换耗时。

返回导出流水线

执行完成后，Bufstream 将 Arrow 输出转换为 NDJSON 日志格式。它会对日志进行批处理、gzip 压缩，并上传到配置的目标地址。不同目标接收数据的速度不同，可能会暂时减慢。Bufstream 会根据目标的并发限制动态调整上传的 goroutine 数量。

通过将 SQL 执行嵌入 Bufstream，Transformers 可复用 Logpush 现有的分布式基础设施、批处理机制和目标控制功能。Cloudflare 的数据流水线处理这些复杂性，而客户只需通过几个步骤即可创建并应用 Transformer。

入门指南

在 Cloudflare 控制台的 Logpush 页面打开 Transformer Studio，然后按照以下步骤操作：

• 选择要转换的 Logpush 数据集。
• 编写 SQL 查询并使用示例记录进行预览。
• 保存 Transformer。
• 将其附加到匹配的 Logpush 任务。

Transformers 支持免费版、专业版、企业版和旗舰版计划。有关设置详情，请参阅 Transformers 文档；有关当前定价信息，请参阅 Logpush 定价文档。

目录列

讨论列

相关标签及社交媒体链接

相关标签

关注社交媒体

• Cloudflare

电子邮件订阅