aboutsummaryrefslogtreecommitdiff
path: root/src/rpc/system.rs
diff options
context:
space:
mode:
authorRoberto Hidalgo <un@rob.mx>2023-05-05 16:18:24 -0600
committerRoberto Hidalgo <un@rob.mx>2023-05-22 08:57:15 -0600
commit02ba9016ab6eca5bb4964549de002573d5a3a07a (patch)
tree85fc01db2fa60605d2016f253da3a64ae8ef75d9 /src/rpc/system.rs
parent9d833bb7efc2c166036db38da89b0b5ac8f466fe (diff)
downloadgarage-02ba9016ab6eca5bb4964549de002573d5a3a07a.tar.gz
garage-02ba9016ab6eca5bb4964549de002573d5a3a07a.zip
register consul services against local agent instead of catalog api
Diffstat (limited to 'src/rpc/system.rs')
-rw-r--r--src/rpc/system.rs69
1 files changed, 66 insertions, 3 deletions
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());