commit
2895a8ae2e
@ -0,0 +1 @@ |
||||
/target |
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,19 @@ |
||||
[package] |
||||
name = "tricot" |
||||
version = "0.1.0" |
||||
edition = "2018" |
||||
authors = ["Alex Auvolat <alex@adnab.me>"] |
||||
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html |
||||
|
||||
[dependencies] |
||||
anyhow = "1.0.28" |
||||
envy = "0.4" |
||||
futures = "0.3.5" |
||||
log = "0.4" |
||||
pretty_env_logger = "0.4" |
||||
regex = "1" |
||||
reqwest = { version = "0.11", features = ["json"] } |
||||
serde = { version = "1.0.107", features = ["derive"] } |
||||
serde_json = "1.0.53" |
||||
tokio = { version = "1.0", default-features = false, features = ["rt", "rt-multi-thread", "io-util", "net", "time", "macros", "sync", "signal", "fs"] } |
@ -0,0 +1 @@ |
||||
hard_tabs = true |
@ -0,0 +1,62 @@ |
||||
use std::collections::HashMap; |
||||
|
||||
use anyhow::Result; |
||||
use log::*; |
||||
use serde::{Deserialize, Serialize}; |
||||
|
||||
// ---- Watch and retrieve Consul catalog ----
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug)] |
||||
pub struct ConsulServiceEntry { |
||||
#[serde(rename = "Address")] |
||||
pub address: String, |
||||
|
||||
#[serde(rename = "Port")] |
||||
pub port: u16, |
||||
|
||||
#[serde(rename = "Tags")] |
||||
pub tags: Vec<String>, |
||||
} |
||||
|
||||
#[derive(Serialize, Deserialize, Debug)] |
||||
pub struct ConsulNodeCatalog { |
||||
#[serde(rename = "Services")] |
||||
pub services: HashMap<String, ConsulServiceEntry>, |
||||
} |
||||
|
||||
#[derive(Clone)] |
||||
pub struct Consul { |
||||
client: reqwest::Client, |
||||
url: String, |
||||
idx: Option<u64>, |
||||
} |
||||
|
||||
impl Consul { |
||||
pub fn new(url: &str) -> Self { |
||||
return Self { |
||||
client: reqwest::Client::new(), |
||||
url: url.to_string(), |
||||
idx: None, |
||||
}; |
||||
} |
||||
|
||||
pub fn watch_node_reset(&mut self) -> () { |
||||
self.idx = None; |
||||
} |
||||
|
||||
pub async fn watch_node(&mut self, host: &str) -> Result<ConsulNodeCatalog> { |
||||
let url = match self.idx { |
||||
Some(i) => format!("{}/v1/catalog/node/{}?index={}", self.url, host, i), |
||||
None => format!("{}/v1/catalog/node/{}", self.url, host), |
||||
}; |
||||
|
||||
let http = self.client.get(&url).send().await?; |
||||
self.idx = match http.headers().get("X-Consul-Index") { |
||||
Some(v) => Some(v.to_str()?.parse::<u64>()?), |
||||
None => return Err(anyhow!("X-Consul-Index header not found")), |
||||
}; |
||||
|
||||
let resp: ConsulNodeCatalog = http.json().await?; |
||||
return Ok(resp); |
||||
} |
||||
} |
@ -0,0 +1,20 @@ |
||||
#[macro_use] |
||||
extern crate anyhow; |
||||
|
||||
mod consul; |
||||
mod proxy_config; |
||||
|
||||
use log::*; |
||||
|
||||
#[tokio::main] |
||||
async fn main() { |
||||
pretty_env_logger::init(); |
||||
info!("Starting Tricot"); |
||||
|
||||
let consul = consul::Consul::new("http://10.42.0.21:8500"); |
||||
let mut rx_proxy_config = proxy_config::spawn_proxy_config_task(consul.clone(), "carcajou"); |
||||
|
||||
while rx_proxy_config.changed().await.is_ok() { |
||||
info!("Proxy config: {:#?}", *rx_proxy_config.borrow()); |
||||
} |
||||
} |
@ -0,0 +1,113 @@ |
||||
use std::net::SocketAddr; |
||||
use std::{cmp, time::Duration}; |
||||
|
||||
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<String>, |
||||
pub priority: u32, |
||||
} |
||||
|
||||
#[derive(Debug)] |
||||
pub struct ProxyConfig { |
||||
pub entries: Vec<ProxyEntry>, |
||||
} |
||||
|
||||
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<ProxyEntry> { |
||||
let splits = tag.split(' ').collect::<Vec<_>>(); |
||||
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, |
||||
}) |
||||
} |
||||
|
||||
fn parse_consul_catalog(catalog: &ConsulNodeCatalog) -> ProxyConfig { |
||||
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); |
||||
} |
||||
} |
||||
} |
||||
|
||||
ProxyConfig { entries } |
||||
} |
||||
|
||||
pub fn spawn_proxy_config_task(mut consul: Consul, node: &str) -> watch::Receiver<ProxyConfig> { |
||||
let (tx, rx) = watch::channel(ProxyConfig { |
||||
entries: Vec::new(), |
||||
}); |
||||
|
||||
let node = node.to_string(); |
||||
|
||||
tokio::spawn(async move { |
||||
let mut retries = 0; |
||||
|
||||
loop { |
||||
let catalog = match consul.watch_node(&node).await { |
||||
Ok(c) => c, |
||||
Err(e) => { |
||||
consul.watch_node_reset(); |
||||
retries = cmp::min(std::u32::MAX - 1, retries) + 1; |
||||
let will_retry_in = retry_to_time(retries, Duration::from_secs(600)); |
||||
error!( |
||||
"Failed to query consul. Will retry in {}s. {}", |
||||
will_retry_in.as_secs(), |
||||
e |
||||
); |
||||
sleep(will_retry_in).await; |
||||
continue; |
||||
} |
||||
}; |
||||
retries = 0; |
||||
|
||||
let config = parse_consul_catalog(&catalog); |
||||
debug!("Extracted configuration: {:#?}", config); |
||||
|
||||
tx.send(config).expect("Internal error"); |
||||
} |
||||
}); |
||||
|
||||
rx |
||||
} |
Loading…
Reference in new issue