390 lines
9.3 KiB
Rust
390 lines
9.3 KiB
Rust
use std::collections::HashMap;
|
|
use std::net::SocketAddr;
|
|
use std::sync::{atomic, Arc};
|
|
use std::time::Duration;
|
|
|
|
use anyhow::Result;
|
|
use opentelemetry::{metrics, KeyValue};
|
|
|
|
use tokio::{select, sync::watch};
|
|
use tracing::*;
|
|
|
|
use crate::consul;
|
|
|
|
// ---- Extract proxy config from Consul catalog ----
|
|
|
|
#[derive(Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
|
|
pub enum HostDescription {
|
|
Hostname(String),
|
|
Pattern(glob::Pattern),
|
|
}
|
|
|
|
impl HostDescription {
|
|
fn new(desc: &str) -> Result<Self> {
|
|
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, "[{}]", 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<String>,
|
|
/// 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,
|
|
}
|
|
|
|
impl PartialEq for ProxyEntry {
|
|
fn eq(&self, other: &Self) -> bool {
|
|
self.host == other.host
|
|
&& self.path_prefix == other.path_prefix
|
|
&& self.priority == other.priority
|
|
&& self.service_name == other.service_name
|
|
&& self.target_addr == other.target_addr
|
|
&& self.https_target == other.https_target
|
|
&& self.flags == other.flags
|
|
&& self.add_headers == other.add_headers
|
|
}
|
|
}
|
|
impl Eq for ProxyEntry {}
|
|
|
|
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
|
|
pub struct ProxyEntryFlags {
|
|
/// Is the target healthy?
|
|
pub healthy: bool,
|
|
|
|
/// 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.healthy {
|
|
write!(f, " UNHEALTHY")?;
|
|
}
|
|
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, PartialEq, Eq)]
|
|
pub struct ProxyConfig {
|
|
pub entries: Vec<ProxyEntry>,
|
|
}
|
|
|
|
fn parse_tricot_tag(
|
|
service_name: String,
|
|
tag: &str,
|
|
target_addr: SocketAddr,
|
|
add_headers: &[(String, String)],
|
|
flags: ProxyEntryFlags,
|
|
) -> Option<ProxyEntry> {
|
|
let splits = tag.split(' ').collect::<Vec<_>>();
|
|
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::<Vec<_>>();
|
|
if splits.len() == 3 && splits[0] == "tricot-add-header" {
|
|
Some((splits[1].to_string(), splits[2].to_string()))
|
|
} else {
|
|
None
|
|
}
|
|
}
|
|
|
|
fn parse_consul_service(
|
|
s: &consul::catalog::HealthServiceNode,
|
|
mut flags: ProxyEntryFlags,
|
|
) -> Vec<ProxyEntry> {
|
|
trace!("Parsing service: {:#?}", s);
|
|
|
|
let mut entries = vec![];
|
|
|
|
let ip_addr = match s.service.address.parse() {
|
|
Ok(ip) => ip,
|
|
_ => match s.node.address.parse() {
|
|
Ok(ip) => ip,
|
|
_ => {
|
|
warn!(
|
|
"Could not get address for service {} at node {}",
|
|
s.service.service, s.node.node
|
|
);
|
|
return vec![];
|
|
}
|
|
},
|
|
};
|
|
let addr = SocketAddr::new(ip_addr, s.service.port);
|
|
|
|
if s.service.tags.contains(&"tricot-global-lb".into()) {
|
|
flags.global_lb = true;
|
|
} else if s.service.tags.contains(&"tricot-site-lb".into()) {
|
|
flags.site_lb = true;
|
|
};
|
|
|
|
let mut add_headers = vec![];
|
|
for tag in s.service.tags.iter() {
|
|
if let Some(pair) = parse_tricot_add_header_tag(tag) {
|
|
add_headers.push(pair);
|
|
}
|
|
}
|
|
|
|
for tag in s.service.tags.iter() {
|
|
if let Some(ent) = parse_tricot_tag(
|
|
s.service.service.clone(),
|
|
tag,
|
|
addr,
|
|
&add_headers[..],
|
|
flags,
|
|
) {
|
|
entries.push(ent);
|
|
}
|
|
}
|
|
|
|
trace!("Result of parsing service:");
|
|
for ent in entries.iter() {
|
|
trace!(" {}", ent);
|
|
}
|
|
|
|
entries
|
|
}
|
|
|
|
pub fn spawn_proxy_config_task(
|
|
consul: consul::Consul,
|
|
local_node: String,
|
|
mut must_exit: watch::Receiver<bool>,
|
|
) -> watch::Receiver<Arc<ProxyConfig>> {
|
|
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 catalog_rx = consul.watch_all_service_health(Duration::from_secs(300));
|
|
let mut local_node_site = None;
|
|
|
|
while !*must_exit.borrow() {
|
|
select! {
|
|
_ = catalog_rx.changed() => (),
|
|
_ = must_exit.changed() => continue,
|
|
};
|
|
|
|
let services = catalog_rx.borrow_and_update().clone();
|
|
if local_node_site.is_none() {
|
|
for (_, svcnodes) in services.iter() {
|
|
for svcnode in svcnodes.iter() {
|
|
if svcnode.node.node == local_node {
|
|
if let Some(site) = svcnode.node.meta.get("site") {
|
|
local_node_site = Some(site.to_string());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut entries = vec![];
|
|
|
|
for (_service, svcnodes) in services.iter() {
|
|
for svcnode in svcnodes.iter() {
|
|
let healthy = !svcnode.checks.iter().any(|x| x.status == "critical");
|
|
|
|
let same_node = svcnode.node.node == local_node;
|
|
let same_site = match (svcnode.node.meta.get("site"), local_node_site.as_ref())
|
|
{
|
|
(Some(s1), Some(s2)) => s1 == s2,
|
|
_ => false,
|
|
};
|
|
|
|
let flags = ProxyEntryFlags {
|
|
healthy,
|
|
same_node,
|
|
same_site,
|
|
site_lb: false,
|
|
global_lb: false,
|
|
};
|
|
|
|
entries.extend(parse_consul_service(&svcnode, flags));
|
|
}
|
|
}
|
|
|
|
entries.sort_by_cached_key(|ent| ent.to_string());
|
|
|
|
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::ObservableGauge<u64>,
|
|
}
|
|
|
|
impl ProxyConfigMetrics {
|
|
fn new(rx: watch::Receiver<Arc<ProxyConfig>>) -> Self {
|
|
let meter = opentelemetry::global::meter("tricot");
|
|
Self {
|
|
_proxy_config_entries: meter
|
|
.u64_observable_gauge("proxy_config_entries")
|
|
.with_callback(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")
|
|
}
|
|
}
|
|
}
|