aboutsummaryrefslogtreecommitdiff
path: root/src/rpc
diff options
context:
space:
mode:
Diffstat (limited to 'src/rpc')
-rw-r--r--src/rpc/Cargo.toml6
-rw-r--r--src/rpc/consul.rs129
-rw-r--r--src/rpc/kubernetes.rs16
-rw-r--r--src/rpc/lib.rs1
-rw-r--r--src/rpc/system.rs70
5 files changed, 119 insertions, 103 deletions
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<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,
+}
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<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;
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<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,
-}
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-07 16:00:13 +0000