From 28fa412a06371afd03dd22922c2da2adf19f60c8 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 28 Aug 2026 06:16:03 +0800 Subject: [PATCH] refactor(ecstore): extract shared audit/notify KVS table constructors (#6756) The audit and notify subsystems each declare their own default KVS table for the same nine delivery targets. For amqp, nats, pulsar, postgres and kafka the two declarations are byte-identical; for redis and mysql they differ only in a single default literal (the pub/sub channel and the destination table). Keeping two copies means every default or key-order change has to be made twice, and a missed edit silently changes what admin config reports for one subsystem only. Add `config::target_defaults` with one constructor per shared table, taking the diverging literal as a parameter for redis and mysql, plus a small `kv` helper that replaces the repeated `KV { .. }` literals. Key order is reproduced exactly because it drives the order the admin API lists keys in. Unit tests pin the full ordered key/value/hidden_if_empty triple of every table against hard-coded literals, and cover both the audit and the notify literal for the two parameterized tables. Webhook and mqtt are deliberately left out: audit's webhook table carries extra batching and retry keys, both webhook tables disagree on key order and on the auth-token hidden_if_empty flag, and mqtt disagrees on qos, keep-alive interval and reconnect interval. Those are real behavioral forks, not duplication, so they stay declared in place. This is the expand step only. Nothing calls the new module yet, so audit.rs and notify.rs are untouched and no default changes; the constructors carry an item-level allow(dead_code) until the migrate step points both files at them. Refs rustfs/backlog#2044 --- crates/ecstore/src/config/mod.rs | 1 + crates/ecstore/src/config/target_defaults.rs | 414 +++++++++++++++++++ 2 files changed, 415 insertions(+) create mode 100644 crates/ecstore/src/config/target_defaults.rs diff --git a/crates/ecstore/src/config/mod.rs b/crates/ecstore/src/config/mod.rs index ff97d03e0..9f029cfb9 100644 --- a/crates/ecstore/src/config/mod.rs +++ b/crates/ecstore/src/config/mod.rs @@ -21,6 +21,7 @@ mod notify; mod oidc; mod scanner; pub mod storageclass; +mod target_defaults; use crate::error::Result; use crate::store::ECStore; diff --git a/crates/ecstore/src/config/target_defaults.rs b/crates/ecstore/src/config/target_defaults.rs new file mode 100644 index 000000000..d75980f86 --- /dev/null +++ b/crates/ecstore/src/config/target_defaults.rs @@ -0,0 +1,414 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Shared default KVS tables for delivery targets that audit and notify declare identically. +//! +//! The audit and notify subsystems register one default KVS per delivery target. For amqp, nats, +//! pulsar, postgres and kafka both sides declare byte-identical tables; for redis and mysql they +//! differ only in a single default literal, which the caller passes in. +//! +//! Webhook and mqtt are deliberately absent: audit's webhook table carries extra batching/retry +//! keys and both tables disagree on key order and on several defaults (mqtt qos, keep-alive and +//! reconnect intervals), so they are real behavioral forks, not duplication. +//! +//! Key order is part of the contract: it drives the order admin config output lists the keys in, +//! so every constructor reproduces the existing order exactly. + +use rustfs_config::server_config::{KV, KVS}; +use rustfs_config::{ + AMQP_EXCHANGE, AMQP_MANDATORY, AMQP_PASSWORD, AMQP_PERSISTENT, AMQP_QUEUE_DIR, AMQP_QUEUE_LIMIT, AMQP_ROUTING_KEY, + AMQP_TLS_CA, AMQP_TLS_CLIENT_CERT, AMQP_TLS_CLIENT_KEY, AMQP_URL, AMQP_USERNAME, COMMENT_KEY, DEFAULT_LIMIT, ENABLE_KEY, + EVENT_DEFAULT_DIR, EnableState, KAFKA_ACKS, KAFKA_BROKERS, KAFKA_QUEUE_DIR, KAFKA_QUEUE_LIMIT, KAFKA_SASL_ENABLE, + KAFKA_SASL_MECHANISM, KAFKA_SASL_PASSWORD, KAFKA_SASL_USERNAME, KAFKA_TLS_CA, KAFKA_TLS_CLIENT_CERT, KAFKA_TLS_CLIENT_KEY, + KAFKA_TLS_ENABLE, KAFKA_TOPIC, MYSQL_DSN_STRING, MYSQL_FORMAT, MYSQL_MAX_OPEN_CONNECTIONS, MYSQL_QUEUE_DIR, + MYSQL_QUEUE_LIMIT, MYSQL_TABLE, MYSQL_TLS_CA, MYSQL_TLS_CLIENT_CERT, MYSQL_TLS_CLIENT_KEY, NATS_ADDRESS, + NATS_CREDENTIALS_FILE, NATS_JETSTREAM_ACK_TIMEOUT_DEFAULT_SECS, NATS_JETSTREAM_ACK_TIMEOUT_SECS, NATS_JETSTREAM_ENABLE, + NATS_JETSTREAM_STREAM_NAME, NATS_PASSWORD, NATS_QUEUE_DIR, NATS_QUEUE_LIMIT, NATS_SUBJECT, NATS_TLS_CA, NATS_TLS_CLIENT_CERT, + NATS_TLS_CLIENT_KEY, NATS_TLS_REQUIRED, NATS_TOKEN, NATS_USERNAME, POSTGRES_DSN_STRING, POSTGRES_FORMAT, POSTGRES_QUEUE_DIR, + POSTGRES_QUEUE_LIMIT, POSTGRES_TABLE, POSTGRES_TLS_CA, POSTGRES_TLS_CLIENT_CERT, POSTGRES_TLS_CLIENT_KEY, + POSTGRES_TLS_REQUIRED, PULSAR_AUTH_TOKEN, PULSAR_BROKER, PULSAR_PASSWORD, PULSAR_QUEUE_DIR, PULSAR_QUEUE_LIMIT, + PULSAR_TLS_ALLOW_INSECURE, PULSAR_TLS_CA, PULSAR_TLS_HOSTNAME_VERIFICATION, PULSAR_TOPIC, PULSAR_USERNAME, REDIS_CHANNEL, + REDIS_CONNECTION_TIMEOUT, REDIS_KEEP_ALIVE_INTERVAL, REDIS_MAX_RETRY_ATTEMPTS, REDIS_MAX_RETRY_DELAY, REDIS_MIN_RETRY_DELAY, + REDIS_PASSWORD, REDIS_PIPELINE_BUFFER_SIZE, REDIS_QUEUE_DIR, REDIS_QUEUE_LIMIT, REDIS_RECONNECT_RETRY_ATTEMPTS, + REDIS_RESPONSE_TIMEOUT, REDIS_TLS_ALLOW_INSECURE, REDIS_TLS_CA, REDIS_TLS_CLIENT_CERT, REDIS_TLS_CLIENT_KEY, + REDIS_TLS_POLICY, REDIS_URL, REDIS_USERNAME, +}; + +/// Builds one default entry. `hidden_if_empty` marks values the admin API elides when unset. +fn kv(key: &str, value: impl Into, hidden_if_empty: bool) -> KV { + KV { + key: key.to_owned(), + value: value.into(), + hidden_if_empty, + } +} + +/// Default KVS for the amqp delivery target. +// Unused until the audit/notify tables are migrated onto these constructors. +#[allow(dead_code)] +pub fn amqp_kvs() -> KVS { + KVS(vec![ + kv(ENABLE_KEY, EnableState::Off.to_string(), false), + kv(AMQP_URL, "", false), + kv(AMQP_EXCHANGE, "", false), + kv(AMQP_ROUTING_KEY, "", false), + kv(AMQP_MANDATORY, EnableState::Off.to_string(), false), + kv(AMQP_PERSISTENT, EnableState::On.to_string(), false), + kv(AMQP_USERNAME, "", false), + kv(AMQP_PASSWORD, "", true), + kv(AMQP_TLS_CA, "", true), + kv(AMQP_TLS_CLIENT_CERT, "", true), + kv(AMQP_TLS_CLIENT_KEY, "", true), + kv(AMQP_QUEUE_DIR, EVENT_DEFAULT_DIR, false), + kv(AMQP_QUEUE_LIMIT, DEFAULT_LIMIT.to_string(), false), + kv(COMMENT_KEY, "", false), + ]) +} + +/// Default KVS for the nats delivery target. +// Unused until the audit/notify tables are migrated onto these constructors. +#[allow(dead_code)] +pub fn nats_kvs() -> KVS { + KVS(vec![ + kv(ENABLE_KEY, EnableState::Off.to_string(), false), + kv(NATS_ADDRESS, "", false), + kv(NATS_SUBJECT, "", false), + kv(NATS_USERNAME, "", false), + kv(NATS_PASSWORD, "", true), + kv(NATS_TOKEN, "", true), + kv(NATS_CREDENTIALS_FILE, "", true), + kv(NATS_TLS_CA, "", true), + kv(NATS_TLS_CLIENT_CERT, "", true), + kv(NATS_TLS_CLIENT_KEY, "", true), + kv(NATS_TLS_REQUIRED, EnableState::Off.to_string(), false), + kv(NATS_QUEUE_DIR, EVENT_DEFAULT_DIR, false), + kv(NATS_QUEUE_LIMIT, DEFAULT_LIMIT.to_string(), false), + kv(NATS_JETSTREAM_ENABLE, EnableState::Off.to_string(), false), + kv(NATS_JETSTREAM_STREAM_NAME, "", false), + kv( + NATS_JETSTREAM_ACK_TIMEOUT_SECS, + NATS_JETSTREAM_ACK_TIMEOUT_DEFAULT_SECS.to_string(), + false, + ), + kv(COMMENT_KEY, "", false), + ]) +} + +/// Default KVS for the pulsar delivery target. +// Unused until the audit/notify tables are migrated onto these constructors. +#[allow(dead_code)] +pub fn pulsar_kvs() -> KVS { + KVS(vec![ + kv(ENABLE_KEY, EnableState::Off.to_string(), false), + kv(PULSAR_BROKER, "", false), + kv(PULSAR_TOPIC, "", false), + kv(PULSAR_AUTH_TOKEN, "", true), + kv(PULSAR_USERNAME, "", false), + kv(PULSAR_PASSWORD, "", true), + kv(PULSAR_TLS_CA, "", true), + kv(PULSAR_TLS_ALLOW_INSECURE, EnableState::Off.to_string(), false), + kv(PULSAR_TLS_HOSTNAME_VERIFICATION, EnableState::On.to_string(), false), + kv(PULSAR_QUEUE_DIR, EVENT_DEFAULT_DIR, false), + kv(PULSAR_QUEUE_LIMIT, DEFAULT_LIMIT.to_string(), false), + kv(COMMENT_KEY, "", false), + ]) +} + +/// Default KVS for the postgres delivery target. +// Unused until the audit/notify tables are migrated onto these constructors. +#[allow(dead_code)] +pub fn postgres_kvs() -> KVS { + KVS(vec![ + kv(ENABLE_KEY, EnableState::Off.to_string(), false), + kv(POSTGRES_DSN_STRING, "", true), + kv(POSTGRES_TABLE, "", false), + kv(POSTGRES_FORMAT, "namespace", false), + kv(POSTGRES_TLS_REQUIRED, EnableState::Off.to_string(), false), + kv(POSTGRES_TLS_CA, "", true), + kv(POSTGRES_TLS_CLIENT_CERT, "", true), + kv(POSTGRES_TLS_CLIENT_KEY, "", true), + kv(POSTGRES_QUEUE_DIR, EVENT_DEFAULT_DIR, false), + kv(POSTGRES_QUEUE_LIMIT, DEFAULT_LIMIT.to_string(), false), + kv(COMMENT_KEY, "", false), + ]) +} + +/// Default KVS for the kafka delivery target. +// Unused until the audit/notify tables are migrated onto these constructors. +#[allow(dead_code)] +pub fn kafka_kvs() -> KVS { + KVS(vec![ + kv(ENABLE_KEY, EnableState::Off.to_string(), false), + kv(KAFKA_BROKERS, "", false), + kv(KAFKA_TOPIC, "", false), + kv(KAFKA_ACKS, "1", false), + kv(KAFKA_TLS_ENABLE, EnableState::Off.to_string(), false), + kv(KAFKA_TLS_CA, "", true), + kv(KAFKA_TLS_CLIENT_CERT, "", true), + kv(KAFKA_TLS_CLIENT_KEY, "", true), + kv(KAFKA_SASL_ENABLE, EnableState::Off.to_string(), false), + kv(KAFKA_SASL_MECHANISM, "", false), + kv(KAFKA_SASL_USERNAME, "", false), + kv(KAFKA_SASL_PASSWORD, "", true), + kv(KAFKA_QUEUE_DIR, EVENT_DEFAULT_DIR, false), + kv(KAFKA_QUEUE_LIMIT, DEFAULT_LIMIT.to_string(), false), + kv(COMMENT_KEY, "", false), + ]) +} + +/// Default KVS for the redis delivery target. `channel` is the subsystem's default pub/sub channel, +/// which is the only value audit and notify disagree on. +// Unused until the audit/notify tables are migrated onto these constructors. +#[allow(dead_code)] +pub fn redis_kvs(channel: &str) -> KVS { + KVS(vec![ + kv(ENABLE_KEY, EnableState::Off.to_string(), false), + kv(REDIS_URL, "", false), + kv(REDIS_CHANNEL, channel, false), + kv(REDIS_USERNAME, "", false), + kv(REDIS_PASSWORD, "", true), + kv(REDIS_KEEP_ALIVE_INTERVAL, "15", false), + kv(REDIS_QUEUE_DIR, EVENT_DEFAULT_DIR, false), + kv(REDIS_QUEUE_LIMIT, DEFAULT_LIMIT.to_string(), false), + kv(REDIS_MAX_RETRY_ATTEMPTS, "3", false), + kv(REDIS_RECONNECT_RETRY_ATTEMPTS, "", false), + kv(REDIS_MIN_RETRY_DELAY, "", false), + kv(REDIS_MAX_RETRY_DELAY, "", false), + kv(REDIS_CONNECTION_TIMEOUT, "", false), + kv(REDIS_RESPONSE_TIMEOUT, "", false), + kv(REDIS_PIPELINE_BUFFER_SIZE, "", false), + kv(REDIS_TLS_POLICY, "", true), + kv(REDIS_TLS_CA, "", true), + kv(REDIS_TLS_CLIENT_CERT, "", true), + kv(REDIS_TLS_CLIENT_KEY, "", true), + kv(REDIS_TLS_ALLOW_INSECURE, EnableState::Off.to_string(), false), + kv(COMMENT_KEY, "", false), + ]) +} + +/// Default KVS for the mysql delivery target. `table` is the subsystem's default destination table, +/// which is the only value audit and notify disagree on. +// Unused until the audit/notify tables are migrated onto these constructors. +#[allow(dead_code)] +pub fn mysql_kvs(table: &str) -> KVS { + KVS(vec![ + kv(ENABLE_KEY, EnableState::Off.to_string(), false), + kv(MYSQL_DSN_STRING, "", true), + kv(MYSQL_TABLE, table, false), + kv(MYSQL_FORMAT, "access", false), + kv(MYSQL_TLS_CA, "", true), + kv(MYSQL_TLS_CLIENT_CERT, "", true), + kv(MYSQL_TLS_CLIENT_KEY, "", true), + kv(MYSQL_QUEUE_DIR, EVENT_DEFAULT_DIR, false), + kv(MYSQL_QUEUE_LIMIT, DEFAULT_LIMIT.to_string(), false), + kv(MYSQL_MAX_OPEN_CONNECTIONS, "2", false), + kv(COMMENT_KEY, "", false), + ]) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Expected values are spelled out as literals on purpose: they mirror the tables currently + /// declared in `audit.rs` and `notify.rs`, so a drift in key order or in any default breaks + /// the test instead of silently changing admin config output. + fn assert_table(actual: &KVS, expected: &[(&str, &str, bool)]) { + let actual: Vec<(&str, &str, bool)> = actual + .0 + .iter() + .map(|kv| (kv.key.as_str(), kv.value.as_str(), kv.hidden_if_empty)) + .collect(); + assert_eq!(actual, expected); + } + + const QUEUE_DIR: &str = "/opt/rustfs/events"; + const QUEUE_LIMIT: &str = "100000"; + + #[test] + fn amqp_table_matches_audit_and_notify() { + assert_table( + &amqp_kvs(), + &[ + ("enable", "off", false), + ("url", "", false), + ("exchange", "", false), + ("routing_key", "", false), + ("mandatory", "off", false), + ("persistent", "on", false), + ("username", "", false), + ("password", "", true), + ("tls_ca", "", true), + ("tls_client_cert", "", true), + ("tls_client_key", "", true), + ("queue_dir", QUEUE_DIR, false), + ("queue_limit", QUEUE_LIMIT, false), + ("comment", "", false), + ], + ); + } + + #[test] + fn nats_table_matches_audit_and_notify() { + assert_table( + &nats_kvs(), + &[ + ("enable", "off", false), + ("address", "", false), + ("subject", "", false), + ("username", "", false), + ("password", "", true), + ("token", "", true), + ("credentials_file", "", true), + ("tls_ca", "", true), + ("tls_client_cert", "", true), + ("tls_client_key", "", true), + ("tls_required", "off", false), + ("queue_dir", QUEUE_DIR, false), + ("queue_limit", QUEUE_LIMIT, false), + ("jetstream_enable", "off", false), + ("jetstream_stream_name", "", false), + ("jetstream_ack_timeout_secs", "30", false), + ("comment", "", false), + ], + ); + } + + #[test] + fn pulsar_table_matches_audit_and_notify() { + assert_table( + &pulsar_kvs(), + &[ + ("enable", "off", false), + ("broker", "", false), + ("topic", "", false), + ("auth_token", "", true), + ("username", "", false), + ("password", "", true), + ("tls_ca", "", true), + ("tls_allow_insecure", "off", false), + ("tls_hostname_verification", "on", false), + ("queue_dir", QUEUE_DIR, false), + ("queue_limit", QUEUE_LIMIT, false), + ("comment", "", false), + ], + ); + } + + #[test] + fn postgres_table_matches_audit_and_notify() { + assert_table( + &postgres_kvs(), + &[ + ("enable", "off", false), + ("dsn_string", "", true), + ("table", "", false), + ("format", "namespace", false), + ("tls_required", "off", false), + ("tls_ca", "", true), + ("tls_client_cert", "", true), + ("tls_client_key", "", true), + ("queue_dir", QUEUE_DIR, false), + ("queue_limit", QUEUE_LIMIT, false), + ("comment", "", false), + ], + ); + } + + #[test] + fn kafka_table_matches_audit_and_notify() { + assert_table( + &kafka_kvs(), + &[ + ("enable", "off", false), + ("brokers", "", false), + ("topic", "", false), + ("acks", "1", false), + ("tls_enable", "off", false), + ("tls_ca", "", true), + ("tls_client_cert", "", true), + ("tls_client_key", "", true), + ("sasl_enable", "off", false), + ("sasl_mechanism", "", false), + ("sasl_username", "", false), + ("sasl_password", "", true), + ("queue_dir", QUEUE_DIR, false), + ("queue_limit", QUEUE_LIMIT, false), + ("comment", "", false), + ], + ); + } + + fn expected_redis(channel: &str) -> Vec<(&str, &str, bool)> { + vec![ + ("enable", "off", false), + ("url", "", false), + ("channel", channel, false), + ("username", "", false), + ("password", "", true), + ("keep_alive_interval", "15", false), + ("queue_dir", QUEUE_DIR, false), + ("queue_limit", QUEUE_LIMIT, false), + ("max_retry_attempts", "3", false), + ("reconnect_retry_attempts", "", false), + ("min_retry_delay", "", false), + ("max_retry_delay", "", false), + ("connection_timeout", "", false), + ("response_timeout", "", false), + ("pipeline_buffer_size", "", false), + ("tls_policy", "", true), + ("tls_ca", "", true), + ("tls_client_cert", "", true), + ("tls_client_key", "", true), + ("tls_allow_insecure", "off", false), + ("comment", "", false), + ] + } + + #[test] + fn redis_table_matches_audit() { + assert_table(&redis_kvs("rustfs_audit_channel"), &expected_redis("rustfs_audit_channel")); + } + + #[test] + fn redis_table_matches_notify() { + assert_table(&redis_kvs("rustfs_notify_channel"), &expected_redis("rustfs_notify_channel")); + } + + fn expected_mysql(table: &str) -> Vec<(&str, &str, bool)> { + vec![ + ("enable", "off", false), + ("dsn_string", "", true), + ("table", table, false), + ("format", "access", false), + ("tls_ca", "", true), + ("tls_client_cert", "", true), + ("tls_client_key", "", true), + ("queue_dir", QUEUE_DIR, false), + ("queue_limit", QUEUE_LIMIT, false), + ("max_open_connections", "2", false), + ("comment", "", false), + ] + } + + #[test] + fn mysql_table_matches_audit() { + assert_table(&mysql_kvs("rustfs_audit_logs"), &expected_mysql("rustfs_audit_logs")); + } + + #[test] + fn mysql_table_matches_notify() { + assert_table(&mysql_kvs("rustfs_events"), &expected_mysql("rustfs_events")); + } +}