From 02ba9016ab6eca5bb4964549de002573d5a3a07a Mon Sep 17 00:00:00 2001 From: Roberto Hidalgo Date: Fri, 5 May 2023 16:18:24 -0600 Subject: register consul services against local agent instead of catalog api --- src/rpc/Cargo.toml | 1 + src/rpc/consul_services.rs | 166 +++++++++++++++++++++++++++++++++++++++++++++ src/rpc/lib.rs | 2 + src/rpc/system.rs | 69 ++++++++++++++++++- 4 files changed, 235 insertions(+), 3 deletions(-) create mode 100644 src/rpc/consul_services.rs (limited to 'src/rpc') 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, +} + +#[derive(Serialize, Clone, Debug)] +struct ConsulPublishService { + #[serde(rename = "ID")] + service_id: String, + #[serde(rename = "Name")] + service_name: String, + #[serde(rename = "Tags")] + tags: Vec, + #[serde(rename = "Address")] + address: IpAddr, + #[serde(rename = "Port")] + port: u16, + #[serde(rename = "Meta")] + meta: HashMap, +} + +// ---- + +pub struct ConsulServiceDiscovery { + config: ConsulServiceConfig, + client: reqwest::Client, +} + +impl ConsulServiceDiscovery { + pub fn new(config: ConsulServiceConfig) -> Result { + 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, 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 = http.json().await?; + + let mut ret = vec![]; + for ent in entries { + let ip = ent.address.parse::().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>, 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, bootstrap_peers: Vec, #[cfg(feature = "consul-discovery")] consul_discovery: Option, + #[cfg(feature = "consul-service-discovery")] + consul_service_discovery: Option, #[cfg(feature = "kubernetes-discovery")] kubernetes_discovery: Option, @@ -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) { + 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) { 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()); -- cgit v1.2.3