cli_v2: migrate cleanupincompleteuploads to Admin API

src/api/admin/api.rs

@@ -61,6 +61,7 @@ admin_endpoints![
 	CreateBucket,
 	UpdateBucket,
 	DeleteBucket,
+	CleanupIncompleteUploads,
 
 	// Operations on permissions for keys on buckets
 	AllowBucketKey,
@@ -496,6 +497,19 @@ pub struct DeleteBucketRequest {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DeleteBucketResponse;
 
+// ---- CleanupIncompleteUploads ----
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CleanupIncompleteUploadsRequest {
+	pub bucket_id: String,
+	pub older_than_secs: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CleanupIncompleteUploadsResponse {
+	pub uploads_deleted: u64,
+}
+
 // **********************************************
 //      Operations on permissions for keys on buckets
 // **********************************************

src/api/admin/bucket.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 
 use async_trait::async_trait;
 
@@ -387,6 +388,26 @@ impl EndpointHandler for UpdateBucketRequest {
 	}
 }
 
+#[async_trait]
+impl EndpointHandler for CleanupIncompleteUploadsRequest {
+	type Response = CleanupIncompleteUploadsResponse;
+
+	async fn handle(self, garage: &Arc<Garage>) -> Result<CleanupIncompleteUploadsResponse, Error> {
+		let duration = Duration::from_secs(self.older_than_secs);
+
+		let bucket_id = parse_bucket_id(&self.bucket_id)?;
+
+		let count = garage
+			.bucket_helper()
+			.cleanup_incomplete_uploads(&bucket_id, duration)
+			.await?;
+
+		Ok(CleanupIncompleteUploadsResponse {
+			uploads_deleted: count as u64,
+		})
+	}
+}
+
 // ---- BUCKET/KEY PERMISSIONS ----
 
 #[async_trait]

src/api/admin/router_v2.rs

@@ -51,6 +51,7 @@ impl AdminApiRequest {
 			POST CreateBucket (body),
 			POST DeleteBucket (query::id),
 			POST UpdateBucket (body_field, query::id),
+			POST CleanupIncompleteUploads (body),
 			// Bucket-key permissions
 			POST AllowBucketKey (body),
 			POST DenyBucketKey (body),

src/garage/admin/bucket.rs

@@ -1,53 +0,0 @@
-use std::fmt::Write;
-
-use garage_model::helper::error::{Error, OkOrBadRequest};
-
-use crate::cli::*;
-
-use super::*;
-
-impl AdminRpcHandler {
-	pub(super) async fn handle_bucket_cmd(&self, cmd: &BucketOperation) -> Result<AdminRpc, Error> {
-		match cmd {
-			BucketOperation::CleanupIncompleteUploads(query) => {
-				self.handle_bucket_cleanup_incomplete_uploads(query).await
-			}
-			_ => unreachable!(),
-		}
-	}
-
-	async fn handle_bucket_cleanup_incomplete_uploads(
-		&self,
-		query: &CleanupIncompleteUploadsOpt,
-	) -> Result<AdminRpc, Error> {
-		let mut bucket_ids = vec![];
-		for b in query.buckets.iter() {
-			bucket_ids.push(
-				self.garage
-					.bucket_helper()
-					.admin_get_existing_matching_bucket(b)
-					.await?,
-			);
-		}
-
-		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))
-	}
-}

src/garage/admin/mod.rs

@@ -1,5 +1,4 @@
 mod block;
-mod bucket;
 
 use std::collections::HashMap;
 use std::fmt::Write;
@@ -39,7 +38,6 @@ pub const ADMIN_RPC_PATH: &str = "garage/admin_rpc.rs/Rpc";
 #[derive(Debug, Serialize, Deserialize)]
 #[allow(clippy::large_enum_variant)]
 pub enum AdminRpc {
-	BucketOperation(BucketOperation),
 	LaunchRepair(RepairOpt),
 	Stats(StatsOpt),
 	Worker(WorkerOperation),
@@ -532,7 +530,6 @@ impl EndpointHandler<AdminRpc> for AdminRpcHandler {
 		_from: NodeID,
 	) -> Result<AdminRpc, Error> {
 		match message {
-			AdminRpc::BucketOperation(bo) => self.handle_bucket_cmd(bo).await,
 			AdminRpc::LaunchRepair(opt) => self.handle_launch_repair(opt.clone()).await,
 			AdminRpc::Stats(opt) => self.handle_stats(opt.clone()).await,
 			AdminRpc::Worker(wo) => self.handle_worker_cmd(wo).await,

src/garage/cli_v2/bucket.rs

@@ -5,7 +5,6 @@ use garage_util::error::*;
 
 use garage_api::admin::api::*;
 
-use crate::cli as cli_v1;
 use crate::cli::structs::*;
 use crate::cli_v2::*;
 
@@ -22,15 +21,9 @@ impl Cli {
 			BucketOperation::Deny(query) => self.cmd_bucket_deny(query).await,
 			BucketOperation::Website(query) => self.cmd_bucket_website(query).await,
 			BucketOperation::SetQuotas(query) => self.cmd_bucket_set_quotas(query).await,
-
+			BucketOperation::CleanupIncompleteUploads(query) => {
-			// TODO
+				self.cmd_cleanup_incomplete_uploads(query).await
-			x => cli_v1::cmd_admin(
+			}
-				&self.admin_rpc_endpoint,
-				self.rpc_host,
-				AdminRpc::BucketOperation(x),
-			)
-			.await
-			.ok_or_message("old error"),
 		}
 	}
 
@@ -520,4 +513,37 @@ impl Cli {
 
 		Ok(())
 	}
+
+	pub async fn cmd_cleanup_incomplete_uploads(
+		&self,
+		opt: CleanupIncompleteUploadsOpt,
+	) -> Result<(), Error> {
+		let older_than = parse_duration::parse::parse(&opt.older_than)
+			.ok_or_message("Invalid duration passed for --older-than parameter")?;
+
+		for b in opt.buckets.iter() {
+			let bucket = self
+				.api_request(GetBucketInfoRequest {
+					id: None,
+					global_alias: None,
+					search: Some(b.clone()),
+				})
+				.await?;
+
+			let res = self
+				.api_request(CleanupIncompleteUploadsRequest {
+					bucket_id: bucket.id.clone(),
+					older_than_secs: older_than.as_secs(),
+				})
+				.await?;
+
+			if res.uploads_deleted > 0 {
+				println!("{:.16}: {} uploads deleted", bucket.id, res.uploads_deleted);
+			} else {
+				println!("{:.16}: no uploads deleted", bucket.id);
+			}
+		}
+
+		Ok(())
+	}
 }