admin api: base infrastructure for local endpoints

admin api: rename EndpointHandler into RequestHandler to avoid confusion with RPC

wip: infrastructure for local api calls

admin api: fix things

admin api: first local endpoint to work with new scheme

admin api: implement SetWorkerVariable

2 files changed, 104 insertions, 15 deletions

diff --git a/src/garage/cli_v2/mod.rs b/src/garage/cli_v2/mod.rs
index 6cc13b2d..b9bf05fe 100644
--- a/src/garage/cli_v2/mod.rs
+++ b/src/garage/cli_v2/mod.rs
@@ -3,6 +3,8 @@ pub mod cluster;
 pub mod key;
 pub mod layout;
 
+pub mod worker;
+
 use std::convert::TryFrom;
 use std::sync::Arc;
 use std::time::Duration;
@@ -13,7 +15,8 @@ use garage_rpc::system::*;
 use garage_rpc::*;
 
 use garage_api_admin::api::*;
-use garage_api_admin::EndpointHandler as AdminApiEndpoint;
+use garage_api_admin::api_server::{AdminRpc as ProxyRpc, AdminRpcResponse as ProxyRpcResponse};
+use garage_api_admin::RequestHandler as AdminApiEndpoint;
 
 use crate::admin::*;
 use crate::cli as cli_v1;
@@ -23,6 +26,7 @@ use crate::cli::Command;
 pub struct Cli {
 	pub system_rpc_endpoint: Arc<Endpoint<SystemRpc, ()>>,
 	pub admin_rpc_endpoint: Arc<Endpoint<AdminRpc, ()>>,
+	pub proxy_rpc_endpoint: Arc<Endpoint<ProxyRpc, ()>>,
 	pub rpc_host: NodeID,
 }
 
@@ -36,6 +40,7 @@ impl Cli {
 			Command::Layout(layout_opt) => self.layout_command_dispatch(layout_opt).await,
 			Command::Bucket(bo) => self.cmd_bucket(bo).await,
 			Command::Key(ko) => self.cmd_key(ko).await,
+			Command::Worker(wo) => self.cmd_worker(wo).await,
 
 			// TODO
 			Command::Repair(ro) => cli_v1::cmd_admin(
@@ -50,13 +55,6 @@ impl Cli {
 					.await
 					.ok_or_message("cli_v1")
 			}
-			Command::Worker(wo) => cli_v1::cmd_admin(
-				&self.admin_rpc_endpoint,
-				self.rpc_host,
-				AdminRpc::Worker(wo),
-			)
-			.await
-			.ok_or_message("cli_v1"),
 			Command::Block(bo) => cli_v1::cmd_admin(
 				&self.admin_rpc_endpoint,
 				self.rpc_host,
@@ -85,14 +83,16 @@ impl Cli {
 		let req = AdminApiRequest::from(req);
 		let req_name = req.name();
 		match self
-			.admin_rpc_endpoint
-			.call(&self.rpc_host, AdminRpc::ApiRequest(req), PRIO_NORMAL)
-			.await?
-			.ok_or_message("rpc")?
+			.proxy_rpc_endpoint
+			.call(&self.rpc_host, ProxyRpc::Proxy(req), PRIO_NORMAL)
+			.await??
 		{
-			AdminRpc::ApiOkResponse(resp) => <T as AdminApiEndpoint>::Response::try_from(resp)
-				.map_err(|_| Error::Message(format!("{} returned unexpected response", req_name))),
-			AdminRpc::ApiErrorResponse {
+			ProxyRpcResponse::ProxyApiOkResponse(resp) => {
+				<T as AdminApiEndpoint>::Response::try_from(resp).map_err(|_| {
+					Error::Message(format!("{} returned unexpected response", req_name))
+				})
+			}
+			ProxyRpcResponse::ApiErrorResponse {
 				http_code,
 				error_code,
 				message,
diff --git a/src/garage/cli_v2/worker.rs b/src/garage/cli_v2/worker.rs
new file mode 100644
index 00000000..0dfe3e96
--- /dev/null
+++ b/src/garage/cli_v2/worker.rs
@@ -0,0 +1,89 @@
+//use bytesize::ByteSize;
+use format_table::format_table;
+
+use garage_util::error::*;
+
+use garage_api_admin::api::*;
+
+use crate::cli::structs::*;
+use crate::cli_v2::*;
+
+impl Cli {
+	pub async fn cmd_worker(&self, cmd: WorkerOperation) -> Result<(), Error> {
+		match cmd {
+			WorkerOperation::Get {
+				all_nodes,
+				variable,
+			} => self.cmd_get_var(all_nodes, variable).await,
+			WorkerOperation::Set {
+				all_nodes,
+				variable,
+				value,
+			} => self.cmd_set_var(all_nodes, variable, value).await,
+			wo => cli_v1::cmd_admin(
+				&self.admin_rpc_endpoint,
+				self.rpc_host,
+				AdminRpc::Worker(wo),
+			)
+			.await
+			.ok_or_message("cli_v1"),
+		}
+	}
+
+	pub async fn cmd_get_var(&self, all: bool, var: Option<String>) -> Result<(), Error> {
+		let res = self
+			.api_request(GetWorkerVariableRequest {
+				node: if all {
+					"*".to_string()
+				} else {
+					hex::encode(self.rpc_host)
+				},
+				body: LocalGetWorkerVariableRequest { variable: var },
+			})
+			.await?;
+
+		let mut table = vec![];
+		for (node, vars) in res.success.iter() {
+			for (key, val) in vars.0.iter() {
+				table.push(format!("{:.16}\t{}\t{}", node, key, val));
+			}
+		}
+		format_table(table);
+
+		for (node, err) in res.error.iter() {
+			eprintln!("{:.16}: error: {}", node, err);
+		}
+
+		Ok(())
+	}
+
+	pub async fn cmd_set_var(
+		&self,
+		all: bool,
+		variable: String,
+		value: String,
+	) -> Result<(), Error> {
+		let res = self
+			.api_request(SetWorkerVariableRequest {
+				node: if all {
+					"*".to_string()
+				} else {
+					hex::encode(self.rpc_host)
+				},
+				body: LocalSetWorkerVariableRequest { variable, value },
+			})
+			.await?;
+
+		let mut table = vec![];
+		for (node, kv) in res.success.iter() {
+			table.push(format!("{:.16}\t{}\t{}", node, kv.variable, kv.value));
+		}
+		format_table(table);
+
+		for (node, err) in res.error.iter() {
+			eprintln!("{:.16}: error: {}", node, err);
+		}
+
+		Ok(())
+	}
+}