Cloudflare 推出 Logpush Transformers，讓日誌即時過濾與重構

Logpush 盡快以批次方式傳送日誌，讓 Cloudflare 能夠幾乎即時提供資訊。在我們最近的 Birthday Week 期間，我們將 Logpush 推出至所有自助方案，並讓 Transformers 一般可用。

這些變更屬於 Cloudflare Observability 的八項主要更新，並體現我們致力於讓每個 Cloudflare 功能都能讓所有人使用的承諾。本篇文章深入探討 Transformers 在 Logpush 匯出流程中的運作。Observability 團隊使用 Logpush 將 Cloudflare 日誌匯出至安全資訊與事件管理系統（SIEMs）、分析平臺以及物件儲存。於日誌抵達目的地前，團隊可能需要移除健康檢查、省略敏感欄位，或為儀錶板與調查重新塑造記錄。如今，這通常需要一個獨立的處理服務，團隊必須部署、擴充、監控並維護。

舉例來說，收集多個 CDN 日誌的團隊可以在記錄抵達其分析平臺前加入一個靜態 provider 欄位以統一格式。團隊還需為匯出、接收並儲存他們一到達即丟棄的記錄付費，因此在傳送前過濾它們可以降低下游資料量與成本。

有了 Transformers，這一處理流程便移入 Logpush 匯出工作流程。團隊定義一個 SQL 查詢以篩選記錄、遮蔽敏感資料，並透過新增、重新命名、刪除或巢狀欄位來重塑輸出。Logpush 在匯出前執行此查詢，確保每個目的地收到符合其需求的資料。我們透過將 SQL 轉換引擎嵌入 Logpush 的高吞吐量匯出管線，並保持效能與資源使用可預測，讓客戶能輕鬆使用此工作流程。

Logpush Transformers 的工作原理

Transformers 是針對特定 Logpush 資料集編寫的 SQL 查詢。查詢會參照該資料集的欄位，並定義 Logpush 應匯出的記錄與欄位。Logpush 在匯出前將查詢套用於每個批次，未符合 WHERE 子句的記錄會被移除。其餘記錄中，SELECT 語句的輸出即為匯出的記錄。Transformers 可以改變每筆記錄的結構與內容，並以 NDJSON 格式匯出轉換後的結果。

客戶可以透過 Transformer Studio 或透過 API 程式化方式，在 Cloudflare 儀錶板上建立、預覽並管理 Transformers。儲存 Transformers 後，他們可以將其附加至使用相同資料集的相容 NDJSON Logpush 工作。每一次更新都會產生新版本，讓客戶能檢視 Transformers 隨時間的變更。

以下範例將 Cloudflare HTTP request log 重新塑造以符合團隊在多個 CDN 提供者之間共用的自訂 schema。它不會在序列化格式之間轉換：輸入與輸出均使用 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"
}

Transformers 重新命名並巢狀欄位，遮蔽客戶端 IP 位址的一部份，加入 provider 與 error 資訊，移除未使用的欄位，並過濾掉內部與健康檢查流量：

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 記錄採用已正規化的 schema：

{
  "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
}

Transformers 如何使用 Apache DataFusion 與 Arrow 執行 SQL

Cloudflare 的資料管線每秒處理超過十億個事件。其 Logpush 傳送服務，我們稱之為 Bufstream，將每秒超過 1.1 億筆 Logpush 記錄送達客戶目的地。它執行在數百個 Kubernetes pod 上，每個 pod 從 Buftee（Cloudflare 內部分散式基於日誌的訊息代理）取得一段時間的資料。此模型將工作水平分散至 Logpush 工作。工作量遵循 Pareto 分佈：大多數工作量較低，而少數高量工作每秒處理數百萬筆日誌。Transformer 必須滿足與 Logpush 其餘部分相同的操作需求：可靠傳遞、高吞吐量，以及可預測的資源使用，而不在匯出熱路徑中引入額外失敗情境。

我們最初的原型在 Cloudflare Workers 中執行客戶定義的 WebAssembly Transformers，Workers 提供全球分散且隔離的執行環境。然而，呼叫 Worker 會在 Logpush 匯出路徑中插入一次網路呼叫。該遠端依賴可能引入暫時性逾時與重試，降低傳遞吞吐量並增加傳遞延遲。

我們決定將 Transformer 執行保持於傳送服務內部程序，以優先考量可靠性與傳遞吞吐量。我們評估了多個 SQL 執行引擎並選擇了 Apache DataFusion，它已經為 Cloudflare Basin 提供支援。

Apache DataFusion 是一個基於 Apache Arrow 的 Rust 查詢引擎，擁有 SQL 計畫器與向量化、列式執行引擎。在基準測試中，對單一 Go goroutine 的合成 5,000 行 http_requests 批次執行預備 Transformer 約耗時 3 毫秒，約等於每秒 1.6 百萬行。

將 Transformer 附加至工作會為其處理路徑增加 CPU 工作。Bufstream 透過每工作轉換並行化，使高量工作能同時執行轉換批次。將 DataFusion 嵌入 Bufstream 使這些批次能利用服務現有的傳送基礎設施，包括排程、批次化與目的地控制，無需引入獨立的轉換層或網路跳躍。

選擇 DataFusion 意味著將使用 Go 編寫的 Bufstream 連線至 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 呼叫 C 程式的機制。

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 資料集定義的 schema 驗證。

每個 Arrow 批次都需要一個描述其欄位與型別的 schema。Bufstream 不是從特定批次的日誌推斷該 schema，而是提供每個 Logpush 資料集規格所產生的 schema。這為 DataFusion 提供了可用欄位與型別的穩定檢視，從而實現查詢驗證。Transformer 的 FROM 子句中命名的表決定查詢使用哪個資料集 schema。例如，以下 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 批次。

Transformer 設計為失敗時關閉。若轉換無法執行，Bufstream 不會推進工作交付位移，也不會送出未轉換的批次。以原始資料取代已轉換資料可能違反下游管道的 schema、過濾或刪除假設，因而 Bufstream 選擇重新嘗試失敗的轉換。

為了避免持續失敗的工作佔用健康工作所需的容量，Bufstream 會逐步延遲重試並在工作上報錯。長時間處於失敗狀態的工作將自動被停用。客戶可以在 Logpush 的 Health 標籤中檢視 Transformer Health 檢視，監控成功與失敗的轉換、轉換前後的資料量，以及轉換持續時間。

回到匯出管線

執行完成後，Bufstream 將 Arrow 輸出轉換為 NDJSON 日誌。它會將日誌批次處理、gzip 壓縮，並上傳至已設定的目的地。不同的目的地接收資料的速度不同，且可能暫時降低速率。Bufstream 在遵守目的地併發限制的同時調整上傳 goroutine 的數量。

將 SQL 執行嵌入 Bufstream，Transformers 能利用 Logpush 現有的分散式基礎設施、批次處理和目的地控制。Cloudflare 的資料管線負責處理這些複雜性，而客戶只需經由幾個步驟即可建立並套用 Transformer。

開始使用

在 Cloudflare 儀錶板中，從 Logpush page 開啟 Transformer Studio，並依照以下步驟操作：

選擇您想要轉換的 Logpush 資料集。
撰寫您的 SQL 查詢，並對樣本記錄進行預覽。
儲存 Transformer。
將其附加到符合條件的 Logpush 工作。

Transformers 可於 Free、Pro、Business 與 Enterprise 方案中使用。請參閱 Transformers documentation 瞭解設定細節，以及 Logpush pricing documentation 以取得最新價格資訊。