diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/autodiscovery.rs | 110 | ||||
| -rw-r--r-- | src/dns_config.rs | 114 | ||||
| -rw-r--r-- | src/main.rs | 1 |
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; |