dfe-transform-vrl — Design Document

Overview

dfe-transform-vrl is an embedded VRL transform engine for Kafka-to-Kafka pipelines. Unlike dfe-transform-vector (which runs Vector as a subprocess), this service embeds the VRL crate directly and owns the Kafka consumer/producer, giving full control over memory, backpressure, and wire format.

Problem Statement

Memory Control

Vector.dev has no configurable memory cap. Its internal buffers grow unbounded based on throughput. In K8s autoscaling, pod memory limits must be set to the maximum Vector might ever use, resulting in 2-4x memory waste compared to DFE services that control their own buffers.

Architecture

Data Flow

Per-Event Processing

Kafka partition message (raw bytes)
  │
  ├─ Parse: sonic_rs::from_slice::<Value>() -- a record that is not JSON is dropped
  │
  ├─ Run VRL program(s) on Value
  │   └─ VRL operates directly on Value — no format conversion
  │
  ├─ Serialise: sonic_rs::to_vec(&value)
  │
  └─ KafkaTransport::send() (scalo) → delivery future

Offset Commit Strategy (At-Least-Once)

Release is batch-level. scalo's batch engine hands a block of records to the transform, sends the result, and releases the block's source offsets once the batch send returns Ok. Under that, scalo's Kafka transport holds each source offset it handed out and commits a partition only up to its lowest unreleased one, so the cumulative commit never passes a record not yet released.

If the process crashes between send and release, the whole block is re-consumed and re-processed (at-least-once, not exactly-once). VRL transforms must be idempotent.

Payload Format

JSON is the only payload format. MessagePack, supported in DFE/XDR 2.0 and 2.1, is deprecated in DFE 2.2 and no longer accepted: the JSON path (SIMD parsing with sonic-rs, zstd on the wire) is fast enough that MessagePack gave no CPU saving.

A record that does not parse as JSON is removed from its block, counted in records_error_total{stage="deserialise"}, and logged (sampled) with the parse error. The block's source is then released Dropped, so the record is not redelivered.

VRL Integration

Compilation

VRL programs are compiled once at startup:

let program = vrl::compiler::compile(source, &vrl::stdlib::all())
    .map_err(|diagnostics| /* format errors */)?;

The compiled Program is reused for every event — zero per-event compilation cost.

Execution

Each event is wrapped in a TargetValueRef and passed to the VRL runtime:

let mut target = TargetValueRef {
    value: &mut event_value,
    metadata: &mut metadata,
    secrets: &mut secrets,
};
runtime.resolve(&mut target, &program)?;

VRL's Value type is serde-compatible, so sonic_rs parses JSON bytes directly into it.

Transform File Format

Transform files contain raw VRL source code (not Vector YAML):

# transforms/01_parse.vrl
.parsed = parse_json!(.message)
del(.message)

# transforms/02_enrich.vrl
.environment = .labels.env ?? "unknown"
.processed_at = now()

Files are loaded in sorted order (by filename) and concatenated into a single VRL program. This matches Vector's remap transform behaviour where multiple VRL statements execute sequentially.

vrl is built without its enable_env_functions and enable_network_functions features, and compile_vrl drops get_hostname from the system functions it keeps for parse_etld. A program calling get_env_var, get_hostname, http_request, dns_lookup or reverse_dns fails to compile as a call to an undefined function.

Bundled Pipeline: filebeat-compat (INTERIM)

pipelines/filebeat/ ships a pre-canned, opt-in port of the DFE 2.1 Vector filebeat templates (Cisco Meraki logs, Cisco IOS, Cisco Umbrella) as one pure-VRL file plus its timezones.csv enrichment table. It is convenience data, not engine capability: opting in means pointing the standard transforms.dir and enrichment_tables config at those files. The engine carries no filebeat-specific code; the integration tests (tests/integration/filebeat_pipeline.rs) drive the bundle with the real elastic/integrations pipeline test corpus, which doubles as a full-engine workload.

INTERIM: elastic compatibility is being replaced by dfe-transform-elastic (Rust-native, in beta). Wiring, routing behaviour, known limitations, and regeneration tooling (scripts/filebeat/) are documented in pipelines/filebeat/README.md.

