aboutsummaryrefslogtreecommitdiff
path: root/src/rpc
diff options
context:
space:
mode:
Diffstat (limited to 'src/rpc')
-rw-r--r--src/rpc/Cargo.toml4
-rw-r--r--src/rpc/consul.rs171
2 files changed, 113 insertions, 62 deletions
diff --git a/src/rpc/Cargo.toml b/src/rpc/Cargo.toml
index dcf44f4a..25ed839d 100644
--- a/src/rpc/Cargo.toml
+++ b/src/rpc/Cargo.toml
@@ -14,11 +14,11 @@ path = "lib.rs"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
-garage_util = { version = "0.8.2", path = "../util" }
+garage_util.workspace = true
arc-swap = "1.0"
bytes = "1.0"
-gethostname = "0.2"
+gethostname = "0.4"
hex = "0.4"
tracing = "0.1"
rand = "0.8"
diff --git a/src/rpc/consul.rs b/src/rpc/consul.rs
index f85f789c..ab8d1112 100644
--- a/src/rpc/consul.rs
+++ b/src/rpc/consul.rs
@@ -8,16 +8,26 @@ use serde::{Deserialize, Serialize};
use netapp::NodeID;
+use garage_util::config::ConsulDiscoveryAPI;
use garage_util::config::ConsulDiscoveryConfig;
+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 = "NodeMeta")]
- node_meta: HashMap<String, String>,
+ #[serde(rename = "ServiceMeta")]
+ meta: HashMap<String, String>,
+}
+
+#[derive(Serialize, Clone, Debug)]
+#[serde(untagged)]
+enum PublishRequest {
+ Catalog(ConsulPublishEntry),
+ Service(ConsulPublishService),
}
#[derive(Serialize, Clone, Debug)]
@@ -26,28 +36,43 @@ struct ConsulPublishEntry {
node: String,
#[serde(rename = "Address")]
address: IpAddr,
- #[serde(rename = "NodeMeta")]
- node_meta: HashMap<String, String>,
#[serde(rename = "Service")]
- service: ConsulPublishService,
+ service: ConsulPublishCatalogService,
}
#[derive(Serialize, Clone, Debug)]
-struct ConsulPublishService {
+struct ConsulPublishCatalogService {
#[serde(rename = "ID")]
service_id: String,
#[serde(rename = "Service")]
service_name: String,
#[serde(rename = "Tags")]
tags: Vec<String>,
+ #[serde(rename = "Meta")]
+ meta: HashMap<String, String>,
#[serde(rename = "Address")]
address: IpAddr,
#[serde(rename = "Port")]
port: u16,
}
-// ----
+#[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 ConsulDiscovery {
config: ConsulDiscoveryConfig,
client: reqwest::Client,
@@ -55,44 +80,48 @@ pub struct ConsulDiscovery {
impl ConsulDiscovery {
pub fn new(config: ConsulDiscoveryConfig) -> Result<Self, ConsulError> {
- 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()?
+ let mut builder: reqwest::ClientBuilder = reqwest::Client::builder().use_rustls_tls();
+ if config.tls_skip_verify {
+ builder = builder.danger_accept_invalid_certs(true);
+ } 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)?;
+ builder =
+ builder.add_root_certificate(reqwest::Certificate::from_pem(&ca_cert_buf[..])?);
+ }
+
+ match &config.api {
+ ConsulDiscoveryAPI::Catalog => 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()[..],
+ )?;
+
+ builder = builder.identity(identity);
+ }
+ (None, None) => {}
+ _ => return Err(ConsulError::InvalidTLSConfig),
+ },
+ ConsulDiscoveryAPI::Agent => {
+ if let Some(token) = &config.token {
+ let mut headers = reqwest::header::HeaderMap::new();
+ headers.insert(
+ "x-consul-token",
+ reqwest::header::HeaderValue::from_str(&token)?,
+ );
+ builder = builder.default_headers(headers);
}
}
- (None, None) => reqwest::Client::new(),
- _ => return Err(ConsulError::InvalidTLSConfig),
};
+ let client: reqwest::Client = builder.build()?;
+
Ok(Self { client, config })
}
@@ -111,8 +140,8 @@ impl ConsulDiscovery {
for ent in entries {
let ip = ent.address.parse::<IpAddr>().ok();
let pubkey = ent
- .node_meta
- .get("pubkey")
+ .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) {
@@ -138,29 +167,49 @@ impl ConsulDiscovery {
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 = 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 {
+ let mut meta = self.config.meta.clone().unwrap_or_default();
+ meta.insert(format!("{}-pubkey", META_PREFIX), hex::encode(node_id));
+ meta.insert(format!("{}-hostname", META_PREFIX), hostname.to_string());
+
+ let url = format!(
+ "{}/v1/{}",
+ self.config.consul_http_addr,
+ (match &self.config.api {
+ ConsulDiscoveryAPI::Catalog => "catalog/register",
+ ConsulDiscoveryAPI::Agent => "agent/service/register?replace-existing-checks",
+ })
+ );
+
+ let req = self.client.put(&url);
+ let advertisement: PublishRequest = match &self.config.api {
+ ConsulDiscoveryAPI::Catalog => PublishRequest::Catalog(ConsulPublishEntry {
+ node: node.clone(),
+ address: rpc_public_addr.ip(),
+ service: ConsulPublishCatalogService {
+ service_id: node.clone(),
+ service_name: self.config.service_name.clone(),
+ tags,
+ meta: meta.clone(),
+ address: rpc_public_addr.ip(),
+ port: rpc_public_addr.port(),
+ },
+ }),
+ ConsulDiscoveryAPI::Agent => PublishRequest::Service(ConsulPublishService {
service_id: node.clone(),
service_name: self.config.service_name.clone(),
- tags: vec!["advertised-by-garage".into(), hostname.into()],
+ tags,
+ meta,
address: rpc_public_addr.ip(),
port: rpc_public_addr.port(),
- },
+ }),
};
-
- let url = format!("{}/v1/catalog/register", self.config.consul_http_addr);
-
- let http = self.client.put(&url).json(&advertisement).send().await?;
+ let http = req.json(&advertisement).send().await?;
http.error_for_status()?;
Ok(())
@@ -176,4 +225,6 @@ pub enum ConsulError {
Reqwest(#[error(source)] reqwest::Error),
#[error(display = "Invalid Consul TLS configuration")]
InvalidTLSConfig,
+ #[error(display = "Token error: {}", _0)]
+ Token(#[error(source)] reqwest::header::InvalidHeaderValue),
}