GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feature/removal of monitor good nodes (#833)

Browse Source 
* Removed separated ipv4 and ipv6 testing

* Testing network using chosen core nodes

This should have probably been like 20 independent commits... sorry...

* SQL migrations for updated schema

* SQL updates

* Using absolute uptime directly

* New uptime calculations

* Config entries, more DB work, some cleanup

* Additional API query routes

* More SQL and API work

* Changed `_` to `-` in new routes

* Removed good topology from config

* Fixed gateways reader yield condition

* Initial gateways pinger

* Minor cleanup and logging level decreases

* Missing trait derivations

* Further logging adjustments

* Unused commented out import

* Claiming additional bandwidth in coconut feature when low

* Fixed build with coconut feature

* Minimum number of test routes

* Making beta/nightly clippy happier
This commit is contained in:
Jędrzej Stuczyński
2021-11-09 12:25:41 +00:00
committed by GitHub
parent 955ef7b871
commit 8bcc931d9b
27 changed files with 2380 additions and 1736 deletions
Download Patch File Download Diff File Expand all files Collapse all files
validator-api/src/node_status_api/mod.rs
+2
View File
@@ -27,6 +27,8 @@ pub(crate) fn stage(database_path: PathBuf) -> AdHoc {
routes::gateway_report,
routes::mixnode_uptime_history,
routes::gateway_uptime_history,
routes::mixnode_core_status_count,
routes::gateway_core_status_count,
],
)
})
validator-api/src/node_status_api/models.rs
+38 -37
View File
@@ -30,7 +30,21 @@ impl Uptime {
return Ok(Self::zero());
}
let uptime = ((numerator as f32 / denominator as f32) * 100.0) as u8;
let uptime = ((numerator as f32 / denominator as f32) * 100.0).round() as u8;
if uptime > 100 {
Err(InvalidUptime)
} else {
Ok(Uptime(uptime))
}
}
pub fn from_uptime_sum(running_sum: f32, count: usize) -> Result<Self, InvalidUptime> {
if count == 0 {
return Ok(Self::zero());
}
let uptime = (running_sum / count as f32).round() as u8;
if uptime > 100 {
Err(InvalidUptime)
@@ -79,14 +93,10 @@ pub struct MixnodeStatusReport {
pub(crate) identity: String,
pub(crate) owner: String,
pub(crate) most_recent_ipv4: bool,
pub(crate) most_recent_ipv6: bool,
pub(crate) most_recent: Uptime,
pub(crate) last_hour_ipv4: Uptime,
pub(crate) last_hour_ipv6: Uptime,
pub(crate) last_day_ipv4: Uptime,
pub(crate) last_day_ipv6: Uptime,
pub(crate) last_hour: Uptime,
pub(crate) last_day: Uptime,
}
impl MixnodeStatusReport {
@@ -94,15 +104,13 @@ impl MixnodeStatusReport {
report_time: OffsetDateTime,
identity: String,
owner: String,
last_day_ipv4: Vec<NodeStatus>,
last_day_ipv6: Vec<NodeStatus>,
last_day: Vec<NodeStatus>,
last_hour_test_runs: usize,
last_day_test_runs: usize,
) -> Self {
let node_uptimes = NodeUptimes::calculate_from_last_day_reports(
report_time,
last_day_ipv4,
last_day_ipv6,
last_day,
last_hour_test_runs,
last_day_test_runs,
);
@@ -110,12 +118,9 @@ impl MixnodeStatusReport {
MixnodeStatusReport {
identity,
owner,
most_recent_ipv4: node_uptimes.most_recent_ipv4,
most_recent_ipv6: node_uptimes.most_recent_ipv6,
last_hour_ipv4: node_uptimes.last_hour_ipv4,
last_hour_ipv6: node_uptimes.last_hour_ipv6,
last_day_ipv4: node_uptimes.last_day_ipv4,
last_day_ipv6: node_uptimes.last_day_ipv6,
most_recent: node_uptimes.most_recent,
last_hour: node_uptimes.last_hour,
last_day: node_uptimes.last_day,
}
}
}
@@ -125,14 +130,10 @@ pub struct GatewayStatusReport {
pub(crate) identity: String,
pub(crate) owner: String,
pub(crate) most_recent_ipv4: bool,
pub(crate) most_recent_ipv6: bool,
pub(crate) most_recent: Uptime,
pub(crate) last_hour_ipv4: Uptime,
pub(crate) last_hour_ipv6: Uptime,
pub(crate) last_day_ipv4: Uptime,
pub(crate) last_day_ipv6: Uptime,
pub(crate) last_hour: Uptime,
pub(crate) last_day: Uptime,
}
impl GatewayStatusReport {
@@ -140,15 +141,13 @@ impl GatewayStatusReport {
report_time: OffsetDateTime,
identity: String,
owner: String,
last_day_ipv4: Vec<NodeStatus>,
last_day_ipv6: Vec<NodeStatus>,
last_day: Vec<NodeStatus>,
last_hour_test_runs: usize,
last_day_test_runs: usize,
) -> Self {
let node_uptimes = NodeUptimes::calculate_from_last_day_reports(
report_time,
last_day_ipv4,
last_day_ipv6,
last_day,
last_hour_test_runs,
last_day_test_runs,
);
@@ -156,12 +155,9 @@ impl GatewayStatusReport {
GatewayStatusReport {
identity,
owner,
most_recent_ipv4: node_uptimes.most_recent_ipv4,
most_recent_ipv6: node_uptimes.most_recent_ipv6,
last_hour_ipv4: node_uptimes.last_hour_ipv4,
last_hour_ipv6: node_uptimes.last_hour_ipv6,
last_day_ipv4: node_uptimes.last_day_ipv4,
last_day_ipv6: node_uptimes.last_day_ipv6,
most_recent: node_uptimes.most_recent,
last_hour: node_uptimes.last_hour,
last_day: node_uptimes.last_day,
}
}
}
@@ -208,8 +204,7 @@ pub struct HistoricalUptime {
// I think this is more than enough, we don't need the uber precision of timezone offsets, etc
pub(crate) date: String,
pub(crate) ipv4_uptime: Uptime,
pub(crate) ipv6_uptime: Uptime,
pub(crate) uptime: Uptime,
}
pub(crate) struct ErrorResponse {
@@ -274,3 +269,9 @@ impl Display for ValidatorApiStorageError {
}
}
}
#[derive(Serialize)]
pub struct CoreNodeStatus {
pub(crate) identity: String,
pub(crate) count: i32,
}
validator-api/src/node_status_api/routes.rs
+35 -1
View File
@@ -2,7 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
use crate::node_status_api::models::{
ErrorResponse, GatewayStatusReport, GatewayUptimeHistory, MixnodeStatusReport,
CoreNodeStatus, ErrorResponse, GatewayStatusReport, GatewayUptimeHistory, MixnodeStatusReport,
MixnodeUptimeHistory,
};
use crate::storage::ValidatorApiStorage;
@@ -57,3 +57,37 @@ pub(crate) async fn gateway_uptime_history(
.map(Json)
.map_err(|err| ErrorResponse::new(err, Status::NotFound))
}
#[get("/mixnode/<pubkey>/core-status-count?<since>")]
pub(crate) async fn mixnode_core_status_count(
storage: &State<ValidatorApiStorage>,
pubkey: &str,
since: Option<i64>,
) -> Json<CoreNodeStatus> {
let count = storage
.get_core_mixnode_status_count(pubkey, since)
.await
.unwrap_or_default();
Json(CoreNodeStatus {
identity: pubkey.to_string(),
count,
})
}
#[get("/gateway/<pubkey>/core-status-count?<since>")]
pub(crate) async fn gateway_core_status_count(
storage: &State<ValidatorApiStorage>,
pubkey: &str,
since: Option<i64>,
) -> Json<CoreNodeStatus> {
let count = storage
.get_core_gateway_status_count(pubkey, since)
.await
.unwrap_or_default();
Json(CoreNodeStatus {
identity: pubkey.to_string(),
count,
})
}
validator-api/src/node_status_api/utils.rs
+66 -61
View File
@@ -5,100 +5,105 @@ use crate::node_status_api::models::Uptime;
use crate::node_status_api::{FIFTEEN_MINUTES, ONE_HOUR};
use crate::storage::models::NodeStatus;
use log::warn;
use std::cmp::min;
use std::convert::TryInto;
use time::OffsetDateTime;
// A temporary helper struct used to produce reports for active nodes.
pub(crate) struct ActiveNodeDayStatuses {
pub(crate) struct ActiveNodeStatuses {
pub(crate) identity: String,
pub(crate) owner: String,
pub(crate) ipv4_statuses: Vec<NodeStatus>,
pub(crate) ipv6_statuses: Vec<NodeStatus>,
pub(crate) statuses: Vec<NodeStatus>,
}
// A helper intermediate struct to remove duplicate code for construction of mixnode and gateway reports
pub(crate) struct NodeUptimes {
pub(crate) most_recent_ipv4: bool,
pub(crate) most_recent_ipv6: bool,
pub(crate) most_recent: Uptime,
pub(crate) last_hour_ipv4: Uptime,
pub(crate) last_hour_ipv6: Uptime,
pub(crate) last_day_ipv4: Uptime,
pub(crate) last_day_ipv6: Uptime,
pub(crate) last_hour: Uptime,
pub(crate) last_day: Uptime,
}
impl NodeUptimes {
pub(crate) fn calculate_from_last_day_reports(
report_time: OffsetDateTime,
last_day_ipv4: Vec<NodeStatus>,
last_day_ipv6: Vec<NodeStatus>,
last_day: Vec<NodeStatus>,
last_hour_test_runs: usize,
last_day_test_runs: usize,
) -> Self {
let hour_ago = (report_time - ONE_HOUR).unix_timestamp();
let fifteen_minutes_ago = (report_time - FIFTEEN_MINUTES).unix_timestamp();
let mut ipv4_day_up = last_day_ipv4.iter().filter(|report| report.up).count();
let mut ipv6_day_up = last_day_ipv6.iter().filter(|report| report.up).count();
let mut ipv4_hour_up = last_day_ipv4
.iter()
.filter(|report| report.up && report.timestamp >= hour_ago)
.count();
let mut ipv6_hour_up = last_day_ipv6
.iter()
.filter(|report| report.up && report.timestamp >= hour_ago)
.count();
// most recent status MUST BE within last 15min
let most_recent_ipv4 = last_day_ipv4
.iter()
.max_by_key(|report| report.timestamp) // find the most recent
.map(|status| status.timestamp >= fifteen_minutes_ago && status.up) // make sure its within last 15min
.unwrap_or_default();
let most_recent_ipv6 = last_day_ipv6
.iter()
.max_by_key(|report| report.timestamp) // find the most recent
.map(|status| status.timestamp >= fifteen_minutes_ago && status.up) // make sure its within last 15min
.unwrap_or_default();
// If somehow we have more "up" reports than the actual test runs it means something weird is going on
// If somehow we have more reports than the actual test runs it means something weird is going on
// (or we just started running this code on old data, so if it appears for first 24h, it's fine and actually expected
// as we would not have any run information from the past)
// Either way, bound the the number of "up" reports by number of test runs and log warnings
// if that happens
if ipv4_hour_up > last_hour_test_runs || ipv6_hour_up > last_hour_test_runs {
warn!(
"We have more 'up' reports than the actual number of test runs in last hour! ({} ipv4 'ups', {} ipv6 'ups' for {} test runs)",
ipv4_hour_up,
ipv6_hour_up,
last_hour_test_runs,
);
ipv4_hour_up = min(ipv4_hour_up, last_hour_test_runs);
ipv6_hour_up = min(ipv6_hour_up, last_hour_test_runs);
}
if ipv4_day_up > last_day_test_runs || ipv6_day_up > last_day_test_runs {
let last_day_sum: f32 = if last_day.len() > last_day_test_runs {
warn!(
"We have more 'up' reports than the actual number of test runs in last day! ({} ipv4 'ups', {} ipv6 'ups' for {} test runs)",
ipv4_day_up,
ipv6_day_up,
"We have more reports than the actual number of test runs in last day! ({} reports for {} test runs)",
last_day.len(),
last_day_test_runs,
);
ipv4_day_up = min(ipv4_day_up, last_day_test_runs);
ipv6_day_up = min(ipv6_day_up, last_day_test_runs);
}
last_day
.iter()
.take(last_day_test_runs)
.map(|report| report.reliability as f32)
.sum()
} else {
// we average over expected number of test runs so if a node was not online for some of them
// it's treated as if it had a "zero" status.
last_day
.iter()
.map(|report| report.reliability as f32)
.sum()
};
let last_hour_reports = last_day
.iter()
.filter(|report| report.timestamp >= hour_ago)
.count();
let last_hour_sum: f32 = if last_hour_reports > last_hour_test_runs {
warn!(
"We have more reports than the actual number of test runs in last hour! ({} reports for {} test runs)",
last_hour_reports,
last_hour_test_runs,
);
last_day
.iter()
.filter(|report| report.timestamp >= hour_ago)
.take(last_hour_test_runs)
.map(|report| report.reliability as f32)
.sum()
} else {
last_day
.iter()
.filter(|report| report.timestamp >= hour_ago)
.map(|report| report.reliability as f32)
.sum()
};
// find the most recent
let most_recent_report = last_day.iter().max_by_key(|report| report.timestamp);
let most_recent = if let Some(most_recent_report) = most_recent_report {
// make sure its within last 15min
if most_recent_report.timestamp >= fifteen_minutes_ago {
most_recent_report.reliability
} else {
0
}
} else {
0
};
// the unwraps in Uptime::from_ratio are fine because it's impossible for us to have more "up" results
// than total test runs as we just bounded them
NodeUptimes {
most_recent_ipv4,
most_recent_ipv6,
last_hour_ipv4: Uptime::from_ratio(ipv4_hour_up, last_hour_test_runs).unwrap(),
last_hour_ipv6: Uptime::from_ratio(ipv6_hour_up, last_hour_test_runs).unwrap(),
last_day_ipv4: Uptime::from_ratio(ipv4_day_up, last_day_test_runs).unwrap(),
last_day_ipv6: Uptime::from_ratio(ipv6_day_up, last_day_test_runs).unwrap(),
most_recent: most_recent.try_into().unwrap(),
last_hour: Uptime::from_uptime_sum(last_hour_sum, last_hour_test_runs).unwrap(),
last_day: Uptime::from_uptime_sum(last_day_sum, last_hour_test_runs).unwrap(),
}
}
}