87 lines
2.5 KiB
Rust
87 lines
2.5 KiB
Rust
use std::cmp;
|
|
use std::time::Duration;
|
|
use log::*;
|
|
use tokio::sync::watch;
|
|
use tokio::time::delay_for;
|
|
use anyhow::Result;
|
|
use serde::{Serialize, Deserialize};
|
|
use serde_lexpr::{from_str,to_string,error};
|
|
use crate::messages;
|
|
use crate::consul;
|
|
|
|
#[derive(Serialize, Deserialize, Debug)]
|
|
pub enum DiplonatParameter {
|
|
port(Vec<u16>)
|
|
}
|
|
|
|
#[derive(Serialize, Deserialize, Debug)]
|
|
pub enum DiplonatConsul {
|
|
diplonat(Vec<DiplonatParameter>)
|
|
}
|
|
|
|
pub struct ConsulActor {
|
|
pub rx_open_ports: watch::Receiver<messages::OpenPorts>,
|
|
|
|
consul: consul::Consul,
|
|
node: String,
|
|
retries: u32,
|
|
tx_open_ports: watch::Sender<messages::OpenPorts>
|
|
}
|
|
|
|
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 from_catalog_to_open_ports(catalog: &consul::CatalogNode) -> messages::OpenPorts {
|
|
let mut op = messages::OpenPorts { ports: Vec::new() };
|
|
for (_, service_info) in &catalog.Services {
|
|
for tag in &service_info.Tags {
|
|
let diplo_conf: error::Result<DiplonatConsul> = from_str(tag);
|
|
match diplo_conf {
|
|
Ok(conf) => {
|
|
let DiplonatConsul::diplonat(c) = conf;
|
|
for parameter in &c {
|
|
let DiplonatParameter::port(p) = parameter;
|
|
op.ports.extend(p);
|
|
}
|
|
}
|
|
Err(e) => debug!("Failed to parse entry {}. {}", tag, e),
|
|
};
|
|
}
|
|
}
|
|
return op;
|
|
}
|
|
|
|
impl ConsulActor {
|
|
pub fn new(url: &str, node: &str) -> Self {
|
|
let (tx, rx) = watch::channel(messages::OpenPorts{ports: Vec::new() });
|
|
return Self {
|
|
consul: consul::Consul::new(url),
|
|
rx_open_ports: rx,
|
|
tx_open_ports: tx,
|
|
node: node.to_string(),
|
|
retries: 0,
|
|
};
|
|
}
|
|
|
|
pub async fn listen(&mut self) -> Result<()> {
|
|
loop {
|
|
let catalog = match self.consul.watch_node(&self.node).await {
|
|
Ok(c) => c,
|
|
Err(e) => {
|
|
self.consul.watch_node_reset();
|
|
self.retries = cmp::min(u32::MAX - 1, self.retries) + 1;
|
|
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);
|
|
delay_for(will_retry_in).await;
|
|
continue;
|
|
}
|
|
};
|
|
self.retries = 0;
|
|
info!("{:#?}", from_catalog_to_open_ports(&catalog));
|
|
}
|
|
}
|
|
}
|