Metrics reporting
This commit is contained in:
Jedrzej Stuczynski
Generated
+122
-10
@@ -273,7 +273,7 @@ dependencies = [
|
||||
"idna 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"publicsuffix 1.5.4 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"try_from 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
@@ -513,6 +513,29 @@ name = "futures"
|
||||
version = "0.1.29"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "futures"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-channel"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-core"
|
||||
version = "0.3.1"
|
||||
@@ -527,6 +550,60 @@ dependencies = [
|
||||
"num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-executor"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-io"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "futures-macro"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"syn 1.0.8 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "futures-sink"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "futures-task"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "futures-util"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"memchr 2.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "generic-array"
|
||||
version = "0.12.3"
|
||||
@@ -911,9 +988,12 @@ dependencies = [
|
||||
"clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"dirs 2.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"pem 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"rand_distr 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"reqwest 0.9.24 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"sfw-provider-requests 0.1.0",
|
||||
"sphinx 0.1.0",
|
||||
"tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
@@ -927,6 +1007,7 @@ dependencies = [
|
||||
"base64 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"nym-client 0.1.0",
|
||||
"sphinx 0.1.0",
|
||||
"tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
@@ -1016,6 +1097,11 @@ name = "pin-project-lite"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "pin-utils"
|
||||
version = "0.1.0-alpha.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "pkg-config"
|
||||
version = "0.3.17"
|
||||
@@ -1026,6 +1112,21 @@ name = "ppv-lite86"
|
||||
version = "0.2.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "proc-macro-hack"
|
||||
version = "0.5.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"syn 1.0.8 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "proc-macro-nested"
|
||||
version = "0.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "proc-macro2"
|
||||
version = "1.0.6"
|
||||
@@ -1262,7 +1363,7 @@ dependencies = [
|
||||
"mime 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"mime_guess 2.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"native-tls 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde_urlencoded 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
@@ -1352,15 +1453,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
|
||||
[[package]]
|
||||
name = "serde"
|
||||
version = "1.0.103"
|
||||
version = "1.0.104"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"serde_derive 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "serde_derive"
|
||||
version = "1.0.103"
|
||||
version = "1.0.104"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
@@ -1375,7 +1476,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"ryu 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1385,7 +1486,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
dependencies = [
|
||||
"dtoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
"url 1.7.2 (registry+https://github.com/rust-lang/crates.io-index)",
|
||||
]
|
||||
|
||||
@@ -1963,8 +2064,16 @@ dependencies = [
|
||||
"checksum fuchsia-zircon 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2e9763c69ebaae630ba35f74888db465e49e259ba1bc0eda7d06f4a067615d82"
|
||||
"checksum fuchsia-zircon-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7"
|
||||
"checksum futures 0.1.29 (registry+https://github.com/rust-lang/crates.io-index)" = "1b980f2816d6ee8673b6517b52cb0e808a180efc92e5c19d02cdda79066703ef"
|
||||
"checksum futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b6f16056ecbb57525ff698bb955162d0cd03bee84e6241c27ff75c08d8ca5987"
|
||||
"checksum futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fcae98ca17d102fd8a3603727b9259fcf7fa4239b603d2142926189bc8999b86"
|
||||
"checksum futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "79564c427afefab1dfb3298535b21eda083ef7935b4f0ecbfcb121f0aec10866"
|
||||
"checksum futures-cpupool 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "ab90cde24b3319636588d0c35fe03b1333857621051837ed769faefb4c2162e4"
|
||||
"checksum futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1e274736563f686a837a0568b478bdabfeaec2dca794b5649b04e2fe1627c231"
|
||||
"checksum futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e676577d229e70952ab25f3945795ba5b16d63ca794ca9d2c860e5595d20b5ff"
|
||||
"checksum futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "52e7c56c15537adb4f76d0b7a76ad131cb4d2f4f32d3b0bcabcbe1c7c5e87764"
|
||||
"checksum futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "171be33efae63c2d59e6dbba34186fe0d6394fb378069a76dfd80fdcffd43c16"
|
||||
"checksum futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0bae52d6b29cf440e298856fec3965ee6fa71b06aa7495178615953fd669e5f9"
|
||||
"checksum futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c0d66274fb76985d3c62c886d1da7ac4c0903a8c9f754e8fe0f35a6a6cc39e76"
|
||||
"checksum generic-array 0.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "c68f0274ae0e023facc3c97b2e00f076be70e254bc851d972503b328db79b2ec"
|
||||
"checksum getrandom 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "e7db7ca94ed4cd01190ceee0d8a8052f08a247aa1b469a7f68c6a3b71afcf407"
|
||||
"checksum h2 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)" = "a5b34c246847f938a410a03c5458c7fee2274436675e76d8b903c08efc29c462"
|
||||
@@ -2015,8 +2124,11 @@ dependencies = [
|
||||
"checksum percent-encoding 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "31010dd2e1ac33d5b46a5b413495239882813e0369f8ed8a5e266f173602f831"
|
||||
"checksum percent-encoding 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"
|
||||
"checksum pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f0af6cbca0e6e3ce8692ee19fb8d734b641899e07b68eb73e9bbbd32f1703991"
|
||||
"checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587"
|
||||
"checksum pkg-config 0.3.17 (registry+https://github.com/rust-lang/crates.io-index)" = "05da548ad6865900e60eaba7f589cc0783590a92e940c26953ff81ddbab2d677"
|
||||
"checksum ppv-lite86 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "74490b50b9fbe561ac330df47c08f3f33073d2d00c150f719147d7c54522fa1b"
|
||||
"checksum proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)" = "ecd45702f76d6d3c75a80564378ae228a85f0b59d2f3ed43c91b4a69eb2ebfc5"
|
||||
"checksum proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "369a6ed065f249a159e06c45752c780bda2fb53c995718f9e484d08daa9eb42e"
|
||||
"checksum proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "9c9e470a8dc4aeae2dee2f335e8f533e2d4b347e1434e5671afc49b054592f27"
|
||||
"checksum publicsuffix 1.5.4 (registry+https://github.com/rust-lang/crates.io-index)" = "3bbaa49075179162b49acac1c6aa45fb4dafb5f13cf6794276d77bc7fd95757b"
|
||||
"checksum quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "053a8c8bcc71fcce321828dc897a98ab9760bef03a4fc36693c231e5b3216cfe"
|
||||
@@ -2052,8 +2164,8 @@ dependencies = [
|
||||
"checksum security-framework-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "e31493fc37615debb8c5090a7aeb4a9730bc61e77ab10b9af59f1a202284f895"
|
||||
"checksum semver 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403"
|
||||
"checksum semver-parser 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3"
|
||||
"checksum serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)" = "1217f97ab8e8904b57dd22eb61cde455fa7446a9c1cf43966066da047c1f3702"
|
||||
"checksum serde_derive 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)" = "a8c6faef9a2e64b0064f48570289b4bf8823b7581f1d6157c1b52152306651d0"
|
||||
"checksum serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)" = "414115f25f818d7dfccec8ee535d76949ae78584fc4f79a6f45a904bf8ab4449"
|
||||
"checksum serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)" = "128f9e303a5a29922045a830221b8f78ec74a5f544944f3d5984f8ec3895ef64"
|
||||
"checksum serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)" = "48c575e0cc52bdd09b47f330f646cf59afc586e9c4e3ccd6fc1f625b8ea1dad7"
|
||||
"checksum serde_urlencoded 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)" = "642dd69105886af2efd227f75a520ec9b44a820d65bc133a9131f7d229fd165a"
|
||||
"checksum sha-1 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)" = "23962131a91661d643c98940b20fcaffe62d776a823247be80a48fcb8b6fce68"
|
||||
|
||||
@@ -10,6 +10,7 @@ edition = "2018"
|
||||
base64 = "0.11.0"
|
||||
clap = "2.33.0"
|
||||
curve25519-dalek = "1.2.3"
|
||||
futures = "0.3.1"
|
||||
nym-client = { path = "../nym-client" }
|
||||
sphinx = { path = "../sphinx" }
|
||||
tokio = { version = "0.2", features = ["full"] }
|
||||
|
||||
@@ -9,6 +9,7 @@ pub struct MixPeer {
|
||||
impl MixPeer {
|
||||
// note that very soon `next_hop_address` will be changed to `next_hop_metadata`
|
||||
pub fn new(next_hop_address: [u8; 32]) -> MixPeer {
|
||||
println!("constructing next hop from: {:?}", next_hop_address);
|
||||
let b = next_hop_address;
|
||||
let host = Ipv4Addr::new(b[0], b[1], b[2], b[3]);
|
||||
let port: u16 = u16::from_be_bytes([b[4], b[5]]);
|
||||
@@ -26,6 +27,7 @@ impl MixPeer {
|
||||
}
|
||||
|
||||
pub fn to_string(&self) -> String {
|
||||
println!("going to report metrics for: {:?}. original: {:?}", self.connection.to_string(), self.connection);
|
||||
self.connection.to_string()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
use std::sync::{RwLock, Arc};
|
||||
use futures::lock::Mutex;
|
||||
use futures::channel::mpsc;
|
||||
use futures::{StreamExt, SinkExt};
|
||||
use std::time::Duration;
|
||||
use nym_client::clients::directory;
|
||||
use nym_client::clients::directory::DirectoryClient;
|
||||
use nym_client::clients::directory::requests::metrics_mixes_post::MetricsMixPoster;
|
||||
use nym_client::clients::directory::metrics::MixMetric;
|
||||
use curve25519_dalek::montgomery::MontgomeryPoint;
|
||||
use std::collections::HashMap;
|
||||
|
||||
const METRICS_INTERVAL: u64 = 3;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct MetricsReporter {
|
||||
received: u64,
|
||||
sent: HashMap<String, u64>
|
||||
}
|
||||
|
||||
impl MetricsReporter {
|
||||
pub(crate) fn new() -> Self {
|
||||
MetricsReporter{
|
||||
received: 0,
|
||||
sent: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn add_arc_mutex(self) -> Arc<Mutex<Self>> {
|
||||
Arc::new(Mutex::new(self))
|
||||
}
|
||||
|
||||
async fn increment_received_metrics(mut metrics: Arc<Mutex<MetricsReporter>>) {
|
||||
let mut unlocked = metrics.lock().await;
|
||||
unlocked.received += 1;
|
||||
println!("new metrics: {:?}", unlocked);
|
||||
}
|
||||
|
||||
pub(crate) async fn run_received_metrics_control(metrics: Arc<Mutex<MetricsReporter>>, mut rx: mpsc::Receiver<()>) {
|
||||
while let Some(received_metric) = rx.next().await {
|
||||
println!("[METRIC] - received new packet!");
|
||||
MetricsReporter::increment_received_metrics(metrics.clone()).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn increment_sent_metrics(mut metrics: Arc<Mutex<MetricsReporter>>, sent_to: String) {
|
||||
let mut unlocked = metrics.lock().await;
|
||||
let receiver_count = unlocked.sent.entry(sent_to).or_insert(0);
|
||||
*receiver_count += 1;
|
||||
|
||||
// unlocked.sent += 1;
|
||||
println!("new metrics: {:?}", unlocked);
|
||||
}
|
||||
|
||||
pub(crate) async fn run_sent_metrics_control(metrics: Arc<Mutex<MetricsReporter>>, mut rx: mpsc::Receiver<String>) {
|
||||
while let Some(sent_metric) = rx.next().await {
|
||||
println!("[METRIC] - sent new packet!");
|
||||
MetricsReporter::increment_sent_metrics(metrics.clone(), sent_metric).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn acquire_and_reset_metrics(mut metrics: Arc<Mutex<MetricsReporter>>) -> (u64, HashMap<String, u64>) {
|
||||
let mut unlocked = metrics.lock().await;
|
||||
let received = unlocked.received;
|
||||
|
||||
println!("RESETTING METRICS!");
|
||||
let sent = std::mem::replace(&mut unlocked.sent, HashMap::new());
|
||||
unlocked.received = 0;
|
||||
|
||||
(received, sent)
|
||||
}
|
||||
|
||||
pub(crate) async fn run_metrics_sender(metrics: Arc<Mutex<MetricsReporter>>, cfg: directory::Config, pub_key_str: String) {
|
||||
let delay_duration = Duration::from_secs(METRICS_INTERVAL);
|
||||
let directory_client= directory::Client::new(cfg);
|
||||
loop {
|
||||
tokio::time::delay_for(delay_duration).await;
|
||||
let (received, sent) = MetricsReporter::acquire_and_reset_metrics(metrics.clone()).await;
|
||||
directory_client.metrics_post.post(&MixMetric{
|
||||
pub_key: pub_key_str.clone(),
|
||||
received,
|
||||
sent,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
+70
-27
@@ -1,18 +1,23 @@
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::Duration;
|
||||
|
||||
use futures::channel::mpsc;
|
||||
use curve25519_dalek::montgomery::MontgomeryPoint;
|
||||
use curve25519_dalek::scalar::Scalar;
|
||||
use sphinx::header::delays::Delay as SphinxDelay;
|
||||
use sphinx::{ProcessedPacket, SphinxPacket};
|
||||
use tokio::prelude::*;
|
||||
use tokio::runtime::Runtime;
|
||||
use futures::lock::Mutex;
|
||||
|
||||
use crate::mix_peer::MixPeer;
|
||||
use crate::node::metrics::MetricsReporter;
|
||||
use futures::SinkExt;
|
||||
use nym_client::clients::directory;
|
||||
|
||||
mod presence;
|
||||
pub mod runner;
|
||||
mod metrics;
|
||||
|
||||
pub struct Config {
|
||||
directory_server: String,
|
||||
@@ -25,8 +30,7 @@ pub struct Config {
|
||||
impl Config {
|
||||
pub fn public_key_string(&self) -> String {
|
||||
let key_bytes = self.public_key.to_bytes().to_vec();
|
||||
let b64 = base64::encode_config(&key_bytes, base64::URL_SAFE);
|
||||
b64.to_string()
|
||||
base64::encode_config(&key_bytes, base64::URL_SAFE)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -50,15 +54,18 @@ struct ForwardingData {
|
||||
packet: SphinxPacket,
|
||||
delay: SphinxDelay,
|
||||
recipient: MixPeer,
|
||||
sent_metrics_tx: mpsc::Sender<String>,
|
||||
}
|
||||
|
||||
// TODO: this will need to be changed if MixPeer will live longer than our Forwarding Data
|
||||
impl ForwardingData {
|
||||
fn new(packet: SphinxPacket, delay: SphinxDelay, recipient: MixPeer) -> Self {
|
||||
fn new(packet: SphinxPacket, delay: SphinxDelay, recipient: MixPeer, sent_metrics_tx: mpsc::Sender<String>,
|
||||
) -> Self {
|
||||
ForwardingData {
|
||||
packet,
|
||||
delay,
|
||||
recipient,
|
||||
sent_metrics_tx,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -66,28 +73,36 @@ impl ForwardingData {
|
||||
// ProcessingData defines all data required to correctly unwrap sphinx packets
|
||||
struct ProcessingData {
|
||||
secret_key: Scalar,
|
||||
received_metrics_tx: mpsc::Sender<()>,
|
||||
sent_metrics_tx: mpsc::Sender<String>,
|
||||
}
|
||||
|
||||
impl ProcessingData {
|
||||
fn new(secret_key: Scalar) -> Self {
|
||||
ProcessingData { secret_key }
|
||||
fn new(secret_key: Scalar, received_metrics_tx: mpsc::Sender<()>, sent_metrics_tx: mpsc::Sender<String>) -> Self {
|
||||
ProcessingData { secret_key, received_metrics_tx, sent_metrics_tx }
|
||||
}
|
||||
|
||||
fn add_arc_rwlock(self) -> Arc<RwLock<Self>> {
|
||||
Arc::new(RwLock::new(self))
|
||||
fn add_arc_mutex(self) -> Arc<Mutex<Self>> {
|
||||
Arc::new(Mutex::new(self))
|
||||
}
|
||||
}
|
||||
|
||||
struct PacketProcessor {}
|
||||
|
||||
impl PacketProcessor {
|
||||
pub fn process_sphinx_data_packet(
|
||||
pub async fn process_sphinx_data_packet(
|
||||
packet_data: &[u8],
|
||||
processing_data: Arc<RwLock<ProcessingData>>,
|
||||
processing_data: Arc<Mutex<ProcessingData>>,
|
||||
) -> Result<ForwardingData, MixProcessingError> {
|
||||
// we received something resembling a sphinx packet, report it!
|
||||
let processing_data = processing_data.lock().await;
|
||||
let mut received_sender = processing_data.received_metrics_tx.clone();
|
||||
|
||||
received_sender.send(()).await;
|
||||
|
||||
let packet = SphinxPacket::from_bytes(packet_data.to_vec())?;
|
||||
let (next_packet, next_hop_address, delay) =
|
||||
match packet.process(processing_data.read().unwrap().secret_key) {
|
||||
match packet.process(processing_data.secret_key) {
|
||||
ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay) => {
|
||||
(packet, address, delay)
|
||||
}
|
||||
@@ -96,35 +111,38 @@ impl PacketProcessor {
|
||||
|
||||
let next_mix = MixPeer::new(next_hop_address);
|
||||
|
||||
let fwd_data = ForwardingData::new(next_packet, delay, next_mix);
|
||||
let fwd_data = ForwardingData::new(next_packet, delay, next_mix, processing_data.sent_metrics_tx.clone());
|
||||
Ok(fwd_data)
|
||||
}
|
||||
|
||||
async fn wait_and_forward(forwarding_data: ForwardingData) {
|
||||
async fn wait_and_forward(mut forwarding_data: ForwardingData) {
|
||||
let delay_duration = Duration::from_nanos(forwarding_data.delay.get_value());
|
||||
println!("client says to wait for {:?}", delay_duration);
|
||||
tokio::time::delay_for(delay_duration).await;
|
||||
println!("waited {:?} - time to forward the packet!", delay_duration);
|
||||
println!("waited {:?} - time to forward the packet! (and update metrics!)", delay_duration);
|
||||
forwarding_data.sent_metrics_tx.send(forwarding_data.recipient.to_string()).await;
|
||||
|
||||
match forwarding_data
|
||||
.recipient
|
||||
.send(forwarding_data.packet.to_bytes())
|
||||
.await
|
||||
{
|
||||
Ok(()) => (),
|
||||
Err(e) => {
|
||||
println!(
|
||||
"failed to write bytes to next mix peer. err = {:?}",
|
||||
e.to_string()
|
||||
);
|
||||
{
|
||||
Ok(()) => (),
|
||||
Err(e) => {
|
||||
println!(
|
||||
"failed to write bytes to next mix peer. err = {:?}",
|
||||
e.to_string()
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// the MixNode will live for whole duration of this program
|
||||
pub struct MixNode {
|
||||
directory_server: String,
|
||||
network_address: SocketAddr,
|
||||
public_key: MontgomeryPoint,
|
||||
secret_key: Scalar,
|
||||
layer: usize,
|
||||
}
|
||||
@@ -132,15 +150,17 @@ pub struct MixNode {
|
||||
impl MixNode {
|
||||
pub fn new(config: &Config) -> Self {
|
||||
MixNode {
|
||||
directory_server: config.directory_server.clone(),
|
||||
network_address: config.socket_address,
|
||||
secret_key: config.secret_key,
|
||||
public_key: config.public_key,
|
||||
layer: config.layer,
|
||||
}
|
||||
}
|
||||
|
||||
async fn process_socket_connection(
|
||||
mut socket: tokio::net::TcpStream,
|
||||
processing_data: Arc<RwLock<ProcessingData>>,
|
||||
processing_data: Arc<Mutex<ProcessingData>>,
|
||||
) {
|
||||
// NOTE: processing_data is copied here!!
|
||||
let mut buf = [0u8; sphinx::PACKET_SIZE];
|
||||
@@ -157,8 +177,8 @@ impl MixNode {
|
||||
let fwd_data = PacketProcessor::process_sphinx_data_packet(
|
||||
buf.as_ref(),
|
||||
processing_data.clone(),
|
||||
)
|
||||
.unwrap();
|
||||
).await
|
||||
.unwrap();
|
||||
PacketProcessor::wait_and_forward(fwd_data).await;
|
||||
}
|
||||
Err(e) => {
|
||||
@@ -175,14 +195,37 @@ impl MixNode {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn start_listening(&self) -> Result<(), Box<dyn std::error::Error>> {
|
||||
pub fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
|
||||
// Create the runtime, probably later move it to MixNode itself?
|
||||
let mut rt = Runtime::new()?;
|
||||
|
||||
let (received_tx, received_rx) = mpsc::channel(1024);
|
||||
let (sent_tx, sent_rx) = mpsc::channel(1024);
|
||||
|
||||
let directory_cfg = directory::Config { base_url: self.directory_server.clone() };
|
||||
let pub_key_str = base64::encode_config(&self.public_key.to_bytes().to_vec(), base64::URL_SAFE);
|
||||
|
||||
let metrics = MetricsReporter::new().add_arc_mutex();
|
||||
rt.spawn(MetricsReporter::run_received_metrics_control(metrics.clone(), received_rx));
|
||||
rt.spawn(MetricsReporter::run_sent_metrics_control(metrics.clone(), sent_rx));
|
||||
rt.spawn(MetricsReporter::run_metrics_sender(metrics, directory_cfg, pub_key_str));
|
||||
// let mut foo = received_tx.clone();
|
||||
// let mut bar = sent_tx.clone();
|
||||
// let's increase some metrics!
|
||||
// rt.spawn(async move {
|
||||
// loop {
|
||||
// tokio::time::delay_for(Duration::from_secs(2)).await;
|
||||
// foo.send(()).await;
|
||||
// bar.send("bar".to_string()).await;
|
||||
// foo.send(()).await;
|
||||
// }
|
||||
// });
|
||||
// let (sent_tx, sent_rx) = mpsc::channel(1024);
|
||||
|
||||
// Spawn the root task
|
||||
rt.block_on(async {
|
||||
let mut listener = tokio::net::TcpListener::bind(self.network_address).await?;
|
||||
let processing_data = ProcessingData::new(self.secret_key).add_arc_rwlock();
|
||||
let processing_data = ProcessingData::new(self.secret_key, received_tx, sent_tx).add_arc_mutex();
|
||||
|
||||
loop {
|
||||
let (socket, _) = listener.accept().await?;
|
||||
|
||||
@@ -20,7 +20,7 @@ impl Notifier {
|
||||
let presence = MixNodePresence {
|
||||
host: node_config.socket_address.to_string(), // note: the directory server formats the real incoming IP itself
|
||||
pub_key: node_config.public_key_string(),
|
||||
layer: 666,
|
||||
layer: node_config.layer as u64,
|
||||
last_seen: 666,
|
||||
};
|
||||
Notifier {
|
||||
|
||||
+1
-1
@@ -21,7 +21,7 @@ pub fn start(matches: &ArgMatches) {
|
||||
let notifier = presence::Notifier::new(&config);
|
||||
notifier.run();
|
||||
});
|
||||
mix.start_listening().unwrap();
|
||||
mix.start().unwrap();
|
||||
}
|
||||
|
||||
fn new_config(matches: &ArgMatches) -> node::Config {
|
||||
|
||||
Reference in New Issue
Block a user