Skip to content

CLOUDFLARE 2026-10-12

Read original ↗

Filter, reshape, and redact logs before export with Transformers

Summary

Cloudflare made Transformers generally available during Birthday Week 2026 and this post is the architecture deep-dive on how they work inside the Logpush export pipeline. A Transformer is a SQL query written against a specific Logpush dataset that Logpush applies to each batch before export: rows failing the WHERE clause are dropped, and the SELECT output becomes the exported record (NDJSON in, NDJSON out). This lets teams filter health-checks, redact sensitive fields, and reshape records (add/rename/remove/nest fields) inside the export workflow instead of running a separate processing service — and because discarded records never leave Cloudflare, it cuts downstream ingest and storage cost. The engineering substance is how the SQL executes at Logpush scale: Cloudflare embedded Apache DataFusion (a Rust, Arrow-native vectorized query engine) in-process inside Bufstream, its Go-based Logpush delivery service, rather than invoking a remote Workers service that would insert a network hop into the export hot path. Typed batches cross the Go↔Rust boundary as Arrow IPC over a C ABI via cgo, with explicit ownership/lifetime rules to avoid cross-runtime memory bugs.

Key takeaways

  • A Transformer is dataset-scoped SQL applied per batch before export. "Records that do not match the WHERE clause are removed. For the remaining records, the output of the SELECT statement becomes the exported record." Both input and output are NDJSON — it reshapes, it does not transcode. (Source: sources/2026-10-12-cloudflare-filter-reshape-redact-logs-transformers)
  • Filtering before export reduces downstream data volume and cost. "Teams also pay to export, ingest, and store records they discard as soon as they arrive, so filtering them before delivery can reduce downstream data volume and cost." The canonical motivating example is dropping internal/health-check traffic (/healthz%, /metrics%) and normalizing a multi-CDN schema with a static 'cloudflare' AS provider field. This is transform-before-load pushed to the producer side. (concepts/egress-cost)
  • The scale numbers. Cloudflare's data pipeline ingests "more than one billion events per second." Bufstream, the Logpush delivery service, delivers "more than 110 million Logpush records per second to customer destinations," running across "hundreds of Kubernetes pods." Job volume follows a Pareto-like distribution — most jobs are small, a few high-volume jobs process millions of logs/second (concepts/partition-skew-data-skew).
  • In-process execution was chosen over a Workers network call to protect the hot path. The prototype ran customer WebAssembly Transformers in Workers, but "invoking a Worker would insert a network call into the Logpush export path. That remote dependency could introduce transient timeouts and retries, reducing delivery throughput and increasing delivery latency." The decision: "keep Transformer execution in-process within our delivery service to prioritize reliability and delivery throughput." (patterns/co-located-inference-in-serving-layer, concepts/hot-path)
  • DataFusion was selected because it is Rust + Arrow-native and already powers Cloudflare Basin. "Apache DataFusion is a Rust query engine built on Apache Arrow, with a SQL planner and vectorized, columnar execution engine." Benchmark: a prepared Transformer against a synthetic 5,000-row http_requests batch from a single Go goroutine took ~3 ms, ≈ 1.6 million rows/second.
  • The Go↔Rust boundary is crossed with Arrow IPC over cgo, with explicit ownership. Bufstream (Go) decompresses a batch of Cap'n Proto records from Buftee, parses the eligible logs, builds a typed Arrow RecordBatch, and passes a pointer to its input Arrow IPC buffer to Rust only for the duration of the cgo call. Rust owns the output buffer; Go copies it into Go-managed memory and then explicitly releases the Rust allocation. "This makes ownership and lifetimes explicit across the language boundary, avoiding cross-runtime memory-management bugs."
  • Arrow IPC avoids per-row parsing. Using Arrow IPC, the Rust DataFusion wrapper reads typed RecordBatch values via an Arrow StreamReader "rather than parse individual rows from an intermediate format such as JSON and rebuild typed Arrow columns." Field types, nullability, and nested values are consistent on both sides, so DataFusion gets data in its native columnar model (concepts/columnar-storage-format). Crossing the boundary still costs serialize/decode/memory work, but Arrow IPC keeps it batch-oriented, a better fit for a high-throughput pipeline.
  • Schema comes from the dataset spec, not inferred per batch. "Rather than infer that schema from the logs in a particular batch, Bufstream supplies a generated schema from each Logpush dataset's specification." The table in the FROM clause selects which dataset schema the query validates against — a stable, validateable view of available fields/types.
  • The plan is prepared once per time slice and reused. The saved SQL lives in R2; Bufstream fetches it through a multi-layer cache and compiles it to a reusable DataFusion logical plan (scan, filters, projections, expressions), prepared once per time slice of a job's data and reused for every Arrow batch in that slice. Per batch, the Rust wrapper binds the batch as the plan's table input, executes, and returns Arrow batches.
  • Transformers fail closed. "If a transform cannot be executed, Bufstream neither advances the job's delivery offset nor sends the untransformed batch," because delivering raw data could violate the schema/filter/redaction assumptions downstream. It re-attempts instead. To stop a persistently failing job from starving healthy ones, Bufstream progressively delays retries (concepts/exponential-backoff-jitter), reports errors on the job, and auto-disables jobs stuck failing for an extended period. (concepts/fail-open-vs-fail-closed)
  • Per-job transform parallelism + adaptive upload concurrency. Because attaching a Transformer adds CPU to a job's path, Bufstream uses per-job transform parallelism so high-volume jobs run transform batches concurrently. On output it converts Arrow → NDJSON, batches, gzip-compresses, and uploads — adjusting the number of upload goroutines while respecting destination concurrency limits (destinations accept data at different rates and can slow down — concepts/backpressure).

