From 8a74e1c2bd7f202dfaa55ecc16a91ad84812fff9 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 6 Jun 2023 15:39:15 +0200 Subject: Split garage/admin.rs into smaller files --- src/garage/admin/bucket.rs | 496 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 496 insertions(+) create mode 100644 src/garage/admin/bucket.rs (limited to 'src/garage/admin/bucket.rs') diff --git a/src/garage/admin/bucket.rs b/src/garage/admin/bucket.rs new file mode 100644 index 00000000..11bb8730 --- /dev/null +++ b/src/garage/admin/bucket.rs @@ -0,0 +1,496 @@ +use std::collections::HashMap; +use std::fmt::Write; + +use garage_util::crdt::*; +use garage_util::time::*; + +use garage_table::*; + +use garage_model::bucket_alias_table::*; +use garage_model::bucket_table::*; +use garage_model::helper::error::{Error, OkOrBadRequest}; +use garage_model::permission::*; + +use crate::cli::*; + +use super::*; + +impl AdminRpcHandler { + pub(super) async fn handle_bucket_cmd(&self, cmd: &BucketOperation) -> Result { + match cmd { + BucketOperation::List => self.handle_list_buckets().await, + BucketOperation::Info(query) => self.handle_bucket_info(query).await, + BucketOperation::Create(query) => self.handle_create_bucket(&query.name).await, + BucketOperation::Delete(query) => self.handle_delete_bucket(query).await, + BucketOperation::Alias(query) => self.handle_alias_bucket(query).await, + BucketOperation::Unalias(query) => self.handle_unalias_bucket(query).await, + BucketOperation::Allow(query) => self.handle_bucket_allow(query).await, + BucketOperation::Deny(query) => self.handle_bucket_deny(query).await, + BucketOperation::Website(query) => self.handle_bucket_website(query).await, + BucketOperation::SetQuotas(query) => self.handle_bucket_set_quotas(query).await, + BucketOperation::CleanupIncompleteUploads(query) => { + self.handle_bucket_cleanup_incomplete_uploads(query).await + } + } + } + + async fn handle_list_buckets(&self) -> Result { + let buckets = self + .garage + .bucket_table + .get_range( + &EmptyKey, + None, + Some(DeletedFilter::NotDeleted), + 10000, + EnumerationOrder::Forward, + ) + .await?; + + Ok(AdminRpc::BucketList(buckets)) + } + + async fn handle_bucket_info(&self, query: &BucketOpt) -> Result { + let bucket_id = self + .garage + .bucket_helper() + .resolve_global_bucket_name(&query.name) + .await? + .ok_or_bad_request("Bucket not found")?; + + let bucket = self + .garage + .bucket_helper() + .get_existing_bucket(bucket_id) + .await?; + + let counters = self + .garage + .object_counter_table + .table + .get(&bucket_id, &EmptyKey) + .await? + .map(|x| x.filtered_values(&self.garage.system.ring.borrow())) + .unwrap_or_default(); + + let mut relevant_keys = HashMap::new(); + for (k, _) in bucket + .state + .as_option() + .unwrap() + .authorized_keys + .items() + .iter() + { + if let Some(key) = self + .garage + .key_table + .get(&EmptyKey, k) + .await? + .filter(|k| !k.is_deleted()) + { + relevant_keys.insert(k.clone(), key); + } + } + for ((k, _), _, _) in bucket + .state + .as_option() + .unwrap() + .local_aliases + .items() + .iter() + { + if relevant_keys.contains_key(k) { + continue; + } + if let Some(key) = self.garage.key_table.get(&EmptyKey, k).await? { + relevant_keys.insert(k.clone(), key); + } + } + + Ok(AdminRpc::BucketInfo { + bucket, + relevant_keys, + counters, + }) + } + + #[allow(clippy::ptr_arg)] + async fn handle_create_bucket(&self, name: &String) -> Result { + if !is_valid_bucket_name(name) { + return Err(Error::BadRequest(format!( + "{}: {}", + name, INVALID_BUCKET_NAME_MESSAGE + ))); + } + + if let Some(alias) = self.garage.bucket_alias_table.get(&EmptyKey, name).await? { + if alias.state.get().is_some() { + return Err(Error::BadRequest(format!("Bucket {} already exists", name))); + } + } + + // ---- done checking, now commit ---- + + let bucket = Bucket::new(); + self.garage.bucket_table.insert(&bucket).await?; + + self.garage + .bucket_helper() + .set_global_bucket_alias(bucket.id, name) + .await?; + + Ok(AdminRpc::Ok(format!("Bucket {} was created.", name))) + } + + async fn handle_delete_bucket(&self, query: &DeleteBucketOpt) -> Result { + let helper = self.garage.bucket_helper(); + + let bucket_id = helper + .resolve_global_bucket_name(&query.name) + .await? + .ok_or_bad_request("Bucket not found")?; + + // Get the alias, but keep in minde here the bucket name + // given in parameter can also be directly the bucket's ID. + // In that case bucket_alias will be None, and + // we can still delete the bucket if it has zero aliases + // (a condition which we try to prevent but that could still happen somehow). + // We just won't try to delete an alias entry because there isn't one. + let bucket_alias = self + .garage + .bucket_alias_table + .get(&EmptyKey, &query.name) + .await?; + + // Check bucket doesn't have other aliases + let mut bucket = helper.get_existing_bucket(bucket_id).await?; + let bucket_state = bucket.state.as_option().unwrap(); + if bucket_state + .aliases + .items() + .iter() + .filter(|(_, _, active)| *active) + .any(|(name, _, _)| name != &query.name) + { + return Err(Error::BadRequest(format!("Bucket {} still has other global aliases. Use `bucket unalias` to delete them one by one.", query.name))); + } + if bucket_state + .local_aliases + .items() + .iter() + .any(|(_, _, active)| *active) + { + return Err(Error::BadRequest(format!("Bucket {} still has other local aliases. Use `bucket unalias` to delete them one by one.", query.name))); + } + + // Check bucket is empty + if !helper.is_bucket_empty(bucket_id).await? { + return Err(Error::BadRequest(format!( + "Bucket {} is not empty", + query.name + ))); + } + + if !query.yes { + return Err(Error::BadRequest( + "Add --yes flag to really perform this operation".to_string(), + )); + } + + // --- done checking, now commit --- + // 1. delete authorization from keys that had access + for (key_id, _) in bucket.authorized_keys() { + helper + .set_bucket_key_permissions(bucket.id, key_id, BucketKeyPerm::NO_PERMISSIONS) + .await?; + } + + // 2. delete bucket alias + if bucket_alias.is_some() { + helper + .purge_global_bucket_alias(bucket_id, &query.name) + .await?; + } + + // 3. delete bucket + bucket.state = Deletable::delete(); + self.garage.bucket_table.insert(&bucket).await?; + + Ok(AdminRpc::Ok(format!("Bucket {} was deleted.", query.name))) + } + + async fn handle_alias_bucket(&self, query: &AliasBucketOpt) -> Result { + let helper = self.garage.bucket_helper(); + let key_helper = self.garage.key_helper(); + + let bucket_id = helper + .resolve_global_bucket_name(&query.existing_bucket) + .await? + .ok_or_bad_request("Bucket not found")?; + + if let Some(key_pattern) = &query.local { + let key = key_helper.get_existing_matching_key(key_pattern).await?; + + helper + .set_local_bucket_alias(bucket_id, &key.key_id, &query.new_name) + .await?; + Ok(AdminRpc::Ok(format!( + "Alias {} now points to bucket {:?} in namespace of key {}", + query.new_name, bucket_id, key.key_id + ))) + } else { + helper + .set_global_bucket_alias(bucket_id, &query.new_name) + .await?; + Ok(AdminRpc::Ok(format!( + "Alias {} now points to bucket {:?}", + query.new_name, bucket_id + ))) + } + } + + async fn handle_unalias_bucket(&self, query: &UnaliasBucketOpt) -> Result { + let helper = self.garage.bucket_helper(); + let key_helper = self.garage.key_helper(); + + if let Some(key_pattern) = &query.local { + let key = key_helper.get_existing_matching_key(key_pattern).await?; + + let bucket_id = key + .state + .as_option() + .unwrap() + .local_aliases + .get(&query.name) + .cloned() + .flatten() + .ok_or_bad_request("Bucket not found")?; + + helper + .unset_local_bucket_alias(bucket_id, &key.key_id, &query.name) + .await?; + + Ok(AdminRpc::Ok(format!( + "Alias {} no longer points to bucket {:?} in namespace of key {}", + &query.name, bucket_id, key.key_id + ))) + } else { + let bucket_id = helper + .resolve_global_bucket_name(&query.name) + .await? + .ok_or_bad_request("Bucket not found")?; + + helper + .unset_global_bucket_alias(bucket_id, &query.name) + .await?; + + Ok(AdminRpc::Ok(format!( + "Alias {} no longer points to bucket {:?}", + &query.name, bucket_id + ))) + } + } + + async fn handle_bucket_allow(&self, query: &PermBucketOpt) -> Result { + let helper = self.garage.bucket_helper(); + let key_helper = self.garage.key_helper(); + + let bucket_id = helper + .resolve_global_bucket_name(&query.bucket) + .await? + .ok_or_bad_request("Bucket not found")?; + let key = key_helper + .get_existing_matching_key(&query.key_pattern) + .await?; + + let allow_read = query.read || key.allow_read(&bucket_id); + let allow_write = query.write || key.allow_write(&bucket_id); + let allow_owner = query.owner || key.allow_owner(&bucket_id); + + helper + .set_bucket_key_permissions( + bucket_id, + &key.key_id, + BucketKeyPerm { + timestamp: now_msec(), + allow_read, + allow_write, + allow_owner, + }, + ) + .await?; + + Ok(AdminRpc::Ok(format!( + "New permissions for {} on {}: read {}, write {}, owner {}.", + &key.key_id, &query.bucket, allow_read, allow_write, allow_owner + ))) + } + + async fn handle_bucket_deny(&self, query: &PermBucketOpt) -> Result { + let helper = self.garage.bucket_helper(); + let key_helper = self.garage.key_helper(); + + let bucket_id = helper + .resolve_global_bucket_name(&query.bucket) + .await? + .ok_or_bad_request("Bucket not found")?; + let key = key_helper + .get_existing_matching_key(&query.key_pattern) + .await?; + + let allow_read = !query.read && key.allow_read(&bucket_id); + let allow_write = !query.write && key.allow_write(&bucket_id); + let allow_owner = !query.owner && key.allow_owner(&bucket_id); + + helper + .set_bucket_key_permissions( + bucket_id, + &key.key_id, + BucketKeyPerm { + timestamp: now_msec(), + allow_read, + allow_write, + allow_owner, + }, + ) + .await?; + + Ok(AdminRpc::Ok(format!( + "New permissions for {} on {}: read {}, write {}, owner {}.", + &key.key_id, &query.bucket, allow_read, allow_write, allow_owner + ))) + } + + async fn handle_bucket_website(&self, query: &WebsiteOpt) -> Result { + let bucket_id = self + .garage + .bucket_helper() + .resolve_global_bucket_name(&query.bucket) + .await? + .ok_or_bad_request("Bucket not found")?; + + let mut bucket = self + .garage + .bucket_helper() + .get_existing_bucket(bucket_id) + .await?; + let bucket_state = bucket.state.as_option_mut().unwrap(); + + if !(query.allow ^ query.deny) { + return Err(Error::BadRequest( + "You must specify exactly one flag, either --allow or --deny".to_string(), + )); + } + + let website = if query.allow { + Some(WebsiteConfig { + index_document: query.index_document.clone(), + error_document: query.error_document.clone(), + }) + } else { + None + }; + + bucket_state.website_config.update(website); + self.garage.bucket_table.insert(&bucket).await?; + + let msg = if query.allow { + format!("Website access allowed for {}", &query.bucket) + } else { + format!("Website access denied for {}", &query.bucket) + }; + + Ok(AdminRpc::Ok(msg)) + } + + async fn handle_bucket_set_quotas(&self, query: &SetQuotasOpt) -> Result { + let bucket_id = self + .garage + .bucket_helper() + .resolve_global_bucket_name(&query.bucket) + .await? + .ok_or_bad_request("Bucket not found")?; + + let mut bucket = self + .garage + .bucket_helper() + .get_existing_bucket(bucket_id) + .await?; + let bucket_state = bucket.state.as_option_mut().unwrap(); + + if query.max_size.is_none() && query.max_objects.is_none() { + return Err(Error::BadRequest( + "You must specify either --max-size or --max-objects (or both) for this command to do something.".to_string(), + )); + } + + let mut quotas = bucket_state.quotas.get().clone(); + + match query.max_size.as_ref().map(String::as_ref) { + Some("none") => quotas.max_size = None, + Some(v) => { + let bs = v + .parse::() + .ok_or_bad_request(format!("Invalid size specified: {}", v))?; + quotas.max_size = Some(bs.as_u64()); + } + _ => (), + } + + match query.max_objects.as_ref().map(String::as_ref) { + Some("none") => quotas.max_objects = None, + Some(v) => { + let mo = v + .parse::() + .ok_or_bad_request(format!("Invalid number specified: {}", v))?; + quotas.max_objects = Some(mo); + } + _ => (), + } + + bucket_state.quotas.update(quotas); + self.garage.bucket_table.insert(&bucket).await?; + + Ok(AdminRpc::Ok(format!( + "Quotas updated for {}", + &query.bucket + ))) + } + + async fn handle_bucket_cleanup_incomplete_uploads( + &self, + query: &CleanupIncompleteUploadsOpt, + ) -> Result { + let mut bucket_ids = vec![]; + for b in query.buckets.iter() { + bucket_ids.push( + self.garage + .bucket_helper() + .resolve_global_bucket_name(b) + .await? + .ok_or_bad_request(format!("Bucket not found: {}", b))?, + ); + } + + let duration = parse_duration::parse::parse(&query.older_than) + .ok_or_bad_request("Invalid duration passed for --older-than parameter")?; + + let mut ret = String::new(); + for bucket in bucket_ids { + let count = self + .garage + .bucket_helper() + .cleanup_incomplete_uploads(&bucket, duration) + .await?; + writeln!( + &mut ret, + "Bucket {:?}: {} incomplete uploads aborted", + bucket, count + ) + .unwrap(); + } + + Ok(AdminRpc::Ok(ret)) + } +} -- cgit v1.2.3 From 511e07ecd489fa72040171fe908323873a57ac19 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Thu, 4 May 2023 11:49:23 +0200 Subject: fix mpu counter (add missing workers) and report info at appropriate places --- src/garage/admin/bucket.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) (limited to 'src/garage/admin/bucket.rs') diff --git a/src/garage/admin/bucket.rs b/src/garage/admin/bucket.rs index 11bb8730..0781cb8b 100644 --- a/src/garage/admin/bucket.rs +++ b/src/garage/admin/bucket.rs @@ -73,6 +73,15 @@ impl AdminRpcHandler { .map(|x| x.filtered_values(&self.garage.system.ring.borrow())) .unwrap_or_default(); + let mpu_counters = self + .garage + .mpu_counter_table + .table + .get(&bucket_id, &EmptyKey) + .await? + .map(|x| x.filtered_values(&self.garage.system.ring.borrow())) + .unwrap_or_default(); + let mut relevant_keys = HashMap::new(); for (k, _) in bucket .state @@ -112,6 +121,7 @@ impl AdminRpcHandler { bucket, relevant_keys, counters, + mpu_counters, }) } -- cgit v1.2.3