WIP: Automatically create node layout, keys and buckets #883
8 changed files with 214 additions and 175 deletions
|
@ -279,7 +279,8 @@ impl DataLayout {
|
|||
u16::from_be_bytes([
|
||||
hash.as_slice()[HASH_DRIVE_BYTES.0],
|
||||
hash.as_slice()[HASH_DRIVE_BYTES.1],
|
||||
]) as usize % DRIVE_NPART
|
||||
]) as usize
|
||||
% DRIVE_NPART
|
||||
}
|
||||
|
||||
fn block_dir_from(&self, hash: &Hash, dir: &PathBuf) -> PathBuf {
|
||||
|
|
|
@ -1,5 +1,8 @@
|
|||
use crate::admin::AdminRpc;
|
||||
use crate::cli::{cmd_apply_layout, cmd_assign_role, fetch_layout, fetch_status, ApplyLayoutOpt, AssignRoleOpt, BucketOperation, BucketOpt, KeyImportOpt, KeyInfoOpt, KeyOperation, PermBucketOpt};
|
||||
use crate::cli::{
|
||||
cmd_apply_layout, cmd_assign_role, fetch_layout, fetch_status, ApplyLayoutOpt, AssignRoleOpt,
|
||||
BucketOperation, BucketOpt, KeyImportOpt, KeyInfoOpt, KeyOperation, PermBucketOpt,
|
||||
};
|
||||
use bytesize::ByteSize;
|
||||
use garage_model::helper::error::Error as HelperError;
|
||||
use garage_net::endpoint::Endpoint;
|
||||
|
@ -12,167 +15,186 @@ use garage_util::data::Uuid;
|
|||
use garage_util::error::Error;
|
||||
|
||||
pub async fn key_exists(
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
key_pattern: String,
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
key_pattern: String,
|
||||
) -> Result<bool, Error> {
|
||||
match rpc_cli
|
||||
.call(&rpc_host, AdminRpc::KeyOperation(
|
||||
KeyOperation::Info(KeyInfoOpt{
|
||||
key_pattern,
|
||||
show_secret: false,
|
||||
})), PRIO_NORMAL)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(true),
|
||||
Err(HelperError::BadRequest(_)) => Ok(false),
|
||||
resp => Err(Error::unexpected_rpc_message(resp)),
|
||||
}
|
||||
match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
AdminRpc::KeyOperation(KeyOperation::Info(KeyInfoOpt {
|
||||
key_pattern,
|
||||
show_secret: false,
|
||||
})),
|
||||
PRIO_NORMAL,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(true),
|
||||
Err(HelperError::BadRequest(_)) => Ok(false),
|
||||
resp => Err(Error::unexpected_rpc_message(resp)),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn bucket_exists(
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
name: String,
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
name: String,
|
||||
) -> Result<bool, Error> {
|
||||
match rpc_cli
|
||||
.call(&rpc_host, AdminRpc::BucketOperation(
|
||||
BucketOperation::Info(BucketOpt{name})
|
||||
), PRIO_NORMAL)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(true),
|
||||
Err(HelperError::BadRequest(_)) => Ok(false),
|
||||
resp => Err(Error::unexpected_rpc_message(resp)),
|
||||
}
|
||||
match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
AdminRpc::BucketOperation(BucketOperation::Info(BucketOpt { name })),
|
||||
PRIO_NORMAL,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(true),
|
||||
Err(HelperError::BadRequest(_)) => Ok(false),
|
||||
resp => Err(Error::unexpected_rpc_message(resp)),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn key_create(
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
params: &AutoKey,
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
params: &AutoKey,
|
||||
) -> Result<(), Error> {
|
||||
match rpc_cli
|
||||
.call(&rpc_host, AdminRpc::KeyOperation(
|
||||
KeyOperation::Import(KeyImportOpt{
|
||||
name: params.name.clone(),
|
||||
secret_key: params.secret.clone(),
|
||||
key_id: params.id.clone(),
|
||||
yes: true,
|
||||
})
|
||||
), PRIO_NORMAL).await?
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(HelperError::BadRequest(msg)) => Err(Error::Message(msg)),
|
||||
resp => Err(Error::unexpected_rpc_message(resp))
|
||||
}
|
||||
match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
AdminRpc::KeyOperation(KeyOperation::Import(KeyImportOpt {
|
||||
name: params.name.clone(),
|
||||
secret_key: params.secret.clone(),
|
||||
key_id: params.id.clone(),
|
||||
yes: true,
|
||||
})),
|
||||
PRIO_NORMAL,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(HelperError::BadRequest(msg)) => Err(Error::Message(msg)),
|
||||
resp => Err(Error::unexpected_rpc_message(resp)),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn bucket_create(
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
params: &AutoBucket,
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
params: &AutoBucket,
|
||||
) -> Result<(), Error> {
|
||||
match rpc_cli
|
||||
.call(&rpc_host, AdminRpc::BucketOperation(
|
||||
BucketOperation::Create(BucketOpt{name: params.name.clone()})
|
||||
), PRIO_NORMAL)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(HelperError::BadRequest(msg)) => Err(Error::Message(msg)),
|
||||
resp => Err(Error::unexpected_rpc_message(resp))
|
||||
}
|
||||
match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
AdminRpc::BucketOperation(BucketOperation::Create(BucketOpt {
|
||||
name: params.name.clone(),
|
||||
})),
|
||||
PRIO_NORMAL,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(HelperError::BadRequest(msg)) => Err(Error::Message(msg)),
|
||||
resp => Err(Error::unexpected_rpc_message(resp)),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn grant_permission(
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
bucket_name: String,
|
||||
perm: &AutoPermission,
|
||||
rpc_cli: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
bucket_name: String,
|
||||
perm: &AutoPermission,
|
||||
) -> Result<(), Error> {
|
||||
match rpc_cli
|
||||
.call(&rpc_host, AdminRpc::BucketOperation(
|
||||
BucketOperation::Allow(PermBucketOpt{
|
||||
key_pattern: perm.key.clone(),
|
||||
read: perm.read,
|
||||
write: perm.write,
|
||||
owner: perm.owner,
|
||||
bucket: bucket_name,
|
||||
})
|
||||
), PRIO_NORMAL)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(HelperError::BadRequest(msg)) => Err(Error::Message(msg)),
|
||||
resp => Err(Error::unexpected_rpc_message(resp))
|
||||
}
|
||||
match rpc_cli
|
||||
.call(
|
||||
&rpc_host,
|
||||
AdminRpc::BucketOperation(BucketOperation::Allow(PermBucketOpt {
|
||||
key_pattern: perm.key.clone(),
|
||||
read: perm.read,
|
||||
write: perm.write,
|
||||
owner: perm.owner,
|
||||
bucket: bucket_name,
|
||||
})),
|
||||
PRIO_NORMAL,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(HelperError::BadRequest(msg)) => Err(Error::Message(msg)),
|
||||
resp => Err(Error::unexpected_rpc_message(resp)),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get_unassigned_nodes(
|
||||
rpc_cli: &Endpoint<SystemRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
rpc_cli: &Endpoint<SystemRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
) -> Result<Option<Vec<Uuid>>, Error> {
|
||||
let status = fetch_status(rpc_cli, rpc_host).await?;
|
||||
let layout = fetch_layout(rpc_cli, rpc_host).await?;
|
||||
let mut nodes: Vec<Uuid> = Vec::new();
|
||||
let status = fetch_status(rpc_cli, rpc_host).await?;
|
||||
let layout = fetch_layout(rpc_cli, rpc_host).await?;
|
||||
let mut nodes: Vec<Uuid> = Vec::new();
|
||||
|
||||
for adv in status.iter().filter(|adv| adv.is_up) {
|
||||
if layout.current().roles.get(&adv.id).is_none() {
|
||||
let prev_role = layout
|
||||
.versions
|
||||
.iter()
|
||||
.rev()
|
||||
.find_map(|x| match x.roles.get(&adv.id) {
|
||||
Some(NodeRoleV(Some(cfg))) => Some(cfg),
|
||||
_ => None,
|
||||
});
|
||||
if prev_role.is_none() {
|
||||
if let Some(NodeRoleV(Some(_))) = layout.staging.get().roles.get(&adv.id) {
|
||||
// Node role assignment is pending, can return immediately.
|
||||
return Ok(None);
|
||||
} else {
|
||||
nodes.push(adv.id.clone());
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Node role is assigned, can return immediately.
|
||||
return Ok(None);
|
||||
}
|
||||
}
|
||||
for adv in status.iter().filter(|adv| adv.is_up) {
|
||||
if layout.current().roles.get(&adv.id).is_none() {
|
||||
let prev_role = layout
|
||||
.versions
|
||||
.iter()
|
||||
.rev()
|
||||
.find_map(|x| match x.roles.get(&adv.id) {
|
||||
Some(NodeRoleV(Some(cfg))) => Some(cfg),
|
||||
_ => None,
|
||||
});
|
||||
if prev_role.is_none() {
|
||||
if let Some(NodeRoleV(Some(_))) = layout.staging.get().roles.get(&adv.id) {
|
||||
// Node role assignment is pending, can return immediately.
|
||||
return Ok(None);
|
||||
} else {
|
||||
nodes.push(adv.id.clone());
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Node role is assigned, can return immediately.
|
||||
return Ok(None);
|
||||
}
|
||||
}
|
||||
|
||||
// Encountered no node with an assignment (pending or applied).
|
||||
// Therefore, all nodes are unassigned.
|
||||
Ok(Some(nodes))
|
||||
// Encountered no node with an assignment (pending or applied).
|
||||
// Therefore, all nodes are unassigned.
|
||||
Ok(Some(nodes))
|
||||
}
|
||||
|
||||
pub async fn assign_node_layout(
|
||||
rpc_cli: &Endpoint<SystemRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
unassigned_nodes: &Vec<Uuid>,
|
||||
auto_nodes: &Vec<AutoNode>,
|
||||
rpc_cli: &Endpoint<SystemRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
unassigned_nodes: &Vec<Uuid>,
|
||||
auto_nodes: &Vec<AutoNode>,
|
||||
) -> Result<(), Error> {
|
||||
if unassigned_nodes.len() != auto_nodes.len() {
|
||||
return Err(Error::Message("Cannot apply auto layout: configured nodes do not match actual nodes".to_string()));
|
||||
}
|
||||
if unassigned_nodes.len() != auto_nodes.len() {
|
||||
return Err(Error::Message(
|
||||
"Cannot apply auto layout: configured nodes do not match actual nodes".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
for (i, node_id) in unassigned_nodes.iter().enumerate() {
|
||||
if let Some(auto) = auto_nodes.get(i) {
|
||||
let capacity = auto.capacity.parse::<ByteSize>()?;
|
||||
cmd_assign_role(rpc_cli, rpc_host, AssignRoleOpt{
|
||||
node_ids: vec![format!("{id:?}", id=node_id)],
|
||||
zone: Some(auto.zone.clone()),
|
||||
capacity: Some(capacity),
|
||||
gateway: false,
|
||||
tags: vec![],
|
||||
replace: vec![],
|
||||
}).await?;
|
||||
}
|
||||
}
|
||||
for (i, node_id) in unassigned_nodes.iter().enumerate() {
|
||||
if let Some(auto) = auto_nodes.get(i) {
|
||||
let capacity = auto.capacity.parse::<ByteSize>()?;
|
||||
cmd_assign_role(
|
||||
rpc_cli,
|
||||
rpc_host,
|
||||
AssignRoleOpt {
|
||||
node_ids: vec![format!("{id:?}", id = node_id)],
|
||||
zone: Some(auto.zone.clone()),
|
||||
capacity: Some(capacity),
|
||||
gateway: false,
|
||||
tags: vec![],
|
||||
replace: vec![],
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
cmd_apply_layout(rpc_cli, rpc_host, ApplyLayoutOpt{
|
||||
version: Some(1),
|
||||
}).await?;
|
||||
cmd_apply_layout(rpc_cli, rpc_host, ApplyLayoutOpt { version: Some(1) }).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
@ -19,7 +19,7 @@ pub async fn cli_command_dispatch(
|
|||
system_rpc_endpoint: &Endpoint<SystemRpc, ()>,
|
||||
admin_rpc_endpoint: &Endpoint<AdminRpc, ()>,
|
||||
rpc_host: NodeID,
|
||||
config: &Config,
|
||||
config: &Config,
|
||||
) -> Result<(), HelperError> {
|
||||
match cmd {
|
||||
Command::Status => Ok(cmd_status(system_rpc_endpoint, rpc_host).await?),
|
||||
|
@ -46,9 +46,15 @@ pub async fn cli_command_dispatch(
|
|||
Command::Meta(mo) => {
|
||||
cmd_admin(admin_rpc_endpoint, rpc_host, AdminRpc::MetaOperation(mo)).await
|
||||
}
|
||||
Command::Auto => {
|
||||
cmd_auto(admin_rpc_endpoint, system_rpc_endpoint, rpc_host, config.auto.as_ref()).await
|
||||
}
|
||||
Command::Auto => {
|
||||
cmd_auto(
|
||||
admin_rpc_endpoint,
|
||||
system_rpc_endpoint,
|
||||
rpc_host,
|
||||
config.auto.as_ref(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
}
|
||||
|
@ -275,9 +281,8 @@ pub async fn cmd_auto(
|
|||
rpc_host: NodeID,
|
||||
config: Option<&AutoConfig>,
|
||||
) -> Result<(), HelperError> {
|
||||
match config {
|
||||
Some(auto) => {
|
||||
|
||||
match config {
|
||||
Some(auto) => {
|
||||
// Assign cluster layout if all nodes are unassigned.
|
||||
// This is to ensure a newly created cluster is readily available.
|
||||
// Further changes to the cluster layout must be done manually.
|
||||
|
@ -286,12 +291,12 @@ pub async fn cmd_auto(
|
|||
}
|
||||
|
||||
// Import keys
|
||||
for key in auto.keys.iter() {
|
||||
let exists = key_exists(rpc_admin, rpc_host, key.id.clone()).await?;
|
||||
if !exists {
|
||||
key_create(rpc_admin, rpc_host, key).await?;
|
||||
}
|
||||
}
|
||||
for key in auto.keys.iter() {
|
||||
let exists = key_exists(rpc_admin, rpc_host, key.id.clone()).await?;
|
||||
if !exists {
|
||||
key_create(rpc_admin, rpc_host, key).await?;
|
||||
}
|
||||
}
|
||||
|
||||
// Import buckets
|
||||
for bucket in auto.buckets.iter() {
|
||||
|
@ -305,12 +310,14 @@ pub async fn cmd_auto(
|
|||
grant_permission(rpc_admin, rpc_host, bucket.name.clone(), perm).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
return Err(HelperError::BadRequest("Auto configuration is missing".to_string()))
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
_ => {
|
||||
return Err(HelperError::BadRequest(
|
||||
"Auto configuration is missing".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---- utility ----
|
||||
|
@ -326,4 +333,4 @@ pub async fn fetch_status(
|
|||
SystemRpc::ReturnKnownNodes(nodes) => Ok(nodes),
|
||||
resp => Err(Error::unexpected_rpc_message(resp)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -1,15 +1,15 @@
|
|||
pub(crate) mod auto;
|
||||
pub(crate) mod cmd;
|
||||
pub(crate) mod init;
|
||||
pub(crate) mod layout;
|
||||
pub(crate) mod structs;
|
||||
pub(crate) mod util;
|
||||
pub(crate) mod auto;
|
||||
|
||||
pub(crate) mod convert_db;
|
||||
|
||||
pub(crate) use auto::*;
|
||||
pub(crate) use cmd::*;
|
||||
pub(crate) use init::*;
|
||||
pub(crate) use layout::*;
|
||||
pub(crate) use structs::*;
|
||||
pub(crate) use util::*;
|
||||
pub(crate) use auto::*;
|
|
@ -60,8 +60,8 @@ pub enum Command {
|
|||
#[structopt(name = "convert-db", version = garage_version())]
|
||||
ConvertDb(convert_db::ConvertDbOpt),
|
||||
|
||||
/// Create preconfigured keys, buckets and node layout.
|
||||
Auto,
|
||||
/// Create preconfigured keys, buckets and node layout.
|
||||
Auto,
|
||||
}
|
||||
|
||||
#[derive(StructOpt, Debug)]
|
||||
|
|
|
@ -284,7 +284,15 @@ async fn cli_command(opt: Opt) -> Result<(), Error> {
|
|||
let system_rpc_endpoint = netapp.endpoint::<SystemRpc, ()>(SYSTEM_RPC_PATH.into());
|
||||
let admin_rpc_endpoint = netapp.endpoint::<AdminRpc, ()>(ADMIN_RPC_PATH.into());
|
||||
|
||||
match cli_command_dispatch(opt.cmd, &system_rpc_endpoint, &admin_rpc_endpoint, id, config.as_ref().unwrap()).await {
|
||||
match cli_command_dispatch(
|
||||
opt.cmd,
|
||||
&system_rpc_endpoint,
|
||||
&admin_rpc_endpoint,
|
||||
id,
|
||||
config.as_ref().unwrap(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Err(HelperError::Internal(i)) => Err(Error::Message(format!("Internal error: {}", i))),
|
||||
Err(HelperError::BadRequest(b)) => Err(Error::Message(b)),
|
||||
Err(e) => Err(Error::Message(format!("{}", e))),
|
||||
|
|
|
@ -279,7 +279,8 @@ impl<'a> LockedHelper<'a> {
|
|||
.local_aliases
|
||||
.get(alias_name)
|
||||
.cloned()
|
||||
.flatten() != Some(bucket_id)
|
||||
.flatten()
|
||||
!= Some(bucket_id)
|
||||
{
|
||||
return Err(GarageError::Message(format!(
|
||||
"Bucket {:?} does not have alias {} in namespace of key {}",
|
||||
|
|
|
@ -129,8 +129,8 @@ pub struct Config {
|
|||
#[serde(default = "Default::default")]
|
||||
pub admin: AdminConfig,
|
||||
|
||||
/// Configuration to apply automatically
|
||||
pub auto: Option<AutoConfig>,
|
||||
/// Configuration to apply automatically
|
||||
pub auto: Option<AutoConfig>,
|
||||
}
|
||||
|
||||
/// Value for data_dir: either a single directory or a list of dirs with attributes
|
||||
|
@ -205,10 +205,10 @@ pub struct AdminConfig {
|
|||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
pub struct AutoConfig {
|
||||
/// Buckets to create automatically
|
||||
pub buckets: Vec<AutoBucket>,
|
||||
|
||||
/// Keys to create automatically
|
||||
pub keys: Vec<AutoKey>,
|
||||
pub buckets: Vec<AutoBucket>,
|
||||
|
||||
/// Keys to create automatically
|
||||
pub keys: Vec<AutoKey>,
|
||||
|
||||
/// Node layout to create automatically
|
||||
pub nodes: Vec<AutoNode>,
|
||||
|
@ -218,27 +218,27 @@ pub struct AutoConfig {
|
|||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
pub struct AutoKey {
|
||||
/// Key name
|
||||
pub name: String,
|
||||
pub name: String,
|
||||
/// Key ID starting with GK
|
||||
pub id: String,
|
||||
pub id: String,
|
||||
/// Secret key
|
||||
pub secret: String,
|
||||
pub secret: String,
|
||||
}
|
||||
|
||||
/// Bucket to create automatically
|
||||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
pub struct AutoBucket {
|
||||
/// Bucket name
|
||||
pub name: String,
|
||||
pub name: String,
|
||||
/// Permissions to grant on bucket to given keys
|
||||
pub allow: Vec<AutoPermission>,
|
||||
pub allow: Vec<AutoPermission>,
|
||||
}
|
||||
|
||||
/// Permission to create automatically
|
||||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
pub struct AutoPermission {
|
||||
/// Key ID or name
|
||||
pub key: String,
|
||||
/// Key ID or name
|
||||
pub key: String,
|
||||
|
||||
/// Grant read permission
|
||||
pub read: bool,
|
||||
|
|
Loading…
Reference in a new issue