Memory Budget

The wrapper controls all memory allocation:

Component	Configuration	Default
Consumer prefetch	source.max_buffer_bytes	64 MiB
Transform chunk	batch_processing.max_chunk_size (env only)	10 000 events
Producer queue	sink.max_buffer_bytes	64 MiB
Total pod memory	K8s resource limit	512 MiB (128 MiB request)

The sum of consumer + producer buffers must fit within the pod memory limit minus overhead (binary, runtime, VRL compiled programs). The wrapper enforces backpressure: if the producer queue is full, the consumer pauses fetching.

Configuration

Same big-dial pattern as dfe-transform-vector, minus Vector subprocess config:

pipeline:
  name: "my-pipeline"

source:
  brokers: ["kafka:9092"]
  topics: ["raw_events"]
  group_id: "dfe-transform-vrl-my-pipeline"
  max_buffer_bytes: 67108864        # 64 MiB consumer buffer

transforms:
  dir: "/etc/dfe-transform-vrl/transforms/"       # directory of .vrl files

sink:
  brokers: ["kafka:9092"]
  topic: "enriched_events"
  max_buffer_bytes: 67108864        # 64 MiB producer buffer

That is the wrapper's whole schema. metrics, logger, scaling, worker_pool, batch_processing, self_regulation, version_check and otel_tracing belong to scalo's cascade, which reads this file as its settings layer when it is passed with --config, so they may sit beside these sections. The env layer (METRICS_ADDR, LOG_LEVEL, DFE_TRANSFORM_VRL_SCALING__*, ...) outranks the file.

Comparison: dfe-transform-vrl vs dfe-transform-vector

Aspect	dfe-transform-vrl	dfe-transform-vector
Transform engine	VRL crate (in-process)	Vector subprocess
Kafka source/sink	rdkafka (wrapper-controlled)	Vector-managed
Memory control	Full (bounded buffers)	None (Vector unbounded)
Supported transforms	VRL only	All Vector transforms
Container image size	~20 MiB (Rust binary)	~170 MiB (Rust + Vector)
Process model	Single process	Wrapper + subprocess
Startup time	Fast (compile VRL)	Slower (spawn + healthcheck Vector)
Crash recovery	Restart binary	Restart subprocess + binary

When to Use Which

• dfe-transform-vrl: VRL-only transforms (the common case), memory-constrained pods, high-density deployments
• dfe-transform-vector: Pipelines needing Vector-native transforms (lua, aggregate, dedupe, throttle, sample), or complex multi-source/sink routing

Reload

Carried in the reloadable subset

ConfigReloader (file polling + SIGHUP) re-reads and re-validates the file and swaps SharedConfig<HotConfig>, so a bad edit is caught without a restart.

Field	What it would control	State
pipeline.batch_size	Events per transform chunk	not applied -- see below
pipeline.batch_timeout_ms	Max wait before flushing partial batch	not applied -- no such timer
sink.key_field	Kafka partition key path	not applied -- scalo-rs#37

Accepted but not applied

The pipeline holds the reloadable subset but reads none of it yet, so a reload changes no pipeline behaviour today. config::INERT_SETTINGS is the list, and the wrapper warns at startup for each one a deployment has set. The dial that does size a chunk is batch_processing.max_chunk_size, set via the env layer.

Requires pod restart

These fields are bound to connections, compiled programs, or server sockets established at startup. Changing them requires a pod restart (which is the standard K8s pattern — ConfigMap changes trigger rolling restart via the checksum/config annotation in the Deployment template).

Field	Why
pipeline.name	Baked into Kafka group_id, metrics labels, tracing spans
source.*	rdkafka consumer: connection, subscription, auth, TLS, buffers
sink.brokers	rdkafka producer connection
sink.topic	Output topic (changing mid-stream risks data loss)
sink.compression	rdkafka compression.type
sink.sasl.* / sink.tls.*	Security protocol
sink.max_buffer_bytes	rdkafka queue.buffering.max.kbytes
sink.librdkafka_options	librdkafka ClientConfig
transforms.*	VRL programs compiled at startup