2021-11-09 11:24:04 +00:00
|
|
|
use garage_util::crdt::Crdt;
|
|
|
|
use garage_util::error::*;
|
2022-05-18 20:24:09 +00:00
|
|
|
use garage_util::formater::format_table;
|
2021-11-09 11:24:04 +00:00
|
|
|
|
|
|
|
use garage_rpc::layout::*;
|
|
|
|
use garage_rpc::system::*;
|
|
|
|
use garage_rpc::*;
|
|
|
|
|
|
|
|
use crate::cli::*;
|
|
|
|
|
|
|
|
pub async fn cli_layout_command_dispatch(
|
|
|
|
cmd: LayoutOperation,
|
|
|
|
system_rpc_endpoint: &Endpoint<SystemRpc, ()>,
|
|
|
|
rpc_host: NodeID,
|
|
|
|
) -> Result<(), Error> {
|
|
|
|
match cmd {
|
|
|
|
LayoutOperation::Assign(configure_opt) => {
|
|
|
|
cmd_assign_role(system_rpc_endpoint, rpc_host, configure_opt).await
|
|
|
|
}
|
|
|
|
LayoutOperation::Remove(remove_opt) => {
|
|
|
|
cmd_remove_role(system_rpc_endpoint, rpc_host, remove_opt).await
|
|
|
|
}
|
|
|
|
LayoutOperation::Show => cmd_show_layout(system_rpc_endpoint, rpc_host).await,
|
|
|
|
LayoutOperation::Apply(apply_opt) => {
|
|
|
|
cmd_apply_layout(system_rpc_endpoint, rpc_host, apply_opt).await
|
|
|
|
}
|
|
|
|
LayoutOperation::Revert(revert_opt) => {
|
|
|
|
cmd_revert_layout(system_rpc_endpoint, rpc_host, revert_opt).await
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn cmd_assign_role(
|
|
|
|
rpc_cli: &Endpoint<SystemRpc, ()>,
|
|
|
|
rpc_host: NodeID,
|
|
|
|
args: AssignRoleOpt,
|
|
|
|
) -> Result<(), Error> {
|
|
|
|
let status = match rpc_cli
|
|
|
|
.call(&rpc_host, &SystemRpc::GetKnownNodes, PRIO_NORMAL)
|
|
|
|
.await??
|
|
|
|
{
|
|
|
|
SystemRpc::ReturnKnownNodes(nodes) => nodes,
|
|
|
|
resp => return Err(Error::Message(format!("Invalid RPC response: {:?}", resp))),
|
|
|
|
};
|
|
|
|
|
2022-05-09 09:14:55 +00:00
|
|
|
let mut layout = fetch_layout(rpc_cli, rpc_host).await?;
|
|
|
|
|
2022-03-16 13:43:04 +00:00
|
|
|
let added_nodes = args
|
|
|
|
.node_ids
|
|
|
|
.iter()
|
2022-05-09 09:14:55 +00:00
|
|
|
.map(|node_id| {
|
|
|
|
find_matching_node(
|
|
|
|
status
|
|
|
|
.iter()
|
|
|
|
.map(|adv| adv.id)
|
|
|
|
.chain(layout.node_ids().iter().cloned()),
|
|
|
|
node_id,
|
|
|
|
)
|
|
|
|
})
|
2022-03-16 13:43:04 +00:00
|
|
|
.collect::<Result<Vec<_>, _>>()?;
|
2021-11-09 11:24:04 +00:00
|
|
|
|
|
|
|
let mut roles = layout.roles.clone();
|
|
|
|
roles.merge(&layout.staging);
|
|
|
|
|
|
|
|
for replaced in args.replace.iter() {
|
|
|
|
let replaced_node = find_matching_node(layout.node_ids().iter().cloned(), replaced)?;
|
|
|
|
match roles.get(&replaced_node) {
|
|
|
|
Some(NodeRoleV(Some(_))) => {
|
|
|
|
layout
|
|
|
|
.staging
|
|
|
|
.merge(&roles.update_mutator(replaced_node, NodeRoleV(None)));
|
|
|
|
}
|
|
|
|
_ => {
|
|
|
|
return Err(Error::Message(format!(
|
|
|
|
"Cannot replace node {:?} as it is not currently in planned layout",
|
|
|
|
replaced_node
|
|
|
|
)));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if args.capacity.is_some() && args.gateway {
|
|
|
|
return Err(Error::Message(
|
|
|
|
"-c and -g are mutually exclusive, please configure node either with c>0 to act as a storage node or with -g to act as a gateway node".into()));
|
|
|
|
}
|
|
|
|
if args.capacity == Some(0) {
|
|
|
|
return Err(Error::Message("Invalid capacity value: 0".into()));
|
|
|
|
}
|
|
|
|
|
2022-03-16 13:43:04 +00:00
|
|
|
for added_node in added_nodes {
|
|
|
|
let new_entry = match roles.get(&added_node) {
|
|
|
|
Some(NodeRoleV(Some(old))) => {
|
|
|
|
let capacity = match args.capacity {
|
|
|
|
Some(c) => Some(c),
|
|
|
|
None if args.gateway => None,
|
|
|
|
None => old.capacity,
|
|
|
|
};
|
|
|
|
let tags = if args.tags.is_empty() {
|
|
|
|
old.tags.clone()
|
|
|
|
} else {
|
|
|
|
args.tags.clone()
|
|
|
|
};
|
|
|
|
NodeRole {
|
|
|
|
zone: args.zone.clone().unwrap_or_else(|| old.zone.to_string()),
|
|
|
|
capacity,
|
|
|
|
tags,
|
|
|
|
}
|
2021-11-09 11:24:04 +00:00
|
|
|
}
|
2022-03-16 13:43:04 +00:00
|
|
|
_ => {
|
|
|
|
let capacity = match args.capacity {
|
|
|
|
Some(c) => Some(c),
|
|
|
|
None if args.gateway => None,
|
|
|
|
None => return Err(Error::Message(
|
|
|
|
"Please specify a capacity with the -c flag, or set node explicitly as gateway with -g".into())),
|
|
|
|
};
|
|
|
|
NodeRole {
|
|
|
|
zone: args
|
|
|
|
.zone
|
|
|
|
.clone()
|
|
|
|
.ok_or("Please specifiy a zone with the -z flag")?,
|
|
|
|
capacity,
|
|
|
|
tags: args.tags.clone(),
|
|
|
|
}
|
2021-11-09 11:24:04 +00:00
|
|
|
}
|
2022-03-16 13:43:04 +00:00
|
|
|
};
|
2021-11-09 11:24:04 +00:00
|
|
|
|
2022-03-16 13:43:04 +00:00
|
|
|
layout
|
|
|
|
.staging
|
|
|
|
.merge(&roles.update_mutator(added_node, NodeRoleV(Some(new_entry))));
|
|
|
|
}
|
2021-11-09 11:24:04 +00:00
|
|
|
|
|
|
|
send_layout(rpc_cli, rpc_host, layout).await?;
|
|
|
|
|
2022-03-16 13:43:04 +00:00
|
|
|
println!("Role changes are staged but not yet commited.");
|
2021-11-09 11:24:04 +00:00
|
|
|
println!("Use `garage layout show` to view staged role changes,");
|
|
|
|
println!("and `garage layout apply` to enact staged changes.");
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn cmd_remove_role(
|
|
|
|
rpc_cli: &Endpoint<SystemRpc, ()>,
|
|
|
|
rpc_host: NodeID,
|
|
|
|
args: RemoveRoleOpt,
|
|
|
|
) -> Result<(), Error> {
|
|
|
|
let mut layout = fetch_layout(rpc_cli, rpc_host).await?;
|
|
|
|
|
|
|
|
let mut roles = layout.roles.clone();
|
|
|
|
roles.merge(&layout.staging);
|
|
|
|
|
|
|
|
let deleted_node =
|
|
|
|
find_matching_node(roles.items().iter().map(|(id, _, _)| *id), &args.node_id)?;
|
|
|
|
|
|
|
|
layout
|
|
|
|
.staging
|
|
|
|
.merge(&roles.update_mutator(deleted_node, NodeRoleV(None)));
|
|
|
|
|
|
|
|
send_layout(rpc_cli, rpc_host, layout).await?;
|
|
|
|
|
|
|
|
println!("Role removal is staged but not yet commited.");
|
|
|
|
println!("Use `garage layout show` to view staged role changes,");
|
|
|
|
println!("and `garage layout apply` to enact staged changes.");
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn cmd_show_layout(
|
|
|
|
rpc_cli: &Endpoint<SystemRpc, ()>,
|
|
|
|
rpc_host: NodeID,
|
|
|
|
) -> Result<(), Error> {
|
|
|
|
let mut layout = fetch_layout(rpc_cli, rpc_host).await?;
|
|
|
|
|
|
|
|
println!("==== CURRENT CLUSTER LAYOUT ====");
|
|
|
|
if !print_cluster_layout(&layout) {
|
|
|
|
println!("No nodes currently have a role in the cluster.");
|
|
|
|
println!("See `garage status` to view available nodes.");
|
|
|
|
}
|
|
|
|
println!();
|
|
|
|
println!("Current cluster layout version: {}", layout.version);
|
|
|
|
|
|
|
|
if print_staging_role_changes(&layout) {
|
|
|
|
layout.roles.merge(&layout.staging);
|
|
|
|
|
|
|
|
println!();
|
|
|
|
println!("==== NEW CLUSTER LAYOUT AFTER APPLYING CHANGES ====");
|
|
|
|
if !print_cluster_layout(&layout) {
|
|
|
|
println!("No nodes have a role in the new layout.");
|
|
|
|
}
|
|
|
|
println!();
|
|
|
|
|
|
|
|
// this will print the stats of what partitions
|
|
|
|
// will move around when we apply
|
|
|
|
if layout.calculate_partition_assignation() {
|
|
|
|
println!("To enact the staged role changes, type:");
|
|
|
|
println!();
|
|
|
|
println!(" garage layout apply --version {}", layout.version + 1);
|
|
|
|
println!();
|
|
|
|
println!(
|
|
|
|
"You can also revert all proposed changes with: garage layout revert --version {}",
|
|
|
|
layout.version + 1
|
|
|
|
);
|
|
|
|
} else {
|
|
|
|
println!("Not enough nodes have an assigned role to maintain enough copies of data.");
|
|
|
|
println!("This new layout cannot yet be applied.");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn cmd_apply_layout(
|
|
|
|
rpc_cli: &Endpoint<SystemRpc, ()>,
|
|
|
|
rpc_host: NodeID,
|
|
|
|
apply_opt: ApplyLayoutOpt,
|
|
|
|
) -> Result<(), Error> {
|
2022-05-24 10:16:39 +00:00
|
|
|
let layout = fetch_layout(rpc_cli, rpc_host).await?;
|
2022-03-16 13:43:04 +00:00
|
|
|
|
2022-05-24 10:16:39 +00:00
|
|
|
let layout = layout.apply_staged_changes(apply_opt.version)?;
|
2021-11-09 11:24:04 +00:00
|
|
|
|
|
|
|
send_layout(rpc_cli, rpc_host, layout).await?;
|
|
|
|
|
|
|
|
println!("New cluster layout with updated role assignation has been applied in cluster.");
|
|
|
|
println!("Data will now be moved around between nodes accordingly.");
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn cmd_revert_layout(
|
|
|
|
rpc_cli: &Endpoint<SystemRpc, ()>,
|
|
|
|
rpc_host: NodeID,
|
|
|
|
revert_opt: RevertLayoutOpt,
|
|
|
|
) -> Result<(), Error> {
|
2022-05-24 10:16:39 +00:00
|
|
|
let layout = fetch_layout(rpc_cli, rpc_host).await?;
|
2021-11-09 11:24:04 +00:00
|
|
|
|
2022-05-24 10:16:39 +00:00
|
|
|
let layout = layout.revert_staged_changes(revert_opt.version)?;
|
2021-11-09 11:24:04 +00:00
|
|
|
|
|
|
|
send_layout(rpc_cli, rpc_host, layout).await?;
|
|
|
|
|
|
|
|
println!("All proposed role changes in cluster layout have been canceled.");
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
// --- utility ---
|
|
|
|
|
|
|
|
pub async fn fetch_layout(
|
|
|
|
rpc_cli: &Endpoint<SystemRpc, ()>,
|
|
|
|
rpc_host: NodeID,
|
|
|
|
) -> Result<ClusterLayout, Error> {
|
|
|
|
match rpc_cli
|
|
|
|
.call(&rpc_host, &SystemRpc::PullClusterLayout, PRIO_NORMAL)
|
|
|
|
.await??
|
|
|
|
{
|
|
|
|
SystemRpc::AdvertiseClusterLayout(t) => Ok(t),
|
|
|
|
resp => Err(Error::Message(format!("Invalid RPC response: {:?}", resp))),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn send_layout(
|
|
|
|
rpc_cli: &Endpoint<SystemRpc, ()>,
|
|
|
|
rpc_host: NodeID,
|
|
|
|
layout: ClusterLayout,
|
|
|
|
) -> Result<(), Error> {
|
|
|
|
rpc_cli
|
|
|
|
.call(
|
|
|
|
&rpc_host,
|
|
|
|
&SystemRpc::AdvertiseClusterLayout(layout),
|
|
|
|
PRIO_NORMAL,
|
|
|
|
)
|
|
|
|
.await??;
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn print_cluster_layout(layout: &ClusterLayout) -> bool {
|
|
|
|
let mut table = vec!["ID\tTags\tZone\tCapacity".to_string()];
|
|
|
|
for (id, _, role) in layout.roles.items().iter() {
|
|
|
|
let role = match &role.0 {
|
|
|
|
Some(r) => r,
|
|
|
|
_ => continue,
|
|
|
|
};
|
|
|
|
let tags = role.tags.join(",");
|
|
|
|
table.push(format!(
|
|
|
|
"{:?}\t{}\t{}\t{}",
|
|
|
|
id,
|
|
|
|
tags,
|
|
|
|
role.zone,
|
|
|
|
role.capacity_string()
|
|
|
|
));
|
|
|
|
}
|
|
|
|
if table.len() == 1 {
|
|
|
|
false
|
|
|
|
} else {
|
|
|
|
format_table(table);
|
|
|
|
true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn print_staging_role_changes(layout: &ClusterLayout) -> bool {
|
2022-05-09 09:14:55 +00:00
|
|
|
let has_changes = layout
|
|
|
|
.staging
|
|
|
|
.items()
|
|
|
|
.iter()
|
|
|
|
.any(|(k, _, v)| layout.roles.get(k) != Some(v));
|
|
|
|
|
|
|
|
if has_changes {
|
2021-11-09 11:24:04 +00:00
|
|
|
println!();
|
|
|
|
println!("==== STAGED ROLE CHANGES ====");
|
|
|
|
let mut table = vec!["ID\tTags\tZone\tCapacity".to_string()];
|
|
|
|
for (id, _, role) in layout.staging.items().iter() {
|
2022-05-09 09:14:55 +00:00
|
|
|
if layout.roles.get(id) == Some(role) {
|
|
|
|
continue;
|
|
|
|
}
|
2021-11-09 11:24:04 +00:00
|
|
|
if let Some(role) = &role.0 {
|
|
|
|
let tags = role.tags.join(",");
|
|
|
|
table.push(format!(
|
|
|
|
"{:?}\t{}\t{}\t{}",
|
|
|
|
id,
|
|
|
|
tags,
|
|
|
|
role.zone,
|
|
|
|
role.capacity_string()
|
|
|
|
));
|
|
|
|
} else {
|
|
|
|
table.push(format!("{:?}\tREMOVED", id));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
format_table(table);
|
|
|
|
true
|
|
|
|
} else {
|
|
|
|
false
|
|
|
|
}
|
|
|
|
}
|