Merge pull request #6093 from nymtech/bugfix/ns-api-node-custom-http-api-port
ns-api: fix scraping bug when operator specifies custom node HTTP API port in bond
This commit is contained in:
Generated
+3
-3
@@ -133,9 +133,9 @@ checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
|
||||
|
||||
[[package]]
|
||||
name = "ammonia"
|
||||
version = "4.1.1"
|
||||
version = "4.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d6b346764dd0814805de8abf899fe03065bcee69bb1a4771c785817e39f3978f"
|
||||
checksum = "17e913097e1a2124b46746c980134e8c954bc17a6a59bb3fde96f088d126dde6"
|
||||
dependencies = [
|
||||
"cssparser",
|
||||
"html5ever",
|
||||
@@ -6546,7 +6546,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "nym-node-status-api"
|
||||
version = "4.0.10"
|
||||
version = "4.0.11-rc1"
|
||||
dependencies = [
|
||||
"ammonia",
|
||||
"anyhow",
|
||||
|
||||
+8
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT\n node_id,\n ed25519_identity_pubkey,\n total_stake,\n ip_addresses as \"ip_addresses!: serde_json::Value\",\n mix_port,\n x25519_sphinx_pubkey,\n node_role as \"node_role: serde_json::Value\",\n supported_roles as \"supported_roles: serde_json::Value\",\n entry as \"entry: serde_json::Value\",\n performance,\n self_described as \"self_described: serde_json::Value\",\n bond_info as \"bond_info: serde_json::Value\"\n FROM\n nym_nodes\n ORDER BY\n node_id\n ",
|
||||
"query": "SELECT\n node_id,\n ed25519_identity_pubkey,\n total_stake,\n ip_addresses as \"ip_addresses!: serde_json::Value\",\n mix_port,\n x25519_sphinx_pubkey,\n node_role as \"node_role: serde_json::Value\",\n supported_roles as \"supported_roles: serde_json::Value\",\n entry as \"entry: serde_json::Value\",\n performance,\n self_described as \"self_described: serde_json::Value\",\n bond_info as \"bond_info: serde_json::Value\",\n http_api_port\n FROM\n nym_nodes\n WHERE\n self_described IS NOT NULL\n AND\n bond_info IS NOT NULL\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -62,6 +62,11 @@
|
||||
"ordinal": 11,
|
||||
"name": "bond_info: serde_json::Value",
|
||||
"type_info": "Jsonb"
|
||||
},
|
||||
{
|
||||
"ordinal": 12,
|
||||
"name": "http_api_port",
|
||||
"type_info": "Int4"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
@@ -79,8 +84,9 @@
|
||||
true,
|
||||
false,
|
||||
true,
|
||||
true,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "c48d04fc3de59dd484f0a63d40336ced54e08785f77e9ef85f3157d004ec85dc"
|
||||
"hash": "0b51df277ed66c6553f66af9b135342dee177abc1c92e4a89147de3c22d3d1a5"
|
||||
}
|
||||
+8
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT\n node_id,\n ed25519_identity_pubkey,\n total_stake,\n ip_addresses as \"ip_addresses!: serde_json::Value\",\n mix_port,\n x25519_sphinx_pubkey,\n node_role as \"node_role: serde_json::Value\",\n supported_roles as \"supported_roles: serde_json::Value\",\n entry as \"entry: serde_json::Value\",\n performance,\n self_described as \"self_described: serde_json::Value\",\n bond_info as \"bond_info: serde_json::Value\"\n FROM\n nym_nodes\n WHERE\n self_described IS NOT NULL\n AND\n bond_info IS NOT NULL\n ",
|
||||
"query": "SELECT\n node_id,\n ed25519_identity_pubkey,\n total_stake,\n ip_addresses as \"ip_addresses!: serde_json::Value\",\n mix_port,\n x25519_sphinx_pubkey,\n node_role as \"node_role: serde_json::Value\",\n supported_roles as \"supported_roles: serde_json::Value\",\n entry as \"entry: serde_json::Value\",\n performance,\n self_described as \"self_described: serde_json::Value\",\n bond_info as \"bond_info: serde_json::Value\",\n http_api_port\n FROM\n nym_nodes\n ORDER BY\n node_id\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -62,6 +62,11 @@
|
||||
"ordinal": 11,
|
||||
"name": "bond_info: serde_json::Value",
|
||||
"type_info": "Jsonb"
|
||||
},
|
||||
{
|
||||
"ordinal": 12,
|
||||
"name": "http_api_port",
|
||||
"type_info": "Int4"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
@@ -79,8 +84,9 @@
|
||||
true,
|
||||
false,
|
||||
true,
|
||||
true,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "283f49a65c7d70bf271702ff6a5c7ad6e68c81932d295ff18ed198c54706a57c"
|
||||
"hash": "3ddc12cc4e1796b787a50c40560d2bd71d1cfe5f5265e6f161b3122d1317a421"
|
||||
}
|
||||
-26
@@ -1,26 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO nym_nodes\n (node_id, ed25519_identity_pubkey,\n total_stake,\n ip_addresses, mix_port,\n x25519_sphinx_pubkey, node_role,\n supported_roles, entry,\n self_described,\n bond_info,\n performance, last_updated_utc\n )\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)\n ON CONFLICT(node_id) DO UPDATE SET\n ed25519_identity_pubkey=excluded.ed25519_identity_pubkey,\n ip_addresses=excluded.ip_addresses,\n mix_port=excluded.mix_port,\n x25519_sphinx_pubkey=excluded.x25519_sphinx_pubkey,\n node_role=excluded.node_role,\n supported_roles=excluded.supported_roles,\n entry=excluded.entry,\n self_described=excluded.self_described,\n bond_info=excluded.bond_info,\n performance=excluded.performance,\n last_updated_utc=excluded.last_updated_utc\n ;",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Int4",
|
||||
"Varchar",
|
||||
"Int8",
|
||||
"Jsonb",
|
||||
"Int4",
|
||||
"Varchar",
|
||||
"Jsonb",
|
||||
"Jsonb",
|
||||
"Jsonb",
|
||||
"Jsonb",
|
||||
"Jsonb",
|
||||
"Varchar",
|
||||
"Int4"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "b010fb91828f7e4f0b72bdfe3b58b2abb437cccdb6ebd2e1087cc822ed737b0e"
|
||||
}
|
||||
+27
@@ -0,0 +1,27 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO nym_nodes\n (node_id, ed25519_identity_pubkey,\n total_stake,\n ip_addresses, mix_port,\n x25519_sphinx_pubkey, node_role,\n supported_roles, entry,\n self_described,\n bond_info,\n performance, last_updated_utc, http_api_port\n )\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)\n ON CONFLICT(node_id) DO UPDATE SET\n ed25519_identity_pubkey=excluded.ed25519_identity_pubkey,\n ip_addresses=excluded.ip_addresses,\n mix_port=excluded.mix_port,\n x25519_sphinx_pubkey=excluded.x25519_sphinx_pubkey,\n node_role=excluded.node_role,\n supported_roles=excluded.supported_roles,\n entry=excluded.entry,\n self_described=excluded.self_described,\n bond_info=excluded.bond_info,\n performance=excluded.performance,\n last_updated_utc=excluded.last_updated_utc,\n http_api_port=excluded.http_api_port\n ;",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Int4",
|
||||
"Varchar",
|
||||
"Int8",
|
||||
"Jsonb",
|
||||
"Int4",
|
||||
"Varchar",
|
||||
"Jsonb",
|
||||
"Jsonb",
|
||||
"Jsonb",
|
||||
"Jsonb",
|
||||
"Jsonb",
|
||||
"Varchar",
|
||||
"Int4",
|
||||
"Int4"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "dde9aff827c34086077927bbe33fa3d5c939e7122ba7c88b78a353f00b271ec2"
|
||||
}
|
||||
@@ -3,7 +3,7 @@
|
||||
|
||||
[package]
|
||||
name = "nym-node-status-api"
|
||||
version = "4.0.10"
|
||||
version = "4.0.11-rc1"
|
||||
authors.workspace = true
|
||||
repository.workspace = true
|
||||
homepage.workspace = true
|
||||
|
||||
+2
@@ -0,0 +1,2 @@
|
||||
ALTER TABLE nym_nodes
|
||||
ADD COLUMN IF NOT EXISTS http_api_port INTEGER;
|
||||
@@ -1,5 +1,5 @@
|
||||
use crate::ticketbook_manager::TicketbookManagerConfig;
|
||||
use clap::Parser;
|
||||
use clap::{Parser, Subcommand};
|
||||
use nym_bin_common::bin_info;
|
||||
use nym_credential_proxy_lib::shared_state::ecash_state::TicketType;
|
||||
use reqwest::Url;
|
||||
@@ -105,6 +105,19 @@ pub(crate) struct Cli {
|
||||
|
||||
#[clap(flatten)]
|
||||
pub(crate) ticketbook: TicketbookArgs,
|
||||
|
||||
#[command(subcommand)]
|
||||
pub(crate) command: Option<Commands>,
|
||||
}
|
||||
|
||||
#[derive(Subcommand, Debug)]
|
||||
pub(crate) enum Commands {
|
||||
/// Scrape a single node and output detailed debug logs
|
||||
ScrapeNode {
|
||||
/// The id of the node to scrape
|
||||
#[arg(long)]
|
||||
node_id: i64,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Debug, Parser)]
|
||||
|
||||
@@ -381,7 +381,7 @@ impl ScrapeNodeKind {
|
||||
pub(crate) struct ScraperNodeInfo {
|
||||
pub node_kind: ScrapeNodeKind,
|
||||
pub hosts: Vec<String>,
|
||||
pub http_api_port: i64,
|
||||
pub http_api_port: Option<u16>,
|
||||
}
|
||||
|
||||
impl ScraperNodeInfo {
|
||||
@@ -395,8 +395,21 @@ impl ScraperNodeInfo {
|
||||
format!("http://{}", host),
|
||||
]);
|
||||
|
||||
if self.http_api_port != DEFAULT_NYM_NODE_HTTP_PORT as i64 {
|
||||
urls.insert(0, format!("http://{}:{}", host, self.http_api_port));
|
||||
if let Some(custom_http_api_port) = self.http_api_port {
|
||||
urls = Vec::new();
|
||||
for host in &self.hosts {
|
||||
urls.append(&mut vec![format!(
|
||||
"http://{}:{}",
|
||||
host, custom_http_api_port
|
||||
)]);
|
||||
}
|
||||
|
||||
// do not fall back to default ports, if the operator sets a custom http api port
|
||||
// in their bond, use it and error out if it's not available
|
||||
// this will correctly handle cases where some operators run multiple nodes
|
||||
// on a single IP address and assign different custom http port apis at bond time
|
||||
|
||||
// urls.insert(0, format!("http://{}:{}", host, custom_http_api_port));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -423,6 +436,7 @@ pub(crate) struct NymNodeDto {
|
||||
pub performance: String,
|
||||
pub self_described: Option<serde_json::Value>,
|
||||
pub bond_info: Option<serde_json::Value>,
|
||||
pub http_api_port: Option<i32>,
|
||||
}
|
||||
|
||||
#[allow(dead_code)] // it's not dead code but clippy doesn't detect usage in sqlx macros
|
||||
@@ -440,6 +454,7 @@ pub(crate) struct NymNodeInsertRecord {
|
||||
pub entry: Option<serde_json::Value>,
|
||||
pub self_described: Option<serde_json::Value>,
|
||||
pub bond_info: Option<serde_json::Value>,
|
||||
pub http_api_port: Option<i32>,
|
||||
pub last_updated_utc: i64,
|
||||
}
|
||||
|
||||
@@ -456,6 +471,12 @@ impl NymNodeInsertRecord {
|
||||
.map(|info| decimal_to_i64(info.total_stake()))
|
||||
.unwrap_or(0);
|
||||
let entry = serialize_opt_to_value!(skimmed_node.entry)?;
|
||||
let http_api_port = bond_info.and_then(|bond| {
|
||||
bond.bond_information
|
||||
.node
|
||||
.custom_http_port
|
||||
.map(|port| port as i32)
|
||||
});
|
||||
let bond_info = serialize_opt_to_value!(bond_info)?;
|
||||
let self_described = serialize_opt_to_value!(self_described)?;
|
||||
|
||||
@@ -472,6 +493,7 @@ impl NymNodeInsertRecord {
|
||||
entry,
|
||||
self_described,
|
||||
bond_info,
|
||||
http_api_port,
|
||||
last_updated_utc: now,
|
||||
};
|
||||
|
||||
|
||||
@@ -35,7 +35,8 @@ pub(crate) async fn get_all_nym_nodes(pool: &DbPool) -> anyhow::Result<Vec<NymNo
|
||||
entry as "entry: serde_json::Value",
|
||||
performance,
|
||||
self_described as "self_described: serde_json::Value",
|
||||
bond_info as "bond_info: serde_json::Value"
|
||||
bond_info as "bond_info: serde_json::Value",
|
||||
http_api_port
|
||||
FROM
|
||||
nym_nodes
|
||||
ORDER BY
|
||||
@@ -72,7 +73,8 @@ pub(crate) async fn get_described_bonded_nym_nodes(
|
||||
entry as "entry: serde_json::Value",
|
||||
performance,
|
||||
self_described as "self_described: serde_json::Value",
|
||||
bond_info as "bond_info: serde_json::Value"
|
||||
bond_info as "bond_info: serde_json::Value",
|
||||
http_api_port
|
||||
FROM
|
||||
nym_nodes
|
||||
WHERE
|
||||
@@ -115,9 +117,9 @@ pub(crate) async fn update_nym_nodes(
|
||||
supported_roles, entry,
|
||||
self_described,
|
||||
bond_info,
|
||||
performance, last_updated_utc
|
||||
performance, last_updated_utc, http_api_port
|
||||
)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
|
||||
ON CONFLICT(node_id) DO UPDATE SET
|
||||
ed25519_identity_pubkey=excluded.ed25519_identity_pubkey,
|
||||
ip_addresses=excluded.ip_addresses,
|
||||
@@ -129,7 +131,8 @@ pub(crate) async fn update_nym_nodes(
|
||||
self_described=excluded.self_described,
|
||||
bond_info=excluded.bond_info,
|
||||
performance=excluded.performance,
|
||||
last_updated_utc=excluded.last_updated_utc
|
||||
last_updated_utc=excluded.last_updated_utc,
|
||||
http_api_port=excluded.http_api_port
|
||||
;",
|
||||
record.node_id,
|
||||
record.ed25519_identity_pubkey,
|
||||
@@ -144,6 +147,7 @@ pub(crate) async fn update_nym_nodes(
|
||||
record.bond_info,
|
||||
record.performance,
|
||||
record.last_updated_utc as i32,
|
||||
record.http_api_port,
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await
|
||||
|
||||
@@ -21,10 +21,11 @@ pub(crate) async fn get_nodes_for_scraping(pool: &DbPool) -> Result<Vec<ScraperN
|
||||
let skimmed_nodes = queries::get_described_bonded_nym_nodes(pool)
|
||||
.await
|
||||
.map(|nodes_dto| {
|
||||
nodes_dto.into_iter().filter_map(|node| {
|
||||
let node_id = node.node_id;
|
||||
match SkimmedNode::try_from(node) {
|
||||
Ok(node) => Some(node),
|
||||
nodes_dto.into_iter().filter_map(|node_dto| {
|
||||
let node_id = node_dto.node_id;
|
||||
let http_api_port = node_dto.http_api_port;
|
||||
match SkimmedNode::try_from(node_dto) {
|
||||
Ok(node) => Some((node, http_api_port)),
|
||||
Err(e) => {
|
||||
tracing::error!("Failed to decode node_id={}: {}", node_id, e);
|
||||
None
|
||||
@@ -33,7 +34,7 @@ pub(crate) async fn get_nodes_for_scraping(pool: &DbPool) -> Result<Vec<ScraperN
|
||||
})
|
||||
})?;
|
||||
|
||||
skimmed_nodes.for_each(|node| {
|
||||
skimmed_nodes.for_each(|(node, http_api_port)| {
|
||||
// TODO: relies on polyfilling: Nym nodes table might contain legacy mixnodes
|
||||
// as well. Categorize them here.
|
||||
let node_kind = if gateway_keys.contains(&node.ed25519_identity_pubkey.to_base58_string()) {
|
||||
@@ -54,7 +55,7 @@ pub(crate) async fn get_nodes_for_scraping(pool: &DbPool) -> Result<Vec<ScraperN
|
||||
.into_iter()
|
||||
.map(|ip| ip.to_string())
|
||||
.collect::<Vec<_>>(),
|
||||
http_api_port: node.mix_port.into(),
|
||||
http_api_port: http_api_port.map(|port| port as u16),
|
||||
})
|
||||
});
|
||||
|
||||
|
||||
@@ -138,6 +138,7 @@ mod db_tests {
|
||||
performance: "1.0".to_string(),
|
||||
self_described: None,
|
||||
bond_info: None,
|
||||
http_api_port: None,
|
||||
};
|
||||
|
||||
let skimmed_node: nym_validator_client::nym_api::SkimmedNode =
|
||||
@@ -362,22 +363,42 @@ fn test_scraper_node_info_contact_addresses() {
|
||||
let node_info = ScraperNodeInfo {
|
||||
node_kind: ScrapeNodeKind::MixingNymNode { node_id: 123 },
|
||||
hosts: vec!["1.1.1.1".to_string(), "example.com".to_string()],
|
||||
http_api_port: 8080,
|
||||
http_api_port: None,
|
||||
};
|
||||
|
||||
let addresses = node_info.contact_addresses();
|
||||
|
||||
// Should generate multiple URLs for each host
|
||||
// Custom port (8080) should be inserted at the beginning
|
||||
// When no custom port is specified only default ports should be used
|
||||
assert!(addresses.contains(&"http://1.1.1.1:8080".to_string()));
|
||||
assert!(addresses.contains(&"http://example.com:8080".to_string()));
|
||||
assert!(addresses.contains(&"http://1.1.1.1:8000".to_string()));
|
||||
assert!(addresses.contains(&"https://1.1.1.1".to_string()));
|
||||
assert!(addresses.contains(&"http://1.1.1.1".to_string()));
|
||||
assert!(addresses.contains(&"http://example.com:8000".to_string()));
|
||||
// Check that URLs follow the expected pattern
|
||||
assert!(addresses.len() >= 8); // At least 4 URLs per host
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_scraper_node_info_contact_addresses_with_custom_http_api_port() {
|
||||
use crate::db::models::{ScrapeNodeKind, ScraperNodeInfo};
|
||||
|
||||
let node_info = ScraperNodeInfo {
|
||||
node_kind: ScrapeNodeKind::MixingNymNode { node_id: 123 },
|
||||
hosts: vec!["1.1.1.1".to_string(), "example.com".to_string()],
|
||||
http_api_port: Some(4444),
|
||||
};
|
||||
|
||||
let addresses = node_info.contact_addresses();
|
||||
|
||||
// Should generate multiple URLs for each host
|
||||
// Custom port (4444) should be the only port in the list
|
||||
assert!(addresses.contains(&"http://1.1.1.1:4444".to_string()));
|
||||
assert!(addresses.contains(&"http://example.com:4444".to_string()));
|
||||
// Check that URLs follow the expected pattern
|
||||
assert!(addresses.len() >= 2); // At least 4 URLs per host
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_scrape_node_kind_node_id() {
|
||||
use crate::db::models::ScrapeNodeKind;
|
||||
@@ -414,6 +435,7 @@ fn test_nym_node_dto_with_invalid_keys() {
|
||||
performance: "1.0".to_string(),
|
||||
self_described: None,
|
||||
bond_info: None,
|
||||
http_api_port: None,
|
||||
};
|
||||
|
||||
let result: Result<nym_validator_client::nym_api::SkimmedNode, _> = nym_node_dto.try_into();
|
||||
@@ -451,6 +473,7 @@ fn test_nym_node_dto_with_invalid_performance() {
|
||||
performance: "invalid_percent".to_string(),
|
||||
self_described: None,
|
||||
bond_info: None,
|
||||
http_api_port: None,
|
||||
};
|
||||
|
||||
let result: Result<nym_validator_client::nym_api::SkimmedNode, _> = nym_node_dto.try_into();
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
use crate::cli::Commands;
|
||||
use crate::monitor::DelegationsCache;
|
||||
use crate::node_scraper::helpers::scrape_and_store_description_by_node_id;
|
||||
use crate::ticketbook_manager::TicketbookManager;
|
||||
use crate::ticketbook_manager::state::TicketbookManagerState;
|
||||
use clap::Parser;
|
||||
@@ -40,11 +42,49 @@ async fn main() -> anyhow::Result<()> {
|
||||
tracing::info!("Registered {} agent keys", agent_key_list.len());
|
||||
|
||||
let connection_url = args.database_url.clone();
|
||||
tracing::debug!("Using config:\n{:#?}", args);
|
||||
if std::env::var("SHOW_CONFIG").ok().is_some() {
|
||||
tracing::debug!("Using config:\n{:#?}", args);
|
||||
}
|
||||
|
||||
let storage = db::Storage::init(connection_url, args.sqlx_busy_timeout_s).await?;
|
||||
let db_pool = storage.pool_owned();
|
||||
|
||||
// node geocache is shared between node monitor and HTTP server
|
||||
let geocache = moka::future::Cache::builder()
|
||||
.time_to_live(args.geodata_ttl)
|
||||
.build();
|
||||
let delegations_cache = DelegationsCache::new();
|
||||
|
||||
let client_config = nym_validator_client::nyxd::Config::try_from_nym_network_details(
|
||||
&nym_network_defaults::NymNetworkDetails::new_from_env(),
|
||||
)?;
|
||||
let nyxd_client = NyxdClient::connect(client_config.clone(), args.nyxd_addr.as_str())
|
||||
.map_err(|err| anyhow::anyhow!("Couldn't connect: {}", err))?;
|
||||
|
||||
match args.command {
|
||||
Some(Commands::ScrapeNode { node_id }) => {
|
||||
if std::env::var("RUN_ONCE_INIT_NODES").ok().is_some() {
|
||||
let geocache_clone = geocache.clone();
|
||||
let delegations_cache_clone = Arc::clone(&delegations_cache);
|
||||
monitor::run_once(
|
||||
db_pool.clone(),
|
||||
args.nym_api_client_timeout,
|
||||
nyxd_client,
|
||||
args.ipinfo_api_token,
|
||||
geocache_clone,
|
||||
delegations_cache_clone,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
tracing::info!("Scraping node with id {node_id}...");
|
||||
scrape_and_store_description_by_node_id(&db_pool, node_id).await?;
|
||||
return Ok(());
|
||||
}
|
||||
None => {
|
||||
// default behaviour
|
||||
}
|
||||
}
|
||||
|
||||
// Start the node scraper
|
||||
let scraper = node_scraper::DescriptionScraper::new(storage.pool_owned());
|
||||
shutdown_manager.spawn_with_shutdown(async move {
|
||||
@@ -58,20 +98,9 @@ async fn main() -> anyhow::Result<()> {
|
||||
scraper.start().await;
|
||||
});
|
||||
|
||||
// node geocache is shared between node monitor and HTTP server
|
||||
let geocache = moka::future::Cache::builder()
|
||||
.time_to_live(args.geodata_ttl)
|
||||
.build();
|
||||
let delegations_cache = DelegationsCache::new();
|
||||
|
||||
// Start the monitor
|
||||
let geocache_clone = geocache.clone();
|
||||
let delegations_cache_clone = Arc::clone(&delegations_cache);
|
||||
let client_config = nym_validator_client::nyxd::Config::try_from_nym_network_details(
|
||||
&nym_network_defaults::NymNetworkDetails::new_from_env(),
|
||||
)?;
|
||||
let nyxd_client = NyxdClient::connect(client_config.clone(), args.nyxd_addr.as_str())
|
||||
.map_err(|err| anyhow::anyhow!("Couldn't connect: {}", err))?;
|
||||
|
||||
shutdown_manager.spawn_with_shutdown(async move {
|
||||
monitor::run_in_background(
|
||||
|
||||
@@ -68,7 +68,7 @@ pub(crate) async fn run_in_background(
|
||||
loop {
|
||||
tracing::info!("Refreshing node info...");
|
||||
|
||||
if let Err(e) = monitor.run().await {
|
||||
if let Err(e) = monitor.run(false).await {
|
||||
tracing::error!(
|
||||
"Monitor run failed: {e}, retrying in {}s...",
|
||||
MONITOR_FAILURE_RETRY_DELAY.as_secs()
|
||||
@@ -84,8 +84,33 @@ pub(crate) async fn run_in_background(
|
||||
}
|
||||
}
|
||||
|
||||
#[instrument(level = "debug", name = "data_monitor", skip_all)]
|
||||
pub(crate) async fn run_once(
|
||||
db_pool: DbPool,
|
||||
nym_api_client_timeout: Duration,
|
||||
nyxd_client: nym_validator_client::QueryHttpRpcNyxdClient,
|
||||
ipinfo_api_token: String,
|
||||
geocache: NodeGeoCache,
|
||||
node_delegations: Arc<RwLock<DelegationsCache>>,
|
||||
) -> anyhow::Result<()> {
|
||||
let ipinfo = IpInfoClient::new(ipinfo_api_token.clone());
|
||||
|
||||
let mut monitor = Monitor {
|
||||
db_pool,
|
||||
network_details: nym_network_defaults::NymNetworkDetails::new_from_env(),
|
||||
nym_api_client_timeout,
|
||||
nyxd_client,
|
||||
ipinfo,
|
||||
geocache,
|
||||
node_delegations,
|
||||
};
|
||||
|
||||
tracing::info!("Refreshing node info...");
|
||||
monitor.run(true).await
|
||||
}
|
||||
|
||||
impl Monitor {
|
||||
async fn run(&mut self) -> anyhow::Result<()> {
|
||||
async fn run(&mut self, exit_early: bool) -> anyhow::Result<()> {
|
||||
self.check_ipinfo_bandwidth().await;
|
||||
|
||||
let default_api_url = self
|
||||
@@ -153,6 +178,11 @@ impl Monitor {
|
||||
tracing::debug!("{} nym nodes written to DB!", inserted);
|
||||
})?;
|
||||
|
||||
// stop here if running once
|
||||
if exit_early {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// refresh geodata for all nodes
|
||||
for node_description in described_nodes.values() {
|
||||
self.location_cached(node_description).await;
|
||||
|
||||
@@ -118,6 +118,17 @@ pub fn sanitize_description(
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn scrape_and_store_description_by_node_id(pool: &DbPool, node_id: i64) -> Result<()> {
|
||||
let nodes = crate::db::queries::get_nodes_for_scraping(pool).await?;
|
||||
match nodes.iter().find(|n| *n.node_kind.node_id() == node_id) {
|
||||
Some(node) => Ok(scrape_and_store_description(pool, node.clone()).await?),
|
||||
None => {
|
||||
error!("Could not find node with id {node_id}");
|
||||
Err(anyhow!("Could not find node with id {node_id}"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn scrape_and_store_description(pool: &DbPool, node: ScraperNodeInfo) -> Result<()> {
|
||||
let client = build_client()?;
|
||||
let urls = node.contact_addresses();
|
||||
@@ -152,7 +163,13 @@ pub async fn scrape_and_store_description(pool: &DbPool, node: ScraperNodeInfo)
|
||||
anyhow::anyhow!("Failed to fetch description from any URL: {}", err_msg)
|
||||
})?;
|
||||
|
||||
let sanitized_description = sanitize_description(description, *node.node_id());
|
||||
let sanitized_description = sanitize_description(description.clone(), *node.node_id());
|
||||
|
||||
trace!("tried_url_list = {tried_url_list:?}");
|
||||
trace!("ndoe_id = {}", node.node_id());
|
||||
trace!("description = {:?}", description);
|
||||
trace!("sanitized_description = {:?}", sanitized_description);
|
||||
|
||||
insert_scraped_node_description(pool, &node.node_kind, &sanitized_description).await?;
|
||||
|
||||
Ok(())
|
||||
|
||||
Reference in New Issue
Block a user