diplonat/src/consul_actor.rs

118 lines
3.1 KiB
Rust
Raw Normal View History

use std::{cmp, collections::HashSet, time::Duration};
use anyhow::Result;
use log::*;
use serde::{Deserialize, Serialize};
use serde_lexpr::{error, from_str};
use tokio::{sync::watch, time::sleep};
2021-12-30 19:42:56 +00:00
use crate::config::RuntimeConfigConsul;
use crate::{consul, messages};
2020-05-22 12:17:48 +00:00
#[derive(Serialize, Deserialize, Debug)]
pub enum DiplonatParameter {
tcp_port(HashSet<u16>),
udp_port(HashSet<u16>),
2020-05-22 12:17:48 +00:00
}
#[derive(Serialize, Deserialize, Debug)]
pub enum DiplonatConsul {
diplonat(Vec<DiplonatParameter>),
2020-05-22 12:17:48 +00:00
}
2020-05-22 09:34:12 +00:00
pub struct ConsulActor {
2020-05-22 14:27:43 +00:00
pub rx_open_ports: watch::Receiver<messages::PublicExposedPorts>,
2020-05-22 10:25:44 +00:00
2020-05-22 12:17:48 +00:00
consul: consul::Consul,
2020-05-22 10:25:44 +00:00
node: String,
retries: u32,
tx_open_ports: watch::Sender<messages::PublicExposedPorts>,
2020-05-22 10:25:44 +00:00
}
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,
));
2020-05-22 09:34:12 +00:00
}
2020-05-22 16:13:32 +00:00
fn to_parameters(catalog: &consul::CatalogNode) -> Vec<DiplonatConsul> {
let mut r = Vec::new();
2020-05-22 14:27:43 +00:00
2020-05-22 12:17:48 +00:00
for (_, service_info) in &catalog.Services {
for tag in &service_info.Tags {
let diplo_conf: error::Result<DiplonatConsul> = from_str(tag);
match diplo_conf {
2020-05-22 16:13:32 +00:00
Ok(conf) => r.push(conf),
2020-05-22 12:17:48 +00:00
Err(e) => debug!("Failed to parse entry {}. {}", tag, e),
};
}
}
2020-05-22 16:13:32 +00:00
return r;
2020-05-22 16:13:32 +00:00
}
fn to_open_ports(params: &Vec<DiplonatConsul>) -> messages::PublicExposedPorts {
let mut op = messages::PublicExposedPorts {
tcp_ports: HashSet::new(),
udp_ports: HashSet::new(),
2020-05-22 16:13:32 +00:00
};
for conf in params {
let DiplonatConsul::diplonat(c) = conf;
for parameter in c {
match parameter {
DiplonatParameter::tcp_port(p) => op.tcp_ports.extend(p),
DiplonatParameter::udp_port(p) => op.udp_ports.extend(p),
};
}
}
return op;
2020-05-22 12:17:48 +00:00
}
2020-05-22 09:34:12 +00:00
impl ConsulActor {
2021-12-30 19:42:56 +00:00
pub fn new(config: &RuntimeConfigConsul, node: &str) -> Self {
let (tx, rx) = watch::channel(messages::PublicExposedPorts {
tcp_ports: HashSet::new(),
udp_ports: HashSet::new(),
2020-05-22 14:27:43 +00:00
});
2020-05-22 09:34:12 +00:00
return Self {
2021-12-30 19:42:56 +00:00
consul: consul::Consul::new(config),
2020-05-22 10:25:44 +00:00
rx_open_ports: rx,
tx_open_ports: tx,
node: node.to_string(),
retries: 0,
};
2020-05-22 09:34:12 +00:00
}
2020-05-22 10:25:44 +00:00
2020-05-22 10:29:55 +00:00
pub async fn listen(&mut self) -> Result<()> {
2020-05-22 10:25:44 +00:00
loop {
let catalog = match self.consul.watch_node(&self.node).await {
Ok(c) => c,
Err(e) => {
2020-05-22 13:19:49 +00:00
self.consul.watch_node_reset();
2020-05-23 14:29:02 +00:00
self.retries = cmp::min(std::u32::MAX - 1, self.retries) + 1;
2020-05-22 10:25:44 +00:00
let will_retry_in = retry_to_time(self.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;
2020-05-22 10:25:44 +00:00
}
};
2020-05-22 13:19:49 +00:00
self.retries = 0;
2020-05-22 16:41:13 +00:00
let msg = to_open_ports(&to_parameters(&catalog));
debug!("Extracted configuration: {:#?}", msg);
self.tx_open_ports.send(msg)?;
2020-05-22 10:25:44 +00:00
}
}
2020-05-22 09:34:12 +00:00
}