aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorAlex Auvolat <alex@adnab.me>2023-04-21 13:51:21 +0200
committerAlex Auvolat <alex@adnab.me>2023-04-21 13:51:21 +0200
commitfb7fde4b09df090f30ded5be4fbc094022aeff7a (patch)
tree089c9d0c2c70f0497aafda6a68a5b82d2cf90b04 /src
parentd906a6ebb5d977f44340b157a520477849ced161 (diff)
downloadD53-fb7fde4b09df090f30ded5be4fbc094022aeff7a.tar.gz
D53-fb7fde4b09df090f30ded5be4fbc094022aeff7a.zip
remove stupid cache and use proper watch on consul kv prefix
Diffstat (limited to 'src')
-rw-r--r--src/autodiscovery.rs110
-rw-r--r--src/dns_config.rs114
-rw-r--r--src/main.rs1
3 files changed, 156 insertions, 69 deletions
diff --git a/src/autodiscovery.rs b/src/autodiscovery.rs
new file mode 100644
index 0000000..9fcc094
--- /dev/null
+++ b/src/autodiscovery.rs
@@ -0,0 +1,110 @@
+//! Fetch autodiscoverd IP addresses stored by Diplonat into Consul
+
+use std::collections::HashMap;
+use std::net::{Ipv4Addr, Ipv6Addr};
+use std::sync::Arc;
+use std::time::{Duration, SystemTime};
+
+use anyhow::{anyhow, bail, Result};
+use regex::Regex;
+use serde::{Deserialize, Serialize};
+use tokio::{select, sync::watch};
+use tracing::*;
+
+use df_consul::*;
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct DiplonatAutodiscoveryResult<A> {
+ pub timestamp: u64,
+ pub address: Option<A>,
+}
+
+#[derive(Default, Debug)]
+pub struct AutodiscoveredAddresses {
+ pub ipv4: HashMap<String, DiplonatAutodiscoveryResult<Ipv4Addr>>,
+ pub ipv6: HashMap<String, DiplonatAutodiscoveryResult<Ipv6Addr>>,
+}
+
+pub fn watch_autodiscovered_ips(
+ consul: Consul,
+ mut must_exit: watch::Receiver<bool>,
+) -> watch::Receiver<Arc<AutodiscoveredAddresses>> {
+ let (tx, rx) = watch::channel(Arc::new(AutodiscoveredAddresses::default()));
+
+ tokio::spawn(async move {
+ let mut last_index = None;
+ let re = Regex::new(r".*autodiscovery/(\w+)/(\w+)$").unwrap();
+
+ while !*must_exit.borrow() {
+ let r = select! {
+ _ = must_exit.changed() => continue,
+ r = consul.kv_get_prefix("diplonat/autodiscovery/", last_index) => r,
+ };
+
+ let entries = match r {
+ Err(e) => {
+ warn!("Error fetching diplonat autodiscovery consul prefix: {}", e);
+ tokio::time::sleep(Duration::from_secs(30)).await;
+ continue;
+ }
+ Ok(r) => {
+ last_index = Some(r.index());
+ r.into_inner()
+ }
+ };
+
+ let mut addresses = AutodiscoveredAddresses::default();
+
+ for (k, v) in entries {
+ if let Err(e) = parse_autodiscovered_address(&re, &mut addresses, &k, &v) {
+ warn!(
+ "Invalid k/v pair in diplonat autodiscovery results: {} = {} ({})",
+ k,
+ std::str::from_utf8(&v).unwrap_or("<???>"),
+ e
+ );
+ }
+ }
+
+ if tx.send(Arc::new(addresses)).is_err() {
+ info!("Autodiscovered addresses watcher terminating");
+ }
+ }
+ });
+
+ rx
+}
+
+fn parse_autodiscovered_address(
+ re: &Regex,
+ addresses: &mut AutodiscoveredAddresses,
+ k: &str,
+ v: &[u8],
+) -> Result<()> {
+ let caps = re.captures(k).ok_or(anyhow!("key does not match regex"))?;
+
+ if let (Some(family), Some(node)) = (caps.get(1), caps.get(2)) {
+ match family.as_str() {
+ "ipv4" => {
+ let r: DiplonatAutodiscoveryResult<Ipv4Addr> = serde_json::from_slice(v)?;
+ addresses.ipv4.insert(node.as_str().to_string(), r);
+ }
+ "ipv6" => {
+ let r: DiplonatAutodiscoveryResult<Ipv6Addr> = serde_json::from_slice(v)?;
+ addresses.ipv6.insert(node.as_str().to_string(), r);
+ }
+ _ => bail!("invalid address family {}", family.as_str()),
+ }
+ } else {
+ bail!("invalid regex captures {:?}", caps);
+ }
+
+ Ok(())
+}
+
+pub fn timestamp() -> u64 {
+ SystemTime::now()
+ .duration_since(SystemTime::UNIX_EPOCH)
+ .expect("clock error")
+ .as_secs()
+}
diff --git a/src/dns_config.rs b/src/dns_config.rs
index 8e82c32..acee8d7 100644
--- a/src/dns_config.rs
+++ b/src/dns_config.rs
@@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet};
use std::fmt;
use std::net::{Ipv4Addr, Ipv6Addr};
use std::sync::Arc;
-use std::time::{Duration, SystemTime};
+use std::time::Duration;
use anyhow::Result;
use serde::{Deserialize, Serialize};
@@ -11,6 +11,8 @@ use tracing::*;
use df_consul::*;
+use crate::autodiscovery::*;
+
const IP_TARGET_METADATA_TAG_PREFIX: &str = "public_";
const CNAME_TARGET_METADATA_TAG: &str = "cname_target";
@@ -66,29 +68,25 @@ pub fn spawn_dns_config_task(
) -> watch::Receiver<Arc<DnsConfig>> {
let (tx, rx) = watch::channel(Arc::new(DnsConfig::new()));
- let fetcher = DnsConfigFetcher {
- consul,
- node_ipv4_cache: HashMap::new(),
- node_ipv6_cache: HashMap::new(),
- };
+ let fetcher = DnsConfigTask { consul };
tokio::spawn(fetcher.task(tx, must_exit));
rx
}
-struct DnsConfigFetcher {
+struct DnsConfigTask {
consul: Consul,
- node_ipv4_cache: HashMap<String, (u64, Option<Ipv4Addr>)>,
- node_ipv6_cache: HashMap<String, (u64, Option<Ipv6Addr>)>,
}
-impl DnsConfigFetcher {
+impl DnsConfigTask {
async fn task(
mut self,
tx: watch::Sender<Arc<DnsConfig>>,
mut must_exit: watch::Receiver<bool>,
) {
+ let mut autodiscovery_rx = watch_autodiscovered_ips(self.consul.clone(), must_exit.clone());
+
let mut catalog_rx = self
.consul
.watch_all_service_health(Duration::from_secs(60));
@@ -96,11 +94,13 @@ impl DnsConfigFetcher {
while !*must_exit.borrow() {
select! {
_ = catalog_rx.changed() => (),
+ _ = autodiscovery_rx.changed() => (),
_ = must_exit.changed() => continue,
};
let services = catalog_rx.borrow_and_update().clone();
- match self.parse_catalog(&services).await {
+ let autodiscovery = autodiscovery_rx.borrow_and_update().clone();
+ match self.parse_catalog(&services, &autodiscovery) {
Ok(dns_config) => tx.send(Arc::new(dns_config)).expect("Internal error"),
Err(e) => {
error!("Error when parsing tags: {}", e);
@@ -109,7 +109,11 @@ impl DnsConfigFetcher {
}
}
- async fn parse_catalog(&mut self, services: &catalog::AllServiceHealth) -> Result<DnsConfig> {
+ fn parse_catalog(
+ &mut self,
+ services: &catalog::AllServiceHealth,
+ autodiscovery: &AutodiscoveredAddresses,
+ ) -> Result<DnsConfig> {
let mut dns_config = DnsConfig::new();
for (_svc, nodes) in services.iter() {
for node in nodes.iter() {
@@ -118,7 +122,7 @@ impl DnsConfigFetcher {
continue;
}
for tag in node.service.tags.iter() {
- if let Some((k, v)) = self.parse_d53_tag(tag, &node.node).await? {
+ if let Some((k, v)) = self.parse_d53_tag(tag, &node.node, autodiscovery)? {
dns_config.add(k, v);
}
}
@@ -128,10 +132,11 @@ impl DnsConfigFetcher {
Ok(dns_config)
}
- async fn parse_d53_tag(
+ fn parse_d53_tag(
&mut self,
tag: &str,
node: &catalog::Node,
+ autodiscovery: &AutodiscoveredAddresses,
) -> Result<Option<(DnsEntryKey, DnsEntryValue)>> {
let splits = tag.split(' ').collect::<Vec<_>>();
if splits.len() != 2 {
@@ -139,14 +144,14 @@ impl DnsConfigFetcher {
}
let (record_type, target) = match splits[0] {
- "d53-a" => match self.get_node_ipv4(&node).await? {
+ "d53-a" => match self.get_node_ipv4(&autodiscovery, &node)? {
Some(tgt) => (DnsRecordType::A, tgt.to_string()),
None => {
warn!("Got d53-a tag `{}` but node {} does not appear to have a known public IPv4 address. Tag is ignored.", tag, node.node);
return Ok(None);
}
},
- "d53-aaaa" => match self.get_node_ipv6(&node).await? {
+ "d53-aaaa" => match self.get_node_ipv6(&autodiscovery, &node)? {
Some(tgt) => (DnsRecordType::AAAA, tgt.to_string()),
None => {
warn!("Got d53-aaaa tag `{}` but node {} does not appear to have a known public IPv6 address. Tag is ignored.", tag, node.node);
@@ -174,77 +179,48 @@ impl DnsConfigFetcher {
)))
}
- async fn get_node_ipv4(&mut self, node: &catalog::Node) -> Result<Option<Ipv4Addr>> {
- Self::get_node_ip(&self.consul, "ipv4", &mut self.node_ipv4_cache, node).await
+ fn get_node_ipv4(
+ &mut self,
+ autodiscovery: &AutodiscoveredAddresses,
+ node: &catalog::Node,
+ ) -> Result<Option<Ipv4Addr>> {
+ Self::get_node_ip("ipv4", &autodiscovery.ipv4, node)
}
- async fn get_node_ipv6(&mut self, node: &catalog::Node) -> Result<Option<Ipv6Addr>> {
- Self::get_node_ip(&self.consul, "ipv6", &mut self.node_ipv6_cache, node).await
+ fn get_node_ipv6(
+ &mut self,
+ autodiscovery: &AutodiscoveredAddresses,
+ node: &catalog::Node,
+ ) -> Result<Option<Ipv6Addr>> {
+ Self::get_node_ip("ipv6", &autodiscovery.ipv6, node)
}
- async fn get_node_ip<A>(
- consul: &Consul,
- family: &str,
- cache: &mut HashMap<String, (u64, Option<A>)>,
+ fn get_node_ip<A>(
+ family: &'static str,
+ autodiscovery: &HashMap<String, DiplonatAutodiscoveryResult<A>>,
node: &catalog::Node,
) -> Result<Option<A>>
where
A: Serialize + for<'de> Deserialize<'de> + std::fmt::Debug + std::str::FromStr + Copy + Eq,
<A as std::str::FromStr>::Err: Send + Sync + std::error::Error + 'static,
{
- match cache.get(&node.node) {
- Some((t, a)) if timestamp() <= t + AUTODISCOVERY_CACHE_DURATION => Ok(*a),
- _ => {
- let kv_key = format!("diplonat/autodiscovery/{}/{}", family, node.node);
- let autodiscovery = consul.kv_get(&kv_key).await?;
-
- if let Some(json) = autodiscovery {
- let a = serde_json::from_slice::<DiplonatAutodiscoveryResult<A>>(&json)?;
- if timestamp() <= a.timestamp + AUTODISCOVERY_CACHE_DURATION {
- if cache.get(&node.node).map(|x| x.1) != Some(a.address) {
- info!(
- "Got {} address for {} from diplonat autodiscovery: {:?}",
- family, node.node, a.address
- );
- }
- cache.insert(node.node.clone(), (a.timestamp, a.address));
- return Ok(a.address);
- } else {
- warn!("{} address for {} from diplonat autodiscovery is outdated (value: {:?}), falling back on value from Consul node meta", family, node.node, a.address);
- }
+ match autodiscovery.get(&node.node) {
+ Some(ar) if timestamp() <= ar.timestamp + AUTODISCOVERY_CACHE_DURATION => {
+ Ok(ar.address)
+ }
+ x => {
+ if let Some(ar) = x {
+ warn!("{} address for {} from diplonat autodiscovery is outdated (value: {:?}), falling back on value from Consul node meta", family, node.node, ar.address);
}
let meta_tag = format!("{}{}", IP_TARGET_METADATA_TAG_PREFIX, family);
- let a = node.meta.get(&meta_tag).map(|x| x.parse()).transpose()?;
-
- if cache.get(&node.node).map(|x| x.1) != Some(a) {
- info!(
- "Got {} address for {} from Consul node meta: {:?}",
- family, node.node, a
- );
- }
- cache.insert(node.node.clone(), (timestamp(), a));
- Ok(a)
+ let addr = node.meta.get(&meta_tag).map(|x| x.parse()).transpose()?;
+ Ok(addr)
}
}
}
}
-// ---- util for interaction with diplonat ----
-
-#[derive(Serialize, Deserialize, Debug)]
-pub struct DiplonatAutodiscoveryResult<A> {
- pub timestamp: u64,
- pub address: Option<A>,
-}
-
-fn timestamp() -> u64 {
- SystemTime::now()
- .duration_since(SystemTime::UNIX_EPOCH)
- .expect("clock error")
- .as_secs()
-}
-
// ---- Display impls ----
impl std::fmt::Display for DnsRecordType {
diff --git a/src/main.rs b/src/main.rs
index a8a6b36..9461596 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -5,6 +5,7 @@ use tokio::select;
use tokio::sync::watch;
use tracing::*;
+mod autodiscovery;
mod dns_config;
mod dns_updater;
mod provider;