diff --git a/.gitignore b/.gitignore index 7aebe3b..d4699c7 100644 --- a/.gitignore +++ b/.gitignore @@ -29,6 +29,9 @@ coverage.lcov .cargo/* !.cargo/config.toml +# written by `dfe-transform-vector emit-chart chart`; the release assembles the chart from the contract +/chart/ + # Local working files (not committed) docs/superpowers/ .claude/ diff --git a/.hyperi-ci.yaml b/.hyperi-ci.yaml index 042883b..2f28d7e 100644 --- a/.hyperi-ci.yaml +++ b/.hyperi-ci.yaml @@ -37,7 +37,6 @@ test: # Building build: enabled: true - type: app strategies: - native rust: @@ -63,9 +62,15 @@ build: bolt: enabled: true # Linux only; skipped on non-Linux targets -# Release (where the artefacts go) +# Release (where the artefacts go). The thin chart is assembled from the +# emitted contract and ships beside the image. release: enabled: true + # Keep `library` at the scalo version in Cargo.toml: the library renders only the contract version its scalo release writes. + helm: + enabled: true + contract: emit + library: "2.14.3" container: enabled: true dockerfile: Dockerfile @@ -76,5 +81,3 @@ release: # by `src/deployment.rs::emit_dockerfile()` (consumer-side post-process # over scalo's base output). When hyperi-ci's overlay framework # lands, this moves to `release.container.overlays:` here. - helm: - enabled: true diff --git a/Cargo.lock b/Cargo.lock index bc08f9a..4ed11d2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3204,9 +3204,9 @@ dependencies = [ [[package]] name = "scalo" -version = "2.14.1" +version = "2.14.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c91d0d04345fda3e6508b44d8478add090f8699946d13d16b31bd778cb1dc1e1" +checksum = "1105393fd98b1d95b1809650b5761794a7df88a21a72dddce4ca535d946e983f" dependencies = [ "backon", "bytes", diff --git a/Cargo.toml b/Cargo.toml index aca91d3..e98251f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -64,7 +64,7 @@ tokio-util = { version = ">=0.7.19, <0.8", features = ["rt"] } # image ships no librdkafka, so nothing here may build an rdkafka client. # # The floor's contract carries no vendor defaults, so the app names its own. -scalo = { version = "2.14.1", features = [ +scalo = { version = "2.14.3", features = [ "cli-service", "config-reload", "config-schema", @@ -85,7 +85,6 @@ scalo = { version = "2.14.1", features = [ # Configuration figment = { version = ">=0.10.19, <0.11", features = ["env"] } clap = { version = ">=4.6.7, <5", features = ["derive", "env"] } -dotenvy = ">=0.15.7, <0.16" # Record payloads across the direct-transport bridge, which moves bytes rather # than copying them between the two gRPC legs. @@ -141,9 +140,11 @@ tar = ">=0.4.46, <0.5" chrono = ">=0.4.45, <0.5" testcontainers = ">=0.27.3, <0.28" testcontainers-modules = { version = ">=0.15, <0.16", features = ["kafka"] } -scalo = { version = "2.14.1", features = [ +scalo = { version = "2.14.3", features = [ "transport-kafka", ] } +# The repo's own .env for the live-broker tests. The binary's .env is read by scalo's cascade. +dotenvy = ">=0.15.7, <0.16" # A stand-in Vector source for the bridge tests; scalo already builds this tonic. tonic = { version = ">=0.14.6, <0.15", features = ["gzip"] } # Reads back what the app counted; the Prometheus exporter already builds it. diff --git a/Dockerfile b/Dockerfile index da3ec8c..ae173fc 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,7 +7,7 @@ # # AUTOGENERATED -- do not edit by hand. # Generated by scalo::deployment::generate_dockerfile() -# Schema version: 3 +# Schema version: 4 # Source contract: dfe-transform-vector::deployment::contract() # Regenerate by calling scalo::deployment::generate_dockerfile() on that contract. diff --git a/README.md b/README.md index 4468372..24387d1 100644 --- a/README.md +++ b/README.md @@ -78,7 +78,8 @@ ceiling, or one Vector refused for good because a sink it feeds rejected it - **Production Kafka tuning**: librdkafka defaults baked in with 4-layer cascade - **Metrics**: Vector's own `vector_*` merged into the wrapper's registry, so one endpoint and one OTLP push carry both -- **Helm chart**: Generated Deployment-based chart with KEDA, HPA, ConfigMap +- **Helm chart**: A thin chart assembled at release from the deployment contract + on the scalo-service library chart ## Quick Start @@ -93,12 +94,18 @@ cargo build --release ./target/release/dfe-transform-vector emit-dockerfile ./target/release/dfe-transform-vector emit-compose -# emit-chart overwrites whatever directory you point it at. The committed -# chart/ carries KEDA hand-edits the generator does not produce, so render -# somewhere scratch and diff rather than regenerating over it. -./target/release/dfe-transform-vector emit-chart /tmp/dfe-transform-vector-chart +# scalo's full chart, for local use. emit-chart overwrites the directory it is given. +./target/release/dfe-transform-vector emit-chart chart ``` +### Helm chart + +No chart is committed here. At release, hyperi-ci runs the binary's `generate-artefacts` for the deployment contract and assembles a thin chart from it on the scalo-service library chart, at the version `release.helm.library` names in `.hyperi-ci.yaml`. Keep that version at the scalo version in `Cargo.toml`: a library renders only the contract version its scalo release writes. + +To see the chart a release would ship, build the binary and run `hyperi-ci chart assemble --binary target/debug/dfe-transform-vector --image ghcr.io/hyperi-io/dfe-transform-vector:@sha256: --version `. It prints the chart directory it wrote. + +The ScaledObject scales on CPU alone, because `contract()` sets `KafkaLagTrigger::disabled()`. Consumer-group lag rises when a downstream stage breaks, and more replicas cannot fix that. A deployment adds a trigger with the library's `keda.extraTriggers` value. + ## Configuration See [config.example.yaml](https://github.com/hyperi-io/dfe-transform-vector/blob/main/config.example.yaml) @@ -113,6 +120,8 @@ Key environment variable overrides (prefix `DFE_TRANSFORM_`): | `DFE_TRANSFORM_SOURCE__BROKERS` | Kafka source broker addresses | | `DFE_TRANSFORM_SOURCE__TOPICS` | Source topic list | | `DFE_TRANSFORM_SOURCE__GROUP_ID` | Consumer group ID | +| `DFE_TRANSFORM_{SOURCE,SINK}_SASL_{USERNAME,PASSWORD}` | Kafka SASL credentials per endpoint; the chart mounts the Kafka Secret's `username` and `password` into all four | +| `DFE_TRANSFORM_KAFKA_SECURITY_PROTOCOL` | Sets `source.tls.enabled` and `sink.tls.enabled` true when the value contains `SSL`; never sets them false | | `DFE_TRANSFORM_SINK_TRANSPORT` | `bus` or `direct` | | `DFE_TRANSFORM_SINK_ENDPOINT` | Next stage's Push listener (direct) | | `DFE_TRANSFORM_SINK__BROKERS` | Kafka sink broker addresses | @@ -253,7 +262,7 @@ For commercial licensing options, see [COMMERCIAL.md](https://github.com/hyperi- 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, a consumer-lag scaling signal and a chart dfe-engine compiles. It is NOT +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 @@ -271,10 +280,9 @@ deployment asks for it. | `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` | The two-binary image override spliced onto scalo's generated Dockerfile | +| `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 | -| `chart/` | The committed Helm chart. Carries KEDA hand-edits, so NOT pure generator output | | `docs/` | `architecture.md` for the shape and the invariants, `DESIGN.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 | @@ -290,8 +298,6 @@ cargo nextest run --test e2e # needs Docker or a live broker cargo nextest run --run-ignored all # everything, opt-in cases included ``` -`cargo nextest run --lib` was run against this branch: 112 passed, 1 skipped. - What green does NOT mean: - **The default run skips the Vector validator.** Seven `#[ignore]` cases need a @@ -312,8 +318,10 @@ What green does NOT mean: | Don't | Do | Why | |---|---|---| -| Point `emit-chart` at `chart/` | Render to a scratch directory and diff | `chart/` carries KEDA hand-edits the generator does not produce, including a ScaledObject addressed at `config.source.*` where the generator emits `config.kafka.*`. A test asserts the committed chart matches the generator except for four exempted files | -| Bump the Vector version in one place | Change `VECTOR_VERSION` in `src/deployment.rs` and carry it to `chart/values.yaml` | The chart once said 0.48.0 while the image baked 0.57.0, nine minor versions apart, and only `version_check: warn` kept pods starting. Under `strict` that pairing refuses to start at all | +| 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 | +| Bump scalo and leave `release.helm.library` behind | Move `release.helm.library` in `.hyperi-ci.yaml` to the same scalo version | A scalo-service release renders only the contract version its scalo release writes | +| Turn the Kafka lag trigger back on in `contract()` | Leave `KafkaLagTrigger::disabled()` and scale on CPU plus scaling pressure | Lag rises when a downstream stage breaks, so a lag trigger adds pods that wait on the same broken stage. The 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 committed chart once said 0.48.0 while the image baked 0.57.0, and only `version_check: warn` kept pods starting. Under `strict` that pairing refuses to start at all | | Mount a whole `templates/*.yaml` as a transform file | Copy only its `transforms:` block | The supervisor generates `sources:` and `sinks:` from the big dials, so Vector would run two sources and two sinks | | 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_{SOURCE,SINK}_SASL_*`; 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 the placeholder reaches the broker as the password | @@ -338,7 +346,9 @@ Inbound -- what this repo depends on: 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. + `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: diff --git a/chart/Chart.yaml b/chart/Chart.yaml deleted file mode 100644 index dfe10a9..0000000 --- a/chart/Chart.yaml +++ /dev/null @@ -1,9 +0,0 @@ -apiVersion: v2 -name: dfe-transform-vector -description: Vector.dev subprocess wrapper for Kafka-to-Kafka transform pipelines -type: application -version: 0.1.0 -appVersion: "1.0.0" - -keywords: - - dfe-transform-vector diff --git a/chart/templates/NOTES.txt b/chart/templates/NOTES.txt deleted file mode 100644 index 3c6db02..0000000 --- a/chart/templates/NOTES.txt +++ /dev/null @@ -1,16 +0,0 @@ -dfe-transform-vector has been deployed. - -1. Get the metrics/health endpoint: - kubectl port-forward svc/{{ include "dfe-transform-vector.fullname" . }} {{ .Values.service.port }}:{{ .Values.service.port }} - curl http://localhost:{{ .Values.service.port }}/livez - curl http://localhost:{{ .Values.service.port }}/metrics - -{{- if .Values.keda.enabled }} - -2. Check KEDA autoscaling status: - kubectl get scaledobject {{ include "dfe-transform-vector.fullname" . }} - kubectl get hpa -{{- end }} - -3. View logs: - kubectl logs -l app.kubernetes.io/name={{ include "dfe-transform-vector.name" . }} -f diff --git a/chart/templates/_helpers.tpl b/chart/templates/_helpers.tpl deleted file mode 100644 index 9ed359c..0000000 --- a/chart/templates/_helpers.tpl +++ /dev/null @@ -1,72 +0,0 @@ -{{/* -Expand the name of the chart. -*/}} -{{- define "dfe-transform-vector.name" -}} -{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} -{{- end }} - -{{/* -Create a default fully qualified app name. -Truncated at 63 chars because some K8s name fields are limited. -*/}} -{{- define "dfe-transform-vector.fullname" -}} -{{- if .Values.fullnameOverride }} -{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }} -{{- else }} -{{- $name := default .Chart.Name .Values.nameOverride }} -{{- if contains $name .Release.Name }} -{{- .Release.Name | trunc 63 | trimSuffix "-" }} -{{- else }} -{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }} -{{- end }} -{{- end }} -{{- end }} - -{{/* -Create chart name and version as used by the chart label. -*/}} -{{- define "dfe-transform-vector.chart" -}} -{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} -{{- end }} - -{{/* -Common labels. -*/}} -{{- define "dfe-transform-vector.labels" -}} -helm.sh/chart: {{ include "dfe-transform-vector.chart" . }} -{{ include "dfe-transform-vector.selectorLabels" . }} -{{- if .Chart.AppVersion }} -app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} -{{- end }} -app.kubernetes.io/managed-by: {{ .Release.Service }} -{{- end }} - -{{/* -Selector labels. -*/}} -{{- define "dfe-transform-vector.selectorLabels" -}} -app.kubernetes.io/name: {{ include "dfe-transform-vector.name" . }} -app.kubernetes.io/instance: {{ .Release.Name }} -{{- end }} - -{{/* -Service account name. -*/}} -{{- define "dfe-transform-vector.serviceAccountName" -}} -{{- if .Values.serviceAccount.create }} -{{- default (include "dfe-transform-vector.fullname" .) .Values.serviceAccount.name }} -{{- else }} -{{- default "default" .Values.serviceAccount.name }} -{{- end }} -{{- end }} - -{{/* -kafka secret name -- use existing or generate from fullname. -*/}} -{{- define "dfe-transform-vector.kafkaSecretName" -}} -{{- if .Values.kafka.existingSecret }} -{{- .Values.kafka.existingSecret }} -{{- else }} -{{- printf "%s-kafka" (include "dfe-transform-vector.fullname" .) }} -{{- end }} -{{- end }} diff --git a/chart/templates/configmap.yaml b/chart/templates/configmap.yaml deleted file mode 100644 index cf5c1ee..0000000 --- a/chart/templates/configmap.yaml +++ /dev/null @@ -1,9 +0,0 @@ -apiVersion: v1 -kind: ConfigMap -metadata: - name: {{ include "dfe-transform-vector.fullname" . }}-config - labels: - {{- include "dfe-transform-vector.labels" . | nindent 4 }} -data: - config.yaml: | - {{- toYaml .Values.config | nindent 4 }} diff --git a/chart/templates/deployment.yaml b/chart/templates/deployment.yaml deleted file mode 100644 index a9cec77..0000000 --- a/chart/templates/deployment.yaml +++ /dev/null @@ -1,164 +0,0 @@ -apiVersion: apps/v1 -kind: Deployment -metadata: - name: {{ include "dfe-transform-vector.fullname" . }} - labels: - {{- include "dfe-transform-vector.labels" . | nindent 4 }} -spec: - {{- if not (or .Values.keda.enabled .Values.autoscaling.enabled) }} - replicas: {{ .Values.replicaCount }} - {{- end }} - selector: - matchLabels: - {{- include "dfe-transform-vector.selectorLabels" . | nindent 6 }} - template: - metadata: - annotations: - checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }} - checksum/secret: {{ include (print $.Template.BasePath "/secret.yaml") . | sha256sum }} - {{- with .Values.podAnnotations }} - {{- toYaml . | nindent 8 }} - {{- end }} - labels: - {{- include "dfe-transform-vector.labels" . | nindent 8 }} - {{- with .Values.podLabels }} - {{- toYaml . | nindent 8 }} - {{- end }} - spec: - {{- with .Values.imagePullSecrets }} - imagePullSecrets: - {{- toYaml . | nindent 8 }} - {{- end }} - serviceAccountName: {{ include "dfe-transform-vector.serviceAccountName" . }} - automountServiceAccountToken: false - {{- with .Values.podSecurityContext }} - securityContext: - {{- toYaml . | nindent 8 }} - {{- end }} - containers: - - name: {{ .Chart.Name }} - image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" - imagePullPolicy: {{ .Values.image.pullPolicy }} - {{- with .Values.securityContext }} - securityContext: - {{- toYaml . | nindent 12 }} - {{- end }} - args: - - "--config" - - "/etc/dfe-transform-vector/config.yaml" - ports: - - name: metrics - containerPort: {{ .Values.service.port }} - protocol: TCP - {{- if eq (default "bus" ((.Values.config).source).transport) "direct" }} - - name: push - containerPort: {{ .Values.service.pushPort | default 6000 }} - protocol: TCP - {{- end }} - env: - - name: OTEL_SERVICE_NAME - value: "dfe-transform-vector" - - name: POD_NAME - valueFrom: - fieldRef: - fieldPath: metadata.name - - name: POD_NAMESPACE - valueFrom: - fieldRef: - fieldPath: metadata.namespace - - name: POD_UID - valueFrom: - fieldRef: - fieldPath: metadata.uid - - name: NODE_NAME - valueFrom: - fieldRef: - fieldPath: spec.nodeName - - name: OTEL_RESOURCE_ATTRIBUTES - value: k8s.pod.name=$(POD_NAME),k8s.namespace.name=$(POD_NAMESPACE),k8s.pod.uid=$(POD_UID),k8s.node.name=$(NODE_NAME) - {{- if (.Values.otel).endpoint }} - - name: OTEL_EXPORTER_OTLP_ENDPOINT - value: {{ .Values.otel.endpoint | quote }} - - name: OTEL_EXPORTER_OTLP_PROTOCOL - value: {{ .Values.otel.protocol | quote }} - {{- end }} - livenessProbe: - httpGet: - path: /livez - port: metrics - initialDelaySeconds: 10 - periodSeconds: 10 - failureThreshold: 3 - readinessProbe: - httpGet: - path: /readyz - port: metrics - initialDelaySeconds: 5 - periodSeconds: 5 - failureThreshold: 2 - startupProbe: - httpGet: - path: /livez - port: metrics - failureThreshold: 30 - periodSeconds: 5 - volumeMounts: - # The file alone: a directory mount hides the image's transforms/, which config.transforms.dir names. - - name: config - mountPath: /etc/dfe-transform-vector/config.yaml - subPath: config.yaml - readOnly: true - # Kafka credentials as files. config.{source,sink}.sasl.secret_dir - # names this path, and Vector's directory secret backend reads the - # two files itself, so no credential reaches the environment, the - # ConfigMap or the assembled Vector config. - - name: kafka-credentials - mountPath: /var/run/secrets/dfe-kafka - readOnly: true - - name: serviceaccount-files - mountPath: /var/run/secrets/kubernetes.io/serviceaccount - readOnly: true - {{- with .Values.resources }} - resources: - {{- toYaml . | nindent 12 }} - {{- end }} - volumes: - - name: config - configMap: - name: {{ include "dfe-transform-vector.fullname" . }}-config - - name: kafka-credentials - secret: - secretName: {{ include "dfe-transform-vector.kafkaSecretName" . }} - # Group-readable: the kubelet gives the files fsGroup, and the - # container runs as that group. - defaultMode: 0440 - items: - - key: {{ .Values.kafka.secretKeys.username }} - path: username - - key: {{ .Values.kafka.secretKeys.password }} - path: password - - name: serviceaccount-files - projected: - sources: - - configMap: - name: kube-root-ca.crt - items: - - key: ca.crt - path: ca.crt - - downwardAPI: - items: - - path: namespace - fieldRef: - fieldPath: metadata.namespace - {{- with .Values.nodeSelector }} - nodeSelector: - {{- toYaml . | nindent 8 }} - {{- end }} - {{- with .Values.affinity }} - affinity: - {{- toYaml . | nindent 8 }} - {{- end }} - {{- with .Values.tolerations }} - tolerations: - {{- toYaml . | nindent 8 }} - {{- end }} diff --git a/chart/templates/hpa.yaml b/chart/templates/hpa.yaml deleted file mode 100644 index 651798c..0000000 --- a/chart/templates/hpa.yaml +++ /dev/null @@ -1,24 +0,0 @@ -{{- if and .Values.autoscaling.enabled (not .Values.keda.enabled) }} -# Standard HPA fallback -- use when KEDA operator is not installed. -# Mutually exclusive with keda.enabled (KEDA creates its own HPA). -apiVersion: autoscaling/v2 -kind: HorizontalPodAutoscaler -metadata: - name: {{ include "dfe-transform-vector.fullname" . }} - labels: - {{- include "dfe-transform-vector.labels" . | nindent 4 }} -spec: - scaleTargetRef: - apiVersion: apps/v1 - kind: Deployment - name: {{ include "dfe-transform-vector.fullname" . }} - minReplicas: {{ .Values.autoscaling.minReplicas }} - maxReplicas: {{ .Values.autoscaling.maxReplicas }} - metrics: - - type: Resource - resource: - name: cpu - target: - type: Utilization - averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }} -{{- end }} diff --git a/chart/templates/keda-scaledobject.yaml b/chart/templates/keda-scaledobject.yaml deleted file mode 100644 index 6468435..0000000 --- a/chart/templates/keda-scaledobject.yaml +++ /dev/null @@ -1,74 +0,0 @@ -{{- /* - HAND-EDITED -- do NOT overwrite this file with `emit-chart` output. - - The generator addresses the Kafka trigger through `.Values.config.kafka.*`. - This app names its consumer side `config.source`, has no `config.kafka` block, - and takes a broker LIST rather than a string. Regenerating renders empty - bootstrapServers, consumerGroup and topic, which KEDA accepts and then never - scales on. Guarded by the `checked_in_keda_scaledobject_survives_emit_chart` - integration test. - - The addressing and the sasl/tls lines both diverge from the generator, which - hardcodes them. Every other line tracks it and must be carried across by hand - when scalo changes it, because the exemption stops `emit-chart` from - delivering it here. -*/}} -{{- if .Values.keda.enabled }} -apiVersion: keda.sh/v1alpha1 -kind: ScaledObject -metadata: - name: {{ include "dfe-transform-vector.fullname" . }} - labels: - {{- include "dfe-transform-vector.labels" . | nindent 4 }} -spec: - scaleTargetRef: - name: {{ include "dfe-transform-vector.fullname" . }} - minReplicaCount: {{ .Values.keda.minReplicaCount }} - maxReplicaCount: {{ .Values.keda.maxReplicaCount }} - pollingInterval: {{ .Values.keda.pollingInterval }} - cooldownPeriod: {{ .Values.keda.cooldownPeriod }} - triggers: - # Kafka consumer group lag (primary scaler) - - type: kafka - authenticationRef: - name: {{ include "dfe-transform-vector.fullname" . }}-kafka-auth - metadata: - bootstrapServers: {{ join "," .Values.config.source.brokers | quote }} - consumerGroup: {{ .Values.keda.kafka.consumerGroup | default .Values.config.source.group_id | quote }} - {{- /* `default (index X 0)` would eagerly evaluate `index nil 0` and fail - lint when no topics are set. Use explicit conditional instead. */}} - {{- if .Values.keda.kafka.topic }} - topic: {{ .Values.keda.kafka.topic | quote }} - {{- else if .Values.config.source.topics }} - topic: {{ (index .Values.config.source.topics 0) | quote }} - {{- else }} - topic: "" - {{- end }} - lagThreshold: {{ .Values.keda.kafka.lagThreshold | quote }} - activationLagThreshold: {{ .Values.keda.kafka.activationLagThreshold | quote }} - {{- /* KEDA reads `sasl`, not `saslType`, and its enum spells the - mechanisms without the underscore the app config uses. These track - the app's own source config so the scaler dials the broker the way - the app does. - - `sasl` and `tls` are optional to the app (SourceConfig is - serde(default)), so reach them through the nil-safe parenthesised - form or an operator who omits either block gets a Helm render - error instead of a chart. Compare against "true" rather than - leaning on Go truthiness, which reads a quoted "false" as true. */}} - {{- $mechanisms := dict "plain" "plaintext" "scram_sha_256" "scram_sha256" "scram_sha_512" "scram_sha512" }} - {{- $configured := (.Values.config.source.sasl).mechanism | default "scram_sha_512" }} - sasl: {{ if eq ((.Values.config.source.sasl).enabled | toString) "true" }}{{ required (printf "keda: no KEDA sasl mechanism for %q" $configured) (get $mechanisms $configured) }}{{ else }}none{{ end }} - tls: {{ if eq ((.Values.config.source.tls).enabled | toString) "true" }}enable{{ else }}disable{{ end }} - {{- /* The app turns skip_verify into enable.ssl.certificate.verification=false. - Without the matching dial KEDA is stricter than the app it scales, and a - private-CA broker the app talks to fine fails verification here. */}} - unsafeSsl: {{ (.Values.config.source.tls).skip_verify | default false | quote }} - {{- if .Values.keda.cpu.enabled }} - # CPU utilisation (secondary scaler) - - type: cpu - metricType: Utilization - metadata: - value: {{ .Values.keda.cpu.threshold | quote }} - {{- end }} -{{- end }} diff --git a/chart/templates/keda-triggerauth.yaml b/chart/templates/keda-triggerauth.yaml deleted file mode 100644 index 6cd64cc..0000000 --- a/chart/templates/keda-triggerauth.yaml +++ /dev/null @@ -1,21 +0,0 @@ -{{- if .Values.keda.enabled }} -apiVersion: keda.sh/v1alpha1 -kind: TriggerAuthentication -metadata: - name: {{ include "dfe-transform-vector.fullname" . }}-kafka-auth - labels: - {{- include "dfe-transform-vector.labels" . | nindent 4 }} -spec: - secretTargetRef: - # KEDA's `sasl` parameter is the MECHANISM, not the username -- it takes - # plaintext | scram_sha256 | scram_sha512 | none, and the username is its - # own parameter. Supplying no `username` at all means SASL authentication - # cannot succeed, so the scaler never reads consumer-group lag and the app - # never scales. The mechanism itself rides the trigger metadata. - - parameter: username - name: {{ include "dfe-transform-vector.kafkaSecretName" . }} - key: {{ .Values.kafka.secretKeys.username }} - - parameter: password - name: {{ include "dfe-transform-vector.kafkaSecretName" . }} - key: {{ .Values.kafka.secretKeys.password }} -{{- end }} diff --git a/chart/templates/secret.yaml b/chart/templates/secret.yaml deleted file mode 100644 index 2e1e9ac..0000000 --- a/chart/templates/secret.yaml +++ /dev/null @@ -1,12 +0,0 @@ -{{- if not .Values.kafka.existingSecret }} -apiVersion: v1 -kind: Secret -metadata: - name: {{ include "dfe-transform-vector.kafkaSecretName" . }} - labels: - {{- include "dfe-transform-vector.labels" . | nindent 4 }} -type: Opaque -data: - {{ .Values.kafka.secretKeys.username }}: {{ .Values.kafka.username | b64enc | quote }} - {{ .Values.kafka.secretKeys.password }}: {{ .Values.kafka.password | b64enc | quote }} -{{- end }} diff --git a/chart/templates/service.yaml b/chart/templates/service.yaml deleted file mode 100644 index 40f2619..0000000 --- a/chart/templates/service.yaml +++ /dev/null @@ -1,21 +0,0 @@ -apiVersion: v1 -kind: Service -metadata: - name: {{ include "dfe-transform-vector.fullname" . }} - labels: - {{- include "dfe-transform-vector.labels" . | nindent 4 }} -spec: - type: {{ .Values.service.type }} - ports: - - port: {{ .Values.service.port }} - targetPort: metrics - protocol: TCP - name: metrics - {{- if eq (default "bus" ((.Values.config).source).transport) "direct" }} - - port: {{ .Values.service.pushPort | default 6000 }} - targetPort: push - protocol: TCP - name: push - {{- end }} - selector: - {{- include "dfe-transform-vector.selectorLabels" . | nindent 4 }} diff --git a/chart/templates/serviceaccount.yaml b/chart/templates/serviceaccount.yaml deleted file mode 100644 index d423796..0000000 --- a/chart/templates/serviceaccount.yaml +++ /dev/null @@ -1,13 +0,0 @@ -{{- if .Values.serviceAccount.create -}} -apiVersion: v1 -kind: ServiceAccount -metadata: - name: {{ include "dfe-transform-vector.serviceAccountName" . }} - labels: - {{- include "dfe-transform-vector.labels" . | nindent 4 }} - {{- with .Values.serviceAccount.annotations }} - annotations: - {{- toYaml . | nindent 4 }} - {{- end }} -automountServiceAccountToken: false -{{- end }} diff --git a/chart/values.yaml b/chart/values.yaml deleted file mode 100644 index f86af21..0000000 --- a/chart/values.yaml +++ /dev/null @@ -1,199 +0,0 @@ -# dfe-transform-vector Helm chart values -# -# Generated by scalo deployment module. -# Contract points validated by cargo test. - -# -- Number of replicas, ignored while a KEDA ScaledObject or the HPA -# fallback renders, because that then owns the replica count. -replicaCount: 1 - -image: - repository: ghcr.io/hyperi-io/dfe-transform-vector - # -- Defaults to Chart appVersion - tag: "" - pullPolicy: IfNotPresent - -imagePullSecrets: [] -nameOverride: "" -fullnameOverride: "" - -serviceAccount: - create: true - annotations: {} - # -- If not set, name is generated from fullname - name: "" - -# -- Pod annotations (Prometheus scrape config included by default) -podAnnotations: - prometheus.io/scrape: "true" - prometheus.io/port: "9090" - prometheus.io/path: "/metrics" - -podLabels: {} - -# -- OTLP export target. Only consulted by an app built with scalo's -# otel features; ignored otherwise. Empty leaves the app's own default -# in place rather than switching anything off. Example: -# http://opentelemetry-collector.observability:4317 -# Spans can carry request attributes, so keep this in-cluster. To leave -# the cluster use https:// and set the OTel SDK's certificate env vars. -otel: - endpoint: "" - protocol: grpc - -# -- Pod-level security context. Matches the uid the generated image -# switches to; change both together or the container will not start. -podSecurityContext: - runAsNonRoot: true - runAsUser: 1000 - runAsGroup: 1000 - fsGroup: 1000 - seccompProfile: - type: RuntimeDefault - -# -- Container-level security context. readOnlyRootFilesystem stays -# false because the spool and DLQ write to disk and every volume -# mounted here is read-only; set it true only for an app that spools -# nowhere. -securityContext: - allowPrivilegeEscalation: false - privileged: false - readOnlyRootFilesystem: false - capabilities: - drop: - - ALL - -resources: - requests: - cpu: 250m - memory: 256Mi - limits: - cpu: "2" - memory: 1Gi - -# -- Metrics service; /livez and /readyz are on this port too -service: - type: ClusterIP - port: 9090 - # -- The scalo Push listener, used on the direct transport only. 6000 is the - # platform convention every DFE stage's listener follows. - pushPort: 6000 - -# -- Application configuration (mounted as /etc/dfe-transform-vector/config.yaml) -config: - pipeline: - name: default - source: - # -- bus consumes topics; direct accepts pushes on listen - transport: bus - listen: 0.0.0.0:6000 - brokers: - - kafka:9092 - topics: - - raw_events - group_id: dfe-transform-vector-default - decoding: - codec: json - sasl: - enabled: true - mechanism: scram_sha_512 - # -- Where the Deployment mounts the kafka secret; Vector reads the - # username and password files from it itself. - secret_dir: /var/run/secrets/dfe-kafka - tls: - enabled: false - # -- Commit an offset, or answer a push, only once the record is delivered. - acknowledgements: - enabled: true - sink: - # -- bus produces to topic; direct pushes to endpoint. topic still applies - # on direct: it is the routing key the next stage picks its table by. - transport: bus - endpoint: http://dfe-loader:6000 - brokers: - - kafka:9092 - topic: enriched_events - key_field: .org_id - encoding: json - compression: zstd - sasl: - enabled: true - mechanism: scram_sha_512 - secret_dir: /var/run/secrets/dfe-kafka - tls: - enabled: false - # -- The two supervisor-to-Vector legs, used on the direct transport only. - # Both live inside the pod, so both stay on loopback. - bridge: - to_vector: 127.0.0.1:6100 - from_vector: 127.0.0.1:6101 - batch_size: 500 - transforms: - dir: /etc/dfe-transform-vector/transforms - vector: - binary: /usr/local/bin/vector - data_dir: /var/lib/vector - # -- Vector's API is unauthenticated and nothing here uses it. When turned - # on it binds loopback only: reach it with kubectl port-forward. - api_enabled: false - api_address: 127.0.0.1:8686 - log_level: info - version: 0.58.0 - version_check: warn - metrics: - address: 0.0.0.0:9090 - # -- Where Vector's prometheus_exporter binds. The wrapper scrapes it and - # merges vector_* into the 9090 registry, so keep it on loopback: it is a - # debug surface, not a scrape target, and needs no container port. - vector_metrics_address: 127.0.0.1:9598 - # -- Scrape ticks a merged vector_* gauge may go unseen before it is zeroed, - # so a component Vector drops stops reporting its last value. - vector_metrics_expiry_ticks: 4 - # -- Seconds the sink may hold records without delivering any before - # /readyz reports not ready. 0 never does. - sink_stall_secs: 60 - logging: - level: info - format: json - -# -- kafka credentials -kafka: - existingSecret: "" - secretKeys: - username: kafka-username - password: kafka-password - username: "" - password: "" - -# -- KEDA autoscaling (requires KEDA operator installed) -keda: - enabled: true - minReplicaCount: 1 - maxReplicaCount: 10 - pollingInterval: 15 - cooldownPeriod: 300 - kafka: - # -- Scale when consumer group lag exceeds this per partition - lagThreshold: "1000" - # -- Wake from zero replicas when lag exceeds this - activationLagThreshold: "0" - # -- Override topic (default: first topic from config) - topic: "" - # -- Override consumer group (default: from config) - consumerGroup: "" - cpu: - enabled: true - # -- CPU utilisation percentage threshold - threshold: "80" - -# -- Standard HPA fallback (when KEDA is not installed) -# Mutually exclusive with keda.enabled -autoscaling: - enabled: false - minReplicas: 1 - maxReplicas: 10 - targetCPUUtilizationPercentage: 80 - -nodeSelector: {} -tolerations: [] -affinity: {} diff --git a/config.example.yaml b/config.example.yaml index 318c4a1..68d3d95 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -7,7 +7,7 @@ # Configuration cascade (highest to lowest priority): # 1. CLI args (--config, --log-level, etc.) # 2. Environment variables (DFE_TRANSFORM_*) -# 3. .env file +# 3. .env file in the working directory (never a parent's) # 4. Config file (this file) # 5. Hard-coded defaults @@ -65,6 +65,8 @@ dfe_source: "my_source" # secret_dir: "/var/run/secrets/dfe-kafka" # tls: # enabled: true +# # DFE_TRANSFORM_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_options: # # fetch.max.bytes: "10485760" @@ -182,8 +184,9 @@ logging: # runtime reads its own `enabled` and `memory_gate_threshold` keys from this # block for the pressure gauge the subprocess circuit gates. # -# What actually scales this app is the chart's KEDA triggers -- consumer-group -# lag and CPU -- gated by the Vector subprocess circuit, which pins pressure to -# 0 while the subprocess is down (more pods cannot help a dead subprocess). +# 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). # scaling: # pressure_threshold: 0.8 diff --git a/docs/DESIGN.md b/docs/DESIGN.md index f161d10..93d237b 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -286,10 +286,11 @@ and 9598 stays a loopback debug surface with no reader outside the pod. ### 4.4 Scaling -KEDA scales based on Kafka consumer lag (ScaledObject targets the consumer group -directly). The wrapper exposes `dfe_pipeline_ready` as a basic readiness signal. -Composite scaling pressure (weighted Kafka lag + memory + error rate) is planned -for a future release via the scalo `scaling` feature. +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. --- @@ -744,111 +745,20 @@ DFE remediation guide. ## 9. Helm Chart -### 9.1 Chart Structure +### 9.1 How the Chart Is Built -```text -chart/ - Chart.yaml - values.yaml - templates/ - _helpers.tpl - deployment.yaml - service.yaml - configmap.yaml # Big-dial config - secret.yaml # Kafka credentials (when existingSecret empty) - serviceaccount.yaml - hpa.yaml # HPA fallback (when KEDA not available) - keda-scaledobject.yaml # KEDA ScaledObject (Kafka lag + CPU) - keda-triggerauth.yaml # KEDA TriggerAuthentication - NOTES.txt -``` - -No dependency on the official Vector Helm chart. This is a purpose-built chart for dfe-transform-vector that follows the same patterns as dfe-loader and dfe-receiver charts. +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. `emit-chart ` still writes scalo's full chart for local use. -### 9.2 Values Shape +### 9.2 What the Contract Fixes -`chart/values.yaml`, abridged to the keys an operator normally sets: - -```yaml -replicaCount: 1 # ignored while KEDA or the HPA fallback owns the count - -image: - repository: ghcr.io/hyperi-io/dfe-transform-vector - tag: "" # defaults to the chart appVersion - pullPolicy: IfNotPresent - -config: # mounted as /etc/dfe-transform-vector/config.yaml - pipeline: - name: default - source: - transport: bus # bus consumes topics, direct accepts pushes on listen - brokers: ["kafka:9092"] - topics: ["raw_events"] - group_id: dfe-transform-vector-default - sasl: - enabled: true - mechanism: scram_sha_512 - secret_dir: /var/run/secrets/dfe-kafka - tls: - enabled: false - sink: - transport: bus # bus produces to topic, direct pushes to endpoint - brokers: ["kafka:9092"] - topic: enriched_events - key_field: .org_id - encoding: json - compression: zstd - sasl: - enabled: true - mechanism: scram_sha_512 - secret_dir: /var/run/secrets/dfe-kafka - tls: - enabled: false - transforms: - dir: /etc/dfe-transform-vector/transforms - vector: - version: 0.58.0 - version_check: warn - metrics: - address: 0.0.0.0:9090 - -kafka: # credentials, mounted as files at sasl.secret_dir - existingSecret: "" - secretKeys: - username: kafka-username - password: kafka-password - -resources: - requests: - cpu: 250m - memory: 256Mi - limits: - cpu: "2" - memory: 1Gi - -keda: - enabled: true - minReplicaCount: 1 - maxReplicaCount: 10 - kafka: - lagThreshold: "1000" - cpu: - enabled: true - threshold: "80" - -autoscaling: # HPA fallback, exclusive with keda.enabled - enabled: false - -serviceAccount: - create: true - annotations: {} - -nodeSelector: {} -tolerations: [] -affinity: {} -``` +- 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`. +- The Kafka Secret's `username` and `password` in `DFE_TRANSFORM_{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. -The chart has no `transforms` value. Transform YAML comes from the image, at `config.transforms.dir`, or from the paths in `config.transforms.files`. +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`. ### 9.3 Parity With dfe-loader/dfe-receiver Charts @@ -858,8 +768,8 @@ The chart has no `transforms` value. Transform YAML comes from the image, at `co | Health probes | /livez, /readyz | /livez, /readyz | /livez, /readyz | | Metrics port | 9090 | 9090 | 9090 | | ConfigMap | config.yaml | config.yaml | config.yaml | -| PVC | data dir | data dir | None (Deployment) | -| KEDA ScaledObject | Kafka lag | Kafka lag | Kafka lag | +| PVC | data dir | data dir | None -- `emptyDir` writable paths | +| KEDA ScaledObject | Kafka lag | Kafka lag | CPU | | PodMonitor | Yes | Yes | Yes | | ServiceAccount + IRSA | Yes | Yes | Yes | | Reloader annotation | Yes | Yes | Yes | diff --git a/docs/MIGRATION.md b/docs/MIGRATION.md index c419e66..424878e 100644 --- a/docs/MIGRATION.md +++ b/docs/MIGRATION.md @@ -83,8 +83,6 @@ config: compression: zstd keda: - kafka: - lagThreshold: "500" maxReplicaCount: 20 ``` @@ -93,24 +91,25 @@ 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** inherited from `common.yaml` — only override pipeline-specific values -- **KEDA** configured declaratively, inherits defaults from common.yaml - **Kafka brokers** come from `DFE_TRANSFORM_SOURCE_BROKERS` and `DFE_TRANSFORM_SINK_BROKERS`. A `${VAR}` in the config is not expanded: neither the wrapper nor Vector 0.57+ interpolates environment variables -- **SASL credentials** are files in the mounted secret at `sasl.secret_dir`, - read by Vector's directory secret backend +- **SASL credentials** come from the Kafka Secret through + `DFE_TRANSFORM_{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 ### What Stays the Same - **Secrets**: the chart creates none. The Kafka SASL secret comes from wherever your deployment already supplies it -- **KEDA TriggerAuthentication**: keeps pointing at the same Kafka credential secret - **Node placement**: set the chart's `nodeSelector` and `tolerations` to match your existing pool ### What Changes | Element | Before | After | |---------|--------|-------| -| Chart source | `helm.vector.dev` v0.42.1 | `dfe-transform-vector.git` `/chart` | +| 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` | diff --git a/docs/architecture.md b/docs/architecture.md index ed69465..bc56b1a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -143,14 +143,7 @@ API, which the supervisor does not do. ## Build, chart and licence invariants -**The Vector version lives in three places and they must agree.** The constant -`VECTOR_VERSION` in `src/deployment.rs` is the source of truth. It is substituted -into the Dockerfile's `ARG VECTOR_VERSION`, and `chart/values.yaml` carries -`config.vector.version` separately. `version_check: strict` refuses to start a -pod when the config's version and the binary's disagree, so a drifted pair is an -outage rather than a warning. A test in `tests/integration/deployment.rs` -compares the committed chart value against the constant, because that pairing -once sat nine minor versions apart and only `version_check: warn` kept pods up. +**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. **`password` is a `String` on purpose, not a `SensitiveString`.** The config goes through a figment serialize-merge-deserialize round trip in @@ -158,10 +151,12 @@ goes through a figment serialize-merge-deserialize round trip in 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: the mounted secret at +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. Changing the type to -look safer breaks authentication. +`/.secrets` from a credential given as text. The released chart +takes the second path: its Kafka Secret arrives in `DFE_TRANSFORM_{SOURCE,SINK}_SASL_*`, +and `` sits on the `run` writable path. Changing the type to look +safer breaks authentication. **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. @@ -171,13 +166,9 @@ 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. -**The committed chart is not pure generator output.** `chart/` carries KEDA -hand-edits the generator does not produce, notably a ScaledObject addressed at -`config.source.*` where the generator emits `config.kafka.*`. A test asserts the -committed chart matches the generator except for four files that are exempted -and asserted to keep diverging, so the exemption cannot go stale unnoticed. A -correction landing in scalo's generator does not reach the ScaledObject by -itself and is carried across by hand. +**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. + +**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. ## Where the transports differ diff --git a/docs/config-schema.json b/docs/config-schema.json index 1a2f3ad..5a1a078 100644 --- a/docs/config-schema.json +++ b/docs/config-schema.json @@ -1,7 +1,7 @@ { "$schema": "https://json-schema.org/draft/2020-12/schema", "title": "Config", - "description": "Main configuration.\n\n## Hot-reload behaviour\n\n**Hot-reloaded (takes effect on next poll cycle / SIGHUP):**\n- Transform YAML file contents (modified/added/removed in watched directory)\n- `transforms.dir` -- watcher switches to new directory after successful reload\n- `transforms.files` -- watcher switches to new file list after successful reload\n\n**Requires pod restart:**\n- `source.*` — the consumer or the Push listener is established at startup\n- `sink.*` — the producer or the Push client is established at startup\n- `bridge.*` — the two supervisor-to-Vector legs bind at startup\n- `pipeline.name` — used in consumer group_id and metrics labels at startup\n- `vector.*` — binary path, data_dir, config_dir, API switch and address,\n log level set at Vector spawn\n- `metrics.address` -- the HTTP server binds at startup\n- `logging.*` -- tracing subscriber configured at startup\n- `reload.poll_interval_secs` -- captured at reload loop start\n\n**Read by nothing in this process:**\n- `scaling.pressure_threshold` -- no reader. Scale-out is driven by the\n chart's KEDA triggers (consumer-group lag and CPU), gated by the\n subprocess circuit in [`crate::metrics::spawn_circuit_gate_task`].\n scalo's runtime reads its own `scaling.enabled` and\n `scaling.memory_gate_threshold` from the cascade, not this key.", + "description": "Main configuration.\n\n## Hot-reload behaviour\n\n**Hot-reloaded (takes effect on next poll cycle / SIGHUP):**\n- Transform YAML file contents (modified/added/removed in watched directory)\n- `transforms.dir` -- watcher switches to new directory after successful reload\n- `transforms.files` -- watcher switches to new file list after successful reload\n\n**Requires pod restart:**\n- `source.*` — the consumer or the Push listener is established at startup\n- `sink.*` — the producer or the Push client is established at startup\n- `bridge.*` — the two supervisor-to-Vector legs bind at startup\n- `pipeline.name` — used in consumer group_id and metrics labels at startup\n- `vector.*` — binary path, data_dir, config_dir, API switch and address,\n log level set at Vector spawn\n- `metrics.address` -- the HTTP server binds at startup\n- `logging.*` -- tracing subscriber configured at startup\n- `reload.poll_interval_secs` -- captured at reload loop start\n\n**Read by nothing in this process:**\n- `scaling.pressure_threshold` -- no reader. Scale-out is driven by the\n chart's KEDA ScaledObject (CPU, plus any trigger a deployment adds),\n gated by the subprocess circuit in [`crate::metrics::spawn_circuit_gate_task`].\n scalo's runtime reads its own `scaling.enabled` and\n `scaling.memory_gate_threshold` from the cascade, not this key.", "type": "object", "properties": { "dfe_source": { @@ -349,7 +349,6 @@ "description": "SASL password. Leave empty when `secret_dir` supplies it.\n\nTaken literally: Vector does not expand `${VAR}` placeholders, so a\nvalue holding one is refused.", "type": "string", "x-scalo-secret": true, - "x-dfe-secret": true, "writeOnly": true, "default": "" }, diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index 9b00425..3ddef40 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -23,8 +23,8 @@ description: |- **Read by nothing in this process:** - `scaling.pressure_threshold` -- no reader. Scale-out is driven by the - chart's KEDA triggers (consumer-group lag and CPU), gated by the - subprocess circuit in [`crate::metrics::spawn_circuit_gate_task`]. + chart's KEDA ScaledObject (CPU, plus any trigger a deployment adds), + gated by the subprocess circuit in [`crate::metrics::spawn_circuit_gate_task`]. scalo's runtime reads its own `scaling.enabled` and `scaling.memory_gate_threshold` from the cascade, not this key. type: object @@ -336,7 +336,6 @@ $defs: value holding one is refused. type: string x-scalo-secret: true - x-dfe-secret: true writeOnly: true default: '' secret_dir: diff --git a/src/config/loader.rs b/src/config/loader.rs index eecd494..58e86cd 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -142,8 +142,8 @@ impl Default for DecodingConfig { /// /// **Read by nothing in this process:** /// - `scaling.pressure_threshold` -- no reader. Scale-out is driven by the -/// chart's KEDA triggers (consumer-group lag and CPU), gated by the -/// subprocess circuit in [`crate::metrics::spawn_circuit_gate_task`]. +/// chart's KEDA ScaledObject (CPU, plus any trigger a deployment adds), +/// gated by the subprocess circuit in [`crate::metrics::spawn_circuit_gate_task`]. /// scalo's runtime reads its own `scaling.enabled` and /// `scaling.memory_gate_threshold` from the cascade, not this key. #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] @@ -663,12 +663,14 @@ use scalo::config::flat_env::{self, ApplyFlatEnv, Normalize}; /// scalo's runtime owns (`version_check`, `metrics` and the rest) resolve from /// the cascade alone. A reload re-enters here, and the cascade is set once per /// process. +/// +/// It also reads `./.env` into the process environment before the env layers +/// in [`Config::load`] run, and no parent directory's `.env`. fn init_cascade(path: Option<&str>) -> Result<()> { let opts = scalo::config::ConfigOptions { env_prefix: ENV_PREFIX.to_string(), config_paths: path.map(std::path::PathBuf::from).into_iter().collect(), - // `Config::load` has already read `.env` into the process environment. - load_dotenv: false, + load_dotenv: true, ..Default::default() }; match scalo::config::setup(opts) { @@ -707,6 +709,16 @@ impl ApplyFlatEnv for Config { self.pipeline.name = v; } + // A TLS listener turns TLS on for both endpoints. Any other protocol + // leaves the setting alone: writing false would switch off TLS the + // config file turned on. + if flat_env::flat_env_string(prefix, "KAFKA_SECURITY_PROTOCOL") + .is_some_and(|protocol| protocol.to_ascii_uppercase().contains("SSL")) + { + self.source.tls.enabled = true; + self.sink.tls.enabled = true; + } + // Source if let Some(v) = flat_env::flat_env_string(prefix, "SOURCE_TRANSPORT") { self.source.transport = parse_transport(&v, self.source.transport); @@ -836,13 +848,10 @@ impl Config { /// 1. CLI args (handled by caller) /// 2. Flat env overrides (`DFE_TRANSFORM_SOURCE_BROKERS`, etc.) /// 3. Figment env vars with `__` nesting (`DFE_TRANSFORM_SOURCE__BROKERS`) - /// 4. `.env` file (via dotenvy) + /// 4. `.env` in the working directory, read by scalo's cascade, never a parent's /// 5. Config YAML file /// 6. Hard-coded defaults pub fn load(config_path: Option<&str>) -> Result { - // Load .env file if present (before any env var reading) - let _ = dotenvy::dotenv(); - // Start with defaults let mut config = Config::default(); let mut loaded_from = None; diff --git a/src/config/mod.rs b/src/config/mod.rs index 407bd1e..8206e71 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -11,7 +11,7 @@ //! Configuration cascade (highest to lowest priority): //! 1. CLI args (--config, --log-level, etc.) //! 2. Environment variables (`DFE_TRANSFORM_*`) -//! 3. `.env` file (via dotenvy) +//! 3. `./.env` in the working directory, never a parent's //! 4. Config file specified by `--config` //! 5. Hard-coded defaults diff --git a/src/deployment.rs b/src/deployment.rs index 30294b3..52286c0 100644 --- a/src/deployment.rs +++ b/src/deployment.rs @@ -13,8 +13,10 @@ //! health paths, ports, secrets, KEDA scaling, and default config. use scalo::deployment::{ - DeploymentContract, HealthContract, ImageProfile, KedaConfig, KedaContract, NativeDepsContract, - PortContract, SecretEnvContract, SecretGroupContract, base_image_from_cascade, + CONTRACT_SCHEMA_VERSION, DeploymentContract, HealthContract, ImageProfile, KafkaLagTrigger, + KedaConfig, KedaContract, NativeDepsContract, PortContract, ResourceList, ResourcesContract, + SecretEnvContract, SecretGroupContract, SecurityContract, WritablePath, + base_image_from_cascade, }; /// Build the deployment contract for dfe-transform-vector. @@ -28,12 +30,11 @@ pub fn contract() -> DeploymentContract { DeploymentContract { app_name: "dfe-transform-vector".into(), binary_name: "dfe-transform-vector".into(), - description: "Vector.dev subprocess wrapper for Kafka-to-Kafka transform pipelines".into(), + description: "Vector subprocess wrapper for Kafka-to-Kafka transform pipelines".into(), metrics_port: 9090, health: HealthContract { - liveness_path: "/livez".into(), - readiness_path: "/readyz".into(), - metrics_path: "/metrics".into(), + startup_budget_seconds: 120, + ..HealthContract::default() }, env_prefix: "DFE_TRANSFORM".into(), metric_prefix: "transform_vector".into(), @@ -47,31 +48,15 @@ pub fn contract() -> DeploymentContract { extra_ports: vec![ PortContract::tcp("push", 6000) .when_equals("config.source.transport", "direct") - .bound_from("source.listen"), + .bound_from("source.listen") + .app_protocol("kubernetes.io/h2c"), ], unbound_listen_paths: vec![], entrypoint_args: vec![ "--config".into(), "/etc/dfe-transform-vector/config.yaml".into(), ], - // Drives the chart's kafka Secret and its `kafka.*` values. The committed - // Deployment mounts that Secret at KAFKA_SECRET_DIR rather than - // exporting these env vars, which nothing reads. - secrets: vec![SecretGroupContract { - group_name: "kafka".into(), - env_vars: vec![ - SecretEnvContract { - env_var: "KAFKA_SASL_USERNAME".into(), - key_name: "username".into(), - secret_key: "kafka-username".into(), - }, - SecretEnvContract { - env_var: "KAFKA_SASL_PASSWORD".into(), - key_name: "password".into(), - secret_key: "kafka-password".into(), - }, - ], - }], + secrets: secrets(), default_config: Some(serde_json::json!({ "pipeline": { "name": "default" @@ -83,14 +68,8 @@ pub fn contract() -> DeploymentContract { "topics": ["raw_events"], "group_id": "dfe-transform-vector-default", "decoding": { "codec": "json" }, - // The chart mounts the kafka secret here, and Vector reads the - // username and password files itself, so no credential sits in - // the ConfigMap or in the assembled Vector config. - "sasl": { - "enabled": true, - "mechanism": "scram_sha_512", - "secret_dir": KAFKA_SECRET_DIR - }, + // The username and password arrive through the env vars `secrets()` declares, never the ConfigMap. + "sasl": { "enabled": true, "mechanism": "scram_sha_512" }, "tls": { "enabled": false }, "acknowledgements": { "enabled": true } }, @@ -102,11 +81,7 @@ pub fn contract() -> DeploymentContract { "key_field": ".org_id", "encoding": "json", "compression": "zstd", - "sasl": { - "enabled": true, - "mechanism": "scram_sha_512", - "secret_dir": KAFKA_SECRET_DIR - }, + "sasl": { "enabled": true, "mechanism": "scram_sha_512" }, "tls": { "enabled": false } }, // The two loopback legs between the supervisor and Vector, used on @@ -148,24 +123,28 @@ pub fn contract() -> DeploymentContract { // values into a `KedaConfig` and convert; the new `scaling_pressure_*` // trigger fields stay at their defaults (OFF -- the Prometheus // serverAddress is cluster-specific and must be set before enabling). - keda: Some(KedaContract::from_config(&KedaConfig { - min_replicas: 1, - max_replicas: 10, - polling_interval: 15, - cooldown_period: 300, - kafka_lag_threshold: 1000, - activation_lag_threshold: 0, - cpu_enabled: true, - cpu_threshold: 80, - ..Default::default() - })), + keda: Some( + KedaContract::from_config(&KedaConfig { + min_replicas: 1, + max_replicas: 10, + polling_interval: 15, + cooldown_period: 300, + kafka_lag_threshold: 1000, + activation_lag_threshold: 0, + cpu_enabled: true, + cpu_threshold: 80, + ..Default::default() + }) + // Raw consumer-group lag rises when a downstream stage breaks, so it never scales this app. + .with_kafka_trigger(KafkaLagTrigger::disabled()), + ), // Org-wide base image via the scalo cascade (deployment.base_image), // defaulting to debian:trixie-slim. NOT pinned per-app -- the ubuntu // pin was a stale pre-trixie-cutover leftover. base_image: base_image_from_cascade(), native_deps: NativeDepsContract::default(), image_profile: ImageProfile::default(), - schema_version: 3, + schema_version: CONTRACT_SCHEMA_VERSION, // scalo writes no vendor, licence or copyright of its own, so the labels // and the generated Dockerfile header carry exactly these. oci_labels: scalo::deployment::OciLabels { @@ -182,9 +161,64 @@ pub fn contract() -> DeploymentContract { // managed Vector transform), not per-transform knobs. config_schema: Some(scalo::deployment::config_schema_json::()), capabilities: capabilities(), + // Vector's data directory, and the assembled config with its credential files, under a read-only root. + writable_paths: vec![ + WritablePath::new("data", "/var/lib/vector"), + WritablePath::new("run", "/var/run/vector"), + ], + termination_grace_seconds: 90, + resources: ResourcesContract { + requests: ResourceList { + cpu: "100m".into(), + memory: "128Mi".into(), + }, + limits: ResourceList { + cpu: "500m".into(), + memory: "512Mi".into(), + }, + }, + security: SecurityContract::default(), + singleton: false, } } +/// The Kafka Secret the chart mounts into each endpoint's SASL env vars. +/// +/// `Config::apply_flat_env` reads these per-endpoint names. Each `key_name` is +/// distinct because the chart renders it as a values key. +fn secrets() -> Vec { + let env = |env_var: &str, key_name: &str, secret_key: &str| SecretEnvContract { + env_var: env_var.into(), + key_name: key_name.into(), + secret_key: secret_key.into(), + }; + vec![SecretGroupContract::new( + "kafka", + vec![ + env( + "DFE_TRANSFORM_SOURCE_SASL_USERNAME", + "source-username", + "username", + ), + env( + "DFE_TRANSFORM_SOURCE_SASL_PASSWORD", + "source-password", + "password", + ), + env( + "DFE_TRANSFORM_SINK_SASL_USERNAME", + "sink-username", + "username", + ), + env( + "DFE_TRANSFORM_SINK_SASL_PASSWORD", + "sink-password", + "password", + ), + ], + )] +} + /// Capability catalog for dfe-transform-vector: the Vector-subprocess wrapper. /// Deliberately thin -- Vector's own YAML config owns the transform/routing /// surface (the documented anomaly), so this describes the wrapper only. @@ -212,9 +246,6 @@ fn capabilities() -> Vec { ] } -/// Where the chart mounts the kafka secret, one file per credential. -pub const KAFKA_SECRET_DIR: &str = "/var/run/secrets/dfe-kafka"; - /// Vector.dev version bundled into the published image. /// /// Pinned. Update deliberately — Vector minor versions can change CLI @@ -326,7 +357,7 @@ mod tests { #[test] fn test_contract_carries_reflectable_config() { let c = contract(); - assert_eq!(c.schema_version, 3); + assert_eq!(c.schema_version, CONTRACT_SCHEMA_VERSION); assert!(c.config_schema.is_some()); assert!(c.capabilities.iter().any(|cap| cap.name == "vector")); } @@ -363,6 +394,57 @@ mod tests { .find(|p| p.name == "push") .expect("the direct transport's Push listener must be a declared port"); assert_eq!(push.port, 6000); + // The Push listener is cleartext gRPC, so a proxy in front of it must speak h2c. + assert_eq!(push.app_protocol, "kubernetes.io/h2c"); + } + + /// KEDA stays on and scales on CPU alone. + #[test] + fn test_keda_scales_on_cpu_without_a_lag_trigger() { + let c = contract(); + let keda = c.keda.as_ref().expect("keda present"); + assert!(keda.enabled); + assert!(keda.cpu_enabled); + assert!( + !keda.kafka_trigger.enabled, + "raw consumer-group lag scales out when a downstream stage is broken" + ); + } + + /// `generate-artefacts` and the generators write nothing for a contract + /// that fails these checks, and the library refuses to render a values + /// path the contract leaves unset -- as it did the lag trigger's + /// `config.kafka.*`, which this app has no block for. + #[test] + fn test_contract_passes_the_generator_checks() { + let c = contract(); + c.validate() + .expect("every generator must accept the contract"); + scalo::deployment::assert_listeners_declared(&c); + let unresolved = c.unresolved_values_paths(); + assert!( + unresolved.is_empty(), + "the chart reads values default_config never sets: {unresolved:?}" + ); + } + + /// The root filesystem is read-only, so every directory the app writes by + /// default sits under an ungated writable path or the pod cannot start. + #[test] + fn test_every_default_write_directory_is_a_writable_path() { + let c = contract(); + assert!(c.security.read_only_root_filesystem); + let vector = crate::config::Config::default().vector; + for dir in [&vector.data_dir, &vector.config_dir] { + assert!( + c.writable_paths + .iter() + .any(|writable| writable.when.is_none() + && std::path::Path::new(dir).starts_with(&writable.path)), + "no writable path covers {dir}: {:?}", + c.writable_paths + ); + } } /// The committed reflectable artefacts under docs/ must not drift from a diff --git a/tests/common/mod.rs b/tests/common/mod.rs index a4ab5df..567aa71 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -21,8 +21,8 @@ //! //! The fixture owns the container (if any) and shuts it down on drop. //! -//! Credentials come from the project `.env` (never hard-coded). `.env` is -//! loaded via `dotenvy` at the first call. +//! Credentials come from the project `.env` (never hard-coded), loaded from +//! the repo root at the first call. use std::env; use std::net::ToSocketAddrs; @@ -150,8 +150,9 @@ impl KafkaFixture { } } +/// Load this repo's own `.env`: `dotenvy::dotenv()` would load the first `.env` in any parent. pub fn load_dotenv() { - let _ = dotenvy::dotenv(); + let _ = dotenvy::from_path(std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(".env")); } /// Build a `KafkaConfig` from environment variables (live-cluster mode). diff --git a/tests/integration/config.rs b/tests/integration/config.rs index 70a0326..12d72e8 100644 --- a/tests/integration/config.rs +++ b/tests/integration/config.rs @@ -17,7 +17,7 @@ use dfe_transform_vector::config::assembler; use dfe_transform_vector::config::loader::*; use tempfile::TempDir; -use crate::integration::config_env::{env_write_guard, load_config}; +use crate::integration::config_env::{env_write_guard, load_config, with_vars}; /// Build a complete test config with realistic values. fn full_config(transforms_dir: Option) -> Config { @@ -483,6 +483,119 @@ fn config_env_override_flat() { assert_eq!(config.pipeline.name, "env-override-test"); } +/// `config` with `DFE_TRANSFORM_KAFKA_SECURITY_PROTOCOL` set to `protocol`, +/// or unset for `None`, through the flat env layer. +fn with_security_protocol(config: Config, protocol: Option<&str>) -> Config { + use scalo::config::flat_env::ApplyFlatEnv; + + with_vars( + &[("DFE_TRANSFORM_KAFKA_SECURITY_PROTOCOL", protocol)], + || { + let mut config = config; + config.apply_flat_env("DFE_TRANSFORM"); + config + }, + ) +} + +/// The chart stamps the broker's protocol, and a TLS listener needs TLS on +/// both endpoints whatever case the protocol is spelled in. +#[test] +fn an_ssl_security_protocol_turns_tls_on_for_both_endpoints() { + for protocol in ["SASL_SSL", "SSL", "sasl_ssl"] { + let config = with_security_protocol(Config::default(), Some(protocol)); + assert!( + config.source.tls.enabled, + "{protocol}: source TLS stayed off" + ); + assert!(config.sink.tls.enabled, "{protocol}: sink TLS stayed off"); + } +} + +#[test] +fn a_plaintext_or_unset_security_protocol_leaves_tls_off() { + for protocol in [Some("SASL_PLAINTEXT"), Some("PLAINTEXT"), None] { + let config = with_security_protocol(Config::default(), protocol); + assert!( + !config.source.tls.enabled, + "{protocol:?}: source TLS came on" + ); + assert!(!config.sink.tls.enabled, "{protocol:?}: sink TLS came on"); + } +} + +/// The env only ever turns TLS on, so a plaintext protocol cannot switch off +/// the TLS a config file asked for. +#[test] +fn a_plaintext_security_protocol_keeps_tls_the_config_file_turned_on() { + let file: Config = serde_yaml_ng::from_str( + "source:\n tls:\n enabled: true\nsink:\n tls:\n enabled: true\n", + ) + .unwrap(); + + let config = with_security_protocol(file, Some("SASL_PLAINTEXT")); + + assert!(config.source.tls.enabled, "source TLS was switched off"); + assert!(config.sink.tls.enabled, "sink TLS was switched off"); +} + +/// `config-check` run by the binary from `dir`, returning what it printed. +fn config_check_in(dir: &std::path::Path) -> String { + let out = std::process::Command::new(env!("CARGO_BIN_EXE_dfe-transform-vector")) + .arg("config-check") + .current_dir(dir) + .env_remove("DFE_TRANSFORM_PIPELINE_NAME") + .env_remove("DFE_TRANSFORM_SINK_TOPIC") + .output() + .expect("the binary runs"); + let printed = String::from_utf8_lossy(&out.stderr).into_owned(); + assert!(out.status.success(), "config-check failed:\n{printed}"); + printed +} + +/// The binary reads the `.env` in its working directory and no other. +/// +/// A `.env` in a parent directory belongs to whatever project sits above, so a +/// search up the tree loads another project's settings and credentials. +#[test] +fn a_dotenv_in_a_parent_directory_is_not_loaded() { + let root = TempDir::new().expect("tempdir"); + let project = root.path().join("project"); + fs::create_dir(&project).expect("project dir"); + fs::write( + project.join("config.yaml"), + "sink:\n topic: from_the_file\n", + ) + .expect("project config"); + fs::write( + root.path().join(".env"), + "DFE_TRANSFORM_PIPELINE_NAME=from_parent_dotenv\n", + ) + .expect("parent .env"); + + let printed = config_check_in(&project); + assert!( + !printed.contains("from_parent_dotenv"), + "a .env in the parent directory reached the config" + ); + + // The project's own .env still loads. + fs::write( + project.join(".env"), + "DFE_TRANSFORM_SINK_TOPIC=from_project_dotenv\n", + ) + .expect("project .env"); + let printed = config_check_in(&project); + assert!( + printed.contains("from_project_dotenv"), + "the project's own .env did not reach the config" + ); + assert!( + !printed.contains("from_parent_dotenv"), + "a .env in the parent directory reached the config" + ); +} + // ========================================================================= // KafkaSource integration tests // ========================================================================= diff --git a/tests/integration/config_env.rs b/tests/integration/config_env.rs index ef78969..2184d7d 100644 --- a/tests/integration/config_env.rs +++ b/tests/integration/config_env.rs @@ -44,3 +44,25 @@ pub fn load_config(path: Option<&str>) -> Result { pub fn env_write_guard() -> RwLockWriteGuard<'static, ()> { CONFIG_ENV.write().unwrap_or_else(|e| e.into_inner()) } + +/// Run `f` with each name set to its value, or unset for `None`, then remove +/// every one of them. +pub fn with_vars(vars: &[(&str, Option<&str>)], f: impl FnOnce() -> T) -> T { + let _exclusive = env_write_guard(); + // SAFETY: `_exclusive` excludes every guarded reader and writer of the + // environment until the names are removed again. + unsafe { + for (name, value) in vars { + match value { + Some(v) => std::env::set_var(name, v), + None => std::env::remove_var(name), + } + } + } + let result = f(); + for (name, _) in vars { + // SAFETY: as above. + unsafe { std::env::remove_var(name) }; + } + result +} diff --git a/tests/integration/deployment.rs b/tests/integration/deployment.rs index 1b1b5dc..1a926a6 100644 --- a/tests/integration/deployment.rs +++ b/tests/integration/deployment.rs @@ -10,6 +10,9 @@ //! compose fragments, and contract JSON match the expected structure. use dfe_transform_vector::deployment; +use scalo::config::flat_env::{ApplyFlatEnv, Normalize}; + +use crate::integration::config_env::with_vars; #[test] fn contract_produces_valid_structure() { @@ -58,20 +61,23 @@ fn contract_produces_valid_structure() { assert!(keda.min_replicas >= 1); assert!(keda.max_replicas > keda.min_replicas); - // Kafka secrets defined - assert!(!c.secrets.is_empty(), "Kafka secrets must be defined"); + // One required Kafka Secret, mounted into both endpoints' SASL names. + assert_eq!(c.secrets.len(), 1, "one Kafka secret group"); + assert_eq!(c.secrets[0].group_name, "kafka"); + assert!(!c.secrets[0].optional); let env_vars: Vec<&str> = c.secrets[0] .env_vars .iter() .map(|s| s.env_var.as_str()) .collect(); - assert!( - env_vars.contains(&"KAFKA_SASL_USERNAME"), - "missing KAFKA_SASL_USERNAME secret" - ); - assert!( - env_vars.contains(&"KAFKA_SASL_PASSWORD"), - "missing KAFKA_SASL_PASSWORD secret" + assert_eq!( + env_vars, + [ + "DFE_TRANSFORM_SOURCE_SASL_USERNAME", + "DFE_TRANSFORM_SOURCE_SASL_PASSWORD", + "DFE_TRANSFORM_SINK_SASL_USERNAME", + "DFE_TRANSFORM_SINK_SASL_PASSWORD", + ] ); // Default config present and parseable @@ -81,6 +87,48 @@ fn contract_produces_valid_structure() { ); } +/// The chart mounts a Secret under every declared name, so a name the config +/// never reads leaves the credential silently unused. +/// +/// Each name spells the field it fills -- `DFE_TRANSFORM_SINK_SASL_PASSWORD` is +/// `sink.sasl.password` -- so a name read into the other endpoint fails here too. +#[test] +fn every_contract_secret_env_var_reaches_the_config() { + use dfe_transform_vector::config::loader::Config; + + let c = deployment::contract(); + let prefix = c.env_prefix.clone(); + + for group in &c.secrets { + for env in &group.env_vars { + let field = env + .env_var + .strip_prefix(&format!("{prefix}_")) + .unwrap_or_else(|| panic!("{} lacks the {prefix} prefix", env.env_var)); + let pointer = format!("/{}", field.to_ascii_lowercase().replace('_', "/")); + + let sentinel = format!("sentinel-{}", env.key_name); + let config = with_vars(&[(env.env_var.as_str(), Some(sentinel.as_str()))], || { + let mut config = Config::default(); + config.apply_flat_env(&prefix); + config + }); + + let applied = serde_json::to_value(&config).expect("config serialises"); + let reached = applied + .pointer(&pointer) + .and_then(serde_json::Value::as_str) + == Some(sentinel.as_str()); + // The message carries the env var and group names only, never a value. + assert!( + reached, + "{} ({}) was set and the config field its name spells did not read it", + env.env_var, group.group_name + ); + } + } +} + #[test] fn emit_dockerfile_produces_valid_output() { let dockerfile = deployment::emit_dockerfile(); @@ -221,435 +269,57 @@ fn emit_chart_generates_without_panic() { assert!(templates.is_dir(), "templates/ not generated"); } -/// The committed KEDA ScaledObject is a deliberate hand-edit of the generated -/// one, and a fresh `emit-chart` over `chart/` silently clobbers it. -/// -/// The generator addresses `.Values.config.kafka.*`, which this app has no -/// block for, and hardcodes the broker's SASL mechanism and TLS mode. A -/// regenerated ScaledObject therefore renders empty `bootstrapServers`, -/// `consumerGroup` and `topic`, and describes an auth posture the app may not -/// be using — all valid YAML that KEDA accepts and then never scales on, with -/// nothing logged. -#[test] -fn checked_in_keda_scaledobject_survives_emit_chart() { - let committed_path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) - .join("chart/templates/keda-scaledobject.yaml"); - let committed = std::fs::read_to_string(&committed_path).expect("read committed ScaledObject"); - - // Assert on the trigger metadata itself, not on a substring anywhere in the - // file, so the explanatory note at the top cannot satisfy or trip it. - let directive = |key: &str| { - committed - .lines() - .map(str::trim) - .find(|l| l.starts_with(&format!("{key}:"))) - .unwrap_or_else(|| panic!("committed ScaledObject has no `{key}:` line")) - .to_string() - }; - - for (key, expected) in [ - ("bootstrapServers", ".Values.config.source.brokers"), - ("consumerGroup", ".Values.config.source.group_id"), - ("sasl", ".Values.config.source.sasl"), - ("tls", ".Values.config.source.tls"), - ("unsafeSsl", ".Values.config.source.tls"), - ] { - let line = directive(key); - assert!( - line.contains(expected), - "chart/templates/keda-scaledobject.yaml has been overwritten by `emit-chart`: \ - `{line}` does not read {expected}. The generator addresses config.kafka.*, which \ - this chart does not define, and hardcodes the mechanism and TLS mode, so KEDA \ - would get an empty or wrong value here and stop scaling with nothing logged." - ); - } - assert!( - committed.contains("index .Values.config.source.topics 0"), - "committed ScaledObject no longer derives the topic from config.source.topics" - ); - - let values = std::fs::read_to_string( - std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/values.yaml"), - ) - .expect("read chart values"); - let values: serde_yaml_ng::Value = - serde_yaml_ng::from_str(&values).expect("parse chart values"); - let source = values - .get("config") - .and_then(|c| c.get("source")) - .expect("chart values define config.source"); - for key in ["brokers", "group_id", "topics"] { - assert!( - source.get(key).is_some(), - "chart values.yaml has no config.source.{key} for the ScaledObject to read" - ); - } - // The trigger reads these to decide the scaler's auth posture. Omitting a - // block is safe -- the parenthesised lookups fall back to SASL off, TLS - // off -- but silently descending to `none` when the app is using SCRAM - // gives a scaler that cannot authenticate, so the values must say. - for (block, key) in [ - ("sasl", "enabled"), - ("sasl", "mechanism"), - ("tls", "enabled"), - ] { - assert!( - source.get(block).and_then(|b| b.get(key)).is_some(), - "chart values.yaml has no config.source.{block}.{key}; the ScaledObject trigger \ - would fall back to an unauthenticated scaler against an authenticated broker" - ); - } - - // The divergence itself: if scalo's generator ever converges on this shape, - // the hand-edit and its note in the template are stale. - let dir = tempfile::tempdir().expect("create temp dir"); - scalo::deployment::generate_chart(&deployment::contract(), dir.path().to_str().unwrap(), None) - .expect("chart generation"); - let generated = std::fs::read_to_string(dir.path().join("templates/keda-scaledobject.yaml")) - .expect("read generated ScaledObject"); - assert_ne!( - generated.trim(), - committed.trim(), - "the generator now emits the committed ScaledObject verbatim — drop the hand-edit \ - note from chart/templates/keda-scaledobject.yaml and this divergence assertion" - ); -} - -/// The committed `chart/` must be what the generator produces, file by file. -/// -/// `checked_in_keda_scaledobject_survives_emit_chart` above guards one file -/// against being clobbered. Nothing guarded the rest, so a scalo release that -/// corrected the generator left the shipped chart on the old output with no -/// test failing: the KEDA TriggerAuthentication bound the username to -/// `parameter: sasl`, which KEDA reads as the mechanism enum, so the scaler had -/// no username, never read consumer-group lag, and the app never scaled. +/// The contract's default config, with the Kafka Secret mounted into the env +/// vars the contract declares, must be a config the app accepts. /// -/// Files in `HAND_FIXED` are asserted to still DIVERGE, so when the generator -/// converges on the hand-edit the test says to drop the exemption rather than -/// leaving a stale divergence nobody rechecks. -#[test] -fn committed_chart_matches_the_generator() { - // Each of these addresses app-specific values the generator does not model: - // the `config.source.*` Kafka addressing, the configurable Push port, and - // the kafka secret mounted as the files Vector reads its credentials from. - const HAND_FIXED: &[&str] = &[ - "values.yaml", - "templates/deployment.yaml", - "templates/service.yaml", - "templates/keda-scaledobject.yaml", - ]; - - let generated = tempfile::tempdir().expect("temp dir"); - scalo::deployment::generate_chart( - &deployment::contract(), - generated.path().to_str().expect("temp dir path"), - None, - ) - .expect("chart generates"); - - let want = read_chart(generated.path()); - let got = read_chart(&std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart")); - - assert_eq!( - want.keys().collect::>(), - got.keys().collect::>(), - "chart/ holds a different set of files from `emit-chart`" - ); - - for (rel, from_generator) in &want { - if HAND_FIXED.contains(&rel.as_str()) { - assert_ne!( - got[rel], *from_generator, - "chart/{rel} is listed as hand-fixed but now matches the generator -- \ - drop it from HAND_FIXED" - ); - continue; - } - assert_eq!( - got[rel], *from_generator, - "chart/{rel} has drifted from `emit-chart` -- fix contract() and regenerate, \ - never hand-edit the output" - ); - } -} - -/// Relative path -> contents for a chart directory (root files plus -/// `templates/`, which is the whole shape the generator emits). -fn read_chart(dir: &std::path::Path) -> std::collections::BTreeMap { - let mut out = std::collections::BTreeMap::new(); - for sub in [None, Some("templates")] { - let here = sub.map_or_else(|| dir.to_path_buf(), |s| dir.join(s)); - for entry in std::fs::read_dir(&here).expect("chart directory readable") { - let entry = entry.expect("chart directory entry"); - if !entry.file_type().expect("file type").is_file() { - continue; - } - let name = entry.file_name().to_string_lossy().into_owned(); - let rel = sub.map_or_else(|| name.clone(), |s| format!("{s}/{name}")); - let body = std::fs::read_to_string(entry.path()).expect("chart file readable"); - out.insert(rel, body); - } - } - out -} - -/// Render the chart and read the trigger Helm actually produces. -/// -/// Every other guard here reads the template as text, which cannot tell a -/// working conditional from one that renders the wrong branch or refuses -/// outright. The nil-safe parentheses and the `eq ... "true"` comparison exist -/// because Go truthiness reads a quoted `"false"` as true and a missing block -/// as a fatal dereference; both are invisible to a substring check. -/// -/// Skips without helm on PATH. It is not on the CI test runner today, so treat -/// this as a developer-machine gate until it is. -#[test] -fn the_keda_trigger_renders_the_auth_posture_the_app_uses() { - let chart = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart"); - let render = |args: &[&str]| -> Result { - let out = std::process::Command::new("helm") - .args(["template", "t"]) - .arg(&chart) - .args(["--show-only", "templates/keda-scaledobject.yaml"]) - .args(args) - .output() - .map_err(|e| format!("run helm: {e}"))?; - if out.status.success() { - Ok(String::from_utf8_lossy(&out.stdout).to_string()) - } else { - Err(String::from_utf8_lossy(&out.stderr).to_string()) - } - }; - - if render(&[]).is_err() - && std::process::Command::new("helm") - .arg("version") - .output() - .is_err() - { - eprintln!("SKIP: helm not on PATH, cannot render the chart"); - return; - } - - for (case, args, expected) in [ - ( - "stock values", - vec![], - vec!["sasl: scram_sha512", "tls: disable", "unsafeSsl: \"false\""], - ), - ( - "both optional blocks omitted", - vec![ - "--set", - "config.source.sasl=null", - "--set", - "config.source.tls=null", - ], - vec!["sasl: none", "tls: disable"], - ), - ( - "sasl off as a quoted string, which Go truthiness reads as true", - vec!["--set-string", "config.source.sasl.enabled=false"], - vec!["sasl: none"], - ), - ( - "plain maps to KEDA's spelling", - vec!["--set", "config.source.sasl.mechanism=plain"], - vec!["sasl: plaintext"], - ), - ( - "tls on with skip_verify reaches KEDA too", - vec![ - "--set", - "config.source.tls.enabled=true", - "--set", - "config.source.tls.skip_verify=true", - ], - vec!["tls: enable", "unsafeSsl: \"true\""], - ), - ] { - let rendered = - render(&args).unwrap_or_else(|e| panic!("{case}: helm refused to render: {e}")); - for want in expected { - assert!( - rendered.contains(want), - "{case}: rendered trigger has no `{want}`:\n{rendered}" - ); - } - } - - // A mechanism the app rejects must stop the chart rather than emit an - // empty `sasl:`, which KEDA would fail on with nothing pointing back here. - let err = render(&["--set", "config.source.sasl.mechanism=oauthbearer"]) - .expect_err("an unmapped mechanism must fail the render"); - assert!( - err.contains("no KEDA sasl mechanism"), - "unmapped mechanism failed for the wrong reason: {err}" - ); -} - -/// The chart's expected Vector version must match the one the Dockerfile bakes. -/// -/// They drifted by nine minor versions once already -- the chart said 0.48.0 -/// against a 0.57.0 binary -- and only `version_check: warn` kept pods -/// starting. Under `strict` that combination refuses to start at all. -#[test] -fn chart_expects_the_vector_version_the_image_ships() { - let values = std::fs::read_to_string( - std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/values.yaml"), - ) - .expect("read chart values"); - let values: serde_yaml_ng::Value = - serde_yaml_ng::from_str(&values).expect("parse chart values"); - - let charted = values - .get("config") - .and_then(|c| c.get("vector")) - .and_then(|v| v.get("version")) - .and_then(serde_yaml_ng::Value::as_str) - .expect("chart values define config.vector.version"); - - assert_eq!( - charted, - deployment::VECTOR_VERSION, - "chart/values.yaml expects Vector {charted}, the image ships {}", - deployment::VECTOR_VERSION - ); -} - -/// The chart's `config` block must deserialise and validate. -/// -/// The ConfigMap is `toYaml .Values.config` verbatim, so that block is the -/// config the pod reads, and anything `validate()` rejects is a crash-loop -/// before Vector starts. Going through the real validator rather than -/// asserting on substrings covers every rule it enforces, not just today's. -/// -/// It does NOT prove a stock install can authenticate: `kafka.username` is -/// empty until an operator sets it, so the mounted secret holds nothing to -/// log in with. This is the config-shape half only. -#[test] -fn chart_default_values_are_a_config_the_app_accepts() { - use dfe_transform_vector::config::loader::Config; - - let values = std::fs::read_to_string( - std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/values.yaml"), - ) - .expect("read chart values"); - let values: serde_yaml_ng::Value = - serde_yaml_ng::from_str(&values).expect("parse chart values"); - let block = values - .get("config") - .expect("chart values define a config block") - .clone(); - - let config: Config = serde_yaml_ng::from_value(block) - .expect("the chart's config block must deserialise into the app's Config"); - - config.validate().expect( - "chart/values.yaml does not validate -- `helm install` with stock values would \ - crash-loop the pod before Vector starts", - ); -} - -/// The same check one level up, on the contract the generator reads. -/// -/// `chart/values.yaml` is emitted from `default_config`, so a chart generated -/// anywhere -- a fresh `emit-chart`, a sibling repo's tooling -- is only as -/// good as this. Guarding the committed file alone would let the next -/// regeneration ship a broken chart and pass. +/// `emit-chart` writes `default_config` into its values, so a chart generated +/// from the contract is only as good as this, and anything `validate()` rejects +/// is a crash-loop before Vector starts. #[test] fn the_contract_default_config_is_one_the_app_accepts() { use dfe_transform_vector::config::loader::Config; - let default_config = deployment::contract() + let c = deployment::contract(); + let default_config = c .default_config + .clone() .expect("contract carries a default config"); - let config: Config = serde_json::from_value(default_config) + let mut config: Config = serde_json::from_value(default_config) .expect("the contract's default config must deserialise into the app's Config"); + let mounted: Vec<(&str, Option<&str>)> = c + .secrets + .iter() + .flat_map(|group| &group.env_vars) + .map(|env| (env.env_var.as_str(), Some("from-the-secret"))) + .collect(); + with_vars(&mounted, || { + config.apply_flat_env(&c.env_prefix); + config.normalize(); + }); + config .validate() .expect("deployment::contract()'s default_config does not validate"); } -/// SASL credentials reach Vector as files in the mounted kafka secret, so both -/// ends of that wiring have to agree. -/// -/// A literal credential in `config.source.sasl` would land in the ConfigMap -/// instead of the Secret, and a `${VAR}` placeholder reaches the broker as the -/// credential itself. A mount path that differs from `secret_dir` leaves Vector -/// failing its start on a file that is not there. -#[test] -fn the_sasl_secret_dir_is_where_the_deployment_mounts_the_secret() { - let values = std::fs::read_to_string( - std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/values.yaml"), - ) - .expect("read chart values"); - let parsed: serde_yaml_ng::Value = - serde_yaml_ng::from_str(&values).expect("parse chart values"); - - for side in ["source", "sink"] { - let sasl = parsed - .get("config") - .and_then(|c| c.get(side)) - .and_then(|s| s.get("sasl")) - .unwrap_or_else(|| panic!("chart values have no config.{side}.sasl")); - assert_eq!( - sasl.get("secret_dir") - .and_then(serde_yaml_ng::Value::as_str), - Some(deployment::KAFKA_SECRET_DIR), - "config.{side}.sasl.secret_dir must name the kafka secret mount" - ); - for key in ["username", "password"] { - assert!( - sasl.get(key).is_none(), - "config.{side}.sasl.{key} must be left to the mounted secret -- \ - `toYaml .Values.config` would put it in the ConfigMap" - ); - } - } - - // The other end: the Deployment mounts the secret there, as the two files. - let deployment_yaml = std::fs::read_to_string( - std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/templates/deployment.yaml"), - ) - .expect("read deployment template"); - assert!( - deployment_yaml.contains(&format!("mountPath: {}", deployment::KAFKA_SECRET_DIR)), - "chart/templates/deployment.yaml no longer mounts the kafka secret at {}", - deployment::KAFKA_SECRET_DIR - ); - for file in ["path: username", "path: password"] { - assert!( - deployment_yaml.contains(file), - "the kafka secret volume must project `{file}`, the name Vector reads" - ); - } - assert!( - !deployment_yaml.contains("KAFKA_SASL_PASSWORD"), - "no credential is exported to the environment any more" - ); -} - /// Vector's API takes unauthenticated requests and nothing here is a client, /// so no shipped artefact may publish its port. #[test] fn no_artefact_publishes_the_vector_api() { let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); - for rel in [ - "chart/templates/deployment.yaml", - "chart/templates/service.yaml", - "Dockerfile", - ] { - let text = std::fs::read_to_string(root.join(rel)).expect("read artefact"); - assert!( - !text.contains("8686"), - "{rel} still publishes Vector's API port" - ); - } - let default_config = deployment::contract() - .default_config - .expect("contract carries default config"); + let dockerfile = std::fs::read_to_string(root.join("Dockerfile")).expect("read Dockerfile"); + assert!( + !dockerfile.contains("8686"), + "Dockerfile still publishes Vector's API port" + ); + let c = deployment::contract(); + assert!( + c.extra_ports.iter().all(|p| p.port != 8686), + "the contract declares Vector's API port: {:?}", + c.extra_ports + ); + let default_config = c.default_config.expect("contract carries default config"); assert_eq!(default_config["vector"]["api_enabled"], false); } @@ -660,15 +330,6 @@ fn shipped_configs() -> Vec<(String, dfe_transform_vector::config::loader::Confi let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")); let mut out = Vec::new(); - let values: serde_yaml_ng::Value = serde_yaml_ng::from_str( - &std::fs::read_to_string(root.join("chart/values.yaml")).expect("read chart values"), - ) - .expect("parse chart values"); - out.push(( - "chart/values.yaml".to_string(), - serde_yaml_ng::from_value::(values["config"].clone()).expect("chart config"), - )); - out.push(( "deployment::contract().default_config".to_string(), serde_json::from_value::( @@ -707,7 +368,7 @@ fn no_env_placeholder_survives_assembly_of_a_shipped_config() { for (label, mut config) in shipped_configs() { let work = tempfile::tempdir().expect("work dir"); - // Where the chart mounts the secret; stand one up so assembly can + // A config that names a credential directory gets one, so assembly can // check the files are there. let mounted = work.path().join("secret"); std::fs::create_dir_all(&mounted).expect("secret dir"); @@ -718,17 +379,17 @@ fn no_env_placeholder_survives_assembly_of_a_shipped_config() { sasl.secret_dir = Some(mounted.to_string_lossy().into_owned()); } } - // A container path is read inside the filesystem the stock chart gives - // the pod, so a mount that hides it fails assembly here as it would there. - let pod = work.path().join("pod"); - stock_pod_filesystem(&pod); + // A container path is read inside the directories the image creates, + // so a config naming one the image lacks fails assembly here. + let image = work.path().join("image"); + image_filesystem(&image); if let Some(dir) = config .transforms .dir .clone() .filter(|d| std::path::Path::new(d).is_absolute()) { - config.transforms.dir = Some(inside(&pod, &dir).to_string_lossy().into_owned()); + config.transforms.dir = Some(inside(&image, &dir).to_string_lossy().into_owned()); } let assembled = work.path().join("config"); @@ -754,59 +415,13 @@ fn no_env_placeholder_survives_assembly_of_a_shipped_config() { } } -/// `path`, a container path, inside the pod filesystem stood up at `root`. +/// `path`, a container path, inside the filesystem stood up at `root`. fn inside(root: &std::path::Path, path: &str) -> std::path::PathBuf { root.join(path.trim_start_matches('/')) } -/// One entry under the Deployment's `volumeMounts`. -#[derive(Debug, Default)] -struct VolumeMount { - path: String, - sub_path: Option, -} - -/// The container's volume mounts, read from the Deployment template. -/// -/// Read as text because the template is not YAML until Helm renders it, and -/// Helm is not on the CI runner. -fn chart_volume_mounts() -> Vec { - let template = std::fs::read_to_string( - std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/templates/deployment.yaml"), - ) - .expect("read deployment template"); - let mut mounts: Vec = Vec::new(); - let mut in_mounts = false; - for line in template.lines().map(str::trim) { - match line { - "volumeMounts:" => in_mounts = true, - "volumes:" => in_mounts = false, - _ if !in_mounts => {} - _ if line.starts_with("- name:") => mounts.push(VolumeMount::default()), - _ => { - let Some(mount) = mounts.last_mut() else { - continue; - }; - if let Some(path) = line.strip_prefix("mountPath:") { - mount.path = path.trim().to_string(); - } else if let Some(sub) = line.strip_prefix("subPath:") { - mount.sub_path = Some(sub.trim().to_string()); - } - } - } - } - assert!( - mounts.iter().all(|m| !m.path.is_empty()), - "a volume mount without a mountPath: {mounts:?}" - ); - mounts -} - -/// Stand up under `root` the filesystem the stock chart gives the app: the -/// directories the image creates, then each volume the Deployment mounts. A -/// directory mount replaces whatever the image put at that path; a `subPath` -/// mount places one file. -fn stock_pod_filesystem(root: &std::path::Path) { +/// Stand up under `root` the directories the image creates. +fn image_filesystem(root: &std::path::Path) { let dockerfile = deployment::emit_dockerfile(); let created = dockerfile .lines() @@ -818,53 +433,6 @@ fn stock_pod_filesystem(root: &std::path::Path) { { std::fs::create_dir_all(inside(root, dir)).expect("image directory"); } - for mount in chart_volume_mounts() { - let at = inside(root, &mount.path); - if mount.sub_path.is_some() { - std::fs::create_dir_all(at.parent().expect("a file mount has a parent")) - .expect("mount parent"); - std::fs::write(&at, "").expect("file mount"); - } else { - if at.exists() { - std::fs::remove_dir_all(&at).expect("clear what the mount hides"); - } - std::fs::create_dir_all(&at).expect("directory mount"); - } - } -} - -/// With stock values the app must find the directory `transforms.dir` names. -/// -/// The ConfigMap mounted over the whole of /etc/dfe-transform-vector hid the -/// image's transforms/ beside config.yaml, and assembly refuses a missing -/// transforms dir, so a stock install crash-looped before Vector started. -#[test] -fn the_stock_chart_leaves_the_transforms_dir_its_config_names() { - let values: serde_yaml_ng::Value = serde_yaml_ng::from_str( - &std::fs::read_to_string( - std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/values.yaml"), - ) - .expect("read chart values"), - ) - .expect("parse chart values"); - let dir = values["config"]["transforms"]["dir"] - .as_str() - .expect("the chart names a transforms dir") - .to_string(); - - let pod = tempfile::tempdir().expect("pod root"); - stock_pod_filesystem(pod.path()); - - assert!( - inside(pod.path(), &dir).is_dir(), - "{dir} is not in the pod: the image creates it, but a volume mount hides it. \ - Mounts: {:?}", - chart_volume_mounts() - ); - assert!( - inside(pod.path(), "/etc/dfe-transform-vector/config.yaml").is_file(), - "the config file the entrypoint reads is not mounted" - ); } #[test]