From fcaee3bea0019123db05356fd82706f38498365c Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Fri, 14 Oct 2022 18:10:36 +0200 Subject: definitively expunge openssl from dependencies everywhere --- src/rpc/Cargo.toml | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) (limited to 'src/rpc') diff --git a/src/rpc/Cargo.toml b/src/rpc/Cargo.toml index d61acea4..883929e8 100644 --- a/src/rpc/Cargo.toml +++ b/src/rpc/Cargo.toml @@ -31,9 +31,8 @@ serde_bytes = "0.11" serde_json = "1.0" # newer version requires rust edition 2021 -kube = { version = "0.62", features = ["runtime", "derive"], optional = true } -k8s-openapi = { version = "0.13", features = ["v1_22"], optional = true } -openssl = { version = "0.10", features = ["vendored"], optional = true } +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 } # newer version requires rust edition 2021 @@ -51,5 +50,5 @@ hyper = { version = "0.14", features = ["client", "http1", "runtime", "tcp"] } [features] -kubernetes-discovery = [ "kube", "k8s-openapi", "openssl", "schemars" ] +kubernetes-discovery = [ "kube", "k8s-openapi", "schemars" ] system-libs = [ "sodiumoxide/use-pkg-config" ] -- cgit v1.2.3 From 002b9fc50c5b69e0e10c84e4db5ecea1b3941fad Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 18 Oct 2022 18:38:20 +0200 Subject: Add TLS support for Consul discovery + refactoring --- src/rpc/Cargo.toml | 6 +-- src/rpc/consul.rs | 129 +++++++++++++++++++++++++++++++------------------- src/rpc/kubernetes.rs | 16 +++---- src/rpc/lib.rs | 1 + src/rpc/system.rs | 70 ++++++++++----------------- 5 files changed, 119 insertions(+), 103 deletions(-) (limited to 'src/rpc') diff --git a/src/rpc/Cargo.toml b/src/rpc/Cargo.toml index 883929e8..aedac82b 100644 --- a/src/rpc/Cargo.toml +++ b/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" ] diff --git a/src/rpc/consul.rs b/src/rpc/consul.rs index 15acbcef..05ed278a 100644 --- a/src/rpc/consul.rs +++ b/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 { + 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, Error> { + consul_config: &ConsulDiscoveryConfig, +) -> Result, 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::>(body.as_ref())?; + let client = make_consul_client(consul_config).await?; + let http = client.get(&url).send().await?; + let entries: Vec = 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("") - ); - - 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, +} diff --git a/src/rpc/kubernetes.rs b/src/rpc/kubernetes.rs index 197245aa..63c6567d 100644 --- a/src/rpc/kubernetes.rs +++ b/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, kube::Error> { let client = Client::try_default().await?; - let nodes: Api = Api::namespaced(client.clone(), kubernetes_namespace); + let nodes: Api = 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 = Api::namespaced(client.clone(), kubernetes_namespace); + let nodes: Api = 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; diff --git a/src/rpc/lib.rs b/src/rpc/lib.rs index 392ff48f..92caf75d 100644 --- a/src/rpc/lib.rs +++ b/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; diff --git a/src/rpc/system.rs b/src/rpc/system.rs index 9e0bfa11..7b4cfbde 100644 --- a/src/rpc/system.rs +++ b/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>, rpc_listen_addr: SocketAddr, + #[cfg(any(feature = "consul-discovery", feature = "kubernetes-discovery"))] rpc_public_addr: Option, bootstrap_peers: Vec, - consul_discovery: Option, + #[cfg(feature = "consul-discovery")] + consul_discovery: Option, #[cfg(feature = "kubernetes-discovery")] - kubernetes_discovery: Option, + kubernetes_discovery: Option, 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) -> 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, -} -- cgit v1.2.3 From 5d8d393054f9aa9364fde6af119366953a54081d Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 18 Oct 2022 19:11:16 +0200 Subject: Load TLS certificates only once --- src/rpc/consul.rs | 243 ++++++++++++++++++++++++++---------------------------- src/rpc/system.rs | 22 +++-- 2 files changed, 133 insertions(+), 132 deletions(-) (limited to 'src/rpc') diff --git a/src/rpc/consul.rs b/src/rpc/consul.rs index 05ed278a..cf050207 100644 --- a/src/rpc/consul.rs +++ b/src/rpc/consul.rs @@ -1,9 +1,8 @@ use std::collections::HashMap; +use std::fs::File; +use std::io::Read; use std::net::{IpAddr, SocketAddr}; -use tokio::fs::File; -use tokio::io::AsyncReadExt; - use err_derive::Error; use serde::{Deserialize, Serialize}; @@ -11,59 +10,6 @@ use netapp::NodeID; use garage_util::config::ConsulDiscoveryConfig; -async fn make_consul_client( - config: &ConsulDiscoveryConfig, -) -> Result { - 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 ---- - #[derive(Deserialize, Clone, Debug)] struct ConsulQueryEntry { #[serde(rename = "Address")] @@ -74,42 +20,6 @@ struct ConsulQueryEntry { node_meta: HashMap, } -pub async fn get_consul_nodes( - consul_config: &ConsulDiscoveryConfig, -) -> Result, ConsulError> { - let url = format!( - "http://{}/v1/catalog/service/{}", - consul_config.consul_host, consul_config.service_name - ); - - let client = make_consul_client(consul_config).await?; - let http = client.get(&url).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 - .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")] @@ -136,43 +46,128 @@ struct ConsulPublishService { port: u16, } -pub async fn publish_consul_service( - consul_config: &ConsulDiscoveryConfig, - 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(), - 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_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_config.consul_host); +pub struct ConsulDiscovery { + config: ConsulDiscoveryConfig, + client: reqwest::Client, +} + +impl ConsulDiscovery { + pub fn new(config: ConsulDiscoveryConfig) -> Result { + 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, ConsulError> { + let url = format!( + "http://{}/v1/catalog/service/{}", + self.config.consul_host, self.config.service_name + ); + + let http = self.client.get(&url).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 + .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) + } - let client = make_consul_client(consul_config).await?; - let http = client.put(&url).json(&advertisement).send().await?; - http.error_for_status()?; + // ---- PUBLISHING TO CONSUL CATALOG ---- - Ok(()) + 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(), + 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!("http://{}/v1/catalog/register", self.config.consul_host); + + let http = self.client.put(&url).json(&advertisement).send().await?; + http.error_for_status()?; + + Ok(()) + } } -/// Regroup all Garage errors +/// Regroup all Consul discovery errors #[derive(Debug, Error)] pub enum ConsulError { #[error(display = "IO error: {}", _0)] diff --git a/src/rpc/system.rs b/src/rpc/system.rs index 7b4cfbde..61e380c9 100644 --- a/src/rpc/system.rs +++ b/src/rpc/system.rs @@ -23,8 +23,6 @@ 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::*; @@ -33,7 +31,7 @@ use garage_util::persister::Persister; use garage_util::time::*; #[cfg(feature = "consul-discovery")] -use crate::consul::{get_consul_nodes, publish_consul_service}; +use crate::consul::ConsulDiscovery; #[cfg(feature = "kubernetes-discovery")] use crate::kubernetes::*; use crate::layout::*; @@ -100,7 +98,7 @@ pub struct System { bootstrap_peers: Vec, #[cfg(feature = "consul-discovery")] - consul_discovery: Option, + consul_discovery: Option, #[cfg(feature = "kubernetes-discovery")] kubernetes_discovery: Option, @@ -302,6 +300,15 @@ impl System { warn!("Kubernetes discovery is not enabled in this build."); } + #[cfg(feature = "consul-discovery")] + let consul_discovery = match &config.consul_discovery { + Some(cfg) => Some( + ConsulDiscovery::new(cfg.clone()) + .ok_or_message("Invalid Consul discovery configuration")?, + ), + None => None, + }; + let sys = Arc::new(System { id: netapp.id.into(), persist_cluster_layout, @@ -324,7 +331,7 @@ impl System { rpc_public_addr, bootstrap_peers: config.bootstrap_peers.clone(), #[cfg(feature = "consul-discovery")] - consul_discovery: config.consul_discovery.clone(), + consul_discovery, #[cfg(feature = "kubernetes-discovery")] kubernetes_discovery: config.kubernetes_discovery.clone(), @@ -440,8 +447,7 @@ impl System { } }; - publish_consul_service( - c, + c.publish_consul_service( self.netapp.id, &self.local_status.load_full().hostname, rpc_public_addr, @@ -638,7 +644,7 @@ impl System { // Fetch peer list from Consul #[cfg(feature = "consul-discovery")] if let Some(c) = &self.consul_discovery { - match get_consul_nodes(c).await { + match c.get_consul_nodes().await { Ok(node_list) => { ping_list.extend(node_list); } -- cgit v1.2.3 From 2da8786f54dcf13bed51549f818a1af3bbebaa8c Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 18 Oct 2022 19:13:52 +0200 Subject: move things around --- src/rpc/system.rs | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) (limited to 'src/rpc') diff --git a/src/rpc/system.rs b/src/rpc/system.rs index 61e380c9..d6576f20 100644 --- a/src/rpc/system.rs +++ b/src/rpc/system.rs @@ -290,6 +290,14 @@ impl System { let system_endpoint = netapp.endpoint(SYSTEM_RPC_PATH.into()); + #[cfg(feature = "consul-discovery")] + let consul_discovery = match &config.consul_discovery { + Some(cfg) => Some( + ConsulDiscovery::new(cfg.clone()) + .ok_or_message("Invalid Consul discovery configuration")?, + ), + None => None, + }; #[cfg(not(feature = "consul-discovery"))] if config.consul_discovery.is_some() { warn!("Consul discovery is not enabled in this build."); @@ -300,15 +308,6 @@ impl System { warn!("Kubernetes discovery is not enabled in this build."); } - #[cfg(feature = "consul-discovery")] - let consul_discovery = match &config.consul_discovery { - Some(cfg) => Some( - ConsulDiscovery::new(cfg.clone()) - .ok_or_message("Invalid Consul discovery configuration")?, - ), - None => None, - }; - let sys = Arc::new(System { id: netapp.id.into(), persist_cluster_layout, -- cgit v1.2.3 From 8bc5caf7aa9bc0e27b741c68113cb7fdde2d54e6 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 18 Oct 2022 21:17:11 +0200 Subject: Fix issue with 'http(s)://' prefix --- src/rpc/consul.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'src/rpc') diff --git a/src/rpc/consul.rs b/src/rpc/consul.rs index cf050207..b1772a1a 100644 --- a/src/rpc/consul.rs +++ b/src/rpc/consul.rs @@ -100,8 +100,8 @@ impl ConsulDiscovery { pub async fn get_consul_nodes(&self) -> Result, ConsulError> { let url = format!( - "http://{}/v1/catalog/service/{}", - self.config.consul_host, self.config.service_name + "{}/v1/catalog/service/{}", + self.config.consul_http_addr, self.config.service_name ); let http = self.client.get(&url).send().await?; @@ -158,7 +158,7 @@ impl ConsulDiscovery { }, }; - let url = format!("http://{}/v1/catalog/register", self.config.consul_host); + 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()?; -- cgit v1.2.3 From 57b5c2c754a8d6d55882eae3dad44e328d961084 Mon Sep 17 00:00:00 2001 From: Alex Auvolat Date: Tue, 18 Oct 2022 22:11:27 +0200 Subject: Change reqwest rustls features --- src/rpc/Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src/rpc') diff --git a/src/rpc/Cargo.toml b/src/rpc/Cargo.toml index aedac82b..2c2ddc0b 100644 --- a/src/rpc/Cargo.toml +++ b/src/rpc/Cargo.toml @@ -35,7 +35,7 @@ err-derive = { version = "0.3", optional = true } 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"] } +reqwest = { version = "0.11", optional = true, default-features = false, features = ["rustls-tls-manual-roots", "json"] } # newer version requires rust edition 2021 pnet_datalink = "0.28" -- cgit v1.2.3