Add TLS support for Consul discovery + refactoring

src/rpc/Cargo.toml

@@ -29,11 +29,13 @@ rmp-serde = "0.15"
 serde = { version = "1.0", default-features = false, features = ["derive", "rc"] }
 serde_bytes = "0.11"
 serde_json = "1.0"
+err-derive = { version = "0.3", optional = true }
 
 # newer version requires rust edition 2021
 kube = { version = "0.75", default-features = false, features = ["runtime", "derive", "client", "rustls-tls"], optional = true }
 k8s-openapi = { version = "0.16", features = ["v1_22"], optional = true }
 schemars = { version = "0.8", optional = true }
+reqwest = { version = "0.11", optional = true, default-features = false, features = ["rustls-tls", "json"] }
 
 # newer version requires rust edition 2021
 pnet_datalink = "0.28"
@@ -46,9 +48,7 @@ opentelemetry = "0.17"
 
 netapp = { version = "0.5.2", features = ["telemetry"] }
 
-hyper = { version = "0.14", features = ["client", "http1", "runtime", "tcp"] }
-
-
 [features]
 kubernetes-discovery = [ "kube", "k8s-openapi", "schemars" ]
+consul-discovery = [ "reqwest", "err-derive" ]
 system-libs = [ "sodiumoxide/use-pkg-config" ]

src/rpc/consul.rs

@@ -1,14 +1,66 @@
 use std::collections::HashMap;
 use std::net::{IpAddr, SocketAddr};
 
-use hyper::client::Client;
-use hyper::StatusCode;
-use hyper::{Body, Method, Request};
+use tokio::fs::File;
+use tokio::io::AsyncReadExt;
+
+use err_derive::Error;
 use serde::{Deserialize, Serialize};
 
 use netapp::NodeID;
 
-use garage_util::error::Error;
+use garage_util::config::ConsulDiscoveryConfig;
+
+async fn make_consul_client(
+	config: &ConsulDiscoveryConfig,
+) -> Result<reqwest::Client, ConsulError> {
+	match (&config.client_cert, &config.client_key) {
+		(Some(client_cert), Some(client_key)) => {
+			let mut client_cert_buf = vec![];
+			File::open(client_cert)
+				.await?
+				.read_to_end(&mut client_cert_buf)
+				.await?;
+
+			let mut client_key_buf = vec![];
+			File::open(client_key)
+				.await?
+				.read_to_end(&mut client_key_buf)
+				.await?;
+
+			let identity = reqwest::Identity::from_pem(
+				&[&client_cert_buf[..], &client_key_buf[..]].concat()[..],
+			)?;
+
+			if config.tls_skip_verify {
+				Ok(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)
+					.await?
+					.read_to_end(&mut ca_cert_buf)
+					.await?;
+
+				Ok(reqwest::Client::builder()
+					.use_rustls_tls()
+					.add_root_certificate(reqwest::Certificate::from_pem(&ca_cert_buf[..])?)
+					.identity(identity)
+					.build()?)
+			} else {
+				Ok(reqwest::Client::builder()
+					.use_rustls_tls()
+					.identity(identity)
+					.build()?)
+			}
+		}
+		(None, None) => Ok(reqwest::Client::new()),
+		_ => Err(ConsulError::InvalidTLSConfig),
+	}
+}
 
 // ---- READING FROM CONSUL CATALOG ----
 
