From 97028bf95033c2d11bcd8e9539f325d3da6b8f5a Mon Sep 17 00:00:00 2001 From: Derek Date: Sun, 15 Nov 2026 22:19:48 +1100 Subject: [PATCH] fix(config): mark memory restart-required and correct stale comments The memory guard is built once at start. Reloads never applied `memory`, though the config said it was hot-reloaded. `memory` is now in the restart group. A reload that changes it logs a restart warning. Correct stale comments about: - where config-check runs - when .env is read - the scaling gate - RowBinary sending JSON columns as text - the pre-warm recheck Remove a duplicated reload list and benchmark figures from a deleted bench. The dev compose file stops mounting a deleted init.sql. config.dev.yaml connects to Kafka's host port. Remove two unlinked reference files that contradicted dfe-schemas. --- config.dev.yaml | 2 +- docker-compose.dev.yaml | 1 - docs/config-schema.json | 54 ++++++++++++------------- docs/config-schema.yaml | 43 ++++++++++---------- reference/common_header.csv | 7 ---- reference/common_header.sql | 78 ------------------------------------ src/clickhouse/config.rs | 7 ++-- src/clickhouse/inserter.rs | 2 +- src/config/loader.rs | 4 +- src/main.rs | 9 ++--- src/pipeline/orchestrator.rs | 38 +++--------------- src/transform/extractor.rs | 11 +---- 12 files changed, 66 insertions(+), 190 deletions(-) delete mode 100644 reference/common_header.csv delete mode 100644 reference/common_header.sql diff --git a/config.dev.yaml b/config.dev.yaml index 64e2970b..9a3b4375 100644 --- a/config.dev.yaml +++ b/config.dev.yaml @@ -6,7 +6,7 @@ kafka: brokers: - - localhost:9092 + - localhost:19092 group: dfe-loader-dev topics: - events diff --git a/docker-compose.dev.yaml b/docker-compose.dev.yaml index 91339d89..d9bae44a 100644 --- a/docker-compose.dev.yaml +++ b/docker-compose.dev.yaml @@ -24,7 +24,6 @@ services: CLICKHOUSE_PASSWORD: "" volumes: - clickhouse-data:/var/lib/clickhouse - - ./tests/fixtures/init.sql:/docker-entrypoint-initdb.d/init.sql:ro healthcheck: test: ["CMD", "clickhouse-client", "--query", "SELECT 1"] interval: 5s diff --git a/docs/config-schema.json b/docs/config-schema.json index 6fd97104..272ad0b5 100644 --- a/docs/config-schema.json +++ b/docs/config-schema.json @@ -195,6 +195,14 @@ "table": "dlq" } }, + "memory": { + "description": "Memory limits. **Restart required.**", + "$ref": "#/$defs/MemoryConfig", + "default": { + "limit_bytes": 0, + "pressure_threshold": 0.8 + } + }, "routing": { "description": "Table and tenant routing. **Hot-reloaded**, `routing.dlq` aside.", "$ref": "#/$defs/RoutingConfig", @@ -233,14 +241,6 @@ "flush_age_secs": 5 } }, - "memory": { - "description": "Memory limits. **Hot-reloaded.**", - "$ref": "#/$defs/MemoryConfig", - "default": { - "limit_bytes": 0, - "pressure_threshold": 0.8 - } - }, "timestamp_dq": { "description": "Timestamp data quality validation. **Hot-reloaded.**", "$ref": "#/$defs/TimestampDqConfig", @@ -813,7 +813,7 @@ } }, "InsertFormat": { - "description": "Insert format for data writes.\n\nControls how rows are encoded and sent to `ClickHouse`. `RowBinary` is the\ndefault — it uses schema reflection to encode `Map` directly\nto binary, so `ClickHouse` skips JSON parsing entirely. This significantly\nreduces CPU load on the `ClickHouse` cluster at scale.\n\n`JsonEachRow` is the fallback — simpler, self-describing, but the `ClickHouse`\nserver pays the cost of parsing every JSON row on ingest.", + "description": "Insert format for data writes.\n\nControls how rows are encoded and sent to `ClickHouse`. `RowBinary` is the\ndefault — it uses schema reflection to encode `Map` directly\nto binary, so `ClickHouse` parses only `JSON` columns, which travel as text.\n\n`JsonEachRow` is the fallback — simpler, self-describing, but the `ClickHouse`\nserver pays the cost of parsing every JSON row on ingest.", "oneOf": [ { "description": "Schema-reflected `RowBinary` (default).\n\nFetches schema from `system.columns`, encodes values to binary client-side.\n`ClickHouse` receives pre-columnarised data — zero server-side parsing.\nTotal CPU (client + cluster) is significantly lower than `JSONEachRow`.", @@ -1367,6 +1367,24 @@ } } }, + "MemoryConfig": { + "type": "object", + "properties": { + "limit_bytes": { + "description": "Maximum memory for buffers in bytes. 0 = auto-detect (67% of available)", + "type": "integer", + "format": "uint", + "minimum": 0, + "default": 0 + }, + "pressure_threshold": { + "description": "Pressure threshold (0.0-1.0) - pause consumption above this", + "type": "number", + "format": "double", + "default": 0.8 + } + } + }, "RoutingConfig": { "type": "object", "properties": { @@ -1585,24 +1603,6 @@ } } }, - "MemoryConfig": { - "type": "object", - "properties": { - "limit_bytes": { - "description": "Maximum memory for buffers in bytes. 0 = auto-detect (67% of available)", - "type": "integer", - "format": "uint", - "minimum": 0, - "default": 0 - }, - "pressure_threshold": { - "description": "Pressure threshold (0.0-1.0) - pause consumption above this", - "type": "number", - "format": "double", - "default": 0.8 - } - } - }, "TimestampDqConfig": { "type": "object", "properties": { diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index a8a5468c..01f24aae 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -190,6 +190,12 @@ properties: - dfe_fetcher_dlq - dfe_transform_dlq table: dlq + memory: + description: Memory limits. **Restart required.** + $ref: "#/$defs/MemoryConfig" + default: + limit_bytes: 0 + pressure_threshold: 0.8 routing: description: Table and tenant routing. **Hot-reloaded**, `routing.dlq` aside. $ref: "#/$defs/RoutingConfig" @@ -221,12 +227,6 @@ properties: flush_bytes: 67108864 flush_rows: 20000 flush_age_secs: 5 - memory: - description: Memory limits. **Hot-reloaded.** - $ref: "#/$defs/MemoryConfig" - default: - limit_bytes: 0 - pressure_threshold: 0.8 timestamp_dq: description: Timestamp data quality validation. **Hot-reloaded.** $ref: "#/$defs/TimestampDqConfig" @@ -727,8 +727,7 @@ $defs: Controls how rows are encoded and sent to `ClickHouse`. `RowBinary` is the default — it uses schema reflection to encode `Map` directly - to binary, so `ClickHouse` skips JSON parsing entirely. This significantly - reduces CPU load on the `ClickHouse` cluster at scale. + to binary, so `ClickHouse` parses only `JSON` columns, which travel as text. `JsonEachRow` is the fallback — simpler, self-describing, but the `ClickHouse` server pays the cost of parsing every JSON row on ingest. @@ -1238,6 +1237,20 @@ $defs: picks for its tenant. type: string default: dlq + MemoryConfig: + type: object + properties: + limit_bytes: + description: Maximum memory for buffers in bytes. 0 = auto-detect (67% of available) + type: integer + format: uint + minimum: 0 + default: 0 + pressure_threshold: + description: Pressure threshold (0.0-1.0) - pause consumption above this + type: number + format: double + default: 0.8 RoutingConfig: type: object properties: @@ -1446,20 +1459,6 @@ $defs: format: uint64 minimum: 0 default: 5 - MemoryConfig: - type: object - properties: - limit_bytes: - description: Maximum memory for buffers in bytes. 0 = auto-detect (67% of available) - type: integer - format: uint - minimum: 0 - default: 0 - pressure_threshold: - description: Pressure threshold (0.0-1.0) - pause consumption above this - type: number - format: double - default: 0.8 TimestampDqConfig: type: object properties: diff --git a/reference/common_header.csv b/reference/common_header.csv deleted file mode 100644 index dd075071..00000000 --- a/reference/common_header.csv +++ /dev/null @@ -1,7 +0,0 @@ -column,type,default,index_order,index_type,comment -timestamp,DateTime64(3),,,dimension,Time the event occurred (milliseconds) NOT NULLABLE -timestamp_load,DateTime64(3),now64(3),,,Load time (ClickHouse DEFAULT) NOT NULLABLE -_uuid,UUID,generateUUIDv7(),,,Unique event ID UUIDv7 (time-ordered) NOT NULLABLE -logoriginal,String,,,,The original unparsed log line NULLABLE -logjson,JSON,,,,Complete message as it was from kafka as a JSON type NULLABLE -_tags,JSON,,,,The message meta information and collector or agent information as a JSON type NULLABLE diff --git a/reference/common_header.sql b/reference/common_header.sql deleted file mode 100644 index cd822773..00000000 --- a/reference/common_header.sql +++ /dev/null @@ -1,78 +0,0 @@ --- Common Header Schema v2 - Reference DDL --- --- This is the minimal required schema for DFE v2 tables. --- Additional dynamic fields will be added via ALTER TABLE as needed. --- --- Usage: --- Replace {db} and {table} with actual database and table names. --- Example: CREATE TABLE acme.events_auth ... --- --- Requirements: --- - ClickHouse 24.8+ for generateUUIDv7() --- - JSON type support (experimental in 24.x, stable in 25.x) - -CREATE TABLE IF NOT EXISTS {db}.{table} -( - -- Event timestamp (millisecond precision) - -- Source: extracted from event data, fallback to now64(3) - `timestamp` DateTime64(3) CODEC(Delta, ZSTD(1)), - - -- Load timestamp (when event was inserted) - -- Generated by ClickHouse DEFAULT - all rows in batch get same value - `timestamp_load` DateTime64(3) DEFAULT now64(3) CODEC(Delta, ZSTD(1)), - - -- Unique event ID (UUIDv7 - time-ordered, sortable) - -- Generated by ClickHouse DEFAULT - no client generation needed - `_uuid` UUID DEFAULT generateUUIDv7(), - - -- Organisation ID (for multi-tenancy and row-level security) - -- Extracted from org_id field in source data - -- Used by ClickHouse row policies for data isolation - -- REQUIRED for shared schema deployments (common database) - `_org_id` String CODEC(ZSTD(1)), - - -- Original unparsed log line (if available) - `logoriginal` Nullable(String) CODEC(ZSTD(3)), - - -- Complete original Kafka message as JSON - -- Captured before any transformation - `logjson` Nullable(JSON) CODEC(ZSTD(3)), - - -- Message metadata and collector/agent information - -- Extracted from tags/meta fields before flattening - `_tags` Nullable(JSON) CODEC(ZSTD(3)) - - -- Dynamic fields added here via ALTER TABLE ADD COLUMN -) -ENGINE = MergeTree() -ORDER BY (timestamp, _org_id, _uuid) -- Include _org_id for RLS performance -PARTITION BY (toYYYYMM(timestamp), _org_id) -- Partition by month + org for pruning -SETTINGS index_granularity = 8192; - --- Notes: --- --- 1. _org_id is REQUIRED for shared schema deployments (common database). --- - Extracted from org_id field in source data (configurable) --- - Used by ClickHouse row policies for data isolation (RLS) --- - See reference/clickhouse_rls.md for row policy setup --- --- 2. Routing fields (event_category) determine the table name. --- - org_id determines database ONLY if in routed_orgs list --- - Default: all data goes to common.{table} --- --- 3. timestamp_load uses ClickHouse DEFAULT now64(3). --- All rows in an INSERT batch get the same timestamp (per-batch, not per-row). --- This is acceptable for windowing queries. --- --- 4. _uuid uses generateUUIDv7() which is thread-monotonic in ClickHouse 24.8+. --- Time-ordered UUIDs are ideal for event IDs (sortable, unique). --- --- 5. JSON columns (logjson, _tags) require ClickHouse JSON type support. --- Use ZSTD(3) compression for good ratio on JSON data. --- --- 6. Underscore-prefixed fields (_uuid, _org_id, _tags) avoid collision with source data. --- The loader preserves these during field sanitization. --- --- 7. ORDER BY includes _org_id for efficient row policy filtering. --- PARTITION BY includes _org_id for partition pruning (queries only read relevant org). --- See reference/clickhouse_rls.md for performance details. diff --git a/src/clickhouse/config.rs b/src/clickhouse/config.rs index 6e2e5136..998b9675 100644 --- a/src/clickhouse/config.rs +++ b/src/clickhouse/config.rs @@ -52,8 +52,7 @@ impl std::fmt::Display for Transport { /// /// Controls how rows are encoded and sent to `ClickHouse`. `RowBinary` is the /// default — it uses schema reflection to encode `Map` directly -/// to binary, so `ClickHouse` skips JSON parsing entirely. This significantly -/// reduces CPU load on the `ClickHouse` cluster at scale. +/// to binary, so `ClickHouse` parses only `JSON` columns, which travel as text. /// /// `JsonEachRow` is the fallback — simpler, self-describing, but the `ClickHouse` /// server pays the cost of parsing every JSON row on ingest. @@ -135,7 +134,7 @@ pub struct ClickHouseConfig { /// Insert format — how rows are encoded for INSERT. /// /// `RowBinary` (default): schema-reflected binary encoding. `ClickHouse` - /// skips JSON parsing. Lower total CPU across client + cluster. + /// parses only `JSON` columns, which travel as text. /// /// `JsonEachRow`: JSON text via HTTP. Self-describing, `ClickHouse` coerces /// types server-side. Higher cluster CPU but simpler. @@ -352,7 +351,7 @@ impl ClickHouseConfig { // JSONEachRow is HTTP-only: the dynamic insert path sends JSONEachRow // through `Client::insert_formatted_with` (HTTP). A native-transport // client has no HTTP endpoint, so this combination fails at insert - // time. Reject it at config time so `config-check` catches it. + // time. Rejected here, when the pipeline starts and before readiness. if self.transport == Transport::Native && self.insert_format == InsertFormat::JsonEachRow { return Err( "insert_format 'json_each_row' requires protocol 'http' (JSONEachRow is \ diff --git a/src/clickhouse/inserter.rs b/src/clickhouse/inserter.rs index a3d82d36..4160a4f3 100644 --- a/src/clickhouse/inserter.rs +++ b/src/clickhouse/inserter.rs @@ -408,7 +408,7 @@ impl Inserter { /// Set the insert format. /// /// - `RowBinary` (default): schema-reflected binary via `DynamicInsert`. - /// `ClickHouse` skips JSON parsing — lower cluster CPU at scale. + /// `ClickHouse` parses only `JSON` columns, which travel as text. /// - `JsonEachRow`: NDJSON via `InsertFormatted`. Self-describing fallback. #[must_use] pub fn with_insert_format(mut self, format: InsertFormat) -> Self { diff --git a/src/config/loader.rs b/src/config/loader.rs index eb29c940..e2cd0f89 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -89,14 +89,14 @@ pub struct Config { pub batch_processing: BatchProcessingConfig, /// Kafka DLQ topics loaded into a table of their own. **Restart required.** pub dlq_load: DlqLoadConfig, + /// Memory limits. **Restart required.** + pub memory: MemoryConfig, // --- Hot-reloaded (takes effect on next batch) --- /// Table and tenant routing. **Hot-reloaded**, `routing.dlq` aside. pub routing: RoutingConfig, /// Buffer flush thresholds. **Hot-reloaded.** pub buffer: BufferConfig, - /// Memory limits. **Hot-reloaded.** - pub memory: MemoryConfig, /// Timestamp data quality validation. **Hot-reloaded.** pub timestamp_dq: TimestampDqConfig, /// Field name sanitisation. **Hot-reloaded.** diff --git a/src/main.rs b/src/main.rs index 8360e2c0..03d422aa 100644 --- a/src/main.rs +++ b/src/main.rs @@ -73,8 +73,8 @@ impl ServiceApp for App { } fn load_config(&self, path: Option<&str>) -> Result { - // run_app has set the cascade up from --config, which also reads - // ./.env into the environment Config::load reads. + // run_app has set the cascade up from --config. Without --config it + // also reads ./.env into the environment Config::load reads. let config = Config::load(path).map_err(|e| CliError::Config(e.to_string()))?; config .validate() @@ -128,9 +128,8 @@ impl ServiceApp for App { .clone() .unwrap_or_else(|| Arc::new(config.scaling.build_pressure())); - // The engine KEDA reads takes its gates from scalo's cascade, which - // reads neither the `--config` file nor the `DFE_LOADER__` env form, - // so a gate set either way is reported here and ignored there (#160). + // The engine KEDA reads takes its gates from scalo's cascade, so a + // gate it did not pick up is reported here. if scaling.is_enabled() != config.scaling.enabled { warn!( configured = config.scaling.enabled, diff --git a/src/pipeline/orchestrator.rs b/src/pipeline/orchestrator.rs index 573f5ccf..ccab9d55 100644 --- a/src/pipeline/orchestrator.rs +++ b/src/pipeline/orchestrator.rs @@ -2348,42 +2348,11 @@ fn memory_guard_config(config: &Config) -> MemoryGuardConfig { } } -// Enrichment helpers moved to super::enrichment (enrichment.rs) - -// ============================================================================= -// Hot-Reload Safety: restart-required change detection -// ============================================================================= -// -// Hot-reloaded (takes effect on next batch): -// routing.* -- router, extractor, default table and pre-warm rebuilt -// timestamp_dq.* -- transformer rebuilt on reload -// metadata.* -- router, extractor, transformer + capture overrides rebuilt -// field_sanitization.* -- transformer rebuilt on reload -// buffer.flush_rows / flush_bytes / flush_age_secs -- buffer thresholds updated -// -// Requires pod restart (connections/state established at startup): -// kafka.* — Kafka consumer created at startup -// grpc.* — gRPC server binds at startup -// transport — transport type bound at startup -// clickhouse.* — HTTP client + clickhouse::Client created at startup -// payload.pipeline_mode — pipeline path chosen at startup -// metrics.* — HTTP metrics server binds at startup -// logging.* — tracing subscriber installed 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 -// computed_columns.* — computed column cache built at startup -// column_directives.* — column directive cache built at startup -// routing.dlq -- DLQ backends opened at startup -// field_mapping.* -- mapping rules built at startup -// enrichment.* -- enrichment pipeline built at startup - /// Log warnings for config fields that changed but require a pod restart. /// /// These fields are bound to connections or state created at startup. Changing /// them via hot-reload has no effect — the old values remain active until the -/// pod is restarted. +/// pod is restarted. Each `Config` field in `config/loader.rs` says which it is. fn warn_restart_required(old: &Config, new: &Config) { if old.transport != new.transport { warn!("transport changed — requires restart to take effect"); @@ -2437,6 +2406,9 @@ fn warn_restart_required(old: &Config, new: &Config) { if old.enrichment != new.enrichment { warn!("enrichment config changed -- requires restart to take effect"); } + if old.memory != new.memory { + warn!("memory config changed -- requires restart to take effect"); + } } /// Report from `pre_warm_with_retry`. @@ -2722,7 +2694,7 @@ async fn pre_warm_schemas( /// Whether a pre-warm answer goes to the event loop. An absent source table is /// held back, because the engine writes a source's routing and creates its /// table together, and an absence recorded in that gap sends the source's rows -/// to the default table until `ABSENT_TABLE_TTL` runs out. Its first batch +/// to the default table until the next live recheck. Its first batch /// resolves it instead. The default table is never recorded absent, so its /// answer goes through and a missing one is warned about at once. fn pre_warm_forwards(table: &str, schema: &SchemaResolution, default_table: &str) -> bool { diff --git a/src/transform/extractor.rs b/src/transform/extractor.rs index 13a4a151..0d89c7c8 100644 --- a/src/transform/extractor.rs +++ b/src/transform/extractor.rs @@ -4,15 +4,8 @@ //! Schema-guided SIMD field extractor for the `json_primary` pipeline mode. //! //! Parses the raw payload **once** with `sonic_rs::from_slice`, then performs -//! O(1) hash lookups for each schema column. -//! -//! Benchmarks (`benches/simdjson_spike.rs`, batch=10K) confirm the full-parse -//! approach outperforms `get_from_slice × N` in all realistic cases: -//! - 2.7× faster at N=30 flat payload (196 ms → 72 ms) -//! - 4.2× faster at N=15 nested payload (each miss still scans the full doc) -//! - Tied at N=15 flat payload with high field hit-rate -//! Misses are free with the DOM approach (O(1) hash lookup returns None) -//! but cost a full document scan with `get_from_slice`. +//! O(1) hash lookups for each schema column. A missing column costs one lookup +//! here, where `get_from_slice` per column rescans the whole document. //! //! ## What this does NOT do //!