diff --git a/.gitignore b/.gitignore index 2a14aea..855c296 100644 --- a/.gitignore +++ b/.gitignore @@ -26,6 +26,8 @@ coverage/ .tmp/ # generated by hyperi-ci's osv-scanner stage from quality.ignore (mirrors deny.toml) osv-scanner.toml +# written by `dfe-loader --emit-helm`; the release assembles the chart from the contract +/chart/ # Rust / cargo build artefacts /target diff --git a/.hyperi-ci.yaml b/.hyperi-ci.yaml index 56d6d65..772eff4 100644 --- a/.hyperi-ci.yaml +++ b/.hyperi-ci.yaml @@ -56,7 +56,6 @@ test: # Building build: enabled: true - type: app # Build binaries (not just crate) strategies: - native rust: @@ -64,14 +63,18 @@ build: - x86_64-unknown-linux-gnu - aarch64-unknown-linux-gnu -# 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 platforms: - linux/amd64 - linux/arm64 - helm: - enabled: true diff --git a/Cargo.lock b/Cargo.lock index 6b2328f..7a5c998 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4373,9 +4373,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 = [ "aes-gcm", "backon", diff --git a/Cargo.toml b/Cargo.toml index 9d83fd1..338f8f6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,7 @@ tokio-util = { version = ">=0.7.19, <0.8", features = ["rt"] } # Shared library with transport abstraction. The floor is the first release whose # contract carries no vendor defaults, so the app names its own. -scalo = { version = "2.14.1", features = ["config-schema", "transport-kafka", "transport-grpc", "dlq-kafka", "config", "config-reload", "deployment", "version-check", "scaling", "cli-service", "top", "logger", "metrics", "service-metrics", "expression", "memory", "worker-batch", "governor", "secrets", "sink-stack"] } +scalo = { version = "2.14.3", features = ["config-schema", "transport-kafka", "transport-grpc", "dlq-kafka", "config", "config-reload", "deployment", "version-check", "scaling", "cli-service", "top", "logger", "metrics", "service-metrics", "expression", "memory", "worker-batch", "governor", "secrets", "sink-stack"] } # Steps scalo's jittered retry schedule (sink_stack backoff). Range MUST track # scalo's: BackoffBuilder only applies to the ExponentialBuilder scalo returns. @@ -60,7 +60,6 @@ csv = ">=1.4.0, <2" # CSV remap file parsing (ecs-mapper format compatible) # 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" # Error handling thiserror = ">=2.0.21, <3" @@ -140,6 +139,9 @@ time = ">=0.3.55, <0.4" tokio-test = ">=0.4.6, <0.5" tempfile = ">=3.27.0, <4" +# The repo's own .env for live-service tests and benches. The binary's .env is read by scalo's cascade. +dotenvy = ">=0.15.7, <0.16" + [[bench]] name = "json_parsing" harness = false diff --git a/Dockerfile b/Dockerfile index b03bbd0..b963fdd 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-loader::deployment::contract() # Regenerate by calling scalo::deployment::generate_dockerfile() on that contract. diff --git a/README.md b/README.md index c1f61e1..1f959f2 100644 --- a/README.md +++ b/README.md @@ -108,7 +108,6 @@ NOT a transform stage, NOT a schema manager. Its only outbound topic is the DLQ. | `src/routing/`, `src/transform/`, `src/enrich/` | Routing, coercion and capture, enrichment | | `src/buffer/`, `src/clickhouse/` | Per-table buffers; query client, schema cache, inserter | | `src/clickhouse_ext/` | Dynamic insert: runtime type parser, RowBinary encoder | -| `chart/` | Helm chart, generated from the deployment contract and committed | | `tests/` | `smoke.rs`, `integration/`, `e2e/` -- see `tests/TESTING.md` | ### Commands that prove a change @@ -127,22 +126,21 @@ Green lies three ways: `skip_if_no_clickhouse!()` and `skip_if_no_kafka!()` retu | Don't | Do | Why | |-------|----|-----| -| Bump scalo and commit without regenerating the chart | `dfe-loader --emit-helm chart` | `committed_chart_matches_the_generator` (`tests/integration/deployment.rs:250`) compares `chart/` to the generator file by file. A failure is the guard working -- dfe-fetcher shipped a chart missing `keda-triggerauth.yaml` that its ScaledObject referenced, and never scaled (dfe-fetcher#71) | -| Gate a test behind a cargo feature without adding it to `.hyperi-ci.yaml` | Add `default,` to the feature-set list | `default = []` (`Cargo.toml:327`), so it compiles out of every CI run and CI still reports green. `helm_contract.rs:106` asserts `transport-memory` and `testcontainers` stay listed | +| Bump scalo and leave `release.helm.library` behind | Move `release.helm.library` in `.hyperi-ci.yaml` to the same scalo version | The release assembles the chart from the emitted contract on that scalo-service version, and a scalo-service release ships the schema for only the contract version its scalo release writes | +| Gate a test behind a cargo feature without adding it to `.hyperi-ci.yaml` | Add `default,` to the feature-set list | `default = []` (`Cargo.toml:304`), so it compiles out of every CI run and CI still reports green. `helm_contract.rs:64` asserts `transport-memory` and `testcontainers` stay listed | | Inject a nested config key as `DFE_LOADER_SECTION_FIELD` | `DFE_LOADER__SECTION__FIELD` | figment strips exactly `DFE_LOADER_`, so it arrived as `_kafka.sasl.username`, matched no field and was dropped silently. Pods ran with no SASL and an empty ClickHouse password (`tests/integration/config_reachability.rs`) | | Set `clickhouse.protocol: native` | `http`, on an 8123-family port | The pinned fork has no TCP row fetch, so schema queries stall silently and messages back up pending schema (#115). `validate()` rejects it by name | | Name `clickhouse.tls.ca_cert_file`, `cert_file`, `key_file` or `skip_verify` | Set `tls.enabled`, mount the CA into the trust store | Only `enabled` reaches a client. The rest parsed and did nothing, so `validate()` now fails naming them | -| Widen the `cel` range past scalo's | Keep it on `>=0.13, <0.14` | `cel::Program` crosses the scalo boundary. Wider resolves two semver-incompatible `cel` crates and `Program` stops being the same type (`Cargo.toml:41-45`) | -| Pin the `clickhouse` fork by branch | Pin by `rev` or tag | The `hyperi-port/*` chain is force-pushed, so a branch pin rots with no warning (`Cargo.toml:313-321`) | +| Widen the `cel` range past scalo's | Keep it on `>=0.14.5, <0.15` | `cel::Program` crosses the scalo boundary. Wider resolves two semver-incompatible `cel` crates and `Program` stops being the same type (`Cargo.toml:46-51`) | +| Pin the `clickhouse` fork by branch | Pin by `rev` or tag | The `hyperi-port/*` chain is force-pushed, so a branch pin rots with no warning (`Cargo.toml:289-300`) | | Mechanically sync dfe-engine's loader validation to `Config::validate()` | Read both, keep the divergence | dfe-engine scopes the broker check to the Kafka transport and adds a `grpc.listen` check this side lacks. A blind sync rejects valid gRPC-only configs at author time | ### Where this sits Inbound, declared in `dfe-infra/suite.yaml`: -- **scalo-rs -> dfe-loader** (`cargo-dep`) -- `Cargo.toml:35` takes `scalo` by range for transports, config cascade, CLI, metrics, deployment contract and DLQ. A scalo release reaches this repo here. -- **scalo-rs -> dfe-loader** (`generated-file`, lockstep) -- `Dockerfile` is emitted by `scalo::deployment::generate_dockerfile()` at schema version 3. Regenerate and commit the diff. - +- **scalo-rs -> dfe-loader** (`cargo-dep`) -- `Cargo.toml:36` takes `scalo` by range for transports, config cascade, CLI, metrics, deployment contract and DLQ. A scalo release reaches this repo here. +- **scalo-rs -> dfe-loader** (`generated-file`, lockstep) -- `Dockerfile` is emitted by `scalo::deployment::generate_dockerfile()` at schema version 4. Regenerate and commit the diff. Outbound, so what a change here can break: - **dfe-loader -> dfe-infra** (`image-pin`, lockstep) -- `dfe-infra/helm/charts/dfe-loader/Chart.yaml:6` pins this image as tag plus digest. diff --git a/benches/insert_bakeoff.rs b/benches/insert_bakeoff.rs index 5eae6bb..3ae99e3 100644 --- a/benches/insert_bakeoff.rs +++ b/benches/insert_bakeoff.rs @@ -115,7 +115,8 @@ struct BenchEnv { impl BenchEnv { fn from_env() -> Option { - dotenvy::dotenv().ok(); + // The repo's own .env only: dotenvy::dotenv() would load the first .env in any parent. + let _ = dotenvy::from_path(std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(".env")); let host = env::var("CLICKHOUSE_HOST").ok()?; if host.is_empty() { return None; diff --git a/chart/Chart.yaml b/chart/Chart.yaml deleted file mode 100644 index e27c103..0000000 --- a/chart/Chart.yaml +++ /dev/null @@ -1,9 +0,0 @@ -apiVersion: v2 -name: dfe-loader -description: High-performance Kafka to ClickHouse data loader -type: application -version: 0.1.0 -appVersion: "1.0.0" - -keywords: - - dfe-loader diff --git a/chart/templates/NOTES.txt b/chart/templates/NOTES.txt deleted file mode 100644 index edf15f4..0000000 --- a/chart/templates/NOTES.txt +++ /dev/null @@ -1,16 +0,0 @@ -dfe-loader has been deployed. - -1. Get the metrics/health endpoint: - kubectl port-forward svc/{{ include "dfe-loader.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-loader.fullname" . }} - kubectl get hpa -{{- end }} - -3. View logs: - kubectl logs -l app.kubernetes.io/name={{ include "dfe-loader.name" . }} -f diff --git a/chart/templates/_helpers.tpl b/chart/templates/_helpers.tpl deleted file mode 100644 index 4f10290..0000000 --- a/chart/templates/_helpers.tpl +++ /dev/null @@ -1,83 +0,0 @@ -{{/* -Expand the name of the chart. -*/}} -{{- define "dfe-loader.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-loader.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-loader.chart" -}} -{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} -{{- end }} - -{{/* -Common labels. -*/}} -{{- define "dfe-loader.labels" -}} -helm.sh/chart: {{ include "dfe-loader.chart" . }} -{{ include "dfe-loader.selectorLabels" . }} -{{- if .Chart.AppVersion }} -app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} -{{- end }} -app.kubernetes.io/managed-by: {{ .Release.Service }} -{{- end }} - -{{/* -Selector labels. -*/}} -{{- define "dfe-loader.selectorLabels" -}} -app.kubernetes.io/name: {{ include "dfe-loader.name" . }} -app.kubernetes.io/instance: {{ .Release.Name }} -{{- end }} - -{{/* -Service account name. -*/}} -{{- define "dfe-loader.serviceAccountName" -}} -{{- if .Values.serviceAccount.create }} -{{- default (include "dfe-loader.fullname" .) .Values.serviceAccount.name }} -{{- else }} -{{- default "default" .Values.serviceAccount.name }} -{{- end }} -{{- end }} - -{{/* -kafka secret name -- use existing or generate from fullname. -*/}} -{{- define "dfe-loader.kafkaSecretName" -}} -{{- if .Values.kafka.existingSecret }} -{{- .Values.kafka.existingSecret }} -{{- else }} -{{- printf "%s-kafka" (include "dfe-loader.fullname" .) }} -{{- end }} -{{- end }} - -{{/* -clickhouse secret name -- use existing or generate from fullname. -*/}} -{{- define "dfe-loader.clickhouseSecretName" -}} -{{- if .Values.clickhouse.existingSecret }} -{{- .Values.clickhouse.existingSecret }} -{{- else }} -{{- printf "%s-clickhouse" (include "dfe-loader.fullname" .) }} -{{- end }} -{{- end }} diff --git a/chart/templates/configmap.yaml b/chart/templates/configmap.yaml deleted file mode 100644 index 69973c6..0000000 --- a/chart/templates/configmap.yaml +++ /dev/null @@ -1,9 +0,0 @@ -apiVersion: v1 -kind: ConfigMap -metadata: - name: {{ include "dfe-loader.fullname" . }}-config - labels: - {{- include "dfe-loader.labels" . | nindent 4 }} -data: - loader.yaml: | - {{- toYaml .Values.config | nindent 4 }} diff --git a/chart/templates/deployment.yaml b/chart/templates/deployment.yaml deleted file mode 100644 index ab387e7..0000000 --- a/chart/templates/deployment.yaml +++ /dev/null @@ -1,161 +0,0 @@ -apiVersion: apps/v1 -kind: Deployment -metadata: - name: {{ include "dfe-loader.fullname" . }} - labels: - {{- include "dfe-loader.labels" . | nindent 4 }} -spec: - {{- if not (or (and .Values.keda.enabled .Values.keda.cpu.enabled) (and .Values.autoscaling.enabled (not .Values.keda.enabled))) }} - replicas: {{ .Values.replicaCount }} - {{- end }} - selector: - matchLabels: - {{- include "dfe-loader.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-loader.labels" . | nindent 8 }} - {{- with .Values.podLabels }} - {{- toYaml . | nindent 8 }} - {{- end }} - spec: - {{- with .Values.imagePullSecrets }} - imagePullSecrets: - {{- toYaml . | nindent 8 }} - {{- end }} - serviceAccountName: {{ include "dfe-loader.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/loader.yaml" - ports: - - name: metrics - containerPort: {{ .Values.service.port }} - protocol: TCP - {{- if eq (toString (.Values.config).transport) "grpc" }} - - name: push - containerPort: 6000 - protocol: TCP - {{- end }} - env: - - name: OTEL_SERVICE_NAME - value: "dfe-loader" - - 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 }} - # kafka credentials via Secret (figment env cascade overrides file config) - - name: DFE_LOADER__KAFKA__SASL__USERNAME - valueFrom: - secretKeyRef: - name: {{ include "dfe-loader.kafkaSecretName" . }} - key: {{ .Values.kafka.secretKeys.username }} - - name: DFE_LOADER__KAFKA__SASL__PASSWORD - valueFrom: - secretKeyRef: - name: {{ include "dfe-loader.kafkaSecretName" . }} - key: {{ .Values.kafka.secretKeys.password }} - # clickhouse credentials via Secret (figment env cascade overrides file config) - - name: DFE_LOADER__CLICKHOUSE__PASSWORD - valueFrom: - secretKeyRef: - name: {{ include "dfe-loader.clickhouseSecretName" . }} - key: {{ .Values.clickhouse.secretKeys.password }} - 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: - - name: config - mountPath: /etc/dfe - 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-loader.fullname" . }}-config - - 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 30c0d86..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-loader.fullname" . }} - labels: - {{- include "dfe-loader.labels" . | nindent 4 }} -spec: - scaleTargetRef: - apiVersion: apps/v1 - kind: Deployment - name: {{ include "dfe-loader.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 112479f..0000000 --- a/chart/templates/keda-scaledobject.yaml +++ /dev/null @@ -1,23 +0,0 @@ -{{- if and .Values.keda.enabled .Values.keda.cpu.enabled }} -apiVersion: keda.sh/v1alpha1 -kind: ScaledObject -metadata: - name: {{ include "dfe-loader.fullname" . }} - labels: - {{- include "dfe-loader.labels" . | nindent 4 }} -spec: - scaleTargetRef: - name: {{ include "dfe-loader.fullname" . }} - minReplicaCount: {{ .Values.keda.minReplicaCount }} - maxReplicaCount: {{ .Values.keda.maxReplicaCount }} - pollingInterval: {{ .Values.keda.pollingInterval }} - cooldownPeriod: {{ .Values.keda.cooldownPeriod }} - triggers: - {{- if .Values.keda.cpu.enabled }} - # CPU utilisation (only 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 6688401..0000000 --- a/chart/templates/keda-triggerauth.yaml +++ /dev/null @@ -1 +0,0 @@ -# No Kafka lag trigger -- KEDA TriggerAuthentication not generated diff --git a/chart/templates/secret.yaml b/chart/templates/secret.yaml deleted file mode 100644 index 1ae6466..0000000 --- a/chart/templates/secret.yaml +++ /dev/null @@ -1,24 +0,0 @@ -{{- if not .Values.kafka.existingSecret }} -apiVersion: v1 -kind: Secret -metadata: - name: {{ include "dfe-loader.kafkaSecretName" . }} - labels: - {{- include "dfe-loader.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 }} ---- -{{- if not .Values.clickhouse.existingSecret }} -apiVersion: v1 -kind: Secret -metadata: - name: {{ include "dfe-loader.clickhouseSecretName" . }} - labels: - {{- include "dfe-loader.labels" . | nindent 4 }} -type: Opaque -data: - {{ .Values.clickhouse.secretKeys.password }}: {{ .Values.clickhouse.password | b64enc | quote }} -{{- end }} diff --git a/chart/templates/service.yaml b/chart/templates/service.yaml deleted file mode 100644 index e8b5002..0000000 --- a/chart/templates/service.yaml +++ /dev/null @@ -1,21 +0,0 @@ -apiVersion: v1 -kind: Service -metadata: - name: {{ include "dfe-loader.fullname" . }} - labels: - {{- include "dfe-loader.labels" . | nindent 4 }} -spec: - type: {{ .Values.service.type }} - ports: - - port: {{ .Values.service.port }} - targetPort: metrics - protocol: TCP - name: metrics - {{- if eq (toString (.Values.config).transport) "grpc" }} - - port: 6000 - targetPort: 6000 - protocol: TCP - name: push - {{- end }} - selector: - {{- include "dfe-loader.selectorLabels" . | nindent 4 }} diff --git a/chart/templates/serviceaccount.yaml b/chart/templates/serviceaccount.yaml deleted file mode 100644 index b16628d..0000000 --- a/chart/templates/serviceaccount.yaml +++ /dev/null @@ -1,13 +0,0 @@ -{{- if .Values.serviceAccount.create -}} -apiVersion: v1 -kind: ServiceAccount -metadata: - name: {{ include "dfe-loader.serviceAccountName" . }} - labels: - {{- include "dfe-loader.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 759158f..0000000 --- a/chart/values.yaml +++ /dev/null @@ -1,338 +0,0 @@ -# dfe-loader 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-loader - # -- 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 and health endpoint service -service: - type: ClusterIP - port: 9090 - -# -- Application configuration (mounted as /etc/dfe/loader.yaml) -config: - transport: '' - kafka: - brokers: - - localhost:9092 - group: dfe-loader - topics: [] - topic_regex: null - client_id: dfe-loader - sasl: null - tls: null - allow_insecure_transport: false - librdkafka_overrides: - statistics.interval.ms: '5000' - acknowledgements: - enabled: true - grpc: - listen: 0.0.0.0:6000 - recv_buffer_size: 10000 - recv_timeout_ms: 100 - max_message_size: 16777216 - compression: false - default_topic: main_land - acknowledgements: - enabled: true - max_hold_ms: 13500 - clickhouse: - hosts: - - localhost:8123 - database: dfe - username: default - password: '' - protocol: http - insert_format: rowbinary - tables: [] - tls: null - payload: - pipeline_mode: json_primary - metrics: - enabled: true - address: 0.0.0.0:9090 - logging: - level: info - format: json - schema: - cache_ttl_secs: 300 - refresh_on_error: true - pre_warm_retry_secs: 60 - pending_max_per_table: 1000 - pending_max_total: 10000 - pending_max_age_secs: 30 - geoip: - enabled: false - provider: db_ip_lite - city_db_path: null - asn_db_path: null - auto_download: - enabled: true - data_dir: /var/lib/dfe/geoip - maxmind_account_id: null - maxmind_license_key: null - ipinfo_token: null - max_age_days: 30 - cache_capacity: 100000 - computed_columns: - columns: {} - overrides: {} - column_directives: - global: {} - tables: {} - hot_reload: - enabled: false - poll_interval_secs: 5 - debounce_ms: 500 - keda: - enabled: true - 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 - scaling: - enabled: true - memory_gate_threshold: 0.8 - weight_kafka_lag: 0.35 - weight_buffer_depth: 0.25 - weight_insert_latency: 0.15 - weight_memory: 0.15 - weight_errors: 0.1 - saturation_kafka_lag: 100000.0 - saturation_buffer_depth: 10000.0 - saturation_insert_latency: 5.0 - saturation_errors: 100.0 - batch_processing: - max_chunk_size: 10000 - routing_field: null - pre_route_filters: [] - parse_error_action: dlq - known_fields: - - _table - - _timestamp - - _source - - host - - source_type - - event_type - routing: - rules: [] - db_fields: [] - table_fields: - - _source - default_db: dfe - default_table: main - org_id_field: org_id - org_routes: [] - source_to_table: {} - mapping_file: null - topic_suffixes: - - _land - - _load - compat_v2_source: false - dlq: - enabled: true - mode: cascade - topic: dfe_loader_dlq - topic_suffix: .dlq - file_enabled: true - file_path: /var/spool/dfe/dlq - kafka_enabled: true - buffer: - flush_bytes: 1048576 - flush_rows: 20000 - flush_age_secs: 5 - memory: - limit_bytes: 0 - pressure_threshold: 0.8 - timestamp_dq: - enabled: true - max_future_seconds: 600 - max_past_seconds: 0 - invalid_action: replace_with_now - correct_known_bad: true - field_sanitization: - strip_at_prefix: true - handle_numeric_prefix: true - numeric_prefix: col_ - collapse_underscores: true - trim_underscores: true - collision_strategy: last_wins - metadata: - enabled: true - inject_timestamp_load: true - extract_timestamp_collector: true - collector_timestamp_path: tags.collector.timestamp - tags_fields: - - tags - - _tags - - meta - - metadata.tags - tags_output: _tags - drop_tags: false - capture_mode: full - table_capture_modes: {} - capture_json: true - json_output: _json - capture_raw: true - raw_source_fields: - - logoriginal - raw_output: _raw - capture_source: true - source_fields: - - _source - source_output: _source - disable_json_tables: [] - disable_raw_tables: [] - remove_routing_fields: true - coercion: - type_mappings: {} - unknown_type_fallback: String - null_handling: default - null_strings: - - 'null' - - 'NULL' - - 'Null' - - None - - nil - - undefined - - \N - - - - NA - - N/A - - n/a - - NaN - default_timezone: UTC - timezone_fields: - - tags_collector_timezone - array_to_json: true - strict: false - field_mapping: - enabled: false - default_action: rename - builtin: none - files: [] - overrides: {} - enrichment: - ip_fields: - - src_ip - - client_ip - - ip - - source_ip - reputation: - enabled: false - cache_capacity: 100000 - blocklist_files: [] - risk_scoring: - enabled: false - preset: global - -# -- kafka credentials -kafka: - existingSecret: "" - secretKeys: - username: kafka-username - password: kafka-password - username: "" - password: "" - -# -- clickhouse credentials -clickhouse: - existingSecret: "" - secretKeys: - password: clickhouse-password - password: "" - -# -- KEDA autoscaling (requires KEDA operator installed) -keda: - enabled: true - minReplicaCount: 1 - maxReplicaCount: 10 - pollingInterval: 15 - cooldownPeriod: 300 - 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 84d61d8..34ab147 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -4,7 +4,7 @@ # Configuration cascade (highest to lowest priority): # 1. CLI args (--config, --host, etc.) # 2. Environment variables (DFE_LOADER_KAFKA_BROKERS, etc.) -# 3. .env file +# 3. .env file in the working directory (never a parent's) # 4. settings.{env}.yaml # 5. settings.yaml # 6. defaults.yaml diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index a7d59c8..5f5d610 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -17,7 +17,7 @@ layers win. Every setting has a safe default, so an empty config boots. flowchart TB CLI["1. CLI args (--clickhouse.protocol=http)"] ENV["2. ENV (DFE_LOADER_CLICKHOUSE__PROTOCOL=http)"] - DOTENV["3. .env (gitignored)"] + DOTENV["3. ./.env in the working directory (gitignored)"] ENVYAML["4. settings.{env}.yaml"] YAML["5. settings.yaml"] DEF["6. defaults.yaml"] @@ -28,6 +28,8 @@ flowchart TB ENV names are auto-derived: `clickhouse.protocol` -> `DFE_LOADER_CLICKHOUSE__PROTOCOL` (a double underscore for each level of nesting). +Only the `.env` in the working directory is read. A `.env` in a parent directory belongs to another project and is never loaded. + The scaling gate (`scaling.enabled`, `scaling.memory_gate_threshold`) is read by scalo's own cascade, which discovers `defaults.yaml` / `settings.yaml` but never the file named by `--config`, so set it with `DFE_LOADER_SCALING__ENABLED` and diff --git a/docs/config-schema.json b/docs/config-schema.json index 8905349..1e81499 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 for dfe-loader.\n\n## Hot-Reload Behaviour\n\n**Hot-reloaded (takes effect on next batch):**\n- `routing.*` — routing rules, table mapping, org routing\n- `timestamp_dq.*` — timestamp validation thresholds\n- `metadata.*` — common header injection, tags, _raw handling\n- `field_sanitization.*` — field name sanitisation rules\n- `buffer.flush_rows` / `buffer.flush_bytes` / `buffer.flush_age_secs`\n- `coercion.*` — type coercion config\n- `enrichment.ip_fields` — which fields to enrich\n- `field_mapping.*` — field mapping overrides\n\n**Requires pod restart (connections/state established at startup):**\n- `transport` — transport type (kafka/grpc) bound at startup\n- `kafka.*` — Kafka consumer created at startup\n- `grpc.*` — gRPC server binds at startup\n- `clickhouse.*` — HTTP client + `clickhouse::Client` created at startup\n- `payload.pipeline_mode` — pipeline path chosen at startup\n- `metrics.*` — HTTP metrics server binds at startup\n- `logging.*` — tracing subscriber installed at startup\n- `scaling.*` / `keda.*` — scaling pressure built at startup\n- `hot_reload.*` — watcher config set at startup\n- `schema.*` — schema cache created at startup\n- `geoip.*` — MMDB readers opened at startup\n- `computed_columns.*` — computed column cache built at startup\n- `column_directives.*` — column directive cache built at startup", + "description": "Main configuration for dfe-loader.\n\n## Hot-Reload Behaviour\n\n**Hot-reloaded (takes effect on next batch):**\n- `routing.*` — routing rules, table mapping, org routing\n- `timestamp_dq.*` — timestamp validation thresholds\n- `metadata.*` — common header injection, tags, _raw handling\n- `field_sanitization.*` — field name sanitisation rules\n- `buffer.flush_rows` / `buffer.flush_bytes` / `buffer.flush_age_secs`\n- `coercion.*` — type coercion config\n- `enrichment.ip_fields` — which fields to enrich\n- `field_mapping.*` — field mapping overrides\n\n**Requires pod restart (connections/state established at startup):**\n- `transport` — transport type (kafka/grpc) bound at startup\n- `kafka.*` — Kafka consumer created at startup\n- `grpc.*` — gRPC server binds at startup\n- `clickhouse.*` — HTTP client + `clickhouse::Client` created at startup\n- `payload.pipeline_mode` — pipeline path chosen at startup\n- `metrics.*` — HTTP metrics server binds at startup\n- `logging.*` — tracing subscriber installed at startup\n- `scaling.*` — scaling pressure built at startup\n- `hot_reload.*` — watcher config set at startup\n- `schema.*` — schema cache created at startup\n- `geoip.*` — MMDB readers opened at startup\n- `computed_columns.*` — computed column cache built at startup\n- `column_directives.*` — column directive cache built at startup", "type": "object", "properties": { "transport": { @@ -142,21 +142,6 @@ "debounce_ms": 500 } }, - "keda": { - "description": "KEDA autoscaling config. **Restart required.**", - "$ref": "#/$defs/KedaConfig", - "default": { - "enabled": true, - "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 - } - }, "scaling": { "description": "Scaling pressure config. **Restart required.**", "$ref": "#/$defs/ScalingConfig", @@ -463,7 +448,6 @@ "password": { "type": "string", "x-scalo-secret": true, - "x-dfe-secret": true, "writeOnly": true, "default": "***REDACTED***" }, @@ -490,7 +474,6 @@ "null" ], "x-scalo-secret": true, - "x-dfe-secret": true, "writeOnly": true, "default": null }, @@ -533,7 +516,6 @@ "null" ], "x-scalo-secret": true, - "x-dfe-secret": true, "writeOnly": true, "default": null }, @@ -544,7 +526,6 @@ "null" ], "x-scalo-secret": true, - "x-dfe-secret": true, "writeOnly": true, "default": null }, @@ -685,7 +666,6 @@ "password": { "type": "string", "x-scalo-secret": true, - "x-dfe-secret": true, "writeOnly": true, "default": "***REDACTED***" }, @@ -937,7 +917,6 @@ "null" ], "x-scalo-secret": true, - "x-dfe-secret": true, "writeOnly": true, "default": null }, @@ -1060,68 +1039,6 @@ } } }, - "KedaConfig": { - "description": "KEDA autoscaling thresholds (deployment-level config).\n\nCHART-GENERATION ONLY. These defaults feed the deployment contract, and the\nchart contract test validates chart/values.yaml against them. The loader\nprocess itself reads NO field of this section: KEDA scales the deployment\nfrom the `ScaledObject` the chart renders, so the value that moves a\nthreshold is the chart's `keda.*` (via `--set` or a values overlay), and a\n`keda:` block in the pod's own config file or a `DFE_LOADER__KEDA__*` env var\nchanges nothing. The runtime scaling signal those thresholds read is\n[`ScalingConfig`], which the pod does load.", - "type": "object", - "properties": { - "enabled": { - "type": "boolean", - "default": true - }, - "min_replicas": { - "type": "integer", - "format": "uint32", - "minimum": 0, - "default": 1 - }, - "max_replicas": { - "type": "integer", - "format": "uint32", - "minimum": 0, - "default": 10 - }, - "polling_interval": { - "description": "Seconds between KEDA polling the scaler", - "type": "integer", - "format": "uint32", - "minimum": 0, - "default": 15 - }, - "cooldown_period": { - "description": "Seconds before scale-down after load drops", - "type": "integer", - "format": "uint32", - "minimum": 0, - "default": 300 - }, - "kafka_lag_threshold": { - "description": "Scale when consumer group lag exceeds this per partition", - "type": "integer", - "format": "uint64", - "minimum": 0, - "default": 1000 - }, - "activation_lag_threshold": { - "description": "Wake from zero replicas when lag exceeds this", - "type": "integer", - "format": "uint64", - "minimum": 0, - "default": 0 - }, - "cpu_enabled": { - "description": "Enable CPU-based scaling trigger", - "type": "boolean", - "default": true - }, - "cpu_threshold": { - "description": "CPU utilisation percentage threshold", - "type": "integer", - "format": "uint32", - "minimum": 0, - "default": 80 - } - } - }, "ScalingConfig": { "description": "Scaling pressure configuration for KEDA autoscaling.\n\nProduces a 0-100 composite metric (`loader_scaling_pressure`) based on\nweighted application signals with two hard gates (circuit breaker, memory).\n\nOverride weights at runtime via env vars:\n `DFE_LOADER__SCALING__WEIGHT_KAFKA_LAG=0.45`\n `DFE_LOADER__SCALING__SATURATION_BUFFER_DEPTH=20000`", "type": "object", diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index 8eeaae0..3a4c019 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -23,7 +23,7 @@ description: |- - `payload.pipeline_mode` — pipeline path chosen at startup - `metrics.*` — HTTP metrics server binds at startup - `logging.*` — tracing subscriber installed at startup - - `scaling.*` / `keda.*` — scaling pressure built at startup + - `scaling.*` — scaling pressure built at startup - `hot_reload.*` — watcher config set at startup - `schema.*` — schema cache created at startup - `geoip.*` — MMDB readers opened at startup @@ -140,19 +140,6 @@ properties: enabled: false poll_interval_secs: 5 debounce_ms: 500 - keda: - description: KEDA autoscaling config. **Restart required.** - $ref: '#/$defs/KedaConfig' - default: - enabled: true - 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 scaling: description: Scaling pressure config. **Restart required.** $ref: '#/$defs/ScalingConfig' @@ -411,7 +398,6 @@ $defs: password: type: string x-scalo-secret: true - x-dfe-secret: true writeOnly: true default: '***REDACTED***' oauth_token_endpoint: @@ -432,7 +418,6 @@ $defs: - string - 'null' x-scalo-secret: true - x-dfe-secret: true writeOnly: true default: null oauth_scope: @@ -465,7 +450,6 @@ $defs: - string - 'null' x-scalo-secret: true - x-dfe-secret: true writeOnly: true default: null aws_session_token: @@ -474,7 +458,6 @@ $defs: - string - 'null' x-scalo-secret: true - x-dfe-secret: true writeOnly: true default: null aws_profile: @@ -613,7 +596,6 @@ $defs: password: type: string x-scalo-secret: true - x-dfe-secret: true writeOnly: true default: '***REDACTED***' protocol: @@ -835,7 +817,6 @@ $defs: - string - 'null' x-scalo-secret: true - x-dfe-secret: true writeOnly: true default: null max_age_days: @@ -951,67 +932,6 @@ $defs: format: uint64 minimum: 0 default: 500 - KedaConfig: - description: |- - KEDA autoscaling thresholds (deployment-level config). - - CHART-GENERATION ONLY. These defaults feed the deployment contract, and the - chart contract test validates chart/values.yaml against them. The loader - process itself reads NO field of this section: KEDA scales the deployment - from the `ScaledObject` the chart renders, so the value that moves a - threshold is the chart's `keda.*` (via `--set` or a values overlay), and a - `keda:` block in the pod's own config file or a `DFE_LOADER__KEDA__*` env var - changes nothing. The runtime scaling signal those thresholds read is - [`ScalingConfig`], which the pod does load. - type: object - properties: - enabled: - type: boolean - default: true - min_replicas: - type: integer - format: uint32 - minimum: 0 - default: 1 - max_replicas: - type: integer - format: uint32 - minimum: 0 - default: 10 - polling_interval: - description: Seconds between KEDA polling the scaler - type: integer - format: uint32 - minimum: 0 - default: 15 - cooldown_period: - description: Seconds before scale-down after load drops - type: integer - format: uint32 - minimum: 0 - default: 300 - kafka_lag_threshold: - description: Scale when consumer group lag exceeds this per partition - type: integer - format: uint64 - minimum: 0 - default: 1000 - activation_lag_threshold: - description: Wake from zero replicas when lag exceeds this - type: integer - format: uint64 - minimum: 0 - default: 0 - cpu_enabled: - description: Enable CPU-based scaling trigger - type: boolean - default: true - cpu_threshold: - description: CPU utilisation percentage threshold - type: integer - format: uint32 - minimum: 0 - default: 80 ScalingConfig: description: |- Scaling pressure configuration for KEDA autoscaling. diff --git a/docs/deployment/PUBLISHING.md b/docs/deployment/PUBLISHING.md index 1a98d27..35e09c6 100644 --- a/docs/deployment/PUBLISHING.md +++ b/docs/deployment/PUBLISHING.md @@ -15,9 +15,7 @@ pushed by CI on every release. Downstream projects (dfe-docker, dfe-operator) reference an immutable OCI tag instead of building from source or downloading loose binaries. -Both artefacts go to GHCR. The image is `ghcr.io/hyperi-io/dfe-loader` and the -chart is pushed to `oci://ghcr.io/hyperi-io/helm-charts`. CI authenticates with -`GITHUB_TOKEN`, so no registry secret is needed. +Both artefacts go to GHCR. The image is `ghcr.io/hyperi-io/dfe-loader` and the chart is pushed to `oci://ghcr.io/hyperi-io/charts`. CI authenticates with `GITHUB_TOKEN`, so no registry secret is needed. ```mermaid flowchart TB @@ -25,38 +23,36 @@ flowchart TB BIN["CI cross-build
linux/amd64 + linux/arm64 binaries"] DOCK["Build Dockerfile
wraps pre-built binary"] PUSH["Push image
version tags"] - HELM["Package + push Helm chart
OCI artifact"] + EMIT["dfe-loader generate-artefacts
emits the deployment contract"] + HELM["Assemble thin chart on scalo-service
package + push OCI artifact"] REG[("GHCR")] REL --> BIN --> DOCK --> PUSH --> REG - BIN --> HELM --> REG + BIN --> EMIT --> HELM --> REG ``` ## How it works -`.github/workflows/ci.yml` calls hyperi-ci's reusable `rust-ci.yml`. On a -release it builds the `Dockerfile` for `linux/amd64` and `linux/arm64`, pushes -the image with version tags, then packages and pushes the Helm chart from -`chart/`. The `release:` block in `.hyperi-ci.yaml` turns both on: +`.github/workflows/ci.yml` calls hyperi-ci's reusable `rust-ci.yml`. On a release it builds the `Dockerfile` for `linux/amd64` and `linux/arm64` and pushes the image with version tags. It also runs the built binary's `generate-artefacts` to emit the deployment contract, assembles a thin chart from it on the scalo-service library chart, and pushes that. No chart is committed to this repo. The `release:` block in `.hyperi-ci.yaml` turns both on: ```yaml release: enabled: true + helm: + enabled: true + contract: emit + library: "2.14.3" container: enabled: true dockerfile: Dockerfile platforms: - linux/amd64 - linux/arm64 - helm: - enabled: true ``` -The `Dockerfile` is generated from the app's deployment contract by scalo and -copies in the binary CI has already cross-compiled. Do not edit it by hand. -Regenerate it with `dfe-loader --emit-dockerfile`, and the chart with -`dfe-loader --emit-helm`. `tests/integration/helm_contract.rs` fails when the -committed `Dockerfile` or `chart/values.yaml` drifts from the contract. +`library` moves with the scalo version in `Cargo.toml`: a scalo-service release ships the schema for, and 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-loader --image ghcr.io/hyperi-io/dfe-loader:@sha256: --version `. It prints the chart directory it wrote. + +The `Dockerfile` is generated from the app's deployment contract by scalo and copies in the binary CI has already cross-compiled. Do not edit it by hand. Regenerate it with `dfe-loader --emit-dockerfile`. `tests/integration/helm_contract.rs` fails when the committed `Dockerfile` drifts from the contract. ## Tags generated diff --git a/src/config/loader.rs b/src/config/loader.rs index 4a2a85f..94d43c7 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -8,7 +8,7 @@ //! 2. Explicit flat env overrides (`DFE_LOADER_KAFKA_BROKERS`, etc.) //! 3. Figment env vars, `__` nesting, either separator after the prefix //! (`DFE_LOADER__KAFKA__BROKERS` or `DFE_LOADER_KAFKA__BROKERS`) -//! 4. .env file (via dotenvy) +//! 4. `./.env` in the working directory, read by scalo's cascade at startup //! 5. Config file specified by --config or `DFE_LOADER_CONFIG` //! 6. Hard-coded defaults @@ -45,7 +45,7 @@ pub use super::pipeline::*; /// - `payload.pipeline_mode` — pipeline path chosen at startup /// - `metrics.*` — HTTP metrics server binds at startup /// - `logging.*` — tracing subscriber installed at startup -/// - `scaling.*` / `keda.*` — scaling pressure built at startup +/// - `scaling.*` — scaling pressure built at startup /// - `hot_reload.*` — watcher config set at startup /// - `schema.*` — schema cache created at startup /// - `geoip.*` — MMDB readers opened at startup @@ -53,11 +53,9 @@ pub use super::pipeline::*; /// - `column_directives.*` — column directive cache built at startup #[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)] #[serde(default)] -#[derive(Default)] pub struct Config { // --- Requires restart (connections/state established at startup) --- /// Transport backend: "kafka" (default) or "grpc". **Restart required.** - #[serde(default = "default_transport")] pub transport: String, /// Kafka consumer config. **Restart required.** pub kafka: KafkaConfig, @@ -81,8 +79,6 @@ pub struct Config { pub column_directives: crate::column_meta::ColumnDirectivesConfig, /// Hot-reload watcher config. **Restart required.** pub hot_reload: HotReloadConfig, - /// KEDA autoscaling config. **Restart required.** - pub keda: KedaConfig, /// Scaling pressure config. **Restart required.** pub scaling: ScalingConfig, /// Batch processing engine config (SIMD parse, pre-route, parallelism). **Restart required.** @@ -120,6 +116,37 @@ fn default_transport() -> String { TRANSPORT_KAFKA.to_string() } +// Written out because a derived Default gives `transport` an empty string, not the bus. +impl Default for Config { + fn default() -> Self { + Self { + transport: default_transport(), + kafka: Default::default(), + grpc: Default::default(), + clickhouse: Default::default(), + payload: Default::default(), + metrics: Default::default(), + logging: Default::default(), + schema: Default::default(), + geoip: Default::default(), + computed_columns: Default::default(), + column_directives: Default::default(), + hot_reload: Default::default(), + scaling: Default::default(), + batch_processing: Default::default(), + routing: Default::default(), + buffer: Default::default(), + memory: Default::default(), + timestamp_dq: Default::default(), + field_sanitization: Default::default(), + metadata: Default::default(), + coercion: Default::default(), + field_mapping: Default::default(), + enrichment: Default::default(), + } + } +} + impl Config { /// Whether records arrive on the Push listener rather than a broker. /// @@ -382,13 +409,11 @@ impl Config { /// 2. Explicit flat env overrides (`DFE_LOADER_KAFKA_BROKERS`, etc.) /// 3. Figment env vars with `__` nesting — `DFE_LOADER__KAFKA__SASL__USERNAME` /// (chart / deployment-contract form) or `DFE_LOADER_KAFKA__SASL__USERNAME` - /// 4. `.env` file (via dotenvy) + /// 4. `.env` file: the binary's `load_config` seeds scalo's cascade first, + /// which reads `./.env` and no parent directory's /// 5. Config file (YAML, specified by `--config` or auto-detected) /// 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(); - // 1. Start with hard-coded defaults let mut config = Config::default(); @@ -553,8 +578,9 @@ impl Config { /// (Dockerfile, Helm chart, Compose fragment). pub fn deployment_contract() -> scalo::deployment::DeploymentContract { use scalo::deployment::{ - DeploymentContract, HealthContract, ImageProfile, NativeDepsContract, OciLabels, - PortContract, SecretEnvContract, SecretGroupContract, base_image_from_cascade, + CONTRACT_SCHEMA_VERSION, DeploymentContract, HealthContract, ImageProfile, + NativeDepsContract, OciLabels, PortCondition, PortContract, ResourceList, + ResourcesContract, SecurityContract, WritablePath, base_image_from_cascade, image_registry_from_cascade, }; @@ -565,16 +591,15 @@ impl Config { image_registry_from_cascade().unwrap_or_else(|| "ghcr.io/hyperi-io".into()); DeploymentContract { - schema_version: 3, + schema_version: CONTRACT_SCHEMA_VERSION, app_name: "dfe-loader".into(), base_image: base_image.clone(), binary_name: "dfe-loader".into(), description: "High-performance Kafka to ClickHouse data loader".into(), metrics_port: 9090, health: HealthContract { - liveness_path: "/livez".into(), - readiness_path: "/readyz".into(), - metrics_path: "/metrics".into(), + startup_budget_seconds: 60, + ..HealthContract::default() }, env_prefix: "DFE_LOADER".into(), metric_prefix: "loader".into(), @@ -584,35 +609,12 @@ impl Config { extra_ports: vec![ PortContract::tcp("push", 6000) .when_equals("config.transport", TRANSPORT_GRPC) - .bound_from("grpc.listen"), + .bound_from("grpc.listen") + .app_protocol("kubernetes.io/h2c"), ], unbound_listen_paths: vec![], entrypoint_args: vec!["--config".into(), "/etc/dfe/loader.yaml".into()], - secrets: vec![ - SecretGroupContract { - group_name: "kafka".into(), - env_vars: vec![ - SecretEnvContract { - env_var: "DFE_LOADER__KAFKA__SASL__USERNAME".into(), - key_name: "username".into(), - secret_key: "kafka-username".into(), - }, - SecretEnvContract { - env_var: "DFE_LOADER__KAFKA__SASL__PASSWORD".into(), - key_name: "password".into(), - secret_key: "kafka-password".into(), - }, - ], - }, - SecretGroupContract { - group_name: "clickhouse".into(), - env_vars: vec![SecretEnvContract { - env_var: "DFE_LOADER__CLICKHOUSE__PASSWORD".into(), - key_name: "password".into(), - secret_key: "clickhouse-password".into(), - }], - }, - ], + secrets: Self::secrets(), // Derived from Config::default(), NOT hand-authored: a json! literal // is not type-checked against the config structs, so it drifts from // (and can contradict) the real defaults and fails config-check on @@ -659,11 +661,71 @@ impl Config { // pipeline transform stages. config_schema: Some(scalo::deployment::config_schema_json::()), capabilities: Self::capabilities(), + // GeoIP enrichment downloads its databases here, and the root filesystem is read-only. + writable_paths: vec![ + WritablePath::new("geoip", GEOIP_DATA_DIR) + .size_limit("1Gi") + .when(PortCondition::Equals { + path: "config.geoip.enabled".into(), + value: "true".into(), + }), + ], + termination_grace_seconds: 45, + resources: ResourcesContract { + requests: ResourceList { + cpu: "200m".into(), + memory: "256Mi".into(), + }, + limits: ResourceList { + cpu: "1".into(), + memory: "512Mi".into(), + }, + }, + security: SecurityContract::default(), + singleton: false, } } - /// KEDA half of the contract, from this crate's own [`KedaConfig`] defaults - /// so the chart contract test compares against the documented numbers. + /// The Secrets the chart mounts as env vars. + /// + /// figment's `__` nesting in [`Config::load`] reads these into + /// `kafka.sasl.username`, `kafka.sasl.password` and `clickhouse.password`. + fn secrets() -> Vec { + use scalo::deployment::{SecretEnvContract, SecretGroupContract}; + + 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_LOADER__KAFKA__SASL__USERNAME", + "username", + "kafka-username", + ), + env( + "DFE_LOADER__KAFKA__SASL__PASSWORD", + "password", + "kafka-password", + ), + ], + ), + SecretGroupContract::new( + "clickhouse", + vec![env( + "DFE_LOADER__CLICKHOUSE__PASSWORD", + "password", + "clickhouse-password", + )], + ), + ] + } + + /// KEDA half of the contract, from this crate's own [`KedaConfig`] defaults. fn keda_contract() -> scalo::deployment::KedaContract { use scalo::deployment::{KafkaLagTrigger, KedaContract}; @@ -1460,18 +1522,21 @@ kafka: assert_eq!(config.logging.format, "json"); } + /// The contract publishes `Config::default()` as its default config, which must + /// name the transport an empty config file gets. #[test] - fn test_default_transport_via_serde() { - // Config::default() uses #[derive(Default)] which gives String::default()="". - // The "kafka" default is ONLY applied via serde's #[serde(default = "...")]. - // This is intentional — Default and serde default are separate paths. - let default_cfg = Config::default(); - assert_eq!(default_cfg.transport, "", "Rust Default trait gives empty"); + fn test_default_transport_is_kafka() { + assert_eq!(Config::default().transport, TRANSPORT_KAFKA); - // But parsing via serde applies the "kafka" default let yaml = "kafka:\n brokers: [\"x:9092\"]\n"; let parsed: Config = serde_yaml_ng::from_str(yaml).unwrap(); - assert_eq!(parsed.transport, "kafka", "serde default applies 'kafka'"); + assert_eq!(parsed.transport, TRANSPORT_KAFKA); + + let contract = Config::deployment_contract(); + let defaults = contract + .default_config + .expect("contract carries the defaults"); + assert_eq!(defaults["transport"], TRANSPORT_KAFKA); } // ======================================================================== @@ -2049,6 +2114,44 @@ logging: ); } + /// A stored config still carrying the `keda:` block the chart never read: it + /// loads, and the block is ignored. + #[test] + fn keda_block_from_an_older_config_still_loads() { + let yaml = "keda:\n max_replicas: 40\nbuffer:\n flush_rows: 123\n"; + let config: Config = serde_yaml_ng::from_str(yaml).expect("the removed block is ignored"); + assert_eq!(config.buffer.flush_rows, 123); + } + + /// The downloader writes to `geoip.auto_download.data_dir` under a read-only + /// root, so a writable path must cover it exactly while `GeoIP` is on. + #[test] + fn the_geoip_downloader_has_somewhere_to_write_while_geoip_is_on() { + let contract = Config::deployment_contract(); + let defaults = contract + .default_config + .as_ref() + .expect("contract carries the defaults"); + let data_dir = defaults + .pointer("/geoip/auto_download/data_dir") + .and_then(serde_json::Value::as_str) + .expect("the defaults name a geoip data_dir"); + + assert!(contract.security.read_only_root_filesystem); + let geoip_on = scalo::deployment::PortCondition::Equals { + path: "config.geoip.enabled".into(), + value: "true".into(), + }; + assert!( + contract.writable_paths.iter().any(|writable| { + writable.when.as_ref() == Some(&geoip_on) + && std::path::Path::new(data_dir).starts_with(&writable.path) + }), + "no writable path gated on geoip covers {data_dir}: {:?}", + contract.writable_paths + ); + } + /// Committed reflectable artefacts under docs/ must not drift. Regenerate /// with `dfe-loader config-schema --dir docs`. #[test] @@ -2060,7 +2163,10 @@ logging: #[test] fn test_deployment_contract_basic_fields() { let contract = Config::deployment_contract(); - assert_eq!(contract.schema_version, 3); + assert_eq!( + contract.schema_version, + scalo::deployment::CONTRACT_SCHEMA_VERSION + ); assert_eq!(contract.app_name, "dfe-loader"); assert_eq!(contract.binary_name, "dfe-loader"); // base_image is cascade-resolved (deployment.base_image config/env wins, diff --git a/src/config/pipeline.rs b/src/config/pipeline.rs index 5179fed..1d01d4a 100644 --- a/src/config/pipeline.rs +++ b/src/config/pipeline.rs @@ -218,6 +218,10 @@ pub enum GeoIpProvider { Custom, } +/// Where the downloaded `GeoIP` databases go by default, and where the deployment +/// contract mounts a writable volume for them. +pub(crate) const GEOIP_DATA_DIR: &str = "/var/lib/dfe/geoip"; + /// Auto-download settings for `GeoIP` databases #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(default)] @@ -245,7 +249,7 @@ impl Default for AutoDownloadConfig { fn default() -> Self { Self { enabled: true, - data_dir: "/var/lib/dfe/geoip".into(), + data_dir: GEOIP_DATA_DIR.into(), maxmind_account_id: None, maxmind_license_key: None, ipinfo_token: None, @@ -792,16 +796,13 @@ impl Default for MetadataConfig { // KEDA Autoscaling Configuration // ============================================================================ -/// KEDA autoscaling thresholds (deployment-level config). +/// KEDA autoscaling thresholds the deployment contract publishes. /// -/// CHART-GENERATION ONLY. These defaults feed the deployment contract, and the -/// chart contract test validates chart/values.yaml against them. The loader -/// process itself reads NO field of this section: KEDA scales the deployment +/// Not part of [`Config`](crate::config::Config): KEDA scales the deployment /// from the `ScaledObject` the chart renders, so the value that moves a -/// threshold is the chart's `keda.*` (via `--set` or a values overlay), and a -/// `keda:` block in the pod's own config file or a `DFE_LOADER__KEDA__*` env var -/// changes nothing. The runtime scaling signal those thresholds read is -/// [`ScalingConfig`], which the pod does load. +/// threshold is the chart's `keda.*`, and a `keda:` block in the pod's own +/// config file changes nothing. The runtime scaling signal those thresholds +/// read is [`ScalingConfig`], which the pod does load. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)] #[serde(default)] pub struct KedaConfig { diff --git a/src/main.rs b/src/main.rs index eae40b6..0965058 100644 --- a/src/main.rs +++ b/src/main.rs @@ -75,7 +75,8 @@ impl ServiceApp for App { // Seed scalo's cascade first: while it is unset every `from_cascade()` // reader -- scaling.*, worker_pool.*, batch_processing.*, // self_regulation.*, version_check.* -- resolves to its hard-coded - // default and no env var moves it (#160). + // default and no env var moves it (#160). The seed is also what reads + // ./.env into the environment Config::load reads. if let Err(e) = scalo::config::setup(self.common.to_config_options(self.env_prefix())) { // The cascade is a OnceLock; a second load keeps the first seed. debug!(error = %e, "scalo config cascade already seeded"); diff --git a/src/pipeline/orchestrator.rs b/src/pipeline/orchestrator.rs index 52fcf59..0582784 100644 --- a/src/pipeline/orchestrator.rs +++ b/src/pipeline/orchestrator.rs @@ -2093,7 +2093,7 @@ fn memory_guard_config(config: &Config) -> MemoryGuardConfig { // payload.pipeline_mode — pipeline path chosen at startup // metrics.* — HTTP metrics server binds at startup // logging.* — tracing subscriber installed at startup -// scaling.* / keda.* — scaling pressure built at startup +// scaling.* — scaling pressure built at startup // hot_reload.* — watcher config set at startup // schema.* — schema cache created at startup // geoip.* — MMDB readers opened at startup @@ -2130,9 +2130,6 @@ fn warn_restart_required(old: &Config, new: &Config) { if old.scaling != new.scaling { warn!("scaling config changed — requires restart to take effect"); } - if old.keda != new.keda { - warn!("keda config changed — requires restart to take effect"); - } if old.hot_reload != new.hot_reload { warn!("hot_reload config changed — requires restart to take effect"); } diff --git a/tests/integration/config_reachability.rs b/tests/integration/config_reachability.rs index ceb4919..96fb97c 100644 --- a/tests/integration/config_reachability.rs +++ b/tests/integration/config_reachability.rs @@ -8,14 +8,11 @@ //! look identical from outside: the process starts, logs a healthy line, and //! runs on a value the operator did not set. //! -//! The generic ones ([`contract_secret_env_vars_reach_the_config_they_name`] and -//! [`committed_chart_injects_every_contract_secret_env_var`]) walk the -//! deployment contract rather than a hand-written list, so a secret added to the -//! contract later is checked without touching this file. +//! The generic one ([`every_declared_secret_env_var_reaches_the_config`]) walks +//! the deployment contract rather than a hand-written list, so a secret added to +//! the contract later is checked without touching this file. -use std::path::Path; - -use dfe_loader::config::{Config, SaslConfig, TlsConfig}; +use dfe_loader::config::{Config, KedaConfig, SaslConfig, TlsConfig}; use scalo::config::sensitive::expose_during; use serde_json::Value; @@ -58,7 +55,7 @@ fn write_config(dir: &tempfile::TempDir, body: &str) -> String { /// around each case. #[test] #[allow(unsafe_code)] -fn contract_secret_env_vars_reach_the_config_they_name() { +fn every_declared_secret_env_var_reaches_the_config() { let contract = Config::deployment_contract(); let prefix = contract.env_prefix.clone(); @@ -74,18 +71,13 @@ fn contract_secret_env_vars_reach_the_config_they_name() { unsafe { std::env::remove_var(&secret.env_var) }; let json = expose_during(|| serde_json::to_value(&loaded)).expect("config serialises"); - let found = at(&json, &key); - - assert_eq!( - found.and_then(Value::as_str), - Some(sentinel.as_str()), - "deployment contract declares {} for secret '{}', but setting it left \ - config key '{}' at {:?}. The contract, the chart and the config reader \ - have to agree on the env var name or the secret never reaches the process.", - secret.env_var, - secret.secret_key, - key, - found, + let reached = at(&json, &key).and_then(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 key its name spells did not read it", + secret.env_var, group.group_name ); checked += 1; } @@ -97,42 +89,14 @@ fn contract_secret_env_vars_reach_the_config_they_name() { ); } -/// The committed chart must inject every secret env var the contract declares. -/// -/// `scalo::deployment::validate_helm_values` only checks that the template -/// mentions the env PREFIX somewhere, so a renamed or dropped secret env var -/// leaves it green. -#[test] -fn committed_chart_injects_every_contract_secret_env_var() { - let contract = Config::deployment_contract(); - let template = std::fs::read_to_string( - Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/templates/deployment.yaml"), - ) - .expect("read committed deployment template"); - - for group in &contract.secrets { - for secret in &group.env_vars { - assert!( - template.contains(&secret.env_var), - "chart/templates/deployment.yaml does not inject '{}' ({} / {}), so that \ - secret never reaches the pod", - secret.env_var, - group.group_name, - secret.secret_key, - ); - } - } -} - /// The KEDA half of the contract must come from this crate's own `KedaConfig`. /// -/// It was `KedaContract::default()` — scalo's own `KedaConfig` — so the chart -/// contract test compared chart/values.yaml against scalo's numbers while -/// `KedaConfig` documented itself as the source. The two agreed, which is -/// exactly why nothing caught it. +/// It was `KedaContract::default()` — scalo's own numbers — while `KedaConfig` +/// documented itself as the source. The two agreed, which is exactly why +/// nothing caught it. #[test] fn keda_contract_tracks_this_crate_s_keda_defaults() { - let keda = Config::default().keda; + let keda = KedaConfig::default(); let contract = Config::deployment_contract() .keda .expect("contract carries a KEDA section"); @@ -151,8 +115,8 @@ fn keda_contract_tracks_this_crate_s_keda_defaults() { assert_eq!(contract.cpu_threshold, keda.cpu_threshold); } -/// Raw consumer-group lag rises when a downstream stage breaks, so neither the -/// contract nor the committed chart may scale the loader on it. +/// Raw consumer-group lag rises when a downstream stage breaks, so the contract +/// must never scale the loader on it. #[test] fn keda_scales_on_cpu_and_never_on_kafka_lag() { let contract = Config::deployment_contract() @@ -162,18 +126,9 @@ fn keda_scales_on_cpu_and_never_on_kafka_lag() { !contract.kafka_trigger.enabled, "the deployment contract turned the Kafka lag trigger back on" ); - - let scaled_object = std::fs::read_to_string( - Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/templates/keda-scaledobject.yaml"), - ) - .expect("read committed ScaledObject template"); - assert!( - !scaled_object.contains("type: kafka"), - "chart/templates/keda-scaledobject.yaml carries a Kafka lag trigger" - ); assert!( - scaled_object.contains("type: cpu"), - "chart/templates/keda-scaledobject.yaml lost its CPU trigger" + contract.cpu_enabled, + "the deployment contract lost its CPU trigger" ); } @@ -382,3 +337,56 @@ fn env_cascade_forms_reach_the_config() { assert_eq!(config.routing.default_table, "from_flag"); } } + +/// `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-loader")) + .arg("config-check") + .current_dir(dir) + .env_remove("DFE_LOADER_CONFIG") + .env_remove("DFE_LOADER_ROUTING_DEFAULT_DB") + .env_remove("DFE_LOADER_ROUTING_DEFAULT_TABLE") + .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 = tempfile::TempDir::new().expect("tempdir"); + let project = root.path().join("project"); + std::fs::create_dir(&project).expect("project dir"); + std::fs::write( + root.path().join(".env"), + "DFE_LOADER_ROUTING_DEFAULT_DB=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. + std::fs::write( + project.join(".env"), + "DFE_LOADER_ROUTING_DEFAULT_TABLE=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" + ); +} diff --git a/tests/integration/deployment.rs b/tests/integration/deployment.rs index f2dc8f5..168c357 100644 --- a/tests/integration/deployment.rs +++ b/tests/integration/deployment.rs @@ -206,96 +206,3 @@ fn write_artifacts_to_tmp() { eprintln!("Generated artifacts written to: {}", base.display()); } - -// ============================================================================ -// Committed chart vs the generator -// ============================================================================ - -/// Collect a chart directory as relative-path -> contents. -fn chart_files(root: &Path) -> std::collections::BTreeMap { - fn walk(dir: &Path, root: &Path, out: &mut std::collections::BTreeMap) { - for entry in std::fs::read_dir(dir).expect("read chart dir") { - let path = entry.expect("dir entry").path(); - if path.is_dir() { - walk(&path, root, out); - } else { - let rel = path - .strip_prefix(root) - .expect("path under root") - .to_string_lossy() - .into_owned(); - out.insert( - rel, - std::fs::read_to_string(&path).expect("read chart file"), - ); - } - } - } - - let mut out = std::collections::BTreeMap::new(); - walk(root, root, &mut out); - out -} - -/// The committed `chart/` must be what the generator produces. -/// -/// The tests above prove the GENERATOR emits a well-formed chart. None of them -/// look at the directory we actually ship, which is what a deployment consumes -/// -- nothing regenerates it at deploy time. That gap is not theoretical: -/// dfe-fetcher shipped a chart missing `keda-triggerauth.yaml` while its -/// ScaledObject kept an unconditional `authenticationRef` to the object that -/// file creates, so KEDA could not resolve the reference and the app never -/// scaled on lag (hyperi-io/dfe-fetcher#71). -#[test] -fn committed_chart_matches_the_generator() { - let tmp = tempfile::tempdir().expect("tempdir"); - generate_chart(&contract(), tmp.path(), None).expect("generate_chart"); - - let generated = chart_files(tmp.path()); - let chart_dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("chart"); - let committed = chart_files(&chart_dir); - - let missing: Vec<_> = generated - .keys() - .filter(|k| !committed.contains_key(*k)) - .collect(); - let extra: Vec<_> = committed - .keys() - .filter(|k| !generated.contains_key(*k)) - .collect(); - assert!( - missing.is_empty() && extra.is_empty(), - "chart/ is out of step with the generator -- regenerate with \ - `dfe-loader --emit-helm chart`\n generated but not committed: {missing:?}\n \ - committed but not generated: {extra:?}" - ); - - for (name, want) in &generated { - let have = committed.get(name).expect("presence checked above"); - if have == want { - continue; - } - // Report the first differing line: dumping two whole charts at a - // reader is the same as reporting nothing. - let (line_no, from_generator, from_commit) = want - .lines() - .zip(have.lines()) - .enumerate() - .find(|(_, (w, h))| w != h) - .map_or_else( - || { - ( - 0, - format!("{} lines", want.lines().count()), - format!("{} lines", have.lines().count()), - ) - }, - |(i, (w, h))| (i + 1, w.to_string(), h.to_string()), - ); - panic!( - "chart/{name} differs from the generator at line {line_no} -- regenerate with \ - `dfe-loader --emit-helm chart`\n generator: {from_generator}\n \ - committed: {from_commit}" - ); - } -} diff --git a/tests/integration/helm_contract.rs b/tests/integration/helm_contract.rs index 34b8405..a87e33c 100644 --- a/tests/integration/helm_contract.rs +++ b/tests/integration/helm_contract.rs @@ -1,14 +1,14 @@ // SPDX-License-Identifier: BUSL-1.1 // Copyright (c) 2026 HYPERI PTY LIMITED -//! Helm chart + Dockerfile contract sync tests +//! Dockerfile contract sync and CI feature-coverage tests //! -//! Uses scalo's `DeploymentContract` to validate that -//! `chart/values.yaml` and `Dockerfile` stay in sync with app defaults. +//! Uses scalo's `DeploymentContract` to validate that the committed +//! `Dockerfile` stays in sync with app defaults. The Helm chart is assembled +//! from the emitted contract at release, so no chart is committed to check. //! -//! If you change a default port, health path, or KEDA threshold in -//! the Rust config, these tests fail until the chart/Dockerfile are -//! updated (or vice versa). +//! If you change a default port or health path in the Rust config, these +//! tests fail until the Dockerfile is regenerated. use std::path::Path; @@ -19,28 +19,6 @@ fn app_contract() -> DeploymentContract { Config::deployment_contract() } -// ============================================================================ -// Helm Chart Validation -// ============================================================================ - -#[test] -fn test_helm_chart_matches_contract() { - let contract = app_contract(); - let chart_dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("chart"); - - let mismatches = scalo::deployment::validate_helm_values(&contract, &chart_dir).unwrap(); - - assert!( - mismatches.is_empty(), - "Helm chart mismatches with app contract:\n{}", - mismatches - .iter() - .map(|m| format!(" - {m}")) - .collect::>() - .join("\n") - ); -} - // ============================================================================ // Dockerfile Validation // ============================================================================ @@ -63,26 +41,6 @@ fn test_dockerfile_matches_contract() { ); } -// ============================================================================ -// Chart Metadata -// ============================================================================ - -#[test] -fn test_chart_yaml_valid() { - let chart_path = Path::new(env!("CARGO_MANIFEST_DIR")).join("chart/Chart.yaml"); - assert!(chart_path.exists(), "chart/Chart.yaml not found"); - - let content = std::fs::read_to_string(&chart_path).expect("Failed to read Chart.yaml"); - let chart: serde_yaml_ng::Value = - serde_yaml_ng::from_str(&content).expect("Failed to parse Chart.yaml"); - - assert_eq!( - chart["apiVersion"].as_str().unwrap(), - "v2", - "Chart apiVersion should be v2" - ); -} - // ============================================================================ // CI feature coverage // ============================================================================