Deuxfleurs/garage
48
55
Fork 73
Code Issues 100 Pull requests 14 Releases 30 Wiki Activity

improve internal item counter mechanisms and implement bucket quotas #326

Merged
lx merged 23 commits from counters into main 2022-06-15 18:20:31 +00:00
Conversation 1 Commits 23 Files changed 32 +97 -5
5 changed files with 97 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit c1baa10202 - Show all commits

55
src/api/admin/bucket.rs
View file

@ -14,6 +14,7 @@ use garage_model::bucket_alias_table::*;
use garage_model::bucket_table::*;
use garage_model::garage::Garage;
use garage_model::permission::*;
use garage_model::s3::object_table::*;
use crate::admin::error::*;
use crate::admin::key::ApiBucketKeyPerm;
@ -32,10 +33,28 @@ pub async fn handle_list_buckets(garage: &Arc<Garage>) -> Result<Response<Body>,
)
.await?;
let ring = garage.system.ring.borrow().clone();
let counters = garage
.object_counter_table
.table
.get_range(
&EmptyKey,
None,
Some((DeletedFilter::NotDeleted, ring.layout.node_id_vec.clone())),
15000,
EnumerationOrder::Forward,
)
.await?
.iter()
.map(|x| (x.sk, x.filtered_values(&ring)))
.collect::<HashMap<_, _>>();
let res = buckets
.into_iter()
.map(|b| {
let state = b.state.as_option().unwrap();
let empty_cnts = HashMap::new();
let cnts = counters.get(&b.id).unwrap_or(&empty_cnts);
ListBucketResultItem {
id: hex::encode(b.id),
global_aliases: state
@ -55,6 +74,9 @@ pub async fn handle_list_buckets(garage: &Arc<Garage>) -> Result<Response<Body>,
alias: n.to_string(),
})
.collect::<Vec<_>>(),
objects: cnts.get(OBJECTS).cloned().unwrap_or_default(),
bytes: cnts.get(BYTES).cloned().unwrap_or_default(),
unfinshed_uploads: cnts.get(UNFINISHED_UPLOADS).cloned().unwrap_or_default(),
}
})
.collect::<Vec<_>>();
@ -68,6 +90,9 @@ struct ListBucketResultItem {
id: String,
global_aliases: Vec<String>,
local_aliases: Vec<BucketLocalAlias>,
objects: i64,
bytes: i64,
unfinshed_uploads: i64,
}
#[derive(Serialize)]
@ -77,6 +102,13 @@ struct BucketLocalAlias {
alias: String,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct ApiBucketQuotas {
max_size: Option<u64>,
max_objects: Option<u64>,
}
pub async fn handle_get_bucket_info(
garage: &Arc<Garage>,
id: Option<String>,
@ -108,6 +140,14 @@ async fn bucket_info_results(
.get_existing_bucket(bucket_id)
.await?;
let counters = garage
.object_counter_table
.table
.get(&EmptyKey, &bucket_id)
.await?
.map(|x| x.filtered_values(&garage.system.ring.borrow()))
.unwrap_or_default();
let mut relevant_keys = HashMap::new();
for (k, _) in bucket
.state
@ -148,6 +188,7 @@ async fn bucket_info_results(
let state = bucket.state.as_option().unwrap();
let quotas = state.quotas.get();
let res =
GetBucketInfoResult {
id: hex::encode(&bucket.id),
@ -191,6 +232,16 @@ async fn bucket_info_results(
}
})
.collect::<Vec<_>>(),
objects: counters.get(OBJECTS).cloned().unwrap_or_default(),
bytes: counters.get(BYTES).cloned().unwrap_or_default(),
unfinshed_uploads: counters
.get(UNFINISHED_UPLOADS)
.cloned()
.unwrap_or_default(),
quotas: ApiBucketQuotas {
max_size: quotas.max_size,
max_objects: quotas.max_objects,
},
};
Ok(json_ok_response(&res)?)
@ -205,6 +256,10 @@ struct GetBucketInfoResult {
#[serde(default)]
website_config: Option<GetBucketInfoWebsiteResult>,
keys: Vec<GetBucketInfoKey>,
objects: i64,
bytes: i64,
unfinshed_uploads: i64,
quotas: ApiBucketQuotas,
}
#[derive(Serialize)]

20
src/garage/cli/util.rs
View file

@ -154,7 +154,7 @@ pub fn print_bucket_info(
let size =
bytesize::ByteSize::b(counters.get(BYTES).cloned().unwrap_or_default() as u64);
println!(
"Size: {} ({})",
"\nSize: {} ({})",
size.to_string_as(true),
size.to_string_as(false)
);
@ -170,7 +170,23 @@ pub fn print_bucket_info(
.unwrap_or_default()
);
println!("Website access: {}", p.website_config.get().is_some());
println!("\nWebsite access: {}", p.website_config.get().is_some());
let quotas = p.quotas.get();
if quotas.max_size.is_some() || quotas.max_objects.is_some() {
println!("\nQuotas:");
if let Some(ms) = quotas.max_size {
let ms = bytesize::ByteSize::b(ms);
println!(
" maximum size: {} ({})",
ms.to_string_as(true),
ms.to_string_as(false)
);
}
if let Some(mo) = quotas.max_objects {
println!(" maximum number of objects: {}", mo);
}
}
println!("\nGlobal aliases:");
for (alias, _, active) in p.aliases.items().iter() {

19
src/model/bucket_table.rs
View file

@ -1,6 +1,6 @@
use serde::{Deserialize, Serialize};
use garage_table::crdt::Crdt;
use garage_table::crdt::*;
use garage_table::*;
use garage_util::data::*;
use garage_util::time::*;
@ -44,6 +44,9 @@ pub struct BucketParams {
pub website_config: crdt::Lww<Option<WebsiteConfig>>,
/// CORS rules
pub cors_config: crdt::Lww<Option<Vec<CorsRule>>>,
/// Bucket quotas
#[serde(default)]
pub quotas: crdt::Lww<BucketQuotas>,
}
#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
@ -62,6 +65,18 @@ pub struct CorsRule {
pub expose_headers: Vec<String>,
}
#[derive(Default, PartialEq, Eq, PartialOrd, Ord, Clone, Debug, Serialize, Deserialize)]
pub struct BucketQuotas {
/// Maximum size in bytes (bucket size = sum of sizes of objects in the bucket)
pub max_size: Option<u64>,
/// Maximum number of non-deleted objects in the bucket
pub max_objects: Option<u64>,
}
impl AutoCrdt for BucketQuotas {
const WARN_IF_DIFFERENT: bool = true;
}
impl BucketParams {
/// Create an empty BucketParams with no authorized keys and no website accesss
pub fn new() -> Self {
@ -72,6 +87,7 @@ impl BucketParams {
local_aliases: crdt::LwwMap::new(),
website_config: crdt::Lww::new(None),
cors_config: crdt::Lww::new(None),
quotas: crdt::Lww::new(BucketQuotas::default()),
}
}
}
@ -86,6 +102,7 @@ impl Crdt for BucketParams {
self.website_config.merge(&o.website_config);
self.cors_config.merge(&o.cors_config);
self.quotas.merge(&o.quotas);
}
}

1
src/model/migrate.rs
View file

@ -77,6 +77,7 @@ impl Migrate {
local_aliases: LwwMap::new(),
website_config: Lww::new(website),
cors_config: Lww::new(None),
quotas: Lww::new(Default::default()),
}),
})
.await?;

7
src/model/s3/object_table.rs
View file

@ -315,9 +315,12 @@ impl CountedItem for Object {
}
fn counts(&self) -> Vec<(&'static str, i64)> {
let n_objects = if self.is_tombstone() { 0 } else { 1 };
let versions = self.versions();
let n_objects = if versions.iter().any(|v| v.is_data()) {
0
} else {
1
};
let n_unfinished_uploads = versions
.iter()
.filter(|v| matches!(v.state, ObjectVersionState::Uploading(_)))