From 36a2afee46983a22d941b3daa6aaffe037edba92 Mon Sep 17 00:00:00 2001 From: Derek Date: Sun, 15 Nov 2026 07:42:29 +1100 Subject: [PATCH] fix(capture): drop the disable lists and no_capture tags capture_mode and table_capture_modes are the only capture switches. Remove disable_json_tables and disable_raw_tables from the metadata config. A table comment's @no_capture_json and @no_capture_raw tags no longer change the capture mode. A table with no @capture_mode tag takes its table_capture_modes entry, else the global capture_mode. Remove the tests for the disable lists and no_capture tags. The tag-removal test also covers a table with its own table_capture_modes entry. The config schema is regenerated with dfe-loader config-schema --dir docs. The loader does not read capture_json. Its doc says so. capture_raw still gates _raw under full mode. --- docs/config-schema.json | 20 +- docs/config-schema.yaml | 16 +- docs/transport/common-header.md | 3 +- src/config/loader.rs | 8 +- src/config/pipeline.rs | 23 +-- src/pipeline/capture.rs | 317 +++++--------------------------- 6 files changed, 57 insertions(+), 330 deletions(-) diff --git a/docs/config-schema.json b/docs/config-schema.json index 2fce35a4..fd06f5ed 100644 --- a/docs/config-schema.json +++ b/docs/config-schema.json @@ -292,8 +292,6 @@ "_source" ], "source_output": "_source", - "disable_json_tables": [], - "disable_raw_tables": [], "remove_routing_fields": true } }, @@ -1718,7 +1716,7 @@ "default": {} }, "capture_json": { - "description": "Store complete original Kafka message as JSON before transformation", + "description": "Not read: `capture_mode` decides whether `_json` is written.", "type": "boolean", "default": true }, @@ -1765,22 +1763,6 @@ "type": "string", "default": "_source" }, - "disable_json_tables": { - "description": "Tables where _json capture is disabled (e.g., [\"dfe.metrics\"])", - "type": "array", - "items": { - "type": "string" - }, - "default": [] - }, - "disable_raw_tables": { - "description": "Tables where _raw capture is disabled (e.g., [\"dfe.metrics\"])", - "type": "array", - "items": { - "type": "string" - }, - "default": [] - }, "remove_routing_fields": { "description": "Remove routing fields from output after extraction", "type": "boolean", diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index 8d3c769a..9e6c981b 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -272,8 +272,6 @@ properties: source_fields: - _source source_output: _source - disable_json_tables: [] - disable_raw_tables: [] remove_routing_fields: true field_mapping: description: Field mapping overrides. **Restart required.** @@ -1553,7 +1551,7 @@ $defs: $ref: "#/$defs/CaptureMode" default: {} capture_json: - description: Store complete original Kafka message as JSON before transformation + description: "Not read: `capture_mode` decides whether `_json` is written." type: boolean default: true json_output: @@ -1592,18 +1590,6 @@ $defs: description: Output field name for _source type: string default: _source - disable_json_tables: - description: Tables where _json capture is disabled (e.g., ["dfe.metrics"]) - type: array - items: - type: string - default: [] - disable_raw_tables: - description: Tables where _raw capture is disabled (e.g., ["dfe.metrics"]) - type: array - items: - type: string - default: [] remove_routing_fields: description: Remove routing fields from output after extraction type: boolean diff --git a/docs/transport/common-header.md b/docs/transport/common-header.md index 473f6a76..5bd26ad0 100644 --- a/docs/transport/common-header.md +++ b/docs/transport/common-header.md @@ -372,8 +372,7 @@ metadata: ``` `capture_mode` controls `_json` and `_raw` population -- see the Capture Modes -section below. Legacy booleans `capture_json` / `capture_raw` are still accepted -for backwards compatibility and mapped onto `capture_mode`. +section below. ### `_tags` diff --git a/src/config/loader.rs b/src/config/loader.rs index 653fc073..eb29c940 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -1135,8 +1135,6 @@ mod tests { assert!(config.capture_raw); assert_eq!(config.raw_output, "_raw"); assert_eq!(config.raw_source_fields, [] as [std::string::String; 0]); - assert_eq!(config.disable_json_tables, [] as [std::string::String; 0]); - assert_eq!(config.disable_raw_tables, [] as [std::string::String; 0]); } #[test] @@ -1158,9 +1156,7 @@ mod tests { "json_output": "_custom_json", "capture_raw": false, "raw_output": "_custom_raw", - "raw_source_fields": ["original_log", "raw_message"], - "disable_json_tables": ["common.metrics"], - "disable_raw_tables": ["common.health"] + "raw_source_fields": ["original_log", "raw_message"] }"#; let config: MetadataConfig = serde_json::from_str(json_str).unwrap(); assert!(!config.capture_json); @@ -1171,8 +1167,6 @@ mod tests { config.raw_source_fields, vec!["original_log", "raw_message"] ); - assert_eq!(config.disable_json_tables, vec!["common.metrics"]); - assert_eq!(config.disable_raw_tables, vec!["common.health"]); } // ======================================================================== diff --git a/src/config/pipeline.rs b/src/config/pipeline.rs index 5d7740a7..a60f3f00 100644 --- a/src/config/pipeline.rs +++ b/src/config/pipeline.rs @@ -1032,15 +1032,15 @@ pub struct MetadataConfig { #[serde(default)] pub table_capture_modes: HashMap, - // _json capture (Common Header v2) — DEPRECATED, use capture_mode - /// Store complete original Kafka message as JSON before transformation + // _json capture (Common Header v2) + /// Not read: `capture_mode` decides whether `_json` is written. #[serde(alias = "capture_logjson", default = "default_true")] pub capture_json: bool, /// Output field name for _json #[serde(alias = "logjson_output")] pub json_output: String, - // _raw field injection (Common Header v2) — DEPRECATED, use capture_mode + // _raw field injection (Common Header v2) // Implements @renamed: first(source_fields...) → raw_output // Silent no-op if destination already present in data /// Enable _raw field injection from source (zero-copy rename) @@ -1061,12 +1061,6 @@ pub struct MetadataConfig { /// Output field name for _source pub source_output: String, - // Per-table capture overrides — DEPRECATED, use table_capture_modes - /// Tables where _json capture is disabled (e.g., ["dfe.metrics"]) - pub disable_json_tables: Vec, - /// Tables where _raw capture is disabled (e.g., ["dfe.metrics"]) - pub disable_raw_tables: Vec, - // Routing field removal (Common Header v2) /// Remove routing fields from output after extraction pub remove_routing_fields: bool, @@ -1095,11 +1089,11 @@ impl Default for MetadataConfig { capture_mode: CaptureMode::Full, table_capture_modes: HashMap::new(), - // _json capture defaults (deprecated — use capture_mode) + // _json capture defaults capture_json: true, json_output: "_json".to_string(), - // _raw capture defaults (deprecated — use capture_mode) + // _raw capture defaults capture_raw: true, raw_source_fields: Vec::new(), raw_output: "_raw".to_string(), @@ -1109,10 +1103,6 @@ impl Default for MetadataConfig { source_fields: vec!["_source".to_string()], source_output: "_source".to_string(), - // Per-table overrides (deprecated — use table_capture_modes) - disable_json_tables: vec![], - disable_raw_tables: vec![], - // Routing field removal defaults remove_routing_fields: true, } @@ -1302,9 +1292,6 @@ pub enum CaptureMode { /// 1. DDL: `@capture_mode: raw_only` in table COMMENT /// 2. Config: `table_capture_modes` map /// 3. Config: global `capture_mode` -/// -/// Legacy fields (`disable_json_tables`, `@no_capture_json`) are deprecated -/// but still supported for one release cycle. #[derive(Debug, Clone, Default, PartialEq)] pub struct TableCaptureConfig { pub mode: CaptureMode, diff --git a/src/pipeline/capture.rs b/src/pipeline/capture.rs index 31bd6291..b5996330 100644 --- a/src/pipeline/capture.rs +++ b/src/pipeline/capture.rs @@ -8,14 +8,11 @@ //! 2. Config `table_capture_modes` map — per-table override //! 3. Config `capture_mode` — global default //! -//! Legacy fields (`disable_json_tables`, `@no_capture_json`, etc.) are deprecated -//! but supported for backward compatibility during migration. -//! //! Key design: [`CaptureOverrides::derive_config`] is a pure `&self` method — safe //! for rayon `par_iter`. The mutable operations (`mark_pending`, `ensure_cached`, //! `update_from_comment`) are called only in the sequential phase. -use rustc_hash::{FxHashMap, FxHashSet}; +use rustc_hash::FxHashMap; use tracing::warn; use crate::config::{CaptureMode, MetadataConfig, TableCaptureConfig}; @@ -29,10 +26,6 @@ pub struct CaptureOverrides { table_modes: FxHashMap, /// Resolved per-table configs (populated by DDL tag resolution). configs: FxHashMap, - /// Legacy: tables with `_json` disabled (deprecated, from config lists). - disable_json_tables: FxHashSet, - /// Legacy: tables with `_raw` disabled (deprecated, from config lists). - disable_raw_tables: FxHashSet, /// Tables needing async DDL tag resolution. pending_tables: Vec, } @@ -40,16 +33,6 @@ pub struct CaptureOverrides { impl CaptureOverrides { /// Create from metadata config. pub fn new(metadata_config: &MetadataConfig) -> Self { - // Warn on deprecated config fields - if !metadata_config.disable_json_tables.is_empty() - || !metadata_config.disable_raw_tables.is_empty() - { - warn!( - "disable_json_tables/disable_raw_tables are deprecated. \ - Use table_capture_modes with capture_mode values instead." - ); - } - Self { global_mode: metadata_config.capture_mode, table_modes: metadata_config @@ -58,12 +41,6 @@ impl CaptureOverrides { .map(|(k, v)| (k.clone(), *v)) .collect(), configs: FxHashMap::default(), - disable_json_tables: metadata_config - .disable_json_tables - .iter() - .cloned() - .collect(), - disable_raw_tables: metadata_config.disable_raw_tables.iter().cloned().collect(), pending_tables: Vec::new(), } } @@ -73,8 +50,7 @@ impl CaptureOverrides { /// Resolution order (highest priority first): /// 1. DDL-resolved config (from `update_from_comment`) /// 2. Per-table config (`table_capture_modes`) - /// 3. Legacy per-table disable lists (deprecated) - /// 4. Global `capture_mode` + /// 3. Global `capture_mode` pub fn derive_config(&self, table: &str) -> TableCaptureConfig { // DDL-resolved overrides take highest precedence if let Some(config) = self.configs.get(table) { @@ -83,31 +59,15 @@ impl CaptureOverrides { self.configured(table) } - /// The table's policy from config alone: steps 2 to 4 of + /// The table's policy from config alone: steps 2 and 3 of /// [`Self::derive_config`]. fn configured(&self, table: &str) -> TableCaptureConfig { - // Per-table config map (new style) - if let Some(&mode) = self.table_modes.get(table) { - return TableCaptureConfig { mode }; - } - - // Legacy per-table disable lists (deprecated, backward compat) - let json_disabled = self.disable_json_tables.contains(table); - let raw_disabled = self.disable_raw_tables.contains(table); - if json_disabled || raw_disabled { - let mode = match (json_disabled, raw_disabled) { - (true, true) => CaptureMode::ExtractedOnly, - (true, false) => CaptureMode::RawOnly, - (false, true) => self.global_mode, // _raw disabled but _json enabled → use global - (false, false) => unreachable!(), - }; - return TableCaptureConfig { mode }; - } - - // Global default - TableCaptureConfig { - mode: self.global_mode, - } + let mode = self + .table_modes + .get(table) + .copied() + .unwrap_or(self.global_mode); + TableCaptureConfig { mode } } /// Populate the config cache for a table if absent. @@ -132,15 +92,12 @@ impl CaptureOverrides { /// Update capture config from DDL table comment tags, read successfully. /// - /// `@capture_mode: ` takes precedence over legacy tags. - /// Legacy `@no_capture_json` / `@no_capture_raw` tags are supported - /// for backward compatibility but superseded by `@capture_mode`. A - /// comment with no capture tag, empty included, drops the table's + /// `@capture_mode: ` overrides the configured policy. A comment + /// with no `@capture_mode` tag, empty included, drops the table's /// override, so the configured policy applies again. pub fn update_from_comment(&mut self, table: &str, comment: &str) { let tags = TableTags::from_comment(comment); - // New-style: @capture_mode tag wins over everything if let Some(mode_str) = tags.get("capture_mode") { let mode = match mode_str { "full" => CaptureMode::Full, @@ -161,25 +118,10 @@ impl CaptureOverrides { return; } - // Legacy tags: @no_capture_json / @no_capture_raw (deprecated) - let json_disabled = tags.get("no_capture_json").is_some_and(|v| v == "true"); - let raw_disabled = tags.get("no_capture_raw").is_some_and(|v| v == "true"); - - if json_disabled || raw_disabled { - let mode = match (json_disabled, raw_disabled) { - (true, true) => CaptureMode::ExtractedOnly, - (true, false) => CaptureMode::RawOnly, - (false, true) => self.global_mode, - (false, false) => unreachable!("one of the two is set"), - }; - self.configs - .insert(table.to_string(), TableCaptureConfig { mode }); - } else { - // Cached rather than removed: an absent entry is what marks a - // table for resolution, and the next message would queue it again. - let configured = self.configured(table); - self.configs.insert(table.to_string(), configured); - } + // Cached rather than removed: an absent entry is what marks a + // table for resolution, and the next message would queue it again. + let configured = self.configured(table); + self.configs.insert(table.to_string(), configured); } } @@ -314,64 +256,6 @@ mod tests { ); } - #[test] - fn test_legacy_disable_json_maps_to_raw_only() { - let mut metadata_config = MetadataConfig::default(); - metadata_config - .disable_json_tables - .push("dfe.metrics".to_string()); - let overrides = CaptureOverrides::new(&metadata_config); - - assert_eq!( - overrides.derive_config("dfe.metrics").mode, - CaptureMode::RawOnly - ); - } - - #[test] - fn test_legacy_disable_both_maps_to_extracted_only() { - let mut metadata_config = MetadataConfig::default(); - metadata_config - .disable_json_tables - .push("dfe.health".to_string()); - metadata_config - .disable_raw_tables - .push("dfe.health".to_string()); - let overrides = CaptureOverrides::new(&metadata_config); - - assert_eq!( - overrides.derive_config("dfe.health").mode, - CaptureMode::ExtractedOnly - ); - } - - #[test] - fn test_legacy_ddl_no_capture_json_maps_to_raw_only() { - let metadata_config = MetadataConfig::default(); - let mut overrides = CaptureOverrides::new(&metadata_config); - - overrides.update_from_comment("dfe.events", "@no_capture_json: true"); - - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::RawOnly - ); - } - - #[test] - fn test_ddl_capture_mode_wins_over_legacy_tags() { - let metadata_config = MetadataConfig::default(); - let mut overrides = CaptureOverrides::new(&metadata_config); - - // Both old and new tags present — new wins - overrides.update_from_comment("dfe.events", "@no_capture_json: true | @capture_mode: full"); - - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::Full - ); - } - #[test] fn test_ensure_cached_prevents_duplicate_pending() { let metadata_config = MetadataConfig::default(); @@ -402,28 +286,41 @@ mod tests { } /// A capture tag taken off the table's DDL takes its override with it, - /// whether the comment is now empty or carries other tags. + /// whether the comment is now empty or carries other tags, and the + /// table's `table_capture_modes` entry applies again ahead of the global + /// mode. #[test] fn removing_the_capture_tag_restores_the_configured_policy() { - for later in ["", "@retention: 30d"] { - let metadata_config = MetadataConfig::default(); - let mut overrides = CaptureOverrides::new(&metadata_config); - overrides.update_from_comment("dfe.events", "@capture_mode: extracted_only"); - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::ExtractedOnly - ); - overrides.update_from_comment("dfe.events", later); - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::Full, - "after {later:?}" - ); - overrides.mark_pending("dfe.events"); - assert!( - overrides.take_pending().is_empty(), - "a resolved table is not queued to resolve again" - ); + for (table_modes, expected) in [ + (HashMap::new(), CaptureMode::Full), + ( + HashMap::from([("dfe.events".to_string(), CaptureMode::JsonOnly)]), + CaptureMode::JsonOnly, + ), + ] { + for later in ["", "@retention: 30d"] { + let metadata_config = MetadataConfig { + table_capture_modes: table_modes.clone(), + ..Default::default() + }; + let mut overrides = CaptureOverrides::new(&metadata_config); + overrides.update_from_comment("dfe.events", "@capture_mode: extracted_only"); + assert_eq!( + overrides.derive_config("dfe.events").mode, + CaptureMode::ExtractedOnly + ); + overrides.update_from_comment("dfe.events", later); + assert_eq!( + overrides.derive_config("dfe.events").mode, + expected, + "after {later:?}" + ); + overrides.mark_pending("dfe.events"); + assert!( + overrides.take_pending().is_empty(), + "a resolved table is not queued to resolve again" + ); + } } } @@ -439,114 +336,6 @@ mod tests { ); } - // ---- Coverage gap: legacy raw_disabled only falls through to global mode ---- - - #[test] - fn test_legacy_disable_raw_only_falls_through_to_global_full() { - // disable_raw_tables has the table, disable_json_tables does not - // Global mode is Full (default) — (false, true) match arm → global_mode - let mut metadata_config = MetadataConfig::default(); - metadata_config - .disable_raw_tables - .push("dfe.events".to_string()); - let overrides = CaptureOverrides::new(&metadata_config); - - // Hits the (false, true) legacy match arm → global_mode (Full) - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::Full - ); - } - - #[test] - fn test_legacy_disable_raw_only_falls_through_to_global_extracted() { - // Same path but different global mode to prove the arm delegates to global - let mut metadata_config = MetadataConfig { - capture_mode: CaptureMode::ExtractedOnly, - ..Default::default() - }; - metadata_config - .disable_raw_tables - .push("dfe.events".to_string()); - let overrides = CaptureOverrides::new(&metadata_config); - - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::ExtractedOnly - ); - } - - // ---- Coverage gap: DDL @no_capture_raw alone falls through to global ---- - - #[test] - fn test_ddl_no_capture_raw_only_uses_global_mode() { - // Global is RawOnly — DDL @no_capture_raw alone should yield global (RawOnly) - let metadata_config = MetadataConfig { - capture_mode: CaptureMode::RawOnly, - ..Default::default() - }; - let mut overrides = CaptureOverrides::new(&metadata_config); - - overrides.update_from_comment("dfe.events", "@no_capture_raw: true"); - - // Hits (false, true) → global_mode (RawOnly) - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::RawOnly - ); - } - - #[test] - fn test_ddl_no_capture_raw_only_default_global_is_full() { - let metadata_config = MetadataConfig::default(); - let mut overrides = CaptureOverrides::new(&metadata_config); - - overrides.update_from_comment("dfe.events", "@no_capture_raw: true"); - - // Hits (false, true) → global_mode (Full) - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::Full - ); - } - - // ---- Coverage gap: DDL legacy no_capture_json and no_capture_raw together ---- - - #[test] - fn test_ddl_legacy_both_no_capture_maps_to_extracted_only() { - let metadata_config = MetadataConfig::default(); - let mut overrides = CaptureOverrides::new(&metadata_config); - - overrides.update_from_comment( - "dfe.events", - "@no_capture_json: true | @no_capture_raw: true", - ); - - // Hits (true, true) → ExtractedOnly - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::ExtractedOnly - ); - } - - #[test] - fn test_ddl_legacy_tags_false_values_noop() { - // @no_capture_json: false → is_some_and(|v| v == "true") returns false - // Both legacy flags false → update does nothing, global default applies - let metadata_config = MetadataConfig::default(); - let mut overrides = CaptureOverrides::new(&metadata_config); - - overrides.update_from_comment( - "dfe.events", - "@no_capture_json: false | @no_capture_raw: false", - ); - - assert_eq!( - overrides.derive_config("dfe.events").mode, - CaptureMode::Full - ); - } - #[test] fn test_ddl_update_full_capture_mode() { // Ensure the "full" variant in update_from_comment is exercised @@ -587,16 +376,6 @@ mod tests { assert_eq!(pending.len(), 3); } - #[test] - fn test_deprecation_warning_emitted_on_legacy_config() { - // Just ensures Self::new doesn't panic when deprecated fields are populated. - // The warn! is a side effect; we don't capture logs here. - let mut metadata_config = MetadataConfig::default(); - metadata_config.disable_json_tables.push("t1".to_string()); - metadata_config.disable_raw_tables.push("t2".to_string()); - let _overrides = CaptureOverrides::new(&metadata_config); - } - #[test] fn test_unknown_capture_mode_preserves_existing_config() { // If update_from_comment gets an unknown value AFTER a valid one,