feat: add console-managed audit and notify module switches (#2690)

rustfs/src/admin/handlers/audit.rs

@@ -17,14 +17,16 @@ use crate::admin::{
     handlers::target_descriptor::{
         AdminTargetSpec, AdminTargetValidator, EndpointKey, TargetDomain, TargetEndpointSource, allowed_target_keys,
         collect_validated_key_values as shared_collect_validated_key_values,
-        merge_target_endpoints as shared_merge_target_endpoints,
+        merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason,
         target_mutation_block_reason as shared_target_mutation_block_reason, target_service_name, target_spec,
         validate_target_request,
     },
     router::{AdminOperation, Operation, S3Router},
 };
 use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::{
+    ADMIN_PREFIX, RemoteAddr, is_audit_module_enabled, refresh_audit_module_enabled, refresh_persisted_module_switches_from_store,
+};
 use futures::stream::{FuturesUnordered, StreamExt};
 use http::{HeaderMap, StatusCode};
 use hyper::Method;
@@ -168,6 +170,17 @@ fn audit_target_mutation_block_reason(config: &Config, target_type: &str, target
     )
 }
 
+async fn audit_target_operation_block_reason(action: &str) -> Option<String> {
+    if let Err(err) = refresh_persisted_module_switches_from_store().await {
+        warn!(
+            error = %err,
+            "failed to reload persisted module switches before checking audit target operation gating"
+        );
+    }
+    refresh_audit_module_enabled();
+    target_module_disabled_reason("audit", rustfs_config::ENV_AUDIT_ENABLE, is_audit_module_enabled(), action)
+}
+
 fn merge_audit_endpoints(config: &Config, runtime_statuses: HashMap<EndpointKey, String>) -> Vec<AuditEndpoint> {
     shared_merge_target_endpoints(&audit_target_specs(), AUDIT_ROUTE_PREFIX, config, runtime_statuses)
         .into_iter()
@@ -272,6 +285,9 @@ impl Operation for AuditTargetConfig {
         let (target_type, target_name) = extract_target_params(&params)?;
 
         authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
+        if let Some(reason) = audit_target_operation_block_reason("managing audit targets from the console").await {
+            return Err(s3_error!(InvalidRequest, "{reason}"));
+        }
         let config_snapshot = load_server_config_from_store().await?;
         if let Some(reason) = audit_target_mutation_block_reason(&config_snapshot, target_type, target_name) {
             return Err(s3_error!(InvalidRequest, "{reason}"));
@@ -373,6 +389,9 @@ impl Operation for RemoveAuditTarget {
         let (target_type, target_name) = extract_target_params(&params)?;
 
         authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
+        if let Some(reason) = audit_target_operation_block_reason("managing audit targets from the console").await {
+            return Err(s3_error!(InvalidRequest, "{reason}"));
+        }
         let config_snapshot = load_server_config_from_store().await?;
         if let Some(reason) = audit_target_mutation_block_reason(&config_snapshot, target_type, target_name) {
             return Err(s3_error!(InvalidRequest, "{reason}"));
@@ -534,6 +553,26 @@ mod tests {
         );
     }
 
+    #[test]
+    fn audit_target_operation_block_reason_requires_audit_module_enable() {
+        with_var(rustfs_config::ENV_AUDIT_ENABLE, Some("false"), || {
+            let reason =
+                futures::executor::block_on(audit_target_operation_block_reason("managing audit targets from the console"));
+            assert!(reason.is_some());
+            assert!(reason.unwrap().contains("set RUSTFS_AUDIT_ENABLE=true"));
+        });
+    }
+
+    #[test]
+    fn audit_target_operation_block_reason_allows_when_audit_module_enabled() {
+        with_var(rustfs_config::ENV_AUDIT_ENABLE, Some("true"), || {
+            assert!(
+                futures::executor::block_on(audit_target_operation_block_reason("managing audit targets from the console"))
+                    .is_none()
+            );
+        });
+    }
+
     #[test]
     fn audit_target_mutation_block_reason_rejects_mixed_target() {
         with_var("RUSTFS_AUDIT_WEBHOOK_ENDPOINT_PRIMARY", Some("https://example.com/hook"), || {
@@ -727,6 +766,10 @@ mod tests {
             put_block.contains("authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"),
             "audit target writes should require SetBucketTargetAction"
         );
+        assert!(
+            put_block.contains("audit_target_operation_block_reason(\"managing audit targets from the console\")"),
+            "audit target writes should reject requests when the audit module is disabled"
+        );
         assert!(
             list_block.contains("authorize_audit_admin_request(&req, AdminAction::GetBucketTargetAction).await?;"),
             "audit target list should require GetBucketTargetAction"
@@ -735,6 +778,10 @@ mod tests {
             delete_block.contains("authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"),
             "audit target deletion should require SetBucketTargetAction"
         );
+        assert!(
+            delete_block.contains("audit_target_operation_block_reason(\"managing audit targets from the console\")"),
+            "audit target deletion should reject requests when the audit module is disabled"
+        );
     }
 
     fn extract_block_between_markers<'a>(src: &'a str, start_marker: &str, end_marker: &str) -> &'a str {

rustfs/src/admin/handlers/event.rs

@@ -17,14 +17,17 @@ use crate::admin::{
     handlers::target_descriptor::{
         AdminTargetSpec, AdminTargetValidator, EndpointKey, TargetDomain, TargetEndpointSource, allowed_target_keys,
         collect_validated_key_values as shared_collect_validated_key_values,
-        merge_target_endpoints as shared_merge_target_endpoints,
+        merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason,
         target_mutation_block_reason as shared_target_mutation_block_reason, target_service_name, target_spec,
         validate_target_request,
     },
     router::{AdminOperation, Operation, S3Router},
 };
 use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::server::{
+    ADMIN_PREFIX, RemoteAddr, is_notify_module_enabled, refresh_notify_module_enabled,
+    refresh_persisted_module_switches_from_store,
+};
 use futures::stream::{FuturesUnordered, StreamExt};
 use http::{HeaderMap, StatusCode};
 use hyper::Method;
@@ -167,6 +170,17 @@ fn target_mutation_block_reason(config: &Config, target_type: &str, target_name:
     )
 }
 
+async fn notification_target_operation_block_reason(action: &str) -> Option<String> {
+    if let Err(err) = refresh_persisted_module_switches_from_store().await {
+        warn!(
+            error = %err,
+            "failed to reload persisted module switches before checking notification target operation gating"
+        );
+    }
+    refresh_notify_module_enabled();
+    target_module_disabled_reason("notify", rustfs_config::ENV_NOTIFY_ENABLE, is_notify_module_enabled(), action)
+}
+
 fn merge_notification_endpoints(config: &Config, runtime_statuses: HashMap<EndpointKey, String>) -> Vec<NotificationEndpoint> {
     shared_merge_target_endpoints(&notification_target_specs(), NOTIFY_ROUTE_PREFIX, config, runtime_statuses)
         .into_iter()
@@ -197,6 +211,9 @@ impl Operation for NotificationTarget {
         let (target_type, target_name) = extract_target_params(&params)?;
 
         authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
+        if let Some(reason) = notification_target_operation_block_reason("managing notification targets from the console").await {
+            return Err(s3_error!(InvalidRequest, "{reason}"));
+        }
         let ns = get_notification_system()?;
         let config_snapshot = ns.config.read().await.clone();
         if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name) {
@@ -291,6 +308,13 @@ impl Operation for ListTargetsArns {
         let span = Span::current();
         let _enter = span.enter();
         authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;
+        if let Some(reason) = notification_target_operation_block_reason(
+            "querying notification target ARNs for bucket associations from the console",
+        )
+        .await
+        {
+            return Err(s3_error!(InvalidRequest, "{reason}"));
+        }
         let ns = get_notification_system()?;
 
         let targets = ns.get_target_values().await;
@@ -336,6 +360,9 @@ impl Operation for RemoveNotificationTarget {
         let (target_type, target_name) = extract_target_params(&params)?;
 
         authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;
+        if let Some(reason) = notification_target_operation_block_reason("managing notification targets from the console").await {
+            return Err(s3_error!(InvalidRequest, "{reason}"));
+        }
         let ns = get_notification_system()?;
         let config_snapshot = ns.config.read().await.clone();
         if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name) {
@@ -543,6 +570,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn notification_target_operation_block_reason_requires_notify_module_enable() {
+        with_var(rustfs_config::ENV_NOTIFY_ENABLE, Some("false"), || {
+            let reason = futures::executor::block_on(notification_target_operation_block_reason(
+                "managing notification targets from the console",
+            ));
+            assert!(reason.is_some());
+            assert!(reason.unwrap().contains("set RUSTFS_NOTIFY_ENABLE=true"));
+        });
+    }
+
+    #[test]
+    fn notification_target_operation_block_reason_allows_when_notify_module_enabled() {
+        with_var(rustfs_config::ENV_NOTIFY_ENABLE, Some("true"), || {
+            assert!(
+                futures::executor::block_on(notification_target_operation_block_reason(
+                    "managing notification targets from the console"
+                ))
+                .is_none()
+            );
+        });
+    }
+
     #[test]
     fn target_mutation_block_reason_rejects_mixed_target() {
         with_var("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("https://example.com/hook"), || {
@@ -705,6 +755,11 @@ mod tests {
             put_block.contains("authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"),
             "notification target writes should require SetBucketTargetAction"
         );
+        assert!(
+            put_block.contains("notification_target_operation_block_reason(")
+                && put_block.contains("\"managing notification targets from the console\""),
+            "notification target writes should reject requests when the notify module is disabled"
+        );
         assert!(
             list_block.contains("authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;"),
             "notification target list should require GetBucketTargetAction"
@@ -713,10 +768,20 @@ mod tests {
             arns_block.contains("authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;"),
             "notification target arn listing should require GetBucketTargetAction"
         );
+        assert!(
+            arns_block.contains("notification_target_operation_block_reason(")
+                && arns_block.contains("\"querying notification target ARNs for bucket associations from the console\""),
+            "notification target arn listing should reject requests when the notify module is disabled"
+        );
         assert!(
             delete_block.contains("authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"),
             "notification target deletion should require SetBucketTargetAction"
         );
+        assert!(
+            delete_block.contains("notification_target_operation_block_reason(")
+                && delete_block.contains("\"managing notification targets from the console\""),
+            "notification target deletion should reject requests when the notify module is disabled"
+        );
     }
 
     #[test]

rustfs/src/admin/handlers/mod.rs

@@ -26,6 +26,7 @@ pub mod kms_dynamic;
 pub mod kms_keys;
 pub mod kms_management;
 pub mod metrics;
+pub mod module_switch;
 pub mod oidc;
 pub mod policies;
 pub mod pools;
@@ -55,6 +56,8 @@ mod tests {
         // Test that handler structs can be created
         let _account_handler = account_info::AccountInfoHandler {};
         let _list_audit_targets = audit::ListAuditTargets {};
+        let _get_module_switches = module_switch::GetModuleSwitchesHandler {};
+        let _update_module_switches = module_switch::UpdateModuleSwitchesHandler {};
         let _service_handler = system::ServiceHandle {};
         let _server_info_handler = system::ServerInfoHandler {};
         let _inspect_data_handler = system::InspectDataHandler {};

rustfs/src/admin/handlers/module_switch.rs

@@ -0,0 +1,228 @@
+// 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.
+
+use crate::admin::{
+    auth::validate_admin_request,
+    router::{AdminOperation, Operation, S3Router},
+};
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{
+    ADMIN_PREFIX, ModuleSwitchSnapshot, ModuleSwitchSource, PersistedModuleSwitches, RemoteAddr, current_module_switch_snapshot,
+    init_event_notifier, refresh_audit_module_enabled, refresh_notify_module_enabled,
+    refresh_persisted_module_switches_from_store, save_persisted_module_switches_to_store, shutdown_event_notifier,
+    start_audit_system, stop_audit_system, validate_module_switch_update,
+};
+use http::{HeaderMap, StatusCode};
+use hyper::Method;
+use matchit::Params;
+use rustfs_audit::AuditError;
+use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
+use rustfs_policy::policy::action::{Action, AdminAction};
+use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
+use serde::{Deserialize, Serialize};
+
+pub fn register_module_switch_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/module-switches").as_str(),
+        AdminOperation(&GetModuleSwitchesHandler {}),
+    )?;
+
+    r.insert(
+        Method::PUT,
+        format!("{}{}", ADMIN_PREFIX, "/v3/module-switches").as_str(),
+        AdminOperation(&UpdateModuleSwitchesHandler {}),
+    )?;
+
+    Ok(())
+}
+
+#[derive(Debug, Deserialize)]
+struct UpdateModuleSwitchesRequest {
+    notify_enabled: bool,
+    audit_enabled: bool,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+struct ModuleSwitchesResponse {
+    notify_enabled: bool,
+    audit_enabled: bool,
+    persisted_notify_enabled: bool,
+    persisted_audit_enabled: bool,
+    notify_source: ModuleSwitchSource,
+    audit_source: ModuleSwitchSource,
+}
+
+impl From<ModuleSwitchSnapshot> for ModuleSwitchesResponse {
+    fn from(value: ModuleSwitchSnapshot) -> Self {
+        Self {
+            notify_enabled: value.notify_enabled,
+            audit_enabled: value.audit_enabled,
+            persisted_notify_enabled: value.persisted_notify_enabled,
+            persisted_audit_enabled: value.persisted_audit_enabled,
+            notify_source: value.notify_source,
+            audit_source: value.audit_source,
+        }
+    }
+}
+
+fn build_response<T: Serialize>(
+    status: StatusCode,
+    body: &T,
+    request_id: Option<&http::HeaderValue>,
+) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let data = serde_json::to_vec(body).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
+    let mut header = HeaderMap::new();
+    header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
+    if let Some(v) = request_id {
+        header.insert("x-request-id", v.clone());
+    }
+    Ok(S3Response::with_headers((status, Body::from(data)), header))
+}
+
+async fn authorize_module_switch_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
+    let Some(input_cred) = &req.credentials else {
+        return Err(s3_error!(InvalidRequest, "authentication required"));
+    };
+
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        vec![Action::AdminAction(action)],
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
+}
+
+async fn refresh_module_switch_snapshot() -> S3Result<ModuleSwitchSnapshot> {
+    // Re-read persisted values before every console read/write so the current
+    // node reflects the latest cluster-wide state instead of stale atomics.
+    refresh_persisted_module_switches_from_store()
+        .await
+        .map_err(|e| s3_error!(InternalError, "failed to reload persisted module switches: {}", e))?;
+    refresh_notify_module_enabled();
+    refresh_audit_module_enabled();
+    Ok(current_module_switch_snapshot())
+}
+
+pub struct GetModuleSwitchesHandler {}
+
+#[async_trait::async_trait]
+impl Operation for GetModuleSwitchesHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_module_switch_request(&req, AdminAction::ServerInfoAdminAction).await?;
+        let snapshot = refresh_module_switch_snapshot().await?;
+        build_response(StatusCode::OK, &ModuleSwitchesResponse::from(snapshot), req.headers.get("x-request-id"))
+    }
+}
+
+pub struct UpdateModuleSwitchesHandler {}
+
+#[async_trait::async_trait]
+impl Operation for UpdateModuleSwitchesHandler {
+    async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_module_switch_request(&req, AdminAction::ConfigUpdateAdminAction).await?;
+        refresh_persisted_module_switches_from_store()
+            .await
+            .map_err(|e| s3_error!(InternalError, "failed to reload persisted module switches: {}", e))?;
+
+        let body = req
+            .input
+            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
+            .await
+            .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?;
+        if body.is_empty() {
+            return Err(s3_error!(InvalidRequest, "request body is required"));
+        }
+
+        let request: UpdateModuleSwitchesRequest =
+            serde_json::from_slice(&body).map_err(|e| s3_error!(InvalidRequest, "invalid JSON: {}", e))?;
+        let switches = PersistedModuleSwitches {
+            notify_enabled: request.notify_enabled,
+            audit_enabled: request.audit_enabled,
+        };
+
+        // Reject conflicting writes early so operators do not persist a console
+        // value that still cannot win over an explicit env override.
+        if let Err(err) = validate_module_switch_update(switches) {
+            let _ = refresh_module_switch_snapshot().await;
+            return Err(s3_error!(InvalidRequest, "{err}"));
+        }
+
+        save_persisted_module_switches_to_store(switches)
+            .await
+            .map_err(|e| s3_error!(InternalError, "failed to save module switches: {}", e))?;
+
+        // Apply the new effective values immediately on this node so the console
+        // response reflects the runtime state after to write completes.
+        if refresh_notify_module_enabled() {
+            init_event_notifier().await;
+        } else {
+            shutdown_event_notifier().await;
+        }
+
+        if refresh_audit_module_enabled() {
+            match start_audit_system().await {
+                Ok(()) | Err(AuditError::AlreadyInitialized) => {}
+                Err(e) => return Err(s3_error!(InternalError, "failed to apply audit module switch: {}", e)),
+            }
+        } else {
+            stop_audit_system()
+                .await
+                .map_err(|e| s3_error!(InternalError, "failed to stop audit module after switch update: {}", e))?;
+        }
+
+        let snapshot = current_module_switch_snapshot();
+        build_response(StatusCode::OK, &ModuleSwitchesResponse::from(snapshot), req.headers.get("x-request-id"))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn module_switch_handlers_require_admin_authorization_contract() {
+        let src = include_str!("module_switch.rs");
+        let get_block = extract_block_between_markers(
+            src,
+            "impl Operation for GetModuleSwitchesHandler",
+            "pub struct UpdateModuleSwitchesHandler",
+        );
+        let put_block = extract_block_between_markers(src, "impl Operation for UpdateModuleSwitchesHandler", "#[cfg(test)]");
+
+        assert!(
+            get_block.contains("authorize_module_switch_request(&req, AdminAction::ServerInfoAdminAction).await?;"),
+            "module switch GET should require ServerInfoAdminAction"
+        );
+        assert!(
+            put_block.contains("authorize_module_switch_request(&req, AdminAction::ConfigUpdateAdminAction).await?;"),
+            "module switch PUT should require ConfigUpdateAdminAction"
+        );
+    }
+
+    fn extract_block_between_markers<'a>(src: &'a str, start_marker: &str, end_marker: &str) -> &'a str {
+        let start = src
+            .find(start_marker)
+            .unwrap_or_else(|| panic!("Expected marker `{start_marker}` in source"));
+        let after_start = &src[start..];
+        let end = after_start
+            .find(end_marker)
+            .unwrap_or_else(|| panic!("Expected end marker `{end_marker}` in source"));
+        &after_start[..end]
+    }
+}

rustfs/src/admin/handlers/target_descriptor.rs

@@ -193,6 +193,14 @@ pub(crate) fn target_mutation_block_reason(
     }
 }
 
+pub(crate) fn target_module_disabled_reason(module_name: &str, env_key: &str, enabled: bool, action: &str) -> Option<String> {
+    (!enabled).then(|| {
+        format!(
+            "{module_name} module is disabled; enable the {module_name} module first in the console or set {env_key}=true before {action}"
+        )
+    })
+}
+
 pub(crate) fn merge_target_endpoints(
     specs: &[AdminTargetSpec],
     route_prefix: &str,

rustfs/src/admin/mod.rs

@@ -25,8 +25,8 @@ mod console_test;
 mod route_registration_test;
 
 use handlers::{
-    audit, bucket_meta, heal, health, kms, oidc, pools, profile_admin, quota, rebalance, replication, site_replication, sts,
-    system, tier, user,
+    audit, bucket_meta, heal, health, kms, module_switch, oidc, pools, profile_admin, quota, rebalance, replication,
+    site_replication, sts, system, tier, user,
 };
 use router::{AdminOperation, S3Router};
 use s3s::route::S3Route;
@@ -56,6 +56,7 @@ pub fn make_admin_route(console_enabled: bool) -> std::io::Result<impl S3Route>
     quota::register_quota_route(&mut r)?;
     bucket_meta::register_bucket_meta_route(&mut r)?;
     audit::register_audit_target_route(&mut r)?;
+    module_switch::register_module_switch_route(&mut r)?;
 
     replication::register_replication_route(&mut r)?;
     site_replication::register_site_replication_route(&mut r)?;

rustfs/src/admin/route_registration_test.rs

@@ -14,8 +14,8 @@
 
 use crate::admin::{
     handlers::{
-        audit, bucket_meta, heal, health, kms, oidc, pools, profile_admin, quota, rebalance, replication, site_replication, sts,
-        system, tier, user,
+        audit, bucket_meta, heal, health, kms, module_switch, oidc, pools, profile_admin, quota, rebalance, replication,
+        site_replication, sts, system, tier, user,
     },
     router::{AdminOperation, S3Router},
 };
@@ -52,6 +52,7 @@ fn register_admin_routes(router: &mut S3Router<AdminOperation>) {
     quota::register_quota_route(router).expect("register quota route");
     bucket_meta::register_bucket_meta_route(router).expect("register bucket meta route");
     audit::register_audit_target_route(router).expect("register audit target route");
+    module_switch::register_module_switch_route(router).expect("register module switch route");
     replication::register_replication_route(router).expect("register replication route");
     site_replication::register_site_replication_route(router).expect("register site replication route");
     profile_admin::register_profiling_route(router).expect("register profile route");
@@ -93,6 +94,8 @@ fn test_register_routes_cover_representative_admin_paths() {
     assert_route(&router, Method::GET, &admin_path("/v3/idp/builtin/policy-entities"));
     assert_route(&router, Method::GET, &admin_path("/v3/target/list"));
     assert_route(&router, Method::GET, &admin_path("/v3/audit/target/list"));
+    assert_route(&router, Method::GET, &admin_path("/v3/module-switches"));
+    assert_route(&router, Method::PUT, &admin_path("/v3/module-switches"));
     assert_route(&router, Method::PUT, &admin_path("/v3/audit/target/audit_webhook/test-audit"));
     assert_route(&router, Method::DELETE, &admin_path("/v3/audit/target/audit_webhook/test-audit/reset"));
     assert_route(&router, Method::GET, &admin_path("/v3/accountinfo"));
@@ -185,6 +188,8 @@ fn test_admin_alias_paths_match_existing_admin_routes() {
         (Method::PUT, compat_admin_alias_path("/v3/set-bucket-quota")),
         (Method::GET, compat_admin_alias_path("/v3/get-bucket-quota")),
         (Method::GET, compat_admin_alias_path("/v3/audit/target/list")),
+        (Method::GET, compat_admin_alias_path("/v3/module-switches")),
+        (Method::PUT, compat_admin_alias_path("/v3/module-switches")),
         (Method::PUT, compat_admin_alias_path("/v3/audit/target/audit_webhook/test-audit")),
         (Method::DELETE, compat_admin_alias_path("/v3/audit/target/audit_webhook/test-audit/reset")),
         (Method::POST, compat_admin_alias_path("/v3/heal/")),

rustfs/src/server/audit.rs

@@ -12,6 +12,7 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use super::{module_switch::resolve_audit_module_state, refresh_persisted_module_switches_from_store};
 use crate::app::context::resolve_server_config;
 use rustfs_audit::{AuditError, AuditResult, audit_system, init_audit_system, system::AuditSystemState};
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -24,7 +25,7 @@ fn server_config_from_context() -> Option<rustfs_ecstore::config::Config> {
 }
 
 pub fn refresh_audit_module_enabled() -> bool {
-    let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_AUDIT_ENABLE, rustfs_config::DEFAULT_AUDIT_ENABLE);
+    let enabled = resolve_audit_module_state().enabled;
     AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
     enabled
 }
@@ -56,11 +57,15 @@ fn has_any_audit_targets(config: &rustfs_ecstore::config::Config) -> bool {
 /// If not configured, it skips the initialization.
 /// It also handles cases where the audit system is already running or if the global configuration is not loaded.
 pub async fn start_audit_system() -> AuditResult<()> {
+    if let Err(err) = refresh_persisted_module_switches_from_store().await {
+        warn!("Failed to refresh persisted audit module switch from store: {}", err);
+    }
+
     let enabled = refresh_audit_module_enabled();
     if !enabled {
         info!(
             target: "rustfs::main::start_audit_system",
-            "Audit module is disabled by RUSTFS_AUDIT_ENABLE=false, audit system initialization is skipped."
+            "Audit module is disabled, audit system initialization is skipped. Enable the audit module first."
         );
         return Ok(());
     }
@@ -106,34 +111,69 @@ pub async fn start_audit_system() -> AuditResult<()> {
         target: "rustfs::main::start_audit_system",
         "Audit subsystem configuration detected and started initializing the audit system."
     );
-    // 3. Initialize and start the audit system
-    let system = init_audit_system();
-    // Check if the audit system is already running
-    let state = system.get_state().await;
-    if state == AuditSystemState::Running {
-        warn!(
-            target: "rustfs::main::start_audit_system",
-            "The audit system is running, skip repeated initialization."
-        );
-        return Err(AuditError::AlreadyInitialized);
-    }
-    // Preparation before starting
-    match system.start(server_config).await {
-        Ok(_) => {
-            info!(
-                target: "rustfs::main::start_audit_system",
-                "Audit system started successfully with time: {}.",
-                jiff::Zoned::now()
-            );
-            Ok(())
+
+    if let Some(system) = audit_system() {
+        match system.get_state().await {
+            AuditSystemState::Running | AuditSystemState::Paused | AuditSystemState::Starting => {
+                // Match notify behavior: prefer reloading the existing singleton
+                // instead of constructing a second lifecycle path on re-enable.
+                match system.reload_config(server_config).await {
+                    Ok(()) => {
+                        info!(
+                            target: "rustfs::main::start_audit_system",
+                            "Audit system reloaded successfully with time: {}.",
+                            jiff::Zoned::now()
+                        );
+                        Ok(())
+                    }
+                    Err(e) => {
+                        warn!(
+                            target: "rustfs::main::start_audit_system",
+                            "Audit system reload failed: {:?}",
+                            e
+                        );
+                        Err(e)
+                    }
+                }
+            }
+            AuditSystemState::Stopped | AuditSystemState::Stopping => match system.start(server_config).await {
+                Ok(()) => {
+                    info!(
+                        target: "rustfs::main::start_audit_system",
+                        "Audit system started successfully with time: {}.",
+                        jiff::Zoned::now()
+                    );
+                    Ok(())
+                }
+                Err(e) => {
+                    warn!(
+                        target: "rustfs::main::start_audit_system",
+                        "Audit system startup failed: {:?}",
+                        e
+                    );
+                    Err(e)
+                }
+            },
         }
-        Err(e) => {
-            warn!(
-                target: "rustfs::main::start_audit_system",
-                "Audit system startup failed: {:?}",
-                e
-            );
-            Err(e)
+    } else {
+        let system = init_audit_system();
+        match system.start(server_config).await {
+            Ok(()) => {
+                info!(
+                    target: "rustfs::main::start_audit_system",
+                    "Audit system started successfully with time: {}.",
+                    jiff::Zoned::now()
+                );
+                Ok(())
+            }
+            Err(e) => {
+                warn!(
+                    target: "rustfs::main::start_audit_system",
+                    "Audit system startup failed: {:?}",
+                    e
+                );
+                Err(e)
+            }
         }
     }
 }

rustfs/src/server/event.rs

@@ -12,6 +12,7 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use super::{module_switch::resolve_notify_module_state, refresh_persisted_module_switches_from_store};
 use crate::app::context::resolve_server_config;
 use rustfs_ecstore::event_notification::{EventArgs as EcstoreEventArgs, register_event_dispatch_hook};
 use rustfs_notify::EventArgs as NotifyEventArgs;
@@ -27,7 +28,7 @@ fn server_config_from_context() -> Option<rustfs_ecstore::config::Config> {
 }
 
 pub fn refresh_notify_module_enabled() -> bool {
-    let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_NOTIFY_ENABLE, rustfs_config::DEFAULT_NOTIFY_ENABLE);
+    let enabled = resolve_notify_module_state().enabled;
     NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
     enabled
 }
@@ -98,11 +99,15 @@ pub async fn shutdown_event_notifier() {
 
 #[instrument]
 pub async fn init_event_notifier() {
+    if let Err(err) = refresh_persisted_module_switches_from_store().await {
+        warn!("Failed to refresh persisted notify module switch from store: {}", err);
+    }
+
     let enabled = refresh_notify_module_enabled();
     if !enabled {
         info!(
             target: "rustfs::main::init_event_notifier",
-            "Notify module is disabled by RUSTFS_NOTIFY_ENABLE=false, event notifier initialization is skipped."
+            "Notify module is disabled, event notifier initialization is skipped. Enable the notify module first."
         );
         return;
     }
@@ -126,9 +131,18 @@ pub async fn init_event_notifier() {
         "Event notifier configuration found, proceeding with initialization."
     );
 
-    // 2. Initialize the notification system asynchronously with a global configuration
-    // Use direct await for better error handling and faster initialization
-    if let Err(e) = rustfs_notify::initialize(server_config).await {
+    if let Some(system) = rustfs_notify::notification_system() {
+        // Reuse the existing global system on re-enable so bucket rules, metrics,
+        // and stream lifecycle stay aligned with the current process singleton.
+        if let Err(e) = system.reload_config(server_config).await {
+            error!("Failed to reload event notifier system: {}", e);
+        } else {
+            info!(
+                target: "rustfs::main::init_event_notifier",
+                "Event notifier system reloaded successfully."
+            );
+        }
+    } else if let Err(e) = rustfs_notify::initialize(server_config).await {
         error!("Failed to initialize event notifier system: {}", e);
     } else {
         install_ecstore_event_dispatch_hook();

rustfs/src/server/mod.rs

@@ -19,6 +19,7 @@ mod event;
 mod http;
 mod hybrid;
 mod layer;
+mod module_switch;
 mod prefix;
 mod readiness;
 mod runtime;
@@ -38,6 +39,10 @@ pub use service_state::ShutdownSignal;
 pub use service_state::wait_for_shutdown;
 
 // Items only used within the library crate (admin handlers, server/http.rs, etc.).
+pub(crate) use module_switch::{
+    ModuleSwitchSnapshot, ModuleSwitchSource, PersistedModuleSwitches, current_module_switch_snapshot,
+    refresh_persisted_module_switches_from_store, save_persisted_module_switches_to_store, validate_module_switch_update,
+};
 pub(crate) use prefix::{
     ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, MINIO_ADMIN_PREFIX,
     MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TONIC_PREFIX, VERSION,

rustfs/src/server/module_switch.rs

@@ -0,0 +1,307 @@
+// 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.
+
+use rustfs_ecstore::{
+    config::com::{read_config, save_config},
+    error::Error as StorageError,
+    new_object_layer_fn,
+};
+use serde::{Deserialize, Serialize};
+use std::sync::atomic::{AtomicBool, Ordering};
+
+const MODULE_SWITCH_CONFIG_PATH: &str = "config/module_switches.json";
+
+// Keep a cheap in-process snapshot so hot-path checks do not need to read
+// cluster metadata after startup or console-triggered refresh.
+static PERSISTED_NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
+static PERSISTED_AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
+static PERSISTED_MODULE_SWITCH_CONFIGURED: AtomicBool = AtomicBool::new(false);
+
+#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Eq, Serialize)]
+pub(crate) struct PersistedModuleSwitches {
+    pub(crate) notify_enabled: bool,
+    pub(crate) audit_enabled: bool,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum ModuleSwitchSource {
+    Env,
+    Console,
+    Default,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct ModuleSwitchResolution {
+    pub(crate) enabled: bool,
+    pub(crate) source: ModuleSwitchSource,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+pub(crate) struct ModuleSwitchSnapshot {
+    pub(crate) notify_enabled: bool,
+    pub(crate) audit_enabled: bool,
+    pub(crate) persisted_notify_enabled: bool,
+    pub(crate) persisted_audit_enabled: bool,
+    pub(crate) notify_source: ModuleSwitchSource,
+    pub(crate) audit_source: ModuleSwitchSource,
+}
+
+pub(crate) fn current_persisted_module_switches() -> PersistedModuleSwitches {
+    PersistedModuleSwitches {
+        notify_enabled: PERSISTED_NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed),
+        audit_enabled: PERSISTED_AUDIT_MODULE_ENABLED.load(Ordering::Relaxed),
+    }
+}
+
+fn persisted_module_switches_configured() -> bool {
+    PERSISTED_MODULE_SWITCH_CONFIGURED.load(Ordering::Relaxed)
+}
+
+pub(crate) fn set_persisted_module_switches(config: PersistedModuleSwitches, configured: bool) {
+    PERSISTED_NOTIFY_MODULE_ENABLED.store(config.notify_enabled, Ordering::Relaxed);
+    PERSISTED_AUDIT_MODULE_ENABLED.store(config.audit_enabled, Ordering::Relaxed);
+    PERSISTED_MODULE_SWITCH_CONFIGURED.store(configured, Ordering::Relaxed);
+}
+
+fn env_override_exists(key: &str) -> bool {
+    std::env::var_os(key).is_some()
+}
+
+fn env_override_value(key: &str) -> Option<bool> {
+    rustfs_utils::get_env_opt_bool(key)
+}
+
+fn effective_module_switch_state(env_key: &str, persisted_enabled: bool, default_enabled: bool) -> ModuleSwitchResolution {
+    // Explicit env remains the highest-priority source so process-level bootstrap
+    // cannot be silently overridden by a later console write.
+    if let Some(env_enabled) = env_override_value(env_key) {
+        return ModuleSwitchResolution {
+            enabled: env_enabled,
+            source: ModuleSwitchSource::Env,
+        };
+    }
+
+    if persisted_module_switches_configured() {
+        return ModuleSwitchResolution {
+            enabled: persisted_enabled,
+            source: ModuleSwitchSource::Console,
+        };
+    }
+
+    ModuleSwitchResolution {
+        enabled: default_enabled,
+        source: ModuleSwitchSource::Default,
+    }
+}
+
+pub(crate) fn resolve_notify_module_state() -> ModuleSwitchResolution {
+    effective_module_switch_state(
+        rustfs_config::ENV_NOTIFY_ENABLE,
+        PERSISTED_NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed),
+        rustfs_config::DEFAULT_NOTIFY_ENABLE,
+    )
+}
+
+pub(crate) fn resolve_audit_module_state() -> ModuleSwitchResolution {
+    effective_module_switch_state(
+        rustfs_config::ENV_AUDIT_ENABLE,
+        PERSISTED_AUDIT_MODULE_ENABLED.load(Ordering::Relaxed),
+        rustfs_config::DEFAULT_AUDIT_ENABLE,
+    )
+}
+
+pub(crate) fn current_module_switch_snapshot() -> ModuleSwitchSnapshot {
+    let persisted = current_persisted_module_switches();
+    let notify = resolve_notify_module_state();
+    let audit = resolve_audit_module_state();
+
+    ModuleSwitchSnapshot {
+        notify_enabled: notify.enabled,
+        audit_enabled: audit.enabled,
+        persisted_notify_enabled: persisted.notify_enabled,
+        persisted_audit_enabled: persisted.audit_enabled,
+        notify_source: notify.source,
+        audit_source: audit.source,
+    }
+}
+
+fn validate_env_override_for_request(env_key: &str, requested: bool, label: &str) -> Result<(), String> {
+    if !env_override_exists(env_key) {
+        return Ok(());
+    }
+
+    match env_override_value(env_key) {
+        // Matching values are safe: we still persist the console value, but the
+        // effective runtime source remains env until the operator changes it.
+        Some(value) if value == requested => Ok(()),
+        Some(value) => Err(format!(
+            "{label} is managed by environment variable {env_key}={value}; update the environment value first, then use the console to refresh the module switch state"
+        )),
+        None => Err(format!(
+            "{label} is managed by environment variable {env_key}, but its value is not a valid boolean; fix the environment value first, then use the console to refresh the module switch state"
+        )),
+    }
+}
+
+pub(crate) fn validate_module_switch_update(requested: PersistedModuleSwitches) -> Result<(), String> {
+    validate_env_override_for_request(rustfs_config::ENV_NOTIFY_ENABLE, requested.notify_enabled, "notify module")?;
+    validate_env_override_for_request(rustfs_config::ENV_AUDIT_ENABLE, requested.audit_enabled, "audit module")?;
+    Ok(())
+}
+
+pub(crate) async fn refresh_persisted_module_switches_from_store() -> Result<PersistedModuleSwitches, String> {
+    let Some(store) = new_object_layer_fn() else {
+        return Err("storage layer not initialized".to_string());
+    };
+
+    let (config, configured) = match read_config(store, MODULE_SWITCH_CONFIG_PATH).await {
+        Ok(data) => (
+            serde_json::from_slice::<PersistedModuleSwitches>(&data)
+                .map_err(|e| format!("failed to deserialize module switch config: {e}"))?,
+            true,
+        ),
+        Err(StorageError::ConfigNotFound) => (PersistedModuleSwitches::default(), false),
+        Err(err) => return Err(format!("failed to load module switch config: {err}")),
+    };
+
+    // Track whether the persisted file exists so the effective state can
+    // distinguish "console configured false" from "never configured, use default".
+    set_persisted_module_switches(config, configured);
+    Ok(config)
+}
+
+pub(crate) async fn save_persisted_module_switches_to_store(config: PersistedModuleSwitches) -> Result<(), String> {
+    let Some(store) = new_object_layer_fn() else {
+        return Err("storage layer not initialized".to_string());
+    };
+
+    let data = serde_json::to_vec(&config).map_err(|e| format!("failed to serialize module switch config: {e}"))?;
+    save_config(store, MODULE_SWITCH_CONFIG_PATH, data)
+        .await
+        .map_err(|e| format!("failed to save module switch config: {e}"))?;
+
+    set_persisted_module_switches(config, true);
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serial_test::serial;
+    use temp_env::{with_var, with_vars};
+
+    #[test]
+    #[serial]
+    fn resolve_module_switch_state_prefers_env_override() {
+        set_persisted_module_switches(
+            PersistedModuleSwitches {
+                notify_enabled: false,
+                audit_enabled: false,
+            },
+            true,
+        );
+
+        with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, Some("true")),
+                (rustfs_config::ENV_AUDIT_ENABLE, Some("false")),
+            ],
+            || {
+                let notify = resolve_notify_module_state();
+                let audit = resolve_audit_module_state();
+
+                assert!(notify.enabled);
+                assert_eq!(notify.source, ModuleSwitchSource::Env);
+                assert!(!audit.enabled);
+                assert_eq!(audit.source, ModuleSwitchSource::Env);
+            },
+        );
+    }
+
+    #[test]
+    #[serial]
+    fn resolve_module_switch_state_falls_back_to_console_value() {
+        set_persisted_module_switches(
+            PersistedModuleSwitches {
+                notify_enabled: true,
+                audit_enabled: false,
+            },
+            true,
+        );
+
+        with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, None::<&str>),
+                (rustfs_config::ENV_AUDIT_ENABLE, None::<&str>),
+            ],
+            || {
+                let notify = resolve_notify_module_state();
+                let audit = resolve_audit_module_state();
+
+                assert!(notify.enabled);
+                assert_eq!(notify.source, ModuleSwitchSource::Console);
+                assert!(!audit.enabled);
+                assert_eq!(audit.source, ModuleSwitchSource::Console);
+            },
+        );
+    }
+
+    #[test]
+    #[serial]
+    fn validate_module_switch_update_rejects_env_conflict() {
+        with_var(rustfs_config::ENV_NOTIFY_ENABLE, Some("true"), || {
+            let err = validate_module_switch_update(PersistedModuleSwitches {
+                notify_enabled: false,
+                audit_enabled: false,
+            })
+            .unwrap_err();
+
+            assert!(err.contains(rustfs_config::ENV_NOTIFY_ENABLE));
+            assert!(err.contains("update the environment value first"));
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn validate_module_switch_update_allows_matching_env_override() {
+        with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, Some("true")),
+                (rustfs_config::ENV_AUDIT_ENABLE, Some("false")),
+            ],
+            || {
+                validate_module_switch_update(PersistedModuleSwitches {
+                    notify_enabled: true,
+                    audit_enabled: false,
+                })
+                .expect("matching env override should be accepted");
+            },
+        );
+    }
+
+    #[test]
+    #[serial]
+    fn validate_module_switch_update_rejects_invalid_env_override() {
+        with_var(rustfs_config::ENV_AUDIT_ENABLE, Some("invalid"), || {
+            let err = validate_module_switch_update(PersistedModuleSwitches {
+                notify_enabled: false,
+                audit_enabled: true,
+            })
+            .unwrap_err();
+
+            assert!(err.contains("not a valid boolean"));
+        });
+    }
+}