use std::time::Duration;
use garage_util::crdt::*;
use garage_util::data::*;
use garage_util::error::{Error as GarageError, OkOrMessage};
use garage_util::time::*;
use garage_table::util::*;
use crate::bucket_alias_table::*;
use crate::bucket_table::*;
use crate::garage::Garage;
use crate::helper::error::*;
use crate::helper::key::KeyHelper;
use crate::key_table::*;
use crate::permission::BucketKeyPerm;
use crate::s3::object_table::*;
pub struct BucketHelper<'a>(pub(crate) &'a Garage);
#[allow(clippy::ptr_arg)]
impl<'a> BucketHelper<'a> {
pub async fn resolve_global_bucket_name(
&self,
bucket_name: &String,
) -> Result<Option<Uuid>, Error> {
let hexbucket = hex::decode(bucket_name.as_str())
.ok()
.and_then(|by| Uuid::try_from(&by));
if let Some(bucket_id) = hexbucket {
Ok(self
.0
.bucket_table
.get(&EmptyKey, &bucket_id)
.await?
.filter(|x| !x.state.is_deleted())
.map(|_| bucket_id))
} else {
Ok(self
.0
.bucket_alias_table
.get(&EmptyKey, bucket_name)
.await?
.and_then(|x| *x.state.get()))
}
}
#[allow(clippy::ptr_arg)]
pub async fn resolve_bucket(&self, bucket_name: &String, api_key: &Key) -> Result<Uuid, Error> {
let api_key_params = api_key
.state
.as_option()
.ok_or_message("Key should not be deleted at this point")?;
if let Some(Some(bucket_id)) = api_key_params.local_aliases.get(bucket_name) {
Ok(*bucket_id)
} else {
Ok(self
.resolve_global_bucket_name(bucket_name)
.await?
.ok_or_else(|| Error::NoSuchBucket(bucket_name.to_string()))?)
}
}
pub async fn get_internal_bucket(&self, bucket_id: Uuid) -> Result<Bucket, Error> {
Ok(self
.0
.bucket_table
.get(&EmptyKey, &bucket_id)
.await?
.ok_or_message(format!("Bucket {:?} does not exist", bucket_id))?)
}
pub async fn get_existing_bucket(&self, bucket_id: Uuid) -> Result<Bucket, Error> {
self.0
.bucket_table
.get(&EmptyKey, &bucket_id)
.await?
.filter(|b| !b.is_deleted())
.ok_or_else(|| Error::NoSuchBucket(hex::encode(bucket_id)))
}
pub async fn set_global_bucket_alias(
&self,
bucket_id: Uuid,
alias_name: &String,
) -> Result<(), Error> {
if !is_valid_bucket_name(alias_name) {
return Err(Error::InvalidBucketName(alias_name.to_string()));
}
let mut bucket = self.get_existing_bucket(bucket_id).await?;
let alias = self.0.bucket_alias_table.get(&EmptyKey, alias_name).await?;
if let Some(existing_alias) = alias.as_ref() {
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
)));
}
}
}
let mut bucket_p = bucket.state.as_option_mut().unwrap();
let alias_ts = increment_logical_clock_2(
bucket_p.aliases.get_timestamp(alias_name),
alias.as_ref().map(|a| a.state.timestamp()).unwrap_or(0),
);
let alias = match alias {
None => BucketAlias::new(alias_name.clone(), alias_ts, Some(bucket_id))
.ok_or_else(|| Error::InvalidBucketName(alias_name.clone()))?,
Some(mut a) => {
a.state = Lww::raw(alias_ts, Some(bucket_id));
a
}
};
self.0.bucket_alias_table.insert(&alias).await?;
bucket_p.aliases = LwwMap::raw_item(alias_name.clone(), alias_ts, true);
self.0.bucket_table.insert(&bucket).await?;
Ok(())
}
pub async fn unset_global_bucket_alias(
&self,
bucket_id: Uuid,
alias_name: &String,
) -> Result<(), Error> {
let mut bucket = self.get_existing_bucket(bucket_id).await?;
let mut bucket_state = bucket.state.as_option_mut().unwrap();
let mut alias = self
.0
.bucket_alias_table
.get(&EmptyKey, alias_name)
.await?
.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
))?;
let has_other_global_aliases = bucket_state
.aliases
.items()
.iter()
.any(|(name, _, active)| name != alias_name && *active);
let has_other_local_aliases = bucket_state
.local_aliases
.items()
.iter()
.any(|(_, _, active)| *active);
if !has_other_global_aliases && !has_other_local_aliases {
return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", alias_name)));
}
let alias_ts = increment_logical_clock_2(
alias.state.timestamp(),
bucket_state.aliases.get_timestamp(alias_name),
);
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);
self.0.bucket_table.insert(&bucket).await?;
Ok(())
}
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_else(|| Error::NoSuchBucket(alias_name.to_string()))?;
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()),
};
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(())
}
pub async fn set_local_bucket_alias(
&self,
bucket_id: Uuid,
key_id: &String,
alias_name: &String,
) -> Result<(), Error> {
let key_helper = KeyHelper(self.0);
if !is_valid_bucket_name(alias_name) {
return Err(Error::InvalidBucketName(alias_name.to_string()));
}
let mut bucket = self.get_existing_bucket(bucket_id).await?;
let mut key = key_helper.get_existing_key(key_id).await?;
let mut key_param = key.state.as_option_mut().unwrap();
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)));
}
}
let mut bucket_p = bucket.state.as_option_mut().unwrap();
let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone());
let alias_ts = increment_logical_clock_2(
key_param.local_aliases.get_timestamp(alias_name),
bucket_p
.local_aliases
.get_timestamp(&bucket_p_local_alias_key),
);
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);
self.0.bucket_table.insert(&bucket).await?;
Ok(())
}
pub async fn unset_local_bucket_alias(
&self,
bucket_id: Uuid,
key_id: &String,
alias_name: &String,
) -> Result<(), Error> {
let key_helper = KeyHelper(self.0);
let mut bucket = self.get_existing_bucket(bucket_id).await?;
let mut key = key_helper.get_existing_key(key_id).await?;
let mut bucket_p = bucket.state.as_option_mut().unwrap();
if key
.state
.as_option()
.unwrap()
.local_aliases
.get(alias_name)
.cloned()
.flatten() != Some(bucket_id)
{
return Err(GarageError::Message(format!(
"Bucket {:?} does not have alias {} in namespace of key {}",
bucket_id, alias_name, key_id
))
.into());
}
let has_other_global_aliases = bucket_p
.aliases
.items()
.iter()
.any(|(_, _, active)| *active);
let has_other_local_aliases = bucket_p
.local_aliases
.items()
.iter()
.any(|((k, n), _, active)| *k == key.key_id && n == alias_name && *active);
if !has_other_global_aliases && !has_other_local_aliases {
return Err(Error::BadRequest(format!("Bucket {} doesn't have other aliases, please delete it instead of just unaliasing.", alias_name)));
}
let mut key_param = key.state.as_option_mut().unwrap();
let bucket_p_local_alias_key = (key.key_id.clone(), alias_name.clone());
let alias_ts = increment_logical_clock_2(
key_param.local_aliases.get_timestamp(alias_name),
bucket_p
.local_aliases
.get_timestamp(&bucket_p_local_alias_key),
);
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);
self.0.bucket_table.insert(&bucket).await?;
Ok(())
}
pub async fn set_bucket_key_permissions(
&self,
bucket_id: Uuid,
key_id: &String,
mut perm: BucketKeyPerm,
) -> Result<(), Error> {
let key_helper = KeyHelper(self.0);
let mut bucket = self.get_internal_bucket(bucket_id).await?;
let mut key = key_helper.get_internal_key(key_id).await?;
if let Some(bstate) = bucket.state.as_option() {
if let Some(kp) = bstate.authorized_keys.get(key_id) {
perm.timestamp = increment_logical_clock_2(perm.timestamp, kp.timestamp);
}
} else if perm.is_any() {
return Err(Error::BadRequest(
"Trying to give permissions on a deleted bucket".into(),
));
}
if let Some(kstate) = key.state.as_option() {
if let Some(bp) = kstate.authorized_buckets.get(&bucket_id) {
perm.timestamp = increment_logical_clock_2(perm.timestamp, bp.timestamp);
}
} else if perm.is_any() {
return Err(Error::BadRequest(
"Trying to give permissions to a deleted key".into(),
));
}
if let Some(bstate) = bucket.state.as_option_mut() {
bstate.authorized_keys = Map::put_mutator(key_id.clone(), perm);
self.0.bucket_table.insert(&bucket).await?;
}
if let Some(kstate) = key.state.as_option_mut() {
kstate.authorized_buckets = Map::put_mutator(bucket_id, perm);
self.0.key_table.insert(&key).await?;
}
Ok(())
}
pub async fn is_bucket_empty(&self, bucket_id: Uuid) -> Result<bool, Error> {
let objects = self
.0
.object_table
.get_range(
&bucket_id,
None,
Some(ObjectFilter::IsData),
10,
EnumerationOrder::Forward,
)
.await?;
if !objects.is_empty() {
return Ok(false);
}
#[cfg(feature = "k2v")]
{
use garage_rpc::ring::Ring;
use std::sync::Arc;
let ring: Arc<Ring> = self.0.system.ring.borrow().clone();
let k2vindexes = self
.0
.k2v
.counter_table
.table
.get_range(
&bucket_id,
None,
Some((DeletedFilter::NotDeleted, ring.layout.node_id_vec.clone())),
10,
EnumerationOrder::Forward,
)
.await?;
if !k2vindexes.is_empty() {
return Ok(false);
}
}
Ok(true)
}
pub async fn cleanup_incomplete_uploads(
&self,
bucket_id: &Uuid,
older_than: Duration,
) -> Result<usize, Error> {
let older_than = now_msec() - older_than.as_millis() as u64;
let mut ret = 0usize;
let mut start = None;
loop {
let objects = self
.0
.object_table
.get_range(
bucket_id,
start,
Some(ObjectFilter::IsUploading),
1000,
EnumerationOrder::Forward,
)
.await?;
let abortions = objects
.iter()
.filter_map(|object| {
let aborted_versions = object
.versions()
.iter()
.filter(|v| v.is_uploading() && v.timestamp < older_than)
.map(|v| ObjectVersion {
state: ObjectVersionState::Aborted,
uuid: v.uuid,
timestamp: v.timestamp,
})
.collect::<Vec<_>>();
if !aborted_versions.is_empty() {
Some(Object::new(
object.bucket_id,
object.key.clone(),
aborted_versions,
))
} else {
None
}
})
.collect::<Vec<_>>();
ret += abortions.len();
self.0.object_table.insert_many(abortions).await?;
if objects.len() < 1000 {
break;
} else {
start = Some(objects.last().unwrap().key.clone());
}
}
Ok(ret)
}
}