From 78bf413e6ab6458f57fc6cc213ee3de228732141 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Wed, 4 Dec 2024 16:49:26 +0000 Subject: [PATCH] introduce UNSTABLE endpoints for returning network monitor run details (#5214) --- .../20241204120000_test_run_report.sql | 23 +++ nym-api/nym-api-requests/src/models.rs | 13 ++ nym-api/src/network_monitor/monitor/mod.rs | 30 ++- .../monitor/summary_producer.rs | 188 +++++++++++------- .../handlers/network_monitor.rs | 16 +- .../src/node_status_api/handlers/unstable.rs | 91 ++++++++- nym-api/src/support/storage/manager.rs | 77 ++++++- nym-api/src/support/storage/mod.rs | 60 +++++- nym-api/src/support/storage/models.rs | 17 ++ 9 files changed, 425 insertions(+), 90 deletions(-) create mode 100644 nym-api/migrations/20241204120000_test_run_report.sql diff --git a/nym-api/migrations/20241204120000_test_run_report.sql b/nym-api/migrations/20241204120000_test_run_report.sql new file mode 100644 index 0000000000..62b820b000 --- /dev/null +++ b/nym-api/migrations/20241204120000_test_run_report.sql @@ -0,0 +1,23 @@ +/* + * Copyright 2024 - Nym Technologies SA + * 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); \ No newline at end of file diff --git a/nym-api/nym-api-requests/src/models.rs b/nym-api/nym-api-requests/src/models.rs index ee87d37f74..8465905b62 100644 --- a/nym-api/nym-api-requests/src/models.rs +++ b/nym-api/nym-api-requests/src/models.rs @@ -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; pub type GatewayTestResultResponse = PaginatedResponse; +#[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, + pub gateway_results: BTreeMap, +} + #[derive(Clone, Debug, Serialize, Deserialize, schemars::JsonSchema, ToSchema)] pub struct NoiseDetails { #[schemars(with = "String")] diff --git a/nym-api/src/network_monitor/monitor/mod.rs b/nym-api/src/network_monitor/monitor/mod.rs index 4d24cb0504..a5ea5ab818 100644 --- a/nym-api/src/network_monitor/monitor/mod.rs +++ b/nym-api/src/network_monitor/monitor/mod.rs @@ -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 Monitor { // 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 Monitor { ) .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 Monitor { ); 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) { diff --git a/nym-api/src/network_monitor/monitor/summary_producer.rs b/nym-api/src/network_monitor/monitor/summary_producer.rs index 5f8449f75f..a55212c89c 100644 --- a/nym-api/src/network_monitor/monitor/summary_producer.rs +++ b/nym-api/src/network_monitor/monitor/summary_producer.rs @@ -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, + pub(crate) gateway_results: HashMap, +} + +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, ) } } diff --git a/nym-api/src/node_status_api/handlers/network_monitor.rs b/nym-api/src/node_status_api/handlers/network_monitor.rs index 5f71eac079..dbd3720089 100644 --- a/nym-api/src/node_status_api/handlers/network_monitor.rs +++ b/nym-api/src/node_status_api/handlers/network_monitor.rs @@ -1,6 +1,8 @@ // Copyright 2021-2024 - Nym Technologies SA // 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 { @@ -84,6 +84,18 @@ pub(super) fn network_monitor_routes() -> Router { 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( diff --git a/nym-api/src/node_status_api/handlers/unstable.rs b/nym-api/src/node_status_api/handlers/unstable.rs index 2b72f72908..1378b6a055 100644 --- a/nym-api/src/node_status_api/handlers/unstable.rs +++ b/nym-api/src/node_status_api/handlers/unstable.rs @@ -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 { + 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 { + 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, + State(state): State, +) -> AxumResult> { + 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, +) -> AxumResult> { + 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}" + ))), + } +} diff --git a/nym-api/src/support/storage/manager.rs b/nym-api/src/support/storage/manager.rs index f34ddb4045..e30b316c0d 100644 --- a/nym-api/src/support/storage/manager.rs +++ b/nym-api/src/support/storage/manager.rs @@ -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, 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, 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, + ) -> 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, 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 diff --git a/nym-api/src/support/storage/mod.rs b/nym-api/src/support/storage/mod.rs index b89062d09a..72ff9aea1d 100644 --- a/nym-api/src/support/storage/mod.rs +++ b/nym-api/src/support/storage/mod.rs @@ -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, gateway_results: Vec, test_routes: Vec, - ) -> Result<(), NymApiStorageError> { + ) -> Result { 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)>, 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, NymApiStorageError> { + Ok(self.manager.get_latest_monitor_run_id().await?) + } + pub(crate) async fn submit_mixnode_statuses_v2( &self, mixnode_results: &[NodeResult], diff --git a/nym-api/src/support/storage/models.rs b/nym-api/src/support/storage/models.rs index 1418cab9f8..e082961b01 100644 --- a/nym-api/src/support/storage/models.rs +++ b/nym-api/src/support/storage/models.rs @@ -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,