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

introduce UNSTABLE endpoints for returning network monitor run details (#5214)

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2024-12-04 16:49:26 +00:00
committed by GitHub
parent 29ea4623c8
commit 78bf413e6a
9 changed files with 425 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-api/migrations/20241204120000_test_run_report.sql
+23
View File
@@ -0,0 +1,23 @@
/*
* Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
* SPDX-License-Identifier: GPL-3.0-only
*/
CREATE TABLE monitor_run_report
(
monitor_run_id INTEGER PRIMARY KEY REFERENCES monitor_run (id),
network_reliability FLOAT NOT NULL,
packets_sent INTEGER NOT NULL,
packets_received INTEGER NOT NULL
);
CREATE TABLE monitor_run_score
(
-- mixnode or gateway
typ TEXT NOT NULL,
monitor_run_id INTEGER NOT NULL REFERENCES monitor_run_report (monitor_run_id),
rounded_score INTEGER NOT NULL,
nodes_count INTEGER NOT NULL
);
CREATE INDEX monitor_run_score_id ON monitor_run_score (monitor_run_id);
nym-api/nym-api-requests/src/models.rs
+13
View File
@@ -32,6 +32,7 @@ use schemars::gen::SchemaGenerator;
use schemars::schema::{InstanceType, Schema, SchemaObject};
use schemars::JsonSchema;
use serde::{Deserialize, Deserializer, Serialize};
use std::collections::BTreeMap;
use std::fmt::{Debug, Display, Formatter};
use std::net::IpAddr;
use std::ops::{Deref, DerefMut};
@@ -1255,6 +1256,18 @@ pub struct PartialTestResult {
pub type MixnodeTestResultResponse = PaginatedResponse<PartialTestResult>;
pub type GatewayTestResultResponse = PaginatedResponse<PartialTestResult>;
#[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
pub struct NetworkMonitorRunDetailsResponse {
pub monitor_run_id: i64,
pub network_reliability: f64,
pub total_sent: usize,
pub total_received: usize,
// integer score to number of nodes with that score
pub mixnode_results: BTreeMap<u8, usize>,
pub gateway_results: BTreeMap<u8, usize>,
}
#[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)]
pub struct NoiseDetails {
#[schemars(with = "String")]
nym-api/src/network_monitor/monitor/mod.rs
+24 -6
View File
@@ -4,7 +4,7 @@
use crate::network_monitor::monitor::preparer::PacketPreparer;
use crate::network_monitor::monitor::processor::ReceivedProcessor;
use crate::network_monitor::monitor::sender::PacketSender;
use crate::network_monitor::monitor::summary_producer::{SummaryProducer, TestSummary};
use crate::network_monitor::monitor::summary_producer::{SummaryProducer, TestReport, TestSummary};
use crate::network_monitor::test_packet::NodeTestMessage;
use crate::network_monitor::test_route::TestRoute;
use crate::storage::NymApiStorage;
@@ -78,10 +78,10 @@ impl<R: MessageReceiver + Send> Monitor<R> {
// while it might have been cleaner to put this into a separate `Notifier` structure,
// I don't see much point considering it's only a single, small, method
async fn submit_new_node_statuses(&mut self, test_summary: TestSummary) {
async fn submit_new_node_statuses(&mut self, test_summary: TestSummary, report: TestReport) {
// indicate our run has completed successfully and should be used in any future
// uptime calculations
if let Err(err) = self
let monitor_run_id = match self
.node_status_storage
.insert_monitor_run_results(
test_summary.mixnode_results,
@@ -94,8 +94,22 @@ impl<R: MessageReceiver + Send> Monitor<R> {
)
.await
{
error!("Failed to submit monitor run information to the database: {err}",);
Ok(id) => id,
Err(err) => {
error!("Failed to submit monitor run information to the database: {err}",);
return;
}
};
if let Err(err) = self
.node_status_storage
.insert_monitor_run_report(report, monitor_run_id)
.await
{
error!("failed to submit monitor run report to the database: {err}",);
}
info!("finished persisting monitor run with id {monitor_run_id}");
}
fn analyse_received_test_route_packets(
@@ -279,9 +293,13 @@ impl<R: MessageReceiver + Send> Monitor<R> {
);
let report = summary.create_report(total_sent, total_received);
info!("{report}");
self.submit_new_node_statuses(summary).await;
let display_report = summary
.create_report(total_sent, total_received)
.to_display_report(&summary.route_results);
info!("{display_report}");
self.submit_new_node_statuses(summary, report).await;
}
async fn test_run(&mut self) {
nym-api/src/network_monitor/monitor/summary_producer.rs
+112 -76
View File
@@ -32,9 +32,118 @@ impl RouteResult {
}
}
#[derive(Default, Debug)]
#[derive(Debug)]
pub(crate) struct TestReport {
pub(crate) network_reliability: f32,
pub(crate) network_reliability: f64,
pub(crate) total_sent: usize,
pub(crate) total_received: usize,
// integer score to number of nodes with that score
pub(crate) mixnode_results: HashMap<u8, usize>,
pub(crate) gateway_results: HashMap<u8, usize>,
}
impl TestReport {
pub(crate) fn new(
total_sent: usize,
total_received: usize,
raw_mixnode_results: &[NodeResult],
raw_gateway_results: &[NodeResult],
) -> Self {
let network_reliability = total_received as f64 / total_sent as f64 * 100.0;
let mut mixnode_results = HashMap::new();
let mut gateway_results = HashMap::new();
for res in raw_mixnode_results {
mixnode_results
.entry(res.reliability)
.and_modify(|c| *c += 1)
.or_insert(1);
}
for res in raw_gateway_results {
gateway_results
.entry(res.reliability)
.and_modify(|c| *c += 1)
.or_insert(1);
}
TestReport {
network_reliability,
total_sent,
total_received,
mixnode_results,
gateway_results,
}
}
pub(crate) fn to_display_report(&self, route_results: &[RouteResult]) -> DisplayTestReport {
let mut exceptional_mixnodes = 0;
let mut exceptional_gateways = 0;
let mut fine_mixnodes = 0;
let mut fine_gateways = 0;
let mut poor_mixnodes = 0;
let mut poor_gateways = 0;
let mut unreliable_mixnodes = 0;
let mut unreliable_gateways = 0;
let mut unroutable_mixnodes = 0;
let mut unroutable_gateways = 0;
for (&score, &count) in &self.mixnode_results {
if score >= EXCEPTIONAL_THRESHOLD {
exceptional_mixnodes += count;
} else if score >= FINE_THRESHOLD {
fine_mixnodes += count;
} else if score >= POOR_THRESHOLD {
poor_mixnodes += count;
} else if score >= UNRELIABLE_THRESHOLD {
unreliable_mixnodes += count;
} else {
unroutable_mixnodes += count;
}
}
for (&score, &count) in &self.gateway_results {
if score >= EXCEPTIONAL_THRESHOLD {
exceptional_gateways += count;
} else if score >= FINE_THRESHOLD {
fine_gateways += count;
} else if score >= POOR_THRESHOLD {
poor_gateways += count;
} else if score >= UNRELIABLE_THRESHOLD {
unreliable_gateways += count;
} else {
unroutable_gateways += count;
}
}
DisplayTestReport {
network_reliability: self.network_reliability,
total_sent: self.total_sent,
total_received: self.total_received,
route_results: route_results.to_vec(),
exceptional_mixnodes,
exceptional_gateways,
fine_mixnodes,
fine_gateways,
poor_mixnodes,
poor_gateways,
unreliable_mixnodes,
unreliable_gateways,
unroutable_mixnodes,
unroutable_gateways,
}
}
}
#[derive(Default, Debug)]
pub(crate) struct DisplayTestReport {
pub(crate) network_reliability: f64,
pub(crate) total_sent: usize,
pub(crate) total_received: usize,
@@ -56,79 +165,7 @@ pub(crate) struct TestReport {
pub(crate) unroutable_gateways: usize,
}
impl TestReport {
fn new(
total_sent: usize,
total_received: usize,
mixnode_results: &[NodeResult],
gateway_results: &[NodeResult],
route_results: &[RouteResult],
) -> Self {
let mut exceptional_mixnodes = 0;
let mut exceptional_gateways = 0;
let mut fine_mixnodes = 0;
let mut fine_gateways = 0;
let mut poor_mixnodes = 0;
let mut poor_gateways = 0;
let mut unreliable_mixnodes = 0;
let mut unreliable_gateways = 0;
let mut unroutable_mixnodes = 0;
let mut unroutable_gateways = 0;
for mixnode_result in mixnode_results {
if mixnode_result.reliability >= EXCEPTIONAL_THRESHOLD {
exceptional_mixnodes += 1;
} else if mixnode_result.reliability >= FINE_THRESHOLD {
fine_mixnodes += 1;
} else if mixnode_result.reliability >= POOR_THRESHOLD {
poor_mixnodes += 1;
} else if mixnode_result.reliability >= UNRELIABLE_THRESHOLD {
unreliable_mixnodes += 1;
} else {
unroutable_mixnodes += 1;
}
}
for gateway_result in gateway_results {
if gateway_result.reliability >= EXCEPTIONAL_THRESHOLD {
exceptional_gateways += 1;
} else if gateway_result.reliability >= FINE_THRESHOLD {
fine_gateways += 1;
} else if gateway_result.reliability >= POOR_THRESHOLD {
poor_gateways += 1;
} else if gateway_result.reliability >= UNRELIABLE_THRESHOLD {
unreliable_gateways += 1;
} else {
unroutable_gateways += 1;
}
}
let network_reliability = total_received as f32 / total_sent as f32 * 100.0;
TestReport {
network_reliability,
total_sent,
total_received,
route_results: route_results.to_vec(),
exceptional_mixnodes,
exceptional_gateways,
fine_mixnodes,
fine_gateways,
poor_mixnodes,
poor_gateways,
unreliable_mixnodes,
unreliable_gateways,
unroutable_mixnodes,
unroutable_gateways,
}
}
}
impl Display for TestReport {
impl Display for DisplayTestReport {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
writeln!(f, "Mix Network Test Report")?;
writeln!(
@@ -218,7 +255,6 @@ impl TestSummary {
total_received,
&self.mixnode_results,
&self.gateway_results,
&self.route_results,
)
}
}
nym-api/src/node_status_api/handlers/network_monitor.rs
+14 -2
View File
@@ -1,6 +1,8 @@
// Copyright 2021-2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use super::unstable;
use crate::node_status_api::handlers::unstable::{latest_monitor_run_report, monitor_run_report};
use crate::node_status_api::handlers::MixIdParam;
use crate::node_status_api::helpers::{
_compute_mixnode_reward_estimation, _gateway_core_status_count, _gateway_report,
@@ -23,8 +25,6 @@ use nym_api_requests::models::{
use serde::Deserialize;
use utoipa::IntoParams;
use super::unstable;
// we want to mark the routes as deprecated in swagger, but still expose them
#[allow(deprecated)]
pub(super) fn network_monitor_routes() -> Router<AppState> {
@@ -84,6 +84,18 @@ pub(super) fn network_monitor_routes() -> Router<AppState> {
axum::routing::get(unstable::gateway_test_results),
),
)
.nest(
"/network-monitor/unstable",
Router::new()
.route(
"/run/:monitor_run_id/details",
axum::routing::get(monitor_run_report),
)
.route(
"/run/latest/details",
axum::routing::get(latest_monitor_run_report),
),
)
}
#[utoipa::path(
nym-api/src/node_status_api/handlers/unstable.rs
+89 -2
View File
@@ -6,15 +6,17 @@ use crate::node_status_api::models::{AxumErrorResponse, AxumResult};
use crate::support::http::helpers::PaginationRequest;
use crate::support::http::state::AppState;
use crate::support::storage::NymApiStorage;
use anyhow::bail;
use axum::extract::{Path, Query, State};
use axum::Json;
use nym_api_requests::models::{
GatewayTestResultResponse, MixnodeTestResultResponse, PartialTestResult, TestNode, TestRoute,
GatewayTestResultResponse, MixnodeTestResultResponse, NetworkMonitorRunDetailsResponse,
PartialTestResult, TestNode, TestRoute,
};
use nym_api_requests::pagination::Pagination;
use nym_mixnet_contract_common::NodeId;
use std::cmp::min;
use std::collections::HashMap;
use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{error, trace};
@@ -301,3 +303,88 @@ pub async fn gateway_test_results(
))),
}
}
async fn _monitor_run_report(
monitor_run_id: i64,
storage: &NymApiStorage,
) -> anyhow::Result<NetworkMonitorRunDetailsResponse> {
let Some((raw_report, raw_scores)) = storage.get_monitor_run_report(monitor_run_id).await?
else {
bail!("no results found for monitor run {monitor_run_id}");
};
let mut mixnode_results = BTreeMap::new();
let mut gateway_results = BTreeMap::new();
for score in raw_scores {
if score.typ == "mixnode" {
mixnode_results.insert(score.rounded_score, score.nodes_count as usize);
} else if score.typ == "gateway" {
gateway_results.insert(score.rounded_score, score.nodes_count as usize);
}
}
Ok(NetworkMonitorRunDetailsResponse {
monitor_run_id,
network_reliability: raw_report.network_reliability,
total_sent: raw_report.packets_sent as usize,
total_received: raw_report.packets_received as usize,
mixnode_results,
gateway_results,
})
}
async fn _latest_monitor_run_report(
storage: &NymApiStorage,
) -> anyhow::Result<NetworkMonitorRunDetailsResponse> {
let Some(latest_id) = storage.get_latest_monitor_run_id().await? else {
bail!("no network monitor run found");
};
_monitor_run_report(latest_id, storage).await
}
#[utoipa::path(
tag = "UNSTABLE - DO **NOT** USE",
get,
params(
PaginationRequest
),
path = "/v1/status/network-monitor/unstable/run/{monitor_run_id}/details",
responses(
(status = 200, body = NetworkMonitorRunDetailsResponse)
)
)]
pub async fn monitor_run_report(
Path(monitor_run_id): Path<i64>,
State(state): State<AppState>,
) -> AxumResult<Json<NetworkMonitorRunDetailsResponse>> {
match _monitor_run_report(monitor_run_id, state.storage()).await {
Ok(res) => Ok(Json(res)),
Err(err) => Err(AxumErrorResponse::internal_msg(format!(
"failed to retrieve monitor run report for run {monitor_run_id}: {err}"
))),
}
}
#[utoipa::path(
tag = "UNSTABLE - DO **NOT** USE",
get,
params(
PaginationRequest
),
path = "/v1/status/network-monitor/unstable/run/latest/details",
responses(
(status = 200, body = NetworkMonitorRunDetailsResponse)
)
)]
pub async fn latest_monitor_run_report(
State(state): State<AppState>,
) -> AxumResult<Json<NetworkMonitorRunDetailsResponse>> {
match _latest_monitor_run_report(state.storage()).await {
Ok(res) => Ok(Json(res)),
Err(err) => Err(AxumErrorResponse::internal_msg(format!(
"failed to retrieve the latest monitor run report: {err}"
))),
}
}
nym-api/src/support/storage/manager.rs
+75 -2
View File
@@ -4,8 +4,9 @@
use crate::node_status_api::models::{HistoricalUptime as ApiHistoricalUptime, Uptime};
use crate::node_status_api::utils::{ActiveGatewayStatuses, ActiveMixnodeStatuses};
use crate::support::storage::models::{
ActiveGateway, ActiveMixnode, GatewayDetails, HistoricalUptime, MixnodeDetails, NodeStatus,
RewardingReport, TestedGatewayStatus, TestedMixnodeStatus, TestingRoute,
ActiveGateway, ActiveMixnode, GatewayDetails, HistoricalUptime, MixnodeDetails,
MonitorRunReport, MonitorRunScore, NodeStatus, RewardingReport, TestedGatewayStatus,
TestedMixnodeStatus, TestingRoute,
};
use crate::support::storage::DbIdCache;
use nym_mixnet_contract_common::{EpochId, IdentityKey, NodeId};
@@ -883,6 +884,78 @@ impl StorageManager {
Ok(res.last_insert_rowid())
}
pub(super) async fn insert_monitor_run_report(
&self,
monitor_run_id: i64,
network_reliability: f64,
total_packets_sent: u32,
total_packets_received: u32,
) -> Result<(), sqlx::Error> {
sqlx::query!(
r#"
INSERT INTO monitor_run_report(
monitor_run_id,
network_reliability,
packets_sent,
packets_received
) VALUES (?, ?, ?, ?)
"#,
monitor_run_id,
network_reliability,
total_packets_sent,
total_packets_received
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(super) async fn get_monitor_run_report(
&self,
monitor_run_id: i64,
) -> Result<Option<MonitorRunReport>, sqlx::Error> {
sqlx::query_as("SELECT * FROM monitor_run_report WHERE monitor_run_id = ?")
.bind(monitor_run_id)
.fetch_optional(&self.connection_pool)
.await
}
pub(super) async fn get_latest_monitor_run_id(&self) -> Result<Option<i64>, sqlx::Error> {
sqlx::query!("SELECT id from monitor_run ORDER BY id DESC limit 1")
.fetch_optional(&self.connection_pool)
.await
.map(|r| r.map(|r| r.id))
}
pub(super) async fn insert_monitor_run_scores(
&self,
scores: Vec<MonitorRunScore>,
) -> Result<(), sqlx::Error> {
let mut query_builder = sqlx::QueryBuilder::new(
"INSERT INTO monitor_run_score (typ, monitor_run_id, rounded_score, nodes_count) ",
);
query_builder.push_values(scores, |mut b, score| {
b.push_bind(score.typ)
.push_bind(score.monitor_run_id)
.push_bind(score.rounded_score)
.push_bind(score.nodes_count);
});
query_builder.build().execute(&self.connection_pool).await?;
Ok(())
}
pub(super) async fn get_monitor_run_scores(
&self,
monitor_run_id: i64,
) -> Result<Vec<MonitorRunScore>, sqlx::Error> {
sqlx::query_as("SELECT * FROM monitor_run_score WHERE monitor_run_id = ?")
.bind(monitor_run_id)
.fetch_all(&self.connection_pool)
.await
}
/// Obtains number of network monitor test runs that have occurred within the specified interval.
///
/// # Arguments
nym-api/src/support/storage/mod.rs
+58 -2
View File
@@ -2,6 +2,7 @@
// SPDX-License-Identifier: GPL-3.0-only
use self::manager::{AvgGatewayReliability, AvgMixnodeReliability};
use crate::network_monitor::monitor::summary_producer::TestReport;
use crate::network_monitor::test_route::TestRoute;
use crate::node_status_api::models::{
GatewayStatusReport, GatewayUptimeHistory, HistoricalUptime as ApiHistoricalUptime,
@@ -11,7 +12,8 @@ use crate::node_status_api::{ONE_DAY, ONE_HOUR};
use crate::storage::manager::StorageManager;
use crate::storage::models::{NodeStatus, TestingRoute};
use crate::support::storage::models::{
GatewayDetails, HistoricalUptime, MixnodeDetails, TestedGatewayStatus, TestedMixnodeStatus,
GatewayDetails, HistoricalUptime, MixnodeDetails, MonitorRunReport, MonitorRunScore,
TestedGatewayStatus, TestedMixnodeStatus,
};
use dashmap::DashMap;
use nym_mixnet_contract_common::NodeId;
@@ -730,7 +732,7 @@ impl NymApiStorage {
mixnode_results: Vec<NodeResult>,
gateway_results: Vec<NodeResult>,
test_routes: Vec<TestRoute>,
) -> Result<(), NymApiStorageError> {
) -> Result<i64, NymApiStorageError> {
info!("Submitting new node results to the database. There are {} mixnode results and {} gateway results", mixnode_results.len(), gateway_results.len());
let now = OffsetDateTime::now_utc().unix_timestamp();
@@ -749,9 +751,63 @@ impl NymApiStorage {
self.insert_test_route(monitor_run_id, test_route).await?;
}
Ok(monitor_run_id)
}
pub(crate) async fn insert_monitor_run_report(
&self,
report: TestReport,
monitor_run_id: i64,
) -> Result<(), NymApiStorageError> {
self.manager
.insert_monitor_run_report(
monitor_run_id,
report.network_reliability,
report.total_sent as u32,
report.total_received as u32,
)
.await?;
let mut scores = Vec::new();
for (score, count) in report.mixnode_results {
scores.push(MonitorRunScore {
typ: "mixnode".to_string(),
monitor_run_id,
rounded_score: score,
nodes_count: count as u32,
})
}
for (score, count) in report.gateway_results {
scores.push(MonitorRunScore {
typ: "gateway".to_string(),
monitor_run_id,
rounded_score: score,
nodes_count: count as u32,
})
}
self.manager.insert_monitor_run_scores(scores).await?;
Ok(())
}
pub(crate) async fn get_monitor_run_report(
&self,
monitor_run_id: i64,
) -> Result<Option<(MonitorRunReport, Vec<MonitorRunScore>)>, NymApiStorageError> {
let Some(report) = self.manager.get_monitor_run_report(monitor_run_id).await? else {
return Ok(None);
};
let scores = self.manager.get_monitor_run_scores(monitor_run_id).await?;
Ok(Some((report, scores)))
}
pub(crate) async fn get_latest_monitor_run_id(
&self,
) -> Result<Option<i64>, NymApiStorageError> {
Ok(self.manager.get_latest_monitor_run_id().await?)
}
pub(crate) async fn submit_mixnode_statuses_v2(
&self,
mixnode_results: &[NodeResult],
nym-api/src/support/storage/models.rs
+17
View File
@@ -6,6 +6,23 @@ use nym_mixnet_contract_common::NodeId;
use sqlx::FromRow;
use time::Date;
#[derive(sqlx::FromRow, Debug, Clone, Copy)]
pub(crate) struct MonitorRunReport {
#[allow(dead_code)]
pub(crate) monitor_run_id: i64,
pub(crate) network_reliability: f64,
pub(crate) packets_sent: i64,
pub(crate) packets_received: i64,
}
#[derive(sqlx::FromRow, Debug, Clone)]
pub(crate) struct MonitorRunScore {
pub(crate) typ: String,
pub(crate) monitor_run_id: i64,
pub(crate) rounded_score: u8,
pub(crate) nodes_count: u32,
}
// Internally used struct to catch results from the database to calculate uptimes for given mixnode/gateway
pub(crate) struct NodeStatus {
pub timestamp: Option<i64>,