forked from Deuxfleurs/garage
register consul services against local agent instead of catalog api
This commit is contained in:
parent
9d833bb7ef
commit
02ba9016ab
8 changed files with 262 additions and 6 deletions
doc/book/development
src
|
@ -25,7 +25,7 @@ git clone https://git.deuxfleurs.fr/Deuxfleurs/garage
|
|||
cd garage
|
||||
```
|
||||
|
||||
*Optionnaly, you can use our nix.conf file to speed up compilations:*
|
||||
*Optionally, you can use our nix.conf file to speed up compilations:*
|
||||
|
||||
```bash
|
||||
sudo mkdir -p /etc/nix
|
||||
|
|
|
@ -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).
|
||||
|
|
|
@ -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")]
|
||||
|
|
|
@ -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" ]
|
||||
|
|
166
src/rpc/consul_services.rs
Normal file
166
src/rpc/consul_services.rs
Normal file
|
@ -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),
|
||||
}
|
|
@ -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;
|
||||
|
||||
|
|
|
@ -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());
|
||||
|
||||
|
|
|
@ -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>,
|
||||
|
@ -152,6 +155,24 @@ pub struct ConsulDiscoveryConfig {
|
|||
pub tls_skip_verify: bool,
|
||||
}
|
||||
|
||||
#[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
|
||||
|
|
Loading…
Reference in a new issue