NS API: add mixnet scraper (#5200)
* ns-api: add mixnode scraper * clippy * rebase
This commit is contained in:
Sachin Kamath
Generated
+153
-1
@@ -134,6 +134,19 @@ version = "0.2.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5c6cb57a04249c6480766f7f7cef5467412af1490f8d1e243141daddada3264f"
|
||||
|
||||
[[package]]
|
||||
name = "ammonia"
|
||||
version = "4.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1ab99eae5ee58501ab236beb6f20f6ca39be615267b014899c89b2f0bc18a459"
|
||||
dependencies = [
|
||||
"html5ever",
|
||||
"maplit",
|
||||
"once_cell",
|
||||
"tendril",
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "android-tzdata"
|
||||
version = "0.1.1"
|
||||
@@ -2595,6 +2608,16 @@ version = "2.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
|
||||
|
||||
[[package]]
|
||||
name = "futf"
|
||||
version = "0.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df420e2e84819663797d1ec6544b13c5be84629e7bb00dc960d6917db2987843"
|
||||
dependencies = [
|
||||
"mac",
|
||||
"new_debug_unreachable",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures"
|
||||
version = "0.3.31"
|
||||
@@ -3095,6 +3118,20 @@ dependencies = [
|
||||
"windows-sys 0.52.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "html5ever"
|
||||
version = "0.27.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c13771afe0e6e846f1e67d038d4cb29998a6779f93c809212e4e9c32efd244d4"
|
||||
dependencies = [
|
||||
"log",
|
||||
"mac",
|
||||
"markup5ever",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.90",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "http"
|
||||
version = "0.2.12"
|
||||
@@ -3985,6 +4022,12 @@ dependencies = [
|
||||
"tracing-subscriber",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mac"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
|
||||
|
||||
[[package]]
|
||||
name = "macroific"
|
||||
version = "1.3.1"
|
||||
@@ -4032,6 +4075,26 @@ dependencies = [
|
||||
"syn 2.0.90",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "maplit"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d"
|
||||
|
||||
[[package]]
|
||||
name = "markup5ever"
|
||||
version = "0.12.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "16ce3abbeba692c8b8441d036ef91aea6df8da2c6b6e21c7e14d3c18e526be45"
|
||||
dependencies = [
|
||||
"log",
|
||||
"phf",
|
||||
"phf_codegen",
|
||||
"string_cache",
|
||||
"string_cache_codegen",
|
||||
"tendril",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "matchers"
|
||||
version = "0.1.0"
|
||||
@@ -4315,6 +4378,12 @@ dependencies = [
|
||||
"log",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "new_debug_unreachable"
|
||||
version = "1.0.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086"
|
||||
|
||||
[[package]]
|
||||
name = "newline-converter"
|
||||
version = "0.2.2"
|
||||
@@ -6070,6 +6139,7 @@ dependencies = [
|
||||
name = "nym-node-status-api"
|
||||
version = "1.0.0-rc.7"
|
||||
dependencies = [
|
||||
"ammonia",
|
||||
"anyhow",
|
||||
"axum 0.7.7",
|
||||
"chrono",
|
||||
@@ -7379,7 +7449,46 @@ version = "0.11.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ade2d8b8f33c7333b51bcf0428d37e217e9f32192ae4772156f65063b8ce03dc"
|
||||
dependencies = [
|
||||
"phf_shared",
|
||||
"phf_shared 0.11.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "phf_codegen"
|
||||
version = "0.11.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "aef8048c789fa5e851558d709946d6d79a8ff88c0440c587967f8e94bfb1216a"
|
||||
dependencies = [
|
||||
"phf_generator 0.11.3",
|
||||
"phf_shared 0.11.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "phf_generator"
|
||||
version = "0.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5d5285893bb5eb82e6aaf5d59ee909a06a16737a8970984dd7746ba9283498d6"
|
||||
dependencies = [
|
||||
"phf_shared 0.10.0",
|
||||
"rand",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "phf_generator"
|
||||
version = "0.11.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d"
|
||||
dependencies = [
|
||||
"phf_shared 0.11.2",
|
||||
"rand",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "phf_shared"
|
||||
version = "0.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b6796ad771acdc0123d2a88dc428b5e38ef24456743ddb1744ed628f9815c096"
|
||||
dependencies = [
|
||||
"siphasher 0.3.11",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -7570,6 +7679,12 @@ version = "0.2.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
|
||||
|
||||
[[package]]
|
||||
name = "precomputed-hash"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "925383efa346730478fb4838dbe9137d2a47675ad789c546d150a6e1dd4ab31c"
|
||||
|
||||
[[package]]
|
||||
name = "pretty_assertions"
|
||||
version = "1.4.1"
|
||||
@@ -9297,6 +9412,32 @@ version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f"
|
||||
|
||||
[[package]]
|
||||
name = "string_cache"
|
||||
version = "0.8.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f91138e76242f575eb1d3b38b4f1362f10d3a43f47d182a5b359af488a02293b"
|
||||
dependencies = [
|
||||
"new_debug_unreachable",
|
||||
"once_cell",
|
||||
"parking_lot",
|
||||
"phf_shared 0.10.0",
|
||||
"precomputed-hash",
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "string_cache_codegen"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6bb30289b722be4ff74a408c3cc27edeaad656e06cb1fe8fa9231fa59c728988"
|
||||
dependencies = [
|
||||
"phf_generator 0.10.0",
|
||||
"phf_shared 0.10.0",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "stringprep"
|
||||
version = "0.1.5"
|
||||
@@ -9653,6 +9794,17 @@ dependencies = [
|
||||
"walkdir",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tendril"
|
||||
version = "0.4.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d24a120c5fc464a3458240ee02c299ebcb9d67b5249c8848b09d639dca8d7bb0"
|
||||
dependencies = [
|
||||
"futf",
|
||||
"mac",
|
||||
"utf-8",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "termcolor"
|
||||
version = "1.4.1"
|
||||
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n INSERT INTO mixnode_description (\n mix_id, moniker, website, security_contact, details, last_updated_utc\n ) VALUES (?, ?, ?, ?, ?, ?)\n ON CONFLICT (mix_id) DO UPDATE SET\n moniker = excluded.moniker,\n website = excluded.website,\n security_contact = excluded.security_contact,\n details = excluded.details,\n last_updated_utc = excluded.last_updated_utc\n ",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Right": 6
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "06065394c157927e4002ddd5c7c1af626ae15728d615f539470cd7c189312385"
|
||||
}
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "UPDATE\n testruns\n SET\n status = ?\n WHERE\n status = ?\n AND\n timestamp_utc < ?\n ",
|
||||
"query": "UPDATE\n testruns\n SET\n status = ?\n WHERE\n status = ?\n AND\n last_assigned_utc < ?\n ",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
@@ -8,5 +8,5 @@
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "f5048d9926a5f5329f7f3b96d43b925e033ceec4f8112258feb4ac9e96fc5924"
|
||||
"hash": "0cf0e4d4f30e90caecffd6255ef85dab12730e538be194438f19ed7f198bd50e"
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "INSERT OR IGNORE INTO gateway_session_stats\n (gateway_identity_key, node_id, day,\n unique_active_clients, session_started, users_hashes,\n vpn_sessions, mixnet_sessions, unknown_sessions)\n VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Right": 9
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "3243cf5646255a9430d1e6710970505d0dbcc62703f40e090e80ff48c77723c4"
|
||||
}
|
||||
+1
-1
@@ -30,7 +30,7 @@
|
||||
"nullable": [
|
||||
true,
|
||||
false,
|
||||
true,
|
||||
false,
|
||||
true
|
||||
]
|
||||
},
|
||||
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "INSERT INTO testruns (gateway_id, status, ip_address, timestamp_utc, log) VALUES (?, ?, ?, ?, ?)",
|
||||
"query": "INSERT INTO testruns (gateway_id, status, ip_address, created_utc, log) VALUES (?, ?, ?, ?, ?)",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
@@ -8,5 +8,5 @@
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "3c584e211d07c511644c8079187965acf3bcfb3f84ba8d24ed645d79976cf784"
|
||||
"hash": "4d865e873c9cb133883da94db72dcdebd4969e1f240def9fb1bf946f4a1f342f"
|
||||
}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "UPDATE testruns\n SET\n status = ?,\n last_assigned_utc = ?\n WHERE rowid =\n (\n SELECT rowid\n FROM testruns\n WHERE status = ?\n ORDER BY created_utc asc\n LIMIT 1\n )\n RETURNING\n id as \"id!\",\n gateway_id\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "id!",
|
||||
"ordinal": 0,
|
||||
"type_info": "Int64"
|
||||
},
|
||||
{
|
||||
"name": "gateway_id",
|
||||
"ordinal": 1,
|
||||
"type_info": "Int64"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 3
|
||||
},
|
||||
"nullable": [
|
||||
true,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "5e9cbb39f5fb53774803270f422989e199aac4d4a71913c7074359b4bd676b02"
|
||||
}
|
||||
+1
-1
@@ -82,7 +82,7 @@
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
true,
|
||||
false,
|
||||
true,
|
||||
true,
|
||||
true,
|
||||
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT packets_received, packets_sent, packets_dropped\n FROM mixnode_packet_stats_raw\n WHERE mix_id = ?\n ORDER BY timestamp_utc DESC\n LIMIT 1 OFFSET 1\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "packets_received",
|
||||
"ordinal": 0,
|
||||
"type_info": "Int64"
|
||||
},
|
||||
{
|
||||
"name": "packets_sent",
|
||||
"ordinal": 1,
|
||||
"type_info": "Int64"
|
||||
},
|
||||
{
|
||||
"name": "packets_dropped",
|
||||
"ordinal": 2,
|
||||
"type_info": "Int64"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 1
|
||||
},
|
||||
"nullable": [
|
||||
true,
|
||||
true,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "7b3e44e1dea0568d331017216e5375684086bbeb80f3b251adc761bae0dafe92"
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n INSERT INTO mixnode_packet_stats_raw (\n mix_id, timestamp_utc, packets_received, packets_sent, packets_dropped\n ) VALUES (?, ?, ?, ?, ?)\n ",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Right": 5
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "84e784b7ceb479f37fa312dd9a8bb0b3debc04cd251cdfaa438bce7aca7c5992"
|
||||
}
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT mix_id as node_id, host, http_api_port \n FROM mixnodes \n WHERE bonded = true\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "node_id",
|
||||
"ordinal": 0,
|
||||
"type_info": "Int64"
|
||||
},
|
||||
{
|
||||
"name": "host",
|
||||
"ordinal": 1,
|
||||
"type_info": "Text"
|
||||
},
|
||||
{
|
||||
"name": "http_api_port",
|
||||
"ordinal": 2,
|
||||
"type_info": "Int64"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 0
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "8c6c1c67df2d06da74825872f448d30505801b56506668499f9da718f1d6b8f8"
|
||||
}
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n SELECT\n total_stake\n FROM mixnodes\n WHERE mix_id = ?\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "total_stake",
|
||||
"ordinal": 0,
|
||||
"type_info": "Int64"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 1
|
||||
},
|
||||
"nullable": [
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "969acbff4b128d3a3dc998bb5db1a4c8b4c106767e0276435dc23fcfd6acfe4f"
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "\n INSERT INTO mixnode_daily_stats (\n mix_id, date_utc, total_stake, packets_received, packets_sent, packets_dropped\n ) VALUES (?, ?, ?, ?, ?, ?)\n ON CONFLICT(mix_id, date_utc) DO UPDATE SET\n total_stake = excluded.total_stake,\n packets_received = mixnode_daily_stats.packets_received + excluded.packets_received,\n packets_sent = mixnode_daily_stats.packets_sent + excluded.packets_sent,\n packets_dropped = mixnode_daily_stats.packets_dropped + excluded.packets_dropped\n ",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Right": 6
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "a14d5767512bc88ac47b667c47a773b548ccb20bc539839766080d8455439339"
|
||||
}
|
||||
+10
-4
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "SELECT\n id as \"id!\",\n gateway_id as \"gateway_id!\",\n status as \"status!\",\n timestamp_utc as \"timestamp_utc!\",\n ip_address as \"ip_address!\",\n log as \"log!\"\n FROM testruns\n WHERE gateway_id = ? AND status != 2\n ORDER BY id DESC\n LIMIT 1",
|
||||
"query": "SELECT\n id as \"id!\",\n gateway_id as \"gateway_id!\",\n status as \"status!\",\n created_utc as \"created_utc!\",\n ip_address as \"ip_address!\",\n log as \"log!\",\n last_assigned_utc\n FROM testruns\n WHERE gateway_id = ? AND status != 2\n ORDER BY id DESC\n LIMIT 1",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -19,7 +19,7 @@
|
||||
"type_info": "Int64"
|
||||
},
|
||||
{
|
||||
"name": "timestamp_utc!",
|
||||
"name": "created_utc!",
|
||||
"ordinal": 3,
|
||||
"type_info": "Int64"
|
||||
},
|
||||
@@ -32,6 +32,11 @@
|
||||
"name": "log!",
|
||||
"ordinal": 5,
|
||||
"type_info": "Text"
|
||||
},
|
||||
{
|
||||
"name": "last_assigned_utc",
|
||||
"ordinal": 6,
|
||||
"type_info": "Int64"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
@@ -43,8 +48,9 @@
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
false
|
||||
false,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "46d76bc6d3fba2dae3b21511a36289dd776749dd7a20cda61b0480f2fba60889"
|
||||
"hash": "a1d9eb816acd1a91ed0975c801c9295c01a78861a2a0597ad28b1579a14bf008"
|
||||
}
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "SELECT\n COUNT(id) as \"count: i64\"\n FROM testruns\n WHERE\n status = ?\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "count: i64",
|
||||
"ordinal": 0,
|
||||
"type_info": "Int"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 1
|
||||
},
|
||||
"nullable": [
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "a79a61b87325f3f1d9c5a4fb386ccd585be0641e5878acb6283b879f22ed2b4c"
|
||||
}
|
||||
+12
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "DELETE FROM gateway_session_stats WHERE day <= ?",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Right": 1
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "c9e61180ec35dfab8623333fafa3b216b93440d0fddc0a37dd1b6c1813741f6a"
|
||||
}
|
||||
+10
-4
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "SELECT\n id as \"id!\",\n gateway_id as \"gateway_id!\",\n status as \"status!\",\n timestamp_utc as \"timestamp_utc!\",\n ip_address as \"ip_address!\",\n log as \"log!\"\n FROM testruns\n WHERE\n id = ?\n AND\n status = ?\n ORDER BY timestamp_utc",
|
||||
"query": "SELECT\n id as \"id!\",\n gateway_id as \"gateway_id!\",\n status as \"status!\",\n created_utc as \"created_utc!\",\n ip_address as \"ip_address!\",\n log as \"log!\",\n last_assigned_utc\n FROM testruns\n WHERE\n id = ?\n AND\n status = ?\n ORDER BY created_utc\n LIMIT 1",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -19,7 +19,7 @@
|
||||
"type_info": "Int64"
|
||||
},
|
||||
{
|
||||
"name": "timestamp_utc!",
|
||||
"name": "created_utc!",
|
||||
"ordinal": 3,
|
||||
"type_info": "Int64"
|
||||
},
|
||||
@@ -32,6 +32,11 @@
|
||||
"name": "log!",
|
||||
"ordinal": 5,
|
||||
"type_info": "Text"
|
||||
},
|
||||
{
|
||||
"name": "last_assigned_utc",
|
||||
"ordinal": 6,
|
||||
"type_info": "Int64"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
@@ -43,8 +48,9 @@
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
false
|
||||
false,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "6d7967b831b355d5f2c77950abc56f816956b0824c66a25da611dce688105d36"
|
||||
"hash": "f0c64794cbaed87a1d3166251d8e6720c9a9fc929144188460849be85d915004"
|
||||
}
|
||||
-26
@@ -1,26 +0,0 @@
|
||||
{
|
||||
"db_name": "SQLite",
|
||||
"query": "UPDATE testruns\n SET status = ?\n WHERE rowid =\n (\n SELECT rowid\n FROM testruns\n WHERE status = ?\n ORDER BY timestamp_utc asc\n LIMIT 1\n )\n RETURNING\n id as \"id!\",\n gateway_id\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"name": "id!",
|
||||
"ordinal": 0,
|
||||
"type_info": "Int64"
|
||||
},
|
||||
{
|
||||
"name": "gateway_id",
|
||||
"ordinal": 1,
|
||||
"type_info": "Int64"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Right": 2
|
||||
},
|
||||
"nullable": [
|
||||
true,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "ff9334ba7b670b218b2f9100e9ab5d2f2d08b2e53203aab9f07ea9b52acbd407"
|
||||
}
|
||||
@@ -13,6 +13,7 @@ license.workspace = true
|
||||
rust-version.workspace = true
|
||||
|
||||
[dependencies]
|
||||
ammonia = "4"
|
||||
anyhow = { workspace = true }
|
||||
axum = { workspace = true, features = ["tokio", "macros"] }
|
||||
chrono = { workspace = true }
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
CREATE TABLE mixnode_packet_stats_raw (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
mix_id INTEGER NOT NULL,
|
||||
timestamp_utc INTEGER NOT NULL,
|
||||
packets_received INTEGER,
|
||||
packets_sent INTEGER,
|
||||
packets_dropped INTEGER,
|
||||
FOREIGN KEY (mix_id) REFERENCES mixnodes (mix_id)
|
||||
);
|
||||
|
||||
CREATE INDEX idx_mixnode_packet_stats_raw_mix_id_timestamp_utc ON mixnode_packet_stats_raw (mix_id, timestamp_utc);
|
||||
@@ -376,3 +376,9 @@ impl TryFrom<GatewaySessionsRecord> for http::models::SessionStats {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct ScraperNodeInfo {
|
||||
pub node_id: i64,
|
||||
pub host: String,
|
||||
pub http_api_port: i64,
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ mod gateways;
|
||||
mod gateways_stats;
|
||||
mod misc;
|
||||
mod mixnodes;
|
||||
pub(crate) mod scraper;
|
||||
mod summary;
|
||||
pub(crate) mod testruns;
|
||||
|
||||
@@ -13,6 +14,7 @@ pub(crate) use misc::insert_summaries;
|
||||
pub(crate) use mixnodes::{
|
||||
ensure_mixnodes_still_bonded, get_all_mixnodes, get_daily_stats, insert_mixnodes,
|
||||
};
|
||||
pub(crate) use scraper::fetch_active_nodes;
|
||||
pub(crate) use summary::{get_summary, get_summary_history};
|
||||
|
||||
pub(crate) use gateways_stats::{delete_old_records, get_sessions_stats, insert_session_records};
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
use crate::db::{models::ScraperNodeInfo, DbPool};
|
||||
use anyhow::Result;
|
||||
|
||||
pub(crate) async fn fetch_active_nodes(pool: &DbPool) -> Result<Vec<ScraperNodeInfo>> {
|
||||
let mut conn = pool.acquire().await?;
|
||||
let nodes = sqlx::query_as!(
|
||||
ScraperNodeInfo,
|
||||
r#"
|
||||
SELECT mix_id as node_id, host, http_api_port
|
||||
FROM mixnodes
|
||||
WHERE bonded = true
|
||||
"#
|
||||
)
|
||||
.fetch_all(&mut *conn)
|
||||
.await?;
|
||||
|
||||
Ok(nodes)
|
||||
}
|
||||
|
||||
// TODO: add stuff for gateways
|
||||
@@ -6,6 +6,7 @@ mod cli;
|
||||
mod db;
|
||||
mod http;
|
||||
mod logging;
|
||||
mod mixnet_scraper;
|
||||
mod monitor;
|
||||
mod node_scraper;
|
||||
mod testruns;
|
||||
@@ -28,6 +29,15 @@ async fn main() -> anyhow::Result<()> {
|
||||
|
||||
let storage = db::Storage::init(connection_url).await?;
|
||||
let db_pool = storage.pool_owned();
|
||||
|
||||
// Start the node scraper
|
||||
let scraper = mixnet_scraper::Scraper::new(storage.pool_owned());
|
||||
tokio::spawn(async move {
|
||||
scraper.start().await;
|
||||
});
|
||||
tracing::info!("Started node scraper task");
|
||||
|
||||
// Start the monitor
|
||||
let args_clone = args.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
|
||||
@@ -0,0 +1,315 @@
|
||||
use crate::db::models::ScraperNodeInfo;
|
||||
use ammonia::Builder;
|
||||
use anyhow::Result;
|
||||
use chrono::{DateTime, Datelike, Utc};
|
||||
use nym_network_defaults::DEFAULT_NYM_NODE_HTTP_PORT;
|
||||
use reqwest;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sqlx::SqlitePool;
|
||||
use std::time::Duration;
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct NodeStats {
|
||||
pub packets_received: i64,
|
||||
pub packets_sent: i64,
|
||||
pub packets_dropped: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct NodeDescriptionResponse {
|
||||
pub moniker: Option<String>,
|
||||
pub website: Option<String>,
|
||||
pub security_contact: Option<String>,
|
||||
pub details: Option<String>,
|
||||
}
|
||||
|
||||
// Eventhough they are `/api/v1/*`, newer nodes have a redirect to the correct endpoints
|
||||
const DESCRIPTION_URL: &str = "/description";
|
||||
const PACKET_STATS_URL: &str = "/stats";
|
||||
|
||||
// We need this as some of the mixnodes respond with float values for the packet statistics (?????)
|
||||
pub fn get_packet_value(response: &serde_json::Value, key: &str) -> Option<i64> {
|
||||
response
|
||||
.get(key)
|
||||
.and_then(|value| value.as_i64().or_else(|| value.as_f64().map(|f| f as i64)))
|
||||
}
|
||||
|
||||
pub fn parse_mixnet_stats(response: serde_json::Value) -> Option<NodeStats> {
|
||||
// Try to parse the response return by old (deprecated) mixnodes
|
||||
if let Some(packets_dropped) =
|
||||
get_packet_value(&response, "packets_explicitly_dropped_since_startup")
|
||||
{
|
||||
return Some(NodeStats {
|
||||
packets_dropped,
|
||||
packets_sent: get_packet_value(&response, "packets_sent_since_startup")
|
||||
.unwrap_or_default(),
|
||||
|
||||
packets_received: get_packet_value(&response, "packets_received_since_startup")
|
||||
.unwrap_or_default(),
|
||||
});
|
||||
}
|
||||
|
||||
// Try to parse the response returned by nym-nodes
|
||||
if let Some(packets_dropped) = get_packet_value(&response, "dropped_since_startup") {
|
||||
return Some(NodeStats {
|
||||
packets_dropped,
|
||||
packets_sent: get_packet_value(&response, "sent_since_startup").unwrap_or_default(),
|
||||
|
||||
packets_received: get_packet_value(&response, "received_since_startup")
|
||||
.unwrap_or_default(),
|
||||
});
|
||||
}
|
||||
|
||||
// If neither format matches, return None
|
||||
None
|
||||
}
|
||||
|
||||
pub fn calculate_packet_difference(current: i64, previous: i64) -> i64 {
|
||||
if current >= previous {
|
||||
current - previous
|
||||
} else {
|
||||
current // Node likely restarted, use current value
|
||||
}
|
||||
}
|
||||
|
||||
pub fn build_client() -> Result<reqwest::Client> {
|
||||
reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(3))
|
||||
// Since we have a mix of TLS and non-TLS nodes, we need to accept invalid certs
|
||||
// when accessing IP:port
|
||||
.danger_accept_invalid_certs(true)
|
||||
.build()
|
||||
.map_err(|e| anyhow::anyhow!("Failed to build HTTP client: {}", e))
|
||||
}
|
||||
|
||||
pub fn sanitize_description(description: NodeDescriptionResponse) -> NodeDescriptionResponse {
|
||||
let mut sanitizer = Builder::new();
|
||||
sanitizer
|
||||
.tags(std::collections::HashSet::new())
|
||||
.generic_attributes(std::collections::HashSet::new())
|
||||
.url_schemes(std::collections::HashSet::new());
|
||||
|
||||
let sanitize_field = |opt: Option<String>| -> Option<String> {
|
||||
Some(
|
||||
opt.filter(|s| !s.trim().is_empty())
|
||||
.map_or_else(|| "N/A".to_string(), |s| sanitizer.clean(&s).to_string()),
|
||||
)
|
||||
};
|
||||
|
||||
NodeDescriptionResponse {
|
||||
moniker: sanitize_field(description.moniker),
|
||||
website: sanitize_field(description.website),
|
||||
security_contact: sanitize_field(description.security_contact),
|
||||
details: sanitize_field(description.details),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeInfo) -> Result<()> {
|
||||
let client = build_client()?;
|
||||
let mut urls = vec![
|
||||
format!("http://{}:{DEFAULT_NYM_NODE_HTTP_PORT}", node.host),
|
||||
format!("http://{}:8000", node.host),
|
||||
format!("https://{}", node.host),
|
||||
format!("http://{}", node.host),
|
||||
];
|
||||
|
||||
if node.http_api_port != DEFAULT_NYM_NODE_HTTP_PORT as i64 {
|
||||
urls.insert(0, format!("http://{}:{}", node.host, node.http_api_port));
|
||||
}
|
||||
|
||||
let mut description = None;
|
||||
let mut error = None;
|
||||
|
||||
for mut url in urls {
|
||||
url = format!("{}{}", url.trim_end_matches('/'), DESCRIPTION_URL);
|
||||
|
||||
match client.get(&url).send().await {
|
||||
Ok(response) => {
|
||||
if let Ok(desc) = response.json::<NodeDescriptionResponse>().await {
|
||||
description = Some(desc);
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(e) => error = Some(e),
|
||||
}
|
||||
}
|
||||
|
||||
let description = description.ok_or_else(|| {
|
||||
let err_msg = error.map_or_else(|| "Unknown error".to_string(), |e| e.to_string());
|
||||
anyhow::anyhow!("Failed to fetch description from any URL: {}", err_msg)
|
||||
})?;
|
||||
|
||||
let sanitized_description = sanitize_description(description);
|
||||
|
||||
let mut conn = pool.acquire().await?;
|
||||
let timestamp = Utc::now().timestamp();
|
||||
|
||||
sqlx::query!(
|
||||
r#"
|
||||
INSERT INTO mixnode_description (
|
||||
mix_id, moniker, website, security_contact, details, last_updated_utc
|
||||
) VALUES (?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT (mix_id) DO UPDATE SET
|
||||
moniker = excluded.moniker,
|
||||
website = excluded.website,
|
||||
security_contact = excluded.security_contact,
|
||||
details = excluded.details,
|
||||
last_updated_utc = excluded.last_updated_utc
|
||||
"#,
|
||||
node.node_id,
|
||||
sanitized_description.moniker,
|
||||
sanitized_description.website,
|
||||
sanitized_description.security_contact,
|
||||
sanitized_description.details,
|
||||
timestamp,
|
||||
)
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn scrape_and_store_packet_stats(
|
||||
pool: &SqlitePool,
|
||||
node: &ScraperNodeInfo,
|
||||
) -> Result<()> {
|
||||
let client = build_client()?;
|
||||
let mut urls = vec![
|
||||
format!("http://{}:{DEFAULT_NYM_NODE_HTTP_PORT}", node.host),
|
||||
format!("http://{}:8000", node.host),
|
||||
format!("https://{}", node.host),
|
||||
format!("http://{}", node.host),
|
||||
];
|
||||
|
||||
if node.http_api_port != DEFAULT_NYM_NODE_HTTP_PORT as i64 {
|
||||
urls.insert(0, format!("http://{}:{}", node.host, node.http_api_port));
|
||||
}
|
||||
|
||||
let mut stats = None;
|
||||
let mut error = None;
|
||||
|
||||
for mut url in urls {
|
||||
url = format!("{}{}", url.trim_end_matches('/'), PACKET_STATS_URL);
|
||||
|
||||
match client.get(&url).send().await {
|
||||
Ok(response) => {
|
||||
if let Some(node_stats) = parse_mixnet_stats(response.json().await?) {
|
||||
stats = Some(node_stats);
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(e) => error = Some(e),
|
||||
}
|
||||
}
|
||||
|
||||
let stats = stats.ok_or_else(|| {
|
||||
let err_msg = error.map_or_else(|| "Unknown error".to_string(), |e| e.to_string());
|
||||
anyhow::anyhow!("Failed to fetch stats from any URL: {}", err_msg)
|
||||
})?;
|
||||
|
||||
let timestamp = Utc::now();
|
||||
let timestamp_utc = timestamp.timestamp();
|
||||
let mut conn = pool.acquire().await?;
|
||||
|
||||
// Store raw stats
|
||||
sqlx::query!(
|
||||
r#"
|
||||
INSERT INTO mixnode_packet_stats_raw (
|
||||
mix_id, timestamp_utc, packets_received, packets_sent, packets_dropped
|
||||
) VALUES (?, ?, ?, ?, ?)
|
||||
"#,
|
||||
node.node_id,
|
||||
timestamp_utc,
|
||||
stats.packets_received,
|
||||
stats.packets_sent,
|
||||
stats.packets_dropped,
|
||||
)
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
|
||||
// Update daily stats
|
||||
update_daily_stats(pool, node.node_id, timestamp, &stats).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn update_daily_stats(
|
||||
pool: &SqlitePool,
|
||||
node_id: i64,
|
||||
timestamp: DateTime<Utc>,
|
||||
current_stats: &NodeStats,
|
||||
) -> Result<()> {
|
||||
let mut conn = pool.acquire().await?;
|
||||
|
||||
let date_utc = format!(
|
||||
"{:04}-{:02}-{:02}",
|
||||
timestamp.year(),
|
||||
timestamp.month(),
|
||||
timestamp.day()
|
||||
);
|
||||
|
||||
let total_stake = sqlx::query_scalar!(
|
||||
r#"
|
||||
SELECT
|
||||
total_stake
|
||||
FROM mixnodes
|
||||
WHERE mix_id = ?
|
||||
"#,
|
||||
node_id
|
||||
)
|
||||
.fetch_one(&mut *conn)
|
||||
.await?;
|
||||
|
||||
// Get previous stats
|
||||
let previous_stats = sqlx::query!(
|
||||
r#"
|
||||
SELECT packets_received, packets_sent, packets_dropped
|
||||
FROM mixnode_packet_stats_raw
|
||||
WHERE mix_id = ?
|
||||
ORDER BY timestamp_utc DESC
|
||||
LIMIT 1 OFFSET 1
|
||||
"#,
|
||||
node_id
|
||||
)
|
||||
.fetch_optional(&mut *conn)
|
||||
.await?;
|
||||
|
||||
let (diff_received, diff_sent, diff_dropped) = if let Some(prev) = previous_stats {
|
||||
(
|
||||
calculate_packet_difference(
|
||||
current_stats.packets_received,
|
||||
prev.packets_received.unwrap_or(0),
|
||||
),
|
||||
calculate_packet_difference(current_stats.packets_sent, prev.packets_sent.unwrap_or(0)),
|
||||
calculate_packet_difference(
|
||||
current_stats.packets_dropped,
|
||||
prev.packets_dropped.unwrap_or(0),
|
||||
),
|
||||
)
|
||||
} else {
|
||||
(0, 0, 0) // No previous stats available
|
||||
};
|
||||
|
||||
sqlx::query!(
|
||||
r#"
|
||||
INSERT INTO mixnode_daily_stats (
|
||||
mix_id, date_utc, total_stake, packets_received, packets_sent, packets_dropped
|
||||
) VALUES (?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(mix_id, date_utc) DO UPDATE SET
|
||||
total_stake = excluded.total_stake,
|
||||
packets_received = mixnode_daily_stats.packets_received + excluded.packets_received,
|
||||
packets_sent = mixnode_daily_stats.packets_sent + excluded.packets_sent,
|
||||
packets_dropped = mixnode_daily_stats.packets_dropped + excluded.packets_dropped
|
||||
"#,
|
||||
node_id,
|
||||
date_utc,
|
||||
total_stake,
|
||||
diff_received,
|
||||
diff_sent,
|
||||
diff_dropped,
|
||||
)
|
||||
.execute(&mut *conn)
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,202 @@
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
pub mod helpers;
|
||||
use anyhow::Result;
|
||||
use helpers::{scrape_and_store_description, scrape_and_store_packet_stats};
|
||||
use sqlx::SqlitePool;
|
||||
use tracing::{debug, error, warn};
|
||||
|
||||
use crate::db::models::ScraperNodeInfo;
|
||||
use crate::db::queries::fetch_active_nodes;
|
||||
|
||||
const DESCRIPTION_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60 * 4); // 4 hours
|
||||
const PACKET_SCRAPE_INTERVAL: Duration = Duration::from_secs(60 * 60); // 1 hour
|
||||
const QUEUE_CHECK_INTERVAL: Duration = Duration::from_millis(250);
|
||||
const MAX_CONCURRENT_TASKS: usize = 5;
|
||||
|
||||
static TASK_COUNTER: AtomicUsize = AtomicUsize::new(0);
|
||||
static TASK_ID_COUNTER: AtomicUsize = AtomicUsize::new(0);
|
||||
|
||||
pub struct Scraper {
|
||||
pool: SqlitePool,
|
||||
description_queue: Arc<Mutex<Vec<ScraperNodeInfo>>>,
|
||||
packet_queue: Arc<Mutex<Vec<ScraperNodeInfo>>>,
|
||||
}
|
||||
|
||||
impl Scraper {
|
||||
pub fn new(pool: SqlitePool) -> Self {
|
||||
Self {
|
||||
pool,
|
||||
description_queue: Arc::new(Mutex::new(Vec::new())),
|
||||
packet_queue: Arc::new(Mutex::new(Vec::new())),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn start(&self) {
|
||||
self.spawn_description_scraper().await;
|
||||
self.spawn_packet_scraper().await;
|
||||
}
|
||||
|
||||
async fn spawn_description_scraper(&self) {
|
||||
let pool = self.pool.clone();
|
||||
let queue = self.description_queue.clone();
|
||||
tracing::info!("Starting description scraper");
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
if let Err(e) = Self::run_description_scraper(&pool, queue.clone()).await {
|
||||
error!("Description scraper failed: {}", e);
|
||||
}
|
||||
tokio::time::sleep(DESCRIPTION_SCRAPE_INTERVAL).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async fn spawn_packet_scraper(&self) {
|
||||
let pool = self.pool.clone();
|
||||
let queue = self.packet_queue.clone();
|
||||
tracing::info!("Starting packet scraper");
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
if let Err(e) = Self::run_packet_scraper(&pool, queue.clone()).await {
|
||||
error!("Packet scraper failed: {}", e);
|
||||
}
|
||||
tokio::time::sleep(PACKET_SCRAPE_INTERVAL).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async fn run_description_scraper(
|
||||
pool: &SqlitePool,
|
||||
queue: Arc<Mutex<Vec<ScraperNodeInfo>>>,
|
||||
) -> Result<()> {
|
||||
let nodes = fetch_active_nodes(pool).await?;
|
||||
if let Ok(mut queue_lock) = queue.lock() {
|
||||
queue_lock.extend(nodes);
|
||||
} else {
|
||||
warn!("Failed to acquire description queue lock");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
Self::process_description_queue(pool, queue).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn run_packet_scraper(
|
||||
pool: &SqlitePool,
|
||||
queue: Arc<Mutex<Vec<ScraperNodeInfo>>>,
|
||||
) -> Result<()> {
|
||||
let nodes = fetch_active_nodes(pool).await?;
|
||||
tracing::info!("Found {} active nodes", nodes.len());
|
||||
if let Ok(mut queue_lock) = queue.lock() {
|
||||
queue_lock.extend(nodes);
|
||||
} else {
|
||||
warn!("Failed to acquire packet queue lock");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
Self::process_packet_queue(pool, queue).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn process_description_queue(
|
||||
pool: &SqlitePool,
|
||||
queue: Arc<Mutex<Vec<ScraperNodeInfo>>>,
|
||||
) -> Result<()> {
|
||||
loop {
|
||||
let running_tasks = TASK_COUNTER.load(Ordering::Relaxed);
|
||||
|
||||
if running_tasks < MAX_CONCURRENT_TASKS {
|
||||
let node = {
|
||||
if let Ok(mut queue_lock) = queue.lock() {
|
||||
if queue_lock.is_empty() {
|
||||
TASK_ID_COUNTER.store(0, Ordering::Relaxed);
|
||||
break;
|
||||
}
|
||||
queue_lock.remove(0)
|
||||
} else {
|
||||
warn!("Failed to acquire description queue lock");
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
TASK_COUNTER.fetch_add(1, Ordering::Relaxed);
|
||||
let task_id = TASK_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
|
||||
let pool = pool.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
match Self::scrape_and_store_description(&pool, &node).await {
|
||||
Ok(_) => debug!(
|
||||
"✅ Description task #{} for node {} complete",
|
||||
task_id, node.node_id
|
||||
),
|
||||
Err(e) => debug!(
|
||||
"❌ Description task #{} for node {} failed: {}",
|
||||
task_id, node.node_id, e
|
||||
),
|
||||
}
|
||||
TASK_COUNTER.fetch_sub(1, Ordering::Relaxed);
|
||||
});
|
||||
} else {
|
||||
tokio::time::sleep(QUEUE_CHECK_INTERVAL).await;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn process_packet_queue(
|
||||
pool: &SqlitePool,
|
||||
queue: Arc<Mutex<Vec<ScraperNodeInfo>>>,
|
||||
) -> Result<()> {
|
||||
loop {
|
||||
let running_tasks = TASK_COUNTER.load(Ordering::Relaxed);
|
||||
|
||||
if running_tasks < MAX_CONCURRENT_TASKS {
|
||||
let node = {
|
||||
if let Ok(mut queue_lock) = queue.lock() {
|
||||
if queue_lock.is_empty() {
|
||||
TASK_ID_COUNTER.store(0, Ordering::Relaxed);
|
||||
break;
|
||||
}
|
||||
queue_lock.remove(0)
|
||||
} else {
|
||||
warn!("Failed to acquire packet queue lock");
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
TASK_COUNTER.fetch_add(1, Ordering::Relaxed);
|
||||
let task_id = TASK_ID_COUNTER.fetch_add(1, Ordering::Relaxed);
|
||||
let pool = pool.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
match Self::scrape_and_store_packet_stats(&pool, &node).await {
|
||||
Ok(_) => debug!(
|
||||
"✅ Packet stats task #{} for node {} complete",
|
||||
task_id, node.node_id
|
||||
),
|
||||
Err(e) => debug!(
|
||||
"❌ Packet stats task #{} for node {} failed: {}",
|
||||
task_id, node.node_id, e
|
||||
),
|
||||
}
|
||||
TASK_COUNTER.fetch_sub(1, Ordering::Relaxed);
|
||||
});
|
||||
} else {
|
||||
tokio::time::sleep(QUEUE_CHECK_INTERVAL).await;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn scrape_and_store_description(pool: &SqlitePool, node: &ScraperNodeInfo) -> Result<()> {
|
||||
scrape_and_store_description(pool, node).await
|
||||
}
|
||||
|
||||
async fn scrape_and_store_packet_stats(
|
||||
pool: &SqlitePool,
|
||||
node: &ScraperNodeInfo,
|
||||
) -> Result<()> {
|
||||
scrape_and_store_packet_stats(pool, node).await
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user