diff --git a/README.md b/README.md index c96cb3c..e99834f 100644 --- a/README.md +++ b/README.md @@ -280,6 +280,17 @@ rc admin rebalance start local rc admin rebalance status local rc admin rebalance stop local +# Site replication across clusters (peer sites given as alias names) +rc admin replicate add site1 site2 +rc admin replicate info site1 +rc admin replicate status site1 +rc admin replicate remove site1 --all + +# Service control (restart/stop perform a graceful shutdown; +# a process manager such as systemd relaunches after restart) +rc admin service restart local +rc admin service stop local + # JSON output rc admin info cluster local --json rc admin heal status local --json diff --git a/crates/cli/src/commands/admin/mod.rs b/crates/cli/src/commands/admin/mod.rs index 8a71baf..1e3d8ff 100644 --- a/crates/cli/src/commands/admin/mod.rs +++ b/crates/cli/src/commands/admin/mod.rs @@ -12,6 +12,8 @@ mod info; mod policy; mod pool; mod rebalance; +mod replicate; +mod service; mod service_account; mod user; @@ -68,6 +70,14 @@ pub enum AdminCommands { /// Inspect access key identities #[command(name = "access-key", subcommand)] AccessKey(access_key::AccessKeyCommands), + + /// Control the server process (restart, stop, freeze, unfreeze) + #[command(subcommand)] + Service(service::ServiceCommands), + + /// Manage site replication across clusters + #[command(subcommand)] + Replicate(replicate::ReplicateCommands), } /// Execute an admin subcommand @@ -92,6 +102,10 @@ pub async fn execute(cmd: AdminCommands, output_config: OutputConfig) -> ExitCod AdminCommands::AccessKey(access_key_cmd) => { access_key::execute(access_key_cmd, &formatter).await } + AdminCommands::Service(service_cmd) => service::execute(service_cmd, &formatter).await, + AdminCommands::Replicate(replicate_cmd) => { + replicate::execute(replicate_cmd, &formatter).await + } } } diff --git a/crates/cli/src/commands/admin/replicate.rs b/crates/cli/src/commands/admin/replicate.rs new file mode 100644 index 0000000..9fa6d96 --- /dev/null +++ b/crates/cli/src/commands/admin/replicate.rs @@ -0,0 +1,319 @@ +//! Site replication commands (`rc admin replicate`). +//! +//! Mirrors `mc admin replicate` against the RustFS native admin API +//! (`/rustfs/admin/v3/site-replication/*`). Peer sites are given as +//! configured alias names; their endpoints and credentials are resolved +//! from the local alias store. + +use clap::Subcommand; + +use super::get_admin_client; +use crate::exit_code::ExitCode; +use crate::output::Formatter; +use rc_core::AliasManager; +use rc_core::admin::{AdminApi, PeerSiteSpec, SiteRemoveSpec, SiteStatusOptions}; + +/// Site replication subcommands +#[derive(Subcommand, Debug)] +pub enum ReplicateCommands { + /// Add sites (given as alias names) to a site replication cluster + Add(AddArgs), + + /// Show site replication configuration + Info(InfoArgs), + + /// Show site replication status + Status(StatusArgs), + + /// Remove sites from the site replication cluster + #[command(alias = "rm")] + Remove(RemoveArgs), +} + +#[derive(clap::Args, Debug)] +pub struct AddArgs { + /// Alias names of the sites to link (first is the request target) + #[arg(required = true, num_args = 2..)] + pub aliases: Vec, +} + +#[derive(clap::Args, Debug)] +pub struct InfoArgs { + /// Alias name of the server + pub alias: String, +} + +#[derive(clap::Args, Debug)] +pub struct StatusArgs { + /// Alias name of the server + pub alias: String, + + /// Include bucket replication status + #[arg(long)] + pub buckets: bool, + + /// Include user replication status + #[arg(long)] + pub users: bool, + + /// Include group replication status + #[arg(long)] + pub groups: bool, + + /// Include policy replication status + #[arg(long)] + pub policies: bool, + + /// Include replication metrics + #[arg(long)] + pub metrics: bool, +} + +#[derive(clap::Args, Debug)] +pub struct RemoveArgs { + /// Alias name of the server to send the request to + pub alias: String, + + /// Site names to remove from the cluster + #[arg(long = "site", conflicts_with = "all")] + pub sites: Vec, + + /// Remove all sites (dissolve the cluster) + #[arg(long)] + pub all: bool, +} + +/// Execute a site replication subcommand +pub async fn execute(cmd: ReplicateCommands, formatter: &Formatter) -> ExitCode { + match cmd { + ReplicateCommands::Add(args) => execute_add(args, formatter).await, + ReplicateCommands::Info(args) => execute_info(args, formatter).await, + ReplicateCommands::Status(args) => execute_status(args, formatter).await, + ReplicateCommands::Remove(args) => execute_remove(args, formatter).await, + } +} + +fn resolve_peer_sites( + aliases: &[String], + formatter: &Formatter, +) -> Result, ExitCode> { + let alias_manager = match AliasManager::new() { + Ok(am) => am, + Err(e) => { + formatter.error(&format!("Failed to load aliases: {e}")); + return Err(ExitCode::GeneralError); + } + }; + + let mut sites = Vec::with_capacity(aliases.len()); + for name in aliases { + let alias = match alias_manager.get(name) { + Ok(a) => a, + Err(e) => { + formatter.error(&format!("Unknown alias `{name}`: {e}")); + return Err(ExitCode::GeneralError); + } + }; + if alias.anonymous || alias.access_key.is_empty() { + formatter.error(&format!( + "Alias `{name}` has no credentials; site replication requires credentialed aliases" + )); + return Err(ExitCode::GeneralError); + } + sites.push(PeerSiteSpec { + name: alias.name.clone(), + endpoint: alias.endpoint.clone(), + access_key: alias.access_key.clone(), + secret_key: alias.secret_key.clone(), + skip_tls_verify: alias.insecure, + ca_cert_pem: String::new(), + }); + } + Ok(sites) +} + +async fn execute_add(args: AddArgs, formatter: &Formatter) -> ExitCode { + let sites = match resolve_peer_sites(&args.aliases, formatter) { + Ok(s) => s, + Err(code) => return code, + }; + + let client = match get_admin_client(&args.aliases[0], formatter) { + Ok(c) => c, + Err(code) => return code, + }; + + match client.site_replication_add(&sites).await { + Ok(result) => { + if formatter.is_json() { + formatter.json(&result); + } else { + formatter.success(&format!( + "Site replication configured across: {}", + args.aliases.join(", ") + )); + if let Some(status) = result.get("status").and_then(|v| v.as_str()) { + formatter.println(&format!(" Status: {status}")); + } + } + ExitCode::Success + } + Err(e) => { + formatter.error(&format!("Failed to add site replication: {e}")); + ExitCode::GeneralError + } + } +} + +async fn execute_info(args: InfoArgs, formatter: &Formatter) -> ExitCode { + let client = match get_admin_client(&args.alias, formatter) { + Ok(c) => c, + Err(code) => return code, + }; + + match client.site_replication_info().await { + Ok(info) => { + if formatter.is_json() { + formatter.json(&info); + } else { + let enabled = info + .get("enabled") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + if !enabled { + formatter.println("Site replication is not configured."); + } else { + if let Some(name) = info.get("name").and_then(|v| v.as_str()) { + formatter.println(&format!("Cluster: {name}")); + } + if let Some(sites) = info.get("sites").and_then(|v| v.as_array()) { + formatter.println("Sites:"); + for site in sites { + let name = site.get("name").and_then(|v| v.as_str()).unwrap_or("-"); + let endpoint = + site.get("endpoint").and_then(|v| v.as_str()).unwrap_or("-"); + formatter.println(&format!(" {name} {endpoint}")); + } + } + } + } + ExitCode::Success + } + Err(e) => { + formatter.error(&format!("Failed to get site replication info: {e}")); + ExitCode::GeneralError + } + } +} + +async fn execute_status(args: StatusArgs, formatter: &Formatter) -> ExitCode { + let client = match get_admin_client(&args.alias, formatter) { + Ok(c) => c, + Err(code) => return code, + }; + + // Default to the summary views when no specific flag is requested, + // matching `mc admin replicate status` behavior. + let none_requested = + !(args.buckets || args.users || args.groups || args.policies || args.metrics); + let options = SiteStatusOptions { + buckets: args.buckets || none_requested, + users: args.users || none_requested, + groups: args.groups || none_requested, + policies: args.policies || none_requested, + metrics: args.metrics, + peer_state: false, + ilm_expiry_rules: false, + }; + + match client.site_replication_status(&options).await { + Ok(status) => { + if formatter.is_json() { + formatter.json(&status); + } else { + print_replication_status(&status, formatter); + } + ExitCode::Success + } + Err(e) => { + formatter.error(&format!("Failed to get site replication status: {e}")); + ExitCode::GeneralError + } + } +} + +fn print_replication_status(status: &serde_json::Value, formatter: &Formatter) { + let enabled = status + .get("enabled") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + if !enabled { + formatter.println("Site replication is not configured."); + return; + } + + if let Some(sites) = status.get("Sites").and_then(|v| v.as_object()) { + formatter.println("Sites:"); + for site in sites.values() { + let name = site.get("name").and_then(|v| v.as_str()).unwrap_or("-"); + let endpoint = site.get("endpoint").and_then(|v| v.as_str()).unwrap_or("-"); + formatter.println(&format!(" {name} {endpoint}")); + } + } + + let count = |key: &str| status.get(key).and_then(|v| v.as_u64()).unwrap_or_default(); + formatter.println(&format!( + "Replicated entities: buckets={} users={} groups={} policies={}", + count("MaxBuckets"), + count("MaxUsers"), + count("MaxGroups"), + count("MaxPolicies"), + )); + + if let Some(errors) = status.get("PeerErrors").and_then(|v| v.as_object()) + && !errors.is_empty() + { + formatter.println(&format!( + "Peer errors reported by {} site(s):", + errors.len() + )); + for (site, error) in errors { + formatter.println(&format!(" {site}: {error}")); + } + } +} + +async fn execute_remove(args: RemoveArgs, formatter: &Formatter) -> ExitCode { + if !args.all && args.sites.is_empty() { + formatter.error("Specify --site (repeatable) or --all"); + return ExitCode::UsageError; + } + + let client = match get_admin_client(&args.alias, formatter) { + Ok(c) => c, + Err(code) => return code, + }; + + let spec = SiteRemoveSpec { + site_names: args.sites.clone(), + remove_all: args.all, + }; + + match client.site_replication_remove(&spec).await { + Ok(result) => { + if formatter.is_json() { + formatter.json(&result); + } else { + formatter.success("Site replication removal requested."); + if let Some(status) = result.get("status").and_then(|v| v.as_str()) { + formatter.println(&format!(" Status: {status}")); + } + } + ExitCode::Success + } + Err(e) => { + formatter.error(&format!("Failed to remove site replication: {e}")); + ExitCode::GeneralError + } + } +} diff --git a/crates/cli/src/commands/admin/service.rs b/crates/cli/src/commands/admin/service.rs new file mode 100644 index 0000000..4372e10 --- /dev/null +++ b/crates/cli/src/commands/admin/service.rs @@ -0,0 +1,76 @@ +//! Service control commands (restart, stop, freeze, unfreeze). +//! +//! These map to `POST /rustfs/admin/v3/service?action=` on the +//! server. `restart` and `stop` both trigger a graceful shutdown; a process +//! manager (systemd, k8s) is responsible for relaunching after `restart`. + +use clap::Subcommand; + +use super::get_admin_client; +use crate::exit_code::ExitCode; +use crate::output::Formatter; +use rc_core::admin::AdminApi; + +/// Service control subcommands +#[derive(Subcommand, Debug)] +pub enum ServiceCommands { + /// Restart the server (graceful shutdown; process manager relaunches) + Restart(ActionArgs), + + /// Stop the server gracefully + Stop(ActionArgs), + + /// Set the service freeze flag (advisory) + Freeze(ActionArgs), + + /// Clear the service freeze flag + Unfreeze(ActionArgs), +} + +#[derive(clap::Args, Debug)] +pub struct ActionArgs { + /// Alias name of the server + pub alias: String, +} + +/// Execute a service subcommand +pub async fn execute(cmd: ServiceCommands, formatter: &Formatter) -> ExitCode { + let (action, args) = match cmd { + ServiceCommands::Restart(args) => ("restart", args), + ServiceCommands::Stop(args) => ("stop", args), + ServiceCommands::Freeze(args) => ("freeze", args), + ServiceCommands::Unfreeze(args) => ("unfreeze", args), + }; + + let client = match get_admin_client(&args.alias, formatter) { + Ok(c) => c, + Err(code) => return code, + }; + + match client.service_action(action).await { + Ok(result) => { + if formatter.is_json() { + formatter.json(&result); + } else if result.accepted { + formatter.success(&format!( + "Service {action} accepted on `{}`: {}", + args.alias, result.message + )); + } else { + formatter.error(&format!( + "Service {action} not accepted on `{}`: {}", + args.alias, result.message + )); + return ExitCode::GeneralError; + } + ExitCode::Success + } + Err(e) => { + formatter.error(&format!( + "Failed to {action} service on `{}`: {e}", + args.alias + )); + ExitCode::GeneralError + } + } +} diff --git a/crates/cli/tests/admin_replicate.rs b/crates/cli/tests/admin_replicate.rs new file mode 100644 index 0000000..e9969e2 --- /dev/null +++ b/crates/cli/tests/admin_replicate.rs @@ -0,0 +1,233 @@ +#![cfg(not(windows))] + +mod admin_support; + +use std::process::Command; +use std::time::Duration; + +use admin_support::{rc_binary, rc_host_alias, start_admin_test_server}; + +#[test] +fn replicate_info_dispatches_to_site_replication_info() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server(r#"{"enabled":false}"#); + + let output = Command::new(rc_binary()) + .args(["--json", "admin", "replicate", "info", "myalias"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let stdout = String::from_utf8(output.stdout).expect("stdout should be UTF-8"); + let payload: serde_json::Value = serde_json::from_str(&stdout).expect("JSON output"); + assert_eq!(payload["enabled"], false); + + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + assert_eq!(request.method, "GET"); + assert_eq!(request.target, "/rustfs/admin/v3/site-replication/info"); + + handle.join().expect("admin test server finished"); +} + +#[test] +fn replicate_info_human_output_lists_sites() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server( + r#"{"enabled":true,"name":"site1","sites":[{"name":"site1","endpoint":"http://10.0.0.5:9000"},{"name":"site2","endpoint":"http://10.0.0.6:9000"}]}"#, + ); + + let output = Command::new(rc_binary()) + .args(["admin", "replicate", "info", "myalias"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let stdout = String::from_utf8(output.stdout).expect("stdout should be UTF-8"); + assert!(stdout.contains("site1"), "stdout: {stdout}"); + assert!(stdout.contains("http://10.0.0.6:9000"), "stdout: {stdout}"); + + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + handle.join().expect("admin test server finished"); +} + +#[test] +fn replicate_status_requests_default_summary_sections() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server( + r#"{"enabled":true,"MaxBuckets":2,"MaxUsers":1,"MaxGroups":0,"MaxPolicies":5,"Sites":{"dep-1":{"name":"site1","endpoint":"http://10.0.0.5:9000"}}}"#, + ); + + let output = Command::new(rc_binary()) + .args(["--json", "admin", "replicate", "status", "myalias"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let stdout = String::from_utf8(output.stdout).expect("stdout should be UTF-8"); + let payload: serde_json::Value = serde_json::from_str(&stdout).expect("JSON output"); + assert_eq!(payload["enabled"], true); + assert_eq!(payload["MaxBuckets"], 2); + + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + assert_eq!(request.method, "GET"); + assert_eq!( + request.target, + "/rustfs/admin/v3/site-replication/status?buckets=true&users=true&groups=true&policies=true" + ); + + handle.join().expect("admin test server finished"); +} + +#[test] +fn replicate_status_forwards_selected_section_flags() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server(r#"{"enabled":true}"#); + + let output = Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "status", + "myalias", + "--buckets", + "--metrics", + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + assert_eq!( + request.target, + "/rustfs/admin/v3/site-replication/status?buckets=true&metrics=true" + ); + + handle.join().expect("admin test server finished"); +} + +#[test] +fn replicate_add_dispatches_with_resolved_alias_sites() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server( + r#"{"success":true,"status":"Requested sites were configured for replication successfully."}"#, + ); + + let output = Command::new(rc_binary()) + .args(["--json", "admin", "replicate", "add", "sitea", "siteb"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_sitea", rc_host_alias(&endpoint)) + .env("RC_HOST_siteb", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let stdout = String::from_utf8(output.stdout).expect("stdout should be UTF-8"); + let payload: serde_json::Value = serde_json::from_str(&stdout).expect("JSON output"); + assert_eq!(payload["success"], true); + + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + assert_eq!(request.method, "PUT"); + assert_eq!(request.target, "/rustfs/admin/v3/site-replication/add"); + + handle.join().expect("admin test server finished"); +} + +#[test] +fn replicate_add_rejects_single_alias() { + let config_dir = tempfile::tempdir().expect("create config dir"); + + let output = Command::new(rc_binary()) + .args(["admin", "replicate", "add", "onlyone"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .output() + .expect("run rc command"); + + assert!(!output.status.success()); +} + +#[test] +fn replicate_remove_all_dispatches_to_site_replication_remove() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server( + r#"{"status":"Requested site(s) were removed from cluster replication successfully."}"#, + ); + + let output = Command::new(rc_binary()) + .args(["--json", "admin", "replicate", "remove", "myalias", "--all"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + assert_eq!(request.method, "PUT"); + assert_eq!(request.target, "/rustfs/admin/v3/site-replication/remove"); + + handle.join().expect("admin test server finished"); +} + +#[test] +fn replicate_remove_requires_site_or_all() { + let config_dir = tempfile::tempdir().expect("create config dir"); + + let output = Command::new(rc_binary()) + .args(["admin", "replicate", "remove", "myalias"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .output() + .expect("run rc command"); + + assert!(!output.status.success()); + assert_eq!(output.status.code(), Some(2)); +} diff --git a/crates/cli/tests/admin_service.rs b/crates/cli/tests/admin_service.rs new file mode 100644 index 0000000..e0bcd65 --- /dev/null +++ b/crates/cli/tests/admin_service.rs @@ -0,0 +1,118 @@ +#![cfg(not(windows))] + +mod admin_support; + +use std::process::Command; +use std::time::Duration; + +use admin_support::{rc_binary, rc_host_alias, start_admin_test_server}; + +fn run_service_action(action: &str, response_body: &'static str) -> (serde_json::Value, String) { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server(response_body); + + let output = Command::new(rc_binary()) + .args(["--json", "admin", "service", action, "myalias"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let stdout = String::from_utf8(output.stdout).expect("stdout should be UTF-8"); + let payload: serde_json::Value = serde_json::from_str(&stdout).expect("JSON output"); + + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + assert_eq!(request.method, "POST"); + handle.join().expect("admin test server finished"); + + (payload, request.target) +} + +#[test] +fn service_stop_dispatches_to_service_action_stop() { + let (payload, target) = run_service_action( + "stop", + r#"{"action":"stop","accepted":true,"effective":true,"message":"graceful shutdown initiated"}"#, + ); + + assert_eq!(target, "/rustfs/admin/v3/service?action=stop"); + assert_eq!(payload["action"], "stop"); + assert_eq!(payload["accepted"], true); + assert_eq!(payload["effective"], true); + assert_eq!(payload["message"], "graceful shutdown initiated"); +} + +#[test] +fn service_restart_dispatches_to_service_action_restart() { + let (payload, target) = run_service_action( + "restart", + r#"{"action":"restart","accepted":true,"effective":true,"message":"graceful shutdown initiated; the supervising process manager is responsible for relaunch"}"#, + ); + + assert_eq!(target, "/rustfs/admin/v3/service?action=restart"); + assert_eq!(payload["action"], "restart"); + assert_eq!(payload["accepted"], true); +} + +#[test] +fn service_freeze_reports_advisory_effectiveness() { + let (payload, target) = run_service_action( + "freeze", + r#"{"action":"freeze","accepted":true,"effective":false,"message":"freeze flag recorded, but RustFS does not yet gate request admission on it (advisory only)"}"#, + ); + + assert_eq!(target, "/rustfs/admin/v3/service?action=freeze"); + assert_eq!(payload["accepted"], true); + assert_eq!(payload["effective"], false); +} + +#[test] +fn service_unfreeze_dispatches_to_service_action_unfreeze() { + let (payload, target) = run_service_action( + "unfreeze", + r#"{"action":"unfreeze","accepted":true,"effective":false,"message":"freeze flag cleared (advisory only)"}"#, + ); + + assert_eq!(target, "/rustfs/admin/v3/service?action=unfreeze"); + assert_eq!(payload["action"], "unfreeze"); +} + +#[test] +fn service_stop_human_output_reports_success() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server( + r#"{"action":"stop","accepted":true,"effective":true,"message":"graceful shutdown initiated"}"#, + ); + + let output = Command::new(rc_binary()) + .args(["admin", "service", "stop", "myalias"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + + let stdout = String::from_utf8(output.stdout).expect("stdout should be UTF-8"); + assert!( + stdout.contains("graceful shutdown initiated"), + "stdout: {stdout}" + ); + + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + handle.join().expect("admin test server finished"); +} diff --git a/crates/core/src/admin/mod.rs b/crates/core/src/admin/mod.rs index 10394b1..969b232 100644 --- a/crates/core/src/admin/mod.rs +++ b/crates/core/src/admin/mod.rs @@ -4,6 +4,7 @@ //! IAM users, policies, groups, service accounts, and cluster operations. mod cluster; +mod site; pub mod tier; mod types; @@ -15,6 +16,7 @@ pub use cluster::{ RebalancePoolProgress, RebalancePoolStatus, RebalanceStartResult, RebalanceStatus, ServerInfo, UsageInfo, }; +pub use site::{PeerSiteSpec, ServiceActionResult, SiteRemoveSpec, SiteStatusOptions}; pub use tier::{ TierAliyun, TierAzure, TierConfig, TierCreds, TierGCS, TierHuaweicloud, TierMinIO, TierR2, TierRustFS, TierS3, TierTencent, TierType, @@ -223,6 +225,28 @@ pub trait AdminApi: Send + Sync { /// Get replication metrics for a bucket async fn replication_metrics(&self, bucket: &str) -> Result; + + // ==================== Service Control Operations ==================== + + /// Request a service action (restart, stop, freeze, unfreeze) + async fn service_action(&self, action: &str) -> Result; + + // ==================== Site Replication Operations ==================== + + /// Get current site replication configuration + async fn site_replication_info(&self) -> Result; + + /// Add peer sites to the site replication cluster + async fn site_replication_add(&self, sites: &[PeerSiteSpec]) -> Result; + + /// Get site replication status + async fn site_replication_status( + &self, + options: &SiteStatusOptions, + ) -> Result; + + /// Remove sites from the site replication cluster + async fn site_replication_remove(&self, spec: &SiteRemoveSpec) -> Result; } #[cfg(test)] diff --git a/crates/core/src/admin/site.rs b/crates/core/src/admin/site.rs new file mode 100644 index 0000000..24c1b2a --- /dev/null +++ b/crates/core/src/admin/site.rs @@ -0,0 +1,65 @@ +//! Service control and site replication admin types. +//! +//! Wire formats mirror the RustFS server's `crates/madmin` definitions +//! (MinIO-admin-compatible JSON field names). + +use serde::{Deserialize, Serialize}; + +/// A peer site definition for `site-replication/add`. +/// +/// Field names follow the MinIO admin wire format (note: the endpoint +/// serializes as `endpoints`). +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct PeerSiteSpec { + #[serde(default)] + pub name: String, + #[serde(rename = "endpoints", default)] + pub endpoint: String, + #[serde(rename = "accessKey", default)] + pub access_key: String, + #[serde(rename = "secretKey", default)] + pub secret_key: String, + #[serde(rename = "skipTlsVerify", default)] + pub skip_tls_verify: bool, + #[serde( + rename = "caCertPem", + default, + skip_serializing_if = "String::is_empty" + )] + pub ca_cert_pem: String, +} + +/// Response from `POST /service?action=...`. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ServiceActionResult { + #[serde(default)] + pub action: String, + #[serde(default)] + pub accepted: bool, + /// Whether the action takes real effect on this build (vs advisory only). + #[serde(default)] + pub effective: bool, + #[serde(default)] + pub message: String, +} + +/// Options for `site-replication/status`. +#[derive(Debug, Clone, Default)] +pub struct SiteStatusOptions { + pub buckets: bool, + pub users: bool, + pub groups: bool, + pub policies: bool, + pub metrics: bool, + pub peer_state: bool, + pub ilm_expiry_rules: bool, +} + +/// Request body for `site-replication/remove`. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct SiteRemoveSpec { + #[serde(rename = "sites", default, skip_serializing_if = "Vec::is_empty")] + pub site_names: Vec, + #[serde(rename = "all", default)] + pub remove_all: bool, +} diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index 8616b0c..7d0d9a7 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -12,9 +12,10 @@ use aws_sigv4::sign::v4; use rc_core::admin::{ AccessKeyInfo, AdminApi, BucketQuota, ClusterInfo, CreateServiceAccountRequest, DecommissionPoolStatus, DecommissionStatus, Group, GroupStatus, HealRuntimeState, HealScanMode, - HealStartRequest, HealStatus, HealTaskRequest, Policy, PolicyEntity, PolicyInfo, PoolStatus, - PoolTarget, RebalanceStartResult, RebalanceStatus, ServiceAccount, - ServiceAccountCreateResponse, UpdateGroupMembersRequest, User, UserStatus, + HealStartRequest, HealStatus, HealTaskRequest, PeerSiteSpec, Policy, PolicyEntity, PolicyInfo, + PoolStatus, PoolTarget, RebalanceStartResult, RebalanceStatus, ServiceAccount, + ServiceAccountCreateResponse, ServiceActionResult, SiteRemoveSpec, SiteStatusOptions, + UpdateGroupMembersRequest, User, UserStatus, }; use rc_core::{Alias, Error, Result}; use reqwest::header::{CONTENT_TYPE, HOST, HeaderMap, HeaderName, HeaderValue}; @@ -1209,6 +1210,51 @@ impl AdminApi for AdminClient { self.request(Method::GET, "/replicationmetrics", Some(query), None) .await } + + async fn service_action(&self, action: &str) -> Result { + let query: &[(&str, &str)] = &[("action", action)]; + self.request(Method::POST, "/service", Some(query), None) + .await + } + + async fn site_replication_info(&self) -> Result { + self.request(Method::GET, "/site-replication/info", None, None) + .await + } + + async fn site_replication_add(&self, sites: &[PeerSiteSpec]) -> Result { + let body = serde_json::to_vec(sites).map_err(Error::Json)?; + self.request(Method::PUT, "/site-replication/add", None, Some(&body)) + .await + } + + async fn site_replication_status( + &self, + options: &SiteStatusOptions, + ) -> Result { + let mut query: Vec<(&str, &str)> = Vec::new(); + for (flag, enabled) in [ + ("buckets", options.buckets), + ("users", options.users), + ("groups", options.groups), + ("policies", options.policies), + ("metrics", options.metrics), + ("peer-state", options.peer_state), + ("ilm-expiry-rules", options.ilm_expiry_rules), + ] { + if enabled { + query.push((flag, "true")); + } + } + self.request(Method::GET, "/site-replication/status", Some(&query), None) + .await + } + + async fn site_replication_remove(&self, spec: &SiteRemoveSpec) -> Result { + let body = serde_json::to_vec(spec).map_err(Error::Json)?; + self.request(Method::PUT, "/site-replication/remove", None, Some(&body)) + .await + } } #[cfg(test)] diff --git a/docs/reference/rc/admin.md b/docs/reference/rc/admin.md index c09d9af..57fe286 100644 --- a/docs/reference/rc/admin.md +++ b/docs/reference/rc/admin.md @@ -2,7 +2,7 @@ ## Purpose -The `rc admin` operation manages the RustFS Admin API, including cluster information, healing, pools, expansion, decommissioning, rebalance workflows, IAM users, policies, groups, and service accounts. +The `rc admin` operation manages the RustFS Admin API, including cluster information, healing, pools, expansion, decommissioning, rebalance workflows, IAM users, policies, groups, service accounts, site replication, and service control. `rc admin` does not implement the MinIO Admin API. MinIO aliases remain available to S3 data commands, but MinIO administrative operations require a MinIO-compatible admin client. @@ -21,6 +21,10 @@ rc admin user ... rc admin policy ... rc admin group ... rc admin service-account ... +rc admin service +rc admin replicate add [...] +rc admin replicate [OPTIONS] +rc admin replicate remove <--all|--site > ``` ## Commands @@ -37,6 +41,8 @@ rc admin service-account ... | `policy` | Manage IAM policies and attachments. | | `group` | Manage IAM groups and group membership. | | `service-account` | Manage service accounts. | +| `service` | Control the server process: restart, stop, freeze, unfreeze. | +| `replicate` | Manage site replication across clusters. | ## Examples @@ -111,6 +117,21 @@ Create a service account with a policy file: rc admin service-account create local SA_ACCESS_KEY SA_SECRET_KEY --policy ./policy.json ``` +Link two sites for site replication and check the result: + +```bash +rc admin replicate add site1 site2 +rc admin replicate info site1 +rc admin replicate status site1 +``` + +Gracefully stop or restart a server: + +```bash +rc admin service stop local +rc admin service restart local +``` + ## Behavior Admin operations use the configured alias to create a RustFS admin client. The credentials behind the alias must have permissions for the requested administrative API. The command accepts aliases with or without a trailing slash. @@ -167,6 +188,43 @@ Use `rc admin pool list ` or `rc admin pool status ` to find pool `rc admin expand` is an alias-oriented workflow for the same post-expansion rebalance step. The `expand` command is also available as `scale`. +## Service Control Workflow + +`rc admin service` controls the server process behind an alias. + +| Command | Description | +| --- | --- | +| `rc admin service restart ` | Request a graceful shutdown for restart. The supervising process manager (systemd, Kubernetes) is responsible for relaunching the binary. | +| `rc admin service stop ` | Request a graceful shutdown. | +| `rc admin service freeze ` | Set the service freeze flag. Currently advisory: the server records the flag but does not yet gate request admission on it. | +| `rc admin service unfreeze ` | Clear the service freeze flag. | + +The server response reports whether the action was `accepted` and whether it is `effective` on the current build. RustFS has no in-process supervisor, so `restart` and `stop` both perform a graceful stop; `restart` relies on the process manager to bring the server back up. + +## Site Replication Workflow + +`rc admin replicate` manages multi-cluster site replication. Peer sites are given as configured alias names; their endpoints and credentials are resolved from the local alias store, so every participating site needs an alias with root credentials before running `add`. + +| Command | Description | +| --- | --- | +| `rc admin replicate add [...]` | Link two or more sites into a site replication cluster. The first alias receives the request. | +| `rc admin replicate info ` | Show the current site replication configuration. | +| `rc admin replicate status [OPTIONS]` | Show replication status. Without flags the buckets, users, groups, and policies summaries are requested. | +| `rc admin replicate remove --all` | Dissolve the entire site replication cluster. | +| `rc admin replicate remove --site ` | Remove one or more named sites. Repeat `--site` per name. | + +`status` accepts these section flags: + +| Option | Description | +| --- | --- | +| `--buckets` | Include the bucket replication summary. | +| `--users` | Include the IAM user replication summary. | +| `--groups` | Include the IAM group replication summary. | +| `--policies` | Include the IAM policy replication summary. | +| `--metrics` | Include replication metrics. | + +Site replication requires bucket versioning support on every site and replicates buckets, objects, IAM users, groups, policies, and service accounts across all linked sites. The server rejects loopback peer endpoints unless the deployment explicitly allows them (`RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET=true`), which is intended for local testing only. + Global options shown in command syntax use the same meaning everywhere: | Option | Description |