From 5b18fd8201af98798f0a47a95dfec2f3ab9777f8 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Fri, 4 Nov 2022 11:55:59 +0100 Subject: Add garage bucket cleanup-incomplete-uploads command --- src/garage/Cargo.toml | 1 + src/garage/admin.rs | 39 ++++++++++++++++++++++++++ src/garage/cli/structs.rs | 15 ++++++++++ src/model/helper/bucket.rs | 69 +++++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 123 insertions(+), 1 deletion(-) (limited to 'src') diff --git a/src/garage/Cargo.toml b/src/garage/Cargo.toml index cbc0dc61..35caf2c1 100644 --- a/src/garage/Cargo.toml +++ b/src/garage/Cargo.toml @@ -33,6 +33,7 @@ garage_web = { version = "0.8.0", path = "../web" } bytes = "1.0" bytesize = "1.1" timeago = "0.3" +parse_duration = "2.1" hex = "0.4" tracing = { version = "0.1.30", features = ["log-always"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/src/garage/admin.rs b/src/garage/admin.rs index 802a8261..c21286e5 100644 --- a/src/garage/admin.rs +++ b/src/garage/admin.rs @@ -85,6 +85,9 @@ impl AdminRpcHandler { 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 + } } } @@ -512,6 +515,42 @@ impl AdminRpcHandler { ))) } + 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("Bucket not found")?, + ); + } + + let duration = parse_duration::parse::parse(&query.older_than) + .ok_or_bad_request("Invalid duration")?; + + 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)) + } + async fn handle_key_cmd(&self, cmd: &KeyOperation) -> Result { match cmd { KeyOperation::List => self.handle_list_keys().await, diff --git a/src/garage/cli/structs.rs b/src/garage/cli/structs.rs index 06548e89..cb085813 100644 --- a/src/garage/cli/structs.rs +++ b/src/garage/cli/structs.rs @@ -189,6 +189,10 @@ pub enum BucketOperation { /// Set the quotas for this bucket #[structopt(name = "set-quotas", version = garage_version())] SetQuotas(SetQuotasOpt), + + /// Clean up (abort) old incomplete multipart uploads + #[structopt(name = "cleanup-incomplete-uploads", version = garage_version())] + CleanupIncompleteUploads(CleanupIncompleteUploadsOpt), } #[derive(Serialize, Deserialize, StructOpt, Debug)] @@ -290,6 +294,17 @@ pub struct SetQuotasOpt { pub max_objects: Option, } +#[derive(Serialize, Deserialize, StructOpt, Debug)] +pub struct CleanupIncompleteUploadsOpt { + /// Abort multipart uploads older than this value + #[structopt(long = "older-than", default_value = "1d")] + pub older_than: String, + + /// Name of bucket(s) to clean up + #[structopt(required = true)] + pub buckets: Vec, +} + #[derive(Serialize, Deserialize, StructOpt, Debug)] pub enum KeyOperation { /// List keys diff --git a/src/model/helper/bucket.rs b/src/model/helper/bucket.rs index 130ba5be..4a488d7f 100644 --- a/src/model/helper/bucket.rs +++ b/src/model/helper/bucket.rs @@ -1,3 +1,5 @@ +use std::time::Duration; + use garage_util::crdt::*; use garage_util::data::*; use garage_util::error::{Error as GarageError, OkOrMessage}; @@ -12,7 +14,7 @@ use crate::helper::error::*; use crate::helper::key::KeyHelper; use crate::key_table::*; use crate::permission::BucketKeyPerm; -use crate::s3::object_table::ObjectFilter; +use crate::s3::object_table::*; pub struct BucketHelper<'a>(pub(crate) &'a Garage); @@ -472,4 +474,69 @@ impl<'a> BucketHelper<'a> { Ok(true) } + + // ---- + + /// Deletes all incomplete multipart uploads that are older than a certain time. + /// Returns the number of uploads aborted + pub async fn cleanup_incomplete_uploads( + &self, + bucket_id: &Uuid, + older_than: Duration, + ) -> Result { + 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::>(); + if !aborted_versions.is_empty() { + Some(Object::new( + object.bucket_id, + object.key.clone(), + aborted_versions, + )) + } else { + None + } + }) + .collect::>(); + + 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) + } } -- cgit v1.2.3 From 8d3bbf5703f97ff1f468bc09fc515b6c12b027a3 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Fri, 4 Nov 2022 16:07:33 +0100 Subject: Clearer error messsages --- src/garage/admin.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'src') diff --git a/src/garage/admin.rs b/src/garage/admin.rs index c21286e5..e973cfe7 100644 --- a/src/garage/admin.rs +++ b/src/garage/admin.rs @@ -526,12 +526,12 @@ impl AdminRpcHandler { .bucket_helper() .resolve_global_bucket_name(b) .await? - .ok_or_bad_request("Bucket not found")?, + .ok_or_bad_request(format!("Bucket not found: {}", b))?, ); } let duration = parse_duration::parse::parse(&query.older_than) - .ok_or_bad_request("Invalid duration")?; + .ok_or_bad_request("Invalid duration passed for --older-than parameter")?; let mut ret = String::new(); for bucket in bucket_ids { -- cgit v1.2.3 From e03d9062f7f21dd0493dd82a7dcf82f2cd035943 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Fri, 4 Nov 2022 16:33:05 +0100 Subject: Show a nice message and a backtrace when Garage panics --- src/garage/Cargo.toml | 1 + src/garage/main.rs | 56 ++++++++++++++++++++++++++++++++++++--------------- 2 files changed, 41 insertions(+), 16 deletions(-) (limited to 'src') diff --git a/src/garage/Cargo.toml b/src/garage/Cargo.toml index 35caf2c1..69852db7 100644 --- a/src/garage/Cargo.toml +++ b/src/garage/Cargo.toml @@ -30,6 +30,7 @@ garage_table = { version = "0.8.0", path = "../table" } garage_util = { version = "0.8.0", path = "../util" } garage_web = { version = "0.8.0", path = "../web" } +backtrace = "0.3" bytes = "1.0" bytesize = "1.1" timeago = "0.3" diff --git a/src/garage/main.rs b/src/garage/main.rs index 5b2a85c0..edda734b 100644 --- a/src/garage/main.rs +++ b/src/garage/main.rs @@ -65,21 +65,6 @@ struct Opt { #[tokio::main] async fn main() { - if std::env::var("RUST_LOG").is_err() { - std::env::set_var("RUST_LOG", "netapp=info,garage=info") - } - tracing_subscriber::fmt() - .with_writer(std::io::stderr) - .with_env_filter(tracing_subscriber::filter::EnvFilter::from_default_env()) - .init(); - sodiumoxide::init().expect("Unable to init sodiumoxide"); - - // Abort on panic (same behavior as in Go) - std::panic::set_hook(Box::new(|panic_info| { - error!("{}", panic_info.to_string()); - std::process::abort(); - })); - // Initialize version and features info let features = &[ #[cfg(feature = "k2v")] @@ -108,12 +93,51 @@ async fn main() { } garage_util::version::init_features(features); - // Parse arguments let version = format!( "{} [features: {}]", garage_util::version::garage_version(), features.join(", ") ); + + // Initialize panic handler that aborts on panic and shows a nice message. + // By default, Tokio continues runing normally when a task panics. We want + // to avoid this behavior in Garage as this would risk putting the process in an + // unknown/uncontrollable state. We prefer to exit the process and restart it + // from scratch, so that it boots back into a fresh, known state. + let panic_version_info = version.clone(); + std::panic::set_hook(Box::new(move |panic_info| { + eprintln!("======== PANIC (internal Garage error) ========"); + eprintln!("{}", panic_info); + eprintln!(); + eprintln!("Panics are internal errors that Garage is unable to handle on its own."); + eprintln!("They can be caused by bugs in Garage's code, or by corrupted data in"); + eprintln!("the node's storage. If you feel that this error is likely to be a bug"); + eprintln!("in Garage, please report it on our issue tracker a the following address:"); + eprintln!(); + eprintln!(" https://git.deuxfleurs.fr/Deuxfleurs/garage/issues"); + eprintln!(); + eprintln!("Please include the last log messages and the the full backtrace below in"); + eprintln!("your bug report, as well as any relevant information on the context in"); + eprintln!("which Garage was running when this error occurred."); + eprintln!(); + eprintln!("GARAGE VERSION: {}", panic_version_info); + eprintln!(); + eprintln!("BACKTRACE:"); + eprintln!("{:?}", backtrace::Backtrace::new()); + std::process::abort(); + })); + + // Initialize logging as well as other libraries used in Garage + if std::env::var("RUST_LOG").is_err() { + std::env::set_var("RUST_LOG", "netapp=info,garage=info") + } + tracing_subscriber::fmt() + .with_writer(std::io::stderr) + .with_env_filter(tracing_subscriber::filter::EnvFilter::from_default_env()) + .init(); + sodiumoxide::init().expect("Unable to init sodiumoxide"); + + // Parse arguments and dispatch command line let opt = Opt::from_clap(&Opt::clap().version(version.as_str()).get_matches()); let res = match opt.cmd { -- cgit v1.2.3