aboutsummaryrefslogtreecommitdiff
path: root/src/dns_config.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/dns_config.rs')
-rw-r--r--src/dns_config.rs138
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 {