diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fba5387..e6321a1 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 @@ -67,38 +67,18 @@ affected (e.g., `api`, `auth`, `core`, `cli`, `docs`). 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: @@ -133,11 +113,11 @@ anniversary of its release. ## 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 @@ -148,15 +128,7 @@ anniversary of its release. ## CI/CD Workflow -When your pull request is merged to `main`: - -1. **semantic-release** analyses commit messages since the last release -2. Determines the next version number based on commit types -3. Generates/updates the CHANGELOG -4. Creates a new GitHub release with release notes -5. Publishes the package (if applicable) - -This happens automatically - no manual intervention required. +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 @@ -166,29 +138,20 @@ If you have questions about contributing, please open an issue or contact us. ### What this is -A Rust supervisor that runs Vector.dev as a child process so a Vector pipeline -behaves like every other DFE app -- one big-dial config, one ops port, an OTLP -push, and a chart assembled from its deployment contract. It is NOT -the transform engine and it does not build Vector: the image downloads the -upstream release binary at the version pinned by `VECTOR_VERSION` in -`src/deployment.rs`, and cargo compiles only the supervisor. It is also not a -singleton and not part of the default deploy -- `dfe-infra/apps.yaml` declares -it `multiplicity: per_config` with `scale_deployed: true` and an empty -`default_in`, so one deployment exists per source config and only when a -deployment asks for it. +A Rust supervisor that runs Vector as a child process so a Vector pipeline behaves like every other DFE app: one big-dial config, one ops port, an OTLP push, and a chart assembled from its deployment contract. It is NOT the transform engine and it does not build Vector. The image downloads the upstream release binary at the version pinned by `VECTOR_VERSION` in `src/deployment.rs`, and cargo compiles only the supervisor. It is not a singleton either: dfe-infra's `apps.yaml` declares it `multiplicity: per_config` with `scale_deployed: true` and `default_in: [docker-single]`, so one deployment exists per source config, and outside `docker-single` only when a deployment asks for it. ### Where things live | Path | What it holds | |---|---| | `src/main.rs` | CLI entry. Fills scalo's `CommonArgs` from the loaded config before `run_app`, which is what makes `metrics.address` and `logging.*` take effect | -| `src/config/` | `loader` -> `validate` -> `generate` -> `wiring` -> `assembler`, plus `reload`, `transforms` and `kafka_defaults` | +| `src/config/` | `loader` (the config and its env layers) and the Vector config it becomes: `generate`, `wiring`, `transforms`, `secrets`, `assembler`, `validate`, plus `reload`, `staging` and `kafka_defaults` | | `src/vector/` | `process` spawn and backoff, `lifecycle` states, and `binary` -- version selection kept pure and NOT on the run path, which its own header explains | | `src/bridge.rs` | The direct-transport translation between scalo `Transport/Push` and Vector `PushEvents`. Does nothing on `bus` | | `src/health.rs`, `src/metrics.rs`, `src/metrics/scrape.rs` | Readiness publishing, and the scrape that merges `vector_*` into scalo's registry | | `src/deployment.rs`, `src/vector-layer.dockerfile` | `contract()`, the source the Dockerfile and the released chart are generated from, and the two-binary image override spliced onto scalo's generated Dockerfile | | `templates/bus.yaml`, `templates/direct.yaml` | Whole runnable Vector topologies, one per transport, commented per field | -| `pipelines/filebeat/` | The shipped filebeat pipeline | +| `pipelines/filebeat/` | The bundled filebeat pipeline. The image does not carry it | | `docs/` | `architecture.md` for why the service is built this way and the invariants, `internals.md` for field-by-field depth, `migration.md`, `librdkafka.md`, the generated `config-schema.*` and `capability-catalog.*` | | `tests/` | `integration`, `e2e`, `smoke`, and `TESTING.md` for how the broker and Vector binary are resolved | | `scripts/fetch-vector.sh`, `scripts/pgo-workload.sh` | Downloads the pinned Vector for tests, and drives the PGO/BOLT workload | @@ -200,42 +163,33 @@ make check # hyperi-ci check -- quality + test cargo nextest run --lib # unit only, no infrastructure cargo nextest run --test integration # config assembly, wiring, lifecycle, metrics cargo nextest run --test smoke # CLI surface -cargo nextest run --test e2e # needs Docker or a live broker +cargo nextest run --test e2e # needs Docker or a live broker, and a Vector binary cargo nextest run --run-ignored all # everything, opt-in cases included ``` What green does NOT mean: -- **The default run skips the Vector validator.** Eight `#[ignore]` cases (seven in `tests/integration/vector_validate.rs`, one in `tests/e2e/kafka.rs`) need a - real Vector binary -- six in `tests/integration/vector_validate.rs` and one in - `tests/e2e/kafka.rs`. Until you pass `--run-ignored all`, nothing has run - Vector's own validator over an assembled config. -- **Locally, a missing broker or Vector binary SKIPS rather than fails.** In CI - that same absence is an assertion failure instead, so CI is stricter than a - green laptop run. -- **`e2e::filebeat_kafka` runs by default but skips without a sibling repo.** The - corpus and elastic's golden events live in dfe-transform-vrl, so it needs that - repo checked out beside this one or `DFE_TRANSFORM_VRL_DIR` pointing at it. -- **A docs-only push runs no CI at all.** `paths-ignore` for `docs/**` and - `**.md` is on the `push` trigger but not on `pull_request`, so the PR is the - only place a docs change is checked. A silent push is not a pass. +- The default run leaves out the `#[ignore]` cases in `tests/integration/vector_validate.rs` and `tests/e2e/kafka.rs`. The `vector_validate` ones run Vector's own validator over each buffer, SASL and TLS variant of an assembled config, so until you pass `--run-ignored all` those variants are unchecked. +- Locally, a missing broker or Vector binary SKIPS rather than fails. In CI that absence is an assertion failure for the cases that run by default, so CI is stricter than a green laptop run. +- `e2e::filebeat_kafka` runs by default and needs a dfe-transform-vrl checkout, for its cases and the bundled pipeline: beside this repo, or named by `DFE_TRANSFORM_VRL_DIR`. Without one it skips locally and fails under CI, which checks the repo out beside this one. +- A docs-only push runs no CI at all. `paths-ignore` for `docs/**` and `**.md` is on the `push` trigger but not on `pull_request`, so the PR is the only place a docs change is checked. A silent push is not a pass. ### What tends to bite | 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 | +| Hand-edit `Dockerfile`, or commit a chart | Fix `src/deployment.rs::contract()` and regenerate | The Dockerfile is generator output, so `checked_in_dockerfile_matches_emit_dockerfile` fails on a hand edit, and the release assembles the chart from the contract | | 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, adding a pressure trigger through the chart's `keda.extraTriggers` where a deployment wants one | Lag rises when a downstream stage breaks, so a lag trigger adds pods that wait on the same broken stage. The library's trigger also reads `config.kafka.*`, which this app does not have, and refuses to render | -| Set `vector.version` in a deployment's config and forget it | Leave it unset, so the default reads `VECTOR_VERSION` | A pinned version that differs from the binary the image bakes refuses to start under `strict`, and runs only because `version_check: warn` lets it | +| Set `vector.version` in a deployment's config and forget it | Leave it unset, so the default reads `VECTOR_VERSION` | A pinned version that differs from the binary the image bakes refuses to start under `strict`, the default, and runs only where `vector.version_check: warn` lets it | | Mount a whole `templates/*.yaml` as a transform file | Copy only its `transforms:` block | The supervisor generates `sources:` and `sinks:` from the big dials and refuses a transform file declaring them | | Change `password` to a `SensitiveString` to look safer | Leave it a `String` | `SensitiveString` serialises as `***REDACTED***` and the figment serialize-merge-deserialize round trip in `apply_figment_env()` destroys the value. Masking happens in logs and `Debug`, and Vector reads the credential from files through its `directory` secret backend, so it never lands in the Vector config | -| Put a `${VAR}` placeholder in a credential or a transform | Mount the secret and set `sasl.secret_dir`, or give the credential through `DFE_TRANSFORM_VECTOR_{SOURCE,SINK}_SASL_*`, and read the environment in VRL with `get_env_var` | Vector 0.57+ expands no `${VAR}` without `--dangerously-allow-env-var-interpolation`, which the supervisor never passes, so a credential holding `${` is refused at load rather than reaching the broker as the password | -| Assume a config key you added is wired because it parses and validates | Prove it changes behaviour, and add it to the standing check that every committed config file loads | A key can parse, validate and reach nothing, so a deployment that moves a port on it keeps the old port and loses every probe | -| Set `enable.auto.commit: false` here, as the shared DFE consumer baseline does | Leave auto-commit on | Vector's kafka source only stores offsets and leaves librdkafka's commit timer to flush them, and that timer is armed only when auto-commit is on, so with it off the stored offsets never reach the broker and consumer lag never clears | -| Read `/readyz` as "carrying traffic" | Read it as "the child was alive 500ms after spawn, and the sink is not stalled" | Readiness is those two signals and nothing more, so a pod can be Ready with no record moving | -| Swap the reload poller for inotify | Keep polling | S3-backed mounts -- s3fs, goofys, Mountpoint for S3 -- generate no filesystem notification events, so inotify works on every laptop and silently stops reloading in production | -| Add or rename a credential field without a schema marker | Schema it as `scalo::SensitiveString` with `#[schemars(with = ...)]`, as `SaslConfig::password` is | That puts `x-scalo-secret` and `writeOnly` on the field in the emitted config-schema, which is what tells a console to mask it. Without it, only dfe-engine's leaf-name fallback (`password`, `secret`, `token`, `api_key`, `private_key`, `passphrase`) redacts it, so a name outside that set returns an operator's Kafka password in the clear | +| Put a `${VAR}` placeholder in a credential or a transform | Mount the secret and set `sasl.secret_dir`, or give the credential through `DFE_TRANSFORM_VECTOR_{SOURCE,SINK}_SASL_*`, and read the environment in VRL with `get_env_var` | Vector 0.57+ expands no `${VAR}` without `--dangerously-allow-env-var-interpolation` or `VECTOR_DANGEROUSLY_ALLOW_ENV_VAR_INTERPOLATION`, and the supervisor sets neither. A credential holding `${` is refused at load rather than reaching the broker as the password. Vector runs without the `DFE_TRANSFORM_VECTOR_*` variables, so `get_env_var` cannot read those | +| Assume a config key you added is wired because it parses and validates | Prove it changes behaviour, and keep `every_committed_config_file_loads_and_validates` passing over each committed config file | A key can parse, validate and reach nothing, so a deployment that moves a port on it keeps the old port and loses every probe | +| Set `enable.auto.commit: false` here, as the shared DFE consumer baseline does | Leave auto-commit on | Vector's kafka source only stores offsets and leaves librdkafka's commit timer to flush them, and that timer is armed only when auto-commit is on, so with it off the stored offsets reach the broker only on a rebalance or a clean shutdown | +| Read `/readyz` as "carrying traffic" | Read it as "Vector survived 500 ms after spawn, and the sink is not stalled" | A pod whose broker is unreachable, with nothing held, stays Ready with no record moving | +| Swap the reload poller for inotify | Keep polling | S3-backed mounts -- s3fs, goofys, Mountpoint for S3 -- generate no filesystem notification events, so inotify works on every laptop and stops reloading in production with nothing logged | +| Add or rename a credential field without a schema marker | Schema it as `scalo::SensitiveString` with `#[schemars(with = ...)]`, as `SaslConfig::password` is | That puts `x-scalo-secret` and `writeOnly` on the field in the emitted config-schema, which is what tells a console to mask it. Without it only dfe-engine's name-based fallback can mask it, so a name that fallback does not read as a credential returns an operator's Kafka password in the clear | ### Where this sits @@ -245,31 +199,14 @@ and `--producer dfe-transform-vector`. Inbound -- what this repo depends on: -- **scalo-rs**, `cargo-dep`. `Cargo.toml` declares the `scalo` crate by range, - with a second range covering the dev dependency. A scalo release reaches this - repo, so widen the range if it does not admit the new version, then rebuild. -- **scalo-rs**, `generated-file`, lockstep. The committed `Dockerfile` is written - by `scalo::deployment::generate_dockerfile()` -- its own header names the - generator and the schema version -- and this repo splices the Vector layer onto - it. When the generator or its schema moves, regenerate with - `dfe-transform-vector emit-dockerfile > Dockerfile` and commit 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`. +- scalo-rs, `cargo-dep`. `Cargo.toml` declares the `scalo` crate by range, with a second range covering the dev dependency. A scalo release reaches this repo, so widen the range if it does not admit the new version, then rebuild. +- scalo-rs, `generated-file`, lockstep. The committed `Dockerfile` is written by `scalo::deployment::generate_dockerfile()` -- its own header names the generator and the schema version -- and this repo splices the Vector layer onto it. When the generator or its schema moves, regenerate with `dfe-transform-vector emit-dockerfile > Dockerfile` and commit 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`. Outbound -- what depends on this repo: -- **dfe-infra**, `image-pin`, lockstep. `dfe-infra/versions.yaml` pins this - repo's container image as a tag plus the digest that makes the tag - immutable, and the thin chart released here, which Argo deploys. Bump the - tag and re-resolve the digest. - -Two more relationships that are real but are NOT declared edges in the suite -graph, so no gate enforces them: - -- **dfe-infra `apps.yaml`** is the SSoT for how this app is deployed -- multiplicity, - scaling, transports, endpoints and which compiler derives its routing. - dfe-engine reads that manifest, so adding or changing the app is a manifest - edit, never an engine release. -- **dfe-transform-vrl** holds the filebeat corpus, the bundled pipeline and the - documented divergences that `e2e::filebeat_kafka` grades against. The - dependency is a test-time repo checkout only, and the case skips without it. +- dfe-infra, `image-pin`, lockstep. `dfe-infra/versions.yaml` pins this repo's container image as a tag plus the digest that makes the tag immutable, and the thin chart released here, which Argo deploys. Move the tag with `dfe-stack bump-app`, which repins the chart, then re-resolve the image digest. + +Two more relationships that are real but are NOT declared edges in the suite graph, so no gate enforces them: + +- dfe-infra `apps.yaml` is the SSoT for how this app is deployed -- multiplicity, scaling, transports, endpoints and which compiler derives its routing. dfe-engine reads that manifest, so adding or changing the app is a manifest edit, never an engine release. +- dfe-transform-vrl holds the filebeat cases (`tests/fixtures/filebeat/cases.yaml`) and the bundled pipeline that `e2e::filebeat_kafka` grades against. The dependency is a test-time checkout only, which CI makes and a laptop may not. diff --git a/README.md b/README.md index b2e2340..b55d90a 100644 --- a/README.md +++ b/README.md @@ -7,20 +7,23 @@ Runs a Vector pipeline as a supervised transform stage in a DFE deployment. ## What it does -- Vector does the transform, and the supervisor supplies the rest a stage needs: config, credentials, health, metrics and restarts. Reach for it when you need one of Vector's own transforms beyond `remap`. For VRL alone, [dfe-transform-vrl](https://github.com/hyperi-io/dfe-transform-vrl) is smaller and in-process. +- Vector does the transform, and the supervisor supplies the rest a stage needs: config, credentials, health, metrics and restarts. Reach for it when you need one of Vector's own transforms beyond `remap`. For VRL alone, [dfe-transform-vrl](https://github.com/hyperi-io/dfe-transform-vrl) runs it in-process with no Vector binary. - One small YAML (pipeline name, brokers, topics, SASL and TLS) generates Vector's source, sink and observability config, so you write only the `transforms:` and `enrichment_tables:` blocks. - Kafka in and out (`bus`), or a direct gRPC push with no broker (`direct`). The transform you author is the same on both. -- A crashed Vector is restarted with backoff from 1 s to 60 s, and readiness follows it. +- A crashed Vector is restarted with exponential backoff, and readiness follows it. - With reload enabled, a transform change is validated, sent to Vector with a SIGHUP, and read back from Vector's own counters as applied, refused (the running config is put back) or unconfirmed. One that fails validation is logged and never sent. - Vector's `vector_*` metrics are merged into the supervisor's, so one endpoint and one OTLP push carry both. ## Quick start ```bash -cargo build --release -./target/release/dfe-transform-vector run --config config.example.yaml +cargo build --release --bin dfe-transform-vector +scripts/fetch-vector.sh # caches the pinned Vector under .tmp/ and prints its path +./target/release/dfe-transform-vector --config config.example.yaml config-check ``` +`--config` goes before the subcommand, and `run` is the default when you leave the subcommand out. To run it off the image, point `vector.binary` at the path `scripts/fetch-vector.sh` printed and `vector.config_dir` and `vector.data_dir` at directories you can write, then `./target/release/dfe-transform-vector --config config.example.yaml`. + [config.example.yaml](config.example.yaml) is the annotated reference. [templates/bus.yaml](templates/bus.yaml) and [templates/direct.yaml](templates/direct.yaml) are the whole Vector topology for each transport, commented per field and runnable as they stand: copy one, replace the `dfe_transform` step, and you have a working pipeline. In a deployment the supervisor generates `sources:` and `sinks:`, and refuses any other top-level key in a transform file. | | `bus` | `direct` | @@ -29,20 +32,20 @@ cargo build --release | Out | Kafka `_load` | a scalo Push client to `sink.endpoint` | | Needs a broker | yes | no | -One port, 9090 by default, carries the ops surface: `/livez`, `/readyz` (200 only while Vector is up and its sink is not stalled) and `/metrics`. Vector's own exporter defaults to `127.0.0.1:9598`, and `metrics.vector_metrics_address` stays on loopback. +One port, 9090 by default, carries the ops surface: `/livez`, `/readyz` and `/metrics`. `/readyz` answers 503 until Vector has survived its first 500 ms, and again while it is down or its sink has held records for `metrics.sink_stall_secs` without delivering any. A bus config with no topics idles instead: Ready, with no Vector running, until the config file names a topic. Vector's own exporter defaults to `127.0.0.1:9598`. Keep `metrics.vector_metrics_address` on loopback, because nothing checks it. ## Configuration -Settings load through a cascade: CLI, then `DFE_TRANSFORM_VECTOR_*` environment variables, then the working directory's `.env`, then the config file, then defaults. The transport dials take a single underscore (`DFE_TRANSFORM_VECTOR_SOURCE_TRANSPORT`, `_SOURCE_LISTEN`, `_SINK_ENDPOINT`, the SASL pairs), and a key inside a section takes two (`DFE_TRANSFORM_VECTOR_SOURCE__BROKERS`, `_SINK__TOPIC`, `_TRANSFORMS__DIR`). [docs/config-schema.yaml](docs/config-schema.yaml) lists every key with its type and default. +The config file (`--config`, else `config.yaml` or `config.yml` in the working directory) sits over the defaults, and `DFE_TRANSFORM_VECTOR_*` environment variables sit over the file. The working directory's `.env` is read into the environment first, and a variable already set wins. The flat names take a single underscore and a comma list (`DFE_TRANSFORM_VECTOR_SOURCE_BROKERS=a:9092,b:9092`, `_SINK_TOPIC`, `_TRANSFORMS_DIR`, the SASL pairs) and beat the double-underscore form. That form reaches any key and takes a list in brackets (`DFE_TRANSFORM_VECTOR_SOURCE__BROKERS='[a:9092,b:9092]'`), and a bare comma list there fails to load. `--log-level`, `--log-format` and `--metrics-addr`, or `LOG_LEVEL`, `LOG_FORMAT` and `METRICS_ADDR`, outrank `logging.*` and `metrics.address`. [docs/config-schema.yaml](docs/config-schema.yaml) lists the app's own keys with their types and defaults. -A transform file change hot-reloads when reload is enabled, after it validates. A change to `source.*`, `sink.*`, `bridge.*`, `pipeline.name`, `vector.*`, `metrics.*` or `logging.*` takes a pod restart, and the watcher logs it as such. The reload path, the readiness rules, the metrics merge and the Kafka tuning are in [docs/internals.md](docs/internals.md) and [docs/librdkafka.md](docs/librdkafka.md). +With reload on, the default, a transform file change applies once it validates, with no restart. Any other change to the app's own keys, `dfe_source`, `pipeline.name` and `scaling.pressure_threshold` included, is logged as needing a pod restart, and no later transform change applies until the pod restarts or the edit is reverted. scalo's own sections in the file, such as `version_check`, `scaling.enabled` and `metrics.namespace`, are read once at startup: an edit there needs a restart too, and the watcher logs a successful reload without applying it. In a chart deployment a config or file set change rolls the pods anyway, because the chart hashes both into the pod template. The reload path, the readiness rules, the metrics merge and the Kafka settings are in [docs/internals.md](docs/internals.md) and [docs/librdkafka.md](docs/librdkafka.md). ## Documentation - [docs/architecture.md](docs/architecture.md): the problem, the ownership boundary and the invariants. -- [docs/internals.md](docs/internals.md): the config engine, the subprocess manager, health and metrics, the Vector version pin, the chart. -- [docs/migration.md](docs/migration.md): moving from the official Vector chart. -- [docs/librdkafka.md](docs/librdkafka.md): Kafka client tuning. +- [docs/internals.md](docs/internals.md): the config engine, hot reload, the Vector subprocess, health and metrics, dead letters, the Vector version pin, the chart. +- [docs/migration.md](docs/migration.md): moving a pipeline off the official Vector chart. +- [docs/librdkafka.md](docs/librdkafka.md): the librdkafka settings Vector's Kafka source and sink are given. - [tests/TESTING.md](tests/TESTING.md): the test tiers, and the e2e cases that need Docker and a Vector binary. The suite's published docs are at , and this repository's pages at . 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 e92cb14..13a69c9 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1,5 +1,5 @@ # ============================================================================= -# dfe-transform-vector — example configuration +# dfe-transform-vector -- example configuration # ============================================================================= # # Copy to config.yaml and modify as needed. @@ -20,7 +20,7 @@ pipeline: # ----------------------------------------------------------------------------- # DFE source name # ----------------------------------------------------------------------------- -# The one setting a working config cannot go without. It derives the platform +# Set it, or set source.topics and sink.topic yourself. It derives the platform # topic names and the consumer group. With pipeline.name "my-pipeline" above: # source.topics -> ["my_source_land"] # sink.topic -> "my_source_load" @@ -49,7 +49,7 @@ dfe_source: "my_source" # auto_offset_reset: "largest" # largest, smallest # session_timeout_ms: 30000 # consumer group session timeout # commit_interval_ms: 5000 # offset commit interval -# drain_timeout_ms: 15000 # max wait during rebalance (default: half session_timeout) +# drain_timeout_ms: 15000 # max wait during rebalance (Vector's default: half session_timeout_ms) # topic_lag_metric: true # expose consumer lag metric # # Commit an offset, or answer a push, only once the record is delivered. # acknowledgements: @@ -69,7 +69,7 @@ dfe_source: "my_source" # enabled: true # # DFE_TRANSFORM_VECTOR_KAFKA_SECURITY_PROTOCOL containing SSL (SASL_SSL, SSL) # # turns this on for source and sink. No value turns it off. -# # librdkafka overrides — production defaults baked in, override here +# # librdkafka overrides over the production profile (docs/librdkafka.md) # # librdkafka_options: # # fetch.max.bytes: "10485760" # # queued.min.messages: "100000" @@ -92,15 +92,15 @@ dfe_source: "my_source" # socket_timeout_ms: 60000 # broker socket timeout # batch: # max_events: 10000 # Vector-level batch size (default 10K) -# # max_bytes: 8388608 # optional byte limit; on direct at most 16 MiB, 8 MiB unset -# timeout_secs: 1 # flush timeout; at most 1 on direct +# # max_bytes: 8388608 # optional byte limit, at most 16 MiB on direct, where unset means 8 MiB +# timeout_secs: 1 # flush timeout, at most 1 on direct # sasl: # enabled: true # mechanism: "scram_sha_512" # secret_dir: "/var/run/secrets/dfe-kafka" # tls: # enabled: true -# # librdkafka overrides — production defaults baked in, override here +# # librdkafka overrides over the production profile (docs/librdkafka.md) # # librdkafka_options: # # batch.size: "8388608" # # linger.ms: "20" @@ -123,7 +123,7 @@ dfe_source: "my_source" # ----------------------------------------------------------------------------- # transforms: # dir: "/etc/dfe/transforms/" -# # Or specify individual files (loaded in order): +# # Or list the files, loaded in order. Read only when dir is unset. # # files: # # - "/etc/dfe/transforms/parse.yaml" # # - "/etc/dfe/transforms/enrich.yaml" @@ -138,15 +138,16 @@ dfe_source: "my_source" # binary: "/usr/local/bin/vector" # data_dir: "/var/lib/vector" # # Where the assembled Vector config is written. The default is the -# # container path; a bare install has no write access there. +# # container path, which a bare install cannot write. # config_dir: "/var/run/vector/config" # # Vector's API is unauthenticated and nothing here uses it, so it is off. # # Turned on, it binds a loopback address only. # api_enabled: false # api_address: "127.0.0.1:8686" # log_level: "info" -# # Version pinning — prevents silent version drift -# version: "0.61.0" +# # The Vector release the binary must report. The default is the release +# # the image ships, so set it only beside a binary you supply. +# version: "" # version_check: "strict" # strict, warn, disabled # # The supervisor runs `binary` and never fetches one, so `preshipped` is # # the only value accepted. To run a different Vector, point `binary` at it @@ -174,11 +175,12 @@ dfe_source: "my_source" # ----------------------------------------------------------------------------- # Logging # ----------------------------------------------------------------------------- -# --log-level/LOG_LEVEL and --log-format/LOG_FORMAT outrank these; -# --verbose/--quiet outrank everything. +# --log-level/LOG_LEVEL and --log-format/LOG_FORMAT outrank these, and +# --verbose/--quiet outrank everything. auto is JSON when +# OTEL_EXPORTER_OTLP_ENDPOINT is set, else text on a terminal or under CI, else JSON. logging: level: "info" # trace, debug, info, warn, error - format: "auto" # json, text, auto (text on a tty) + format: "auto" # json, text, auto # ----------------------------------------------------------------------------- # Scaling pressure @@ -186,11 +188,11 @@ logging: # NOTHING IN THIS APP READS `pressure_threshold`. It is accepted for parity # with the `scaling:` block every DFE service takes, and range-checked. scalo's # runtime reads its own `enabled` and `memory_gate_threshold` keys from this -# block for the pressure gauge the subprocess circuit gates. +# block for the scaling_pressure gauge. Nothing feeds that gauge a reading here, +# so it stays 0 and the memory gate never fires. # -# What actually scales this app is the chart's KEDA ScaledObject -- CPU, plus -# any trigger a deployment adds -- gated by the Vector subprocess circuit, which -# pins pressure to 0 while the subprocess is down (more pods cannot help a dead -# subprocess). +# What scales this app is the chart's KEDA ScaledObject: CPU, plus any trigger +# a deployment adds. The subprocess circuit pins scaling_pressure to 0 while +# Vector is down, which matters only to a trigger that reads it. # scaling: # pressure_threshold: 0.8 diff --git a/docs/README.md b/docs/README.md index 30a63e7..df21a65 100644 --- a/docs/README.md +++ b/docs/README.md @@ -3,9 +3,9 @@ How dfe-transform-vector is built and configured, and how to move a pipeline onto it from Vector. - [Architecture](architecture.md) -- [DFE Transform Vector -- Design Document](internals.md) -- [librdkafka Configuration Reference](librdkafka.md) -- [Migration Guide: Vector -> dfe-transform-vector](migration.md) +- [Internals](internals.md) +- [librdkafka settings](librdkafka.md) +- [Moving off the Vector chart](migration.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 630277a..7d034ee 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1,65 +1,33 @@ # Architecture -Why this repo exists, what owns what, and the invariants you cannot read off a -single file. For the field-by-field design -- every config key, every generated -block, every state transition -- read [internals.md](internals.md). This document sits -above it. +Why this repo exists, what owns what, and the invariants you cannot read off a single file. For the field-by-field design -- every config key, every generated file, every state transition -- read [internals.md](internals.md). ## The problem -DFE needs a transform stage between its sources and its loader. Vector.dev -already does the per-record transform work well, so building another transform -engine was never the question. The question was the other half. +DFE needs a transform stage between its sources and its loader, and Vector already does the per-record transform work well. What it lacks is the rest of a DFE app. Vector has its own config format, its own metrics endpoint, its own process lifecycle and its own idea of what "started" means. Every DFE app presents one control surface instead: a big-dial YAML, one ops port carrying `/livez`, `/readyz` and `/metrics`, an OTLP push, KEDA scaling and a chart assembled from its deployment contract. -Vector on its own is opaque to the platform around it. It has its own config -format, its own metrics endpoint, its own process lifecycle and its own idea of -what "started" means. Every other DFE app presents one control surface: a -big-dial YAML in the config registry, one ops port carrying `/livez`, `/readyz` -and `/metrics`, an OTLP push, a consumer-lag scaling signal and a Helm chart -compiled by dfe-engine. Adopting Vector as it ships would mean dfe-engine, -GitOps and the observability stack each learning a special case. - -So Vector is kept, and wrapped. The Rust supervisor is PID 1 and Vector is its -child. The supervisor supplies the production half -- config, credentials, -health, metrics, restarts -- and Vector does the transform. From dfe-engine's -side this app is indistinguishable from dfe-loader or dfe-receiver. +So Vector is kept, and wrapped. The Rust supervisor is PID 1 and Vector is its child. The supervisor supplies the production half -- config, credentials, health, metrics, restarts -- and Vector does the transform. ## Why a subprocess -Vector is not designed as a library and has no stable public API for embedding, -so the image downloads the upstream release binary at a pinned version and the -supervisor spawns it with `--config-dir`. Against compiling Vector in as a -crate, the deciding factor was upgrades: a subprocess swaps the binary, a crate -rebuilds against every new Vector version. +Vector is not designed as a library and has no stable public API for embedding, so the image downloads the upstream release binary at a pinned version and the supervisor spawns it with `--config-dir`. A Vector upgrade is then a version bump and a new image, where a compiled-in Vector would mean rebuilding against every Vector release. -That choice has one visible consequence. This is the only DFE Rust app whose -runtime image carries two binaries -- its own, which scalo's deployment contract -builds, and `vector`, which has no slot in that contract. The Dockerfile is -therefore generated by scalo and then spliced: `src/deployment.rs::emit_dockerfile()` -appends `src/vector-layer.dockerfile` with the version substituted in. The other -five DFE Rust apps are single-binary images and need no such override. +That choice has one visible consequence. This is the only DFE Rust app whose runtime image carries two binaries -- its own, which scalo's deployment contract builds, and `vector`, which has no slot in that contract. The Dockerfile is therefore generated by scalo and then spliced: `src/deployment.rs::emit_dockerfile()` puts `src/vector-layer.dockerfile` in before the generated `USER` line, with the version and the two archive checksums substituted. The other DFE Rust apps are single-binary images and need no such override. -The fragment is a real Dockerfile rather than a Rust string literal so hadolint -and shellcheck can read it. Shell embedded in `format!` is validated by nothing -until the image build runs. +The fragment is a real Dockerfile rather than a Rust string literal so hadolint and shellcheck can read it. Shell embedded in `format!` is checked by nothing until the image build runs. ## Who owns what -The boundary is the thing to get right, because most confusion here is a -question asked of the wrong layer. +Most confusion here is a question asked of the wrong layer. | Layer | Owns | |---|---| | Vector (the child) | Per-record transform work, the Kafka client on the `bus` transport, its own internal metrics | | This supervisor | Config assembly, process lifecycle and crash recovery, health reporting, the metrics merge, the direct-transport bridge, the deployment artefacts | -| dfe-engine | What the app IS -- it reads `dfe-infra/apps.yaml`, compiles the Helm values and holds the config in its registry | +| dfe-engine | What the app IS -- it reads dfe-infra's `apps.yaml`, compiles the app's routing into its chart values and holds the config in its registry | | dfe-infra | The image pin and the GitOps deployment | -Nothing about how this app is deployed is coded into dfe-engine. It is a `per_config` -app with `scale_deployed: true`, declared in `dfe-infra/apps.yaml` -- one -deployment per source config, not a singleton, and KEDA drives the replica -count. Its `default_in` list names only `docker-single`, so every other deploy -composition gets it only when a deployment asks for it. +Nothing about how this app is deployed is coded into dfe-engine. dfe-infra's `apps.yaml` declares it `per_config` with `scale_deployed: true` -- one deployment per source config, not a singleton, and KEDA drives the replica count. Its `default_in` list names only `docker-single`, so every other deploy composition gets it only when a deployment asks for it. ## The pieces @@ -68,116 +36,55 @@ flowchart LR subgraph CFG["Config engine"] L["loader"] --> V["validate"] --> G["generate"] --> W["wiring"] --> A["assembler"] end - A -->|"--config-dir"| VEC["Vector subprocess"] + A -->|config dir| VEC["Vector subprocess"] PM["process + lifecycle"] --> VEC - RL["reload (poll)"] --> A - VEC -->|"exporter, loopback"| MM["metrics merge"] + RLD["reload (poll)"] --> A + VEC -->|exporter on loopback| MM["metrics merge"] PM --> H["health registry"] MM --> OPS["ops port 9090"] H --> OPS ``` -The config engine turns a handful of big dials into a Vector config directory: -a numbered source file, the user's transform files, a sink file and an -observability file. The user authors only the `transforms:` block. `wiring` -parses the component labels and `inputs` references in those files and joins -`dfe_source` to `dfe_sink` through them, so a transform author does not hand-wire -the topology. +The config engine turns a handful of big dials into a Vector config directory: a global file, a source file, the user's transform files, a sink file and an observability file. The user authors only the `transforms:` and `enrichment_tables:` blocks. `wiring` parses the component labels and `inputs` references in those files and joins `dfe_source` to `dfe_sink` through them, so a transform author does not hand-wire the topology. -The process manager spawns Vector, forwards signals, and restarts it with -exponential backoff between one and sixty seconds. +The process manager spawns Vector, forwards SIGTERM at shutdown, and restarts it after a crash with exponential backoff from one to sixty seconds. -The bridge exists on the `direct` transport only. With no broker in the path the -two protocols do not meet: the rest of DFE speaks scalo's `Transport/Push` and -Vector speaks `vector.Vector/PushEvents`. The supervisor translates in both -directions, and both inner legs stay on loopback inside the one pod. On `bus` -the bridge does nothing and Vector talks to Kafka itself. +The bridge exists on the `direct` transport only. With no broker in the path the two protocols do not meet: the rest of DFE speaks scalo's `Transport/Push` and Vector speaks `vector.Vector/PushEvents`. The supervisor translates in both directions, and both inner legs stay on loopback inside the one pod. On `bus` the bridge does nothing and Vector talks to Kafka itself. ## Runtime invariants -These are the facts that cost someone a debugging session. None of them is -visible from reading one file. - -**Reload polls, and must.** The watcher polls rather than using inotify because -S3-backed mounts -- s3fs, goofys, Mountpoint for S3 -- generate no filesystem -notification events. Swapping in inotify would work on every developer laptop -and silently stop reloading in production. +**Reload polls, and must.** The watcher polls rather than using inotify because S3-backed mounts -- s3fs, goofys, Mountpoint for S3 -- generate no filesystem notification events. Inotify would work on every laptop and stop reloading in production with nothing logged. -**Only transforms hot-reload.** A transform change is validated, SIGHUPed into Vector, and reported from Vector's reload counters, since the signal succeeds even when Vector refuses it. Everything else needs a pod restart, and not by policy: the consumer, the Push listener, the bridge legs, the consumer group id, the metrics labels and the ops listener are all set at startup. A change that classifies as transforms-only when it is not logs a successful reload for work it never did. +**Only transforms hot-reload.** A transform change is validated, SIGHUPed into Vector, and reported from Vector's reload counters, since the signal succeeds even when Vector refuses it. Everything else is set at startup -- the consumer, the Push listener, the bridge legs, the group id, the ops listener -- and needs a pod restart. A reload compares the app's own config with the running one, so a restart-only change in the file blocks every later transform change until a restart or a revert. An edit to scalo's sections in the file is not compared: it logs a successful reload and waits for a restart. -**A push on the direct path is answered only once the next hop has it.** Both -bridge listeners are built armed, so a push is held until the pipeline releases -its records. A hop that refuses is retried until the hold runs out, and then the -push is answered `Unavailable` and its sender retries. The holds nest so an -inner hop is answered while the one around it can still answer its own sender: -18 s at the intake, 16.5 s on the leg back from Vector, and 15 s for each send -to the next stage. A record no retry would get through is released dropped and -counted rather than resent forever: one over the next stage's message-size -ceiling, or one Vector refused for good because a sink it feeds rejected it. +**A push on the direct path is answered only once the next hop has it.** Both bridge listeners are built armed, so a push is held until the pipeline releases its records. A hop that refuses is retried until the hold runs out, and then the push is answered `Unavailable` and its sender retries. The holds nest, 18 s at the intake, 16.5 s on the leg back from Vector and 15 s per send onward, so an inner hop is answered while the outer one can still answer its sender. A record no resend could deliver is not resent forever: one over the next stage's message-size ceiling goes to the DLQ, or is dropped and counted with the DLQ off, and one Vector refused for good is released dropped and counted. -**Only the intake sheds under pressure.** The leg back from Vector drains -everything the intake holds, so shedding it would stall the stage until the -holds expired into duplicates. Both listeners lease held bytes on the runtime's -memory guard. +**Only the intake sheds under pressure.** The leg back from Vector drains everything the intake holds, so shedding it would stall the stage until the holds expired into duplicates. Both listeners lease held bytes on the runtime's memory guard. -**A failed bridge leg stops the process.** Its senders would otherwise retry for -ever against a pod that stayed up, so the service shuts down and exits non-zero. +**A failed bridge leg stops the process.** Its senders would otherwise retry for ever against a live pod, so the service exits non-zero. **Transforms cannot reach the Kafka credentials.** Vector runs whatever a transform file declares, with read access to the credentials. So a transform file declares only `transforms` (never `lua`) and `enrichment_tables`, each table resolves inside `transforms.data_dir` after `..` and symlinks and outside every credential path (the config directory, a mounted `sasl.secret_dir`, a TLS key), and a `SECRET[...]` naming an assembler backend is refused. Vector also runs without the `DFE_TRANSFORM_VECTOR_*` variables. ## Health and metrics invariants -**Port 9598 is a loopback debug surface, not a scrape target.** Vector's own -`prometheus_exporter` binds it. The supervisor GETs it on scalo's metrics -interval and registers every sample on the registry scalo already serves and -pushes over OTLP, so one endpoint carries both. No second Prometheus target and -no PodMonitor selector are involved. The merge is also what makes the app's own -throughput counters non-zero, since Vector owns the Kafka client and the -supervisor never sees a record on the bus transport. - -**`/readyz` proves the subprocess exists, not that it carries traffic.** The gate -is that the child was still alive 500ms after spawn, which catches bad argv, an -unreadable config and a pod in crash-recovery backoff. Vector's own startup -routinely takes longer than that, so there is a window where the pod reports -ready and Vector is still coming up. Proving traffic would mean reading Vector's -API, which the supervisor does not do. +**Port 9598 is a loopback debug surface, not a scrape target.** Vector's own `prometheus_exporter` binds it. The supervisor GETs it on scalo's metrics interval and registers every sample on the registry scalo already serves and pushes over OTLP, so one endpoint carries both. No second Prometheus target and no PodMonitor selector are involved. The merge is also what makes the app's own throughput counters non-zero, since Vector owns the Kafka client and the supervisor never sees a record on the bus transport. + +**`/readyz` proves Vector runs and its sink moves, not that records flow.** The gate is that the child was still alive 500 ms after spawn, which catches bad argv, an unreadable config and a pod in crash-recovery backoff, and that the sink has not held records for `metrics.sink_stall_secs` without delivering any. A large program can take Vector longer than 500 ms to build, so the pod can report ready while Vector is still coming up. A pod whose broker is unreachable, with nothing held, stays Ready. So does a bus config with no topics, which idles with no Vector running. ## Build, chart and licence invariants -**The Vector version has one source.** The constant `VECTOR_VERSION` in `src/deployment.rs` is substituted into the Dockerfile's `ARG VECTOR_VERSION`, and both `VectorConfig::default()` and the contract's `default_config` read it for `vector.version`. `version_check: strict` refuses to start a pod when the config's version and the binary's disagree, so a deployment that sets `vector.version` itself has to move it with the image. dfe-infra's `versions.yaml` records the same release as `services.vector`, and dfe-fetcher's container Vector extractor runs that image. +**The Vector version has one source.** The constant `VECTOR_VERSION` in `src/deployment.rs` is substituted into the Dockerfile's `ARG VECTOR_VERSION`, and both `VectorConfig::default()` and the contract's `default_config` read it for `vector.version`. `vector.version_check: strict` refuses to start a pod when the config's version and the binary's disagree, so a deployment that sets `vector.version` itself has to move it with the image. dfe-infra's `versions.yaml` records the same release as `services.vector`, and dfe-fetcher's container Vector extractor runs that image. -**`password` is a `String` on purpose, not a `SensitiveString`.** The config -goes through a figment serialize-merge-deserialize round trip in -`apply_figment_env()`, and `SensitiveString` serialises as `***REDACTED***`, -which destroys the value in transit. The protection is elsewhere: the env var is -masked in logs, `Debug` redacts the field, and no credential is written into the -Vector config. The generated components name it `SECRET[.]` and -Vector's `directory` secret backend reads it from files: a mounted secret at -`sasl.secret_dir`, or owner-only files the assembler writes under -`/.secrets` from a credential given as text. The released chart -takes the second path: its Kafka Secret arrives in `DFE_TRANSFORM_VECTOR_{SOURCE,SINK}_SASL_*`, -and `` sits on the `run` writable path. Changing the type to look -safer breaks authentication. +**`password` is a `String` on purpose, not a `SensitiveString`.** The config goes through a figment serialize-merge-deserialize round trip in `apply_figment_env()`, and `SensitiveString` serialises as `***REDACTED***`, which destroys the value in transit. The protection is elsewhere: the env var is masked in logs, `Debug` redacts the field, and no credential is written into the Vector config. The generated components name it `SECRET[.]` and Vector's `directory` secret backend reads it from files: a mounted secret at `sasl.secret_dir`, or owner-only files the assembler writes under `/.secrets` from a credential given as text. The released chart takes the second path: its Kafka Secret arrives in `DFE_TRANSFORM_VECTOR_{SOURCE,SINK}_SASL_*`, and `` sits on the `run` writable path. -**Vector expands no `${VAR}`, and `vector validate` resolves no `SECRET[...]`.** Since 0.57 interpolation needs `--dangerously-allow-env-var-interpolation`, which the supervisor never passes, and `validate()` refuses a `${...}` in a SASL credential. Validate would hand the broker the literal `SECRET[...]` reference, so the supervisor runs it with `--no-environment`, and a missing secret file fails only when Vector starts -- which is why the assembler checks `secret_dir` itself. +**Vector expands no `${VAR}`, and `vector validate` resolves no `SECRET[...]`.** Since 0.57 interpolation needs `--dangerously-allow-env-var-interpolation` or `VECTOR_DANGEROUSLY_ALLOW_ENV_VAR_INTERPOLATION`, and the supervisor sets neither. `validate()` refuses a `${...}` in a SASL credential. Validate would hand the broker the literal `SECRET[...]` reference, so the supervisor runs it with `--no-environment`, and a missing secret file fails only when Vector starts -- which is why the assembler checks `secret_dir` itself. -**Vector is redistributed unmodified under MPL-2.0.** That carries two -obligations -- ship the licence text and tell recipients where the source is. -Both are discharged inside the image from the release archive's own `LICENSE`, -`NOTICE` and `licenses` tree, so the attribution always matches the exact binary -shipped. Do not replace that with a separate download. +Vector is redistributed unmodified under MPL-2.0, which carries two obligations -- ship the licence text and tell recipients where the source is. Both are discharged inside the image from the release archive's own `LICENSE`, `NOTICE` and `licenses` tree, so the attribution always matches the exact binary shipped. Do not replace that with a separate download. -**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, the writable paths `data` at `/var/lib/vector` and `run` at `/var/run/vector` under a read-only root, requests 100m/128Mi, limits 500m/512Mi and a 90 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`. `emit-contract` prints what the contract fixes: the startup budget, `kubernetes.io/h2c` on the Push port, the Kafka Secret mounted into the per-endpoint SASL names, the writable paths `data`, `run` and `dlq` under a read-only root, the requests and limits, and the grace period. -**The ScaledObject scales on CPU alone.** `contract()` sets `KafkaLagTrigger::disabled()`: consumer-group lag rises when a downstream stage breaks, and the library's lag trigger reads `config.kafka.*`, a block this app does not have, so the library refused to render it. A deployment adds a trigger with the library's `keda.extraTriggers` value. +**The ScaledObject scales on CPU alone.** `contract()` sets `KafkaLagTrigger::disabled()`: consumer-group lag rises when a downstream stage breaks, and the library's lag trigger reads `config.kafka.*`, a block this app does not have, so the library refuses to render it. A deployment adds a trigger with the library's `keda.extraTriggers` value. ## Where the transports differ -One deployment runs one transport and the transform you author is identical on -both. What changes is who hands the record over: on `bus` it is a Kafka topic at -each end and Vector uses its own `kafka` source and sink, on `direct` it is a -scalo Push listener in and a Push client out with Vector on `vector` components -over loopback. `templates/bus.yaml` and `templates/direct.yaml` are the whole -topology for each, runnable as they stand. In a DFE deployment the supervisor -generates the `sources:` and `sinks:` blocks, so a whole template mounted as a -transform file is refused. +One deployment runs one transport and the transform you author is identical on both. What changes is who hands the record over: on `bus` it is a Kafka topic at each end and Vector uses its own `kafka` source and sink, on `direct` it is a scalo Push listener in and a Push client out with Vector on `vector` components over loopback. `templates/bus.yaml` and `templates/direct.yaml` are the whole topology for each, and `vector validate` passes on both as they stand. In a DFE deployment the supervisor generates the `sources:` and `sinks:` blocks, so a whole template mounted as a transform file is refused. diff --git a/docs/internals.md b/docs/internals.md index 020fcbc..8f3a6f7 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -1,213 +1,23 @@ # dfe-transform-vector internals -## 1. Purpose - -dfe-transform-vector wraps Vector.dev as a managed subprocess to provide **Kafka -> transform -> Kafka** pipelines that the DFE platform manages, deploys and observes the same way as dfe-loader and dfe-receiver. - -Vector is powerful but opaque. This wrapper makes it behave like every other DFE service: the same health contract, metric names and scaling signal, a chart generated from its deployment contract, and the same Argo CD lifecycle. How the app is deployed is declared in dfe-infra's `apps.yaml`, not coded into dfe-engine. - ---- - -## 2. Architecture - -The pod runs the chart generated from `contract()` (section 9), with the config the deployment mounts. - -```text -┌──────────────────────────────────────────────────────────────────┐ -│ K8s Pod (Deployment) │ -│ │ -│ ┌────────────────────────────────────────────────────────────┐ │ -│ │ dfe-transform-vector (Rust, PID 1) │ │ -│ │ │ │ -│ │ ┌─────────────┐ ┌──────────────┐ ┌──────────────────┐ │ │ -│ │ │ Config │ │ Process │ │ Observability │ │ │ -│ │ │ Engine │ │ Manager │ │ Server │ │ │ -│ │ │ │ │ │ │ │ │ │ -│ │ │ • Load big │ │ • Spawn │ │ • /livez │ │ │ -│ │ │ dials │ │ vector │ │ • /readyz │ │ │ -│ │ │ • Generate │ │ • Signal │ │ • /metrics │ │ │ -│ │ │ source/ │ │ forwarding │ │ • scaling_ │ │ │ -│ │ │ sink YAML │ │ • Crash │ │ pressure │ │ │ -│ │ │ • Load user │ │ recovery │ │ │ │ │ -│ │ │ transforms│ │ • Health │ │ │ │ │ -│ │ │ • Wire DAG │ │ polling │ │ │ │ │ -│ │ │ • Validate │ │ │ │ │ │ │ -│ │ │ • Assemble │ │ │ │ │ │ │ -│ │ └──────┬──────┘ └──────┬───────┘ └──────────────────┘ │ │ -│ │ │ │ │ │ -│ │ │ config dir │ child process │ │ -│ │ ▼ ▼ │ │ -│ │ ┌────────────────────────────────────────────────────┐ │ │ -│ │ │ vector (subprocess) │ │ │ -│ │ │ --config-dir /var/run/vector/config/ │ │ │ -│ │ │ (no watcher of its own: reloads arrive as SIGHUP) │ │ │ -│ │ │ │ │ │ -│ │ │ Kafka source ──► transforms ──► Kafka sink │ │ │ -│ │ │ ──► internal_metrics ──► prometheus_exp │ │ │ -│ │ └────────────────────────────────────────────────────┘ │ │ -│ └────────────────────────────────────────────────────────────┘ │ -│ │ -│ Volumes: │ -│ • /etc/dfe-transform-vector/config.yaml (ConfigMap, big dials) │ -│ • /etc/dfe-transform-vector/transforms/ (empty dir from image) │ -│ • /var/run/vector/config/ (image dir — assembled Vector config) │ -└──────────────────────────────────────────────────────────────────┘ -``` - ---- - -## 3. Dead Letters - -On `direct`, the supervisor's bridge sends a record the next stage refuses for good to the DLQ `dlq` configures: scalo's `DlqConfig`, by default NDJSON files under `/var/spool/scalo/dlq`. With `dlq.enabled: false` such a record is dropped and counted in `pipeline_dead_letters_dropped_total`: `reason="too_large"` for one over the next stage's message-size ceiling, `reason="rejected"` for one the sink refused for good. On `bus`, Vector's own Kafka sink delivers and the bridge DLQ is not used. - ---- - -## 4. Health & Observability Contract - -### 4.1 Endpoints - -All DFE services expose the same three endpoints, on the one ops port, so a pod -has a single answer to "are you ready". dfe-transform-vector is no exception. - -| Endpoint | Port | Purpose | Response | -|---|---|---|---| -| `GET /livez` | 9090 | K8s liveness + startup probes | `200 OK` if the supervisor process is running | -| `GET /readyz` | 9090 | K8s readiness probe | `200 OK` only while the Vector child process is up | -| `GET /metrics` | 9090 | Prometheus scrape | Wrapper metrics + proxied Vector metrics | - -### 4.2 Liveness vs Readiness - -```text -/livez → always 200 if the Rust process is running (fast, no deps) -/readyz → 200 only when the lifecycle is Running or Reloading, which means: - 1. Config loaded, assembled and `vector validate`-clean - 2. The Vector child was spawned and still existed 500ms later - 3. Not in crash-recovery backoff -``` +Field-level depth under [architecture.md](architecture.md): how a config becomes the directory Vector reads, how reloads, crashes and shutdown run, what the ops port reports, and what the chart fixes. -The supervisor publishes its lifecycle into scalo's health registry, which the -ops listener consults per request -- so `/readyz` tracks the subprocess rather -than the supervisor that outlives it. - -A spawn returning success is not a running subprocess. The lifecycle only -reaches `Running` once the child has survived a short settle window -(`SPAWN_SETTLE`, 500ms), so a crash-on-start (bad argv, unreadable config) -reports 503 instead of a healthy start. - -**What the settle window does NOT prove.** The check is `try_wait()` returning -"still running" at t+500ms -- the process EXISTS. Vector's own startup routinely -takes longer than that, so `/readyz` can answer 200 while Vector is still -wiring up its topology and consuming nothing. The signal for "carrying -traffic" is the sink's progress in Vector's own metrics: `/readyz` also reports -not ready while the sink holds records and has delivered none of them for -`metrics.sink_stall_secs` (60 by default). Vector's API is off by default and, -when turned on, binds loopback only. - -During config reload, readiness stays healthy -- but only if it was healthy -already. `Reloading` counts as ready, so it is entered by compare-and-set from -`Running`: a transform change landing mid crash-backoff leaves the `Crashed` -state alone rather than advertising a subprocess that is not there. During crash -recovery, readiness returns 503 until Vector restarts successfully. - -### 4.3 Metrics - -**Wrapper metrics** (emitted by the Rust binary). Names are bare unless -`metrics.namespace` is set, and none is by default: - -```text -# DFE platform standard -pipeline_ready 1 # 1=ready, 0=not ready - -# Process lifecycle -transform_vector_lifecycle_state{state="running"} 1 -uptime_seconds 3600 - -# Subprocess health -crashes_total 0 -restarts_total 1 - -# Config management -transform_vector_config_reloads_total{result="success"} 5 -transform_vector_config_reloads_total{result="error"} 0 -transform_vector_config_reloads_total{result="rejected"} 0 -transform_vector_config_reloads_total{result="unconfirmed"} 0 -config_validation_errors_total 0 -``` - -`success` means Vector reported the reload. `error` (a config that does not -load or validate, a failed assembly or publish, a SIGHUP that cannot be sent, -or a reload Vector refused) and `rejected` (a change outside the transform -files, which needs a restart) mean the running Vector did not take the change. -`unconfirmed` means the change is published but Vector did not report whether -it runs it. - -Vector, not the wrapper, owns the Kafka client, so the record counters -(`records_received_total`, `records_processed_total`, `records_delivered_total`) -are derived from Vector's own component counters by the scrape below. - -**Merged Vector metrics** (scraped from Vector's `prometheus_exporter` on -loopback `metrics.vector_metrics_address`, registered on the same registry -:9090 serves and scalo pushes over OTLP): - -```text -# Names and labels as Vector emits them -vector_component_received_events_total{component_id="dfe_source",component_type="kafka"} 150000 -vector_component_sent_events_total{component_id="dfe_sink",component_kind="sink"} 149950 -vector_component_errors_total{component_id="parse"} 50 -vector_buffer_byte_size{component_id="dfe_sink"} 1048576 -# ... every other Vector internal metric -``` - -Counters register as counters via `absolute`, carried across Vector restarts: -a total that falls is a restarted Vector, and the count it had reached becomes -a base the new count adds to. A histogram arrives -pre-aggregated, so each cumulative `_bucket` re-registers as a counter carrying -its `le` label: the metrics crate's `Histogram` takes observations, and -replaying buckets through it would change the numbers. - -One scrape of :9090 therefore carries wrapper state and Vector pipeline state, -and 9598 stays a loopback debug surface with no reader outside the pod. - -### 4.4 Scaling +## The config -The ScaledObject scales on CPU alone. `contract()` sets `KafkaLagTrigger::disabled()`, -because consumer-group lag rises when a downstream stage breaks and more replicas -cannot fix that. A deployment adds a trigger with the library chart's -`keda.extraTriggers` value. `spawn_circuit_gate_task` pins scalo's scaling -pressure to 0 while the Vector subprocess is down. +The app's own config layers in this order, each over the one before: -In a DFE deployment, dfe-infra's values set the CPU target to 70% and add no -pressure trigger in any profile, so this app scales 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. +1. The defaults, which [config-schema.yaml](config-schema.yaml) lists. +2. The config file: `--config`, else `config.yaml` or `config.yml` in the working directory. +3. `DFE_TRANSFORM_VECTOR_
__`, which reaches any key. A list takes brackets: `DFE_TRANSFORM_VECTOR_SOURCE__BROKERS='[a:9092,b:9092]'`. +4. The flat names, such as `DFE_TRANSFORM_VECTOR_SOURCE_BROKERS=a:9092,b:9092`, `_SINK_TOPIC`, `_TRANSFORMS_DIR`, `_VECTOR_BINARY` and the SASL pairs. -scalo's runtime serves the pressure at `/scaling/pressure` and writes it into the -`scaling_pressure` gauge. With no pressure components and no memory reading fed -to it, the value is 0 while Vector runs too. +The working directory's `.env` is read into the environment before the env layers, and a variable already set wins. `--log-level`, `--log-format` and `--metrics-addr`, or `LOG_LEVEL`, `LOG_FORMAT` and `METRICS_ADDR`, outrank `logging.*` and `metrics.address`. Nothing reads `DFE_TRANSFORM_VECTOR_CONFIG`: name the file with `--config`, before any subcommand. ---- - -## 5. Config Engine - -### 5.1 Big-Dial Config (Rust side) - -The Rust binary loads its own config via the standard scalo 7-layer cascade: - -```text -1. CLI args (highest priority) -2. Env vars (DFE_TRANSFORM_VECTOR_*) -3. .env file -4. settings.{env}.yaml -5. settings.yaml -6. defaults.yaml -7. Hard-coded defaults (lowest priority) -``` +The same file seeds scalo's own cascade, so scalo's sections in it apply too: `version_check`, `scaling.enabled`, `scaling.memory_gate_threshold` and `metrics.namespace`. `dfe_source` fills in `_land`, `_load` and the consumer group, `dfe-transform-vector-` or `dfe-transform-vector-` without a pipeline name, wherever the config leaves them at their defaults. ```yaml -# /etc/dfe-transform-vector/config.yaml — the big dials pipeline: name: syslog-enrichment - source: brokers: ["kafka:9092"] # or DFE_TRANSFORM_VECTOR_SOURCE_BROKERS topics: @@ -216,507 +26,144 @@ source: sasl: mechanism: scram_sha_512 # A mounted Secret holding `username` and `password` files. Vector reads - # them through its directory secret backend; it expands no ${VAR}. + # them through its directory secret backend, and expands no ${VAR}. secret_dir: /var/run/secrets/dfe-kafka tls: enabled: true - sink: brokers: ["kafka:9092"] # or DFE_TRANSFORM_VECTOR_SINK_BROKERS topic: enriched_syslog_land key_field: ".org_id" - encoding: json compression: zstd - -transforms: - dir: /etc/dfe-transform-vector/transforms - -vector: - binary: /usr/local/bin/vector - data_dir: /var/lib/vector - api_enabled: false # loopback only when turned on - log_level: info - -metrics: - address: "0.0.0.0:9090" # also serves /livez and /readyz - -scaling: - pressure_threshold: 0.8 -``` - -### 5.2 Config Assembly - -The config engine takes the big dials and produces a Vector config directory: - -```text -/var/run/vector/config/ - 00_global.yaml # data_dir, api switch, SASL secret backends - 00_source.yaml # Generated from source big dials - 50_000_parse.yaml # User transforms, flat (Vector --config-dir doesn't recurse) - 50_001_enrich.yaml # Prefixed with 50_NNN_ for stable sort order - 50_002_filter.yaml - 90_sink.yaml # Generated from sink big dials - 99_observability.yaml # Generated: internal_metrics + prometheus_exporter -``` - -**Generated source (`00_source.yaml`):** - -```yaml -sources: - dfe_source: - type: kafka - bootstrap_servers: "kafka-1:9092,kafka-2:9092" - topics: - - raw_syslog_land - group_id: dfe-transform-vector-syslog-enrichment - decoding: - codec: json - sasl: - enabled: true - mechanism: SCRAM-SHA-512 - username: SECRET[dfe_source_sasl.username] - password: SECRET[dfe_source_sasl.password] - tls: - enabled: true - librdkafka_options: - partition.assignment.strategy: cooperative-sticky -``` - -**Generated sink (`90_sink.yaml`):** - -```yaml + sasl: # each endpoint has its own SASL and TLS + mechanism: scram_sha_512 + secret_dir: /var/run/secrets/dfe-kafka + tls: + enabled: true transforms: - dfe_size_cap: # drops a record over message.max.bytes - type: filter - inputs: [""] # Auto-wired - condition: length(encode_json(.)) <= 16777088 -sinks: - dfe_sink: - type: kafka - inputs: ["dfe_size_cap"] - bootstrap_servers: "kafka-1:9092,kafka-2:9092" - topic: enriched_syslog_land - key_field: ".org_id" - encoding: - codec: json - compression: zstd - message_timeout_ms: 0 # an outage holds rather than rejects - sasl: - enabled: true - mechanism: SCRAM-SHA-512 - username: SECRET[dfe_sink_sasl.username] - password: SECRET[dfe_sink_sasl.password] - tls: - enabled: true - librdkafka_options: - enable.idempotence: "true" - message.max.bytes: "16777216" # scalo's record ceiling unless a layer sets one -``` - -The size cap and `message.max.bytes` read one value: scalo's pipeline-wide record ceiling of 16 MiB, the one every other stage's producer and every topic scalo creates carry, unless the central profile or `sink.librdkafka_options` sets `message.max.bytes`. The cap keeps 128 bytes under it for the key, headers and framing. librdkafka's own default of 1,000,000 bytes would refuse a record the rest of the chain carries. - -**Generated observability (`99_observability.yaml`):** - -```yaml -sources: - internal_metrics: - type: internal_metrics - -sinks: - prometheus_exporter: - type: prometheus_exporter - inputs: - - internal_metrics - address: "127.0.0.1:9598" # metrics.vector_metrics_address -``` - -### 5.3 DAG Auto-Wiring - -The wrapper understands the pipeline topology and wires components automatically: - -1. **Source label is always `dfe_source`** — canonical, predictable -2. **Sink label is always `dfe_sink`** — canonical, predictable -3. **First transform**: a transform with no or empty `inputs` reads `dfe_source`, and an input written `source` is rewritten to `dfe_source` in the file Vector reads -4. **Intermediate transforms**: left as authored (user controls the chain). An input names a component (`parse`) or one of its named outputs (`split.accepted`) -5. **Last transforms**: every output nothing reads is injected as `dfe_sink.inputs`. A `route` or `exclusive_route` has named outputs only, so each unread route and its `_unmatched` reach the sink -6. **Extra sources/sinks**: refused, with every other top-level key bar `transforms` and `enrichment_tables` - -**Validation rules:** - -- All `inputs` references must resolve to a defined component, and to an output that component has -- No orphaned components (every component must be reachable from a source) -- No cycles in the DAG -- At least one path from `dfe_source` to `dfe_sink` -- `vector validate --config-dir` as final gate (catches VRL syntax, type mismatches) - -### 5.4 Hot-Reload - -```text -Config change detected (file watcher or SIGHUP) - │ - ├─ Transform file change (safe) - │ ├─ Re-read transform YAMLs, re-run DAG wiring + validation into .next - │ ├─ Check every enrichment table file can be read, then `vector validate` .next - │ ├─ If invalid: log error, increment metric, drop .next (Vector's dir untouched) - │ ├─ If valid: swap .next and in one rename → SIGHUP to Vector child - │ ├─ Read vector_reloaded_total / component_errors_total{error_code} for the outcome - │ ├─ If refused: log error, increment metric, swap the running config dir back - │ └─ Emit config_reloads_total{result=success|error|unconfirmed} - │ - └─ Any other config change (unsafe — requires pod restart) - ├─ Log warning: "config change requires pod restart" - ├─ Emit config_reloads_total{result="rejected"} - └─ Keep old config running — operator must restart the pod -``` - -The swap is one `renameat2(RENAME_EXCHANGE)`, so a Vector restarted mid-reload reads the old directory or the new one, never a mix. A filesystem that cannot swap directories, such as an overlay image layer with no volume mounted at `/var/run/vector`, gets the validated files copied in place instead, which is not atomic. - ---- - -## 6. Subprocess Manager - -### 6.1 Lifecycle States - -```text -Initialising ──► Validating ──► Starting ──► Running ──► ShuttingDown - │ - ├──► Reloading ──► Running - │ - ▼ - Crashed - (wrapper restarts Vector with backoff; - K8s handles pod-level restart) -``` - -### 6.2 Signal Handling - -```text -Wrapper (PID 1) Vector (child) - │ │ - │◄──── SIGTERM (from kubelet) ──────│ - │ │ - │──── SIGTERM ──────────────────────►│ - │ │ - │ (Vector drains: stops sources, │ - │ flushes buffers, exits) │ - │ │ - │◄──── exit(0) ─────────────────────│ - │ │ - │ cleanup (metrics flush, etc.) │ - │ exit(0) │ -``` - -**Timeouts:** - -- `terminationGracePeriodSeconds` in K8s: **65s** -- Wrapper sends SIGTERM to Vector, waits: **55s** -- If Vector doesn't exit in 55s: SIGKILL child, then exit(1) -- Remaining 10s: wrapper cleanup + K8s buffer - -### 6.3 Crash Recovery - -```rust -// Pseudocode -loop { - let child = spawn_vector(&config_dir)?; - let start_time = Instant::now(); - - match child.wait().await { - Ok(status) if status.success() => break, // clean exit (SIGTERM) - - Ok(status) => { - metrics.crashes_total.inc(); - warn!("Vector exited with {status}, restarting in {backoff}"); - - if start_time.elapsed() > HEALTHY_THRESHOLD { - backoff.reset(); // was stable, reset backoff - } - - sleep(backoff.next()).await; - metrics.restarts_total.inc(); - } - - Err(e) => { - error!("Vector process error: {e}"); - sleep(backoff.next()).await; - } - } -} + dir: /etc/dfe-transform-vector-transforms # a file set's mountPath, see The chart ``` -**Backoff schedule:** 1s → 2s → 4s → 8s → 16s → 32s → 60s (cap) -**Reset after:** 5 minutes of healthy running -**Max crashes:** unlimited by default (K8s pod restart handles the outer loop) - ---- +## The assembled directory -## 7. Vector Binary Version Management +`dfe-transform-vector --config assemble ` writes what `run` hands Vector, without starting it. -A key robustness concern: the Vector binary in the container must be tracked, pinned, and updatable without rebuilding the wrapper. - -### 7.1 Version Pinning - -The wrapper's config declares a **pinned Vector version**: - -```yaml -vector: - version: "" # Expected version, default deployment::VECTOR_VERSION - version_check: warn # strict | warn | disabled - binary: /usr/local/bin/vector -``` - -**At startup**, the wrapper runs `vector --version`, parses the output, and compares: - -| `version_check` | Mismatch behavior | +| File | What it holds | |---|---| -| `strict` | Refuse to start. Exit with error. Config or image is wrong. | -| `warn` | Log warning, emit metric, continue. For dev/staging. | -| `disabled` | Skip check entirely. | - -Version check failures are logged at startup. No runtime version metrics are -emitted. - -### 7.2 Version Update Strategy - -Vector upgrades are a **container image concern**, not a wrapper code concern. This is the key advantage of the subprocess model. - -```text - Wrapper Image (Rust) Vector Binary - ───────────────────── ───────────── -Release cadence: When wrapper changes When Vector releases -Build trigger: Code change in this repo Upstream release / security fix -Update path: New wrapper image tag New base image layer -Coupling: None — wrapper is version- Binary in /usr/local/bin/ - agnostic (validates via pinned via Dockerfile ARG - config pin) -``` - -**Dockerfile pattern:** +| `00_global.yaml` | `timezone: UTC`, `data_dir`, the API switch, one `directory` secret backend per credential source | +| `00_source.yaml` | `dfe_source`: a `kafka` source on bus, carrying the librdkafka layers in [librdkafka.md](librdkafka.md), or a `vector` source on `bridge.to_vector` on direct | +| `50_NNN_` | Each transform file, in filename order, re-serialised with its wired `inputs` | +| `90_sink.yaml` | `dfe_sink`: on bus a `kafka` sink behind the `dfe_size_cap` filter, on direct a `vector` sink dialling `bridge.from_vector` | +| `99_observability.yaml` | `internal_metrics` into a `prometheus_exporter` on `metrics.vector_metrics_address` | +| `.secrets/` | Owner-only files for a SASL credential given as text | -The Dockerfile is generated via `emit-dockerfile` from the `DeploymentContract`. -CI builds the Rust binary externally (cross-compile amd64 + arm64), then the -Dockerfile copies it into the runtime image alongside the Vector binary. +Credentials appear only as references: `SECRET[dfe_source_sasl.username]` for a `secret_dir`, `SECRET[dfe_credentials.source_sasl_username]` for a credential given as text. The assembler refuses a `secret_dir` that lacks its `username` or `password` file. -```dockerfile -# Runtime image — CI builds the wrapper binary externally -FROM debian:trixie-slim -ARG VECTOR_VERSION= -ARG TARGETARCH +`dfe_size_cap` drops a record whose encoded length is over the producer's `message.max.bytes` less 128 bytes for the key, headers and framing. `message.max.bytes` is scalo's pipeline-wide record ceiling of 16 MiB unless the central profile or `sink.librdkafka_options` sets it. librdkafka's own default of 1,000,000 bytes would refuse records the rest of the chain carries. -# Wrapper binary (built by CI, copied in) -COPY dfe-transform-vector /usr/local/bin/dfe-transform-vector +The wiring follows these rules: -# Vector binary downloaded inside the build, per $TARGETARCH, with its -# LICENSE / NOTICE / licenses tree copied from the same release archive -RUN curl -fsSL "https://github.com/vectordotdev/vector/releases/download/v${VECTOR_VERSION}/..." +- The source is always `dfe_source` and the sink always `dfe_sink`. +- A transform with no or empty `inputs` reads `dfe_source`, and an input written `source` becomes `dfe_source` in the file Vector reads. +- An input names a component (`parse`) or one of its named outputs (`split.accepted`). A `route` or `exclusive_route` has named outputs only, so a bare reference to one is refused. +- Every output nothing reads feeds the sink, through `dfe_size_cap` on bus, each unread route and its `_unmatched` included. +- A transform file declares only `transforms` and `enrichment_tables`. A `sources:` key, a `lua` transform, or a table outside `transforms.data_dir` or on a credential path is refused. +- Every `inputs` reference must resolve, every transform must be reachable from `dfe_source`, and the graph must have no cycle. The error names the component: `component 'enrich' references undefined input 'parser'`. +- Once every enrichment table file proves readable, `vector validate --no-environment` over the directory is the last gate. -RUN mkdir -p /var/lib/vector /var/run/vector/config /etc/dfe-transform-vector/transforms -USER appuser -ENTRYPOINT ["dfe-transform-vector"] -``` - -**To update Vector without touching wrapper code:** +## Hot reload -1. Change `VECTOR_VERSION` in `src/deployment.rs` — that const is the ONLY - pin. Do NOT edit the Dockerfile: it is autogenerated, and - `checked_in_dockerfile_matches_emit_dockerfile` fails if it drifts. -2. Regenerate: `cargo run -- emit-dockerfile > Dockerfile` - Then move `services.vector` in dfe-infra's `versions.yaml` to the same - release (its tag is `-alpine`). dfe-fetcher's container Vector - extractor runs that image, so the two must name one release. -3. Rebuild image — only the vector and runtime stages rebuild (layer cache) -4. CI runs integration tests against the new Vector version -5. Deploy +The watcher polls the config file and the `.yaml` and `.yml` files of `transforms.dir`, or the listed `transforms.files`, every `reload.poll_interval_secs` (30 by default). A SIGHUP to the supervisor starts the same reload. Enrichment table files are not watched, and with `reload.enabled: false` nothing is, and a SIGHUP is logged and ignored. -The default config's `vector.version` needs no separate edit — the loader -reads `deployment::VECTOR_VERSION`, so it follows step 1 on its own. +1. The config is re-read and validated. +2. A change to the app's own keys outside the `transforms` block is refused. The watcher logs `non-transform config changed -- requires pod restart`, counts `rejected`, and keeps the running config. The change stays in the file, so every later transform change is refused too until the pod restarts or the edit is reverted, which lets the next transform change reload. scalo's sections in the file are read once at startup and not compared: an edit to one needs a restart, and the watcher reloads the transforms and counts `success` without applying it. +3. A transform change is assembled into `.next`, its table files checked and `vector validate` run there. A failure is logged and counted `error`, and the directory Vector reads is untouched. +4. The two directories swap in one `renameat2(RENAME_EXCHANGE)`, so a Vector restarted mid-reload reads the old one or the new one. A filesystem that cannot swap directories gets the files copied in place, which is not atomic. +5. Vector gets a SIGHUP, and its exporter is read for the outcome: `vector_reloaded_total` counts `success`, a refusal in `vector_component_errors_total` counts `error` and swaps the old directory back, and no answer within 60 s counts `unconfirmed`. -### 7.3 Compatibility Testing +A failed attempt is retried on the same files after one poll interval, doubling to five minutes. `transform_vector_config_reloads_total{result}` carries `success`, `error`, `rejected` and `unconfirmed`. scalo's own `config_reloads_total{result}` counts `success` and `error`, a rejection included. -Each CI build validates the wrapper against the pinned Vector version: +In a chart deployment an edit to the config or a file set rolls the pods anyway, because the chart hashes both into the pod template. -```yaml -# CI step -- name: Vector compatibility test - run: | - vector --version # Verify binary exists - # Generate a test config from fixture big dials - dfe-transform-vector assemble --config test/fixtures/basic.yaml --output /tmp/test-config/ - # Validate with the actual Vector binary - vector validate --config-dir /tmp/test-config/ - # Start Vector, wait for health, stop - dfe-transform-vector --config test/fixtures/basic.yaml & - sleep 5 - curl -f http://localhost:9090/readyz - kill %1 -``` +## The Vector subprocess -### 7.4 Version Drift Detection - -Version drift is detected at startup via the `version_check` config setting. The -wrapper runs `vector --version`, parses the output, and compares against the -pinned version. In `strict` mode, a mismatch prevents startup. In `warn` mode, -it logs and continues. Periodic release checking is not implemented — version -tracking is a container image pipeline concern. - ---- - -## 8. Robustness - -### 8.1 Failure Modes & Mitigations - -| Failure Mode | Detection | Mitigation | -|---|---|---| -| **Bad config YAML** | `vector validate` returns exit 78 | Refuse to start. Clear error in logs + metric. Old config kept on reload. | -| **Broken DAG wiring** | Wrapper DAG analysis before validate | Detailed error: "transform 'enrich' references unknown input 'parser' (did you mean 'parse'?)" | -| **Vector crash** | Child process exits non-zero | Restart with exponential backoff. Readiness goes unhealthy. Metrics track crash count. | -| **Vector hang** | Only through the sink: records held and none delivered for `metrics.sink_stall_secs` | Readiness goes unhealthy. Nothing restarts a hung Vector, and liveness stays 200. | -| **OOM** | Child killed by cgroup (exit 137) | Restart with backoff, counted as a crash. The scaling circuit opens, pinning scaling pressure to 0 until Vector runs again. | -| **Kafka unreachable** | Vector consumer lag stops advancing | Scaling pressure stays low (no lag movement). Vector's own retry handles reconnection. | -| **Config reload: new config invalid** | Enrichment table files readable, then `vector validate` on new config | Keep old config running, and put its config dir back. Log error. Emit `config_reloads_total{result="error"}`. | -| **Config reload: Vector rejects SIGHUP** | Vector keeps old config and counts `component_errors_total{error_code="reload"}` | Wrapper reads the counter, emits `config_reloads_total{result="error"}` and puts back the config dir Vector runs. | -| **Version mismatch** | Startup version check | `strict` mode: refuse to start. `warn` mode: log + metric. | -| **Disk buffer corruption** | Vector exits on startup | Wrapper restarts Vector. If persistent, Vector's WAL recovery handles it. | - -### 8.2 Graceful Degradation - -The wrapper prioritizes **availability over correctness of configuration updates**: - -1. **Config reload fails** → old config keeps running (no disruption) -2. **Vector crashes** → readiness unhealthy, K8s stops routing traffic, wrapper restarts Vector -3. **Wrapper crashes** → K8s restartPolicy restarts the pod (outer safety net) -4. **Both crash** → K8s handles it — same as any other pod failure - -The two-layer restart model (wrapper restarts Vector, K8s restarts wrapper) provides defense in depth that bare Vector-in-a-pod does not have. - -### 8.3 Startup Sequence - -```text -1. Load config (cascade) -2. Validate config (Pydantic-equivalent in Rust) -3. Check Vector binary exists at configured path -4. Check Vector version matches pin (if strict) -5. Generate source/sink YAML from big dials -6. Load user transform YAMLs -7. Run DAG wiring + validation -8. Assemble config directory -9. Run `vector validate --no-environment --config-dir` (the broker is Vector's to reach, at runtime) -10. Publish the lifecycle into scalo's health registry (readiness: NOT READY) -11. Spawn Vector child process -12. Wait out the settle window (`SPAWN_SETTLE`, 500ms) -13. Child still alive → lifecycle Running, so readiness reads READY -14. Enter steady state (monitor child, watch config files) +```mermaid +stateDiagram-v2 + [*] --> Initialising + Initialising --> Validating + Validating --> Starting + Starting --> Running: alive 500 ms after spawn + Running --> Reloading + Reloading --> Running + Starting --> Crashed + Running --> Crashed + Crashed --> Starting: after backoff + Running --> ShuttingDown + ShuttingDown --> [*] ``` -The ops listener is scalo's, started by `ServiceRuntime` before step 1; the -wrapper never binds a port of its own. See §4.2 for what the settle window -does and does not prove. - -If any step 1–9 fails, the wrapper exits immediately with a clear error. Steps 1–9 are **pre-flight checks** — nothing runs until they all pass. This means a bad config never reaches Vector. - -### 8.4 Data Safety - -Vector's Kafka source with `acknowledgements: true` only commits offsets after downstream sinks confirm delivery. This means: - -- **At-least-once delivery** is maintained through crashes and restarts -- Disk buffers (WAL in `data_dir`) survive pod restarts via PVC -- No data loss on graceful shutdown — Vector drains buffers before exiting -- On crash: some messages may be reprocessed (Kafka re-delivers from last committed offset) - -The wrapper does not touch the data path — it only manages config and process lifecycle. +Startup runs in this order: the config loads and validates, the runtime binds the ops listener, `/readyz` starts reporting the lifecycle, `vector --version` is checked against `vector.version`, the directory is assembled, `vector validate` passes, the bridge binds its listeners on direct, and Vector is spawned. A failure before the spawn stops the process with its error. -### 8.5 Memory Backpressure (MemoryGuard) +Any exit Vector was not asked for is a crash, exit 0 included. `crashes_total` counts it, the lifecycle reads `Crashed`, and Vector is respawned after a backoff that starts at 1 s, doubles to 60 s and resets once a run has lasted five minutes. There is no crash limit, and `restarts_total` counts each respawn. One Vector runs per pod, and replicas scale out. -**Not applicable to this project.** The `scalo` `MemoryGuard` provides -cgroup-aware memory backpressure for services that buffer data in-process -(dfe-receiver, dfe-loader, dfe-transform-vrl, dfe-fetcher, dfe-archiver). +On SIGTERM or SIGINT the intake stops, Vector gets a SIGTERM and 55 s to exit before a SIGKILL, the bridge drains for up to 15 s, and the process exits 0. A failed bridge leg runs the same shutdown and exits non-zero. -dfe-transform-vector does not buffer data — Vector manages all Kafka I/O, -buffering, and memory allocation as an independent subprocess. The Rust -wrapper's own memory footprint is constant and negligible (config structs, -HTTP servers, prometheus registry). Vector's memory is controlled via its -own `buffer` config (`max_events`, `max_size`) and container resource limits. +At startup the supervisor also runs `vector --version` and compares the result with `vector.version`. `strict` and `warn` emit a security data-quality event on a mismatch, and no metric records it. -If the wrapper ever adds its own data buffering, `MemoryGuard` should be -adopted at that point using pattern D (transport backpressure) from the -DFE remediation guide. - ---- - -## 9. Helm Chart - -### 9.1 How the Chart Is Built +| `vector.version_check` | On a mismatch | +|---|---| +| `strict` (the default) | Logs it and refuses to start | +| `warn` | Logs a warning and carries on | +| `disabled` | Runs no check, not even `vector --version` | -No chart is committed. 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`. The library renders the Deployment, Service, ConfigMap, ServiceAccount and KEDA ScaledObject, and its README documents the values. +## Health and metrics -### 9.2 What the Contract Fixes +| Path | Answers | +|---|---| +| `/livez` | 200 while the ops listener answers | +| `/readyz` | 200 or 503, below | +| `/metrics` | The supervisor's metrics and every merged `vector_*` | +| `/scaling/pressure` | scalo's scaling pressure | -- Probes on 9090 (`/livez`, `/readyz`) with a 120 s startup budget. -- The Push port 6000, `appProtocol: kubernetes.io/h2c`, only while `config.source.transport` is `direct` or `grpc`. -- The Kafka Secret's `username` and `password` in `DFE_TRANSFORM_VECTOR_{SOURCE,SINK}_SASL_{USERNAME,PASSWORD}`. The assembler writes them to owner-only files under `vector.config_dir`, which Vector reads through its `directory` secret backend. -- Writable paths `data` at `/var/lib/vector` and `run` at `/var/run/vector`, beside the library's `/tmp`, under a read-only root filesystem. -- Requests 100m/128Mi, limits 500m/512Mi, a 90 s grace period, uid and gid 1000. -- KEDA on CPU at 80 %, 1 to 10 replicas, no Kafka lag trigger. +All of them are on `metrics.address`, 9090 by default. `/readyz` answers 503 `{"status":"starting"}` until the runtime hands the service over. A bus config with no topics then idles: 200, `pipeline_idle` 1, no Vector, until the config file names a topic. Once running, it answers 503 unless the lifecycle is `Running` or `Reloading`, and while the sink has held records for `metrics.sink_stall_secs` (60 by default) without delivering any. On direct the bridge's gRPC transports report there too. `Reloading` is entered only from `Running`, so a change landing mid crash-backoff cannot turn it to 200. -The mounted config is the chart's `config` with `configOverrides` merged over it. The contract's `default_config` is not merged in, so the app's own defaults apply to anything a deployment leaves unset. The library mounts the ConfigMap over the config file's whole directory unless `configMount.subPath` is set, and that hides the image's `/etc/dfe-transform-vector/transforms`. +Metric names are bare unless `metrics.namespace` is set, and `dfe-transform-vector metrics-manifest` prints the catalogue. The supervisor's own: -### 9.3 Parity With dfe-loader/dfe-receiver Charts +- `transform_vector_lifecycle_state{state}`, 1 on the current state, and `pipeline_ready`. +- `crashes_total`, `restarts_total` and `uptime_seconds`, since the service started. +- `transform_vector_config_reloads_total{result}`, `config_reloads_total{result}` and `config_validation_errors_total`. +- `transform_vector_scrape_failures_total` and `transform_vector_sink_errors_total`. +- `records_received_total`, `records_processed_total`, `records_delivered_total` and `records_error_total`, derived from Vector's component counters on `dfe_source`, `dfe_sink` and `dfe_size_cap`. -All three are generated on the same scalo-service library chart, so they share the same features: +The scrape reads Vector's exporter on scalo's metrics interval and re-registers every sample. A counter is set with `absolute` and carried across Vector restarts: a total that falls is a new Vector, and the old count becomes a base. A histogram arrives pre-aggregated, so each `_bucket` registers as a counter with its `le` label. A gauge Vector stops reporting is zeroed after `metrics.vector_metrics_expiry_ticks` scrapes, a counter is forgotten after 240, and at most 50,000 series merge. -| Feature | dfe-loader | dfe-receiver | dfe-transform-vector | -|---|---|---|---| -| Workload type | Deployment | Deployment | Deployment | -| Health probes | /livez, /readyz | /livez, /readyz | /livez, /readyz | -| Metrics port | 9090 | 9090 | 9090 | -| ConfigMap | config.yaml | config.yaml | config.yaml | -| KEDA ScaledObject (contract) | CPU, no Kafka lag trigger | CPU, no Kafka lag trigger | CPU, no Kafka lag trigger | +The contract's ScaledObject scales on CPU and has no Kafka lag trigger. dfe-infra's values set a CPU target and, on its single, scale and mesh profiles, add a pressure trigger through `keda.extraTriggers`. The `scaling_pressure` this app publishes stays 0 while Vector runs, since nothing feeds it a component or a memory reading, and `spawn_circuit_gate_task` pins it to 0 while Vector is down. ---- +## Delivery and dead letters -## 10. Multi-Instance Deployment +With `source.acknowledgements.enabled`, the default, delivery is at-least-once on both transports. On bus the sink's acknowledgements hold Vector's source offsets until delivery, so a crash redelivers from the last committed offset. On direct a push is answered once the next hop has it. With acknowledgements off, both happen at receipt. -Like dfe-loader, multiple instances of dfe-transform-vector can run simultaneously for different pipelines: +On direct a record over the next stage's message-size ceiling goes to scalo's file DLQ, NDJSON under `/var/spool/scalo/dlq` on the contract's `dlq` writable path. With `dlq.enabled: false` it is dropped and counted in `pipeline_dead_letters_dropped_total{reason="too_large"}`. A record Vector refuses for good is dropped and counted under `reason="rejected"`, whatever `dlq` says. On bus no DLQ runs: the size cap's drops count under `too_large` and the Kafka sink's under `rejected`. Off the image the DLQ directory is not writable, so a direct run logs `DLQ enabled but no backend started` and leaves dead letters unacknowledged at their source: point `dlq.file.path` at a directory you can write. -```text -transform-vector-syslog # Kafka(raw_syslog) → parse+enrich → Kafka(enriched_syslog) -transform-vector-netflow # Kafka(raw_netflow) → decode+geoip → Kafka(enriched_netflow) -transform-vector-dns # Kafka(raw_dns) → parse+filter → Kafka(enriched_dns) -``` +On direct both bridge listeners lease the bytes they hold on scalo's memory guard. On bus Vector holds the records, bounded by its `buffer` settings and the container's memory limit. -dfe-infra's `apps.yaml` declares the app `multiplicity: per_config`, so each source config gets its own: +`vector.data_dir` sits on the contract's `data` writable path, an emptyDir unless `writablePaths.data.persistence.enabled` gives it a claim, so a disk buffer outlives a Vector restart but not a pod. The supervisor logs `DISK BUFFER HAS NO DURABLE STORAGE` at startup when a disk buffer sits on ephemeral storage. -- Config file: `transform-vector-syslog.yaml` -- Kafka consumer group: `dfe-transform-vector-syslog` -- Helm release: separate Deployment -- KEDA ScaledObject: independent scaling +## Upgrading Vector -The app is in the default composition of `docker-single` only. Elsewhere an operator adds an instance per source. +`VECTOR_VERSION` and the two archive checksums in `src/deployment.rs` are the only pin. The Dockerfile, the default `vector.version`, the contract's `default_config` and `scripts/fetch-vector.sh` all read the constant. ---- +1. Change `VECTOR_VERSION`, and `VECTOR_SHA256_AMD64` and `VECTOR_SHA256_ARM64` from Vector's published `SHA256SUMS` for that release. +2. Regenerate the Dockerfile: `cargo run --bin dfe-transform-vector -- emit-dockerfile > Dockerfile`. `checked_in_dockerfile_matches_emit_dockerfile` fails while it drifts. +3. Move `services.vector` in dfe-infra's `versions.yaml` to the same release, tagged `-alpine`. dfe-fetcher's container Vector extractor runs that image. +4. Release this repo. The full test tier, nightly and before a release, runs the `vector_validate` cases against the new Vector. -## 11. Component Boundary Summary +## The chart -| Concern | Owner | Notes | -|---|---|---| -| App deployment | dfe-infra `apps.yaml` | Multiplicity, scaling, transports, source binding | -| Config schema + validation | Wrapper (Rust) | `docs/config-schema.json` is generated from the config structs. dfe-engine judges a stored config against that mounted contract | -| K8s manifests | Chart generated from `contract()` | On the scalo-service library chart (section 9) | -| Config assembly (big dials → Vector YAML) | Wrapper (Rust) | The core novel work | -| DAG wiring + validation | Wrapper (Rust) | Auto-wire source/sink, validate refs | -| Vector process lifecycle | Wrapper (Rust) | Spawn, monitor, restart, shutdown | -| Data processing | Vector (subprocess) | Kafka consume → transform → Kafka produce | -| Health/metrics aggregation | Wrapper (Rust) | Composite health, proxied + own metrics | -| Scaling signals | Wrapper (Rust) | `scaling_pressure` is 0 while Vector runs and pinned to 0 while it is down. KEDA scales on CPU | -| Vector binary version | Dockerfile ARG + config pin | Decoupled from wrapper releases | +No chart is committed. At release hyperi-ci runs `generate-artefacts` for the deployment contract and assembles a thin chart on the scalo-service library chart, at the version `release.helm.library` names in `.hyperi-ci.yaml`. The library renders the Deployment, Service, ConfigMap, ServiceAccount and KEDA ScaledObject, and its README documents the values. ---- +`dfe-transform-vector emit-contract` prints what the contract fixes: the probes on the metrics port and their startup budget, the Push port while `config.source.transport` is `direct` or `grpc`, the Kafka Secret in `DFE_TRANSFORM_VECTOR_{SOURCE,SINK}_SASL_{USERNAME,PASSWORD}` and `DFE_TRANSFORM_VECTOR_KAFKA_SASL_MECHANISM`, the writable paths `data`, `run` and `dlq` beside the library's `/tmp` under a read-only root, the requests and limits, the grace period, and KEDA on CPU with no Kafka lag trigger. -## 12. What's NOT In Scope +The mounted config is the chart's `config` with `configOverrides` merged over it. The contract's `default_config` is never merged in, so the app's own defaults apply to anything a deployment leaves unset, `vector.version_check: strict` included. The ConfigMap mounts over the config file's whole directory unless `configMount.subPath` is set, which hides the image's `/etc/dfe-transform-vector/transforms`: deliver transform files with a `fileSets` entry and point `transforms.dir` at its `mountPath`. -- **VRL authoring tools** — users write their own transform VRL, we just validate it -- **Transform library/marketplace** — future consideration, not MVP -- **Multi-Vector-process** — one Vector child per wrapper instance (scale via K8s replicas) -- **Kafka topic creation** — handled by dfe-engine's source registry -- **ClickHouse DDL** — not applicable (this is Kafka→Kafka, not Kafka→ClickHouse) -- **Kubernetes operator / CRDs** — we use Helm + Argo CD, not a custom operator +dfe-infra's `apps.yaml` declares the app `per_config`, so each source config gets its own Deployment, consumer group and ScaledObject. The app is in the default composition of `docker-single` only. diff --git a/docs/librdkafka.md b/docs/librdkafka.md index 4d8bebc..825dc2d 100644 --- a/docs/librdkafka.md +++ b/docs/librdkafka.md @@ -1,245 +1,105 @@ -# librdkafka Configuration Reference +# librdkafka settings -DFE services use librdkafka (via Vector's Kafka source/sink or via rdkafka crate -directly). This document covers the shared baseline defaults, per-service -overrides, and how to customise settings through the config cascade. +On the bus transport Vector owns the Kafka clients, and the supervisor writes their librdkafka settings into the generated source and sink as `librdkafka_options`. This page covers where those settings come from and how to change them. -## Design Philosophy +## Layers -**Only set values that differ from librdkafka defaults.** Every non-default -setting must have clear justification. Fewer overrides means fewer surprises, -easier debugging, and automatic benefit from upstream librdkafka improvements. +Three layers merge into `librdkafka_options`, highest last: -## Configuration Cascade +| Layer | Source | Who manages it | +|-------|--------|----------------| +| 1. Baseline profile | `consumer.production` or `producer.production` in the central file when it defines that profile, else scalo's `CONSUMER_PRODUCTION` or `PRODUCER_PRODUCTION` | Platform operator, or scalo | +| 2. Service overrides | Hardcoded in `src/config/generate.rs` | This repo | +| 3. Per-instance options | `source.librdkafka_options`, `sink.librdkafka_options` | Whoever writes the config | -Three layers merge in priority order (highest wins): +Layer 1 is one layer, not two. A profile in the central file REPLACES the matching scalo constant set outright, so a central `producer.production` that names only `linger.ms` ships only `linger.ms`, and every other baseline setting reverts to what Vector writes itself, below, or else librdkafka's default. Layer 2 sits on whichever source won, so a central setting cannot undo a service override: `queue.buffering.max.kbytes` set centrally takes no effect here. -| Layer | Source | Who Manages | -|-------|--------|-------------| -| 3. User config YAML | `librdkafka_options:` in service config | Platform operator | -| 2. Service-specific | Hardcoded in each service's generator | Service developer | -| 1. Baseline profile | The central config file if it defines the profile, else `scalo::kafka_config` constants | Platform team / shared library | +Vector applies `librdkafka_options` over the settings it writes itself from the source and sink fields, so every layer above outranks those: -**Layer 1 is one layer, not two, and the two sources are mutually exclusive.** -A profile named in `librdkafka.yaml` REPLACES the matching scalo constant set -outright -- it does not merge over it. A central profile that names only -`linger.ms` therefore ships only `linger.ms`, and every other baseline setting -reverts to the librdkafka default. Restate the whole profile in the central -file, or leave the profile out and let the constants stand. +| librdkafka key | Written by Vector from | +|---|---| +| `session.timeout.ms`, `auto.commit.interval.ms`, `auto.offset.reset` | `source.session_timeout_ms` (30000), `source.commit_interval_ms` (5000), `source.auto_offset_reset` (`largest`) | +| `enable.auto.commit`, `enable.auto.offset.store`, `statistics.interval.ms` | Vector's own source, as `true`, `false` and `1000` | +| `message.timeout.ms`, `socket.timeout.ms` | `sink.message_timeout_ms` (0, no limit), `sink.socket_timeout_ms` (60000) | +| `queue.buffering.max.ms`, `batch.num.messages`, `batch.size` | `sink.batch.timeout_secs`, `.max_events` and, where set, `.max_bytes` | -**A service override beats the central file.** Layer 2 is applied on top of -whichever layer-1 source won, so a platform-wide central setting cannot undo a -service's hardcoded one -- see Service-Specific Overrides for what each service -pins. Only layer 3, the per-instance `librdkafka_options:`, is above it. That -matters when tuning: `queue.buffering.max.kbytes` is a transform-vector service -override, so setting it centrally will NOT take effect. +`linger.ms` is librdkafka's other name for `queue.buffering.max.ms`. rdkafka applies a client's options from a hash map, so with the production profile's `linger.ms` beside Vector's `queue.buffering.max.ms`, which one a producer ends up with is not fixed. -### The big dials sit above all three +A few keys are derived from the big dials and written after the merge, so `librdkafka_options` cannot reach them: -A handful of librdkafka keys are DERIVED from the big-dial config and written -after the whole merge, so `librdkafka_options` cannot reach them: - -| Key | Owned by | Applies when | +| Key | Owned by | Written when | |---|---|---| | `compression.type` | `sink.compression` | always, on the sink | | `security.protocol` | `{source,sink}.sasl.enabled` + `.tls.enabled` | either is on | | `sasl.mechanism` | `{source,sink}.sasl.mechanism` | SASL is on | | `enable.ssl.certificate.verification` | `{source,sink}.tls.skip_verify` | TLS is on and verification is skipped | -`config-check` refuses a config that sets one of these in `librdkafka_options`, -and names the big dial that owns the key. -The conditions above are part of the rule: with SASL off nothing derives -`sasl.mechanism`, so setting it by hand there is still allowed. +`config-check` refuses a config that sets one of these in `librdkafka_options`, and names the big dial that owns the key. The conditions are part of the rule: with SASL off nothing derives `sasl.mechanism`, so setting it by hand there is allowed. Any other key can be set per instance: -### Example +```yaml +source: + librdkafka_options: + fetch.min.bytes: "2097152" # 2 MiB instead of 1 MiB -A transform-vector pod producing to Kafka, with no central config file: +sink: + librdkafka_options: + queue.buffering.max.kbytes: "524288" # 512 MiB +``` -1. Baseline: scalo `PRODUCER_PRODUCTION` sets `linger.ms=100`, - `compression.type=zstd`, `socket.nagle.disable=true`, - `statistics.interval.ms=1000` -- or, if `librdkafka.yaml` defines - `producer.production`, exactly what that profile lists and nothing else -2. Service override adds `queue.buffering.max.kbytes=262144` (256 MiB cap) -3. User config YAML sets `linger.ms=20` for a specific pipeline -4. The big dial `sink.compression: lz4` is applied last +## Example -Result: `linger.ms=20`, `compression.type=lz4`, -`queue.buffering.max.kbytes=262144`, `socket.nagle.disable=true`, -`statistics.interval.ms=1000`, and `message.max.bytes=16777216` because no layer -set it +For a producer with no central file, `sink.compression: lz4` and `sink.librdkafka_options: {linger.ms: "20"}`, the supervisor writes `linger.ms=20`, `compression.type=lz4`, `queue.buffering.max.kbytes=262144`, `enable.idempotence=true`, `socket.nagle.disable=true`, `statistics.interval.ms=1000` and `message.max.bytes=16777216`, the last because no layer set it. -Add a central `producer.production` naming only `linger.ms=50` and the result -becomes `linger.ms=50`, `compression.type=lz4`, -`queue.buffering.max.kbytes=262144` -- `socket.nagle.disable` and -`statistics.interval.ms` are gone, because the central profile replaced the -constants rather than merging over them. +Add a central `producer.production` naming only `linger.ms: "50"` and `socket.nagle.disable` and `statistics.interval.ms` drop out of `librdkafka_options`, because the central profile replaced the constants rather than merging over them. Vector still writes `statistics.interval.ms=1000` itself. `linger.ms` stays `20`: the per-instance option outranks the central file. Drop that option and it becomes `50`. -## Consumer Profiles +`dfe-transform-vector --config assemble ` writes `90_sink.yaml` and `00_source.yaml`, so you can read the result for any config. -### Production (`CONSUMER_PRODUCTION`) +## The production profiles -High-throughput consumer baseline. Used by dfe-loader, dfe-transform-vector, -and any service consuming from Kafka in production. +This app reads `CONSUMER_PRODUCTION` for the source and `PRODUCER_PRODUCTION` for the sink, and no other profile. The librdkafka defaults below are those of the librdkafka release Vector bundles. -| Setting | Value | librdkafka Default | Justification | +| Consumer setting | Value | librdkafka default | Justification | |---------|-------|--------------------|---------------| | `partition.assignment.strategy` | `cooperative-sticky` | `range,roundrobin` | KIP-429: incremental rebalances avoid stop-the-world pauses. Partitions stay assigned during rebalance -- only moving partitions are revoked. | | `fetch.min.bytes` | `1048576` (1 MiB) | `1` | Batch fetches for throughput. Broker waits until 1 MiB of data is available before responding, reducing fetch round-trips. | -| `fetch.wait.max.ms` | `100` | `500` | Upper bound on fetch latency when `fetch.min.bytes` threshold isn't met. Prevents 500 ms stalls on low-volume topics. | -| `queued.min.messages` | `20000` | `100000` | Pre-fetch queue depth. 10-20K is the efficiency sweet spot -- large enough for batching, small enough to avoid excessive memory and rebalance lag. | -| `enable.auto.commit` | `false` | `true` | Suits a service that commits by hand after processing, which is how dfe-loader withholds offsets on a failed insert. transform-vector overrides this back to `true` -- see Service-Specific Overrides. | -| `statistics.interval.ms` | `1000` | `0` (disabled) | Enable librdkafka internal metrics at 1-second granularity for Prometheus scraping. | - -**Settings intentionally left at default:** - -| Setting | Default | Why Not Changed | -|---------|---------|-----------------| -| `queued.max.messages.kbytes` | `65536` (64 MiB) | Adequate for DFE message sizes | -| `socket.receive.buffer.bytes` | `0` (OS default) | OS auto-tuning is better than static values | -| `auto.offset.reset` | `latest` | Controlled per-service via big-dial config, not librdkafka | -| `session.timeout.ms` | `45000` | Default is sensible for K8s environments | -| `heartbeat.interval.ms` | `3000` | Default (session.timeout / 15) is correct | -| `max.poll.interval.ms` | `300000` | Default 5 min is generous enough | -| `fetch.max.bytes` | `52428800` (50 MiB) | Default is adequate | - -### Development/Test (`CONSUMER_DEVTEST`) - -Fast iteration, low memory footprint. For local dev and CI. - -| Setting | Value | librdkafka Default | Justification | -|---------|-------|--------------------|---------------| -| `partition.assignment.strategy` | `cooperative-sticky` | `range,roundrobin` | Consistent with production | -| `queued.min.messages` | `1000` | `100000` | Lower memory on dev machines | -| `enable.auto.commit` | `false` | `true` | Consistent with production | -| `reconnect.backoff.ms` | `10` | `100` | Fast reconnect for quick iteration | -| `reconnect.backoff.max.ms` | `100` | `10000` | Cap quickly -- don't wait 10s in dev | -| `log.connection.close` | `true` | `false` | Debug-friendly connection logging | -| `statistics.interval.ms` | `1000` | `0` | Metrics available even in dev | - -### Low-Latency (`CONSUMER_LOW_LATENCY`) - -Minimal fetch delay for latency-sensitive pipelines. - -| Setting | Value | librdkafka Default | Justification | -|---------|-------|--------------------|---------------| -| `partition.assignment.strategy` | `cooperative-sticky` | `range,roundrobin` | Consistent across environments | -| `fetch.wait.max.ms` | `10` | `500` | Return from fetch after 10 ms regardless of data available | -| `queued.min.messages` | `1000` | `100000` | Smaller pre-fetch queue -- process sooner | -| `enable.auto.commit` | `false` | `true` | DFE manages commits | -| `reconnect.backoff.ms` | `10` | `100` | Fast reconnect | -| `reconnect.backoff.max.ms` | `100` | `10000` | Cap quickly | -| `statistics.interval.ms` | `1000` | `0` | Enable metrics | - -## Producer Profiles - -### Production (`PRODUCER_PRODUCTION`) - -High-throughput producer. Default for all DFE services writing to Kafka. - -| Setting | Value | librdkafka Default | Justification | -|---------|-------|--------------------|---------------| -| `linger.ms` | `100` | `5` | Accumulate larger batches -- 100 ms latency trade-off for significantly higher throughput. Most DFE pipelines are throughput-optimised. | -| `compression.type` | `zstd` | `none` | Best compression ratio with acceptable CPU cost. Reduces network I/O and broker storage. | -| `socket.nagle.disable` | `true` | `false` | Disable TCP Nagle's algorithm. Kafka already batches at the application level (`linger.ms`), so Nagle just adds latency. | -| `statistics.interval.ms` | `1000` | `0` | Enable librdkafka metrics for Prometheus. | - -**Settings intentionally left at default:** - -| Setting | Default | Why Not Changed | -|---------|---------|-----------------| -| `acks` | `all` (-1) | Default is already the safest -- all ISR replicas must acknowledge. | -| `batch.size` | `1000000` (1 MiB) | Adequate for most workloads | -| `batch.num.messages` | `10000` | Default is reasonable | -| `queue.buffering.max.kbytes` | `1048576` (1 GiB) | Default is fine for most services. transform-vector overrides to 256 MiB due to lower pod memory. | -| `message.max.bytes` | `1000000` | Left out of the profile. dfe-transform-vector writes scalo's 16 MiB pipeline ceiling where no layer sets it -- see Service-Specific Overrides. | -| `retries` | `2147483647` (max) | Default retry-forever with idempotence is correct. | -| `delivery.timeout.ms` | `300000` (5 min) | Generous timeout for transient broker issues | - -### Exactly-Once (`PRODUCER_EXACTLY_ONCE`) - -Idempotent producer with ordering guarantees. - -| Setting | Value | librdkafka Default | Justification | -|---------|-------|--------------------|---------------| -| `enable.idempotence` | `true` | `false` | Exactly-once semantics within a partition. Broker deduplicates by producer ID + sequence number. | -| `acks` | `all` | `all` (-1) | Already default but stated explicitly -- invariant for EOS. | -| `max.in.flight.requests.per.connection` | `5` | `1000000` | Maximum value that preserves ordering with idempotent producer (librdkafka requirement). | -| `linger.ms` | `20` | `5` | Moderate batching -- lower than production to reduce latency for EOS workloads. | -| `compression.type` | `zstd` | `none` | Consistent with production | -| `socket.nagle.disable` | `true` | `false` | Consistent with production | -| `statistics.interval.ms` | `1000` | `0` | Enable metrics | - -### Low-Latency (`PRODUCER_LOW_LATENCY`) - -Minimal delay, leader-ack only. Trades durability for speed. - -| Setting | Value | librdkafka Default | Justification | -|---------|-------|--------------------|---------------| -| `acks` | `1` | `all` (-1) | Leader acknowledgement only -- skip ISR replication wait. Risk: data loss if leader crashes before replication. | -| `linger.ms` | `0` | `5` | Send immediately, no batching delay. | -| `compression.type` | `lz4` | `none` | LZ4 is the fastest compression codec -- adds minimal CPU for significant bandwidth reduction. | -| `socket.nagle.disable` | `true` | `false` | No TCP coalescing | -| `statistics.interval.ms` | `1000` | `0` | Enable metrics | - -### Development/Test (`PRODUCER_DEVTEST`) - -Fast acks, no compression overhead. For local dev and CI. +| `fetch.wait.max.ms` | `100` | `500` | Upper bound on fetch latency when the `fetch.min.bytes` threshold isn't met. Prevents 500 ms stalls on low-volume topics. | +| `queued.min.messages` | `20000` | `100000` | Pre-fetch queue depth, large enough for batching and small enough to bound memory and rebalance lag. | +| `enable.auto.commit` | `false` | `true` | Suits a service that commits by hand after processing. This app overrides it back to `true` -- see Service overrides. | +| `statistics.interval.ms` | `1000` | `0` (disabled) | librdkafka internal metrics at 1-second granularity. | -| Setting | Value | librdkafka Default | Justification | +| Producer setting | Value | librdkafka default | Justification | |---------|-------|--------------------|---------------| -| `acks` | `1` | `all` (-1) | Faster for dev -- don't wait for full ISR | -| `socket.nagle.disable` | `true` | `false` | Consistent across profiles | -| `statistics.interval.ms` | `1000` | `0` | Enable metrics even in dev | +| `linger.ms` | `100` | `5` | Larger batches -- 100 ms of latency for higher throughput. See Layers for how it meets Vector's `queue.buffering.max.ms`. | +| `compression.type` | `zstd` | `none` | Best compression ratio with acceptable CPU cost. `sink.compression` replaces it on every sink. | +| `socket.nagle.disable` | `true` | `true` | Already librdkafka's default. The profile states it. | +| `statistics.interval.ms` | `1000` | `0` | librdkafka metrics. | -## Service-Specific Overrides +Left at librdkafka's default, unless a layer or a Vector field above writes them: -Each DFE service may add overrides on top of the shared baseline. These are -hardcoded in the service and applied at layer 2 (below user config, above -baseline). +| Setting | Default | +|---------|---------| +| `queued.max.messages.kbytes` | `65536` (64 MiB) | +| `socket.receive.buffer.bytes` | `0` (OS default) | +| `heartbeat.interval.ms` | `3000` | +| `max.poll.interval.ms` | `300000` | +| `fetch.max.bytes` | `52428800` (50 MiB) | +| `acks` | `all` (-1), which idempotence requires anyway | +| `retries` | `2147483647` | -### dfe-transform-vector +## Service overrides | Setting | Side | Value | Reason | |---------|------|-------|--------| -| `queue.buffering.max.kbytes` | producer | `262144` (256 MiB) | The default 1 GiB producer queue is too large for a transform-vector pod. | -| `enable.auto.commit` | consumer | `true` | Vector's kafka source stores offsets on delivery and leaves the periodic flush to librdkafka's commit timer, which the baseline's `false` never arms. Vector still commits on rebalance and clean shutdown, so the baseline stalled reported lag between those points rather than losing offsets. Lag is what KEDA scales on here. | -| `message.max.bytes` | producer | `16777216` (16 MiB), scalo's `MESSAGE_MAX_BYTES` | The record ceiling every other stage's producer and every topic scalo creates carry; librdkafka's own `1000000` refuses a larger record. Written only where neither layer 1 nor `sink.librdkafka_options` sets the key, so either of those still decides it. The size cap in front of the sink reads the same value. | +| `queue.buffering.max.kbytes` | producer | `262144` (256 MiB) | librdkafka's default 1 GiB producer queue is too large for a transform-vector pod. | +| `enable.idempotence` | producer | `true` | The sink retries without limit, and a retry on a non-idempotent producer can reorder and duplicate records within a partition. | +| `enable.auto.commit` | consumer | `true` | Vector's kafka source stores offsets on delivery and leaves the periodic flush to librdkafka's commit timer, which the baseline's `false` never arms. Vector still commits on rebalance and clean shutdown, so the baseline left reported lag stuck between those points rather than losing offsets. | +| `message.max.bytes` | producer | `16777216` (16 MiB), scalo's `MESSAGE_MAX_BYTES` | The record ceiling every other stage's producer and every topic scalo creates carry, where librdkafka's own `1000000` refuses a larger record. Written only where neither layer 1 nor `sink.librdkafka_options` sets the key. The size cap in front of the sink reads the same value. | -### dfe-loader +## Central config file -No service-specific overrides currently. Uses production baseline as-is. +The central file is `$DFE_CONFIG_DIR/shared/librdkafka.yaml`, read only when `DFE_CONFIG_DIR` is set. This app reads it if it is there and falls back to scalo's constants if it is not, and no other DFE app reads it. It is layer 1, so the service overrides still win over any key it sets. -### dfe-receiver - -No service-specific overrides currently. Uses production baseline as-is. - -## Central Config File - -A single `librdkafka.yaml` in the deployment's private config repo -is the management point for the platform-wide baseline. All DFE services -read this file at startup and fall back to scalo coded-in constants if -it isn't present. It is layer 1, so service-specific constants (layer 2) -still override any key it sets. - -**Location:** `/shared/librdkafka.yaml` - -**Discovery:** Via `DFE_CONFIG_DIR` environment variable. The service reads -`$DFE_CONFIG_DIR/shared/librdkafka.yaml`. This follows the same pattern -dfe-engine uses for `services/`, `deployment/`, etc. - -**Activation:** - -```bash -cd /path/to/config-repo -cp shared/librdkafka.yaml.example shared/librdkafka.yaml -# Edit as needed, then commit -``` - -**Caching:** Profiles are loaded once on first access via `OnceLock` and -cached for the process lifetime. Changes require a service restart (or -future hot-reload support). - -**Format:** +The profiles load once, on first access, and are cached for the process lifetime, so a change to the file needs a restart: a hot reload keeps the cached profiles. ```yaml consumer: @@ -248,64 +108,20 @@ consumer: fetch.min.bytes: "1048576" # ... add/remove/change any librdkafka setting - devtest: - # ... - - low_latency: - # ... - producer: production: linger.ms: "100" compression.type: zstd # ... - - exactly_once: - # ... - - low_latency: - # ... - - devtest: - # ... ``` -Profile names match the scalo constant names (`production`, `devtest`, -`low_latency`, `exactly_once`). If a profile exists in the file, it -completely replaces the scalo constant for that profile. If a profile -is missing from the file, the scalo constant is used. - -## User Config Overrides - -Any setting can be overridden per-instance via the service's config YAML: - -```yaml -source: - librdkafka_options: - fetch.min.bytes: "2097152" # 2 MiB instead of 1 MiB - custom.vendor.setting: "value" - -sink: - librdkafka_options: - queue.buffering.max.kbytes: "524288" # 512 MiB -``` - -These are layer 3 (highest priority) and override everything below, including -the service-specific constants -- but NOT the derived keys in the table above. -`compression.type` belongs to `sink.compression`, and putting it here is a -config error rather than a silent no-op. - -## Source Code References - -- **Shared baseline constants:** `scalo/src/kafka_config.rs` -- **Central config loader:** `src/config/kafka_defaults.rs` (YAML loading, `OnceLock` cache, fallback) -- **Profile accessors:** `kafka_defaults::consumer_profile()`, `kafka_defaults::producer_profile()` -- **3-layer merge:** `kafka_defaults::merge_layers()` (base + service overrides + user overrides) -- **YAML generator:** `src/config/generate.rs` (`build_source_librdkafka`, `build_sink_librdkafka`) -- **Config schema:** `src/config/loader.rs` (`SourceConfig.librdkafka_options`, `SinkConfig.librdkafka_options`) -- **Central config file:** `/shared/librdkafka.yaml.example` +Only the `production` profile of each side is read. A profile in the file replaces the scalo constant set of that name, and a missing one leaves the constants in force. So does an empty one, or one holding no string, number or boolean value: it is skipped. -## Reference +## Source code references -- [librdkafka CONFIGURATION.md](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md) -- full property reference with defaults -- [KIP-429](https://cwiki.apache.org/confluence/display/KAFKA/KIP-429%3A+Kafka+Consumer+Incremental+Rebalance+Protocol) -- cooperative-sticky rebalance protocol +- Profile constants: scalo-rs `src/kafka_config.rs` +- Central file loader: `src/config/kafka_defaults.rs` (`consumer_profile()`, `producer_profile()`, the `OnceLock` cache, `merge_layers()`) +- Layer merge and derived keys: `src/config/generate.rs` (`build_source_librdkafka`, `build_sink_librdkafka`, `SERVICE_CONSUMER_OVERRIDES`, `SERVICE_PRODUCER_OVERRIDES`) +- Config fields: `src/config/loader.rs` (`SourceConfig.librdkafka_options`, `SinkConfig.librdkafka_options`) +- [librdkafka CONFIGURATION.md](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md) -- every property, with its default +- [KIP-429](https://cwiki.apache.org/confluence/display/KAFKA/KIP-429%3A+Kafka+Consumer+Incremental+Rebalance+Protocol) -- the cooperative-sticky rebalance protocol diff --git a/docs/migration.md b/docs/migration.md index f5f55a2..9adca0d 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -1,22 +1,17 @@ -# Migration Guide: Vector → dfe-transform-vector +# Moving off the Vector chart -This guide covers migrating existing Vector pipelines from the official -`timberio/vector` Helm chart to the `dfe-transform-vector` wrapper. +This guide covers moving an existing Vector pipeline from the official Vector Helm chart, served from `helm.vector.dev`, to dfe-transform-vector. -## Pipeline Values Migration +## Pipeline values -A values file for the official Vector chart carries the whole Vector config. -dfe-transform-vector's chart takes the "big-dial" config pattern instead, one -values file per pipeline. Both releases can run side by side while you move. +A values file for the official Vector chart carries the whole Vector config. dfe-transform-vector's chart takes the big-dial config instead, one values file per pipeline, and both releases can run side by side while you move. -### Values Format - -**Before** (official Vector chart — `vector-syslog.yaml`): +Before, the official Vector chart's `vector-syslog.yaml`: ```yaml replicas: 3 image: - tag: "0.48.0-debian" + tag: "-debian" persistence: size: 20Gi resources: @@ -50,11 +45,9 @@ customConfig: # ... ``` -**After** (dfe-transform-vector — `dfe-transform-vector-syslog.yaml`): +After, dfe-transform-vector's `dfe-transform-vector-syslog.yaml`: ```yaml -replicaCount: 3 - resources: requests: cpu: "2" @@ -67,77 +60,70 @@ config: pipeline: name: syslog-enrichment source: + brokers: ["kafka:9092"] topics: - raw_syslog_land group_id: dfe-transform-vector-syslog + tls: + enabled: true sink: + brokers: ["kafka:9092"] topic: enriched_syslog_land key_field: .org_id compression: zstd + tls: + enabled: true + transforms: + dir: /etc/dfe-transform-vector-transforms + +fileSets: + transforms: + mountPath: /etc/dfe-transform-vector-transforms + files: + - name: enrich.yaml + content: | + transforms: + enrich: + type: remap + source: .env = "prod" + +secrets: + kafka: + existingSecret: kafka-sasl # holds username, password and sasl.mechanism keda: + minReplicaCount: 3 maxReplicaCount: 20 ``` Key differences: -- **No inline VRL** — transforms are separate YAML files the wrapper loads from `config.transforms.dir` (the image's `/etc/dfe-transform-vector/transforms`) or lists in `config.transforms.files` -- **No persistence** — Deployment (not StatefulSet), data dir loss just re-reads from Kafka -- **SASL/TLS** can sit in one values file every pipeline layers over, so each pipeline file carries only its own values -- **Kafka brokers** come from `DFE_TRANSFORM_VECTOR_SOURCE_BROKERS` and - `DFE_TRANSFORM_VECTOR_SINK_BROKERS`. A `${VAR}` in the config is not expanded: - neither the wrapper nor Vector 0.57+ interpolates environment variables -- **SASL credentials** come from the Kafka Secret through - `DFE_TRANSFORM_VECTOR_{SOURCE,SINK}_SASL_{USERNAME,PASSWORD}`, and the wrapper hands - them to Vector's directory secret backend as owner-only files -- **KEDA** scales on CPU, with no Kafka lag trigger and no - TriggerAuthentication +- No inline VRL. Transforms are YAML files the app loads from `config.transforms.dir`, or lists in `config.transforms.files`. Deliver them with a `fileSets` entry or an image of your own: the chart's ConfigMap mounts over `/etc/dfe-transform-vector/` and hides the image's `transforms` directory there. +- No persistence by default. The workload is a Deployment, and `vector.data_dir` is an emptyDir unless `writablePaths.data.persistence.enabled` gives it a claim, so records not yet acknowledged are read again from Kafka after a pod restart. +- SASL and TLS can sit in one values file every pipeline layers over, so each pipeline file carries only its own values. +- A `${VAR}` in the config is not expanded: neither the app nor Vector 0.57+ interpolates environment variables. Brokers and topics are big dials, and in a DFE deployment the `DFE_TRANSFORM_VECTOR_SOURCE_BROKERS` and `DFE_TRANSFORM_VECTOR_SINK_BROKERS` variables set the brokers. +- SASL credentials come from the Kafka Secret through `DFE_TRANSFORM_VECTOR_{SOURCE,SINK}_SASL_{USERNAME,PASSWORD}`, and the app hands them to Vector's directory secret backend as owner-only files. +- `replicaCount` applies only while nothing autoscales the workload. KEDA scales on CPU, between `keda.minReplicaCount` and `keda.maxReplicaCount`, with no Kafka lag trigger and no TriggerAuthentication. -### What Stays the Same - -- **Secrets**: the chart creates none. The Kafka SASL secret comes from wherever your deployment already supplies it -- **Node placement**: set the chart's `nodeSelector` and `tolerations` to match your existing pool - -### What Changes +Two things stay as they were. The chart creates no Secret, so the Kafka SASL secret comes from wherever your deployment already supplies it. Node placement carries over through the chart's `nodeSelector` and `tolerations`. | Element | Before | After | |---------|--------|-------| -| Chart source | `helm.vector.dev` v0.42.1 | The thin chart each `dfe-transform-vector` release publishes, on scalo-service | -| Image | `timberio/vector:0.48.0-debian` | `ghcr.io/hyperi-io/dfe-transform-vector` | -| Workload type | StatefulSet | Deployment | -| Health endpoint | `/health` on :8686 | `/livez`, `/readyz` on :9090, beside `/metrics` | -| Metrics | Vector native on :9090 | Wrapper + merged Vector on :9090 | -| Config format | Full Vector YAML (customConfig) | Big-dial config (source/sink/transforms) | -| Transform delivery | Inline VRL in `customConfig` | YAML files in `transforms.dir` or listed in `transforms.files`. The chart has no init container or download step: bake the files into your image or mount them at that path | -| DAG wiring | Manual (inline in values) | Auto-wired by wrapper | -| Hot-reload | Not supported | Poll-based file watcher + SIGHUP | -| Crash recovery | K8s pod restart only | Wrapper restarts Vector with backoff | - -## Cutover Procedure - -### Step 1: Prepare Pipeline Values - -For each existing Vector values file, write a dfe-transform-vector values file -in the big-dial format above. - -### Step 2: Deploy in Parallel - -Run both releases at once. The new pipeline uses a different consumer group -ID, so it reads from the same source topic independently without affecting -the existing pipeline. - -### Step 3: Validate - -- Compare output topic records between old and new pipelines -- Monitor metrics: throughput, latency, error rates -- Check health endpoints respond correctly -- Verify KEDA scaling behaviour - -### Step 4: Switch Traffic - -Once validated: - -1. Scale down old pipeline replicas to 0 -2. Update downstream consumers to read from new output topic (if topic name changed) -3. Monitor for 24-48 hours -4. Remove the old Vector release and its values file +| Chart source | `helm.vector.dev` | The thin chart each `dfe-transform-vector` release publishes, on scalo-service | +| Image | `timberio/vector` | `ghcr.io/hyperi-io/dfe-transform-vector` | +| Workload type | StatefulSet, the `Aggregator` role | Deployment | +| Health endpoint | `/health` on Vector's API, 8686 by default | `/livez`, `/readyz` on 9090, beside `/metrics` | +| Metrics | Vector's own exporter | The app's and Vector's, merged on 9090 | +| Config format | Full Vector YAML (`customConfig`) | Big-dial config (source/sink/transforms) | +| Transform delivery | Inline VRL in `customConfig` | YAML files in `transforms.dir`, from a `fileSets` entry or an image of your own | +| DAG wiring | Manual (inline in values) | Auto-wired by the app | +| Reload | Vector's `--watch-config`, where the args set it | A transform file changed in place reloads. A chart change rolls the pods | +| Crash recovery | K8s pod restart only | The app restarts Vector with backoff, inside the pod | + +## Cutover procedure + +1. For each existing Vector values file, write a dfe-transform-vector values file in the big-dial format above. +2. Run both releases at once, the new one with a `sink.topic` of its own. The new pipeline uses a different consumer group ID, so it reads the same source topic independently of the existing pipeline, and its own output topic keeps the two from writing every record twice into one topic. A new group starts at the newest offset, because `source.auto_offset_reset` defaults to `largest`, so its output holds only what arrives after it starts. Set `smallest` to have it read the topic's backlog too. +3. Validate: compare the output topics of the old and new pipelines, watch throughput, latency and error rates, check the health endpoints, and check KEDA scales. +4. Scale the old pipeline to 0 replicas and let downstream consumers drain its output topic, then point them at the new output topic. Records both pipelines carried during the overlap are in both topics, so a consumer that starts the new topic at its beginning reads those twice. Start it past the last record it took from the old topic, or deduplicate. +5. Monitor for 24-48 hours, then remove the old Vector release and its values file. diff --git a/pipelines/filebeat/README.md b/pipelines/filebeat/README.md index 9e40727..db221c2 100644 --- a/pipelines/filebeat/README.md +++ b/pipelines/filebeat/README.md @@ -10,10 +10,7 @@ Copyright: (c) 2026 HYPERI PTY LIMITED # Bundled pipeline: filebeat-compat (INTERIM) -A pre-canned transform file that keeps DFE 2.1 filebeat feeds working on -dfe-transform-vector: the same VRL dfe-transform-vrl runs, wrapped in a Vector -`remap` and carrying the enrichment table it needs. Point `transforms.dir` at -this directory and the pipeline runs -- there is nothing to author. +A pre-canned transform file that keeps DFE 2.1 filebeat feeds working on dfe-transform-vector: the same VRL dfe-transform-vrl runs, wrapped in a Vector `remap` and carrying the enrichment table it needs. Point `transforms.dir` at this directory, mount the table, and the pipeline runs -- there is nothing to author. The modules, the routing rules, the departures from DFE 2.1 and the known limitations are documented once, in dfe-transform-vrl `pipelines/filebeat/README.md`, and apply here unchanged. > **INTERIM.** Elastic compatibility is being replaced by > `dfe-transform-elastic` (Rust-native). New integrations @@ -28,31 +25,12 @@ transforms: dir: "/etc/dfe-transform-vector/transforms" ``` -Mount `filebeat.yaml` into that directory and `timezones.csv` at the path the -pipeline's `enrichment_tables` entry names -(`/etc/dfe-transform-vector/data/timezones.csv`), or edit that one path to -wherever the table lives. Both files are here. +Mount `filebeat.yaml` into that directory and `timezones.csv` at the path the pipeline's `enrichment_tables` entry names (`/etc/dfe-transform-vector/data/timezones.csv`), or edit that one path to wherever the table lives inside `transforms.data_dir`. Both files are here. -The wrapper writes a transform file into Vector's config dir verbatim, which is -what lets a transform file declare a top-level `enrichment_tables` key: Vector -merges the declaration and the VRL's timezone lookups resolve. `vector validate` -runs before Vector starts, so a pipeline that cannot compile stops the app -rather than passing nothing on. +The wrapper writes each transform file into Vector's config dir with its wired `inputs` added, and a transform file may declare a top-level `enrichment_tables` key, so Vector merges the declaration and the VRL's timezone lookups resolve. `vector validate` runs before Vector starts, so a pipeline that cannot compile stops the app rather than passing nothing on. -Events are the DFE 2.1 Kafka shape (`{message, tags, timestamp}`), and the -optional `._conf.tz_offset` / `._conf.tz_map` fields drive the ios and meraki -timezone handling. - -## What it covers, and how it routes - -The modules, the routing rules, the departures from DFE 2.1 and the known -divergences from the elastic goldens are documented once, in dfe-transform-vrl -`pipelines/filebeat/README.md`. They apply here unchanged: this is the same VRL. +Events carry the DFE 2.1 Kafka fields (`{message, tags, timestamp}`), and the optional `._conf.tz_offset` / `._conf.tz_map` fields drive the ios and meraki timezone handling. ## Provenance -`filebeat.yaml` embeds dfe-transform-vrl `pipelines/filebeat/filebeat.vrl` -@ `e0b7087`, byte-identical, and `timezones.csv` is that repo's table copied. -Refresh by copying both again rather than editing here -- the two transform -apps grade against the same corpus (`tests/e2e/filebeat_kafka.rs` in each), and -that only means something while the program is the same one. +`filebeat.yaml` embeds dfe-transform-vrl `pipelines/filebeat/filebeat.vrl` @ `e0b7087`, the same program bar two lines of its header comment, and `timezones.csv` is that repo's table copied. Refresh by copying both again rather than editing here -- the two transform apps grade against the same cases (`tests/e2e/filebeat_kafka.rs` in each), and that only means something while the program is the same one. diff --git a/tests/TESTING.md b/tests/TESTING.md index 5352b0b..41119f6 100644 --- a/tests/TESTING.md +++ b/tests/TESTING.md @@ -1,26 +1,19 @@ # Testing -The house rules are hyperi-ai's testing standard, `standards/universal/testing.md`. +Which tier runs what, and how the tests find a broker and a Vector binary. ## Tiers -- **Core** runs on every push and PR: `cargo nextest run --all-features`. Unit, integration, smoke, and the e2e cases that are not `#[ignore]`. -- **Full** runs nightly and before every release (`test.full.required_for_release` in `.hyperi-ci.yaml`): core plus every `#[ignore]` case, `cargo nextest run --all-features --run-ignored all`. Those are the `vector_validate` cases and the opt-in `e2e::kafka` pipelines. +- Core runs on every push and PR: `cargo nextest run --all-features`. Unit, integration, smoke, and the e2e cases that are not `#[ignore]`. +- Full runs nightly and before every release (`test.full.required_for_release` in `.hyperi-ci.yaml`): core plus every `#[ignore]` case, `cargo nextest run --all-features --run-ignored all --ignore-default-filter`. Those are the `vector_validate` cases, the opt-in `e2e::kafka` pipelines and one timing measurement in `src/metrics/scrape.rs`. -## Backends +## Brokers and the Vector binary There is no mode switch. `KafkaFixture::acquire` resolves a broker on its own: -1. **A live cluster**, if `KAFKA_BROKERS` (and any SASL settings) in `.env` - authenticate against it. Faster, and a real multi-broker cluster. -2. **The run's shared broker**, otherwise -- one single-node Apache Kafka that - scalo's `shared_service` starts for the whole run and every test process - joins. The image is scalo's `KAFKA_IMAGE`. - The first test that needs it starts it on the local - Docker daemon, locally and in CI alike, and a watcher removes it when the - run's last test process exits. -3. **Skip**, when neither is available. In CI that is an assertion failure - instead: a test that skips when its environment disappears is not a gate. +1. A live cluster, if `KAFKA_BROKERS` (and any SASL settings) in `.env` authenticate against it. Faster, and a real multi-broker cluster. +2. Otherwise the run's shared broker: one single-node Apache Kafka, scalo's `KAFKA_IMAGE`, that scalo's `shared_service` starts for the whole run and every test process joins. The first test that needs it starts it on the local Docker daemon, locally and in CI alike, and a watcher removes it when the run's last test process exits. +3. A skip, when neither is available. In CI that is an assertion failure instead: a test that skips when its environment disappears is not a gate. `KafkaFixture::hermetic` skips step 1 outright and takes the shared broker. Anything using the real DFE topic names takes that path, because @@ -35,11 +28,7 @@ leftovers from a killed run with: docker rm -f $(docker ps -aq --filter label=io.hyperi.test.suite=dfe-transform-vector-integration) ``` -## The Vector binary - -Tests that spawn Vector resolve it through `scripts/fetch-vector.sh`, which -downloads the pinned build once into `.tmp/`, and fall back to `vector` on -`PATH`. Without either they skip. +Tests that spawn Vector resolve it through `scripts/fetch-vector.sh`, which downloads the pinned build once into `.tmp/`, and fall back to `vector` on `PATH`. Without either they skip locally, and the cases that run by default fail under CI. ## Suites @@ -53,16 +42,14 @@ cargo nextest run --test integration # CLI surface cargo nextest run --test smoke -# End to end -- needs Docker or a live cluster, plus a Vector binary +# The e2e tests -- need Docker or a live cluster, plus a Vector binary cargo nextest run --test e2e # Everything, opt-in cases included cargo nextest run --run-ignored all ``` -Opt-in (`#[ignore]`) cases are the full tier: `vector_validate` runs the real -validator over each assembled config, and two `e2e::kafka` pipelines drive a -bare Vector process against a broker. +Opt-in (`#[ignore]`) cases are the full tier: `vector_validate` runs the real validator over each buffer, SASL and TLS variant of an assembled config, and two `e2e::kafka` pipelines drive a bare Vector process against a broker. ## The WS21 acceptance case