Temporarily moved healthcheck from validator to separate common crate
This commit is contained in:
Jedrzej Stuczynski
@@ -3,19 +3,5 @@ use serde_derive::Deserialize;
|
||||
#[derive(Deserialize, Debug)]
|
||||
pub struct Config {
|
||||
#[serde(rename(deserialize = "healthcheck"))]
|
||||
pub health_check: HealthCheck,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug)]
|
||||
pub struct HealthCheck {
|
||||
#[serde(rename(deserialize = "directory-server"))]
|
||||
pub directory_server: String,
|
||||
|
||||
pub interval: f64, // in seconds
|
||||
|
||||
#[serde(rename(deserialize = "resolution-timeout"))]
|
||||
pub resolution_timeout: f64, // in seconds
|
||||
|
||||
#[serde(rename(deserialize = "test-packets-per-node"))]
|
||||
pub num_test_packets: usize,
|
||||
pub health_check: healthcheck::config::HealthCheck,
|
||||
}
|
||||
|
||||
@@ -1,84 +0,0 @@
|
||||
use crate::validator::config;
|
||||
use crate::validator::health_check::result::HealthCheckResult;
|
||||
use directory_client::requests::presence_topology_get::PresenceTopologyGetRequester;
|
||||
use directory_client::DirectoryClient;
|
||||
use log::{debug, error, info, trace};
|
||||
use std::time::Duration;
|
||||
use topology::NymTopologyError;
|
||||
|
||||
mod path_check;
|
||||
mod result;
|
||||
mod score;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum HealthCheckerError {
|
||||
FailedToObtainTopologyError,
|
||||
InvalidTopologyError,
|
||||
}
|
||||
|
||||
impl From<topology::NymTopologyError> for HealthCheckerError {
|
||||
fn from(_: NymTopologyError) -> Self {
|
||||
use HealthCheckerError::*;
|
||||
InvalidTopologyError
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct HealthChecker {
|
||||
directory_client: directory_client::Client,
|
||||
interval: Duration,
|
||||
num_test_packets: usize,
|
||||
resolution_timeout: Duration,
|
||||
}
|
||||
|
||||
impl HealthChecker {
|
||||
pub fn new(config: config::HealthCheck) -> Self {
|
||||
debug!(
|
||||
"healthcheck will be using the following directory server: {:?}",
|
||||
config.directory_server
|
||||
);
|
||||
let directory_client_config = directory_client::Config::new(config.directory_server);
|
||||
HealthChecker {
|
||||
directory_client: directory_client::Client::new(directory_client_config),
|
||||
interval: Duration::from_secs_f64(config.interval),
|
||||
resolution_timeout: Duration::from_secs_f64(config.resolution_timeout),
|
||||
num_test_packets: config.num_test_packets,
|
||||
}
|
||||
}
|
||||
|
||||
async fn do_check(&self) -> Result<HealthCheckResult, HealthCheckerError> {
|
||||
trace!("going to perform a healthcheck!");
|
||||
let current_topology = match self.directory_client.presence_topology.get() {
|
||||
Ok(topology) => topology,
|
||||
Err(err) => {
|
||||
error!("failed to obtain topology - {:?}", err);
|
||||
return Err(HealthCheckerError::FailedToObtainTopologyError);
|
||||
}
|
||||
};
|
||||
trace!("current topology: {:?}", current_topology);
|
||||
|
||||
let mut healthcheck_result = HealthCheckResult::calculate(
|
||||
current_topology,
|
||||
self.num_test_packets,
|
||||
self.resolution_timeout,
|
||||
)
|
||||
.await;
|
||||
healthcheck_result.sort_scores();
|
||||
Ok(healthcheck_result)
|
||||
}
|
||||
|
||||
pub async fn run(self) -> Result<(), HealthCheckerError> {
|
||||
debug!(
|
||||
"healthcheck will run every {:?} and will send {} packets to each node",
|
||||
self.interval, self.num_test_packets
|
||||
);
|
||||
|
||||
loop {
|
||||
match self.do_check().await {
|
||||
Ok(health) => info!("current network health: \n{}", health),
|
||||
Err(err) => error!("failed to perform healthcheck - {:?}", err),
|
||||
};
|
||||
|
||||
tokio::time::delay_for(self.interval).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,237 +0,0 @@
|
||||
use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair, MixnetIdentityPublicKey};
|
||||
use itertools::Itertools;
|
||||
use log::{debug, error, warn};
|
||||
use mix_client::MixClient;
|
||||
use provider_client::ProviderClient;
|
||||
use sphinx::header::delays::Delay;
|
||||
use sphinx::route::{Destination, Node as SphinxNode};
|
||||
use std::collections::HashMap;
|
||||
use topology::MixProviderNode;
|
||||
|
||||
#[derive(Debug, PartialEq, Clone)]
|
||||
pub enum PathStatus {
|
||||
Healthy,
|
||||
Unhealthy,
|
||||
Pending,
|
||||
}
|
||||
|
||||
pub(crate) struct PathChecker {
|
||||
provider_clients: HashMap<[u8; 32], Option<ProviderClient>>,
|
||||
// currently this is an overkill as MixClient is extremely cheap to create,
|
||||
// however, once we introduce persistent connection between client and layer one mixes,
|
||||
// this will be extremely helpful to have
|
||||
layer_one_clients: HashMap<[u8; 32], Option<MixClient>>,
|
||||
paths_status: HashMap<Vec<u8>, PathStatus>,
|
||||
our_destination: Destination,
|
||||
}
|
||||
|
||||
impl PathChecker {
|
||||
pub(crate) async fn new(
|
||||
providers: Vec<MixProviderNode>,
|
||||
ephemeral_keys: DummyMixIdentityKeyPair,
|
||||
) -> Self {
|
||||
let mut provider_clients = HashMap::new();
|
||||
|
||||
let mut temporary_address = [0u8; 32];
|
||||
let public_key_bytes = ephemeral_keys.public_key().to_bytes();
|
||||
temporary_address.copy_from_slice(&public_key_bytes[..]);
|
||||
|
||||
for provider in providers {
|
||||
let mut provider_client =
|
||||
ProviderClient::new(provider.client_listener, temporary_address, None);
|
||||
let insertion_result = match provider_client.register().await {
|
||||
Ok(token) => {
|
||||
debug!("registered at provider {}", provider.pub_key);
|
||||
provider_client.update_token(token);
|
||||
provider_clients.insert(provider.get_pub_key_bytes(), Some(provider_client))
|
||||
}
|
||||
Err(err) => {
|
||||
warn!(
|
||||
"failed to register at provider {} - {:?}",
|
||||
provider.pub_key, err
|
||||
);
|
||||
provider_clients.insert(provider.get_pub_key_bytes(), None)
|
||||
}
|
||||
};
|
||||
|
||||
if insertion_result.is_some() {
|
||||
error!("provider {} already existed!", provider.pub_key);
|
||||
}
|
||||
}
|
||||
|
||||
PathChecker {
|
||||
provider_clients,
|
||||
layer_one_clients: HashMap::new(),
|
||||
our_destination: Destination::new(temporary_address, Default::default()),
|
||||
paths_status: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
// iteration is used to distinguish packets sent through the same path (as the healthcheck
|
||||
// may try to send say 10 packets through given path)
|
||||
fn unique_path_key(path: &Vec<SphinxNode>, iteration: u8) -> Vec<u8> {
|
||||
std::iter::once(iteration)
|
||||
.chain(
|
||||
path.iter()
|
||||
.map(|node| node.pub_key.to_bytes().to_vec())
|
||||
.flatten(),
|
||||
)
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub(crate) fn path_key_to_node_keys(path_key: Vec<u8>) -> Vec<[u8; 32]> {
|
||||
assert_eq!(path_key.len() % 32, 1);
|
||||
path_key
|
||||
.into_iter()
|
||||
.skip(1) // remove first byte as it represents the iteration number which we do not care about now
|
||||
.chunks(32)
|
||||
.into_iter()
|
||||
.map(|key_chunk| {
|
||||
let key_chunk_vec: Vec<_> = key_chunk.collect();
|
||||
let mut key = [0u8; 32];
|
||||
key.copy_from_slice(&key_chunk_vec);
|
||||
key
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn update_path_statuses(&mut self, messages: Vec<Vec<u8>>) {
|
||||
for msg in messages.into_iter() {
|
||||
// mark path as healthy
|
||||
let previous_status = self.paths_status.insert(msg, PathStatus::Healthy);
|
||||
match previous_status {
|
||||
None => warn!("we received information about unknown path! - perhaps somebody is messing with healthchecker?"),
|
||||
Some(status) => {
|
||||
if status != PathStatus::Pending {
|
||||
warn!("we received information about path that WASN'T in PENDING state! (it was in {:?}", status);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// consume path_checker and return all path statuses
|
||||
pub(crate) fn get_all_statuses(self) -> HashMap<Vec<u8>, PathStatus> {
|
||||
self.paths_status
|
||||
}
|
||||
|
||||
// pull messages from given provider until there are no more 'real' messages
|
||||
async fn resolve_pending_provider_checks(
|
||||
&self,
|
||||
provider_client: &ProviderClient,
|
||||
) -> Vec<Vec<u8>> {
|
||||
// keep getting messages until we encounter the dummy message
|
||||
let mut provider_messages = Vec::new();
|
||||
loop {
|
||||
match provider_client.retrieve_messages().await {
|
||||
Err(err) => {
|
||||
error!("failed to fetch provider messages! - {:?}", err);
|
||||
break;
|
||||
}
|
||||
Ok(messages) => {
|
||||
let mut should_stop = false;
|
||||
for msg in messages.into_iter() {
|
||||
if msg == sfw_provider_requests::DUMMY_MESSAGE_CONTENT {
|
||||
// finish iterating the loop as the messages might not be ordered
|
||||
should_stop = true;
|
||||
} else {
|
||||
provider_messages.push(msg);
|
||||
}
|
||||
}
|
||||
if should_stop {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
provider_messages
|
||||
}
|
||||
|
||||
pub(crate) async fn resolve_pending_checks(&mut self) {
|
||||
// not sure how to nicely put it into an iterator due to it being async calls
|
||||
let mut provider_messages = Vec::new();
|
||||
for provider_client in self.provider_clients.values() {
|
||||
// if it was none all associated paths were already marked as unhealthy
|
||||
if provider_client.is_some() {
|
||||
let pc = provider_client.as_ref().unwrap();
|
||||
provider_messages.extend(self.resolve_pending_provider_checks(pc).await);
|
||||
}
|
||||
}
|
||||
|
||||
self.update_path_statuses(provider_messages);
|
||||
}
|
||||
|
||||
pub(crate) async fn send_test_packet(&mut self, path: &Vec<SphinxNode>, iteration: u8) {
|
||||
debug!("Checking path: {:?} ({})", path, iteration);
|
||||
let path_identifier = PathChecker::unique_path_key(path, iteration);
|
||||
|
||||
// check if there is even any point in sending the packet
|
||||
|
||||
// does provider exist?
|
||||
let provider_client = self
|
||||
.provider_clients
|
||||
.get(&path.last().unwrap().pub_key.to_bytes())
|
||||
.unwrap();
|
||||
|
||||
if provider_client.is_none() {
|
||||
debug!("we can ignore this path as provider itself is inaccessible");
|
||||
self.paths_status
|
||||
.insert(path_identifier, PathStatus::Unhealthy)
|
||||
.unwrap();
|
||||
return;
|
||||
}
|
||||
|
||||
let layer_one_mix = path.first().unwrap();
|
||||
let first_node_key = layer_one_mix.pub_key.to_bytes();
|
||||
let first_node_address =
|
||||
addressing::socket_address_from_encoded_bytes(layer_one_mix.address.to_bytes());
|
||||
|
||||
let first_node_client = self
|
||||
.layer_one_clients
|
||||
.entry(first_node_key)
|
||||
.or_insert(Some(mix_client::MixClient::new()));
|
||||
|
||||
if first_node_client.is_none() {
|
||||
debug!("we can ignore this path as layer one mix is inaccessible");
|
||||
self.paths_status
|
||||
.insert(path_identifier, PathStatus::Unhealthy)
|
||||
.unwrap();
|
||||
return;
|
||||
}
|
||||
|
||||
let first_node_client = first_node_client.as_ref().unwrap();
|
||||
|
||||
let delays: Vec<_> = path.iter().map(|_| Delay::new(0)).collect();
|
||||
|
||||
let packet = sphinx::SphinxPacket::new(
|
||||
path_identifier.clone(),
|
||||
&path[..],
|
||||
&self.our_destination,
|
||||
&delays,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
debug!("sending test packet to {}", first_node_address);
|
||||
match first_node_client.send(packet, first_node_address).await {
|
||||
Err(err) => {
|
||||
warn!("failed to send packet to {} - {}", first_node_address, err);
|
||||
if self
|
||||
.paths_status
|
||||
.insert(path_identifier, PathStatus::Unhealthy)
|
||||
.is_some()
|
||||
{
|
||||
panic!("Overwriting path checks!")
|
||||
}
|
||||
}
|
||||
Ok(_) => {
|
||||
if self
|
||||
.paths_status
|
||||
.insert(path_identifier, PathStatus::Pending)
|
||||
.is_some()
|
||||
{
|
||||
panic!("Overwriting path checks!")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,109 +0,0 @@
|
||||
use crate::validator::health_check::path_check::{PathChecker, PathStatus};
|
||||
use crate::validator::health_check::score::NodeScore;
|
||||
use crypto::identity::{DummyMixIdentityKeyPair, MixnetIdentityKeyPair};
|
||||
use log::{debug, info, warn};
|
||||
use std::collections::HashMap;
|
||||
use std::fmt::{Error, Formatter};
|
||||
use std::time::Duration;
|
||||
use topology::NymTopology;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct HealthCheckResult(Vec<NodeScore>);
|
||||
|
||||
impl std::fmt::Display for HealthCheckResult {
|
||||
fn fmt(&self, f: &mut Formatter) -> Result<(), Error> {
|
||||
write!(f, "NETWORK HEALTH\n==============\n")?;
|
||||
self.0
|
||||
.iter()
|
||||
.for_each(|score| write!(f, "{}\n", score).unwrap());
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl HealthCheckResult {
|
||||
pub fn sort_scores(&mut self) {
|
||||
self.0.sort();
|
||||
}
|
||||
|
||||
fn zero_score<T: NymTopology>(topology: T) -> Self {
|
||||
warn!("The network is unhealthy, could not send any packets - returning zero score!");
|
||||
let mixes = topology.get_mix_nodes();
|
||||
let providers = topology.get_mix_provider_nodes();
|
||||
|
||||
let health = mixes
|
||||
.into_iter()
|
||||
.map(|node| NodeScore::from_mixnode(node))
|
||||
.chain(
|
||||
providers
|
||||
.into_iter()
|
||||
.map(|node| NodeScore::from_provider(node)),
|
||||
)
|
||||
.collect();
|
||||
|
||||
HealthCheckResult(health)
|
||||
}
|
||||
|
||||
pub async fn calculate<T: NymTopology>(
|
||||
topology: T,
|
||||
iterations: usize,
|
||||
resolution_timeout: Duration,
|
||||
) -> Self {
|
||||
// currently healthchecker supports only up to 255 iterations - if we somehow
|
||||
// find we need more, it's relatively easy change
|
||||
assert!(iterations <= 255);
|
||||
|
||||
let all_paths = match topology.all_paths() {
|
||||
Ok(paths) => paths,
|
||||
Err(_) => return Self::zero_score(topology),
|
||||
};
|
||||
|
||||
// create entries for all nodes
|
||||
let mut score_map = HashMap::new();
|
||||
topology.get_mix_nodes().into_iter().for_each(|node| {
|
||||
score_map.insert(node.get_pub_key_bytes(), NodeScore::from_mixnode(node));
|
||||
});
|
||||
|
||||
topology
|
||||
.get_mix_provider_nodes()
|
||||
.into_iter()
|
||||
.for_each(|node| {
|
||||
score_map.insert(node.get_pub_key_bytes(), NodeScore::from_provider(node));
|
||||
});
|
||||
|
||||
let ephemeral_keys = DummyMixIdentityKeyPair::new();
|
||||
let providers = topology.get_mix_provider_nodes();
|
||||
|
||||
let mut path_checker = PathChecker::new(providers, ephemeral_keys).await;
|
||||
for i in 0..iterations {
|
||||
debug!("running healthcheck iteration {} / {}", i + 1, iterations);
|
||||
for path in &all_paths {
|
||||
path_checker.send_test_packet(&path, i as u8).await;
|
||||
// increase sent count for each node
|
||||
for node in path {
|
||||
let current_node_score = score_map.get_mut(&node.pub_key.0).unwrap();
|
||||
current_node_score.increase_sent_packet_count();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
info!(
|
||||
"waiting {:?} for pending requests to resolve",
|
||||
resolution_timeout
|
||||
);
|
||||
tokio::time::delay_for(resolution_timeout).await;
|
||||
path_checker.resolve_pending_checks().await;
|
||||
|
||||
let all_statuses = path_checker.get_all_statuses();
|
||||
for (path_key, status) in all_statuses.into_iter() {
|
||||
let node_keys = PathChecker::path_key_to_node_keys(path_key);
|
||||
for node in node_keys {
|
||||
if status == PathStatus::Healthy {
|
||||
let current_node_score = score_map.get_mut(&node).unwrap();
|
||||
current_node_score.increase_received_packet_count();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
HealthCheckResult(score_map.into_iter().map(|(_, v)| v).collect())
|
||||
}
|
||||
}
|
||||
@@ -1,130 +0,0 @@
|
||||
use log::error;
|
||||
use sphinx::route::NodeAddressBytes;
|
||||
use std::cmp::Ordering;
|
||||
use std::fmt::Formatter;
|
||||
use std::net::SocketAddr;
|
||||
use topology::{MixNode, MixProviderNode};
|
||||
|
||||
#[derive(Debug, Eq)]
|
||||
pub(crate) struct NodeScore {
|
||||
pub_key: NodeAddressBytes,
|
||||
addresses: Vec<SocketAddr>,
|
||||
version: String,
|
||||
layer: String,
|
||||
packets_sent: u64,
|
||||
packets_received: u64,
|
||||
}
|
||||
|
||||
impl Ord for NodeScore {
|
||||
// order by: version, layer, sent, received, pubkey; ignore addresses
|
||||
fn cmp(&self, other: &Self) -> Ordering {
|
||||
if self.version > other.version {
|
||||
return Ordering::Greater;
|
||||
} else if self.version < other.version {
|
||||
return Ordering::Less;
|
||||
}
|
||||
if self.layer > other.layer {
|
||||
return Ordering::Greater;
|
||||
} else if self.layer < other.layer {
|
||||
return Ordering::Less;
|
||||
}
|
||||
if self.packets_sent > other.packets_sent {
|
||||
return Ordering::Greater;
|
||||
} else if self.packets_sent < other.packets_sent {
|
||||
return Ordering::Less;
|
||||
}
|
||||
if self.packets_received > other.packets_received {
|
||||
return Ordering::Greater;
|
||||
} else if self.packets_received < other.packets_received {
|
||||
return Ordering::Less;
|
||||
}
|
||||
if self.pub_key > other.pub_key {
|
||||
return Ordering::Greater;
|
||||
} else if self.pub_key < other.pub_key {
|
||||
return Ordering::Less;
|
||||
}
|
||||
|
||||
Ordering::Equal
|
||||
}
|
||||
}
|
||||
|
||||
impl PartialOrd for NodeScore {
|
||||
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
|
||||
Some(self.cmp(other))
|
||||
}
|
||||
}
|
||||
|
||||
impl PartialEq for NodeScore {
|
||||
fn eq(&self, other: &Self) -> bool {
|
||||
self.pub_key == other.pub_key
|
||||
&& self.addresses == other.addresses
|
||||
&& self.version == other.version
|
||||
&& self.layer == other.layer
|
||||
&& self.packets_sent == other.packets_sent
|
||||
&& self.packets_received == other.packets_received
|
||||
}
|
||||
}
|
||||
|
||||
impl NodeScore {
|
||||
pub(crate) fn from_mixnode(node: MixNode) -> Self {
|
||||
NodeScore {
|
||||
pub_key: NodeAddressBytes::from_b64_string(node.pub_key),
|
||||
addresses: vec![node.host],
|
||||
version: node.version,
|
||||
layer: format!("layer {}", node.layer),
|
||||
packets_sent: 0,
|
||||
packets_received: 0,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn from_provider(node: MixProviderNode) -> Self {
|
||||
NodeScore {
|
||||
pub_key: NodeAddressBytes::from_b64_string(node.pub_key),
|
||||
addresses: vec![node.mixnet_listener, node.client_listener],
|
||||
version: node.version,
|
||||
layer: format!("provider"),
|
||||
packets_sent: 0,
|
||||
packets_received: 0,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn increase_sent_packet_count(&mut self) {
|
||||
self.packets_sent += 1;
|
||||
}
|
||||
|
||||
pub(crate) fn increase_received_packet_count(&mut self) {
|
||||
self.packets_received += 1;
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Display for NodeScore {
|
||||
fn fmt(&self, f: &mut Formatter) -> Result<(), std::fmt::Error> {
|
||||
let fmtd_addresses = match self.addresses.len() {
|
||||
1 => format!("{}", self.addresses[0]),
|
||||
2 => format!("{}, {}", self.addresses[0], self.addresses[1]),
|
||||
n => {
|
||||
error!(
|
||||
"could not format score - node has {} addresses while only 1 or 2 are allowed!",
|
||||
n
|
||||
);
|
||||
return Err(std::fmt::Error);
|
||||
}
|
||||
};
|
||||
let health_percentage = match self.packets_sent {
|
||||
0 => 0.0,
|
||||
_ => (self.packets_received as f64 / self.packets_sent as f64) * 100.0,
|
||||
};
|
||||
let stringified_key = self.pub_key.to_b64_string();
|
||||
write!(
|
||||
f,
|
||||
"{}/{}\t({}%)\t|| {}\tv{} <{}> - {}",
|
||||
self.packets_received,
|
||||
self.packets_sent,
|
||||
health_percentage,
|
||||
self.layer,
|
||||
self.version,
|
||||
fmtd_addresses,
|
||||
stringified_key,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -1,10 +1,9 @@
|
||||
use crate::validator::config::Config;
|
||||
use crate::validator::health_check::HealthChecker;
|
||||
use healthcheck::HealthChecker;
|
||||
use log::debug;
|
||||
use tokio::runtime::Runtime;
|
||||
|
||||
pub mod config;
|
||||
mod health_check;
|
||||
|
||||
pub struct Validator {
|
||||
heath_check: HealthChecker,
|
||||
|
||||
Reference in New Issue
Block a user