@@ -23,27 +75,16 @@ struct ConsulQueryEntry {
 }
 
 pub async fn get_consul_nodes(
-	consul_host: &str,
-	consul_service_name: &str,
-) -> Result<Vec<(NodeID, SocketAddr)>, Error> {
+	consul_config: &ConsulDiscoveryConfig,
+) -> Result<Vec<(NodeID, SocketAddr)>, ConsulError> {
 	let url = format!(
 		"http://{}/v1/catalog/service/{}",
-		consul_host, consul_service_name
+		consul_config.consul_host, consul_config.service_name
 	);
-	let req = Request::builder()
-		.uri(url)
-		.method(Method::GET)
-		.body(Body::default())?;
 
-	let client = Client::new();
-
-	let resp = client.request(req).await?;
-	if resp.status() != StatusCode::OK {
-		return Err(Error::Message(format!("HTTP error {}", resp.status())));
-	}
-
-	let body = hyper::body::to_bytes(resp.into_body()).await?;
-	let entries = serde_json::from_slice::<Vec<ConsulQueryEntry>>(body.as_ref())?;
+	let client = make_consul_client(consul_config).await?;
+	let http = client.get(&url).send().await?;
+	let entries: Vec<ConsulQueryEntry> = http.json().await?;
 
 	let mut ret = vec![];
 	for ent in entries {
@@ -96,15 +137,14 @@ struct ConsulPublishService {
 }
 
 pub async fn publish_consul_service(
-	consul_host: &str,
-	consul_service_name: &str,
+	consul_config: &ConsulDiscoveryConfig,
 	node_id: NodeID,
 	hostname: &str,
 	rpc_public_addr: SocketAddr,
-) -> Result<(), Error> {
+) -> Result<(), ConsulError> {
 	let node = format!("garage:{}", hex::encode(&node_id[..8]));
 
-	let advertisment = ConsulPublishEntry {
+	let advertisement = ConsulPublishEntry {
 		node: node.clone(),
 		address: rpc_public_addr.ip(),
 		node_meta: [
@@ -116,36 +156,29 @@ pub async fn publish_consul_service(
 		.collect(),
 		service: ConsulPublishService {
 			service_id: node.clone(),
-			service_name: consul_service_name.to_string(),
+			service_name: consul_config.service_name.clone(),
 			tags: vec!["advertised-by-garage".into(), hostname.into()],
 			address: rpc_public_addr.ip(),
 			port: rpc_public_addr.port(),
 		},
 	};
 
-	let url = format!("http://{}/v1/catalog/register", consul_host);
-	let req_body = serde_json::to_string(&advertisment)?;
-	debug!("Request body for consul adv: {}", req_body);
+	let url = format!("http://{}/v1/catalog/register", consul_config.consul_host);
 
-	let req = Request::builder()
-		.uri(url)
-		.method(Method::PUT)
-		.body(Body::from(req_body))?;
-
-	let client = Client::new();
-
-	let resp = client.request(req).await?;
-	debug!("Response of advertising to Consul: {:?}", resp);
-	let resp_code = resp.status();
-	let resp_bytes = &hyper::body::to_bytes(resp.into_body()).await?;
-	debug!(
-		"{}",
-		std::str::from_utf8(resp_bytes).unwrap_or("<invalid utf8>")
-	);
-
-	if resp_code != StatusCode::OK {
-		return Err(Error::Message(format!("HTTP error {}", resp_code)));
-	}
+	let client = make_consul_client(consul_config).await?;
+	let http = client.put(&url).json(&advertisement).send().await?;
+	http.error_for_status()?;
 
 	Ok(())
 }
+
+/// Regroup all Garage 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 Consul TLS configuration")]
+	InvalidTLSConfig,
+}

src/rpc/kubernetes.rs

@@ -12,6 +12,8 @@ use serde::{Deserialize, Serialize};
 
 use netapp::NodeID;
 
+use garage_util::config::KubernetesDiscoveryConfig;
+
 static K8S_GROUP: &str = "deuxfleurs.fr";
 
 #[derive(CustomResource, Debug, Serialize, Deserialize, Clone, JsonSchema)]
@@ -41,15 +43,14 @@ pub async fn create_kubernetes_crd() -> Result<(), kube::Error> {
 }
 
 pub async fn get_kubernetes_nodes(
-	kubernetes_service_name: &str,
-	kubernetes_namespace: &str,
+	kubernetes_config: &KubernetesDiscoveryConfig,
 ) -> Result<Vec<(NodeID, SocketAddr)>, kube::Error> {
 	let client = Client::try_default().await?;
-	let nodes: Api<GarageNode> = Api::namespaced(client.clone(), kubernetes_namespace);
+	let nodes: Api<GarageNode> = Api::namespaced(client.clone(), &kubernetes_config.namespace);
 
 	let lp = ListParams::default().labels(&format!(
 		"garage.{}/service={}",
-		K8S_GROUP, kubernetes_service_name
+		K8S_GROUP, kubernetes_config.service_name
 	));
 
 	let nodes = nodes.list(&lp).await?;
@@ -73,8 +74,7 @@ pub async fn get_kubernetes_nodes(
 }
 
 pub async fn publish_kubernetes_node(
-	kubernetes_service_name: &str,
-	kubernetes_namespace: &str,
+	kubernetes_config: &KubernetesDiscoveryConfig,
 	node_id: NodeID,
 	hostname: &str,
 	rpc_public_addr: SocketAddr,
@@ -93,13 +93,13 @@ pub async fn publish_kubernetes_node(
 	let labels = node.metadata.labels.insert(BTreeMap::new());
 	labels.insert(
 		format!("garage.{}/service", K8S_GROUP),
-		kubernetes_service_name.to_string(),
+		kubernetes_config.service_name.to_string(),
 	);
 
 	debug!("Node object to be applied: {:#?}", node);
 
 	let client = Client::try_default().await?;
-	let nodes: Api<GarageNode> = Api::namespaced(client.clone(), kubernetes_namespace);
+	let nodes: Api<GarageNode> = Api::namespaced(client.clone(), &kubernetes_config.namespace);
 
 	if let Ok(old_node) = nodes.get(&node_pubkey).await {
 		node.metadata.resource_version = old_node.metadata.resource_version;

src/rpc/lib.rs

@@ -3,6 +3,7 @@
 #[macro_use]
 extern crate tracing;
 
+#[cfg(feature = "consul-discovery")]
 mod consul;
 #[cfg(feature = "kubernetes-discovery")]
 mod kubernetes;

src/rpc/system.rs

@@ -23,12 +23,17 @@ use netapp::{NetApp, NetworkKey, NodeID, NodeKey};
 
 use garage_util::background::BackgroundRunner;
 use garage_util::config::Config;
+#[cfg(feature = "consul-discovery")]
+use garage_util::config::ConsulDiscoveryConfig;
+#[cfg(feature = "kubernetes-discovery")]
+use garage_util::config::KubernetesDiscoveryConfig;
 use garage_util::data::*;
 use garage_util::error::*;
 use garage_util::persister::Persister;
 use garage_util::time::*;
 
-use crate::consul::*;
+#[cfg(feature = "consul-discovery")]
+use crate::consul::{get_consul_nodes, publish_consul_service};
 #[cfg(feature = "kubernetes-discovery")]
 use crate::kubernetes::*;
 use crate::layout::*;
@@ -90,12 +95,14 @@ pub struct System {
 	system_endpoint: Arc<Endpoint<SystemRpc, System>>,
 
 	rpc_listen_addr: SocketAddr,
+	#[cfg(any(feature = "consul-discovery", feature = "kubernetes-discovery"))]
 	rpc_public_addr: Option<SocketAddr>,
 	bootstrap_peers: Vec<String>,
 
-	consul_discovery: Option<ConsulDiscoveryParam>,
+	#[cfg(feature = "consul-discovery")]
+	consul_discovery: Option<ConsulDiscoveryConfig>,
 	#[cfg(feature = "kubernetes-discovery")]
-	kubernetes_discovery: Option<KubernetesDiscoveryParam>,
+	kubernetes_discovery: Option<KubernetesDiscoveryConfig>,
 
 	replication_factor: usize,
 
@@ -285,29 +292,13 @@ impl System {
 
 		let system_endpoint = netapp.endpoint(SYSTEM_RPC_PATH.into());
 
-		let consul_discovery = match (&config.consul_host, &config.consul_service_name) {
-			(Some(ch), Some(csn)) => Some(ConsulDiscoveryParam {
-				consul_host: ch.to_string(),
-				service_name: csn.to_string(),
-			}),
-			_ => None,
-		};
-
-		#[cfg(feature = "kubernetes-discovery")]
-		let kubernetes_discovery = match (
-			&config.kubernetes_service_name,
-			&config.kubernetes_namespace,
-		) {
-			(Some(ksn), Some(kn)) => Some(KubernetesDiscoveryParam {
-				service_name: ksn.to_string(),
-				namespace: kn.to_string(),
-				skip_crd: config.kubernetes_skip_crd,
-			}),
-			_ => None,
-		};
+		#[cfg(not(feature = "consul-discovery"))]
+		if config.consul_discovery.is_some() {
+			warn!("Consul discovery is not enabled in this build.");
+		}
 
 		#[cfg(not(feature = "kubernetes-discovery"))]
-		if config.kubernetes_service_name.is_some() || config.kubernetes_namespace.is_some() {
+		if config.kubernetes_discovery.is_some() {
 			warn!("Kubernetes discovery is not enabled in this build.");
 		}
 
@@ -329,11 +320,13 @@ impl System {
 			system_endpoint,
 			replication_factor,
 			rpc_listen_addr: config.rpc_bind_addr,
+			#[cfg(any(feature = "consul-discovery", feature = "kubernetes-discovery"))]
 			rpc_public_addr,
 			bootstrap_peers: config.bootstrap_peers.clone(),
-			consul_discovery,
+			#[cfg(feature = "consul-discovery")]
+			consul_discovery: config.consul_discovery.clone(),
 			#[cfg(feature = "kubernetes-discovery")]
-			kubernetes_discovery,
+			kubernetes_discovery: config.kubernetes_discovery.clone(),
 
 			ring,
 			update_ring: Mutex::new(update_ring),
@@ -432,6 +425,7 @@ impl System {
 
 	// ---- INTERNALS ----
 
+	#[cfg(feature = "consul-discovery")]
 	async fn advertise_to_consul(self: Arc<Self>) -> Result<(), Error> {
 		let c = match &self.consul_discovery {
 			Some(c) => c,
@@ -447,8 +441,7 @@ impl System {
 		};
 
 		publish_consul_service(
-			&c.consul_host,
-			&c.service_name,
+			c,
 			self.netapp.id,
 			&self.local_status.load_full().hostname,
 			rpc_public_addr,
@@ -473,8 +466,7 @@ impl System {
 		};
 
 		publish_kubernetes_node(
-			&k.service_name,
-			&k.namespace,
+			k,
 			self.netapp.id,
 			&self.local_status.load_full().hostname,
 			rpc_public_addr,
@@ -644,8 +636,9 @@ impl System {
 				}
 
 				// Fetch peer list from Consul
+				#[cfg(feature = "consul-discovery")]
 				if let Some(c) = &self.consul_discovery {
-					match get_consul_nodes(&c.consul_host, &c.service_name).await {
+					match get_consul_nodes(c).await {
 						Ok(node_list) => {
 							ping_list.extend(node_list);
 						}
@@ -667,7 +660,7 @@ impl System {
 						};
 					}
 
-					match get_kubernetes_nodes(&k.service_name, &k.namespace).await {
+					match get_kubernetes_nodes(k).await {
 						Ok(node_list) => {
 							ping_list.extend(node_list);
 						}
@@ -691,6 +684,7 @@ impl System {
 				warn!("Could not save peer list to file: {}", e);
 			}
 
+			#[cfg(feature = "consul-discovery")]
 			self.background.spawn(self.clone().advertise_to_consul());
 
 			#[cfg(feature = "kubernetes-discovery")]
@@ -785,15 +779,3 @@ async fn resolve_peers(peers: &[String]) -> Vec<(NodeID, SocketAddr)> {
 
 	ret
 }
-
-struct ConsulDiscoveryParam {
-	consul_host: String,
-	service_name: String,
-}
-
-#[cfg(feature = "kubernetes-discovery")]
-struct KubernetesDiscoveryParam {
-	service_name: String,
-	namespace: String,
-	skip_crd: bool,
-}