aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/consul.rs6
-rw-r--r--src/https.rs2
-rw-r--r--src/proxy_config.rs59
3 files changed, 56 insertions, 11 deletions
diff --git a/src/consul.rs b/src/consul.rs
index 14df1ee..cba435a 100644
--- a/src/consul.rs
+++ b/src/consul.rs
@@ -23,6 +23,8 @@ pub struct ConsulNode {
pub node: String,
#[serde(rename = "Address")]
pub address: String,
+ #[serde(rename = "Meta")]
+ pub meta: HashMap<String, String>,
}
#[derive(Serialize, Deserialize, Debug)]
@@ -117,14 +119,14 @@ impl Consul {
})
}
- pub async fn list_nodes(&self) -> Result<Vec<String>> {
+ pub async fn list_nodes(&self) -> Result<Vec<ConsulNode>> {
debug!("list_nodes");
let url = format!("{}/v1/catalog/nodes", self.url);
let http = self.client.get(&url).send().await?;
let resp: Vec<ConsulNode> = http.json().await?;
- Ok(resp.into_iter().map(|n| n.node).collect::<Vec<_>>())
+ Ok(resp)
}
pub async fn watch_node(
diff --git a/src/https.rs b/src/https.rs
index a9f2cc1..34e3f85 100644
--- a/src/https.rs
+++ b/src/https.rs
@@ -137,6 +137,8 @@ async fn handle(
.as_ref()
.map(|x| x.len() as i32)
.unwrap_or(0),
+ ent.same_node,
+ ent.same_site,
-ent.calls.load(Ordering::SeqCst),
)
});
diff --git a/src/proxy_config.rs b/src/proxy_config.rs
index deb2ffe..a64e5aa 100644
--- a/src/proxy_config.rs
+++ b/src/proxy_config.rs
@@ -40,12 +40,27 @@ impl HostDescription {
#[derive(Debug)]
pub struct ProxyEntry {
- pub target_addr: SocketAddr,
- pub https_target: bool,
-
+ /// 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,
+
+ /// 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,
+
+ /// 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,
+
+ /// Add the following headers to all responses returned
+ /// when matching this rule
pub add_headers: Vec<(String, String)>,
// Counts the number of times this proxy server has been called to
@@ -70,6 +85,11 @@ impl std::fmt::Display for ProxyEntry {
self.path_prefix.as_ref().unwrap_or(&String::new()),
self.priority
)?;
+ if self.same_node {
+ write!(f, " OURSELF")?;
+ } else if self.same_site {
+ write!(f, " SAME_SITE")?;
+ }
if !self.add_headers.is_empty() {
write!(f, " +Headers: {:?}", self.add_headers)?;
}
@@ -96,6 +116,8 @@ fn parse_tricot_tag(
tag: &str,
target_addr: SocketAddr,
add_headers: &[(String, String)],
+ same_node: bool,
+ same_site: bool,
) -> Option<ProxyEntry> {
let splits = tag.split(' ').collect::<Vec<_>>();
if (splits.len() != 2 && splits.len() != 3)
@@ -129,6 +151,8 @@ fn parse_tricot_tag(
target_addr,
https_target: (splits[0] == "tricot-https"),
host,
+ same_node,
+ same_site,
path_prefix,
priority,
add_headers: add_headers.to_vec(),
@@ -145,7 +169,11 @@ fn parse_tricot_add_header_tag(tag: &str) -> Option<(String, String)> {
}
}
-fn parse_consul_catalog(catalog: &ConsulNodeCatalog) -> Vec<ProxyEntry> {
+fn parse_consul_catalog(
+ catalog: &ConsulNodeCatalog,
+ same_node: bool,
+ same_site: bool,
+) -> Vec<ProxyEntry> {
trace!("Parsing node catalog: {:#?}", catalog);
let mut entries = vec![];
@@ -174,7 +202,7 @@ fn parse_consul_catalog(catalog: &ConsulNodeCatalog) -> Vec<ProxyEntry> {
}
for tag in svc.tags.iter() {
- if let Some(ent) = parse_tricot_tag(tag, addr, &add_headers[..]) {
+ if let Some(ent) = parse_tricot_tag(tag, addr, &add_headers[..], same_node, same_site) {
entries.push(ent);
}
}
@@ -206,12 +234,15 @@ pub fn spawn_proxy_config_task(consul: Consul) -> watch::Receiver<Arc<ProxyConfi
let mut nodes = HashMap::new();
let mut watches = FuturesUnordered::<BoxFuture<'static, (String, Result<_>)>>::new();
+ let mut node_site = HashMap::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) {
+ 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();
@@ -222,6 +253,9 @@ pub fn spawn_proxy_config_task(consul: Consul) -> watch::Receiver<Arc<ProxyConfi
(node, res)
}));
}
+ if let Some(site) = consul_node.meta.get("site") {
+ node_site.insert(node.clone(), site.clone());
+ }
}
}
Err(e) => {
@@ -277,9 +311,16 @@ pub fn spawn_proxy_config_task(consul: Consul) -> watch::Receiver<Arc<ProxyConfi
}
let mut entries = vec![];
- for (_, watch_state) in nodes.iter() {
+ for (node_name, watch_state) in nodes.iter() {
if let Some(catalog) = &watch_state.last_catalog {
- entries.extend(parse_consul_catalog(catalog));
+ let same_node = *node_name == consul.local_node;
+ let same_site =
+ match (node_site.get(node_name), node_site.get(&consul.local_node)) {
+ (Some(s1), Some(s2)) => s1 == s2,
+ _ => false,
+ };
+
+ entries.extend(parse_consul_catalog(catalog, same_node, same_site));
}
}
let config = ProxyConfig { entries };