diff --git a/crates/targets/src/config/loader.rs b/crates/targets/src/config/loader.rs index a47646351..89d24a5dd 100644 --- a/crates/targets/src/config/loader.rs +++ b/crates/targets/src/config/loader.rs @@ -37,6 +37,22 @@ pub fn try_collect_target_configs( try_collect_target_configs_from_env(config, route_prefix, target_type, valid_fields, std::env::vars()) } +/// Collects enabled target instance configs with per-instance fault isolation. +/// +/// Unlike [`try_collect_target_configs`], an instance with an invalid +/// configuration (for example an unparseable `enable` value) does not abort the +/// whole target type. The enabled `(id, config)` pairs are returned alongside a +/// summary line for every instance that could not be collected, so callers can +/// load the healthy instances while still surfacing the failed ones. +pub fn collect_target_config_results( + config: &Config, + route_prefix: &str, + target_type: &str, + valid_fields: &HashSet, +) -> (Vec<(String, KVS)>, Vec) { + collect_target_config_results_from_env(config, route_prefix, target_type, valid_fields, std::env::vars()) +} + fn is_sensitive_target_field(field_name: &str) -> bool { let field_name = field_name.to_ascii_lowercase(); field_name.contains("password") @@ -171,6 +187,35 @@ where .collect()) } +pub fn collect_target_config_results_from_env( + config: &Config, + route_prefix: &str, + target_type: &str, + valid_fields: &HashSet, + env_vars: I, +) -> (Vec<(String, KVS)>, Vec) +where + I: IntoIterator, +{ + let mut configs = Vec::new(); + let mut failures = Vec::new(); + for result in collect_merged_target_config_results_from_env( + config, + &format!("{route_prefix}{target_type}").to_lowercase(), + route_prefix, + target_type, + valid_fields, + env_vars, + ) { + match result { + Ok(record) if record.enabled => configs.push((record.instance_id, record.effective_config)), + Ok(_) => {} + Err((record, err)) => failures.push(format!("{target_type}/{}: {err}", record.instance_id)), + } + } + (configs, failures) +} + pub(crate) fn collect_merged_target_configs_from_env( config: &Config, section_name: &str, @@ -330,8 +375,8 @@ where #[cfg(test)] mod tests { use super::{ - collect_env_target_instance_ids_from_env, collect_target_configs_from_env, redact_target_field_value, - redacted_target_config, try_collect_target_configs_from_env, + collect_env_target_instance_ids_from_env, collect_target_config_results_from_env, collect_target_configs_from_env, + redact_target_field_value, redacted_target_config, try_collect_target_configs_from_env, }; use crate::TargetError; use rustfs_config::notify::{ @@ -516,6 +561,48 @@ mod tests { assert_eq!(configs[0].0, "good"); } + #[test] + fn collect_target_config_results_isolates_invalid_instance_and_reports_it() { + let (configs, failures) = collect_target_config_results_from_env( + &Config(HashMap::new()), + NOTIFY_ROUTE_PREFIX, + "webhook", + &HashSet::from([ENABLE_KEY.to_string(), WEBHOOK_ENDPOINT.to_string()]), + vec![ + ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_GOOD".to_string(), "on".to_string()), + ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_GOOD".to_string(), "https://example.com/good".to_string()), + // "enable" is a common typo: EnableState accepts "enabled"/"on", not "enable". + ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_BAD".to_string(), "enable".to_string()), + ], + ); + + // The healthy instance still loads even though a sibling is malformed... + assert_eq!(configs.len(), 1); + assert_eq!(configs[0].0, "good"); + // ...and the malformed instance is surfaced (not silently dropped, not + // aborting the whole target type) so callers can fail-report it. + assert_eq!(failures.len(), 1); + assert!( + failures[0].contains("webhook/bad") && failures[0].contains("enable"), + "unexpected failure summary: {}", + failures[0] + ); + } + + #[test] + fn collect_target_config_results_does_not_report_validly_disabled_instances() { + let (configs, failures) = collect_target_config_results_from_env( + &Config(HashMap::new()), + NOTIFY_ROUTE_PREFIX, + "webhook", + &HashSet::from([ENABLE_KEY.to_string(), WEBHOOK_ENDPOINT.to_string()]), + vec![("RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY".to_string(), "off".to_string())], + ); + + assert!(configs.is_empty()); + assert!(failures.is_empty(), "a validly disabled instance is not a failure"); + } + #[test] fn collect_target_configs_preserves_whitespace_padded_legacy_value() { let configs = try_collect_target_configs_from_env( diff --git a/crates/targets/src/config/mod.rs b/crates/targets/src/config/mod.rs index df083edec..843ec7762 100644 --- a/crates/targets/src/config/mod.rs +++ b/crates/targets/src/config/mod.rs @@ -24,8 +24,9 @@ pub use instance::{ try_normalize_target_plugin_instances, try_normalize_target_plugin_instances_from_env, }; pub use loader::{ - collect_env_target_instance_ids, collect_env_target_instance_ids_from_env, collect_target_configs, - collect_target_configs_from_env, try_collect_target_configs, try_collect_target_configs_from_env, + collect_env_target_instance_ids, collect_env_target_instance_ids_from_env, collect_target_config_results, + collect_target_config_results_from_env, collect_target_configs, collect_target_configs_from_env, try_collect_target_configs, + try_collect_target_configs_from_env, }; pub use target_args::{ build_amqp_args, build_kafka_args, build_mqtt_args, build_mysql_args, build_nats_args, build_postgres_args, diff --git a/crates/targets/src/plugin.rs b/crates/targets/src/plugin.rs index a95d414ce..6c72590a7 100644 --- a/crates/targets/src/plugin.rs +++ b/crates/targets/src/plugin.rs @@ -14,7 +14,7 @@ use crate::{ PluginRuntimeAdapter, RuntimeActivation, Target, TargetError, - config::try_collect_target_configs, + config::collect_target_config_results, manifest::{TargetPluginManifest, builtin_target_manifest}, target::with_deferred_queue_store_open, }; @@ -347,7 +347,16 @@ where for (target_type, plugin) in &self.plugins { info!(target_type = %target_type, "Start working on target type"); - for (id, merged_config) in try_collect_target_configs(config, route_prefix, target_type, plugin.valid_fields_set())? { + // Per-instance fault isolation: an invalid instance (e.g. an + // unparseable `enable` value) is recorded as a failure and skipped, + // never aborting the remaining instances or other target types. + let (collected, invalid_instances) = + collect_target_config_results(config, route_prefix, target_type, plugin.valid_fields_set()); + for detail in invalid_instances { + error!(target_type = %target_type, reason = "invalid_config", detail = %detail, "Skipping target instance with invalid configuration"); + failures.push(detail); + } + for (id, merged_config) in collected { info!(target_type = %target_type, instance_id = %id, "Target is enabled, ready to create"); let created = if defer_store_open { with_deferred_queue_store_open(|| self.create_target(target_type, id.clone(), &merged_config)) @@ -505,4 +514,61 @@ mod tests { assert_eq!(activation.targets[0].id().to_string(), "primary:test"); assert!(activation.replay_workers.is_empty()); } + + // Regression: a single instance with a malformed `enable` value must not + // abort the remaining instances or unrelated target types. Before this fix + // the collector short-circuited the whole create path, so one typo took + // down every notify/audit target. + #[tokio::test] + async fn create_dormant_isolates_invalid_enable_and_still_loads_other_targets() { + let mut registry = TargetPluginRegistry::::new(); + for target_type in ["alpha", "beta"] { + registry.register(TargetPluginDescriptor::new( + target_type, + &[ENABLE_KEY, "endpoint"], + |_config| Ok(()), + move |id, _config| { + Ok(Box::new(TestTarget { + id: crate::arn::TargetID::new(id, target_type.to_string()), + })) + }, + )); + } + + let mut cfg = Config(HashMap::new()); + + // alpha: one healthy instance plus one with a malformed `enable` value + // ("enable" is a typo -- EnableState accepts "enabled"/"on", not "enable"). + let mut alpha = HashMap::new(); + let mut alpha_good = KVS::new(); + alpha_good.insert(ENABLE_KEY.to_string(), "on".to_string()); + alpha_good.insert("endpoint".to_string(), "https://example.com/alpha".to_string()); + alpha.insert("good".to_string(), alpha_good); + let mut alpha_bad = KVS::new(); + alpha_bad.insert(ENABLE_KEY.to_string(), "enable".to_string()); + alpha.insert("bad".to_string(), alpha_bad); + cfg.0.insert("notify_alpha".to_string(), alpha); + + // beta: a healthy instance in a different target type must survive. + let mut beta = HashMap::new(); + let mut beta_primary = KVS::new(); + beta_primary.insert(ENABLE_KEY.to_string(), "on".to_string()); + beta_primary.insert("endpoint".to_string(), "https://example.com/beta".to_string()); + beta.insert("primary".to_string(), beta_primary); + cfg.0.insert("notify_beta".to_string(), beta); + + let (targets, failures) = registry + .create_dormant_targets_from_config(&cfg, "notify_") + .await + .expect("a malformed instance must not abort target creation"); + + let mut created: Vec = targets.iter().map(|target| target.id().to_string()).collect(); + created.sort(); + assert_eq!(created, vec!["good:alpha".to_string(), "primary:beta".to_string()]); + + // The malformed instance is surfaced (so an Admin write can't report a + // false success) rather than silently dropped or fatally aborting. + assert_eq!(failures.len(), 1); + assert!(failures[0].contains("alpha/bad"), "unexpected failure summary: {}", failures[0]); + } }