Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
14 changes: 14 additions & 0 deletions crates/cli/src/commands/admin/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@ mod info;
mod policy;
mod pool;
mod rebalance;
mod replicate;
mod service;
mod service_account;
mod user;

Expand Down Expand Up @@ -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
Expand All @@ -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
}
}
}

Expand Down
319 changes: 319 additions & 0 deletions crates/cli/src/commands/admin/replicate.rs
Original file line number Diff line number Diff line change
@@ -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
}
}
}
Loading
Loading