use std::collections::HashMap; use std::net::SocketAddr; use std::sync::{atomic, Arc}; use std::{cmp, time::Duration}; use anyhow::Result; use opentelemetry::{metrics, KeyValue}; use futures::future::BoxFuture; use futures::stream::{FuturesUnordered, StreamExt}; use log::*; use tokio::{select, sync::watch, time::sleep}; use crate::consul::*; // ---- Extract proxy config from Consul catalog ---- #[derive(Debug)] pub enum HostDescription { Hostname(String), Pattern(glob::Pattern), } impl HostDescription { fn new(desc: &str) -> Result { if desc.chars().any(|x| matches!(x, '*' | '?' | '[' | ']')) { Ok(Self::Pattern(glob::Pattern::new(desc)?)) } else { Ok(Self::Hostname(desc.to_string())) } } pub fn matches(&self, v: &str) -> bool { match self { Self::Pattern(p) => p.matches(v), Self::Hostname(s) => s == v, } } } impl std::fmt::Display for HostDescription { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { HostDescription::Hostname(h) => write!(f, "{}", h), HostDescription::Pattern(p) => write!(f, "Pattern('{}')", p.as_str()), } } } #[derive(Debug)] pub struct ProxyEntry { /// Publicly exposed TLS hostnames for matching this rule pub host: HostDescription, /// Path prefix for matching this rule pub path_prefix: Option, /// Priority with which this rule is considered (highest first) pub priority: u32, /// Consul service name pub service_name: String, /// Node address (ip+port) to handle requests that match this entry pub target_addr: SocketAddr, /// Is the target serving HTTPS instead of HTTP? pub https_target: bool, /// Flags for target selection pub flags: ProxyEntryFlags, /// Add the following headers to all responses returned /// when matching this rule pub add_headers: Vec<(String, String)>, /// Number of calls in progress, used to deprioritize slow back-ends pub calls_in_progress: atomic::AtomicI64, /// Time of last call, used for round-robin selection pub last_call: atomic::AtomicI64, } #[derive(Debug, Clone, Copy)] pub struct ProxyEntryFlags { /// Is the target the same node as we are running on? /// (if yes priorize it over other matching targets) pub same_node: bool, /// Is the target the same site as this node? /// (if yes priorize it over other matching targets) pub same_site: bool, /// Is site-wide load balancing enabled for this service? pub site_lb: bool, /// Is global load balancing enabled for this service? pub global_lb: bool, } impl std::fmt::Display for ProxyEntry { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { if self.https_target { write!(f, "https://")?; } write!(f, "{} ", self.target_addr)?; write!( f, "{}{} {}", self.host, self.path_prefix.as_deref().unwrap_or_default(), self.priority )?; if self.flags.same_node { write!(f, " OURSELF")?; } else if self.flags.same_site { write!(f, " SAME_SITE")?; } if self.flags.global_lb { write!(f, " GLOBAL-LB")?; } else if self.flags.site_lb { write!(f, " SITE-LB")?; } if !self.add_headers.is_empty() { write!(f, " +Headers: {:?}", self.add_headers)?; } Ok(()) } } #[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 Duration::from_secs(cmp::min( max_time.as_secs(), 1.2f64.powf(retries as f64) as u64, )) } fn parse_tricot_tag( service_name: String, tag: &str, target_addr: SocketAddr, add_headers: &[(String, String)], flags: ProxyEntryFlags, ) -> Option { let splits = tag.split(' ').collect::>(); if (splits.len() != 2 && splits.len() != 3) || (splits[0] != "tricot" && splits[0] != "tricot-https") { return None; } let (host, path_prefix) = match splits[1].find('/') { Some(i) => { let (host, pp) = splits[1].split_at(i); (host, Some(pp.to_string())) } None => (splits[1], None), }; let priority = match splits.len() { 3 => splits[2].parse().ok()?, _ => 100, }; let host = match HostDescription::new(host) { Ok(h) => h, Err(e) => { warn!("Invalid hostname pattern {}: {}", host, e); return None; } }; Some(ProxyEntry { service_name, target_addr, https_target: (splits[0] == "tricot-https"), host, flags, path_prefix, priority, add_headers: add_headers.to_vec(), last_call: atomic::AtomicI64::from(0), calls_in_progress: atomic::AtomicI64::from(0), }) } fn parse_tricot_add_header_tag(tag: &str) -> Option<(String, String)> { let splits = tag.splitn(3, ' ').collect::>(); if splits.len() == 3 && splits[0] == "tricot-add-header" { Some((splits[1].to_string(), splits[2].to_string())) } else { None } } fn parse_consul_catalog( catalog: &ConsulNodeCatalog, same_node: bool, same_site: bool, ) -> Vec { trace!("Parsing node catalog: {:#?}", catalog); let mut entries = vec![]; for (_, svc) in catalog.services.iter() { let ip_addr = match svc.address.parse() { Ok(ip) => ip, _ => match catalog.node.address.parse() { Ok(ip) => ip, _ => { warn!( "Could not get address for service {} at node {}", svc.service, catalog.node.node ); continue; } }, }; let addr = SocketAddr::new(ip_addr, svc.port); let (site_lb, global_lb) = if svc.tags.contains(&"tricot-global-lb".into()) { (false, true) } else if svc.tags.contains(&"tricot-site-lb".into()) { (true, false) } else { (false, false) }; let flags = ProxyEntryFlags { same_node, same_site, site_lb, global_lb, }; let mut add_headers = vec![]; for tag in svc.tags.iter() { if let Some(pair) = parse_tricot_add_header_tag(tag) { add_headers.push(pair); } } for tag in svc.tags.iter() { if let Some(ent) = parse_tricot_tag(svc.service.clone(), tag, addr, &add_headers[..], flags) { entries.push(ent); } } } trace!("Result of parsing catalog:"); for ent in entries.iter() { trace!(" {}", ent); } entries } #[derive(Default)] struct NodeWatchState { last_idx: Option, last_catalog: Option, retries: u32, } pub fn spawn_proxy_config_task( consul: Consul, local_node: String, mut must_exit: watch::Receiver, ) -> watch::Receiver> { let (tx, rx) = watch::channel(Arc::new(ProxyConfig { entries: Vec::new(), })); let metrics = ProxyConfigMetrics::new(rx.clone()); let consul = Arc::new(consul); tokio::spawn(async move { let mut nodes = HashMap::new(); let mut watches = FuturesUnordered::)>>::new(); let mut node_site = HashMap::new(); while !*must_exit.borrow() { let list_nodes = select! { ln = consul.list_nodes() => ln, _ = must_exit.changed() => continue, }; match list_nodes { Ok(consul_nodes) => { info!("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) })); } if let Some(site) = consul_node.meta.get("site") { node_site.insert(node.clone(), site.clone()); } } } Err(e) => { warn!("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 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 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 entries = vec![]; for (node_name, watch_state) in nodes.iter() { if let Some(catalog) = &watch_state.last_catalog { let same_node = *node_name == local_node; let same_site = match (node_site.get(node_name), node_site.get(&local_node)) { (Some(s1), Some(s2)) => s1 == s2, _ => false, }; entries.extend(parse_consul_catalog(catalog, same_node, same_site)); } } let config = ProxyConfig { entries }; tx.send(Arc::new(config)).expect("Internal error"); } drop(metrics); // ensure Metrics lives up to here }); rx } // ---- struct ProxyConfigMetrics { _proxy_config_entries: metrics::ValueObserver, } impl ProxyConfigMetrics { fn new(rx: watch::Receiver>) -> Self { let meter = opentelemetry::global::meter("tricot"); Self { _proxy_config_entries: meter .u64_value_observer("proxy_config_entries", move |observer| { let mut patterns = HashMap::new(); for ent in rx.borrow().entries.iter() { let attrs = ( ent.host.to_string(), ent.path_prefix.clone().unwrap_or_default(), ent.service_name.clone(), ); *patterns.entry(attrs).or_default() += 1; } for ((host, prefix, svc), num) in patterns { observer.observe( num, &[ KeyValue::new("host", host), KeyValue::new("path_prefix", prefix), KeyValue::new("service", svc), ], ); } }) .with_description("Number of proxy entries (back-ends) configured in Tricot") .init(), } } } // ---- #[cfg(test)] mod tests { use super::*; #[test] fn test_parse_tricot_add_header_tag() { match parse_tricot_add_header_tag("tricot-add-header Content-Security-Policy default-src 'none'; img-src 'self'; script-src 'self'; style-src 'self'") { Some((name, value)) => { assert_eq!(name, "Content-Security-Policy"); assert_eq!(value, "default-src 'none'; img-src 'self'; script-src 'self'; style-src 'self'"); } _ => panic!("Passed a valid tag but the function says it is not valid") } } }