feat(admin): add service control and site replication commands (#257)

Author: overtrue

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

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
+        }
     }
 }
 

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<String>,
+}
+
+#[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<String>,
+
+    /// 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<Vec<PeerSiteSpec>, 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 <name> (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
+        }
+    }
+}