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

Validator API storage fixes and adjustments

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2021-07-22 15:15:08 +01:00
parent 432e1e6065
commit c81454e00b
14 changed files with 1273 additions and 1206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
validator-api/migrations/20210714120000_create_initial_tables.sql → validator-api/migrations/20210722120000_create_initial_tables.sql
+6 -6
View File
@@ -1,15 +1,15 @@
CREATE TABLE mixnode_details
(
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
owner VARCHAR NOT NULL UNIQUE,
pub_key VARCHAR NOT NULL UNIQUE
owner VARCHAR NOT NULL,
identity VARCHAR NOT NULL UNIQUE
);
CREATE TABLE gateway_details
(
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
owner VARCHAR NOT NULL UNIQUE,
pub_key VARCHAR NOT NULL UNIQUE
owner VARCHAR NOT NULL,
identity VARCHAR NOT NULL UNIQUE
);
create TABLE mixnode_historical_uptime
@@ -87,5 +87,5 @@ INDEX `gateway_ipv4_status_index` ON `gateway_ipv4_status` (`gateway_details_id`
CREATE
INDEX `gateway_ipv6_status_index` ON `gateway_ipv6_status` (`gateway_details_id`, `timestamp` desc);
CREATE INDEX `mixnode_identity_index` ON `mixnode_details` (`id`, `pub_key`);
CREATE INDEX `gateway_identity_index` ON `gateway_details` (`id`, `pub_key`);
CREATE INDEX `mixnode_identity_index` ON `mixnode_details` (`id`, `identity`);
CREATE INDEX `gateway_identity_index` ON `gateway_details` (`id`, `identity`);
validator-api/sqlx-data.json
+132 -132
View File
@@ -1,17 +1,7 @@
{
"db": "SQLite",
"01de0a1f0b7c432dfdcad2cc2dedd2c1fa2c9618a0166207a1920e3b25d2b7e7": {
"query": "DELETE FROM mixnode_ipv4_status WHERE timestamp < ?",
"describe": {
"columns": [],
"parameters": {
"Right": 1
},
"nullable": []
}
},
"1915fb818696abb354f1bc7915399be69fb7b97204f7fb81baba4f4be1777fce": {
"query": "\n SELECT date, ipv4_uptime, ipv6_uptime\n FROM gateway_historical_uptime\n JOIN gateway_details\n ON gateway_historical_uptime.gateway_details_id = gateway_details.id\n WHERE gateway_details.pub_key = ?\n ORDER BY date ASC\n ",
"000bae8722e573327cbd660fc3df5c96d0b6d0f56a872e486557bbbe7e423638": {
"query": "\n SELECT date, ipv4_uptime, ipv6_uptime\n FROM gateway_historical_uptime\n JOIN gateway_details\n ON gateway_historical_uptime.gateway_details_id = gateway_details.id\n WHERE gateway_details.identity = ?\n ORDER BY date ASC\n ",
"describe": {
"columns": [
{
@@ -40,6 +30,16 @@
]
}
},
"01de0a1f0b7c432dfdcad2cc2dedd2c1fa2c9618a0166207a1920e3b25d2b7e7": {
"query": "DELETE FROM mixnode_ipv4_status WHERE timestamp < ?",
"describe": {
"columns": [],
"parameters": {
"Right": 1
},
"nullable": []
}
},
"20b7eee88b4bfb196d0dbbbef19d789d9ffd0790349b7ae6bea6b536123be3b3": {
"query": "INSERT INTO mixnode_historical_uptime(mixnode_details_id, date, ipv4_uptime, ipv6_uptime) VALUES (?, ?, ?, ?)",
"describe": {
@@ -50,6 +50,24 @@
"nullable": []
}
},
"21671ecfb386373f57a260385b6ca5ae0eead07b371805a5b1d37ae0092625f9": {
"query": "SELECT owner FROM mixnode_details WHERE identity = ?",
"describe": {
"columns": [
{
"name": "owner",
"ordinal": 0,
"type_info": "Text"
}
],
"parameters": {
"Right": 1
},
"nullable": [
false
]
}
},
"26a6cec333e3187d1bee9a583529b4126cb12d711771c5276e9598ccaa0a144d": {
"query": "\n INSERT INTO mixnode_ipv4_status (mixnode_details_id, up, timestamp) VALUES (?, ?, ?);\n ",
"describe": {
@@ -70,18 +88,18 @@
"nullable": []
}
},
"2d87cc44c28330492c7cf90589e5d8113ccff07d6f67c485226bbe59246db522": {
"query": "\n INSERT OR IGNORE INTO gateway_details(pub_key, owner) VALUES (?, ?);\n SELECT id FROM gateway_details WHERE pub_key = ?;\n ",
"32bb50f08facfb033a0eeb61b74ce546bd36263f364fe0dfebcd0c9ac8818af6": {
"query": "SELECT owner FROM gateway_details WHERE identity = ?",
"describe": {
"columns": [
{
"name": "id",
"name": "owner",
"ordinal": 0,
"type_info": "Int64"
"type_info": "Text"
}
],
"parameters": {
"Right": 3
"Right": 1
},
"nullable": [
false
@@ -136,6 +154,36 @@
]
}
},
"49b59f0fcede59a095a47be3acb9035f441d6d533443ae73f85c553b346be356": {
"query": "\n SELECT DISTINCT identity, owner, id\n FROM mixnode_details\n JOIN mixnode_ipv4_status\n ON mixnode_details.id = mixnode_ipv4_status.mixnode_details_id\n WHERE EXISTS (\n SELECT 1 FROM mixnode_ipv4_status WHERE timestamp > ?\n )\n ",
"describe": {
"columns": [
{
"name": "identity",
"ordinal": 0,
"type_info": "Text"
},
{
"name": "owner",
"ordinal": 1,
"type_info": "Text"
},
{
"name": "id",
"ordinal": 2,
"type_info": "Int64"
}
],
"parameters": {
"Right": 1
},
"nullable": [
false,
false,
false
]
}
},
"4c32c31247672549621c5b817625228f1a777919ae711d0d29a192494183e9d3": {
"query": "DELETE FROM gateway_ipv6_status WHERE timestamp < ?",
"describe": {
@@ -156,30 +204,6 @@
"nullable": []
}
},
"6b421289099f59d2f61287e87152621039d47b311dd2474887dcc91c0ca7db7e": {
"query": "\n SELECT timestamp, up\n FROM mixnode_ipv4_status\n JOIN mixnode_details\n ON mixnode_ipv4_status.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.pub_key=? AND mixnode_ipv4_status.timestamp > ?;\n ",
"describe": {
"columns": [
{
"name": "timestamp",
"ordinal": 0,
"type_info": "Int64"
},
{
"name": "up",
"ordinal": 1,
"type_info": "Bool"
}
],
"parameters": {
"Right": 2
},
"nullable": [
false,
false
]
}
},
"76c9645a1d79ce4294bff73e0fb60d211490bd810687f8e7d9963d05e7970cee": {
"query": "DELETE FROM gateway_ipv4_status WHERE timestamp < ?",
"describe": {
@@ -190,8 +214,56 @@
"nullable": []
}
},
"9212a4484ce59c30e9ef7d9e6796eb837e00e618db6f21b711cbf17661011054": {
"query": "\n SELECT timestamp, up\n FROM mixnode_ipv6_status\n JOIN mixnode_details\n ON mixnode_ipv6_status.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.pub_key=? AND mixnode_ipv6_status.timestamp > ?;\n ",
"786cf93daf23b03a5e6ffcf15f03013375be1f94a43aa8b614c9240d7dee6da4": {
"query": "\n INSERT OR IGNORE INTO gateway_details(identity, owner) VALUES (?, ?);\n SELECT id FROM gateway_details WHERE identity = ?;\n ",
"describe": {
"columns": [
{
"name": "id",
"ordinal": 0,
"type_info": "Int64"
}
],
"parameters": {
"Right": 3
},
"nullable": [
false
]
}
},
"7b37a1a850def3f82ab64c4825af256ef79a6e8a7edad122d4770be8bc84e32a": {
"query": "\n SELECT DISTINCT identity, owner, id\n FROM gateway_details\n JOIN gateway_ipv4_status\n ON gateway_details.id = gateway_ipv4_status.gateway_details_id\n WHERE EXISTS (\n SELECT 1 FROM gateway_ipv4_status WHERE timestamp > ?\n )\n ",
"describe": {
"columns": [
{
"name": "identity",
"ordinal": 0,
"type_info": "Text"
},
{
"name": "owner",
"ordinal": 1,
"type_info": "Text"
},
{
"name": "id",
"ordinal": 2,
"type_info": "Int64"
}
],
"parameters": {
"Right": 1
},
"nullable": [
false,
false,
false
]
}
},
"8395b08fa17d904217b19c5d03ab4d66dd0a0f648487d81ae09b1f926111d614": {
"query": "\n SELECT timestamp, up\n FROM mixnode_ipv4_status\n JOIN mixnode_details\n ON mixnode_ipv4_status.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.identity=? AND mixnode_ipv4_status.timestamp > ?;\n ",
"describe": {
"columns": [
{
@@ -248,44 +320,8 @@
"nullable": []
}
},
"97e108e08c4a8aa2c0e2a32f93ae7c65193a26be2d06f8f97ab8fdc0f7004fa4": {
"query": "SELECT owner FROM gateway_details WHERE pub_key = ?",
"describe": {
"columns": [
{
"name": "owner",
"ordinal": 0,
"type_info": "Text"
}
],
"parameters": {
"Right": 1
},
"nullable": [
false
]
}
},
"9cd914e2f1e8058c6935a50f2a02c0e04c4ee5a1c4a400f08d9d0eb47f5665d8": {
"query": "SELECT owner FROM mixnode_details WHERE pub_key = ?",
"describe": {
"columns": [
{
"name": "owner",
"ordinal": 0,
"type_info": "Text"
}
],
"parameters": {
"Right": 1
},
"nullable": [
false
]
}
},
"aa9708202a664aa8b85e8881deb9f0137efd91f88eba18a7fa05ee1ad7e8e621": {
"query": "\n SELECT date, ipv4_uptime, ipv6_uptime\n FROM mixnode_historical_uptime\n JOIN mixnode_details\n ON mixnode_historical_uptime.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.pub_key = ?\n ORDER BY date ASC\n ",
"96bf94cf7b81a527718b74bd456587f00f12827c72792eff9346e0b1683feede": {
"query": "\n SELECT date, ipv4_uptime, ipv6_uptime\n FROM mixnode_historical_uptime\n JOIN mixnode_details\n ON mixnode_historical_uptime.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.identity = ?\n ORDER BY date ASC\n ",
"describe": {
"columns": [
{
@@ -314,38 +350,26 @@
]
}
},
"b34e1f8595764d2706e80b317c7aed9ee9a11b23028aa44ff56ce840349e0ff9": {
"query": "\n SELECT DISTINCT pub_key, owner, id\n FROM gateway_details\n JOIN gateway_ipv4_status\n ON gateway_details.id = gateway_ipv4_status.gateway_details_id\n WHERE EXISTS (\n SELECT 1 FROM gateway_ipv4_status WHERE timestamp > ?\n )\n ",
"ba7d4f4134902f6ccec1973c9acfc5052539d35911ba238607ce8ca6111f219d": {
"query": "\n INSERT OR IGNORE INTO mixnode_details(identity, owner) VALUES (?, ?);\n SELECT id FROM mixnode_details WHERE identity = ?;\n ",
"describe": {
"columns": [
{
"name": "pub_key",
"ordinal": 0,
"type_info": "Text"
},
{
"name": "owner",
"ordinal": 1,
"type_info": "Text"
},
{
"name": "id",
"ordinal": 2,
"ordinal": 0,
"type_info": "Int64"
}
],
"parameters": {
"Right": 1
"Right": 3
},
"nullable": [
false,
false,
false
]
}
},
"b8e05a65bf3a6f0d4173e0c27cb5491e4f5d1615407c9b46651fcc2dd5970feb": {
"query": "\n SELECT timestamp, up\n FROM gateway_ipv4_status\n JOIN gateway_details\n ON gateway_ipv4_status.gateway_details_id = gateway_details.id\n WHERE gateway_details.pub_key=? AND gateway_ipv4_status.timestamp > ?;\n ",
"bcd2069178c3f006a640983a1bc8d8d0293d163044f30b859b8878a96787333b": {
"query": "\n SELECT timestamp, up\n FROM gateway_ipv4_status\n JOIN gateway_details\n ON gateway_ipv4_status.gateway_details_id = gateway_details.id\n WHERE gateway_details.identity=? AND gateway_ipv4_status.timestamp > ?;\n ",
"describe": {
"columns": [
{
@@ -378,56 +402,32 @@
"nullable": []
}
},
"d378bade0e786e3efdda4a02a5f403da9a23c20f4eecb86fceb18930cf2cceec": {
"query": "\n SELECT DISTINCT pub_key, owner, id\n FROM mixnode_details\n JOIN mixnode_ipv4_status\n ON mixnode_details.id = mixnode_ipv4_status.mixnode_details_id\n WHERE EXISTS (\n SELECT 1 FROM mixnode_ipv4_status WHERE timestamp > ?\n )\n ",
"bfc4a0aec14ac76da012f011afb8ff637666b67d8585f2cbb681520a93f3b472": {
"query": "\n SELECT timestamp, up\n FROM mixnode_ipv6_status\n JOIN mixnode_details\n ON mixnode_ipv6_status.mixnode_details_id = mixnode_details.id\n WHERE mixnode_details.identity=? AND mixnode_ipv6_status.timestamp > ?;\n ",
"describe": {
"columns": [
{
"name": "pub_key",
"name": "timestamp",
"ordinal": 0,
"type_info": "Text"
"type_info": "Int64"
},
{
"name": "owner",
"name": "up",
"ordinal": 1,
"type_info": "Text"
},
{
"name": "id",
"ordinal": 2,
"type_info": "Int64"
"type_info": "Bool"
}
],
"parameters": {
"Right": 1
"Right": 2
},
"nullable": [
false,
false,
false
]
}
},
"d4b9a3e22a1230635523aec12a0b5a980e99949232369ebd748c70fed3f4532e": {
"query": "\n INSERT OR IGNORE INTO mixnode_details(pub_key, owner) VALUES (?, ?);\n SELECT id FROM mixnode_details WHERE pub_key = ?;\n ",
"describe": {
"columns": [
{
"name": "id",
"ordinal": 0,
"type_info": "Int64"
}
],
"parameters": {
"Right": 3
},
"nullable": [
false
]
}
},
"d85a0985f6e0d818217f6733a1df0c9b83821ecf0e824b0a606c3723ad87362a": {
"query": "\n SELECT timestamp, up\n FROM gateway_ipv6_status\n JOIN gateway_details\n ON gateway_ipv6_status.gateway_details_id = gateway_details.id\n WHERE gateway_details.pub_key=? AND gateway_ipv6_status.timestamp > ?;\n ",
"d10fd4560017ab9229567ab3419e4fe6912f61aa4c17ffd87e3f724ec8518023": {
"query": "\n SELECT timestamp, up\n FROM gateway_ipv6_status\n JOIN gateway_details\n ON gateway_ipv6_status.gateway_details_id = gateway_details.id\n WHERE gateway_details.identity=? AND gateway_ipv6_status.timestamp > ?;\n ",
"describe": {
"columns": [
{
validator-api/src/main.rs
+2 -1
View File
@@ -8,7 +8,7 @@ use crate::cache::ValidatorCacheRefresher;
use crate::config::Config;
use crate::network_monitor::new_monitor_runnables;
use crate::network_monitor::tested_network::good_topology::parse_topology_file;
use crate::node_status_api::storage::NodeStatusStorage;
use crate::storage::NodeStatusStorage;
use ::config::NymConfig;
use anyhow::Result;
use cache::ValidatorCache;
@@ -22,6 +22,7 @@ pub(crate) mod cache;
pub(crate) mod config;
mod network_monitor;
mod node_status_api;
pub(crate) mod storage;
const MONITORING_ENABLED: &str = "enable-monitor";
const V4_TOPOLOGY_ARG: &str = "v4-topology-filepath";
validator-api/src/network_monitor/mod.rs
+1 -1
View File
@@ -14,7 +14,7 @@ use crate::network_monitor::monitor::sender::PacketSender;
use crate::network_monitor::monitor::summary_producer::SummaryProducer;
use crate::network_monitor::monitor::Monitor;
use crate::network_monitor::tested_network::TestedNetwork;
use crate::node_status_api::storage::NodeStatusStorage;
use crate::storage::NodeStatusStorage;
use crypto::asymmetric::{encryption, identity};
use futures::channel::mpsc;
use nymsphinx::addressing::clients::Recipient;
validator-api/src/network_monitor/monitor/mod.rs
+1 -1
View File
@@ -8,7 +8,7 @@ use crate::network_monitor::monitor::sender::PacketSender;
use crate::network_monitor::monitor::summary_producer::{NodeResult, SummaryProducer, TestReport};
use crate::network_monitor::test_packet::NodeType;
use crate::network_monitor::tested_network::TestedNetwork;
use crate::node_status_api::storage::NodeStatusStorage;
use crate::storage::NodeStatusStorage;
use log::{debug, info};
use std::process;
use tokio::time::{sleep, Duration, Instant};
validator-api/src/network_monitor/monitor/summary_producer.rs
+4 -3
View File
@@ -6,8 +6,9 @@ use crate::network_monitor::test_packet::{NodeType, TestPacket};
use crate::PENALISE_OUTDATED;
use std::collections::HashMap;
#[derive(Debug)]
pub(crate) struct NodeResult {
pub(crate) pub_key: String,
pub(crate) identity: String,
pub(crate) owner: String,
pub(crate) working_ipv4: bool,
pub(crate) working_ipv6: bool,
@@ -20,9 +21,9 @@ struct NodeStatus {
}
impl NodeStatus {
fn into_node_status(self, pub_key: String, owner: String) -> NodeResult {
fn into_node_status(self, identity: String, owner: String) -> NodeResult {
NodeResult {
pub_key,
identity,
owner,
working_ipv4: self.ip_v4_compatible,
working_ipv6: self.ip_v6_compatible,
validator-api/src/node_status_api/mod.rs
+1 -1
View File
@@ -1,6 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::storage;
use rocket::fairing::AdHoc;
use std::path::PathBuf;
use std::time::Duration;
@@ -8,7 +9,6 @@ use std::time::Duration;
pub(crate) mod local_guard;
pub(crate) mod models;
pub(crate) mod routes;
pub(crate) mod storage;
pub(crate) mod utils;
pub(crate) const FIFTEEN_MINUTES: Duration = Duration::from_secs(900);
validator-api/src/node_status_api/models.rs
+2 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::node_status_api::utils::{NodeStatus, NodeUptimes};
use crate::node_status_api::utils::NodeUptimes;
use rocket::http::{ContentType, Status};
use rocket::response::{self, Responder, Response};
use rocket::Request;
@@ -9,6 +9,7 @@ use serde::{Deserialize, Serialize};
use std::convert::TryFrom;
use std::fmt::{self, Display, Formatter};
use std::io::Cursor;
use crate::storage::models::NodeStatus;
// todo: put into some error enum
#[derive(Debug)]
validator-api/src/node_status_api/routes.rs
+1 -1
View File
@@ -6,7 +6,7 @@ use crate::node_status_api::models::{
ErrorResponse, GatewayStatusReport, GatewayUptimeHistory, MixnodeStatusReport,
MixnodeUptimeHistory,
};
use crate::node_status_api::storage::NodeStatusStorage;
use crate::storage::NodeStatusStorage;
use rocket::http::Status;
use rocket::serde::json::Json;
use rocket::State;
validator-api/src/node_status_api/storage.rs
-1045
View File
File diff suppressed because it is too large Load Diff
validator-api/src/node_status_api/utils.rs
+2 -14
View File
@@ -3,24 +3,12 @@
use crate::node_status_api::models::Uptime;
use crate::node_status_api::{FIFTEEN_MINUTES, ONE_HOUR};
use crate::storage::models::NodeStatus;
use sqlx::types::time::OffsetDateTime;
// Internally used struct to catch results from the database to calculate uptimes for given mixnode/gateway
pub(crate) struct NodeStatus {
pub(crate) timestamp: i64,
pub(crate) up: bool,
}
// Internally used struct to catch results from the database to find active mixnodes/gateways
pub(crate) struct ActiveNode {
pub(crate) id: i64,
pub(crate) pub_key: String,
pub(crate) owner: String,
}
// A temporary helper struct used to produce reports for active nodes.
pub(crate) struct ActiveNodeDayStatuses {
pub(crate) pub_key: String,
pub(crate) identity: String,
pub(crate) owner: String,
pub(crate) node_id: i64,
validator-api/src/storage/manager.rs
+645
View File
@@ -0,0 +1,645 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::network_monitor::monitor::summary_producer::NodeResult;
use crate::node_status_api::models::{HistoricalUptime, Uptime};
use crate::node_status_api::utils::ActiveNodeDayStatuses;
use crate::node_status_api::ONE_DAY;
use crate::storage::models::{ActiveNode, NodeStatus};
use crate::storage::UnixTimestamp;
use sqlx::types::time::OffsetDateTime;
use std::convert::TryFrom;
#[derive(Clone)]
pub(crate) struct StorageManager {
pub(super) connection_pool: sqlx::SqlitePool,
}
// all SQL goes here
impl StorageManager {
/// Tries to obtain owner value of given mixnode given its identity
pub(crate) async fn get_mixnode_owner(
&self,
identity: &str,
) -> Result<Option<String>, sqlx::Error> {
let owner = sqlx::query!(
"SELECT owner FROM mixnode_details WHERE identity = ?",
identity
)
.fetch_optional(&self.connection_pool)
.await?
.map(|row| row.owner);
Ok(owner)
}
/// Tries to obtain owner value of given gateway given its identity
pub(crate) async fn get_gateway_owner(
&self,
identity: &str,
) -> Result<Option<String>, sqlx::Error> {
let owner = sqlx::query!(
"SELECT owner FROM gateway_details WHERE identity = ?",
identity
)
.fetch_optional(&self.connection_pool)
.await?
.map(|row| row.owner);
Ok(owner)
}
/// Gets all ipv4 statuses for mixnode with particular identity that were inserted
/// into the database after the specified unix timestamp.
pub(crate) async fn get_mixnode_ipv4_statuses_since(
&self,
identity: &str,
timestamp: UnixTimestamp,
) -> Result<Vec<NodeStatus>, sqlx::Error> {
sqlx::query_as!(
NodeStatus,
r#"
SELECT timestamp, up
FROM mixnode_ipv4_status
JOIN mixnode_details
ON mixnode_ipv4_status.mixnode_details_id = mixnode_details.id
WHERE mixnode_details.identity=? AND mixnode_ipv4_status.timestamp > ?;
"#,
identity,
timestamp,
)
.fetch_all(&self.connection_pool)
.await
}
/// Gets all ipv6 statuses for mixnode with particular identity that were inserted
/// into the database after the specified unix timestamp.
pub(crate) async fn get_mixnode_ipv6_statuses_since(
&self,
identity: &str,
timestamp: UnixTimestamp,
) -> Result<Vec<NodeStatus>, sqlx::Error> {
sqlx::query_as!(
NodeStatus,
r#"
SELECT timestamp, up
FROM mixnode_ipv6_status
JOIN mixnode_details
ON mixnode_ipv6_status.mixnode_details_id = mixnode_details.id
WHERE mixnode_details.identity=? AND mixnode_ipv6_status.timestamp > ?;
"#,
identity,
timestamp
)
.fetch_all(&self.connection_pool)
.await
}
/// Gets all ipv4 statuses for gateway with particular identity that were inserted
/// into the database after the specified unix timestamp.
pub(crate) async fn get_gateway_ipv4_statuses_since(
&self,
identity: &str,
timestamp: UnixTimestamp,
) -> Result<Vec<NodeStatus>, sqlx::Error> {
sqlx::query_as!(
NodeStatus,
r#"
SELECT timestamp, up
FROM gateway_ipv4_status
JOIN gateway_details
ON gateway_ipv4_status.gateway_details_id = gateway_details.id
WHERE gateway_details.identity=? AND gateway_ipv4_status.timestamp > ?;
"#,
identity,
timestamp,
)
.fetch_all(&self.connection_pool)
.await
}
/// Gets all ipv6 statuses for gateway with particular identity that were inserted
/// into the database after the specified unix timestamp.
pub(crate) async fn get_gateway_ipv6_statuses_since(
&self,
identity: &str,
timestamp: UnixTimestamp,
) -> Result<Vec<NodeStatus>, sqlx::Error> {
sqlx::query_as!(
NodeStatus,
r#"
SELECT timestamp, up
FROM gateway_ipv6_status
JOIN gateway_details
ON gateway_ipv6_status.gateway_details_id = gateway_details.id
WHERE gateway_details.identity=? AND gateway_ipv6_status.timestamp > ?;
"#,
identity,
timestamp
)
.fetch_all(&self.connection_pool)
.await
}
/// Gets the historical daily uptime associated with the particular mixnode
pub(crate) async fn get_mixnode_historical_uptimes(
&self,
identity: &str,
) -> Result<Vec<HistoricalUptime>, sqlx::Error> {
let uptimes = sqlx::query!(
r#"
SELECT date, ipv4_uptime, ipv6_uptime
FROM mixnode_historical_uptime
JOIN mixnode_details
ON mixnode_historical_uptime.mixnode_details_id = mixnode_details.id
WHERE mixnode_details.identity = ?
ORDER BY date ASC
"#,
identity
)
.fetch_all(&self.connection_pool)
.await?
.into_iter()
// filter out nodes with valid uptime (in theory all should be 100% valid since we insert them ourselves, but
// better safe than sorry and not use an unwrap)
.filter_map(|row| {
Uptime::try_from(row.ipv4_uptime)
.ok()
.map(|ipv4_uptime| {
Uptime::try_from(row.ipv6_uptime)
.ok()
.map(|ipv6_uptime| HistoricalUptime {
date: row.date,
ipv4_uptime,
ipv6_uptime,
})
})
.flatten()
})
.collect();
Ok(uptimes)
}
/// Gets the historical daily uptime associated with the particular gateway
pub(crate) async fn get_gateway_historical_uptimes(
&self,
identity: &str,
) -> Result<Vec<HistoricalUptime>, sqlx::Error> {
let uptimes = sqlx::query!(
r#"
SELECT date, ipv4_uptime, ipv6_uptime
FROM gateway_historical_uptime
JOIN gateway_details
ON gateway_historical_uptime.gateway_details_id = gateway_details.id
WHERE gateway_details.identity = ?
ORDER BY date ASC
"#,
identity
)
.fetch_all(&self.connection_pool)
.await?
.into_iter()
// filter out nodes with valid uptime (in theory all should be 100% valid since we insert them ourselves, but
// better safe than sorry and not use an unwrap)
.filter_map(|row| {
Uptime::try_from(row.ipv4_uptime)
.ok()
.map(|ipv4_uptime| {
Uptime::try_from(row.ipv6_uptime)
.ok()
.map(|ipv6_uptime| HistoricalUptime {
date: row.date,
ipv4_uptime,
ipv6_uptime,
})
})
.flatten()
})
.collect();
Ok(uptimes)
}
/// Gets all ipv4 statuses for mixnode with particular id that were inserted
/// into the database after the specified unix timestamp.
pub(crate) async fn get_mixnode_ipv4_statuses_since_by_id(
&self,
id: i64,
timestamp: UnixTimestamp,
) -> Result<Vec<NodeStatus>, sqlx::Error> {
sqlx::query_as!(
NodeStatus,
r#"
SELECT timestamp, up
FROM mixnode_ipv4_status
WHERE mixnode_details_id=? AND timestamp > ?;
"#,
id,
timestamp
)
.fetch_all(&self.connection_pool)
.await
}
/// Gets all ipv6 statuses for mixnode with particular id that were inserted
/// into the database after the specified unix timestamp.
pub(crate) async fn get_mixnode_ipv6_statuses_since_by_id(
&self,
id: i64,
timestamp: UnixTimestamp,
) -> Result<Vec<NodeStatus>, sqlx::Error> {
sqlx::query_as!(
NodeStatus,
r#"
SELECT timestamp, up
FROM mixnode_ipv6_status
WHERE mixnode_details_id=? AND timestamp > ?;
"#,
id,
timestamp
)
.fetch_all(&self.connection_pool)
.await
}
/// Gets all ipv4 statuses for gateway with particular id that were inserted
/// into the database after the specified unix timestamp.
pub(crate) async fn get_gateway_ipv4_statuses_since_by_id(
&self,
id: i64,
timestamp: UnixTimestamp,
) -> Result<Vec<NodeStatus>, sqlx::Error> {
sqlx::query_as!(
NodeStatus,
r#"
SELECT timestamp, up
FROM gateway_ipv4_status
WHERE gateway_details_id=? AND timestamp > ?;
"#,
id,
timestamp
)
.fetch_all(&self.connection_pool)
.await
}
/// Gets all ipv6 statuses for gateway with particular id that were inserted
/// into the database after the specified unix timestamp.
pub(crate) async fn get_gateway_ipv6_statuses_since_by_id(
&self,
id: i64,
timestamp: UnixTimestamp,
) -> Result<Vec<NodeStatus>, sqlx::Error> {
sqlx::query_as!(
NodeStatus,
r#"
SELECT timestamp, up
FROM gateway_ipv6_status
WHERE gateway_details_id=? AND timestamp > ?;
"#,
id,
timestamp
)
.fetch_all(&self.connection_pool)
.await
}
/// Tries to submit mixnode [`NodeResult`] from the network monitor to the database.
pub(crate) async fn submit_mixnode_statuses(
&self,
timestamp: UnixTimestamp,
mixnode_results: Vec<NodeResult>,
) -> Result<(), sqlx::Error> {
// insert it all in a transaction to make sure all nodes are updated at the same time
// (plus it's a nice guard against new nodes)
let mut tx = self.connection_pool.begin().await?;
for mixnode_result in mixnode_results {
let mixnode_id = sqlx::query!(
r#"
INSERT OR IGNORE INTO mixnode_details(identity, owner) VALUES (?, ?);
SELECT id FROM mixnode_details WHERE identity = ?;
"#,
mixnode_result.identity,
mixnode_result.owner,
mixnode_result.identity,
)
.fetch_one(&mut tx)
.await?
.id;
// insert ipv4 status
sqlx::query!(
r#"
INSERT INTO mixnode_ipv4_status (mixnode_details_id, up, timestamp) VALUES (?, ?, ?);
"#,
mixnode_id,
mixnode_result.working_ipv4,
timestamp
)
.execute(&mut tx)
.await?;
// insert ipv6 status
sqlx::query!(
r#"
INSERT INTO mixnode_ipv6_status (mixnode_details_id, up, timestamp) VALUES (?, ?, ?);
"#,
mixnode_id,
mixnode_result.working_ipv6,
timestamp
)
.execute(&mut tx)
.await?;
}
// finally commit the transaction
tx.commit().await
}
/// Tries to submit gateway [`NodeResult`] from the network monitor to the database.
pub(crate) async fn submit_gateway_statuses(
&self,
timestamp: UnixTimestamp,
gateway_results: Vec<NodeResult>,
) -> Result<(), sqlx::Error> {
// insert it all in a transaction to make sure all nodes are updated at the same time
// (plus it's a nice guard against new nodes)
let mut tx = self.connection_pool.begin().await?;
for gateway_result in gateway_results {
// if gateway info doesn't exist, insert it and get its id
// same ID "problem" as described for mixnode insertion
let gateway_id = sqlx::query!(
r#"
INSERT OR IGNORE INTO gateway_details(identity, owner) VALUES (?, ?);
SELECT id FROM gateway_details WHERE identity = ?;
"#,
gateway_result.identity,
gateway_result.owner,
gateway_result.identity,
)
.fetch_one(&mut tx)
.await?
.id;
// insert ipv4 status
sqlx::query!(
r#"
INSERT INTO gateway_ipv4_status (gateway_details_id, up, timestamp) VALUES (?, ?, ?);
"#,
gateway_id,
gateway_result.working_ipv4,
timestamp
)
.execute(&mut tx)
.await?;
// insert ipv6 status
sqlx::query!(
r#"
INSERT INTO gateway_ipv6_status (gateway_details_id, up, timestamp) VALUES (?, ?, ?);
"#,
gateway_id,
gateway_result.working_ipv6,
timestamp
)
.execute(&mut tx)
.await?;
}
// finally commit the transaction
tx.commit().await
}
/// Checks whether there are already any historical uptimes with this particular date.
pub(crate) async fn check_for_historical_uptime_existence(
&self,
today_iso_8601: &str,
) -> Result<bool, sqlx::Error> {
sqlx::query!(
"SELECT EXISTS (SELECT 1 FROM mixnode_historical_uptime WHERE date = ?) AS 'exists'",
today_iso_8601
)
.fetch_one(&self.connection_pool)
.await
.map(|result| result.exists == 1)
}
/// Creates new entry for mixnode historical uptime
pub(crate) async fn insert_mixnode_historical_uptime(
&self,
node_id: i64,
date: &str,
ipv4_uptime: u8,
ipv6_uptime: u8,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"INSERT INTO mixnode_historical_uptime(mixnode_details_id, date, ipv4_uptime, ipv6_uptime) VALUES (?, ?, ?, ?)",
node_id,
date,
ipv4_uptime,
ipv6_uptime,
).execute(&self.connection_pool).await?;
Ok(())
}
/// Creates new entry for gatewy historical uptime
pub(crate) async fn insert_gateway_historical_uptime(
&self,
node_id: i64,
date: &str,
ipv4_uptime: u8,
ipv6_uptime: u8,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"INSERT INTO gateway_historical_uptime(gateway_details_id, date, ipv4_uptime, ipv6_uptime) VALUES (?, ?, ?, ?)",
node_id,
date,
ipv4_uptime,
ipv6_uptime,
).execute(&self.connection_pool).await?;
Ok(())
}
pub(crate) async fn purge_old_mixnode_ipv4_statuses(
&self,
timestamp: UnixTimestamp,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"DELETE FROM mixnode_ipv4_status WHERE timestamp < ?",
timestamp
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn purge_old_mixnode_ipv6_statuses(
&self,
timestamp: UnixTimestamp,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"DELETE FROM mixnode_ipv6_status WHERE timestamp < ?",
timestamp
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn purge_old_gateway_ipv4_statuses(
&self,
timestamp: UnixTimestamp,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"DELETE FROM gateway_ipv4_status WHERE timestamp < ?",
timestamp
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
pub(crate) async fn purge_old_gateway_ipv6_statuses(
&self,
timestamp: UnixTimestamp,
) -> Result<(), sqlx::Error> {
sqlx::query!(
"DELETE FROM gateway_ipv6_status WHERE timestamp < ?",
timestamp
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
// ####################################################################################################
// ALL THE METHODS BELOW ARE TEMPORARY AND WILL BE REMOVED ONCE PAYMENTS ARE DONE INSIDE VALIDATOR API
// ####################################################################################################
// NOTE: this method will go away once we move payments into the validator-api
// it just helps us to get rid of having to query for reports of each node individually
/// Returns public key, owner and id of all mixnodes that have had any ipv4 statuses submitted
/// since provided timestamp.
pub(crate) async fn get_all_active_mixnodes(
&self,
timestamp: UnixTimestamp,
) -> Result<Vec<ActiveNode>, sqlx::Error> {
// find mixnode details of all nodes that have had at least 1 ipv4 status since the provided
// timestamp
// TODO: I dont know if theres a potential issue of if we have a lot of inactive nodes that
// haven't mixed in ages, they might increase the query times?
sqlx::query_as!(
ActiveNode,
r#"
SELECT DISTINCT identity, owner, id
FROM mixnode_details
JOIN mixnode_ipv4_status
ON mixnode_details.id = mixnode_ipv4_status.mixnode_details_id
WHERE EXISTS (
SELECT 1 FROM mixnode_ipv4_status WHERE timestamp > ?
)
"#,
timestamp
)
.fetch_all(&self.connection_pool)
.await
}
// NOTE: this method will go away once we move payments into the validator-api
// it just helps us to get rid of having to query for reports of each node individually
/// Returns public key, owner and id of all gateways that have had any ipv4 statuses submitted
/// since provided timestamp.
pub(crate) async fn get_all_active_gateways(
&self,
timestamp: UnixTimestamp,
) -> Result<Vec<ActiveNode>, sqlx::Error> {
sqlx::query_as!(
ActiveNode,
r#"
SELECT DISTINCT identity, owner, id
FROM gateway_details
JOIN gateway_ipv4_status
ON gateway_details.id = gateway_ipv4_status.gateway_details_id
WHERE EXISTS (
SELECT 1 FROM gateway_ipv4_status WHERE timestamp > ?
)
"#,
timestamp
)
.fetch_all(&self.connection_pool)
.await
}
// NOTE: this method will go away once we move payments into the validator-api
// it just helps us to get rid of having to query for reports of each node individually
// TODO: should that live on the 'Inner' struct or should it rather exist on the actual storage struct
// since technically it doesn't touch any SQL directly
pub(crate) async fn get_all_active_mixnodes_statuses(
&self,
) -> Result<Vec<ActiveNodeDayStatuses>, sqlx::Error> {
let now = OffsetDateTime::now_utc();
let day_ago = (now - ONE_DAY).unix_timestamp();
let active_nodes = self.get_all_active_mixnodes(day_ago).await?;
let mut active_day_statuses = Vec::with_capacity(active_nodes.len());
for active_node in active_nodes.into_iter() {
let ipv4_statuses = self
.get_mixnode_ipv4_statuses_since_by_id(active_node.id, day_ago)
.await?;
let ipv6_statuses = self
.get_mixnode_ipv6_statuses_since_by_id(active_node.id, day_ago)
.await?;
let statuses = ActiveNodeDayStatuses {
identity: active_node.identity,
owner: active_node.owner,
node_id: active_node.id,
ipv4_statuses,
ipv6_statuses,
};
active_day_statuses.push(statuses);
}
Ok(active_day_statuses)
}
// NOTE: this method will go away once we move payments into the validator-api
// it just helps us to get rid of having to query for reports of each node individually
// TODO: should that live on the 'Inner' struct or should it rather exist on the actual storage struct
// since technically it doesn't touch any SQL directly
pub(crate) async fn get_all_active_gateways_statuses(
&self,
) -> Result<Vec<ActiveNodeDayStatuses>, sqlx::Error> {
let now = OffsetDateTime::now_utc();
let day_ago = (now - ONE_DAY).unix_timestamp();
let active_nodes = self.get_all_active_gateways(day_ago).await?;
let mut active_day_statuses = Vec::with_capacity(active_nodes.len());
for active_node in active_nodes.into_iter() {
let ipv4_statuses = self
.get_gateway_ipv4_statuses_since_by_id(active_node.id, day_ago)
.await?;
let ipv6_statuses = self
.get_gateway_ipv6_statuses_since_by_id(active_node.id, day_ago)
.await?;
let statuses = ActiveNodeDayStatuses {
identity: active_node.identity,
owner: active_node.owner,
node_id: active_node.id,
ipv4_statuses,
ipv6_statuses,
};
active_day_statuses.push(statuses);
}
Ok(active_day_statuses)
}
}
validator-api/src/storage/mod.rs
+461
View File
@@ -0,0 +1,461 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::network_monitor::monitor::summary_producer::NodeResult;
use crate::node_status_api::models::{
GatewayStatusReport, GatewayUptimeHistory, MixnodeStatusReport, MixnodeUptimeHistory,
NodeStatusApiError, Uptime,
};
use crate::node_status_api::ONE_DAY;
use crate::storage::manager::StorageManager;
use crate::storage::models::NodeStatus;
use rocket::fairing::{self, AdHoc};
use rocket::{Build, Rocket};
use sqlx::types::time::OffsetDateTime;
use sqlx::ConnectOptions;
use std::path::PathBuf;
pub(crate) mod manager;
pub(crate) mod models;
// A type alias to be more explicit about type of timestamp used.
type UnixTimestamp = i64;
// note that clone here is fine as upon cloning the same underlying pool will be used
#[derive(Clone)]
pub(crate) struct NodeStatusStorage {
manager: StorageManager,
}
impl NodeStatusStorage {
async fn init(rocket: Rocket<Build>, database_path: PathBuf) -> fairing::Result {
// TODO: we can inject here more stuff based on our validator-api global config
// struct. Maybe different pool size or timeout intervals?
let mut opts = sqlx::sqlite::SqliteConnectOptions::new()
.filename(&database_path)
.create_if_missing(true);
// TODO: do we want auto_vacuum ?
opts.disable_statement_logging();
let connection_pool = match sqlx::SqlitePool::connect_with(opts).await {
Ok(db) => db,
Err(e) => {
error!("Failed to connect to SQLx database: {}", e);
return Err(rocket);
}
};
if let Err(e) = sqlx::migrate!("./migrations").run(&connection_pool).await {
error!("Failed to initialize SQLx database: {}", e);
return Err(rocket);
}
info!("Database migration finished!");
let storage = NodeStatusStorage {
manager: StorageManager { connection_pool },
};
Ok(rocket.manage(storage))
}
pub(crate) fn stage(database_path: PathBuf) -> AdHoc {
AdHoc::try_on_ignite("SQLx Database", |rocket| {
NodeStatusStorage::init(rocket, database_path)
})
}
/// Gets all statuses for particular mixnode (ipv4 and ipv6) that were inserted in last 24h.
async fn get_mixnode_daily_statuses(
&self,
identity: &str,
) -> Result<(Vec<NodeStatus>, Vec<NodeStatus>), NodeStatusApiError> {
let now = OffsetDateTime::now_utc();
let day_ago = now - ONE_DAY;
let ipv4_statuses = self
.manager
.get_mixnode_ipv4_statuses_since(identity, day_ago.unix_timestamp())
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
let ipv6_statuses = self
.manager
.get_mixnode_ipv6_statuses_since(identity, day_ago.unix_timestamp())
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
Ok((ipv4_statuses, ipv6_statuses))
}
/// Gets all statuses for particular gateway (ipv4 and ipv6) that were inserted in last 24h.
async fn get_gateway_daily_statuses(
&self,
identity: &str,
) -> Result<(Vec<NodeStatus>, Vec<NodeStatus>), NodeStatusApiError> {
let now = OffsetDateTime::now_utc();
let day_ago = now - ONE_DAY;
let ipv4_statuses = self
.manager
.get_gateway_ipv4_statuses_since(identity, day_ago.unix_timestamp())
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
let ipv6_statuses = self
.manager
.get_gateway_ipv6_statuses_since(identity, day_ago.unix_timestamp())
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
Ok((ipv4_statuses, ipv6_statuses))
}
/// Tries to construct a status report for mixnode with the specified identity.
pub(crate) async fn construct_mixnode_report(
&self,
identity: &str,
) -> Result<MixnodeStatusReport, NodeStatusApiError> {
let (ipv4_statuses, ipv6_statuses) = self.get_mixnode_daily_statuses(identity).await?;
// if we have no statuses, the node doesn't exist (or monitor is down), but either way, we can't make a report
if ipv4_statuses.is_empty() {
return Err(NodeStatusApiError::MixnodeReportNotFound(
identity.to_owned(),
));
}
// now, technically this is not a critical error, but this should have NEVER happened in the first place
// so something super weird is going on
if ipv4_statuses.len() != ipv6_statuses.len() {
error!("Somehow we have different number of ipv4 and ipv6 statuses for mixnode {}! (ipv4: {}, ipv6: {})",
identity,
ipv4_statuses.len(),
ipv6_statuses.len(),
)
}
let mixnode_owner = self
.manager
.get_mixnode_owner(identity)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?
.expect("The node doesn't have an owner even though we have status information on it!");
Ok(MixnodeStatusReport::construct_from_last_day_reports(
identity.to_owned(),
mixnode_owner,
ipv4_statuses,
ipv6_statuses,
))
}
pub(crate) async fn construct_gateway_report(
&self,
identity: &str,
) -> Result<GatewayStatusReport, NodeStatusApiError> {
let (ipv4_statuses, ipv6_statuses) = self.get_gateway_daily_statuses(identity).await?;
// if we have no statuses, the node doesn't exist (or monitor is down), but either way, we can't make a report
if ipv4_statuses.is_empty() {
return Err(NodeStatusApiError::GatewayReportNotFound(
identity.to_owned(),
));
}
// now, technically this is not a critical error, but this should have NEVER happened in the first place
// so something super weird is going on
if ipv4_statuses.len() != ipv6_statuses.len() {
error!("Somehow we have different number of ipv4 and ipv6 statuses for gateway {}! (ipv4: {}, ipv6: {})",
identity,
ipv4_statuses.len(),
ipv6_statuses.len(),
)
}
let gateway_owner = self
.manager
.get_gateway_owner(identity)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?
.expect(
"The gateway doesn't have an owner even though we have status information on it!",
);
Ok(GatewayStatusReport::construct_from_last_day_reports(
identity.to_owned(),
gateway_owner,
ipv4_statuses,
ipv6_statuses,
))
}
pub(crate) async fn get_mixnode_uptime_history(
&self,
identity: &str,
) -> Result<MixnodeUptimeHistory, NodeStatusApiError> {
let history = self
.manager
.get_mixnode_historical_uptimes(identity)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
if history.is_empty() {
return Err(NodeStatusApiError::MixnodeUptimeHistoryNotFound(
identity.to_owned(),
));
}
let mixnode_owner = self
.manager
.get_mixnode_owner(identity)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?
.expect("The node doesn't have an owner even though we have uptime history for it!");
Ok(MixnodeUptimeHistory::new(
identity.to_owned(),
mixnode_owner,
history,
))
}
pub(crate) async fn get_gateway_uptime_history(
&self,
identity: &str,
) -> Result<GatewayUptimeHistory, NodeStatusApiError> {
let history = self
.manager
.get_gateway_historical_uptimes(identity)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
if history.is_empty() {
return Err(NodeStatusApiError::GatewayUptimeHistoryNotFound(
identity.to_owned(),
));
}
let gateway_owner = self
.manager
.get_gateway_owner(identity)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?
.expect("The gateway doesn't have an owner even though we have uptime history for it!");
Ok(GatewayUptimeHistory::new(
identity.to_owned(),
gateway_owner,
history,
))
}
// NOTE: this method will go away once we move payments into the validator-api
// it just helps us to get rid of having to query for reports of each node individually
pub(crate) async fn get_all_mixnode_reports(
&self,
) -> Result<Vec<MixnodeStatusReport>, NodeStatusApiError> {
let reports = self
.manager
.get_all_active_mixnodes_statuses()
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?
.into_iter()
.map(|statuses| {
MixnodeStatusReport::construct_from_last_day_reports(
statuses.identity,
statuses.owner,
statuses.ipv4_statuses,
statuses.ipv6_statuses,
)
})
.collect();
Ok(reports)
}
// NOTE: this method will go away once we move payments into the validator-api
// it just helps us to get rid of having to query for reports of each node individually
pub(crate) async fn get_all_gateway_reports(
&self,
) -> Result<Vec<GatewayStatusReport>, NodeStatusApiError> {
let reports = self
.manager
.get_all_active_gateways_statuses()
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?
.into_iter()
.map(|statuses| {
GatewayStatusReport::construct_from_last_day_reports(
statuses.identity,
statuses.owner,
statuses.ipv4_statuses,
statuses.ipv6_statuses,
)
})
.collect();
Ok(reports)
}
// Used by network monitor
pub(crate) async fn submit_new_statuses(
&self,
mixnode_results: Vec<NodeResult>,
gateway_results: Vec<NodeResult>,
) -> Result<(), NodeStatusApiError> {
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();
self.manager
.submit_mixnode_statuses(now, mixnode_results)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
self.manager
.submit_gateway_statuses(now, gateway_results)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)
}
// Called on timer/reward script
async fn update_historical_uptimes(
&self,
today_iso_8601: &str,
) -> Result<(), NodeStatusApiError> {
// get statuses for all active mixnodes...
let active_mixnodes_statuses = self
.manager
.get_all_active_mixnodes_statuses()
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
for statuses in active_mixnodes_statuses.into_iter() {
let ipv4_day_up = statuses
.ipv4_statuses
.iter()
.filter(|status| status.up)
.count();
let ipv6_day_up = statuses
.ipv6_statuses
.iter()
.filter(|status| status.up)
.count();
// calculate their uptimes for the last 24h
let ipv4_uptime = Uptime::from_ratio(ipv4_day_up, statuses.ipv4_statuses.len())
.unwrap()
.u8();
let ipv6_uptime = Uptime::from_ratio(ipv6_day_up, statuses.ipv6_statuses.len())
.unwrap()
.u8();
// and insert into the database
self.manager
.insert_mixnode_historical_uptime(
statuses.node_id,
today_iso_8601,
ipv4_uptime,
ipv6_uptime,
)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
}
// get statuses for all active gateways...
let active_gateways_statuses = self
.manager
.get_all_active_gateways_statuses()
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
for statuses in active_gateways_statuses.into_iter() {
let ipv4_day_up = statuses
.ipv4_statuses
.iter()
.filter(|status| status.up)
.count();
let ipv6_day_up = statuses
.ipv6_statuses
.iter()
.filter(|status| status.up)
.count();
// calculate their uptimes for the last 24h
let ipv4_uptime = Uptime::from_ratio(ipv4_day_up, statuses.ipv4_statuses.len())
.unwrap()
.u8();
let ipv6_uptime = Uptime::from_ratio(ipv6_day_up, statuses.ipv6_statuses.len())
.unwrap()
.u8();
// and insert into the database
self.manager
.insert_gateway_historical_uptime(
statuses.node_id,
today_iso_8601,
ipv4_uptime,
ipv6_uptime,
)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
}
Ok(())
}
async fn check_if_historical_uptimes_exist_for_date(
&self,
date_iso_8601: &str,
) -> Result<bool, NodeStatusApiError> {
self.manager
.check_for_historical_uptime_existence(date_iso_8601)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)
}
// Called on timer/reward script
async fn purge_old_statuses(&self) -> Result<(), NodeStatusApiError> {
let now = OffsetDateTime::now_utc();
let two_days_ago = (now - 2 * ONE_DAY).unix_timestamp();
self.manager
.purge_old_mixnode_ipv4_statuses(two_days_ago)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
self.manager
.purge_old_mixnode_ipv6_statuses(two_days_ago)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
self.manager
.purge_old_gateway_ipv4_statuses(two_days_ago)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)?;
self.manager
.purge_old_gateway_ipv6_statuses(two_days_ago)
.await
.map_err(|_| NodeStatusApiError::InternalDatabaseError)
}
pub(crate) async fn daily_chores(&self) -> Result<bool, NodeStatusApiError> {
let today_iso_8601 = OffsetDateTime::now_utc().date().to_string();
// if we have already performed the update for today's date, don't do anything
if self
.check_if_historical_uptimes_exist_for_date(&today_iso_8601)
.await?
{
Ok(false)
} else {
info!(
"Updating historical daily uptimes of all nodes and purging old status reports..."
);
self.update_historical_uptimes(&today_iso_8601).await?;
self.purge_old_statuses().await?;
Ok(true)
}
}
}
validator-api/src/storage/models.rs
+15
View File
@@ -0,0 +1,15 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
// Internally used struct to catch results from the database to calculate uptimes for given mixnode/gateway
pub(crate) struct NodeStatus {
pub(crate) timestamp: i64,
pub(crate) up: bool,
}
// Internally used struct to catch results from the database to find active mixnodes/gateways
pub(crate) struct ActiveNode {
pub(crate) id: i64,
pub(crate) identity: String,
pub(crate) owner: String,
}