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

This commit is contained in:
Alex 2023-02-02 16:30:00 +01:00
parent 86c255dfea
commit aaff9b7d4a
4 changed files with 27 additions and 126 deletions

6
Cargo.lock generated
View file

@ -114,15 +114,17 @@ dependencies = [
[[package]] [[package]]
name = "df-consul" name = "df-consul"
version = "0.2.0" version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a351d00f138e768845cdefb9ae27b79aeed97c698745c73bb2805cad1167aa81" checksum = "0e38cfbab431b53dfd2d09f2a9902510c636d3d7397645bac5cf1959cfde2999"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"bytes", "bytes",
"futures",
"log", "log",
"reqwest", "reqwest",
"serde", "serde",
"tokio",
] ]
[[package]] [[package]]

View file

@ -12,8 +12,9 @@ anyhow = "1.0.66"
futures = "0.3" futures = "0.3"
tracing = { version = "0.1.30" } tracing = { version = "0.1.30" }
tracing-subscriber = { version = "0.3", features = ["env-filter"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] }
df-consul = "0.2.0"
structopt = "0.3" structopt = "0.3"
tokio = { version = "1.22", default-features = false, features = ["rt", "rt-multi-thread", "io-util", "net", "time", "macros", "sync", "signal", "fs"] } tokio = { version = "1.22", default-features = false, features = ["rt", "rt-multi-thread", "io-util", "net", "time", "macros", "sync", "signal", "fs"] }
reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls-webpki-roots" ] } reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls-webpki-roots" ] }
serde = { version = "1.0.107", features = ["derive"] } serde = { version = "1.0.107", features = ["derive"] }
df-consul = "0.3.3"

View file

@ -1,14 +1,9 @@
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::fmt; use std::fmt;
use std::sync::Arc; use std::sync::Arc;
use std::{cmp, time::Duration}; use std::time::Duration;
use anyhow::Result; use tokio::{select, sync::watch};
use futures::future::BoxFuture;
use futures::stream::{FuturesUnordered, StreamExt};
use tokio::{select, sync::watch, time::sleep};
use tracing::*; use tracing::*;
use df_consul::*; 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<_>>(); let splits = tag.split(' ').collect::<Vec<_>>();
if splits.len() != 2 { if splits.len() != 2 {
return None; return None;
@ -102,122 +97,35 @@ 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( pub fn spawn_dns_config_task(
consul: Consul, consul: &Consul,
mut must_exit: watch::Receiver<bool>, mut must_exit: watch::Receiver<bool>,
) -> watch::Receiver<Arc<DnsConfig>> { ) -> watch::Receiver<Arc<DnsConfig>> {
let (tx, rx) = watch::channel(Arc::new(DnsConfig::new())); 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 { tokio::spawn(async move {
let mut nodes = HashMap::new();
let mut watches = FuturesUnordered::<BoxFuture<'static, (String, Result<_>)>>::new();
while !*must_exit.borrow() { while !*must_exit.borrow() {
let list_nodes = select! { select! {
ln = consul.list_nodes() => ln, _ = catalog_rx.changed() => (),
_ = must_exit.changed() => continue, _ = must_exit.changed() => continue,
}; };
match list_nodes { let services = catalog_rx.borrow_and_update();
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();
watches.push(Box::pin(async move {
let res = consul.watch_node(&node, None).await;
(node, res)
}));
}
}
}
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(); let mut dns_config = DnsConfig::new();
for (_, watch_state) in nodes.iter() { for (_svc, nodes) in services.iter() {
if let Some(catalog) = &watch_state.last_catalog { for node in nodes.iter() {
parse_consul_catalog(catalog, &mut dns_config); // 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);
}
}
} }
} }
@ -228,16 +136,6 @@ pub fn spawn_dns_config_task(
rx 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 ---- // ---- Display impls ----
impl std::fmt::Display for DnsRecordType { impl std::fmt::Display for DnsRecordType {

View file

@ -76,7 +76,7 @@ async fn main() {
let (exit_signal, _) = watch_ctrl_c(); let (exit_signal, _) = watch_ctrl_c();
let consul_config = df_consul::ConsulConfig { let consul_config = df_consul::Config {
addr: opt.consul_addr.clone(), addr: opt.consul_addr.clone(),
ca_cert: opt.consul_ca_cert.clone(), ca_cert: opt.consul_ca_cert.clone(),
tls_skip_verify: opt.consul_tls_skip_verify, tls_skip_verify: opt.consul_tls_skip_verify,
@ -111,7 +111,7 @@ async fn main() {
.map(ToString::to_string) .map(ToString::to_string)
.collect::<Vec<_>>(); .collect::<Vec<_>>();
let rx_dns_config = dns_config::spawn_dns_config_task(consul.clone(), exit_signal.clone()); let rx_dns_config = dns_config::spawn_dns_config_task(&consul, exit_signal.clone());
let updater_task = tokio::spawn(dns_updater::dns_updater_task( let updater_task = tokio::spawn(dns_updater::dns_updater_task(
rx_dns_config.clone(), rx_dns_config.clone(),