diff --git a/rustfs/src/admin/handlers/audit.rs b/rustfs/src/admin/handlers/audit.rs index 900bbddbf..4b8192dc2 100644 --- a/rustfs/src/admin/handlers/audit.rs +++ b/rustfs/src/admin/handlers/audit.rs @@ -14,13 +14,14 @@ use crate::admin::{ auth::authorize_admin_request, - handlers::audit_runtime_config::{load_server_config_from_store, update_audit_config_and_reload}, + handlers::audit_runtime_config::update_audit_config_and_reload, handlers::target_descriptor::{ AdminTargetSpec, EndpointKey, RuntimeHealthStatus, TargetEndpointSource, admin_target_spec_from_builtin, - build_enabled_target_kvs, build_json_response, collect_runtime_statuses, extract_supported_target_params, + build_enabled_target_kvs, build_json_response, extract_supported_target_params, merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason, }, + handlers::target_read::{collect_audit_runtime_statuses, load_server_config_from_store, load_target_read_snapshot}, router::{AdminOperation, Operation, S3Router}, }; use crate::server::{ @@ -29,7 +30,6 @@ use crate::server::{ use http::StatusCode; use hyper::Method; use matchit::Params; -use rustfs_audit::audit_system; use rustfs_config::audit::AUDIT_ROUTE_PREFIX; use rustfs_config::server_config::Config; use rustfs_config::{AUDIT_DEFAULT_DIR, MAX_ADMIN_REQUEST_BODY_SIZE}; @@ -344,12 +344,7 @@ impl Operation for ListAuditTargets { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { authorize_audit_admin_request(&req, AdminAction::GetBucketTargetAction).await?; - let mut runtime_statuses = HashMap::new(); - if let Some(system) = audit_system() { - runtime_statuses = collect_runtime_statuses(system.get_target_values().await).await; - } - - let config = load_server_config_from_store().await?; + let (config, runtime_statuses) = load_target_read_snapshot(collect_audit_runtime_statuses()).await?; let audit_endpoints = merge_audit_endpoints(&config, runtime_statuses)?; let data = serde_json::to_vec(&AuditEndpointsResponse { audit_endpoints }).map_err(|e| { log_audit_target_request_failed!("list_audit_targets", "serialize_audit_targets_failed", None, None, e); diff --git a/rustfs/src/admin/handlers/audit_runtime_config.rs b/rustfs/src/admin/handlers/audit_runtime_config.rs index ad58b726a..3ebe8de83 100644 --- a/rustfs/src/admin/handlers/audit_runtime_config.rs +++ b/rustfs/src/admin/handlers/audit_runtime_config.rs @@ -17,8 +17,8 @@ use crate::admin::handlers::target_descriptor::AdminTargetSpec; use crate::admin::runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context}; use crate::admin::service::config::with_runtime_config_reload_lock; use crate::admin::storage_api::config::{ - read_admin_config_without_migrate, read_admin_server_config_snapshot, read_existing_admin_server_config_no_lock, - save_admin_server_config_snapshot, with_admin_server_config_read_lock, + read_admin_server_config_snapshot, read_existing_admin_server_config_no_lock, save_admin_server_config_snapshot, + with_admin_server_config_read_lock, }; use rustfs_audit::{audit_system, start_audit_system as start_global_audit_system, system::AuditSystemState}; use rustfs_config::DEFAULT_DELIMITER; @@ -26,21 +26,6 @@ use rustfs_config::server_config::Config; use s3s::{S3Error, S3Result, s3_error}; use tracing::warn; -pub(crate) async fn load_server_config_from_store_for_context(context: Option<&AppContext>) -> S3Result { - let Some(store) = current_object_store_handle_for_context(context) else { - return Ok(Config::new()); - }; - - read_admin_config_without_migrate(store) - .await - .map_err(|e| s3_error!(InternalError, "failed to read server config: {}", e)) -} - -pub(crate) async fn load_server_config_from_store() -> S3Result { - let context = current_app_context(); - load_server_config_from_store_for_context(context.as_deref()).await -} - fn has_any_audit_targets(specs: &[AdminTargetSpec], config: &Config) -> bool { specs.iter().any(|spec| { config @@ -188,7 +173,7 @@ mod tests { use super::*; use crate::admin::handlers::target_descriptor::admin_target_spec_from_builtin; use crate::admin::runtime_sources::{IamInterface, KmsInterface}; - use crate::admin::storage_api::config::save_admin_server_config; + use crate::admin::storage_api::config::{read_admin_config_without_migrate, save_admin_server_config}; use crate::admin::storage_api::error::StorageError; use rustfs_config::audit::AUDIT_WEBHOOK_SUB_SYS; use rustfs_config::server_config::KVS; diff --git a/rustfs/src/admin/handlers/event.rs b/rustfs/src/admin/handlers/event.rs index 7db37475a..7ac3c1c56 100644 --- a/rustfs/src/admin/handlers/event.rs +++ b/rustfs/src/admin/handlers/event.rs @@ -14,7 +14,7 @@ use crate::admin::{ auth::authorize_admin_request, - handlers::notify_runtime_access::{get_notification_system, load_notification_config_snapshot}, + handlers::notify_runtime_access::{get_notification_system, load_notification_mutation_snapshot}, handlers::supervise_admin_mutation, handlers::target_descriptor::{ AdminTargetSpec, EndpointKey, RuntimeHealthStatus, TargetEndpointSource, admin_target_spec_from_builtin, @@ -22,6 +22,7 @@ use crate::admin::{ merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason, }, + handlers::target_read::{collect_notification_runtime_statuses, load_target_read_snapshot}, router::{AdminOperation, Operation, S3Router}, runtime_sources::object_store_from_extensions, runtime_sources::{AppContext, app_context_from_req}, @@ -464,7 +465,7 @@ impl Operation for NotificationTarget { log_notification_target_operation_blocked!("set_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } - let (ns, config_snapshot) = load_notification_config_snapshot().await?; + let (ns, config_snapshot) = load_notification_mutation_snapshot().await?; if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name)? { log_notification_target_operation_blocked!("set_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); @@ -537,8 +538,7 @@ impl Operation for ListNotificationTargets { s3_error!(InternalError, "failed to refresh notification module state") })?; let notify_enabled = refresh_notify_module_enabled(); - let (ns, config) = load_notification_config_snapshot().await?; - let runtime_statuses = collect_runtime_statuses(ns.get_target_values().await).await; + let (config, runtime_statuses) = load_target_read_snapshot(collect_notification_runtime_statuses()).await?; let notification_endpoints = merge_notification_endpoints(&config, runtime_statuses)?; let data = serde_json::to_vec(&NotificationEndpointsResponse { @@ -610,7 +610,7 @@ impl Operation for RemoveNotificationTarget { log_notification_target_operation_blocked!("remove_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); } - let (ns, config_snapshot) = load_notification_config_snapshot().await?; + let (ns, config_snapshot) = load_notification_mutation_snapshot().await?; if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name)? { log_notification_target_operation_blocked!("remove_target_config", Some(target_type), Some(target_name), &reason); return Err(s3_error!(InvalidRequest, "{reason}")); @@ -676,6 +676,268 @@ mod tests { } } + #[tokio::test] + #[serial] + async fn target_list_reads_durable_config_without_runtime_reconciliation() { + use crate::admin::handlers::{ + audit::ListAuditTargets, + extensions::ListExtensionInstancesHandler, + plugins_instances::{GetPluginInstanceHandler, ListPluginInstancesHandler}, + target_read::load_target_read_snapshot, + }; + use crate::admin::storage_api::config::{ + read_admin_config, save_admin_config, save_admin_server_config, with_admin_server_config_write_lock, + }; + use crate::server::{mark_event_notifier_unreconciled, with_notify_runtime_reconcile_lock}; + use http_body_util::BodyExt as _; + use std::sync::Arc; + use std::time::Duration; + + async fn response_json( + call: impl std::future::Future>>, + ) -> S3Result { + let response = tokio::time::timeout(Duration::from_secs(10), call) + .await + .expect("destination inventory must not wait for runtime reconciliation")?; + assert_eq!(response.output.0, StatusCode::OK); + let body = response + .output + .1 + .collect() + .await + .expect("collect inventory response") + .to_bytes(); + Ok(serde_json::from_slice(&body).expect("decode inventory response")) + } + + let _ = rustfs_credentials::init_global_action_credentials( + Some("notification-list-root".to_string()), + Some("notification-list-secret".to_string()), + ); + let temp_dir = tempfile::tempdir().expect("create notification list test directory"); + let env = rustfs_test_utils::TestECStoreEnv::builder() + .base_dir(temp_dir.path()) + .disk_count(1) + .init_bucket_metadata(false) + .build() + .await; + crate::runtime_sources::install_test_app_context(Arc::clone(&env.ecstore)).await; + + let credentials = crate::runtime_sources::current_action_credentials().expect("test root credentials"); + let request = |uri: &str| S3Request { + input: Body::empty(), + method: Method::GET, + uri: uri.parse().expect("target inventory URI"), + headers: http::HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: Some(s3s::auth::Credentials { + access_key: credentials.access_key.clone(), + secret_key: credentials.secret_key.clone().into(), + }), + region: None, + service: None, + trailing_headers: None, + }; + + temp_env::async_with_vars([(rustfs_config::ENV_NOTIFY_ENABLE, Some("true"))], async { + let mut config = Config::new(); + let mut kvs = enabled_kvs("on"); + kvs.insert(rustfs_config::WEBHOOK_ENDPOINT.to_string(), "http://127.0.0.1:9/hook".to_string()); + kvs.insert(rustfs_config::WEBHOOK_AUTH_TOKEN.to_string(), "inventory-secret-token".to_string()); + config + .0 + .entry(rustfs_config::audit::AUDIT_WEBHOOK_SUB_SYS.to_string()) + .or_default() + .insert("audit-unavailable".to_string(), kvs.clone()); + config + .0 + .entry(NOTIFY_WEBHOOK_SUB_SYS.to_string()) + .or_default() + .insert("unavailable".to_string(), kvs); + save_admin_server_config(env.ecstore.clone(), &config) + .await + .expect("persist target without loading it into the runtime"); + + let (locked_tx, locked_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let store = env.ecstore.clone(); + let writer = tokio::spawn(async move { + with_admin_server_config_write_lock(store, move |_guard| async move { + locked_tx.send(()).expect("signal config write lock acquisition"); + release_rx.await.expect("release config writer"); + }) + .await + .expect("hold config write lock"); + }); + locked_rx.await.expect("config write lock should be held"); + let (observing_tx, observing_rx) = tokio::sync::oneshot::channel(); + let snapshot = tokio::spawn(load_target_read_snapshot(async move { + observing_tx.send(()).expect("signal runtime observation"); + })); + tokio::time::timeout(Duration::from_secs(10), observing_rx) + .await + .expect("runtime observation must start while the durable read is blocked") + .expect("runtime observer should start"); + assert!(!snapshot.is_finished(), "the snapshot must still await its blocked durable read"); + release_tx.send(()).expect("release config write lock"); + writer.await.expect("config writer task"); + let (observed_config, ()) = snapshot.await.expect("snapshot task").expect("read durable snapshot"); + assert!(observed_config.0[NOTIFY_WEBHOOK_SUB_SYS].contains_key("unavailable")); + mark_event_notifier_unreconciled(); + + with_notify_runtime_reconcile_lock(async { + for initialize_runtime in [false, true] { + if initialize_runtime { + rustfs_notify::ensure_live_events(); + } + let value = + response_json(ListNotificationTargets {}.call(request("/rustfs/admin/v3/target/list"), Params::new())) + .await + .expect("read configured destinations while the runtime is unavailable"); + assert_eq!(value["notify_enabled"], true); + let endpoints = value["notification_endpoints"].as_array().expect("destination inventory"); + let endpoint = endpoints + .iter() + .find(|endpoint| endpoint["account_id"] == "unavailable" && endpoint["service"] == "webhook") + .expect("persisted destination remains visible"); + assert_eq!(endpoint["status"], "offline"); + assert_eq!(endpoint["health_state"], "offline"); + assert_eq!(endpoint["health_reason"], "not_loaded_in_runtime"); + assert_eq!(endpoint["source"], "config"); + + let audit = + response_json(ListAuditTargets {}.call(request("/rustfs/admin/v3/audit/target/list"), Params::new())) + .await + .expect("read audit destinations without runtime initialization"); + assert!( + audit["audit_endpoints"] + .as_array() + .expect("audit inventory") + .iter() + .any(|entry| { + entry["account_id"] == "audit-unavailable" && entry["health_reason"] == "not_loaded_in_runtime" + }) + ); + + for value in [ + response_json( + ListPluginInstancesHandler {}.call(request("/rustfs/admin/v4/plugins/instances"), Params::new()), + ) + .await + .expect("read plugin inventory without runtime reconciliation"), + response_json( + ListExtensionInstancesHandler {} + .call(request("/rustfs/admin/v4/extensions/instances"), Params::new()), + ) + .await + .expect("read extension inventory without runtime reconciliation"), + ] { + let instances = value["instances"].as_array().expect("instance inventory"); + for (domain, account_id) in [("notify", "unavailable"), ("audit", "audit-unavailable")] { + let entry = instances + .iter() + .find(|entry| entry["domain"] == domain && entry["account_id"] == account_id) + .expect("both target domains remain visible"); + assert_eq!(entry["status"], "offline"); + assert_eq!(entry["config"][rustfs_config::WEBHOOK_AUTH_TOKEN], "***redacted***"); + } + assert!(!value.to_string().contains("inventory-secret-token")); + } + + let mut router = Router::new(); + router.insert("/{id}", ()).expect("plugin detail route"); + let matched = router + .at("/builtin:webhook:notify:unavailable") + .expect("plugin detail params"); + let detail = response_json(GetPluginInstanceHandler {}.call( + request("/rustfs/admin/v4/plugins/instances/builtin:webhook:notify:unavailable"), + matched.params, + )) + .await + .expect("read plugin details without runtime reconciliation"); + assert_eq!(detail["config"][rustfs_config::WEBHOOK_AUTH_TOKEN], "***redacted***"); + assert!(!detail.to_string().contains("inventory-secret-token")); + assert!(!crate::server::is_event_notifier_reconciled()); + if let Some(system) = rustfs_notify::notification_system() { + assert_eq!(system.runtime_lifecycle_state(), rustfs_notify::NotificationRuntimeState::LiveOnly); + } + } + + let target = config.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].clone(); + let targets = config.0.get_mut(NOTIFY_WEBHOOK_SUB_SYS).expect("notification target config"); + targets.remove("unavailable"); + targets.insert("latest".to_string(), target); + save_admin_server_config(env.ecstore.clone(), &config) + .await + .expect("persist a newer target configuration while runtime stays unreconciled"); + let latest = response_json( + ListPluginInstancesHandler {}.call(request("/rustfs/admin/v4/plugins/instances"), Params::new()), + ) + .await + .expect("inventory must read the latest durable config"); + let instances = latest["instances"].as_array().expect("latest inventory"); + assert!(instances.iter().any(|entry| entry["account_id"] == "latest")); + assert!(!instances.iter().any(|entry| entry["account_id"] == "unavailable")); + + let valid_config = read_admin_config(env.ecstore.clone(), "config/config.json") + .await + .expect("read valid config bytes for restoration"); + save_admin_config(env.ecstore.clone(), "config/config.json", b"invalid-server-config".to_vec()) + .await + .expect("inject corrupt persisted configuration"); + let error = ListNotificationTargets {} + .call(request("/rustfs/admin/v3/target/list"), Params::new()) + .await + .expect_err("unreadable persisted config must not become an empty inventory"); + assert_eq!(error.code(), &S3ErrorCode::InternalError); + + let error = response_json( + ListPluginInstancesHandler {}.call(request("/rustfs/admin/v4/plugins/instances"), Params::new()), + ) + .await + .expect_err("plugin inventory must not hide corrupt durable config"); + assert_eq!(error.code(), &S3ErrorCode::InternalError); + let error = + tokio::time::timeout(Duration::from_secs(10), load_target_read_snapshot(std::future::pending::<()>())) + .await + .expect("a config read failure must cancel stalled runtime observations") + .expect_err("config read failure must remain visible"); + assert_eq!(error.code(), &S3ErrorCode::InternalError); + + let mut unsigned = request("/rustfs/admin/v3/target/list"); + unsigned.credentials = None; + let error = ListNotificationTargets {} + .call(unsigned, Params::new()) + .await + .expect_err("authorization must precede persisted configuration access"); + assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); + assert_eq!(error.message(), Some("credentials not found")); + + save_admin_config(env.ecstore.clone(), "config/config.json", valid_config) + .await + .expect("restore valid target configuration"); + save_admin_config(env.ecstore.clone(), "config/module_switches.json", b"invalid-module-switches".to_vec()) + .await + .expect("inject corrupt persisted module switches"); + let error = response_json( + ListPluginInstancesHandler {}.call(request("/rustfs/admin/v4/plugins/instances"), Params::new()), + ) + .await + .expect_err("plugin diagnostics must not silently reuse stale module switches"); + assert_eq!(error.code(), &S3ErrorCode::InternalError); + let mut unsigned = request("/rustfs/admin/v4/plugins/instances"); + unsigned.credentials = None; + let error = response_json(ListPluginInstancesHandler {}.call(unsigned, Params::new())) + .await + .expect_err("plugin authorization must precede module and config access"); + assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); + assert_eq!(error.message(), Some("authentication required")); + }) + .await; + }) + .await; + } + fn subscription_router() -> Router<()> { let mut router = Router::new(); router @@ -1319,7 +1581,7 @@ mod tests { .find("let notify_enabled = refresh_notify_module_enabled();") .expect("target list should resolve the effective env and persisted module state"); let load_index = list_block - .find("load_notification_config_snapshot().await?") + .find("load_target_read_snapshot(collect_notification_runtime_statuses()).await?") .expect("target list should load notification config"); assert!( authorize_index < refresh_index && refresh_index < effective_state_index && effective_state_index < load_index, diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index fd9afef19..43bb78730 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -69,6 +69,7 @@ pub mod sts; pub mod system; pub mod table_catalog; mod target_descriptor; +mod target_read; pub mod tier; pub mod tls_certificate; pub mod tls_debug; diff --git a/rustfs/src/admin/handlers/notify_runtime_access.rs b/rustfs/src/admin/handlers/notify_runtime_access.rs index cbaf5572d..59a527c44 100644 --- a/rustfs/src/admin/handlers/notify_runtime_access.rs +++ b/rustfs/src/admin/handlers/notify_runtime_access.rs @@ -36,7 +36,7 @@ pub(crate) async fn get_notification_system() -> S3Result S3Result<(Arc, Config)> { +pub(crate) async fn load_notification_mutation_snapshot() -> S3Result<(Arc, Config)> { let system = get_notification_system().await?; let config = system.config_snapshot().await; Ok((system, config)) diff --git a/rustfs/src/admin/handlers/plugins_instances.rs b/rustfs/src/admin/handlers/plugins_instances.rs index 548720651..e4f7d12a6 100644 --- a/rustfs/src/admin/handlers/plugins_instances.rs +++ b/rustfs/src/admin/handlers/plugins_instances.rs @@ -14,20 +14,25 @@ use crate::admin::{ auth::authorize_admin_request, - handlers::audit_runtime_config::{load_server_config_from_store, remove_audit_target_config, set_audit_target_config}, + handlers::audit_runtime_config::{remove_audit_target_config, set_audit_target_config}, handlers::notify_runtime_access::{ - load_notification_config_snapshot, remove_notification_target_config, set_notification_target_config, + load_notification_mutation_snapshot, remove_notification_target_config, set_notification_target_config, }, handlers::target_descriptor::{ AdminTargetSpec, RuntimeHealthStatus, TargetEndpointSource, TargetInstanceReadModel, admin_target_spec_from_builtin, - build_enabled_target_kvs, build_json_response, collect_runtime_statuses, collect_target_instances, find_target_instance, + build_enabled_target_kvs, build_json_response, collect_target_instances, find_target_instance, target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason, }, + handlers::target_read::{ + collect_audit_runtime_statuses, collect_notification_runtime_statuses, load_server_config_from_store, + load_target_read_snapshot, + }, plugin_contract::{ PluginContractDomain, PluginInstanceDetail, PluginInstanceDiagnostic, PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, PluginInstanceEntry, PluginInstanceSource, PluginInstancesResponse, }, router::{AdminOperation, Operation, S3Router}, + storage_api::s3::{self, S3ErrorCode}, }; use crate::server::{ ADMIN_PREFIX, is_audit_module_enabled, is_notify_module_enabled, refresh_audit_module_enabled, refresh_notify_module_enabled, @@ -35,7 +40,6 @@ use crate::server::{ }; use hyper::{Method, StatusCode}; use matchit::Params; -use rustfs_audit::audit_system; use rustfs_config::audit::AUDIT_ROUTE_PREFIX; use rustfs_config::notify::NOTIFY_ROUTE_PREFIX; use rustfs_config::server_config::{Config, KVS}; @@ -597,13 +601,15 @@ fn plugin_instance_mutation_block_reason( /// Reloads persisted module switches from the config store. Call once per /// admin request before evaluating [`module_disabled_block_reason`] so that /// list/detail paths do not hit the store once per domain or per instance. -async fn refresh_persisted_module_switches() { +async fn refresh_persisted_module_switches() -> S3Result<()> { if let Err(err) = refresh_persisted_module_switches_from_store().await { warn!( error = %err, "failed to reload persisted module switches before checking plugin instance operation gating" ); + return Err(s3::error(S3ErrorCode::InternalError, "failed to refresh target module state")); } + Ok(()) } fn module_disabled_block_reason(domain: PluginContractDomain, action: &str) -> Option { @@ -619,53 +625,53 @@ fn module_disabled_block_reason(domain: PluginContractDomain, action: &str) -> O } } -async fn plugin_instance_operation_block_reason(context: PluginInstanceDomainContext, action: &str) -> Option { - refresh_persisted_module_switches().await; - module_disabled_block_reason(context.domain, action) +async fn plugin_instance_operation_block_reason(context: PluginInstanceDomainContext, action: &str) -> S3Result> { + refresh_persisted_module_switches().await?; + Ok(module_disabled_block_reason(context.domain, action)) } async fn plugin_instance_runtime_statuses( context: PluginInstanceDomainContext, -) -> S3Result> { +) -> HashMap<(String, String), RuntimeHealthStatus> { match context.domain { - PluginContractDomain::Notify => { - let (ns, _) = load_notification_config_snapshot().await?; - Ok(collect_runtime_statuses(ns.get_target_values().await).await) - } - PluginContractDomain::Audit => { - let mut runtime_statuses = HashMap::new(); - if let Some(system) = audit_system() { - runtime_statuses = collect_runtime_statuses(system.get_target_values().await).await; - } - Ok(runtime_statuses) - } + PluginContractDomain::Notify => collect_notification_runtime_statuses().await, + PluginContractDomain::Audit => collect_audit_runtime_statuses().await, } } -async fn plugin_instance_config_snapshot(context: PluginInstanceDomainContext) -> S3Result { +async fn plugin_instance_mutation_config_snapshot(context: PluginInstanceDomainContext) -> S3Result { match context.domain { - PluginContractDomain::Notify => load_notification_config_snapshot().await.map(|(_, config)| config), + PluginContractDomain::Notify => load_notification_mutation_snapshot().await.map(|(_, config)| config), PluginContractDomain::Audit => load_server_config_from_store().await, } } -async fn collect_domain_instances(context: PluginInstanceDomainContext) -> S3Result> { - let runtime_statuses = plugin_instance_runtime_statuses(context).await?; - let config = plugin_instance_config_snapshot(context).await?; +fn collect_domain_instances( + context: PluginInstanceDomainContext, + config: &Config, + runtime_statuses: HashMap<(String, String), RuntimeHealthStatus>, +) -> S3Result> { let module_disabled_reason = module_disabled_block_reason(context.domain, "listing plugin instances"); let mut entries = Vec::new(); - for instance in collect_target_instances(context.specs, context.route_prefix, &config, runtime_statuses)? { + for instance in collect_target_instances(context.specs, context.route_prefix, config, runtime_statuses)? { entries.push(plugin_instance_list_entry(instance, module_disabled_reason.clone())); } Ok(entries) } pub(super) async fn collect_all_instances() -> S3Result> { - refresh_persisted_module_switches().await; - let (mut notify_instances, audit_instances) = tokio::try_join!( - collect_domain_instances(plugin_instance_domain_context(PluginContractDomain::Notify)), - collect_domain_instances(plugin_instance_domain_context(PluginContractDomain::Audit)) - )?; + refresh_persisted_module_switches().await?; + let notify_context = plugin_instance_domain_context(PluginContractDomain::Notify); + let audit_context = plugin_instance_domain_context(PluginContractDomain::Audit); + let (config, (notify_statuses, audit_statuses)) = load_target_read_snapshot(async { + tokio::join!( + plugin_instance_runtime_statuses(notify_context), + plugin_instance_runtime_statuses(audit_context) + ) + }) + .await?; + let mut notify_instances = collect_domain_instances(notify_context, &config, notify_statuses)?; + let audit_instances = collect_domain_instances(audit_context, &config, audit_statuses)?; notify_instances.extend(audit_instances); notify_instances.sort_by(|a, b| a.service.cmp(&b.service).then_with(|| a.account_id.cmp(&b.account_id))); Ok(notify_instances) @@ -673,8 +679,7 @@ pub(super) async fn collect_all_instances() -> S3Result async fn find_plugin_instance(instance_id: &str) -> S3Result> { let context = plugin_instance_domain_context(parse_plugin_instance_id(instance_id)?.1); - let runtime_statuses = plugin_instance_runtime_statuses(context).await?; - let config = plugin_instance_config_snapshot(context).await?; + let (config, runtime_statuses) = load_target_read_snapshot(plugin_instance_runtime_statuses(context)).await?; find_target_instance(context.specs, context.route_prefix, &config, runtime_statuses, instance_id) } @@ -738,7 +743,7 @@ impl Operation for GetPluginInstanceHandler { .get("id") .ok_or_else(|| s3_error!(InvalidArgument, "missing required parameter: 'id'"))?; - refresh_persisted_module_switches().await; + refresh_persisted_module_switches().await?; let instance = find_plugin_instance(instance_id) .await? .ok_or_else(|| s3_error!(NoSuchKey, "plugin instance not found"))?; @@ -761,7 +766,8 @@ impl Operation for PutPluginInstanceHandler { let resolved = resolve_plugin_instance_target(instance_id)?; let context = resolved.context; - if let Some(reason) = plugin_instance_operation_block_reason(context, "managing plugin instances from the console").await + if let Some(reason) = + plugin_instance_operation_block_reason(context, "managing plugin instances from the console").await? { return Err(s3_error!(InvalidRequest, "{reason}")); } @@ -774,7 +780,7 @@ impl Operation for PutPluginInstanceHandler { let mut body: PluginInstanceBody = serde_json::from_slice(&body_bytes) .map_err(|e| s3_error!(InvalidArgument, "invalid json body for plugin instance config: {}", e))?; - let config_snapshot = plugin_instance_config_snapshot(context).await?; + let config_snapshot = plugin_instance_mutation_config_snapshot(context).await?; if let Some(reason) = plugin_instance_mutation_block_reason( context, &config_snapshot, @@ -821,12 +827,13 @@ impl Operation for DeletePluginInstanceHandler { let resolved = resolve_plugin_instance_target(instance_id)?; let context = resolved.context; - if let Some(reason) = plugin_instance_operation_block_reason(context, "managing plugin instances from the console").await + if let Some(reason) = + plugin_instance_operation_block_reason(context, "managing plugin instances from the console").await? { return Err(s3_error!(InvalidRequest, "{reason}")); } - let config_snapshot = plugin_instance_config_snapshot(context).await?; + let config_snapshot = plugin_instance_mutation_config_snapshot(context).await?; if let Some(reason) = plugin_instance_mutation_block_reason( context, &config_snapshot, diff --git a/rustfs/src/admin/handlers/target_read.rs b/rustfs/src/admin/handlers/target_read.rs new file mode 100644 index 000000000..9b6ddff60 --- /dev/null +++ b/rustfs/src/admin/handlers/target_read.rs @@ -0,0 +1,61 @@ +// 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::{ + handlers::target_descriptor::{EndpointKey, RuntimeHealthStatus, collect_runtime_statuses}, + runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context}, + storage_api::{ + config::read_admin_config_without_migrate, + s3::{self, S3Error, S3ErrorCode, S3Result}, + }, +}; +use rustfs_config::server_config::Config; +use std::{collections::HashMap, future::Future}; + +pub(crate) async fn load_server_config_from_store_for_context(context: Option<&AppContext>) -> S3Result { + let Some(store) = current_object_store_handle_for_context(context) else { + return Ok(Config::new()); + }; + + read_admin_config_without_migrate(store) + .await + .map_err(|e| s3::error(S3ErrorCode::InternalError, format!("failed to read server config: {e}"))) +} + +pub(crate) async fn load_server_config_from_store() -> S3Result { + let context = current_app_context(); + load_server_config_from_store_for_context(context.as_deref()).await +} + +/// Reads one durable configuration alongside passive runtime observations. +/// Neither source waits for target initialization or lifecycle convergence. +pub(crate) async fn load_target_read_snapshot(runtime: impl Future) -> S3Result<(Config, T)> { + tokio::try_join!(load_server_config_from_store(), async { Ok::<_, S3Error>(runtime.await) }) +} + +pub(crate) async fn collect_notification_runtime_statuses() -> HashMap { + if let Some(system) = rustfs_notify::notification_system() { + collect_runtime_statuses(system.get_target_values().await).await + } else { + HashMap::new() + } +} + +pub(crate) async fn collect_audit_runtime_statuses() -> HashMap { + if let Some(system) = rustfs_audit::audit_system() { + collect_runtime_statuses(system.get_target_values().await).await + } else { + HashMap::new() + } +} diff --git a/scripts/check_s3s_footprint.sh b/scripts/check_s3s_footprint.sh index 6413d65e5..46316efc1 100755 --- a/scripts/check_s3s_footprint.sh +++ b/scripts/check_s3s_footprint.sh @@ -65,7 +65,7 @@ cd "$(dirname "$0")/.." # other reductions. The merged tree measures 209 import files and 1586 # invocation lines, tightening main's reviewed 1592-line baseline by six. S3S_IMPORT_FILES_BASELINE=209 -S3_ERROR_LINES_BASELINE=1572 +S3_ERROR_LINES_BASELINE=1571 # ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not # know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming* # client was extracted to crates/s3-client, where s3s usage is legitimate;