diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 6029a98..9761c41 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -31,7 +31,7 @@ versioning and changelog generation. All commits must follow this format: | `build` | Changes to build system or dependencies | None | | `ci` | Changes to CI configuration | None | | `chore` | Other changes that don't modify src or test files | None | -| `revert` | Reverts a previous commit | Varies | +| `revert` | Reverts a previous commit | Patch (0.0.X) | ### Breaking Changes @@ -53,38 +53,18 @@ affected (e.g., `config`, `engine`, `kafka`, `pipeline`). This project follows [Semantic Versioning 2.0.0](https://semver.org/): -- **MAJOR** (X.0.0): Breaking changes that require users to modify their code -- **MINOR** (0.X.0): New features that are backwards-compatible -- **PATCH** (0.0.X): Bug fixes and minor improvements +- MAJOR (X.0.0): breaking changes that require users to modify their code +- MINOR (0.X.0): new features that are backwards-compatible +- PATCH (0.0.X): bug fixes and minor improvements Versions are automatically determined by semantic-release based on commit messages. Do not manually update version numbers. ## Developer Certificate of Origin -This project uses the Developer Certificate of Origin (DCO) to ensure that -contributors have the right to submit their contributions. +This project uses the [Developer Certificate of Origin 1.1](https://developercertificate.org/). A `Signed-off-by` line on a commit certifies, for that contribution, the four statements in that text. -By making a contribution to this project, you certify that: - -1. The contribution was created in whole or in part by you and you have the - right to submit it under the license indicated in the file; or - -2. The contribution is based upon previous work that, to the best of your - knowledge, is covered under an appropriate open source license and you - have the right under that license to submit that work with modifications; - or - -3. The contribution was provided directly to me by some other person who - certified (1), (2) or (3) and you have not modified it. - -4. You understand and agree that this project and the contribution are public - and that a record of the contribution (including all personal information - you submit with it, including your sign-off) is maintained indefinitely - and may be redistributed consistent with this project or the license(s) - involved. - -## How to Sign Off Your Commits +### How to Sign Off Your Commits You must sign off each commit to indicate your acceptance of the DCO. Combine the signoff with your conventional commit message: @@ -118,11 +98,11 @@ Change License on the Change Date specified in [LICENSE](LICENSE). ## How to Contribute -1. **Fork the repository** and create your branch from `main` -2. **Make your changes** following the commit message format above -3. **Sign off your commits** with the DCO -4. **Test your changes** to ensure they work as expected -5. **Submit a pull request** with a clear description of what you've done +1. Fork the repository and create your branch from `main` +2. Make your changes following the commit message format above +3. Sign off your commits with the DCO +4. Run the tests that cover your change +5. Submit a pull request with a clear description of what you've done ### Pull Request Checklist @@ -145,6 +125,10 @@ cargo build --release cargo nextest run ``` +## CI/CD Workflow + +A merge to `main` runs CI and releases nothing by itself. A release is cut when the squash-merge commit ends with a `Release: true` trailer, or when a maintainer dispatches the CI workflow with `from-head: true`. semantic-release then reads the commit messages since the last release and picks the next version, the release commit updates `CHANGELOG.md` and `VERSION`, and hyperi-ci builds and publishes the release. + ## Questions If you have questions about contributing, please open an issue or contact us. @@ -174,30 +158,29 @@ turn. | `src/enrichment/` | Table loading (CSV, JSON, YAML, MMDB, STIX, SQLite), refresh, the custom VRL functions | | `src/deployment.rs` | `contract()` -- the single source the Dockerfile and the released chart are generated from | | `Dockerfile` | Generator output, not hand-authored | -| `pipelines/filebeat/` | Opt-in data bundle (212,218 bytes of VRL plus a lookup table), not engine capability | -| `tests/` | `integration/` runs mostly without Kafka, `e2e/` needs a broker, `TESTING.md` explains the modes | +| `pipelines/filebeat/` | Opt-in data bundle (one VRL file plus a lookup table), not engine capability | +| `tests/` | `integration/` runs without a broker, `e2e/` needs a broker container or a live broker, `TESTING.md` explains the tiers | | `docs/architecture.md` | Why the service is built this way, and the invariants | ### Commands that prove a change ```bash make check # hyperi-ci check -- quality + test, the pre-push gate -cargo nextest run --features enrichment-mmdb,enrichment-sqlite # what CI actually runs -cargo nextest run --all-features --run-ignored # adds the ignored Kafka e2e tests -cargo run -- config-check --config config.yaml # validate a config without starting +cargo nextest run --features enrichment-mmdb,enrichment-sqlite # the feature set CI tests with +cargo nextest run --all-features --run-ignored all --ignore-default-filter # the full tier, ignored tests included +cargo run --bin dfe-transform-vrl -- --config config.yaml config-check # validate a config without starting cargo run --bin dfe-transform-vrl -- emit-dockerfile > Dockerfile # regenerate the Dockerfile ``` +`cargo run` needs `--bin`, because `pgo-driver` is a second binary, and `--config` goes before the subcommand. + Green lies here in three ways, and all three are on by default. `default = []` in `Cargo.toml`, so a bare `cargo nextest run` does not compile the MMDB or SQLite enrichment tests in at all. `.hyperi-ci.yaml` adds `enrichment-mmdb` and `enrichment-sqlite` to both the test and the build feature sets -- the build too, because the Dockerfile only copies the binary, so a default build ships a container that rejects `type: mmdb` and `type: sqlite` tables at startup. -Kafka-dependent tests call `skip_if_no_kafka!()` and skip cleanly when no broker -is reachable. They report as not-failed, which is not the same as proven. +The core-tier e2e tests run on broker containers scalo's test support starts. Without Docker they skip locally and fail only under `CI`, so a local run without Docker is green without having reached a broker. -The Kafka e2e tests are `#[ignore]` and need `--run-ignored` plus a broker, and the other e2e tests run by default and start their own containers. Set -`TEST_MODE=docker` for the dfe-docker infra profile on `localhost:19092`, or -`TEST_MODE=remote` for the cluster endpoints. +The tests that may reach a broker somebody else owns are `#[ignore]` and run only under `--run-ignored all` or `only`. They take a live broker first when `KAFKA_BROKERS` answers. `TEST_MODE=docker` points the harness at `localhost:19092` with no auth, the port dfe-docker publishes its broker on, and otherwise it reads `KAFKA_BROKERS`, `KAFKA_SECURITY_PROTOCOL` and the `KAFKA_SASL_*` names from the environment or the repo's `.env`. One more: the push trigger in `.github/workflows/ci.yml` carries `paths-ignore: docs/**, **.md`, so a docs-only push runs no jobs. The @@ -208,32 +191,34 @@ checked. | Don't | Do | Why | |---|---|---| -| Hand-edit `Dockerfile`, or commit a chart | Fix `src/deployment.rs::contract()` and regenerate | The Dockerfile is generator output and the release assembles the chart from the contract, so a hand edit is reverted or never ships, and `test_every_contract_secret_env_var_reaches_the_config` holds every declared name to the field it spells | +| Hand-edit `Dockerfile`, or commit a chart | Fix `src/deployment.rs::contract()` and regenerate | The Dockerfile is generator output, and `checked_in_dockerfile_matches_emit_dockerfile` fails on a hand edit. The release assembles the chart from the contract, so a committed chart never ships, and `test_every_contract_secret_env_var_reaches_the_config` holds every declared name to the field it spells | | Bump scalo and leave `release.helm.library` behind | Move `release.helm.library` in `.hyperi-ci.yaml` to the same scalo version | A scalo-service release renders only the contract version its scalo release writes | -| Turn the Kafka lag trigger back on in `contract()` | Leave `KafkaLagTrigger::disabled()` and scale on CPU plus scaling pressure | Lag rises when a downstream stage breaks, so a lag trigger adds pods that wait on the same broken stage. The lag trigger was also the one that kept reading a kafka block this app's values do not have | -| Put a scalo section (`metrics`, `logger`, `scaling`, `worker_pool`, `batch_processing`, `self_regulation`, `version_check`, `otel_tracing`) in a `config.yaml` the binary finds in its working directory | Pass the file with `--config`, or set the section through the env layer on a **double** underscore | scalo reads the `--config` file as its settings layer but finds other files only by fixed base name (`settings.yaml`, `defaults.yaml`), so a working-directory `config.yaml` reaches the wrapper alone. Full list under [Environment variable overrides](docs/internals.md#environment-variable-overrides) | +| Turn the Kafka lag trigger back on in `contract()` | Leave `KafkaLagTrigger::disabled()`, so the contract scales on CPU and a deployment adds the scaling-pressure trigger | Lag rises when a downstream stage breaks, so a lag trigger adds pods that wait on the same broken stage. The lag trigger also reads `config.kafka.*`, which this app's values do not have | +| Put a scalo section (`metrics`, `logger`, `memory`, `scaling`, `worker_pool`, `batch_processing`, `self_regulation`, `version_check`, `otel_tracing`) in a `config.yaml` the binary finds in its working directory | Pass the file with `--config`, or set the section through the env layer on a **double** underscore | scalo reads the `--config` file as its settings layer but finds other files only by fixed base name (`settings.yaml`, `defaults.yaml`), so a working-directory `config.yaml` reaches the wrapper alone. Full list under [Environment variable overrides](docs/internals.md#environment-variable-overrides) | | Trust `pipeline.batch_size`, `pipeline.batch_timeout_ms`, `sink.key_field` or `source.commit_interval_ms` | Size a chunk with `batch_processing.max_chunk_size` | `config::INERT_SETTINGS` -- accepted, validated, reaching nothing. Table under [Reload](docs/internals.md#reload) | | Call a bare `cargo nextest run` green | Pass `--features enrichment-mmdb,enrichment-sqlite` | `default = []`, so the MMDB and SQLite tests are not compiled in and the run is green without having tested them | -| Set `sasl.enabled` with an empty username or password | Supply both, or neither | librdkafka's SCRAM check is a NULL check that an empty string passes, so that pod authenticated against nothing and still reported Ready. Refused at startup now | -| Add a serialise path that redacts by field name | Keep the password a `scalo::SensitiveString` | Redaction is by type on every path -- `Debug`, the `/config` dump, the emitted schema. The figment round-trip has to be wrapped in `expose_during` or a file-sourced password reaches the broker as the literal `***REDACTED***` | -| Size the container from steady state when the program is large | Leave headroom for the compile | Compilation scales with the VRL, and the bundled filebeat program is OOM-killed under a 32 MiB limit. Below the floor the kernel kills the process mid-compile, seen as an exit-137 restart loop with nothing in the log. `engine::budget` refuses first and names all three numbers | +| Set `sasl.enabled` with an empty username or password | Supply both, or neither | librdkafka's SCRAM check is a NULL check that an empty string passes, so the pod would authenticate against nothing and still report Ready. `validate` refuses it at startup | +| Add a serialise path that redacts by field name | Keep the password a `scalo::SensitiveString` | Redaction is by type on every path -- `Debug`, the `config-check` dump, the emitted schema. The figment round-trip has to be wrapped in `expose_during` or a file-sourced password reaches the broker as the literal `***REDACTED***` | +| Size the container from steady state when the program is large | Leave headroom for the compile | Compilation scales with the VRL, and below the floor `engine::budget` computes the kernel kills the process mid-compile, seen as an exit-137 restart loop with nothing in the log. `engine::budget` refuses first and names all three numbers | | Pick the optimisation tier with `publish-target` | Use `build.skip_optimize` in `.hyperi-ci.yaml`, or the `skip-optimize` dispatch input for one run | hyperi-ci ignores `publish-target`. A build that ships gets PGO and BOLT from `scripts/pgo-workload.sh` unless optimisation is skipped, and a stable release that skips it is refused unless the dispatch passes `release-unoptimized: true` | ### Where this sits Inbound -- what this repo depends on: -- **scalo-rs** (`cargo-dep`). `Cargo.toml` declares the `scalo` crate by range, a +- scalo-rs (`cargo-dep`). `Cargo.toml` declares the `scalo` crate by range, a second range covering the dev dependency. A scalo release arrives through that range: `cargo update -p scalo` and rebuild if it admits the version, widen the range first if not. -- **scalo-rs** (`generated-file`, lockstep). The `Dockerfile` is written by - scalo's generator from this crate's `deployment::contract()`, at contract - schema version 4. A generator or schema change upstream means regenerating - with the command in the file's own header and committing the diff. The - released chart is assembled on the scalo-service library chart at - `release.helm.library`, which moves with the scalo version in `Cargo.toml`. -- **dfe-infra** (`apps.yaml`, deploy-time authority). The suite manifest declares +- scalo-rs (`generated-file`, lockstep). The `Dockerfile` is written by + scalo's generator from this crate's `deployment::contract()`, and its header + names the generator and the contract schema version. A generator or schema + change upstream means regenerating with + `cargo run --bin dfe-transform-vrl -- emit-dockerfile > Dockerfile` and + committing the diff. The released chart is assembled on the scalo-service + library chart at `release.helm.library`, which moves with the scalo version in + `Cargo.toml`. +- dfe-infra (`apps.yaml`, deploy-time authority). The suite manifest declares what this app is: `multiplicity: per_config` (one deployment per source config, never a singleton), `scale_deployed: true`, both transports, and a source binding deriving `{source}_land` in, `{source}_load` out and @@ -242,10 +227,7 @@ Inbound -- what this repo depends on: Outbound -- what depends on this repo: -- **dfe-infra** (`image-pin`, lockstep). Pins this repo's ghcr image as a tag plus +- dfe-infra (`image-pin`, lockstep). Pins this repo's ghcr image as a tag plus the digest that makes it immutable. A release here means bumping the tag and re-resolving the digest there, with `check_versions_drift.py` confirming the chart's appVersion and the digest mirror agree. - -dfe-transform-splack runs this repo's engine image with a rule-config chart of -its own. It is out of suite scope and no work here is driven by it. diff --git a/README.md b/README.md index 91b57a8..2e81d1b 100644 --- a/README.md +++ b/README.md @@ -11,19 +11,24 @@ Runs your VRL transforms over a DFE deployment's event stream, in-process. - Kafka in and out, or a direct gRPC push on a deployment with no broker. The transform you write is the same on both. - At-least-once by default: the source is acknowledged only once the sink has the transformed records. `source.acknowledgements.enabled: false` acknowledges at receipt instead. - Enrichment tables from CSV, JSON, YAML and STIX, read from VRL, plus MMDB and SQLite with the `enrichment-mmdb` and `enrichment-sqlite` build features. A table with a `refresh` setting reloads while the service runs. MMDB loads once at start. -- The host and network functions (`get_env_var`, `get_hostname`, `http_request`, `dns_lookup`, `reverse_dns`) do not compile, so a transform cannot reach outside the process. +- The host and network functions (`get_env_var`, `get_hostname`, `http_request`, `dns_lookup`, `reverse_dns`) do not compile, so a transform cannot read the host's environment or reach the network. JSON is the only payload format. A record that is not JSON is dropped, counted in `records_error_total{stage="deserialise"}`, and not redelivered. For Vector's own transforms beyond `remap`, use [dfe-transform-vector](https://github.com/hyperi-io/dfe-transform-vector). ## Quick start +On Linux the build needs `protoc`, a C toolchain, `pkg-config` and librdkafka. scalo compiles its gRPC `.proto` files, `aws-lc-sys` compiles C, and `rdkafka-sys` links librdkafka through pkg-config. It refuses a librdkafka older than the version after the `+` in its `Cargo.lock` entry, and a Debian or Ubuntu package can be older than that, so there [build librdkafka from source](https://github.com/confluentinc/librdkafka#build-from-source) first. + ```bash +sudo apt-get install -y protobuf-compiler build-essential pkg-config cargo build --release -cp config.example.yaml config.yaml # set the brokers, topics and transforms.dir -./target/release/dfe-transform-vrl config-check --config config.yaml +cp config.example.yaml config.yaml # uncomment source, sink and transforms, then set the brokers, topics and transforms.dir +./target/release/dfe-transform-vrl --config config.yaml config-check ./target/release/dfe-transform-vrl --config config.yaml ``` +`--config` goes before the subcommand, or `config-check` refuses it as an unexpected argument. + The smallest config names the source, a directory holding at least one `.vrl` file, and the sink: ```yaml @@ -53,9 +58,11 @@ del(.message) Health and metrics answer on port 9090: `/livez`, `/readyz` (503 while memory is under pressure or a transport reports itself unhealthy, and 200 once the pipeline runs, an idle one included) and `/metrics`. +Outside the image, set `dlq.file.path` to a writable directory. The default, `/var/spool/scalo/dlq`, is one the image creates, and without a writable one the service logs that the DLQ has no backend and holds any dead letter unacknowledged at its source. + ## Configuration -[config.example.yaml](config.example.yaml) is the annotated reference, and [docs/config-schema.yaml](docs/config-schema.yaml) lists every key with its type and default. The big dials have flat environment overrides for Kubernetes, `DFE_TRANSFORM_VRL_SOURCE_BROKERS`, `DFE_TRANSFORM_VRL_SINK_TOPIC`, `DFE_TRANSFORM_VRL_TRANSFORMS_DIR` and the SASL pair among them. The whole list, Kafka over TLS with a private CA and the scalo sections that share the file on a double underscore are under [Operating notes](docs/internals.md#operating-notes) in docs/internals.md, and the settings that parse but reach nothing under [Reload](docs/internals.md#reload). +[config.example.yaml](config.example.yaml) is the annotated reference, and [docs/config-schema.yaml](docs/config-schema.yaml) lists every key of the transform's own sections with its type and default. The big dials have flat environment overrides for Kubernetes, `DFE_TRANSFORM_VRL_SOURCE_BROKERS`, `DFE_TRANSFORM_VRL_SINK_TOPIC`, `DFE_TRANSFORM_VRL_TRANSFORMS_DIR` and the SASL pair among them. The whole list, Kafka over TLS with a private CA and the scalo sections that share the file on a double underscore are under [Operating notes](docs/internals.md#operating-notes) in docs/internals.md, and the settings that parse but reach nothing under [Reload](docs/internals.md#reload). `pipelines/filebeat/` ships a pure-VRL port of the DFE 2.1 filebeat templates (Cisco Meraki, IOS, Umbrella) with its `timezones.csv` table. Opt in by pointing `transforms.dir` and `enrichment_tables` at it. dfe-transform-elastic carries the same sources as compiled Rust. diff --git a/SECURITY.md b/SECURITY.md index 8d4fe10..0a2d3ce 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -17,12 +17,10 @@ Please include: ## What to Expect -- **Acknowledgment**: We will acknowledge receipt of your report within - 5 business days -- **Investigation**: We will investigate and keep you informed of our progress -- **Resolution**: We will work to resolve confirmed vulnerabilities promptly -- **Disclosure**: We will coordinate with you on an appropriate disclosure - timeline +- We will acknowledge receipt of your report within 5 business days +- We will investigate and keep you informed of our progress +- We will work to resolve confirmed vulnerabilities promptly +- We will coordinate with you on an appropriate disclosure timeline ## Safe Harbour @@ -39,12 +37,12 @@ We will not pursue legal action against security researchers who: ## Recognition With your permission, we will credit you for the discovery of confirmed -vulnerabilities. We do not currently offer monetary bounties, but we value +vulnerabilities. We do not offer monetary bounties, but we value and appreciate responsible disclosure. ## Out of Scope -The following are generally out of scope: +These are generally out of scope: - Social engineering or phishing attacks - Denial of service (DoS/DDoS) attacks diff --git a/config.example.yaml b/config.example.yaml index e9260ed..433719a 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1,5 +1,5 @@ # ============================================================================= -# dfe-transform-vrl — example configuration +# dfe-transform-vrl -- example configuration # ============================================================================= # # Copy to config.yaml and modify as needed. @@ -7,7 +7,8 @@ # Configuration cascade (highest to lowest priority): # 1. CLI args (--config, --log-level, etc.) # 2. Environment variables (DFE_TRANSFORM_VRL_*) -# 3. .env file in the working directory (never a parent's) +# 3. .env file in the working directory (never a parent's), read only when +# no --config is given # 4. Config file (this file) # 5. Hard-coded defaults @@ -18,8 +19,8 @@ pipeline: name: "my-pipeline" # NOT APPLIED. Parsed and validated, then read by nothing -- the batch engine # takes its chunk size from batch_processing.max_chunk_size (see the bottom of - # this file), and the governed driver has no partial-batch timer. Setting - # either logs a warning at startup. + # this file), and the governed driver has no partial-batch timer. Changing + # either from its default logs a warning at startup. batch_size: 1000 batch_timeout_ms: 100 @@ -94,7 +95,7 @@ pipeline: # compression: "zstd" # none, gzip, lz4, snappy, zstd # max_buffer_bytes: 67108864 # 64 MiB producer buffer # # Producer delivery timeout. Capped at 15000 while a direct source holds -# # its acknowledgements, so the report lands inside the 18 s hold; set +# # its acknowledgements, so the report lands inside the 18 s hold. Set # # librdkafka_options message.timeout.ms to override the cap. # message_timeout_ms: 300000 # sasl: @@ -126,7 +127,7 @@ pipeline: # --- OPTIONAL: bundled DFE 2.1 filebeat-compat pipeline (INTERIM) ------------ # Pre-canned pure-VRL port of the DFE 2.1 filebeat templates (Cisco Meraki / -# IOS / Umbrella). Being replaced by dfe-transform-elastic (beta) - see +# IOS / Umbrella). Being replaced by dfe-transform-elastic - see # pipelines/filebeat/README.md before adopting. The stock image ships no # pipelines/ directory - copy or mount the two files from the repo's # pipelines/filebeat/ first, then point the generic knobs at them: @@ -146,7 +147,7 @@ pipeline: # one match errors, and the table name must be a literal naming a table listed # here or the program fails to compile. # -# key_columns index the lookup; a condition on any other column still filters, +# key_columns index the lookup. A condition on any other column still filters, # it just does not use the index. # # schema coerces text columns, which is what makes a numeric comparison or a @@ -175,12 +176,13 @@ pipeline: # ----------------------------------------------------------------------------- # scalo's own settings -- /livez and /readyz are served by the metrics listener # ----------------------------------------------------------------------------- -# metrics, logger, scaling, worker_pool, batch_processing, self_regulation, -# version_check and otel_tracing belong to scalo's config cascade, which reads -# the file passed with --config as its settings layer, once at startup. They -# take effect from this file when it is passed that way. A config.yaml found in -# the working directory without --config reaches the wrapper only, and the -# wrapper warns at startup for each scalo section in it. +# metrics, logger, memory, scaling, worker_pool, batch_processing, +# self_regulation, version_check and otel_tracing belong to scalo's config +# cascade, which reads the file passed with --config as its settings layer, once +# at startup. They take effect from this file when it is passed that way. A +# config.yaml found in the working directory without --config reaches the +# wrapper only, and the wrapper warns at startup for each of those sections in +# it but memory. # # scaling: # memory_gate_threshold: 0.8 @@ -209,5 +211,5 @@ pipeline: # DFE_TRANSFORM_VRL_METRICS__OTEL__ENABLED=false # # batch_processing.max_chunk_size is the events-per-chunk dial the batch engine -# actually reads. pipeline.batch_size above is NOT it -- see the "accepted but -# not applied" list in the README. +# actually reads. pipeline.batch_size above is NOT it -- see Reload in +# docs/internals.md. diff --git a/docs/README.md b/docs/README.md index 82b599e..892c7ae 100644 --- a/docs/README.md +++ b/docs/README.md @@ -3,7 +3,7 @@ The architecture and design of dfe-transform-vrl. - [Architecture](architecture.md) -- [dfe-transform-vrl -- Design Document](internals.md) +- [Internals](internals.md) - [config-schema.yaml](config-schema.yaml) -- the configuration as a JSON Schema (YAML) - [config-schema.json](config-schema.json) -- the configuration as a JSON Schema (JSON) - [capability-catalog.yaml](capability-catalog.yaml) -- what the app can do, as a catalogue (YAML) diff --git a/docs/architecture.md b/docs/architecture.md index 191e11a..b92bbb2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -12,8 +12,6 @@ That cost buys access to Vector's non-VRL transforms, which most pipelines never use. So the answer is to keep VRL and drop Vector. Compile the VRL crate into the binary, and let the wrapper own the buffers and the wire format. -### What it is, and what it is not - One Rust binary. It reads records, runs a compiled VRL program over each one, and writes the results out. Kafka in and out on the `bus` transport, gRPC in and out on the `direct` transport (`0.0.0.0:6000` listener, default sink endpoint @@ -46,12 +44,12 @@ flowchart TB SINK["produce sink"] COMP["VRL compiler + enrichment registry"] end - DRV -->|"WorkBatch of Record"| PROC - PROC -->|"out-batch"| SINK + DRV -->|WorkBatch of Record| PROC + PROC -->|out-batch| SINK SINK --> DRV - COMP -.->|"compiled Program, once at startup"| PROC - POOL -.->|"parallel VRL execution"| PROC - GOV -.->|"pauses intake under memory pressure"| DRV + COMP -.->|compiled Program, once at startup| PROC + POOL -.->|parallel VRL execution| PROC + GOV -.->|pauses intake under memory pressure| DRV ``` Per block of records: @@ -61,9 +59,9 @@ Per block of records: pauses under memory pressure by pausing partitions -- the member stays in the consumer group and no rebalance fires. On the direct transport it is the Push listener, which refuses pushes `UNAVAILABLE` under the same pressure. -2. `process` deserialises each record into a VRL `Value` (format auto-sensed), - runs the compiled program on the worker pool in parallel, and reserialises - survivors back to the format they arrived in. +2. `process` parses each record's JSON into a VRL `Value`, runs the compiled + program on the worker pool in parallel, and serialises the survivors back to + JSON. 3. The loop sends the whole out-batch through `TransportSender::send_batch`. 4. The loop releases the block's source only after the send returns `Ok`: Kafka commits the offsets, the Push listener answers its sender OK. @@ -96,21 +94,22 @@ into a single program. Changing a transform means rolling the pod -- hence recompile path at runtime. **The compile memory floor is checked before the compiler runs.** Compilation is -the one startup cost that scales with the deployment's own data, and the bundled -filebeat pipeline is 212,218 bytes of VRL. Below a certain container limit the -kernel kills the process mid-compile, which an operator sees as an exit-137 -restart loop with nothing in the log naming the cause. `engine::budget` reads the -limit first and refuses with all three numbers -- program size, limit, required -floor -- in one message. The constants come from measurement on the shipped -image, not from theory: 96 resident bytes per source byte over a 24 MiB reserve. +the one startup cost that scales with the deployment's own data. Below a certain +container limit the kernel kills the process mid-compile, which an operator sees +as an exit-137 restart loop with nothing in the log naming the cause. +`engine::budget` reads the limit first and refuses with all three numbers -- +program size, limit, required floor -- in one message. Its two constants, a +reserve and a cost per source byte, come from measurement on the shipped image, +and their doc comments name the measured points. **Idle is a valid, Ready state.** A transform instance is deployed before its program exists -- whatever creates it writes the config naming its topics, and -the VRL arrives afterwards. With no topics or no program the app starts, reports -Ready, holds `pipeline_idle` at 1, joins no consumer group, and logs the reason -once. scalo's idle gate re-polls, so the program that lands later starts the -service without a restart. Only reaching the loader with work still to do is a -failure. +the VRL arrives afterwards. With no topics on the bus, or no program, the app +starts, reports Ready, holds `pipeline_idle` at 1, joins no consumer group, and +logs the reason once. scalo's idle gate re-reads the config when the config file +changes or the process gets a SIGHUP, and a program written into +`transforms.dir` alone does not wake it. In Kubernetes a new program is a change +to the transforms file set, which rolls the pod. **Metric names are emitted bare.** The `MetricsManager` namespace is empty unless `metrics.namespace` is set, and none is by default. A set namespace is @@ -128,52 +127,53 @@ compile time -- which is why a table cannot appear after startup. ## Two config authorities, one config file The `--config` file plus `DFE_TRANSFORM_VRL_*` env vars resolve this crate's own -`Config`: `pipeline`, `source`, `sink`, `transforms`, `enrichment_tables`. +`Config`: `pipeline`, `source`, `sink`, `transforms`, `enrichment_tables`, `dlq`. -scalo resolves its own sections -- `metrics`, `logger`, `scaling`, +scalo resolves its own sections -- `metrics`, `logger`, `memory`, `scaling`, `worker_pool`, `batch_processing`, `self_regulation`, `version_check`, -`otel_tracing` -- from a separate cascade, which the wrapper seeds with the -`--config` file as its settings layer. So the mounted config carries both, each -authority taking the sections it owns. The env layer outranks the file for both, -and there a scalo section nests on a double underscore. A `config.yaml` read -from the working directory with no `--config` reaches the wrapper only, so the -wrapper warns at startup for each scalo section in it, rather than letting the -deployment believe it took. +`otel_tracing` -- from a separate cascade, which scalo's `run_app` sets up with +the `--config` file as its settings layer before the wrapper loads its own. So +the mounted config carries both, each authority taking the sections it owns. The +env layer outranks the file for both, and there a scalo section nests on a +double underscore. Only `--config` names the file: `DFE_TRANSFORM_VRL_CONFIG` +reaches neither cascade. A `config.yaml` read from the working directory with no +`--config` reaches the wrapper only, so the wrapper warns at startup for each +section `config::SCALO_CASCADE_SECTIONS` lists, rather than letting the +deployment believe it took. That list has no `memory`, so a `memory` section +there goes unapplied without a warning. A second list, `config::INERT_SETTINGS`, covers fields this crate accepts and validates but does not act on, each with the reason and the replacement where one exists. `HotConfig` and `ConfigReloader` re-read and re-validate the file on -change or SIGHUP, so a bad edit is caught without a restart, but every field the -reloadable subset carries is on the inert list -- no pipeline behaviour changes -on reload. +change or SIGHUP, so a bad edit logs `Config reload failed, keeping current +config` without a restart, but every field the reloadable subset carries is on +the inert list -- no pipeline behaviour changes on reload. ## Components | Path | Holds | |---|---| -| `src/main.rs`, `src/cli.rs` | Process entry, subcommands, env seeding for scalo's cascade | +| `src/main.rs`, `src/cli.rs` | Process entry, subcommands, the scalo `ServiceApp` wiring | | `src/pipeline.rs` | The `process` closure and sink handed to scalo's batch engine | | `src/engine/compiler.rs` | Loads and compiles transform files into one program | | `src/engine/runner.rs` | Runs a compiled program against one event `Value` | | `src/engine/budget.rs` | Compile memory floor, checked before compiling | | `src/config/` | `Config`, validation, `HotConfig`, the inert and cascade lists | | `src/enrichment/` | Table loading (CSV, JSON, YAML, MMDB, STIX, SQLite), refresh, the VRL functions | -| `src/kafka/mod.rs` | Wraps scalo's `KafkaTransport` and `FormatDetector` | +| `src/kafka/mod.rs` | The consumer and producer configs for scalo's `KafkaTransport`, and the DFE topic and group names | | `src/deployment.rs` | `contract()` -- the source the Dockerfile and the released chart are generated from | | `src/metrics.rs` | Three metric layers, bare names | | `pipelines/filebeat/` | Opt-in data bundle, not engine capability | ## Deployment -### Generated artefacts - -The `Dockerfile` is generated from `deployment.rs::contract()` by scalo's generator, and carries a header naming the generator and the regeneration command. A hand edit is reverted by the next regeneration, so a fix belongs in `contract()`. +The `Dockerfile` is generated from `deployment.rs::contract()` by scalo's generator, and carries a header naming the generator and the contract schema version. `emit-dockerfile` regenerates it, and `checked_in_dockerfile_matches_emit_dockerfile` fails on a hand edit, so a fix belongs in `contract()`. -No chart is committed. At release, hyperi-ci emits the contract with `generate-artefacts` and assembles a thin chart from it on the scalo-service library chart at `release.helm.library`. The contract is schema version 4: a 120 s startup budget, `kubernetes.io/h2c` on the Push port, the Kafka Secret mounted into the per-endpoint SASL names, no writable path beyond the library's scratch `/tmp` under a read-only root, requests 100m/128Mi, limits 500m/512Mi and a 45 s grace period. +No chart is committed. At release, hyperi-ci emits the contract with `generate-artefacts` and assembles a thin chart from it on the scalo-service library chart at `release.helm.library`. `dfe-transform-vrl emit-contract` prints the contract: the startup budget, `kubernetes.io/h2c` on the Push port, the Kafka Secret mounted into the per-endpoint SASL names and the shared mechanism name, the DLQ's directory as the one writable path under a read-only root beside the library's scratch `/tmp`, the requests and limits, and the grace period. -The Push port (6000) is gated on `config.source.transport` being `direct` or `grpc`, the two names of the only transport that binds the listener, so the bus default (`bus` or `kafka`) publishes no port nothing answers on. The contract's ScaledObject scales on CPU alone (80%): consumer-group lag rises when a downstream stage breaks, so it is not a trigger. A deployment adds a trigger with the library's `keda.extraTriggers` value. +The Push port (6000) is gated on `config.source.transport` being `direct` or `grpc`, the two names of the only transport that binds the listener, so the bus default (`bus` or `kafka`) publishes no port nothing answers on. The contract's ScaledObject scales on CPU alone: consumer-group lag rises when a downstream stage breaks, so it is not a trigger. A deployment adds a trigger with the library's `keda.extraTriggers` value. -In a DFE deployment, dfe-infra's values set the CPU target to 70% in every profile. The single, scale and mesh profiles add the scaling-pressure trigger (target 70, read through dfe-keda-shim), and slim runs on CPU alone. The engine's [KEDA scaling page](https://github.com/hyperi-io/dfe-engine/blob/main/docs/deployment/keda-scaling.md) covers the triggers for every app. +In a DFE deployment, dfe-infra's values set the CPU target in every profile. The single, scale and mesh profiles add the scaling-pressure trigger, read through dfe-keda-shim, and slim runs on CPU alone. The engine's [KEDA scaling page](https://github.com/hyperi-io/dfe-engine/blob/main/docs/deployment/keda-scaling.md) covers the triggers for every app. ### How many instances run @@ -187,8 +187,8 @@ group. Adding a source is a manifest edit, not a change here. ## Not covered here `docs/internals.md` carries the deeper design detail: the vrl-versus-vector -comparison in full, format detection behaviour, the memory budget table, and the -field-by-field reload matrix. +comparison in full, the per-record parse and serialise path, the memory budget +table, and the field-by-field reload matrix. Two things are known gaps rather than omissions. Readiness reads scalo's health registry, so a broker refusing the credentials, a missing topic or a consumer diff --git a/docs/internals.md b/docs/internals.md index 2f47ad1..7bf48a5 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -1,105 +1,54 @@ -# dfe-transform-vrl -- Design Document +# Internals -## Overview +How a record moves through dfe-transform-vrl, and the settings behind each step. [architecture.md](architecture.md) covers why the service is built this way and the invariants it keeps. -`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. +## Data flow -## 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 +One deployment runs one transport. On the bus, records come from a Kafka consumer and go to a Kafka producer. On the direct transport, they arrive on a Push listener and go on through a gRPC client to the next stage's listener. ```mermaid -flowchart TB - subgraph DP["Data path"] - direction LR - KS["Kafka source (scalo)
JSON"] -->|consume| VRL["VRL engine
Value in/out"] - VRL --> KP["Kafka sink (scalo)
JSON"] - KP -.->|offset commit after delivery, at-least-once| KS +flowchart LR + subgraph SRC["Source"] + KC["Kafka consumer
bus"] + PL["Push listener
direct"] end - subgraph OPS["Operational endpoints (same process)"] - direction LR - MS["Metrics :9090
/metrics, /livez, /readyz"] + subgraph PROC["process closure"] + PARSE["sonic-rs parse
JSON to Value"] + VRL["compiled VRL program"] + SER["sonic-rs serialise
Value to JSON"] end + subgraph SNK["Sink"] + KP["Kafka producer
bus"] + GC["gRPC client
direct"] + end + KC -->|WorkBatch| PARSE + PL -->|WorkBatch| PARSE + PARSE -->|Value| VRL + VRL -->|survivors| SER + SER -->|send_batch| KP + SER -->|send_batch| GC ``` -## Data Flow - -### Per-Event Processing - -```text -Kafka partition message (raw bytes) - │ - ├─ Parse: sonic_rs::from_slice::() -- 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. +Per record: -### Payload Format +1. `sonic_rs::from_slice` parses the payload straight into a VRL `Value`, which is serde-compatible. A record that is not JSON, or nests deeper than sonic-rs accepts, is dropped. +2. The compiled program runs on the value in place, in parallel on scalo's worker pool. +3. `sonic_rs::to_vec` serialises each survivor back to JSON, keyed by the sink topic. +4. The loop sends the whole out-batch with `TransportSender::send_batch`. -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. +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. -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. +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. -## VRL Integration +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"}`, logged at debug level with the parse error (sampled), and reported to the security log by reason, with nothing from the record. The block's source is then released `Dropped`, so the record is not redelivered. -### Compilation +## VRL integration -VRL programs are compiled once at startup: - -```rust -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: - -```rust -let mut target = TargetValueRef { - value: &mut event_value, - metadata: &mut metadata, - secrets: &mut secrets, -}; -runtime.resolve(&mut target, &program)?; -``` +`compile_vrl` compiles the source once, at startup, against VRL's stdlib less `get_hostname`, plus the two enrichment functions, with the enrichment registry in the compile config. A compile error quotes the line that failed and stops startup. The compiled `Program` is reused for every event -- zero per-event compilation cost. -VRL's `Value` type is serde-compatible, so `sonic_rs` parses JSON bytes -directly into it. +`run_vrl` wraps each event in a `TargetValueRef` with an empty metadata object, and resolves it with UTC as the default timezone, so a zone-less `parse_timestamp` reads the same on every node. `abort` drops the event and counts it as filtered. Any other runtime error drops it and counts it in `records_error_total{stage="transform"}`. A panic inside VRL, which its grok and arithmetic functions can raise on record data, fails that record rather than the process, because the release profile unwinds. -### Transform File Format +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. `parse_proto` and `encode_proto` stay, and read the descriptor file their literal path names at compile time. `validate_json_schema` stays too, and reads its schema file when the first event reaches the call, keeping it once it compiles. A missing or invalid schema passes compilation and then fails each such event as a transform error. Transform files contain raw VRL source code (not Vector YAML): @@ -109,21 +58,11 @@ Transform files contain raw VRL source code (not Vector YAML): del(.message) # transforms/02_enrich.vrl -.environment = .labels.env ?? "unknown" +.environment = string(.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) +The `.vrl` files in `transforms.dir` load in sorted filename order, then the paths in `transforms.files` in the order given, and they concatenate into a single VRL program. This matches Vector's remap transform behaviour where multiple VRL statements execute sequentially. `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 @@ -136,81 +75,42 @@ 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 +(Rust-native). Wiring, routing behaviour, known limitations, and regeneration tooling (`scripts/filebeat/`) are documented in [pipelines/filebeat/README.md](../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 | what `src/deployment.rs::contract()` sets | +## Memory budget -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. +The wrapper bounds the buffers it owns. [config-schema.yaml](config-schema.yaml) carries the defaults for its own keys, and scalo's `BatchProcessingConfig` the default chunk size. -## Configuration - -Same big-dial pattern as `dfe-transform-vector`, minus Vector subprocess config: - -```yaml -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 -``` +| Component | Configuration | +|-----------|--------------| +| Consumer prefetch | `source.max_buffer_bytes`, set as librdkafka's `queued.max.messages.kbytes` | +| Transform chunk | `batch_processing.max_chunk_size`, a scalo section, from the `--config` file or the env layer | +| Producer queue | `sink.max_buffer_bytes`, set as librdkafka's `queue.buffering.max.kbytes` | +| Total pod memory | The container memory limit, whose chart default `src/deployment.rs::contract()` sets | -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. +The consumer and producer buffers have to fit within the pod memory limit beside the binary, the runtime, the compiled program and the enrichment tables. A full producer queue comes back from the send as backpressure, and the loop retries that block before it reads another, so intake waits on the sink and librdkafka's prefetch stops at `source.max_buffer_bytes`. Under memory pressure scalo's governor pauses the assigned partitions, and on the direct transport the listener refuses pushes `UNAVAILABLE`. ## 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) | +| Kafka source/sink | scalo's `KafkaTransport` over rdkafka (wrapper-controlled) | Vector-managed | +| Memory control | Bounded consumer and producer buffers, scalo's memory guard | None (Vector unbounded) | | Supported transforms | VRL only | All Vector transforms | | Container image size | Smaller (Rust binary) | Larger (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 +Use dfe-transform-vrl for VRL-only transforms (the common case), memory-constrained pods and high-density deployments. Use dfe-transform-vector for 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`, so a bad edit is caught without a restart. +swaps `SharedConfig`. A bad edit logs `Config reload failed, keeping current config` and counts in `config_reloads_total{result="error"}`, and the service keeps running. The reloadable subset carries three fields: | Field | What it would control | State | |-------|----------------------|-------| @@ -218,16 +118,12 @@ swaps `SharedConfig`, so a bad edit is caught without a restart. | `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, so a reload changes no pipeline behaviour. `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. +the wrapper warns at startup for each one a deployment has moved off its default. The dial that +does size a chunk is `batch_processing.max_chunk_size`, a scalo section. -### Requires pod restart - -These fields are bound to connections, compiled programs, or server sockets +The fields below 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). @@ -235,7 +131,8 @@ standard K8s pattern -- ConfigMap changes trigger rolling restart via the | Field | Why | |-------|-----| | `pipeline.name` | Baked into Kafka group_id, metrics labels, tracing spans | -| `source.*` | rdkafka consumer: connection, subscription, auth, TLS, buffers | +| `source.*` | rdkafka consumer or Push listener: connection, subscription, auth, TLS, buffers | +| `sink.transport`, `sink.endpoint` | The sink built at startup | | `sink.brokers` | rdkafka producer connection | | `sink.topic` | Output topic (changing mid-stream risks data loss) | | `sink.compression` | rdkafka `compression.type` | @@ -243,13 +140,16 @@ standard K8s pattern -- ConfigMap changes trigger rolling restart via the | `sink.max_buffer_bytes` | rdkafka `queue.buffering.max.kbytes` | | `sink.librdkafka_options` | librdkafka `ClientConfig` | | `transforms.*` | VRL programs compiled at startup | +| `enrichment_tables` | Tables loaded at startup, and the compiled program checks its lookups against them | +| `dlq.*` | The dead-letter queue spawned at startup | ## Operating notes +The transform's own sections are `pipeline`, `source`, `sink`, `transforms`, `enrichment_tables` and `dlq`, and [config-schema.yaml](config-schema.yaml) lists every key in them. Only `--config` names the file: `DFE_TRANSFORM_VRL_CONFIG` reaches neither the wrapper nor scalo, and with no `--config` the wrapper reads `config.yaml` or `config.yml` from the working directory. + ### Environment variable overrides -The big dials have flat env var overrides for K8s. This is the whole list -- -anything not here is not read: +The big dials have flat env var overrides for K8s. These are every flat name `apply_flat_env` in `src/config/loader.rs` reads: | Env Var | Config Field | |---------|-------------| @@ -258,11 +158,15 @@ anything not here is not read: | `DFE_TRANSFORM_VRL_KAFKA_SASL_PASSWORD` | `source.sasl.password` + `sink.sasl.password` | | `DFE_TRANSFORM_VRL_KAFKA_SASL_MECHANISM` | `source.sasl.mechanism` + `sink.sasl.mechanism` | | `DFE_TRANSFORM_VRL_KAFKA_SECURITY_PROTOCOL` | `source.tls.enabled` + `sink.tls.enabled`, set true when the value contains `SSL`, never set false | +| `DFE_TRANSFORM_VRL_SOURCE_TRANSPORT` | `source.transport` (`bus`, `kafka`, `direct` or `grpc`) | +| `DFE_TRANSFORM_VRL_SOURCE_LISTEN` | `source.listen` | | `DFE_TRANSFORM_VRL_SOURCE_BROKERS` | `source.brokers` | | `DFE_TRANSFORM_VRL_SOURCE_TOPICS` | `source.topics` | | `DFE_TRANSFORM_VRL_SOURCE_GROUP_ID` | `source.group_id` | | `DFE_TRANSFORM_VRL_SOURCE_SASL_USERNAME` | `source.sasl.username` | | `DFE_TRANSFORM_VRL_SOURCE_SASL_PASSWORD` | `source.sasl.password` | +| `DFE_TRANSFORM_VRL_SINK_TRANSPORT` | `sink.transport` (`bus`, `kafka`, `direct` or `grpc`) | +| `DFE_TRANSFORM_VRL_SINK_ENDPOINT` | `sink.endpoint` | | `DFE_TRANSFORM_VRL_SINK_BROKERS` | `sink.brokers` | | `DFE_TRANSFORM_VRL_SINK_TOPIC` | `sink.topic` | | `DFE_TRANSFORM_VRL_SINK_KEY_FIELD` | `sink.key_field` | @@ -271,6 +175,8 @@ anything not here is not read: | `DFE_TRANSFORM_VRL_SINK_SASL_PASSWORD` | `sink.sasl.password` | | `DFE_TRANSFORM_VRL_TRANSFORMS_DIR` | `transforms.dir` | +An empty value reads as unset. Every other key of the same sections takes the double-underscore form, `DFE_TRANSFORM_VRL_SINK__TOPIC` for `sink.topic`, and a flat name outranks it. + The chart mounts the Kafka Secret's `username` and `password` into both endpoint pairs, and its `sasl.mechanism` (`PLAIN`, `SCRAM-SHA-512`) into `KAFKA_SASL_MECHANISM`. Outside the chart the shared `KAFKA_SASL_*` names reach both endpoints, and the per-endpoint names override them. Either half turns SASL on, an enabled block still missing one once the env @@ -300,7 +206,7 @@ scales on CPU and opens no Kafka connection, so it needs no CA. ### scalo's own settings share the config file -`metrics`, `logger`, `scaling`, `worker_pool`, `batch_processing`, +`metrics`, `logger`, `memory`, `scaling`, `worker_pool`, `batch_processing`, `self_regulation`, `version_check` and `otel_tracing` are resolved by scalo, not by the wrapper. scalo reads the file passed with `--config` as its settings layer, once at @@ -315,8 +221,7 @@ batch_processing: ``` A `config.yaml` picked up from the working directory with no `--config` is read -by the wrapper alone, so a scalo section in it changes nothing and the wrapper -warns at startup. Pass the file with `--config`. +by the wrapper alone, so a scalo section in it changes nothing. The wrapper warns at startup for each section `config::SCALO_CASCADE_SECTIONS` lists, and that list has no `memory`. Pass the file with `--config`. The env layer outranks the file, and there the section nests on a **double** underscore: @@ -332,9 +237,9 @@ DFE_TRANSFORM_VRL_METRICS__OTEL__ENABLED=false # OTLP metric push, on by defau ``` A single underscore (`DFE_TRANSFORM_VRL_METRICS_ADDRESS`) produces a flat key that -matches no section and is ignored, and that spelling also warns. +matches no section and is ignored, and that spelling also warns. The memory guard is the exception: scalo also reads `DFE_TRANSFORM_VRL_MEMORY_LIMIT_BYTES`, `DFE_TRANSFORM_VRL_MEMORY_PRESSURE_THRESHOLD` and `DFE_TRANSFORM_VRL_MEMORY_CGROUP_HEADROOM` on a single underscore. -### Helm chart +## Helm chart No chart is committed here. At release, hyperi-ci runs the binary's `generate-artefacts` for the deployment contract and assembles a thin chart from it on the scalo-service library chart, at the version `release.helm.library` names in `.hyperi-ci.yaml`. Keep that version at the scalo version in `Cargo.toml`: a library renders only the contract version its scalo release writes. diff --git a/pipelines/filebeat/README.md b/pipelines/filebeat/README.md index 0e00e6b..5b568da 100644 --- a/pipelines/filebeat/README.md +++ b/pipelines/filebeat/README.md @@ -23,8 +23,8 @@ Single-file port of the DFE 2.1 Vector templates (dfe-vector-templates | 143..149-transform-filebeat-cisco-umbrella-*-logs | Cisco Umbrella exports (audit, cloud firewall, dlp, dns, intrusion, ip, proxy) | The seven umbrella templates were one body split across files (Vector -large-VRL workarounds) differing only in a hardcoded `.log.file.path` seed; -they collapse here into one body behind log-type detection. +large-VRL workarounds) differing only in a hardcoded `.log.file.path` seed. +They collapse here into one body behind log-type detection. ## Opting in @@ -39,11 +39,11 @@ enrichment_tables: key_columns: ["abbreviation"] ``` -Both files are in this directory; mount or copy them wherever the pod can +Both files are in this directory. Mount or copy them wherever the pod can read them. Nothing else is needed - the engine treats this like any other `transforms.dir`. -Events are the DFE 2.1 Kafka shape: `{message, tags, timestamp}`. Two +Events carry the DFE 2.1 Kafka fields: `{message, tags, timestamp}`. Two optional event fields drive ios/meraki timezone handling: `._conf.tz_offset` (an offset like `-0600` or an abbreviation from `timezones.csv`) and `._conf.tz_map` (a list of `{tz_short, tz_long}` mappings from log @@ -60,46 +60,49 @@ per event, so this port detects the module first: 1. `<134>1 ` prefix -> Meraki 2. any other `` syslog prefix -> Cisco IOS -3. `%FACILITY-SEVERITY-MNEMONIC` marker (on a non-CSV-shaped line) -> - Cisco IOS -4. otherwise CSV shape -> Umbrella; log type from `.log.file.path` when the - upstream provides one, else column-shape detection: 7 cols = ip, - 9 with the timestamp in column 1 = audit, 13 (or 10-12 with the - ` (TYPE)` query-type marker in column 6) = dns, 14/16 = cloud - firewall, 17 = intrusion when columns 3-4 are numeric gid/sid else - dlp, >=20 = proxy; every rule also requires a leading timestamp column +3. `%FACILITY-SEVERITY-MNEMONIC` marker (on a line that does not open like + an Umbrella CSV row) -> Cisco IOS +4. otherwise CSV columns -> Umbrella. The log type comes from + `.log.file.path` when the upstream provides one, else from the column + count: 7 cols = ip, 9 with the timestamp in column 1 = audit, 13 (or + 10-12 with the ` (TYPE)` query-type marker in column 6) = dns, 14/16 = + cloud firewall, 17 = intrusion when columns 3-4 are numeric gid/sid else + dlp, >=20 = proxy. Every rule but audit also requires a timestamp in + column 0 5. anything else: tagged `filebeat_unmatched` and passed through unchanged (the 2.1 fan-out silently dropped these) -Detection picks the branch; parsing failures INSIDE the chosen branch +Detection picks the branch, and parsing failures INSIDE the chosen branch still error the event out (dropped and counted), matching the 2.1 -`drop_on_abort` behaviour. Only shape-unmatched events pass through. +`drop_on_abort` behaviour. Only events no rule matches pass through. ## Deliberate departures from DFE 2.1 -- **No geoip enrichment.** dfe-loader owns geoip in the current pipeline - layout; the `geoip_city`/`geoip_asn` lookups are stripped. -- **Unmatched events pass through tagged**, never silently dropped. -- **`.log.file.path` is rewritten** to `/filebeat/cisco_umbrella/` +- No geoip enrichment. dfe-loader owns geoip in the current pipeline + layout, so the `geoip_city`/`geoip_asn` lookups are stripped. +- Unmatched events pass through tagged, never silently dropped. +- `.log.file.path` is rewritten to `/filebeat/cisco_umbrella/` in the umbrella branch (2.1 also overwrote it, with `/test/path/`). -- **IANA zone names resolve correctly in the ios branch.** The 2.1 +- IANA zone names resolve correctly in the ios branch. The 2.1 templates pushed a `tz_map`-mapped name like `Australia/Sydney` into the - abbreviation CSV lookup, which always failed and dropped the event; the + abbreviation CSV lookup, which always failed and dropped the event. The port routes slash-bearing zone names through `parse_timestamp`'s `timezone:` argument (DST-aware via the tz database). ## Known limitations (verified against the elastic golden corpus) - `user_agent` classification follows VRL's parser, which differs - cosmetically from elastic's ("Mac OSX" vs "Mac OS X"); a + cosmetically from elastic's ("Mac OSX" vs "Mac OS X"). A whitespace-only UA string is treated as absent. - `url.query` is rebuilt with alphabetised parameters (VRL object maps are - ordered); elastic preserves raw order. Same parameters, different order. + ordered), and elastic preserves raw order. Same parameters, different order. +- `url.path` of a URL with no path is `/`, where elastic keeps it empty. - `dns.question.registered_domain` uses the full public-suffix list - including the private section; elastic uses the ICANN section only. + including the private section, and elastic uses the ICANN section only. - Ambiguous timezone abbreviations (the CSV carries duplicates such as ADT, AMT, AST, CST, IST) never resolve - the 2.1 "ambiguous timezone abbreviation" path is preserved. +- A Cisco IOS timestamp with no year takes the current year. ## Provenance and regeneration diff --git a/tests/TESTING.md b/tests/TESTING.md index 15d33a4..d5d6abb 100644 --- a/tests/TESTING.md +++ b/tests/TESTING.md @@ -1,55 +1,14 @@ # Testing -## Test Modes +## Tiers -Tests support two backends, controlled by `TEST_MODE` in `.env`: - -### Remote (default) - -Uses the remote cluster endpoints from `.env`. Tests skip if endpoints are unreachable. - -```bash -TEST_MODE=remote cargo nextest run --all-features -``` - -### Docker-local - -Uses `dfe-docker` infra profile (Kafka on localhost, no auth, no TLS). - -```bash -# Start infrastructure (once, stays running) -cd ../dfe-docker -docker compose --profile infra up -d - -# Run tests -TEST_MODE=docker cargo nextest run --all-features - -# Tear down (when done) -docker compose --profile infra down -``` - -Docker-local Kafka: `localhost:19092` (PLAINTEXT, no SASL) - -## Test Categories - -### Unit Tests - -Run without any external infrastructure. Embedded in source files (`#[cfg(test)]`). +Unit tests run without any external infrastructure, embedded in the source files (`#[cfg(test)]`): ```bash cargo nextest run --lib ``` -### Integration Tests - -In `tests/` directory. Most run without Kafka (VRL compilation, transforms, config). -Kafka-dependent tests use `skip_if_no_kafka!()` and skip cleanly when Kafka is unavailable. - -```bash -cargo nextest run --all-features -``` - -### Core and full tiers +The integration tests in `tests/` run without a broker: VRL compilation, transforms, config, enrichment, the filebeat corpus, the direct transport on local ports, and the pipeline over scalo's in-memory transport. The core tier is every test that is not `#[ignore]`: unit, integration, smoke, and the e2e tests that use a broker container scalo starts (`filebeat_kafka`, `held_acks`, `kafka_tls`). PR and push runs use it, and so does `hyperi-ci check`. @@ -57,41 +16,41 @@ The core tier is every test that is not `#[ignore]`: unit, integration, smoke, a cargo nextest run --all-features ``` -The full tier adds the `#[ignore]` e2e tests (`kafka`, `cli_service`, `container_hygiene`), which prefer a live broker when `KAFKA_BROKERS` answers. A release runs it (`test.full.required_for_release`), and so does the nightly CI run. +The full tier adds the `#[ignore]` e2e tests. `kafka`, `cli_service` and `container_hygiene` prefer a live broker when `KAFKA_BROKERS` answers, else the run's shared container, and `connectivity` checks a live broker only and skips when none answers. A release runs the full tier (`test.full.required_for_release`), and so does the nightly CI run. ```bash cargo nextest run --all-features --run-ignored all --ignore-default-filter hyperi-ci check --tier full ``` -### Broker containers +## Broker containers scalo's test support starts every broker, from the image `scalo::transport::kafka::test_support::KAFKA_IMAGE` pins. Locally and in CI the same way: the first test process that needs one starts it through Docker, and nothing has to be running beforehand. - One PLAINTEXT broker serves the whole run (`shared_kafka`). Every nextest process joins it, and a watcher removes it when the run ends. - `held_acks` holds a broker alone, because it pauses it, and removes it when the test ends. -- `kafka_tls` holds a TLS, mTLS and SASL_SSL broker alone (`TlsBroker`), with certificates from `alpine/openssl:3.5.9`. Its clients name no CA file and trust the bundle `SSL_CERT_FILE` names. +- `kafka_tls` holds a TLS, mTLS and SASL_SSL broker alone (`TlsBroker`), with certificates from the openssl image scalo's test support pins. Its clients name no CA file and trust the bundle `SSL_CERT_FILE` names. + +Containers are named `dfe-transform-vrl-test-integration-*` (or `HYPERI_TEST_CONTAINER_PREFIX`) and labelled `io.hyperi.test.suite=dfe-transform-vrl-integration`. A core-tier test that needs a broker skips without Docker locally and fails under `CI`. -Containers are named `dfe-transform-vrl-test-integration-*` (or `HYPERI_TEST_CONTAINER_PREFIX`) and labelled `io.hyperi.test.suite=dfe-transform-vrl-integration`. A test that needs a broker skips without Docker locally and fails under `CI`. +## A live broker -The testing standard these tiers follow: `standards/universal/testing.md` in hyperi-ai. +`TEST_MODE`, read from the environment or this repo's own `.env`, picks where a test looks for a live broker. Unset or `remote`, it reads `KAFKA_BROKERS` (default `localhost:9092`) with `KAFKA_SECURITY_PROTOCOL`, `KAFKA_SASL_MECHANISM`, `KAFKA_SASL_USER` and `KAFKA_SASL_PASSWORD`. `docker` means `localhost:19092`, PLAINTEXT with no SASL, the host port dfe-docker publishes its broker on. A test that finds no live broker there falls back to the run's shared container, except `connectivity`, which skips. -## Test Infrastructure +```bash +TEST_MODE=docker cargo nextest run --all-features --run-ignored all --ignore-default-filter +``` -### `tests/common/mod.rs` +`.env` is gitignored and holds a developer's own values. `.env.example`, committed, lists the names the harness and the app read, with placeholders. -Shared helpers used by integration and e2e tests: +## Shared helpers -- `TestMode::detect()` — reads `TEST_MODE` from `.env` -- `kafka_test_config()` — returns correct broker/auth config for the active mode -- `test_topic("suffix")` — generates unique topic names with timestamp -- `skip_if_no_kafka!()` — skips test if Kafka is unreachable -- `ensure_docker_infra()` — starts dfe-docker containers if needed +`tests/common/mod.rs` holds the helpers the integration and e2e tests share: + +- `TestMode::detect()` -- reads `TEST_MODE` once this repo's `.env` is loaded +- `kafka_test_config()` -- the live-broker config for the active mode +- `test_topic("suffix")` -- a unique topic name with a timestamp, prefixed by `TEST_TOPIC_PREFIX` - `KafkaTestEnv::ensure()` -- a live broker when one answers, else the run's shared broker container. `hermetic()` is the shared container alone - `KafkaTestEnv::hermetic_on_low_port()` -- a broker container the test holds alone, published below port 10240. `pause()` and `unpause()` freeze and resume it +- `ensure_kafka_or_skip!()` -- `ensure()`, skipping when it finds neither and failing under `CI` - `ports::free_port()` -- a host port below 10240 for a test listener, outside the range the OS hands out ephemeral ports from - -### `.env` and `.env.example` - -`.env` contains real credentials (gitignored). `.env.example` has the same structure -with placeholder values (committed). diff --git a/tests/fixtures/filebeat/README.md b/tests/fixtures/filebeat/README.md index 7773213..c903900 100644 --- a/tests/fixtures/filebeat/README.md +++ b/tests/fixtures/filebeat/README.md @@ -25,12 +25,12 @@ the working tree): ## Provenance - Source repo: `github.com/elastic/integrations` -- Commit: `c7bc53023288c7d34afd631c0c3e3552f6219419` (pinned -- the same - vintage the DFE 2.1 VRL templates were generated against, and the same - pin the dfe-transform-elastic testdata and elastic_to_vrl submodule use) +- Commit: `c7bc53023288c7d34afd631c0c3e3552f6219419` (pinned in + `scripts/filebeat/fetch_testdata.py` -- the same vintage the DFE 2.1 VRL + templates were generated against) - Paths: `packages/cisco_{umbrella,ios,meraki}/data_stream/log/_dev/test/pipeline/` - Licence: Elastic License 2.0 (test data redistributed unmodified for - compatibility testing; see `LICENSE.txt` inside the archive) + compatibility testing, see `LICENSE.txt` inside the archive) ## Refreshing @@ -46,19 +46,15 @@ tar --sort=name --owner=0 --group=0 --numeric-owner \ cisco_umbrella cisco_ios cisco_meraki LICENSE.txt licenses ``` -The tar flags make the archive byte-reproducible; a refresh only churns +Those tar flags make the archive byte-reproducible, so a refresh only churns the blob when content actually changed. Re-validate the -`ACCEPTED_MISMATCHES` line indices in -`tests/integration/filebeat_pipeline.rs` after any refresh - they are -keyed by line position within each log file. - -Keep the pin aligned with the dfe-transform-elastic testdata unless the -bundled VRL is regenerated against a newer template vintage. +`ACCEPTED_MISMATCHES` line indices in `tests/common/filebeat.rs` after any +refresh - they are keyed by line position within each log file. ## Golden-output caveats The `-expected.json` files are the outputs of the ELASTIC ingest pipelines -at that pin. The bundled VRL is the DFE 2.1 port, which deliberately +at that pin. This repo's VRL is the DFE 2.1 port, which deliberately differs: geoip enrichment is stripped (dfe-loader owns geoip), `related.*` / ECS versions may lag the newest integration revisions, and events carry DFE 2.1 conventions (`_conf.tz_offset`, tag seeding). Tests therefore