Update df_consul dependency and avoid advertising failed backends (fix #2)

1 files changed, 18 insertions, 120 deletions

diff --git a/src/dns_config.rs b/src/dns_config.rs
index 0697270..f4a95be 100644
--- a/src/dns_config.rs
+++ b/src/dns_config.rs
@@ -1,14 +1,9 @@
 use std::collections::{HashMap, HashSet};
 use std::fmt;
 use std::sync::Arc;
-use std::{cmp, time::Duration};
+use std::time::Duration;
 
-use anyhow::Result;
-
-use futures::future::BoxFuture;
-use futures::stream::{FuturesUnordered, StreamExt};
-
-use tokio::{select, sync::watch, time::sleep};
+use tokio::{select, sync::watch};
 use tracing::*;
 
 use df_consul::*;
@@ -59,7 +54,7 @@ impl DnsConfig {
     }
 }
 
-fn parse_d53_tag(tag: &str, node: &ConsulNode) -> Option<(DnsEntryKey, DnsEntryValue)> {
+fn parse_d53_tag(tag: &str, node: &catalog::Node) -> Option<(DnsEntryKey, DnsEntryValue)> {
     let splits = tag.split(' ').collect::<Vec<_>>();
     if splits.len() != 2 {
         return None;
@@ -102,123 +97,36 @@ fn parse_d53_tag(tag: &str, node: &ConsulNode) -> Option<(DnsEntryKey, DnsEntryV
     ))
 }
 
-fn parse_consul_catalog(catalog: &ConsulNodeCatalog, dns_config: &mut DnsConfig) {
-    trace!("Parsing node catalog: {:#?}", catalog);
-
-    for (_, svc) in catalog.services.iter() {
-        for tag in svc.tags.iter() {
-            if let Some((k, v)) = parse_d53_tag(tag, &catalog.node) {
-                dns_config.add(k, v);
-            }
-        }
-    }
-}
-
-#[derive(Default)]
-struct NodeWatchState {
-    last_idx: Option<usize>,
-    last_catalog: Option<ConsulNodeCatalog>,
-    retries: u32,
-}
-
 pub fn spawn_dns_config_task(
-    consul: Consul,
+    consul: &Consul,
     mut must_exit: watch::Receiver<bool>,
 ) -> watch::Receiver<Arc<DnsConfig>> {
     let (tx, rx) = watch::channel(Arc::new(DnsConfig::new()));
 
-    let consul = Arc::new(consul);
+    let mut catalog_rx = consul.watch_all_service_health(Duration::from_secs(60));
 
     tokio::spawn(async move {
-        let mut nodes = HashMap::new();
-        let mut watches = FuturesUnordered::<BoxFuture<'static, (String, Result<_>)>>::new();
-
         while !*must_exit.borrow() {
-            let list_nodes = select! {
-                ln = consul.list_nodes() => ln,
+            select! {
+                _ = catalog_rx.changed() => (),
                 _ = must_exit.changed() => continue,
             };
 
-            match list_nodes {
-                Ok(consul_nodes) => {
-                    debug!("Watched consul nodes: {:?}", consul_nodes);
-                    for consul_node in consul_nodes {
-                        let node = &consul_node.node;
-                        if !nodes.contains_key(node) {
-                            nodes.insert(node.clone(), NodeWatchState::default());
-
-                            let node = node.to_string();
-                            let consul = consul.clone();
+            let services = catalog_rx.borrow_and_update();
 
-                            watches.push(Box::pin(async move {
-                                let res = consul.watch_node(&node, None).await;
-                                (node, res)
-                            }));
+            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);
                         }
                     }
                 }
-                Err(e) => {
-                    error!("Could not get Consul node list: {}", e);
-                }
-            }
-
-            let next_watch = select! {
-                nw = watches.next() => nw,
-                _ = must_exit.changed() => continue,
-            };
-
-            let (node, res): (String, Result<_>) = match next_watch {
-                Some(v) => v,
-                None => {
-                    warn!("No nodes currently watched in dns_config.rs");
-                    sleep(Duration::from_secs(10)).await;
-                    continue;
-                }
-            };
-
-            match res {
-                Ok((catalog, new_idx)) => {
-                    let mut watch_state = nodes.get_mut(&node).unwrap();
-                    watch_state.last_idx = Some(new_idx);
-                    watch_state.last_catalog = catalog;
-                    watch_state.retries = 0;
-
-                    let idx = watch_state.last_idx;
-                    let consul = consul.clone();
-                    watches.push(Box::pin(async move {
-                        let res = consul.watch_node(&node, idx).await;
-                        (node, res)
-                    }));
-                }
-                Err(e) => {
-                    let mut watch_state = nodes.get_mut(&node).unwrap();
-                    watch_state.retries += 1;
-                    watch_state.last_idx = None;
-
-                    let will_retry_in =
-                        retry_to_time(watch_state.retries, Duration::from_secs(600));
-                    error!(
-                        "Failed to query consul for node {}. Will retry in {}s. {}",
-                        node,
-                        will_retry_in.as_secs(),
-                        e
-                    );
-
-                    let consul = consul.clone();
-                    watches.push(Box::pin(async move {
-                        sleep(will_retry_in).await;
-                        let res = consul.watch_node(&node, None).await;
-                        (node, res)
-                    }));
-                    continue;
-                }
-            }
-
-            let mut dns_config = DnsConfig::new();
-            for (_, watch_state) in nodes.iter() {
-                if let Some(catalog) = &watch_state.last_catalog {
-                    parse_consul_catalog(catalog, &mut dns_config);
-                }
             }
 
             tx.send(Arc::new(dns_config)).expect("Internal error");
@@ -228,16 +136,6 @@ pub fn spawn_dns_config_task(
     rx
 }
 
-fn retry_to_time(retries: u32, max_time: Duration) -> Duration {
-    // 1.2^x seems to be a good value to exponentially increase time at a good pace
-    // eg. 1.2^32 = 341 seconds ~= 5 minutes - ie. after 32 retries we wait 5
-    // minutes
-    Duration::from_secs(cmp::min(
-        max_time.as_secs(),
-        1.2f64.powf(retries as f64) as u64,
-    ))
-}
-
 // ---- Display impls ----
 
 impl std::fmt::Display for DnsRecordType {