New model for buckets #172

Merged
lx merged 19 commits from new-buckets into main 2022-01-10 11:32:42 +00:00
12 changed files with 86 additions and 62 deletions
Showing only changes of commit 1bcd6fabbd - Show all commits

View file

@ -7,7 +7,6 @@ use hyper::server::conn::AddrStream;
use hyper::service::{make_service_fn, service_fn};
use hyper::{Body, Request, Response, Server};
use garage_util::crdt;
use garage_util::data::*;
use garage_util::error::Error as GarageError;
@ -306,8 +305,7 @@ async fn resolve_bucket(
.as_option()
.ok_or_else(|| Error::Forbidden("Operation is not allowed for this key.".to_string()))?;
if let Some(crdt::Deletable::Present(bucket_id)) = api_key_params.local_aliases.get(bucket_name)
{
if let Some(Some(bucket_id)) = api_key_params.local_aliases.get(bucket_name) {
Ok(*bucket_id)
} else {
Ok(garage

View file

@ -65,8 +65,8 @@ pub async fn handle_list_buckets(garage: &Garage, api_key: &Key) -> Result<Respo
if *active {
let alias_ent = garage.bucket_alias_table.get(&EmptyKey, alias).await?;
if let Some(alias_ent) = alias_ent {
if let Some(alias_p) = alias_ent.state.get().as_option() {
if alias_p.bucket_id == *bucket_id {
if let Some(alias_bucket) = alias_ent.state.get() {
if alias_bucket == bucket_id {
aliases.insert(alias_ent.name().to_string(), *bucket_id);
lx marked this conversation as resolved Outdated

this is a lot of indentation. It may be possible to remove a few layers by using Option::and_then, and doing param.aliases.items().filter(|(_,_,active)| active)

this is a lot of indentation. It may be possible to remove a few layers by using Option::and_then, and doing `param.aliases.items().filter(|(_,_,active)| active)`
}
}
@ -78,8 +78,8 @@ pub async fn handle_list_buckets(garage: &Garage, api_key: &Key) -> Result<Respo
}
}
for (alias, _, id) in key_state.local_aliases.items() {
if let Some(id) = id.as_option() {
for (alias, _, id_opt) in key_state.local_aliases.items() {
if let Some(id) = id_opt {
aliases.insert(alias.clone(), *id);
}
}

View file

@ -140,7 +140,7 @@ impl AdminRpcHandler {
}
if let Some(alias) = self.garage.bucket_alias_table.get(&EmptyKey, name).await? {
if !alias.state.get().is_deleted() {
if alias.state.get().is_some() {
return Err(Error::BadRequest(format!("Bucket {} already exists", name)));
}
}
@ -229,7 +229,7 @@ impl AdminRpcHandler {
// 2. delete bucket alias
if bucket_alias.is_some() {
helper
.unset_global_bucket_alias(bucket_id, &query.name)
.purge_global_bucket_alias(bucket_id, &query.name)
.await?;
}
@ -281,7 +281,7 @@ impl AdminRpcHandler {
.unwrap()
.local_aliases
.get(&query.name)
.map(|a| a.into_option())
.cloned()
.flatten()
.ok_or_bad_request("Bucket not found")?;
@ -484,20 +484,26 @@ impl AdminRpcHandler {
let state = key.state.as_option_mut().unwrap();
// --- done checking, now commit ---
// (the step at unset_local_bucket_alias will fail if a bucket
// does not have another alias, the deletion will be
// interrupted in the middle if that happens)
// 1. Delete local aliases
for (alias, _, to) in state.local_aliases.items().iter() {
if let Deletable::Present(bucket_id) = to {
if let Some(bucket_id) = to {
helper
.unset_local_bucket_alias(*bucket_id, &key.key_id, alias)
.await?;
}
}
// 2. Delete authorized buckets
// 2. Remove permissions on all authorized buckets
for (ab_id, _auth) in state.authorized_buckets.items().iter() {
helper
.set_bucket_key_permissions(*ab_id, &key.key_id, BucketKeyPerm::no_permissions())
.await?;
}
// 3. Actually delete key
key.state = Deletable::delete();
self.garage.key_table.insert(&key).await?;

View file

@ -168,8 +168,8 @@ pub async fn cmd_admin(
println!("List of buckets:");
let mut table = vec![];
for alias in bl {
if let Some(p) = alias.state.get().as_option() {
table.push(format!("\t{}\t{:?}", alias.name(), p.bucket_id));
if let Some(alias_bucket) = alias.state.get() {
table.push(format!("\t{}\t{:?}", alias.name(), alias_bucket));
}
}
format_table(table);

View file

@ -34,7 +34,7 @@ pub fn print_key_info(key: &Key, relevant_buckets: &HashMap<Uuid, Bucket>) {
println!("\nKey-specific bucket aliases:");
let mut table = vec![];
for (alias_name, _, alias) in p.local_aliases.items().iter() {
if let Some(bucket_id) = alias.as_option() {
if let Some(bucket_id) = alias {
table.push(format!(
"\t{}\t{}\t{}",
alias_name,
@ -55,7 +55,7 @@ pub fn print_key_info(key: &Key, relevant_buckets: &HashMap<Uuid, Bucket>) {
.local_aliases
.items()
.iter()
.filter(|(_, _, a)| a.as_option() == Some(bucket_id))
.filter(|(_, _, a)| *a == Some(*bucket_id))
.map(|(a, _, _)| a.clone())
.collect::<Vec<_>>()
.join(", ");

View file

@ -10,32 +10,23 @@ use garage_table::*;
#[derive(PartialEq, Clone, Debug, Serialize, Deserialize)]
pub struct BucketAlias {
name: String,
pub state: crdt::Lww<crdt::Deletable<AliasParams>>,
}
#[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)]
pub struct AliasParams {
pub bucket_id: Uuid,
}
impl AutoCrdt for AliasParams {
const WARN_IF_DIFFERENT: bool = true;
pub state: crdt::Lww<Option<Uuid>>,
}
impl BucketAlias {
pub fn new(name: String, ts: u64, bucket_id: Uuid) -> Option<Self> {
pub fn new(name: String, ts: u64, bucket_id: Option<Uuid>) -> Option<Self> {
if !is_valid_bucket_name(&name) {
None
} else {
Some(BucketAlias {
name,
state: crdt::Lww::raw(ts, crdt::Deletable::present(AliasParams { bucket_id })),
state: crdt::Lww::raw(ts, bucket_id),
})
}
}
pub fn is_deleted(&self) -> bool {
self.state.get().is_deleted()
self.state.get().is_none()
}
pub fn name(&self) -> &str {
&self.name

View file

@ -63,6 +63,7 @@ impl BucketParams {
impl Crdt for BucketParams {
fn merge(&mut self, o: &Self) {
self.creation_date = std::cmp::min(self.creation_date, o.creation_date);
self.authorized_keys.merge(&o.authorized_keys);
self.website_config.merge(&o.website_config);
self.aliases.merge(&o.aliases);

View file

@ -46,7 +46,7 @@ impl<'a> BucketHelper<'a> {
.bucket_alias_table
.get(&EmptyKey, bucket_name)
.await?
.map(|x| x.state.get().as_option().map(|x| x.bucket_id))
.map(|x| *x.state.get())
.flatten())
}
}
@ -154,11 +154,11 @@ impl<'a> BucketHelper<'a> {
let alias = self.0.bucket_alias_table.get(&EmptyKey, alias_name).await?;
if let Some(existing_alias) = alias.as_ref() {
if let Some(p) = existing_alias.state.get().as_option() {
if p.bucket_id != bucket_id {
if let Some(p_bucket) = existing_alias.state.get() {
if *p_bucket != bucket_id {
return Err(Error::BadRequest(format!(
"Alias {} already exists and points to different bucket: {:?}",
alias_name, p.bucket_id
alias_name, p_bucket
)));
}
}
@ -176,10 +176,10 @@ impl<'a> BucketHelper<'a> {
// writes are now done and all writes use timestamp alias_ts
let alias = match alias {
None => BucketAlias::new(alias_name.clone(), alias_ts, bucket_id)
None => BucketAlias::new(alias_name.clone(), alias_ts, Some(bucket_id))
.ok_or_bad_request(format!("{}: {}", alias_name, INVALID_BUCKET_NAME_MESSAGE))?,
Some(mut a) => {
a.state = Lww::raw(alias_ts, Deletable::present(AliasParams { bucket_id }));
a.state = Lww::raw(alias_ts, Some(bucket_id));
a
}
};
@ -209,13 +209,7 @@ impl<'a> BucketHelper<'a> {
.bucket_alias_table
.get(&EmptyKey, alias_name)
.await?
.filter(|a| {
a.state
.get()
.as_option()
.map(|x| x.bucket_id == bucket_id)
.unwrap_or(false)
})
.filter(|a| a.state.get().map(|x| x == bucket_id).unwrap_or(false))
.ok_or_message(format!(
"Internal error: alias not found or does not point to bucket {:?}",
bucket_id
@ -244,7 +238,7 @@ impl<'a> BucketHelper<'a> {
// ---- timestamp-ensured causality barrier ----
// writes are now done and all writes use timestamp alias_ts
alias.state = Lww::raw(alias_ts, Deletable::delete());
alias.state = Lww::raw(alias_ts, None);
self.0.bucket_alias_table.insert(&alias).await?;
bucket_state.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, false);
@ -253,6 +247,51 @@ impl<'a> BucketHelper<'a> {
Ok(())
}
/// Ensures a bucket does not have a certain global alias.
/// Contrarily to unset_global_bucket_alias, this does not
/// fail on any condition other than:
/// - bucket cannot be found (its fine if it is in deleted state)
/// - alias cannot be found (its fine if it points to nothing or
/// to another bucket)
pub async fn purge_global_bucket_alias(
&self,
bucket_id: Uuid,
alias_name: &String,
) -> Result<(), Error> {
let mut bucket = self.get_internal_bucket(bucket_id).await?;
let mut alias = self
.0
.bucket_alias_table
.get(&EmptyKey, alias_name)
.await?
.ok_or_message(format!("Alias {} not found", alias_name))?;
// Checks ok, remove alias
let alias_ts = match bucket.state.as_option() {
Some(bucket_state) => increment_logical_clock_2(
alias.state.timestamp(),
bucket_state.aliases.get_timestamp(alias_name),
),
None => increment_logical_clock(alias.state.timestamp()),
};
// ---- timestamp-ensured causality barrier ----
// writes are now done and all writes use timestamp alias_ts
if alias.state.get() == &Some(bucket_id) {
alias.state = Lww::raw(alias_ts, None);
self.0.bucket_alias_table.insert(&alias).await?;
}
if let Some(mut bucket_state) = bucket.state.as_option_mut() {
bucket_state.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, false);
self.0.bucket_table.insert(&bucket).await?;
}
Ok(())
}
/// Sets a new alias for a bucket in the local namespace of a key.
/// This function fails if:
/// - alias name is not valid according to S3 spec
@ -277,7 +316,7 @@ impl<'a> BucketHelper<'a> {
let mut key_param = key.state.as_option_mut().unwrap();
if let Some(Deletable::Present(existing_alias)) = key_param.local_aliases.get(alias_name) {
if let Some(Some(existing_alias)) = key_param.local_aliases.get(alias_name) {
if *existing_alias != bucket_id {
return Err(Error::BadRequest(format!("Alias {} already exists in namespace of key {} and points to different bucket: {:?}", alias_name, key.key_id, existing_alias)));
}
@ -301,8 +340,7 @@ impl<'a> BucketHelper<'a> {
// ---- timestamp-ensured causality barrier ----
// writes are now done and all writes use timestamp alias_ts
key_param.local_aliases =
LwwMap::raw_item(alias_name.clone(), alias_ts, Deletable::present(bucket_id));
key_param.local_aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, Some(bucket_id));
self.0.key_table.insert(&key).await?;
bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, true);
@ -334,8 +372,8 @@ impl<'a> BucketHelper<'a> {
.unwrap()
.local_aliases
.get(alias_name)
.map(|x| x.as_option())
.flatten() != Some(&bucket_id)
.cloned()
.flatten() != Some(bucket_id)
{
return Err(GarageError::Message(format!(
"Bucket {:?} does not have alias {} in namespace of key {}",
@ -372,8 +410,7 @@ impl<'a> BucketHelper<'a> {
// ---- timestamp-ensured causality barrier ----
// writes are now done and all writes use timestamp alias_ts
key_param.local_aliases =
LwwMap::raw_item(alias_name.clone(), alias_ts, Deletable::delete());
key_param.local_aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, None);
self.0.key_table.insert(&key).await?;
bucket_p.local_aliases = LwwMap::raw_item(bucket_p_local_alias_key, alias_ts, false);

View file

@ -31,7 +31,7 @@ pub struct Key {
pub struct KeyParams {
pub allow_create_bucket: crdt::Lww<bool>,
pub authorized_buckets: crdt::Map<Uuid, BucketKeyPerm>,
pub local_aliases: crdt::LwwMap<String, crdt::Deletable<Uuid>>,
pub local_aliases: crdt::LwwMap<String, Option<Uuid>>,
}
impl KeyParams {

View file

@ -1,5 +1,3 @@
use crate::data::*;
/// Definition of a CRDT - all CRDT Rust types implement this.
///
/// A CRDT is defined as a merge operator that respects a certain set of axioms.
@ -87,7 +85,3 @@ impl AutoCrdt for String {
impl AutoCrdt for bool {
const WARN_IF_DIFFERENT: bool = true;
}
impl AutoCrdt for FixedBytes32 {
const WARN_IF_DIFFERENT: bool = true;
}

View file

@ -5,7 +5,7 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::fmt;
/// An array of 32 bytes
#[derive(Default, PartialOrd, Ord, Clone, Hash, PartialEq, Copy)]
#[derive(Default, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Copy)]
pub struct FixedBytes32([u8; 32]);
impl From<[u8; 32]> for FixedBytes32 {
@ -20,8 +20,6 @@ impl std::convert::AsRef<[u8]> for FixedBytes32 {
}
}
impl Eq for FixedBytes32 {}
impl fmt::Debug for FixedBytes32 {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}…", hex::encode(&self.0[..8]))

View file

@ -86,9 +86,8 @@ async fn serve_file(garage: Arc<Garage>, req: Request<Body>) -> Result<Response<
.bucket_alias_table
.get(&EmptyKey, &bucket_name.to_string())
.await?
.map(|x| x.state.take().into_option())
.map(|x| x.state.take())
.flatten()
.map(|param| param.bucket_id)
.ok_or(Error::NotFound)?;
// Check bucket isn't deleted and has website access enabled