From be403c6ee86a4a1ff4584d87c8a4514973335398 Mon Sep 17 00:00:00 2001 From: durch Date: Fri, 4 Jul 2025 14:50:53 +0200 Subject: [PATCH] feat(nym-node-status-agent): add multi-API support with random selection Agents can now connect to multiple APIs and randomly select one for each testrun: - Accept multiple --server arguments in format "address:port:auth_key" - Randomly shuffle server list before attempting connections - Try each server until a testrun is obtained - Submit results back only to the API that provided the testrun - Continue to next server if one is down or has no testruns available --- .../nym-node-status-agent/src/cli/mod.rs | 75 ++++++++++++------ .../src/cli/run_probe.rs | 79 +++++++++++++++---- .../nym-node-status-api/src/db/mod.rs | 12 +-- .../src/db/queries/summary.rs | 2 +- 4 files changed, 120 insertions(+), 48 deletions(-) diff --git a/nym-node-status-api/nym-node-status-agent/src/cli/mod.rs b/nym-node-status-api/nym-node-status-agent/src/cli/mod.rs index 75601548ed..57a5e33e6f 100644 --- a/nym-node-status-api/nym-node-status-agent/src/cli/mod.rs +++ b/nym-node-status-api/nym-node-status-agent/src/cli/mod.rs @@ -1,17 +1,45 @@ use crate::probe::GwProbe; use clap::{Parser, Subcommand}; use nym_bin_common::bin_info; +use nym_crypto::asymmetric::ed25519::PrivateKey; use std::sync::OnceLock; pub(crate) mod generate_keypair; pub(crate) mod run_probe; +#[derive(Debug)] +pub(crate) struct ServerConfig { + pub(crate) address: String, + pub(crate) port: u16, + pub(crate) auth_key: PrivateKey, +} + // Helper for passing LONG_VERSION to clap fn pretty_build_info_static() -> &'static str { static PRETTY_BUILD_INFORMATION: OnceLock = OnceLock::new(); PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print()) } +fn parse_server_config(s: &str) -> Result { + let parts: Vec<&str> = s.split(':').collect(); + if parts.len() != 3 { + return Err("Server config must be in format 'address:port:auth_key'".to_string()); + } + + let address = parts[0].to_string(); + let port = parts[1] + .parse::() + .map_err(|_| "Invalid port number".to_string())?; + let auth_key = + PrivateKey::from_base58_string(parts[2]).map_err(|_| "Invalid auth key".to_string())?; + + Ok(ServerConfig { + address, + port, + auth_key, + }) +} + #[derive(Parser, Debug)] #[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)] pub(crate) struct Args { @@ -22,15 +50,10 @@ pub(crate) struct Args { #[derive(Subcommand, Debug)] pub(crate) enum Command { RunProbe { - #[arg(short, long, env = "NODE_STATUS_AGENT_SERVER_ADDRESS")] - server_address: String, - - #[arg(short = 'p', long, env = "NODE_STATUS_AGENT_SERVER_PORT")] - server_port: u16, - - /// base58-encoded private key - #[arg(long, env = "NODE_STATUS_AGENT_AUTH_KEY")] - ns_api_auth_key: String, + /// Server configurations in format "address:port:auth_key" + /// Can be specified multiple times for multiple servers + #[arg(short, long, required = true)] + server: Vec, /// path of binary to run #[arg(long, env = "NODE_STATUS_AGENT_PROBE_PATH")] @@ -54,22 +77,28 @@ impl Args { pub(crate) async fn execute(&self) -> anyhow::Result<()> { match &self.command { Command::RunProbe { - server_address, - server_port, - ns_api_auth_key, + server, probe_path, probe_extra_args, - } => run_probe::run_probe( - server_address, - server_port.to_owned(), - ns_api_auth_key, - probe_path, - probe_extra_args, - ) - .await - .inspect_err(|err| { - tracing::error!("{err}"); - })?, + } => { + // Parse server configs + let mut servers = Vec::new(); + for s in server { + match parse_server_config(s) { + Ok(config) => servers.push(config), + Err(e) => { + tracing::error!("Invalid server config '{}': {}", s, e); + anyhow::bail!("Invalid server config '{}': {}", s, e); + } + } + } + + run_probe::run_probe(&servers, probe_path, probe_extra_args) + .await + .inspect_err(|err| { + tracing::error!("{err}"); + })? + } Command::GenerateKeypair { path } => { let path = path .to_owned() diff --git a/nym-node-status-api/nym-node-status-agent/src/cli/run_probe.rs b/nym-node-status-api/nym-node-status-agent/src/cli/run_probe.rs index 8cf779cfbc..8479a51825 100644 --- a/nym-node-status-api/nym-node-status-agent/src/cli/run_probe.rs +++ b/nym-node-status-api/nym-node-status-agent/src/cli/run_probe.rs @@ -1,33 +1,80 @@ -use crate::cli::GwProbe; +use crate::cli::{GwProbe, ServerConfig}; use anyhow::Context; -use nym_crypto::asymmetric::ed25519::PrivateKey; +use rand::seq::SliceRandom; pub(crate) async fn run_probe( - server_ip: &str, - server_port: u16, - ns_api_auth_key: &str, + servers: &[ServerConfig], probe_path: &str, probe_extra_args: &Vec, ) -> anyhow::Result<()> { - let auth_key = PrivateKey::from_base58_string(ns_api_auth_key) - .context("Couldn't parse auth key, exiting")?; - - let ns_api_client = nym_node_status_client::NsApiClient::new(server_ip, server_port, auth_key); + if servers.is_empty() { + anyhow::bail!("No servers configured"); + } let probe = GwProbe::new(probe_path.to_string()); let version = probe.version().await; tracing::info!("Probe version:\n{}", version); - if let Some(testrun) = ns_api_client.request_testrun().await? { - let log = probe.run_and_get_log(&Some(testrun.gateway_identity_key), probe_extra_args); + // Create indices and shuffle them for random selection + let mut indices: Vec = (0..servers.len()).collect(); + indices.shuffle(&mut rand::thread_rng()); - ns_api_client - .submit_results(testrun.testrun_id, log, testrun.assigned_at_utc) - .await?; - } else { - tracing::info!("No testruns available, exiting") + for idx in indices { + let server = &servers[idx]; + tracing::info!("Trying server: {}:{}", server.address, server.port); + + // Clone the auth key by converting to/from bytes + let auth_key = + nym_crypto::asymmetric::ed25519::PrivateKey::from_bytes(&server.auth_key.to_bytes()) + .expect("Failed to clone auth key"); + let ns_api_client = + nym_node_status_client::NsApiClient::new(&server.address, server.port, auth_key); + + match ns_api_client.request_testrun().await { + Ok(Some(testrun)) => { + tracing::info!( + "Received testrun {} from {}:{}", + testrun.testrun_id, + server.address, + server.port + ); + + let log = + probe.run_and_get_log(&Some(testrun.gateway_identity_key), probe_extra_args); + + ns_api_client + .submit_results(testrun.testrun_id, log, testrun.assigned_at_utc) + .await + .context("Failed to submit results")?; + + tracing::info!( + "Successfully submitted results to {}:{}", + server.address, + server.port + ); + return Ok(()); + } + Ok(None) => { + tracing::info!( + "No testruns available from {}:{}", + server.address, + server.port + ); + continue; + } + Err(e) => { + tracing::warn!( + "Failed to contact {}:{} - {}", + server.address, + server.port, + e + ); + continue; + } + } } + tracing::info!("No testruns available from any API"); Ok(()) } diff --git a/nym-node-status-api/nym-node-status-api/src/db/mod.rs b/nym-node-status-api/nym-node-status-api/src/db/mod.rs index ecc4e85446..f9d7c34596 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/mod.rs @@ -18,11 +18,7 @@ use sqlx::{ }; #[cfg(feature = "pg")] -use sqlx::{ - migrate::Migrator, - postgres::PgConnectOptions, - ConnectOptions, PgPool, -}; +use sqlx::{migrate::Migrator, postgres::PgConnectOptions, ConnectOptions, PgPool}; #[cfg(feature = "sqlite")] static MIGRATOR: Migrator = sqlx::migrate!("./migrations"); @@ -72,8 +68,8 @@ impl Storage { #[cfg(feature = "pg")] pub async fn init(connection_url: String, _busy_timeout: Duration) -> Result { - let connect_options = PgConnectOptions::from_str(&connection_url)? - .disable_statement_logging(); + let connect_options = + PgConnectOptions::from_str(&connection_url)?.disable_statement_logging(); let pool = sqlx::PgPool::connect_with(connect_options) .await @@ -112,4 +108,4 @@ impl Storage { Ok(()) } -} \ No newline at end of file +} diff --git a/nym-node-status-api/nym-node-status-api/src/db/queries/summary.rs b/nym-node-status-api/nym-node-status-api/src/db/queries/summary.rs index 5c5ea976bc..3695c4481b 100644 --- a/nym-node-status-api/nym-node-status-api/src/db/queries/summary.rs +++ b/nym-node-status-api/nym-node-status-api/src/db/queries/summary.rs @@ -55,7 +55,7 @@ async fn get_summary_dto(pool: &DbPool) -> anyhow::Result> { key, value_json, last_updated_utc - FROM summary"# + FROM summary"#, ) .fetch(&mut *conn) .try_collect::>()