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

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

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

## なにが発表されたの？

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

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

## 今までどうだったの？

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

## これで何が変わるの？

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

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

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

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

```sql
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 をプロセス内で動かして高速に処理する
- 全プランで一般提供されている

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