Cloudflare Logpush Transformers：失敗時は遮断、生ログでの代替送信なし

NDJSONジョブに付ける単一のSQLクエリ

TransformerはLogpushがバッチを配信する前に、レコードへSQLクエリを一つ適用する。WHEREが残すレコードを選び、SELECTの出力が配信されるレコードになる。

接続先は同じデータセットを使うNDJSON出力のジョブでなければならない。FROMのテーブル名がジョブのデータセットと異なると接続に失敗する。

範囲の決まったレコード変換であり、汎用のSQLデータウェアハウスではない。結合、サブクエリ、集約、ウィンドウ関数、ORDER BYには対応しない。

Logpushの配信プロセス内で実行される変換

技術記事によれば、変換は別のリモート呼び出しを使わず、Logpushの配信サービス内でRust製クエリエンジンApache DataFusionにより実行される。初期の試作はWorkersを呼んでいたが、配信経路にネットワーク呼び出しが入るためこの方式を選んだ。

変換を実行できない場合、ジョブの配信オフセットは進まず、未変換のバッチも送られない。Cloudflareはこれを失敗時に遮断する（fail closed）設計と説明している。

Logpushは再試行の間隔を徐々に延ばし、ジョブにエラーを報告する。失敗状態が長く続いたジョブは自動で無効化される。

注意点：再試行は保持の約束ではない

ドキュメントは、失敗が続くと該当バッチのレコードは最終的に破棄され、復旧できないと記している。オフセットを保って再試行しても、無制限の保持や最終的な配信は保証されない。

最後のエラーはジョブのlast_errorフィールドに表示される。ドキュメントが挙げる主な原因は出力サイズの上限超過とCloudflare内部のエラーだ。

トレードオフは明確だ。未変換のログが外へ出るのを防ぐ代わりに、障害調査に必要なストリームが途切れることがある。

フィルターからTransformerへ移すときの条件

一つのLogpushジョブに置けるのはフィルターかTransformerのどちらかだけだ。ダッシュボードのTransformer Studioは既存のフィルターを同等のSQL条件式に変換し、それをWHEREに移すことでフィルターの動作が保たれる。

APIはこの変換を行わない。フィルターを自分でSQL条件にしてWHEREに入れたうえで、同じジョブ更新リクエストでfilterを空にしtransformer_idを指定する。

{
	"filter": "",
	"transformer_id": 42
}

filterを空にせずtransformer_idだけを指定すると、APIは400を返す。既存の選択ルールを移さずに新しいSQLだけをつなぐと、外へ出るイベントが変わる。

プレビューはデータセットごとのサンプルレコード一件で実行される。プレビューの成功は、本番のすべてのケースを通った証拠ではない。

SQL構文より先に決める出力条件

導入は最短のクエリより、望む出力条件から始めるのが合っている。外へ送ってよいフィールド、調査用に残すフィールド、意図的に除くレコードを先に決める。

欠損値や特殊な内容を含む代表的なレコードで試し、実際の配信結果を確かめる。目立つ識別子を一つ消したり名前を変えたりしても、残りのフィールドの機微情報まで消えたわけではない。

この部分はSQLの構文チェックでは終わらない、ポリシーとデータのレビューだ。

変換の成功とレコードの到着を分けて見る監視

監視は、変換が成功しているかと、予想したレコードが届いているかを分けて答える必要がある。

送信量の減少は、フィルタリングの効果かもしれないし、広すぎる除外条件や止まったジョブかもしれない。承認したクエリと予想出力の例を保管しておけば、変更を同じ基準で判断できる。

生ログでの代替送信は復旧計画ではない。失敗した変換は診断して直す対象だ。

失敗時の遮断は、配信の監視とデータ欠落への対応計画がそろってこそ価値を持ち、そのどちらの代わりにもならない。