Merge pull request 'Support for loading IP addresses from DiploNAT autodiscovery' (#4) from diplonat-autodiscovery into maindocker-4

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/D53/pulls/4

1 files changed, 148 insertions, 63 deletions

diff --git a/src/dns_config.rs b/src/dns_config.rs
index f4a95be..acee8d7 100644
--- a/src/dns_config.rs
+++ b/src/dns_config.rs
@@ -1,17 +1,23 @@
 use std::collections::{HashMap, HashSet};
 use std::fmt;
+use std::net::{Ipv4Addr, Ipv6Addr};
 use std::sync::Arc;
 use std::time::Duration;
 
+use anyhow::Result;
+use serde::{Deserialize, Serialize};
 use tokio::{select, sync::watch};
 use tracing::*;
 
 use df_consul::*;
 
-const IPV4_TARGET_METADATA_TAG: &str = "public_ipv4";
-const IPV6_TARGET_METADATA_TAG: &str = "public_ipv6";
+use crate::autodiscovery::*;
+
+const IP_TARGET_METADATA_TAG_PREFIX: &str = "public_";
 const CNAME_TARGET_METADATA_TAG: &str = "cname_target";
 
+const AUTODISCOVERY_CACHE_DURATION: u64 = 600; // 10 minutes
+
 // ---- Extract DNS config from Consul catalog ----
 
 #[derive(Debug, Eq, PartialEq, Default)]
@@ -54,86 +60,165 @@ impl DnsConfig {
     }
 }
 
-fn parse_d53_tag(tag: &str, node: &catalog::Node) -> Option<(DnsEntryKey, DnsEntryValue)> {
-    let splits = tag.split(' ').collect::<Vec<_>>();
-    if splits.len() != 2 {
-        return None;
-    }
-
-    let (record_type, targets) = match splits[0] {
-        "d53-a" => match node.meta.get(IPV4_TARGET_METADATA_TAG) {
-            Some(tgt) => (DnsRecordType::A, [tgt.to_string()].into_iter().collect()),
-            None => {
-                warn!("Got d53-a tag `{}` but node {} does not have a {} metadata value. Tag is ignored.", tag, node.node, IPV4_TARGET_METADATA_TAG);
-                return None;
-            }
-        },
-        "d53-aaaa" => match node.meta.get(IPV6_TARGET_METADATA_TAG) {
-            Some(tgt) => (DnsRecordType::AAAA, [tgt.to_string()].into_iter().collect()),
-            None => {
-                warn!("Got d53-aaaa tag `{}` but node {} does not have a {} metadata value. Tag is ignored.", tag, node.node, IPV6_TARGET_METADATA_TAG);
-                return None;
-            }
-        },
-        "d53-cname" => match node.meta.get(CNAME_TARGET_METADATA_TAG) {
-            Some(tgt) => (
-                DnsRecordType::CNAME,
-                [tgt.to_string()].into_iter().collect(),
-            ),
-            None => {
-                warn!("Got d53-cname tag `{}` but node {} does not have a {} metadata value. Tag is ignored.", tag, node.node, CNAME_TARGET_METADATA_TAG);
-                return None;
-            }
-        },
-        _ => return None,
-    };
-
-    Some((
-        DnsEntryKey {
-            dns_path: splits[1].to_string(),
-            record_type,
-        },
-        DnsEntryValue { targets },
-    ))
-}
+// ---- fetcher and autodiscovery cache ----
 
 pub fn spawn_dns_config_task(
-    consul: &Consul,
-    mut must_exit: watch::Receiver<bool>,
+    consul: Consul,
+    must_exit: watch::Receiver<bool>,
 ) -> watch::Receiver<Arc<DnsConfig>> {
     let (tx, rx) = watch::channel(Arc::new(DnsConfig::new()));
 
-    let mut catalog_rx = consul.watch_all_service_health(Duration::from_secs(60));
+    let fetcher = DnsConfigTask { consul };
+
+    tokio::spawn(fetcher.task(tx, must_exit));
+
+    rx
+}
+
+struct DnsConfigTask {
+    consul: Consul,
+}
+
+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));
 
-    tokio::spawn(async move {
         while !*must_exit.borrow() {
             select! {
                 _ = catalog_rx.changed() => (),
+                _ = autodiscovery_rx.changed() => (),
                 _ = must_exit.changed() => continue,
             };
 
-            let services = catalog_rx.borrow_and_update();
+            let services = catalog_rx.borrow_and_update().clone();
+            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);
+                }
+            };
+        }
+    }
 
-            let mut dns_config = DnsConfig::new();
-            for (_svc, nodes) in services.iter() {
-                for node in nodes.iter() {
-                    // Do not take into account backends if any have status critical
-                    if node.checks.iter().any(|x| x.status == "critical") {
-                        continue;
-                    }
-                    for tag in node.service.tags.iter() {
-                        if let Some((k, v)) = parse_d53_tag(tag, &node.node) {
-                            dns_config.add(k, v);
-                        }
+    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() {
+                // Do not take into account backends if any have status critical
+                if node.checks.iter().any(|x| x.status == "critical") {
+                    continue;
+                }
+                for tag in node.service.tags.iter() {
+                    if let Some((k, v)) = self.parse_d53_tag(tag, &node.node, autodiscovery)? {
+                        dns_config.add(k, v);
                     }
                 }
             }
+        }
 
-            tx.send(Arc::new(dns_config)).expect("Internal error");
+        Ok(dns_config)
+    }
+
+    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 {
+            return Ok(None);
         }
-    });
 
-    rx
+        let (record_type, target) = match splits[0] {
+            "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(&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);
+                    return Ok(None);
+                }
+            },
+            "d53-cname" => match node.meta.get(CNAME_TARGET_METADATA_TAG) {
+                Some(tgt) => (DnsRecordType::CNAME, tgt.to_string()),
+                None => {
+                    warn!("Got d53-cname tag `{}` but node {} does not have a {} metadata value. Tag is ignored.", tag, node.node, CNAME_TARGET_METADATA_TAG);
+                    return Ok(None);
+                }
+            },
+            _ => return Ok(None),
+        };
+
+        Ok(Some((
+            DnsEntryKey {
+                dns_path: splits[1].to_string(),
+                record_type,
+            },
+            DnsEntryValue {
+                targets: [target].into_iter().collect(),
+            },
+        )))
+    }
+
+    fn get_node_ipv4(
+        &mut self,
+        autodiscovery: &AutodiscoveredAddresses,
+        node: &catalog::Node,
+    ) -> Result<Option<Ipv4Addr>> {
+        Self::get_node_ip("ipv4", &autodiscovery.ipv4, node)
+    }
+
+    fn get_node_ipv6(
+        &mut self,
+        autodiscovery: &AutodiscoveredAddresses,
+        node: &catalog::Node,
+    ) -> Result<Option<Ipv6Addr>> {
+        Self::get_node_ip("ipv6", &autodiscovery.ipv6, node)
+    }
+
+    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 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 addr = node.meta.get(&meta_tag).map(|x| x.parse()).transpose()?;
+                Ok(addr)
+            }
+        }
+    }
 }
 
 // ---- Display impls ----