diff --git a/.gitignore b/.gitignore index 8953d2a2da..817ddac4f2 100644 --- a/.gitignore +++ b/.gitignore @@ -53,3 +53,4 @@ nym-network-monitor/topology.json nym-network-monitor/__pycache__ nym-network-monitor/*.key nym-network-monitor/.envrc +nym-network-monitor/.envrc diff --git a/Cargo.lock b/Cargo.lock index 2fd2fbf19c..d84ab9916c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2475,6 +2475,12 @@ dependencies = [ "once_cell", ] +[[package]] +name = "fallible-iterator" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7" + [[package]] name = "fancy_constructor" version = "1.2.2" @@ -5796,6 +5802,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-postgres", "tokio-util", "utoipa", "utoipa-swagger-ui", @@ -7251,6 +7258,24 @@ dependencies = [ "indexmap 2.2.6", ] +[[package]] +name = "phf" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ade2d8b8f33c7333b51bcf0428d37e217e9f32192ae4772156f65063b8ce03dc" +dependencies = [ + "phf_shared", +] + +[[package]] +name = "phf_shared" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90fcb95eef784c2ac79119d1dd819e162b5da872ce6f3c3abe1e8ca1c082f72b" +dependencies = [ + "siphasher 0.3.11", +] + [[package]] name = "pin-project" version = "1.1.6" @@ -7389,6 +7414,35 @@ version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7170ef9988bc169ba16dd36a7fa041e5c4cbeb6a35b76d4c03daded371eae7c0" +[[package]] +name = "postgres-protocol" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "acda0ebdebc28befa84bee35e651e4c5f09073d668c7aed4cf7e23c3cda84b23" +dependencies = [ + "base64 0.22.1", + "byteorder", + "bytes", + "fallible-iterator", + "hmac", + "md-5", + "memchr", + "rand", + "sha2 0.10.8", + "stringprep", +] + +[[package]] +name = "postgres-types" +version = "0.2.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f66ea23a2d0e5734297357705193335e0a957696f34bed2f2faefacb2fec336f" +dependencies = [ + "bytes", + "fallible-iterator", + "postgres-protocol", +] + [[package]] name = "powerfmt" version = "0.2.0" @@ -9723,6 +9777,32 @@ dependencies = [ "syn 2.0.90", ] +[[package]] +name = "tokio-postgres" +version = "0.7.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b5d3742945bc7d7f210693b0c58ae542c6fd47b17adbbda0885f3dcb34a6bdb" +dependencies = [ + "async-trait", + "byteorder", + "bytes", + "fallible-iterator", + "futures-channel", + "futures-util", + "log", + "parking_lot", + "percent-encoding", + "phf", + "pin-project-lite", + "postgres-protocol", + "postgres-types", + "rand", + "socket2", + "tokio", + "tokio-util", + "whoami", +] + [[package]] name = "tokio-rustls" version = "0.24.1" @@ -10913,6 +10993,7 @@ checksum = "372d5b87f58ec45c384ba03563b03544dc5fadc3983e434b286913f5b4a9bb6d" dependencies = [ "redox_syscall 0.5.1", "wasite", + "web-sys", ] [[package]] diff --git a/nym-network-monitor/Cargo.toml b/nym-network-monitor/Cargo.toml index bd537d5483..a6def79278 100644 --- a/nym-network-monitor/Cargo.toml +++ b/nym-network-monitor/Cargo.toml @@ -27,6 +27,7 @@ tokio = { workspace = true, features = ["macros", "time"] } tokio-util = { workspace = true } utoipa = { workspace = true, features = ["axum_extras"] } utoipa-swagger-ui = { workspace = true, features = ["axum"] } +tokio-postgres = "0.7" # internal nym-bin-common = { path = "../common/bin-common" } diff --git a/nym-network-monitor/entrypoint.sh b/nym-network-monitor/entrypoint.sh index 91a0830639..62b527944d 100755 --- a/nym-network-monitor/entrypoint.sh +++ b/nym-network-monitor/entrypoint.sh @@ -9,13 +9,14 @@ network=${NYM_NETWORK:-mainnet} timeout=${LOCUST_TIMEOUT:-600} users=${LOCUST_USERS:-10} processes=${LOCUST_PROCESSES:-4} +_database_url=${DATABASE_URL} -RUST_LOG=info nym-network-monitor --env envs/"${network}".env --private-key "${_private_key}" & +RUST_LOG=info nym-network-monitor --env envs/"${network}".env --private-key "${_private_key}" --database-url "${_database_url}" & nnm_pid=$! sleep 10 -python -m locust -H http://${NYM_NETWORK_MONITOR_HOST}:${NYM_NETWORK_MONITOR_PORT} --processes "${processes}" --autostart --autoquit 60 -u "${users}" -t "${timeout}"s & +python -m locust -H http://"${NYM_NETWORK_MONITOR_HOST}":"${NYM_NETWORK_MONITOR_PORT}" --processes "${processes}" --autostart --autoquit 60 -u "${users}" -t "${timeout}"s & locust_pid=$! wait $locust_pid diff --git a/nym-network-monitor/src/accounting.rs b/nym-network-monitor/src/accounting.rs index b87f795762..2fcacee75a 100644 --- a/nym-network-monitor/src/accounting.rs +++ b/nym-network-monitor/src/accounting.rs @@ -1,7 +1,10 @@ -use std::collections::{HashMap, HashSet}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; use anyhow::Result; -use futures::{stream::FuturesUnordered, StreamExt}; +use futures::{pin_mut, stream::FuturesUnordered, StreamExt}; use log::{debug, info}; use nym_sphinx::chunking::{monitoring, SentFragment}; use nym_topology::{gateway, mix, NymTopology}; @@ -10,6 +13,7 @@ use nym_validator_client::nym_api::routes::{API_VERSION, STATUS, SUBMIT_GATEWAY, use rand::SeedableRng; use rand_chacha::ChaCha8Rng; use serde::{Deserialize, Serialize}; +use tokio_postgres::{binary_copy::BinaryCopyInWriter, types::Type, Client}; use utoipa::ToSchema; use crate::{NYM_API_URL, PRIVATE_KEY, TOPOLOGY}; @@ -23,20 +27,20 @@ struct HydratedRoute { struct GatewayStats(u32, u32); impl GatewayStats { - fn new(sent: u32, recv: u32) -> Self { - GatewayStats(sent, recv) + fn new(success: u32, failure: u32) -> Self { + GatewayStats(success, failure) } fn success(&self) -> u32 { self.0 } - fn failed(&self) -> u32 { + fn failure(&self) -> u32 { self.1 } fn reliability(&self) -> f64 { - self.success() as f64 / (self.success() + self.failed()) as f64 + self.success() as f64 / (self.success() + self.failure()) as f64 } fn incr_success(&mut self) { @@ -321,48 +325,125 @@ pub async fn monitor_mixnode_results() -> anyhow::Result> { .collect()) } -pub async fn submit_metrics() -> anyhow::Result<()> { - let node_stats = monitor_mixnode_results().await?; - let gateway_stats = monitor_gateway_results().await?; +async fn submit_node_stats_to_db(client: Arc) -> anyhow::Result<()> { + let client = Arc::clone(&client); + let node_stats = all_node_stats().await?; - info!("Submitting metrics to {}", *NYM_API_URL); - let client = reqwest::Client::new(); + let sink = client + .copy_in("COPY node_stats (node_id, identity, reliability, complete_routes, incomplete_routes) FROM STDIN BINARY") + .await?; - let node_submit_url = format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_NODE}", &*NYM_API_URL); - let gateway_submit_url = format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_GATEWAY}", &*NYM_API_URL); + let writer = BinaryCopyInWriter::new( + sink, + &[Type::INT4, Type::TEXT, Type::FLOAT8, Type::INT8, Type::INT8], + ); + pin_mut!(writer); - info!("Submitting {} mixnode measurements", node_stats.len()); + for stat in node_stats { + writer + .as_mut() + .write(&[ + &(stat.mix_id as i32), + &stat.identity, + &stat.reliability, + &(stat.complete_routes as i64), + &(stat.incomplete_routes as i64), + ]) + .await?; + } - node_stats - .chunks(10) - .map(|chunk| { - let monitor_message = - MonitorMessage::new(chunk.to_vec(), PRIVATE_KEY.get().expect("We've set this!")); - client.post(&node_submit_url).json(&monitor_message).send() - }) - .collect::>() - .collect::>>() - .await - .into_iter() - .collect::, _>>()?; + writer.finish().await?; - info!("Submitting {} gateway measurements", gateway_stats.len()); + Ok(()) +} - gateway_stats - .chunks(10) - .map(|chunk| { - let monitor_message = - MonitorMessage::new(chunk.to_vec(), PRIVATE_KEY.get().expect("We've set this!")); - client - .post(&gateway_submit_url) - .json(&monitor_message) - .send() - }) - .collect::>() - .collect::>>() - .await - .into_iter() - .collect::, _>>()?; +async fn submit_gateway_stats_to_db(client: Arc) -> anyhow::Result<()> { + let client = Arc::clone(&client); + let network_account = NetworkAccount::finalize()?; + let gateway_stats = network_account.gateway_stats; + + let sink = client + .copy_in("COPY gateway_stats (identity, reliability, success, failure) FROM STDIN BINARY") + .await?; + + let writer = BinaryCopyInWriter::new(sink, &[Type::TEXT, Type::FLOAT8, Type::INT8, Type::INT8]); + pin_mut!(writer); + + for (key, stats) in gateway_stats { + writer + .as_mut() + .write(&[ + &key, + &stats.reliability(), + &(stats.success() as i64), + &(stats.failure() as i64), + ]) + .await?; + } + + writer.finish().await?; + + Ok(()) +} + +pub async fn submit_metrics_to_db(client: Arc) -> anyhow::Result<()> { + let client = Arc::clone(&client); + let client2 = Arc::clone(&client); + submit_node_stats_to_db(client).await?; + submit_gateway_stats_to_db(client2).await?; + Ok(()) +} + +pub async fn submit_metrics(client: Option>) -> anyhow::Result<()> { + if let Some(client) = client { + submit_metrics_to_db(client).await?; + } + + if let Some(private_key) = PRIVATE_KEY.get() { + let node_stats = monitor_mixnode_results().await?; + let gateway_stats = monitor_gateway_results().await?; + + info!("Submitting metrics to {}", *NYM_API_URL); + let client = reqwest::Client::new(); + + let node_submit_url = format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_NODE}", &*NYM_API_URL); + let gateway_submit_url = + format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_GATEWAY}", &*NYM_API_URL); + + info!("Submitting {} mixnode measurements", node_stats.len()); + + node_stats + .chunks(10) + .map(|chunk| { + let monitor_message = MonitorMessage::new(chunk.to_vec(), private_key); + client.post(&node_submit_url).json(&monitor_message).send() + }) + .collect::>() + .collect::>>() + .await + .into_iter() + .collect::, _>>()?; + + info!("Submitting {} gateway measurements", gateway_stats.len()); + + gateway_stats + .chunks(10) + .map(|chunk| { + let monitor_message = MonitorMessage::new( + chunk.to_vec(), + PRIVATE_KEY.get().expect("We've set this!"), + ); + client + .post(&gateway_submit_url) + .json(&monitor_message) + .send() + }) + .collect::>() + .collect::>>() + .await + .into_iter() + .collect::, _>>()?; + } NetworkAccount::empty_buffers(); diff --git a/nym-network-monitor/src/main.rs b/nym-network-monitor/src/main.rs index 76ec69440d..89fee6df6f 100644 --- a/nym-network-monitor/src/main.rs +++ b/nym-network-monitor/src/main.rs @@ -2,7 +2,7 @@ use crate::http::HttpServer; use accounting::submit_metrics; use anyhow::Result; use clap::Parser; -use log::{info, warn}; +use log::{error, info, warn}; use nym_client_core::ForgetMe; use nym_crypto::asymmetric::ed25519::PrivateKey; use nym_network_defaults::setup_env; @@ -22,6 +22,7 @@ use std::{ }; use tokio::sync::OnceCell; use tokio::{signal::ctrl_c, sync::RwLock}; +use tokio_postgres::NoTls; use tokio_util::sync::CancellationToken; static NYM_API_URL: LazyLock = LazyLock::new(|| { @@ -136,7 +137,10 @@ struct Args { generate_key_pair: bool, #[arg(long)] - private_key: String, + private_key: Option, + + #[arg(long, env = "DATABASE_URL")] + database_url: Option, } fn generate_key_pair() -> Result<()> { @@ -174,8 +178,10 @@ async fn main() -> Result<()> { std::process::exit(0); } - let pk = PrivateKey::from_base58_string(&args.private_key)?; - PRIVATE_KEY.set(pk).ok(); + if let Some(private_key) = args.private_key { + let pk = PrivateKey::from_base58_string(&private_key)?; + PRIVATE_KEY.set(pk).ok(); + } TOPOLOGY .set(if let Some(topology_file) = args.topology { @@ -203,16 +209,31 @@ async fn main() -> Result<()> { info!("Waiting for message (ctrl-c to exit)"); + let client = if let Some(database_url) = args.database_url { + let (client, connection) = tokio_postgres::connect(&database_url, NoTls).await?; + + tokio::spawn(async move { + if let Err(e) = connection.await { + error!("Postgres connection error: {}", e); + } + }); + + Some(Arc::new(client)) + } else { + None + }; + loop { + let client = client.as_ref().map(Arc::clone); match tokio::time::timeout(Duration::from_secs(600), ctrl_c()).await { Ok(_) => { info!("Received kill signal, shutting down, submitting final batch of metrics"); - submit_metrics().await?; + submit_metrics(client).await?; break; } Err(_) => { info!("Submitting metrics, cleaning metric buffers"); - submit_metrics().await?; + submit_metrics(client).await?; } }; }