cli_v2: implement ListWorkers and GetWorkerInfo

src/api/admin/api.rs

@@ -10,6 +10,7 @@ use serde::{Deserialize, Serialize};
 use garage_rpc::*;
 
 use garage_model::garage::Garage;
+use garage_util::error::Error as GarageError;
 
 use garage_api_common::common_error::CommonErrorDerivative;
 use garage_api_common::helpers::is_default;
@@ -78,11 +79,46 @@ admin_endpoints![
 	RemoveBucketAlias,
 
 	// Worker operations
+	ListWorkers,
+	GetWorkerInfo,
 	GetWorkerVariable,
 	SetWorkerVariable,
 ];
 
-local_admin_endpoints![GetWorkerVariable, SetWorkerVariable,];
+local_admin_endpoints![
+	// Background workers
+	ListWorkers,
+	GetWorkerInfo,
+	GetWorkerVariable,
+	SetWorkerVariable,
+];
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MultiRequest<RB> {
+	pub node: String,
+	pub body: RB,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct MultiResponse<RB> {
+	pub success: HashMap<String, RB>,
+	pub error: HashMap<String, String>,
+}
+
+impl<RB> MultiResponse<RB> {
+	pub fn into_single_response(self) -> Result<RB, GarageError> {
+		if let Some((_, e)) = self.error.into_iter().next() {
+			return Err(GarageError::Message(e));
+		}
+		if self.success.len() != 1 {
+			return Err(GarageError::Message(format!(
+				"{} responses returned, expected 1",
+				self.success.len()
+			)));
+		}
+		Ok(self.success.into_iter().next().unwrap().1)
+	}
+}
 
 // **********************************************
 //      Special endpoints
@@ -596,6 +632,61 @@ pub struct RemoveBucketAliasResponse(pub GetBucketInfoResponse);
 //      Worker operations
 // **********************************************
 
+// ---- GetWorkerList ----
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct LocalListWorkersRequest {
+	#[serde(default)]
+	pub busy_only: bool,
+	#[serde(default)]
+	pub error_only: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct LocalListWorkersResponse(pub Vec<WorkerInfoResp>);
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WorkerInfoResp {
+	pub id: u64,
+	pub name: String,
+	pub state: WorkerStateResp,
+	pub errors: u64,
+	pub consecutive_errors: u64,
+	pub last_error: Option<WorkerLastError>,
+	pub tranquility: Option<u32>,
+	pub progress: Option<String>,
+	pub queue_length: Option<u64>,
+	pub persistent_errors: Option<u64>,
+	pub freeform: Vec<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum WorkerStateResp {
+	Busy,
+	Throttled { duration_secs: f32 },
+	Idle,
+	Done,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WorkerLastError {
+	pub message: String,
+	pub secs_ago: u64,
+}
+
+// ---- GetWorkerList ----
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct LocalGetWorkerInfoRequest {
+	pub id: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct LocalGetWorkerInfoResponse(pub WorkerInfoResp);
+
 // ---- GetWorkerVariable ----
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/api/admin/error.rs

@@ -25,6 +25,10 @@ pub enum Error {
 	#[error(display = "Access key not found: {}", _0)]
 	NoSuchAccessKey(String),
 
+	/// The requested worker does not exist
+	#[error(display = "Worker not found: {}", _0)]
+	NoSuchWorker(u64),
+
 	/// In Import key, the key already exists
 	#[error(
 		display = "Key {} already exists in data store. Even if it is deleted, we can't let you create a new key with the same ID. Sorry.",
@@ -53,6 +57,7 @@ impl Error {
 		match self {
 			Error::Common(c) => c.aws_code(),
 			Error::NoSuchAccessKey(_) => "NoSuchAccessKey",
+			Error::NoSuchWorker(_) => "NoSuchWorker",
 			Error::KeyAlreadyExists(_) => "KeyAlreadyExists",
 		}
 	}
@@ -63,7 +68,7 @@ impl ApiError for Error {
 	fn http_status_code(&self) -> StatusCode {
 		match self {
 			Error::Common(c) => c.http_status_code(),
-			Error::NoSuchAccessKey(_) => StatusCode::NOT_FOUND,
+			Error::NoSuchAccessKey(_) | Error::NoSuchWorker(_) => StatusCode::NOT_FOUND,
 			Error::KeyAlreadyExists(_) => StatusCode::CONFLICT,
 		}
 	}

src/api/admin/macros.rs

@@ -111,19 +111,11 @@ macro_rules! local_admin_endpoints {
             }
 
             $(
-                #[derive(Debug, Clone, Serialize, Deserialize)]
-                pub struct [< $endpoint Request >] {
-                    pub node: String,
-                    pub body: [< Local $endpoint Request >],
-                }
+                pub type [< $endpoint Request >] = MultiRequest< [< Local $endpoint Request >] >;
 
                 pub type [< $endpoint RequestBody >] = [< Local $endpoint Request >];
 
-                #[derive(Debug, Clone, Serialize, Deserialize)]
-                pub struct [< $endpoint Response >] {
-                    pub success: HashMap<String, [< Local $endpoint Response >] >,
-                    pub error: HashMap<String, String>,
-                }
+                pub type [< $endpoint Response >] = MultiResponse< [< Local $endpoint Response >] >;
 
                 impl From< [< Local $endpoint Request >] > for LocalAdminApiRequest {
                     fn from(req: [< Local $endpoint Request >]) -> LocalAdminApiRequest {

src/api/admin/router_v2.rs

@@ -60,7 +60,10 @@ impl AdminApiRequest {
 			POST AddBucketAlias (body),
 			POST RemoveBucketAlias (body),
 			// Worker APIs
+			POST ListWorkers (body_field, query::node),
+			POST GetWorkerInfo (body_field, query::node),
 			POST GetWorkerVariable (body_field, query::node),
+			POST SetWorkerVariable (body_field, query::node),
 		]);
 
 		if let Some(message) = query.nonempty_message() {

src/api/admin/worker.rs

@@ -3,12 +3,59 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 
+use garage_util::background::*;
+use garage_util::time::now_msec;
+
 use garage_model::garage::Garage;
 
 use crate::api::*;
 use crate::error::Error;
 use crate::{Admin, RequestHandler};
 
+#[async_trait]
+impl RequestHandler for LocalListWorkersRequest {
+	type Response = LocalListWorkersResponse;
+
+	async fn handle(
+		self,
+		_garage: &Arc<Garage>,
+		admin: &Admin,
+	) -> Result<LocalListWorkersResponse, Error> {
+		let workers = admin.background.get_worker_info();
+		let info = workers
+			.into_iter()
+			.filter(|(_, w)| {
+				(!self.busy_only
+					|| matches!(w.state, WorkerState::Busy | WorkerState::Throttled(_)))
+					&& (!self.error_only || w.errors > 0)
+			})
+			.map(|(id, w)| worker_info_to_api(id as u64, w))
+			.collect::<Vec<_>>();
+		Ok(LocalListWorkersResponse(info))
+	}
+}
+
+#[async_trait]
+impl RequestHandler for LocalGetWorkerInfoRequest {
+	type Response = LocalGetWorkerInfoResponse;
+
+	async fn handle(
+		self,
+		_garage: &Arc<Garage>,
+		admin: &Admin,
+	) -> Result<LocalGetWorkerInfoResponse, Error> {
+		let info = admin
+			.background
+			.get_worker_info()
+			.get(&(self.id as usize))
+			.ok_or(Error::NoSuchWorker(self.id))?
+			.clone();
+		Ok(LocalGetWorkerInfoResponse(worker_info_to_api(
+			self.id, info,
+		)))
+	}
+}
+
 #[async_trait]
 impl RequestHandler for LocalGetWorkerVariableRequest {
 	type Response = LocalGetWorkerVariableResponse;
@@ -48,3 +95,30 @@ impl RequestHandler for LocalSetWorkerVariableRequest {
 		})
 	}
 }
+
+// ---- helper functions ----
+
+fn worker_info_to_api(id: u64, info: WorkerInfo) -> WorkerInfoResp {
+	WorkerInfoResp {
+		id: id,
+		name: info.name,
+		state: match info.state {
+			WorkerState::Busy => WorkerStateResp::Busy,
+			WorkerState::Throttled(t) => WorkerStateResp::Throttled { duration_secs: t },
+			WorkerState::Idle => WorkerStateResp::Idle,
+			WorkerState::Done => WorkerStateResp::Done,
+		},
+		errors: info.errors as u64,
+		consecutive_errors: info.consecutive_errors as u64,
+		last_error: info.last_error.map(|(message, t)| WorkerLastError {
+			message,
+			secs_ago: (std::cmp::max(t, now_msec()) - t) / 1000,
+		}),
+
+		tranquility: info.status.tranquility,
+		progress: info.status.progress,
+		queue_length: info.status.queue_length,
+		persistent_errors: info.status.persistent_errors,
+		freeform: info.status.freeform,
+	}
+}