aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/garage/Cargo.toml1
-rw-r--r--src/garage/main.rs2
-rw-r--r--src/rpc/Cargo.toml1
-rw-r--r--src/rpc/consul_services.rs166
-rw-r--r--src/rpc/lib.rs2
-rw-r--r--src/rpc/system.rs69
-rw-r--r--src/util/config.rs25
7 files changed, 261 insertions, 5 deletions
diff --git a/src/garage/Cargo.toml b/src/garage/Cargo.toml
index 52d0ea79..aaf028a8 100644
--- a/src/garage/Cargo.toml
+++ b/src/garage/Cargo.toml
@@ -88,6 +88,7 @@ sqlite = [ "garage_model/sqlite" ]
# Automatic registration and discovery via Consul API
consul-discovery = [ "garage_rpc/consul-discovery" ]
+consul-service-discovery = [ "garage_rpc/consul-service-discovery" ]
# Automatic registration and discovery via Kubernetes API
kubernetes-discovery = [ "garage_rpc/kubernetes-discovery" ]
# Prometheus exporter (/metrics endpoint).
diff --git a/src/garage/main.rs b/src/garage/main.rs
index e8aee892..103c027d 100644
--- a/src/garage/main.rs
+++ b/src/garage/main.rs
@@ -95,6 +95,8 @@ async fn main() {
"sqlite",
#[cfg(feature = "consul-discovery")]
"consul-discovery",
+ #[cfg(feature = "consul-service-discovery")]
+ "consul-service-discovery",
#[cfg(feature = "kubernetes-discovery")]
"kubernetes-discovery",
#[cfg(feature = "metrics")]
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<String, String>,
+}
+
+#[derive(Serialize, Clone, Debug)]
+struct ConsulPublishService {
+ #[serde(rename = "ID")]
+ service_id: String,
+ #[serde(rename = "Name")]
+ service_name: String,
+ #[serde(rename = "Tags")]
+ tags: Vec<String>,
+ #[serde(rename = "Address")]
+ address: IpAddr,
+ #[serde(rename = "Port")]
+ port: u16,
+ #[serde(rename = "Meta")]
+ meta: HashMap<String, String>,
+}
+
+// ----
+
+pub struct ConsulServiceDiscovery {
+ config: ConsulServiceConfig,
+ client: reqwest::Client,
+}
+
+impl ConsulServiceDiscovery {
+ pub fn new(config: ConsulServiceConfig) -> Result<Self, ConsulError> {
+ 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<Vec<(NodeID, SocketAddr)>, 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<ConsulQueryEntry> = http.json().await?;
+
+ let mut ret = vec![];
+ for ent in entries {
+ let ip = ent.address.parse::<IpAddr>().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<Endpoint<SystemRpc, System>>,
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<SocketAddr>,
bootstrap_peers: Vec<String>,
#[cfg(feature = "consul-discovery")]
consul_discovery: Option<ConsulDiscovery>,
+ #[cfg(feature = "consul-service-discovery")]
+ consul_service_discovery: Option<ConsulServiceDiscovery>,
#[cfg(feature = "kubernetes-discovery")]
kubernetes_discovery: Option<KubernetesDiscoveryConfig>,
@@ -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<Self>) {
+ 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<Self>) {
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());
diff --git a/src/util/config.rs b/src/util/config.rs
index 95835bbb..4b32f8ba 100644
--- a/src/util/config.rs
+++ b/src/util/config.rs
@@ -56,6 +56,9 @@ pub struct Config {
/// Configuration for automatic node discovery through Consul
#[serde(default)]
pub consul_discovery: Option<ConsulDiscoveryConfig>,
+ /// Configuration for automatic node discovery through Consul
+ #[serde(default)]
+ pub consul_service_discovery: Option<ConsulServiceConfig>,
/// Configuration for automatic node discovery through Kubernetes
#[serde(default)]
pub kubernetes_discovery: Option<KubernetesDiscoveryConfig>,
@@ -153,6 +156,24 @@ pub struct ConsulDiscoveryConfig {
}
#[derive(Deserialize, Debug, Clone)]
+pub struct ConsulServiceConfig {
+ /// Consul http or https address to connect to to discover more peers
+ pub consul_http_addr: String,
+ /// Token to use for connecting to consul
+ pub consul_http_token: Option<String>,
+ /// Consul service name to use
+ pub service_name: String,
+ /// CA TLS certificate to use when connecting to Consul
+ pub ca_cert: Option<String>,
+ // Additional tags to add to the service
+ #[serde(default)]
+ pub tags: Vec<String>,
+ /// Skip TLS hostname verification
+ #[serde(default)]
+ pub tls_skip_verify: bool,
+}
+
+#[derive(Deserialize, Debug, Clone)]
pub struct KubernetesDiscoveryConfig {
/// Kubernetes namespace the service discovery resources are be created in
pub namespace: String,
@@ -344,7 +365,7 @@ mod tests {
replication_mode = "3"
rpc_bind_addr = "[::]:3901"
rpc_secret_file = "{}"
-
+
[s3_api]
s3_region = "garage"
api_bind_addr = "[::]:3900"
@@ -388,7 +409,7 @@ mod tests {
rpc_bind_addr = "[::]:3901"
rpc_secret= "dummy"
rpc_secret_file = "dummy"
-
+
[s3_api]
s3_region = "garage"
api_bind_addr = "[::]:3900"
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-27 00:05:48 +0000