aboutsummaryrefslogtreecommitdiff
path: root/src/rpc/consul.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/rpc/consul.rs')
-rw-r--r--src/rpc/consul.rs234
1 files changed, 131 insertions, 103 deletions
diff --git a/src/rpc/consul.rs b/src/rpc/consul.rs
index 15acbcef..b1772a1a 100644
--- a/src/rpc/consul.rs
+++ b/src/rpc/consul.rs
@@ -1,16 +1,14 @@
use std::collections::HashMap;
+use std::fs::File;
+use std::io::Read;
use std::net::{IpAddr, SocketAddr};
-use hyper::client::Client;
-use hyper::StatusCode;
-use hyper::{Body, Method, Request};
+use err_derive::Error;
use serde::{Deserialize, Serialize};
use netapp::NodeID;
-use garage_util::error::Error;
-
-// ---- READING FROM CONSUL CATALOG ----
+use garage_util::config::ConsulDiscoveryConfig;
#[derive(Deserialize, Clone, Debug)]
struct ConsulQueryEntry {
@@ -22,53 +20,6 @@ struct ConsulQueryEntry {
node_meta: HashMap<String, String>,
}
-pub async fn get_consul_nodes(
- consul_host: &str,
- consul_service_name: &str,
-) -> Result<Vec<(NodeID, SocketAddr)>, Error> {
- let url = format!(
- "http://{}/v1/catalog/service/{}",
- consul_host, consul_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 mut ret = vec![];
- for ent in entries {
- let ip = ent.address.parse::<IpAddr>().ok();
- let pubkey = ent
- .node_meta
- .get("pubkey")
- .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 ----
-
#[derive(Serialize, Clone, Debug)]
struct ConsulPublishEntry {
#[serde(rename = "Node")]
@@ -95,57 +46,134 @@ struct ConsulPublishService {
port: u16,
}
-pub async fn publish_consul_service(
- consul_host: &str,
- consul_service_name: &str,
- node_id: NodeID,
- hostname: &str,
- rpc_public_addr: SocketAddr,
-) -> Result<(), Error> {
- let node = format!("garage:{}", hex::encode(&node_id[..8]));
-
- let advertisment = 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 {
- service_id: node.clone(),
- service_name: consul_service_name.to_string(),
- tags: vec!["advertised-by-garage".into(), hostname.into()],
+// ----
+
+pub struct ConsulDiscovery {
+ config: ConsulDiscoveryConfig,
+ client: reqwest::Client,
+}
+
+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()?
+ }
+ }
+ (None, None) => reqwest::Client::new(),
+ _ => return Err(ConsulError::InvalidTLSConfig),
+ };
+
+ Ok(Self { client, config })
+ }
+
+ // ---- READING FROM CONSUL CATALOG ----
+
+ pub async fn get_consul_nodes(&self) -> Result<Vec<(NodeID, SocketAddr)>, ConsulError> {
+ let url = format!(
+ "{}/v1/catalog/service/{}",
+ self.config.consul_http_addr, self.config.service_name
+ );
+
+ let http = self.client.get(&url).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
+ .node_meta
+ .get("pubkey")
+ .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 advertisement = ConsulPublishEntry {
+ node: node.clone(),
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 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)));
+ node_meta: [
+ ("pubkey".to_string(), hex::encode(node_id)),
+ ("hostname".to_string(), hostname.to_string()),
+ ]
+ .iter()
+ .cloned()
+ .collect(),
+ service: ConsulPublishService {
+ service_id: node.clone(),
+ service_name: self.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!("{}/v1/catalog/register", self.config.consul_http_addr);
+
+ let http = self.client.put(&url).json(&advertisement).send().await?;
+ http.error_for_status()?;
+
+ Ok(())
}
+}
- 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 Consul TLS configuration")]
+ InvalidTLSConfig,
}
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-01 08:32:17 +0000