Merge pull request 'feature: Register consul services with agent API' (#567) from unrob/garage:roberto/consul-agent-registration into main

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/567
Reviewed-by: Alex <alex@adnab.me>

2 files changed, 132 insertions, 62 deletions

diff --git a/src/rpc/consul.rs b/src/rpc/consul.rs
index f85f789c..ab8d1112 100644
--- a/src/rpc/consul.rs
+++ b/src/rpc/consul.rs
@@ -8,16 +8,26 @@ use serde::{Deserialize, Serialize};
 
 use netapp::NodeID;
 
+use garage_util::config::ConsulDiscoveryAPI;
 use garage_util::config::ConsulDiscoveryConfig;
 
+const META_PREFIX: &str = "fr-deuxfleurs-garagehq";
+
 #[derive(Deserialize, Clone, Debug)]
 struct ConsulQueryEntry {
 	#[serde(rename = "Address")]
 	address: String,
 	#[serde(rename = "ServicePort")]
 	service_port: u16,
-	#[serde(rename = "NodeMeta")]
-	node_meta: HashMap<String, String>,
+	#[serde(rename = "ServiceMeta")]
+	meta: HashMap<String, String>,
+}
+
+#[derive(Serialize, Clone, Debug)]
+#[serde(untagged)]
+enum PublishRequest {
+	Catalog(ConsulPublishEntry),
+	Service(ConsulPublishService),
 }
 
 #[derive(Serialize, Clone, Debug)]
@@ -26,28 +36,43 @@ struct ConsulPublishEntry {
 	node: String,
 	#[serde(rename = "Address")]
 	address: IpAddr,
-	#[serde(rename = "NodeMeta")]
-	node_meta: HashMap<String, String>,
 	#[serde(rename = "Service")]
-	service: ConsulPublishService,
+	service: ConsulPublishCatalogService,
 }
 
 #[derive(Serialize, Clone, Debug)]
-struct ConsulPublishService {
+struct ConsulPublishCatalogService {
 	#[serde(rename = "ID")]
 	service_id: String,
 	#[serde(rename = "Service")]
 	service_name: String,
 	#[serde(rename = "Tags")]
 	tags: Vec<String>,
+	#[serde(rename = "Meta")]
+	meta: HashMap<String, String>,
 	#[serde(rename = "Address")]
 	address: IpAddr,
 	#[serde(rename = "Port")]
 	port: u16,
 }
 
-// ----
+#[derive(Serialize, Clone, Debug)]
+struct ConsulPublishService {
+	#[serde(rename = "ID")]
+	service_id: String,
+	#[serde(rename = "Name")]
+	service_name: String,
+	#[serde(rename = "Tags")]
+	tags: Vec<String>,
+	#[serde(rename = "Address")]
+	address: IpAddr,
+	#[serde(rename = "Port")]
+	port: u16,
+	#[serde(rename = "Meta")]
+	meta: HashMap<String, String>,
+}
 
+// ----
 pub struct ConsulDiscovery {
 	config: ConsulDiscoveryConfig,
 	client: reqwest::Client,
@@ -55,44 +80,48 @@ pub struct ConsulDiscovery {
 
 impl ConsulDiscovery {
 	pub fn new(config: ConsulDiscoveryConfig) -> Result<Self, ConsulError> {
-		let client = match (&config.client_cert, &config.client_key) {
-			(Some(client_cert), Some(client_key)) => {
-				let mut client_cert_buf = vec![];
-				File::open(client_cert)?.read_to_end(&mut client_cert_buf)?;
-
-				let mut client_key_buf = vec![];
-				File::open(client_key)?.read_to_end(&mut client_key_buf)?;
-
-				let identity = reqwest::Identity::from_pem(
-					&[&client_cert_buf[..], &client_key_buf[..]].concat()[..],
-				)?;
-
-				if config.tls_skip_verify {
-					reqwest::Client::builder()
-						.use_rustls_tls()
-						.danger_accept_invalid_certs(true)
-						.identity(identity)
-						.build()?
-				} else if let Some(ca_cert) = &config.ca_cert {
-					let mut ca_cert_buf = vec![];
-					File::open(ca_cert)?.read_to_end(&mut ca_cert_buf)?;
-
-					reqwest::Client::builder()
-						.use_rustls_tls()
-						.add_root_certificate(reqwest::Certificate::from_pem(&ca_cert_buf[..])?)
-						.identity(identity)
-						.build()?
-				} else {
-					reqwest::Client::builder()
-						.use_rustls_tls()
-						.identity(identity)
-						.build()?
+		let mut builder: reqwest::ClientBuilder = reqwest::Client::builder().use_rustls_tls();
+		if config.tls_skip_verify {
+			builder = builder.danger_accept_invalid_certs(true);
+		} else if let Some(ca_cert) = &config.ca_cert {
+			let mut ca_cert_buf = vec![];
+			File::open(ca_cert)?.read_to_end(&mut ca_cert_buf)?;
+			builder =
+				builder.add_root_certificate(reqwest::Certificate::from_pem(&ca_cert_buf[..])?);
+		}
+
+		match &config.api {
+			ConsulDiscoveryAPI::Catalog => match (&config.client_cert, &config.client_key) {
+				(Some(client_cert), Some(client_key)) => {
+					let mut client_cert_buf = vec![];
+					File::open(client_cert)?.read_to_end(&mut client_cert_buf)?;
+
+					let mut client_key_buf = vec![];
+					File::open(client_key)?.read_to_end(&mut client_key_buf)?;
+
+					let identity = reqwest::Identity::from_pem(
+						&[&client_cert_buf[..], &client_key_buf[..]].concat()[..],
+					)?;
+
+					builder = builder.identity(identity);
+				}
+				(None, None) => {}
+				_ => return Err(ConsulError::InvalidTLSConfig),
+			},
+			ConsulDiscoveryAPI::Agent => {
+				if let Some(token) = &config.token {
+					let mut headers = reqwest::header::HeaderMap::new();
+					headers.insert(
+						"x-consul-token",
+						reqwest::header::HeaderValue::from_str(&token)?,
+					);
+					builder = builder.default_headers(headers);
 				}
 			}
-			(None, None) => reqwest::Client::new(),
-			_ => return Err(ConsulError::InvalidTLSConfig),
 		};
 
+		let client: reqwest::Client = builder.build()?;
+
 		Ok(Self { client, config })
 	}
 
@@ -111,8 +140,8 @@ impl ConsulDiscovery {
 		for ent in entries {
 			let ip = ent.address.parse::<IpAddr>().ok();
 			let pubkey = ent
-				.node_meta
-				.get("pubkey")
+				.meta
+				.get(&format!("{}-pubkey", META_PREFIX))
 				.and_then(|k| hex::decode(k).ok())
 				.and_then(|k| NodeID::from_slice(&k[..]));
 			if let (Some(ip), Some(pubkey)) = (ip, pubkey) {
@@ -138,29 +167,49 @@ impl ConsulDiscovery {
 		rpc_public_addr: SocketAddr,
 	) -> Result<(), ConsulError> {
 		let node = format!("garage:{}", hex::encode(&node_id[..8]));
+		let tags = [
+			vec!["advertised-by-garage".into(), hostname.into()],
+			self.config.tags.clone(),
+		]
+		.concat();
 
-		let advertisement = ConsulPublishEntry {
-			node: node.clone(),
-			address: rpc_public_addr.ip(),
-			node_meta: [
-				("pubkey".to_string(), hex::encode(node_id)),
-				("hostname".to_string(), hostname.to_string()),
-			]
-			.iter()
-			.cloned()
-			.collect(),
-			service: ConsulPublishService {
+		let mut meta = self.config.meta.clone().unwrap_or_default();
+		meta.insert(format!("{}-pubkey", META_PREFIX), hex::encode(node_id));
+		meta.insert(format!("{}-hostname", META_PREFIX), hostname.to_string());
+
+		let url = format!(
+			"{}/v1/{}",
+			self.config.consul_http_addr,
+			(match &self.config.api {
+				ConsulDiscoveryAPI::Catalog => "catalog/register",
+				ConsulDiscoveryAPI::Agent => "agent/service/register?replace-existing-checks",
+			})
+		);
+
+		let req = self.client.put(&url);
+		let advertisement: PublishRequest = match &self.config.api {
+			ConsulDiscoveryAPI::Catalog => PublishRequest::Catalog(ConsulPublishEntry {
+				node: node.clone(),
+				address: rpc_public_addr.ip(),
+				service: ConsulPublishCatalogService {
+					service_id: node.clone(),
+					service_name: self.config.service_name.clone(),
+					tags,
+					meta: meta.clone(),
+					address: rpc_public_addr.ip(),
+					port: rpc_public_addr.port(),
+				},
+			}),
+			ConsulDiscoveryAPI::Agent => PublishRequest::Service(ConsulPublishService {
 				service_id: node.clone(),
 				service_name: self.config.service_name.clone(),
-				tags: vec!["advertised-by-garage".into(), hostname.into()],
+				tags,
+				meta,
 				address: rpc_public_addr.ip(),
 				port: rpc_public_addr.port(),
-			},
+			}),
 		};
-
-		let url = format!("{}/v1/catalog/register", self.config.consul_http_addr);
-
-		let http = self.client.put(&url).json(&advertisement).send().await?;
+		let http = req.json(&advertisement).send().await?;
 		http.error_for_status()?;
 
 		Ok(())
@@ -176,4 +225,6 @@ pub enum ConsulError {
 	Reqwest(#[error(source)] reqwest::Error),
 	#[error(display = "Invalid Consul TLS configuration")]
 	InvalidTLSConfig,
+	#[error(display = "Token error: {}", _0)]
+	Token(#[error(source)] reqwest::header::InvalidHeaderValue),
 }
diff --git a/src/util/config.rs b/src/util/config.rs
index 95835bbb..1da95b2f 100644
--- a/src/util/config.rs
+++ b/src/util/config.rs
@@ -135,8 +135,19 @@ pub struct AdminConfig {
 	pub trace_sink: Option<String>,
 }
 
+#[derive(Deserialize, Debug, Clone, Default)]
+#[serde(rename_all = "lowercase")]
+pub enum ConsulDiscoveryAPI {
+	#[default]
+	Catalog,
+	Agent,
+}
+
 #[derive(Deserialize, Debug, Clone)]
 pub struct ConsulDiscoveryConfig {
+	/// The consul api to use when registering: either `catalog` (the default) or `agent`
+	#[serde(default)]
+	pub api: ConsulDiscoveryAPI,
 	/// Consul http or https address to connect to to discover more peers
 	pub consul_http_addr: String,
 	/// Consul service name to use
@@ -147,9 +158,17 @@ pub struct ConsulDiscoveryConfig {
 	pub client_cert: Option<String>,
 	/// Client TLS key to use when connecting to Consul
 	pub client_key: Option<String>,
+	/// /// Token to use for connecting to consul
+	pub token: Option<String>,
 	/// Skip TLS hostname verification
 	#[serde(default)]
 	pub tls_skip_verify: bool,
+	/// Additional tags to add to the service
+	#[serde(default)]
+	pub tags: Vec<String>,
+	/// Additional service metadata to add
+	#[serde(default)]
+	pub meta: Option<std::collections::HashMap<String, String>>,
 }
 
 #[derive(Deserialize, Debug, Clone)]
@@ -344,7 +363,7 @@ mod tests {
 			replication_mode = "3"
 			rpc_bind_addr = "[::]:3901"
 			rpc_secret_file = "{}"
-		
+
 			[s3_api]
 			s3_region = "garage"
 			api_bind_addr = "[::]:3900"
@@ -388,7 +407,7 @@ mod tests {
 			rpc_bind_addr = "[::]:3901"
 			rpc_secret= "dummy"
 			rpc_secret_file = "dummy"
-		
+
 			[s3_api]
 			s3_region = "garage"
 			api_bind_addr = "[::]:3900"