use std::net::SocketAddr; use std::sync::{atomic, Arc}; use std::collections::HashMap; use std::{cmp, time::Duration}; use anyhow::Result; use futures::stream::{FuturesUnordered, StreamExt}; use futures::future::{BoxFuture}; use log::*; use tokio::{sync::watch, time::sleep}; use crate::consul::*; // ---- Extract proxy config from Consul catalog ---- #[derive(Debug)] pub struct ProxyEntry { pub target_addr: SocketAddr, pub host: String, pub path_prefix: Option, pub priority: u32, // Counts the number of times this proxy server has been called to // This implements a round-robin load balancer if there are multiple // entries for the same host and same path prefix. pub calls: atomic::AtomicU64, } #[derive(Debug)] pub struct ProxyConfig { pub entries: Vec, } 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 return Duration::from_secs(cmp::min( max_time.as_secs(), 1.2f64.powf(retries as f64) as u64, )); } fn parse_tricot_tag(target_addr: SocketAddr, tag: &str) -> Option { let splits = tag.split(' ').collect::>(); if (splits.len() != 2 && splits.len() != 3) || splits[0] != "tricot" { return None; } let (host, path_prefix) = match splits[1].split_once('/') { Some((h, p)) => (h, Some(p.to_string())), None => (splits[1], None), }; let priority = match splits.len() { 3 => splits[2].parse().ok()?, _ => 100, }; Some(ProxyEntry { target_addr, host: host.to_string(), path_prefix, priority, calls: atomic::AtomicU64::from(0), }) } fn parse_consul_catalog(catalog: &ConsulNodeCatalog) -> Vec { let mut entries = vec![]; for (_, svc) in catalog.services.iter() { let ip_addr = match svc.address.parse() { Ok(ip) => ip, _ => continue, }; let addr = SocketAddr::new(ip_addr, svc.port); for tag in svc.tags.iter() { if let Some(ent) = parse_tricot_tag(addr, tag) { entries.push(ent); } } } entries } #[derive(Default)] struct NodeWatchState { last_idx: Option, last_catalog: Option, retries: u32, } pub fn spawn_proxy_config_task(consul: Consul) -> watch::Receiver> { let (tx, rx) = watch::channel(Arc::new(ProxyConfig { entries: Vec::new(), })); let consul = Arc::new(consul); tokio::spawn(async move { let mut nodes = HashMap::new(); let mut watches = FuturesUnordered::)>>::new(); loop { match consul.list_nodes().await { Ok(consul_nodes) => { info!("Watched consul nodes: {:?}", consul_nodes); for node in consul_nodes { 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) => { warn!("Could not get Consul node list: {}", e); } } let (node, res): (String, Result<_>) = match watches.next().await { Some(v) => v, None => { warn!("No nodes currently watched in proxy_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 = Some(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. Will retry in {}s. {}", 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 entries = vec![]; for (_, watch_state) in nodes.iter() { if let Some(catalog) = &watch_state.last_catalog { entries.extend(parse_consul_catalog(catalog)); } } let config = ProxyConfig { entries }; debug!("Extracted configuration: {:#?}", config); tx.send(Arc::new(config)).expect("Internal error"); } }); rx }