The worked example (verbatim)

Input http_requests record (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"
}

The Transformer SQL (renames + nests fields, redacts the client IP, adds provider + is_server_error, drops unused fields, filters internal + health-check traffic):

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%'

Output (normalized 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
}

The in-process Transformer pipeline (end to end)

  1. Ingest — Bufstream receives compressed Logpush batches of Cap'n Proto records from Buftee, Cloudflare's distributed log-based in-house message broker. Each of the hundreds of Bufstream pods claims a time slice of a job's data from Buftee, distributing work horizontally across Logpush jobs.
  2. Decode → columns — for a job with a Transformer attached, Bufstream decompresses, parses eligible logs, and builds a typed Arrow RecordBatch using the dataset's generated schema.
  3. Plan once — the SQL (stored in R2, fetched through a multi-layer cache) is compiled to a reusable DataFusion logical plan, prepared once per time slice.
  4. Execute per batch — the Rust wrapper binds each Arrow batch as the plan's table input, executes (vectorized/columnar), returns Arrow batches. Fail-closed on error.
  5. Return to export — Arrow output → NDJSON → batch → gzip → upload to the destination with adaptive upload-goroutine concurrency.

Operational numbers

  • 1 billion events/second ingested by Cloudflare's data pipeline.

  • 110 million Logpush records/second delivered by Bufstream.

  • Hundreds of Kubernetes pods; each claims a time slice from Buftee.
  • ~3 ms to run a prepared Transformer over a 5,000-row synthetic http_requests batch from one Go goroutine ≈ 1.6M rows/s.
  • 64 KiB-irrelevant here; this is NDJSON/Arrow, not block storage.
  • Transformers available on Free, Pro, Business, and Enterprise plans.

Caveats / what's not disclosed

  • No end-to-end latency figures for a Transformer-attached job (only the isolated 1.6M rows/s micro-benchmark).
  • No numbers on the memory cost of the Arrow IPC copy at the boundary, or on cgo call overhead at 110M records/s.
  • The "multi-layer cache" for Transformer SQL is named but not detailed (layers, invalidation on version change).
  • Auto-disable threshold ("an extended period") is not quantified.
  • Each Transformer update creates a new version (reviewable over time) but version-migration/rollback mechanics are not described.
  • The prototype's WebAssembly-in-Workers approach is described as abandoned for the hot-path concern, not benchmarked against.

Source

Last updated · 854 distilled / 2,448 read