register consul services against local agent instead of catalog api

4 files changed, 235 insertions, 3 deletions

diff --git a/src/rpc/Cargo.toml b/src/rpc/Cargo.toml
index f0fde7a7..d168e246 100644
--- a/src/rpc/Cargo.toml
+++ b/src/rpc/Cargo.toml
@@ -50,4 +50,5 @@ netapp = { version = "0.5.2", features = ["telemetry"] }
 [features]
 kubernetes-discovery = [ "kube", "k8s-openapi", "schemars" ]
 consul-discovery = [ "reqwest", "err-derive" ]
+consul-service-discovery = [ "reqwest", "err-derive" ]
 system-libs = [ "sodiumoxide/use-pkg-config" ]
diff --git a/src/rpc/consul_services.rs b/src/rpc/consul_services.rs
new file mode 100644
index 00000000..461bb195
--- /dev/null
+++ b/src/rpc/consul_services.rs
@@ -0,0 +1,166 @@
+use std::collections::HashMap;
+use std::fs::File;
+use std::io::Read;
+use std::net::{IpAddr, SocketAddr};
+
+use err_derive::Error;
+use serde::{Deserialize, Serialize};
+
+use netapp::NodeID;
+
+use garage_util::config::ConsulServiceConfig;
+
+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 = "ServiceMeta")]
+	service_meta: HashMap<String, String>,
+}
+
+#[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 ConsulServiceDiscovery {
+	config: ConsulServiceConfig,
+	client: reqwest::Client,
+}
+
+impl ConsulServiceDiscovery {
+	pub fn new(config: ConsulServiceConfig) -> Result<Self, ConsulError> {
+		let mut builder: reqwest::ClientBuilder = match &config.ca_cert {
+			Some(client_ca) => {
+				let mut ca_cert_buf = vec![];
+				File::open(client_ca)?.read_to_end(&mut ca_cert_buf)?;
+
+				let req: reqwest::ClientBuilder = reqwest::Client::builder()
+					.add_root_certificate(reqwest::Certificate::from_pem(&ca_cert_buf[..])?)
+					.use_rustls_tls();
+
+				if config.tls_skip_verify {
+					req.danger_accept_invalid_certs(true)
+				} else {
+					req
+				}
+			}
+			None => reqwest::Client::builder(),
+		};
+
+		if let Some(token) = &config.consul_http_token {
+			let mut headers = reqwest::header::HeaderMap::new();
+			headers.insert("x-consul-token", reqwest::header::HeaderValue::from_str(&token)?);
+			builder = builder.default_headers(headers);
+		}
+
+		let client = builder.build()?;
+
+		Ok(Self { client, config })
+	}
+
+	// ---- READING FROM CONSUL CATALOG ----
+
+	pub async fn get_consul_services(&self) -> Result<Vec<(NodeID, SocketAddr)>, ConsulError> {
+		let url = format!(
+			"{}/v1/catalog/service/{}",
+			self.config.consul_http_addr, self.config.service_name
+		);
+
+		let req = self.client.get(&url);
+		let http = req.send().await?;
+		let entries: Vec<ConsulQueryEntry> = http.json().await?;
+
+		let mut ret = vec![];
+		for ent in entries {
+			let ip = ent.address.parse::<IpAddr>().ok();
+			let pubkey = ent
+				.service_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) {
+				ret.push((pubkey, SocketAddr::new(ip, ent.service_port)));
+			} else {
+				warn!(
+					"Could not process node spec from Consul: {:?} (invalid IP or public key)",
+					ent
+				);
+			}
+		}
+		debug!("Got nodes from Consul: {:?}", ret);
+
+		Ok(ret)
+	}
+
+	// ---- PUBLISHING TO CONSUL CATALOG ----
+
+	pub async fn publish_consul_service(
+		&self,
+		node_id: NodeID,
+		hostname: &str,
+		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: ConsulPublishService = ConsulPublishService {
+			service_id: node.clone(),
+			service_name: self.config.service_name.clone(),
+			tags,
+			meta: [
+				(format!("{}-pubkey", META_PREFIX), hex::encode(node_id)),
+				(format!("{}-hostname", META_PREFIX), hostname.to_string()),
+			]
+			.iter()
+			.cloned()
+			.collect(),
+			address: rpc_public_addr.ip(),
+			port: rpc_public_addr.port(),
+		};
+
+		let url = format!(
+			"{}/v1/agent/service/register?replace-existing-checks",
+			self.config.consul_http_addr
+		);
+
+		let req = self.client.put(&url);
+		let http = req.json(&advertisement).send().await?;
+		http.error_for_status()?;
+
+		Ok(())
+	}
+}
+
+/// Regroup all Consul discovery errors
+#[derive(Debug, Error)]
+pub enum ConsulError {
+	#[error(display = "IO error: {}", _0)]
+	Io(#[error(source)] std::io::Error),
+	#[error(display = "HTTP error: {}", _0)]
+	Reqwest(#[error(source)] reqwest::Error),
+	#[error(display = "Invalid HTTP header error: {}", _0)]
+	HeaderValue(#[error(source)] reqwest::header::InvalidHeaderValue),
+}
diff --git a/src/rpc/lib.rs b/src/rpc/lib.rs
index 5aec92c0..2c5ed107 100644
--- a/src/rpc/lib.rs
+++ b/src/rpc/lib.rs
@@ -8,6 +8,8 @@ mod system_metrics;
 
 #[cfg(feature = "consul-discovery")]
 mod consul;
+#[cfg(feature = "consul-service-discovery")]
+mod consul_services;
 #[cfg(feature = "kubernetes-discovery")]
 mod kubernetes;
 
diff --git a/src/rpc/system.rs b/src/rpc/system.rs
index b42e49fc..b0fef6f2 100644
--- a/src/rpc/system.rs
+++ b/src/rpc/system.rs
@@ -32,6 +32,8 @@ use garage_util::time::*;
 
 #[cfg(feature = "consul-discovery")]
 use crate::consul::ConsulDiscovery;
+#[cfg(feature = "consul-service-discovery")]
+use crate::consul_services::ConsulServiceDiscovery;
 #[cfg(feature = "kubernetes-discovery")]
 use crate::kubernetes::*;
 use crate::layout::*;
@@ -98,12 +100,14 @@ pub struct System {
 	system_endpoint: Arc<Endpoint<SystemRpc, System>>,
 
 	rpc_listen_addr: SocketAddr,
-	#[cfg(any(feature = "consul-discovery", feature = "kubernetes-discovery"))]
+	#[cfg(any(feature = "consul-discovery", feature = "consul-service-discovery", feature = "kubernetes-discovery"))]
 	rpc_public_addr: Option<SocketAddr>,
 	bootstrap_peers: Vec<String>,
 
 	#[cfg(feature = "consul-discovery")]
 	consul_discovery: Option<ConsulDiscovery>,
+	#[cfg(feature = "consul-service-discovery")]
+	consul_service_discovery: Option<ConsulServiceDiscovery>,
 	#[cfg(feature = "kubernetes-discovery")]
 	kubernetes_discovery: Option<KubernetesDiscoveryConfig>,
 
@@ -346,6 +350,19 @@ impl System {
 			warn!("Consul discovery is not enabled in this build.");
 		}
 
+		#[cfg(feature = "consul-service-discovery")]
+		let consul_service_discovery = match &config.consul_service_discovery {
+			Some(cfg) => Some(
+				ConsulServiceDiscovery::new(cfg.clone())
+					.ok_or_message("Invalid Consul service discovery configuration")?,
+			),
+			None => None,
+		};
+		#[cfg(not(feature = "consul-service-discovery"))]
+		if config.consul_service_discovery.is_some() {
+			warn!("Consul service discovery is not enabled in this build.");
+		}
+
 		#[cfg(not(feature = "kubernetes-discovery"))]
 		if config.kubernetes_discovery.is_some() {
 			warn!("Kubernetes discovery is not enabled in this build.");
@@ -369,11 +386,13 @@ impl System {
 			replication_mode,
 			replication_factor,
 			rpc_listen_addr: config.rpc_bind_addr,
-			#[cfg(any(feature = "consul-discovery", feature = "kubernetes-discovery"))]
+			#[cfg(any(feature = "consul-discovery", feature = "consul-service-discovery", feature = "kubernetes-discovery"))]
 			rpc_public_addr,
 			bootstrap_peers: config.bootstrap_peers.clone(),
 			#[cfg(feature = "consul-discovery")]
 			consul_discovery,
+			#[cfg(feature = "consul-service-discovery")]
+			consul_service_discovery,
 			#[cfg(feature = "kubernetes-discovery")]
 			kubernetes_discovery: config.kubernetes_discovery.clone(),
 			metrics,
@@ -555,6 +574,34 @@ impl System {
 		}
 	}
 
+	#[cfg(feature = "consul-service-discovery")]
+	async fn advertise_to_consul(self: Arc<Self>) {
+		let c = match &self.consul_service_discovery {
+			Some(c) => c,
+			_ => return,
+		};
+
+		let rpc_public_addr = match self.rpc_public_addr {
+			Some(addr) => addr,
+			None => {
+				warn!("Not advertising to Consul because rpc_public_addr is not defined in config file and could not be autodetected.");
+				return;
+			}
+		};
+
+		if let Err(e) = c
+			.publish_consul_service(
+				self.netapp.id,
+				&self.local_status.load_full().hostname,
+				rpc_public_addr,
+			)
+			.await
+		{
+			error!("Error while publishing Consul service: {}", e);
+		}
+	}
+
+
 	#[cfg(feature = "kubernetes-discovery")]
 	async fn advertise_to_kubernetes(self: Arc<Self>) {
 		let k = match &self.kubernetes_discovery {
@@ -744,7 +791,7 @@ impl System {
 					ping_list.extend(peers.0.iter().map(|(id, addr)| ((*id).into(), *addr)))
 				}
 
-				// Fetch peer list from Consul
+				// Fetch peer list from Consul Nodes
 				#[cfg(feature = "consul-discovery")]
 				if let Some(c) = &self.consul_discovery {
 					match c.get_consul_nodes().await {
@@ -757,6 +804,19 @@ impl System {
 					}
 				}
 
+				// Fetch peer list from Consul Services
+				#[cfg(feature = "consul-service-discovery")]
+				if let Some(c) = &self.consul_service_discovery {
+					match c.get_consul_services().await {
+						Ok(node_list) => {
+							ping_list.extend(node_list);
+						}
+						Err(e) => {
+							warn!("Could not retrieve service list from Consul: {}", e);
+						}
+					}
+				}
+
 				// Fetch peer list from Kubernetes
 				#[cfg(feature = "kubernetes-discovery")]
 				if let Some(k) = &self.kubernetes_discovery {
@@ -796,6 +856,9 @@ impl System {
 			#[cfg(feature = "consul-discovery")]
 			tokio::spawn(self.clone().advertise_to_consul());
 
+			#[cfg(feature = "consul-service-discovery")]
+			tokio::spawn(self.clone().advertise_to_consul());
+
 			#[cfg(feature = "kubernetes-discovery")]
 			tokio::spawn(self.clone().advertise_to_kubernetes());