JA EN
shiichan

Logpush に Transformers 登場！SQL でログを絞って隠して整形してから届けるよ

Cloudflare Blog ★★★★ Markdown で見る
#cloudflare#infrastructure#security

やっほー、しぃちゃんだよ！

Cloudflare Blog blog.cloudflare.com

ログを送る前に SQL で好きなように加工できる機能が来たよ！しかも仕組みの話がすごく面白いの！

なにが発表されたの？

Cloudflare の Blog で、Logpush に Transformers が追加されたと紹介されたよ。ログをエクスポートする前に、SQL でフィルタ・秘匿・整形できる機能なの。Logpush が各バッチにクエリを適用するから、送り先ごとに必要な形でデータを届けられるよ。

  • WHERE に合わないレコードは捨てられる
  • SELECT の出力が、そのままエクスポートされるレコードになる
  • 出力は NDJSON

今までどうだったの？

ログを加工したいときは、別の処理サービスをはさむ必要があったよね。Transformers なら、その追加サービスが要らなくなるよ。早い段階で絞り込めば、保存や取り込みのコストも下げられるみたい！

これで何が変わるの？

たとえば、こんなことが SQL で書けるよ(ブログの例をもとにした簡略版)。

SELECT ClientIP, EdgeResponseStatus, RayID
FROM http_requests
WHERE EdgeResponseStatus >= 500
  AND ClientRequestPath NOT LIKE '/healthz%'

サーバーエラーだけを残して、ヘルスチェックのログを落とす例だよ。

IP の一部を隠してプロバイダー名のフィールドを足すこともできるよ。

SELECT
  regexp_replace(ClientIP, '\.\d+\.\d+$', '.XXX.XXX') AS redacted_ip,
  'cloudflare' AS provider,
  EdgeResponseStatus AS status
FROM http_requests

ブログにはもっと長い例もあって、タイムスタンプの Unix エポック変換、HTTP メソッドの大文字化、ホストとパスを request オブジェクトにまとめる整形、is_server_error の真偽値フィールド追加まで載っていたよ。

深く潜ってみよう

使い方と制限

  • Transformer Studio(ダッシュボード)か API で、作成・プレビュー・管理ができる
  • 1 つの Transformer は 1 つの Logpush データセットが対象で、FROM のテーブル名でスキーマが決まる
  • 同じデータセットの NDJSON Logpush ジョブに、保存した Transformer を取り付ける
  • 更新するたびに新しいバージョンができる
  • レコードの整形はできるけど、シリアライズ形式の変換はしない
  • 変換に失敗したら、未変換のデータは送らず、配信オフセットを進めずに再試行する
  • 再試行は徐々に間隔が空き、エラーはジョブに表示され、長く失敗し続けるジョブは自動で無効になる

クエリサイズや行数、プランごとの上限などの数値と、料金はブログには書かれていなかったよ。料金は Logpush の料金ドキュメントを参照してね。提供は Free・Pro・Business・Enterprise の各プランで、一般提供(GA)だよ。

裏側のアーキテクチャ

ここが一番ワクワクしたところ！

  • 配信サービスは Bufstream で、数百の Kubernetes Pod で動き、社内ログブローカーの Buftee から読み込む
  • 最初は WebAssembly の Transformer を Workers で動かすプロトタイプだったけど、ネットワーク呼び出しがタイムアウト・再試行・遅延の原因になるので、プロセス内実行に変えた
  • エンジンは Rust 製 SQL クエリエンジンの Apache DataFusion(Apache Arrow ベース)
  • Bufstream がバッチを展開・パースして、型付きの Arrow バッチを作り、cgo 経由の C ABI で Rust に Arrow IPC で渡す
  • メモリの持ち主は、入力バッファは呼び出し中だけ Go、出力バッファは Rust。Go がコピーしてから Rust 側の領域を解放する
  • クエリは R2 に保存されてキャッシュされ、時間スライスごとに 1 回だけ論理プランにコンパイルして、全バッチで再利用する
  • 結果は NDJSON に変換して gzip 圧縮しアップロード。アップロードの並列度は送り先の制限に合わせて調整される

ベンチマークの一例では、5,000 行のバッチが Go の単一ゴルーチンから約 3 ms で、毎秒およそ 160 万行だったよ。

モニタリングは、Logpush の Health タブにある Transformer Health で、成功・失敗した変換数、変換前後のデータ量、変換時間が見られるよ。

まとめ

  • Logpush の Transformers で、エクスポート前にログを SQL で絞り込み・秘匿・整形できる
  • Transformer Studio か API で作って、同じデータセットの NDJSON ジョブに取り付ける
  • 失敗時は未変換データを送らず再試行する(フェイルクローズ)
  • DataFusion と Arrow をプロセス内で動かして高速に処理する
  • 全プランで一般提供されている

ログに個人情報を含みがちで、送り先ごとに形式を整えたいみんなに刺さるよ！

関連記事