GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge commit '3cad2dbb341c23592d22856ced5c8020a7ea20b7' into simon/instrumented

Browse Source
This commit is contained in:
Simon Wicky
2023-03-28 13:21:46 +00:00
parent 263db0dbc3 3cad2dbb34
commit e5af0f5d5e
13 changed files with 384 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+295
View File
@@ -115,6 +115,17 @@ version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8da52d66c7071e2e3fa2a1e5c6d088fec47b593032b254f5e980de8ea54454d6"
[[package]]
name = "async-channel"
version = "1.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf46fee83e5ccffc220104713af3292ff9bc7c64c7de289f66dae8e38d826833"
dependencies = [
"concurrent-queue",
"event-listener",
"futures-core",
]
[[package]]
name = "async-stream"
version = "0.3.4"
@@ -455,6 +466,12 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
[[package]]
name = "castaway"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a2698f953def977c68f935bb0dfa959375ad4638570e969e2f1e9f433cbf1af6"
[[package]]
name = "cc"
version = "1.0.79"
@@ -732,6 +749,15 @@ dependencies = [
"unicode-width",
]
[[package]]
name = "concurrent-queue"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c278839b831783b70278b14df4d45e1beb1aad306c07bb796637de9a0e323e8e"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "console-api"
version = "0.4.0"
@@ -1156,6 +1182,36 @@ dependencies = [
"rustc_version 0.2.3",
]
[[package]]
name = "curl"
version = "0.4.44"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "509bd11746c7ac09ebd19f0b17782eae80aadee26237658a6b4808afb5c11a22"
dependencies = [
"curl-sys",
"libc",
"openssl-probe",
"openssl-sys",
"schannel",
"socket2",
"winapi",
]
[[package]]
name = "curl-sys"
version = "0.4.61+curl-8.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "14d05c10f541ae6f3bc5b3d923c20001f47db7d5f0b2bc6ad16490133842db79"
dependencies = [
"cc",
"libc",
"libz-sys",
"openssl-sys",
"pkg-config",
"vcpkg",
"winapi",
]
[[package]]
name = "curve25519-dalek"
version = "3.2.0"
@@ -1842,6 +1898,21 @@ version = "0.3.27"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "89d422fa3cbe3b40dca574ab087abb5bc98258ea57eea3fd6f1fa7162c778b91"
[[package]]
name = "futures-lite"
version = "1.12.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7694489acd39452c77daa48516b894c153f192c3578d5a839b62c58099fcbf48"
dependencies = [
"fastrand",
"futures-core",
"futures-io",
"memchr",
"parking",
"pin-project-lite",
"waker-fn",
]
[[package]]
name = "futures-macro"
version = "0.3.27"
@@ -2532,6 +2603,12 @@ dependencies = [
"web-sys",
]
[[package]]
name = "integer-encoding"
version = "3.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02"
[[package]]
name = "io-lifetimes"
version = "1.0.9"
@@ -2579,6 +2656,31 @@ dependencies = [
"windows-sys 0.45.0",
]
[[package]]
name = "isahc"
version = "1.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "334e04b4d781f436dc315cb1e7515bd96826426345d498149e4bde36b67f8ee9"
dependencies = [
"async-channel",
"castaway",
"crossbeam-utils",
"curl",
"curl-sys",
"event-listener",
"futures-lite",
"http",
"log",
"once_cell",
"polling",
"slab",
"sluice",
"tracing",
"tracing-futures",
"url",
"waker-fn",
]
[[package]]
name = "isocountry"
version = "0.3.2"
@@ -2910,6 +3012,7 @@ dependencies = [
"nym-task",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
@@ -2935,6 +3038,7 @@ dependencies = [
"thiserror",
"tokio",
"tokio-util",
"tracing",
"url",
"validator-client",
]
@@ -3525,6 +3629,8 @@ dependencies = [
"nym-task",
"nym-topology",
"nym-types",
"opentelemetry",
"opentelemetry-jaeger",
"pretty_env_logger",
"rand 0.7.3",
"rocket",
@@ -3534,6 +3640,10 @@ dependencies = [
"tokio",
"tokio-util",
"toml",
"tracing",
"tracing-flame",
"tracing-opentelemetry",
"tracing-subscriber",
"url",
"validator-client",
]
@@ -4118,6 +4228,105 @@ dependencies = [
"vcpkg",
]
[[package]]
name = "opentelemetry"
version = "0.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69d6c3d7288a106c0a363e4b0e8d308058d56902adefb16f4936f417ffef086e"
dependencies = [
"opentelemetry_api",
"opentelemetry_sdk",
]
[[package]]
name = "opentelemetry-http"
version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1edc79add46364183ece1a4542592ca593e6421c60807232f5b8f7a31703825d"
dependencies = [
"async-trait",
"bytes",
"http",
"isahc",
"opentelemetry_api",
]
[[package]]
name = "opentelemetry-jaeger"
version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e785d273968748578931e4dc3b4f5ec86b26e09d9e0d66b55adda7fce742f7a"
dependencies = [
"async-trait",
"futures",
"futures-executor",
"http",
"isahc",
"once_cell",
"opentelemetry",
"opentelemetry-http",
"opentelemetry-semantic-conventions",
"thiserror",
"thrift",
"tokio",
]
[[package]]
name = "opentelemetry-semantic-conventions"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b02e0230abb0ab6636d18e2ba8fa02903ea63772281340ccac18e0af3ec9eeb"
dependencies = [
"opentelemetry",
]
[[package]]
name = "opentelemetry_api"
version = "0.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c24f96e21e7acc813c7a8394ee94978929db2bcc46cf6b5014fc612bf7760c22"
dependencies = [
"fnv",
"futures-channel",
"futures-util",
"indexmap",
"js-sys",
"once_cell",
"pin-project-lite",
"thiserror",
]
[[package]]
name = "opentelemetry_sdk"
version = "0.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ca41c4933371b61c2a2f214bf16931499af4ec90543604ec828f7a625c09113"
dependencies = [
"async-trait",
"crossbeam-channel",
"dashmap 5.4.0",
"fnv",
"futures-channel",
"futures-executor",
"futures-util",
"once_cell",
"opentelemetry_api",
"percent-encoding",
"rand 0.8.5",
"thiserror",
"tokio",
"tokio-stream",
]
[[package]]
name = "ordered-float"
version = "1.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3305af35278dd29f46fcdd139e0b1fbfae2153f0e5928b39b035542dd31e37b7"
dependencies = [
"num-traits",
]
[[package]]
name = "os_str_bytes"
version = "6.5.0"
@@ -4148,6 +4357,12 @@ dependencies = [
"group 0.11.0",
]
[[package]]
name = "parking"
version = "2.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "427c3892f9e783d91cc128285287e70a59e206ca452770ece88a76f7a3eddd72"
[[package]]
name = "parking_lot"
version = "0.11.2"
@@ -4399,6 +4614,22 @@ dependencies = [
"plotters-backend",
]
[[package]]
name = "polling"
version = "2.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7e1f879b2998099c2d69ab9605d145d5b661195627eccc680002c4918a7fb6fa"
dependencies = [
"autocfg 1.1.0",
"bitflags",
"cfg-if",
"concurrent-queue",
"libc",
"log",
"pin-project-lite",
"windows-sys 0.45.0",
]
[[package]]
name = "poly1305"
version = "0.8.0"
@@ -5563,6 +5794,17 @@ dependencies = [
"autocfg 1.1.0",
]
[[package]]
name = "sluice"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d7400c0eff44aa2fcb5e31a5f24ba9716ed90138769e4977a2ba6014ae63eb5"
dependencies = [
"async-channel",
"futures-core",
"futures-io",
]
[[package]]
name = "smallvec"
version = "0.6.14"
@@ -6159,6 +6401,28 @@ dependencies = [
"once_cell",
]
[[package]]
name = "threadpool"
version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d050e60b33d41c19108b32cea32164033a9013fe3b46cbd4457559bfbf77afaa"
dependencies = [
"num_cpus",
]
[[package]]
name = "thrift"
version = "0.16.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09678c4cdbb4eed72e18b7c2af1329c69825ed16fcbac62d083fc3e2b0590ff0"
dependencies = [
"byteorder",
"integer-encoding",
"log",
"ordered-float",
"threadpool",
]
[[package]]
name = "time"
version = "0.1.45"
@@ -6452,6 +6716,17 @@ dependencies = [
"valuable",
]
[[package]]
name = "tracing-flame"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0bae117ee14789185e129aaee5d93750abe67fdc5a9a62650452bfe4e122a3a9"
dependencies = [
"lazy_static",
"tracing",
"tracing-subscriber",
]
[[package]]
name = "tracing-futures"
version = "0.2.5"
@@ -6473,6 +6748,20 @@ dependencies = [
"tracing-core",
]
[[package]]
name = "tracing-opentelemetry"
version = "0.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "21ebb87a95ea13271332df069020513ab70bdb5637ca42d6e492dc3bbbad48de"
dependencies = [
"once_cell",
"opentelemetry",
"tracing",
"tracing-core",
"tracing-log",
"tracing-subscriber",
]
[[package]]
name = "tracing-subscriber"
version = "0.3.16"
@@ -6797,6 +7086,12 @@ version = "0.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f"
[[package]]
name = "waker-fn"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9d5b2c62b4012a3e1eca5a7e077d13b3bf498c4073e33ccd58626607748ceeca"
[[package]]
name = "walkdir"
version = "2.3.3"
common/client-libs/mixnet-client/Cargo.toml
+1
View File
@@ -8,6 +8,7 @@ edition = "2021"
[dependencies]
futures = "0.3"
tracing = "0.1.37"
log = { workspace = true }
tokio = { version = "1.24.1", features = ["time", "net", "rt"] }
tokio-util = { version = "0.7.4", features = ["codec"] }
common/client-libs/mixnet-client/src/client.rs
+6 -2
View File
@@ -3,11 +3,12 @@
use futures::channel::mpsc;
use futures::StreamExt;
use log::*;
use tracing::*;
use nym_sphinx::framing::codec::SphinxCodec;
use nym_sphinx::framing::packet::FramedSphinxPacket;
use nym_sphinx::params::PacketMode;
use nym_sphinx::{addressing::nodes::NymNodeRoutingAddress, SphinxPacket};
use nym_sphinx::params::packet_sizes::PacketSize;
use std::collections::HashMap;
use std::io;
use std::net::SocketAddr;
@@ -197,6 +198,7 @@ impl Client {
}
impl SendWithoutResponse for Client {
#[instrument(level="info", skip(self, packet), "Sending packet to mixnet", fields(packet_size))]
fn send_without_response(
&mut self,
address: NymNodeRoutingAddress,
@@ -204,13 +206,15 @@ impl SendWithoutResponse for Client {
packet_mode: PacketMode,
) -> io::Result<()> {
trace!("Sending packet to {:?}", address);
let packet_size = PacketSize::get_type(packet.len()).unwrap();
Span::current().record("packet_size", field::debug(packet_size));
let framed_packet =
FramedSphinxPacket::new(packet, packet_mode, self.config.use_legacy_version);
if let Some(sender) = self.conn_new.get_mut(&address) {
if let Err(err) = sender.channel.try_send(framed_packet) {
if err.is_full() {
debug!("Connection to {} seems to not be able to handle all the traffic - dropping the current packet", address);
info!("Connection to {} seems to not be able to handle all the traffic - dropping the current packet", address);
// it's not a 'big' error, but we did not manage to send the packet
// if the queue is full, we can't really do anything but to drop the packet
Err(io::Error::new(
common/mixnode-common/Cargo.toml
+1
View File
@@ -16,6 +16,7 @@ serde = { version = "1.0", features = ["derive"] }
tokio = { version = "1.24.1", features = ["time", "macros", "rt", "net", "io-util"] }
tokio-util = { version = "0.7.4", features = ["codec"] }
url = "2.2"
tracing = "0.1.37"
thiserror = "1.0.37"
nym-crypto = { path = "../crypto" }
common/mixnode-common/src/packet_processor/processor.rs
+8 -2
View File
@@ -2,7 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
use crate::packet_processor::error::MixProcessingError;
use log::*;
use tracing::*;
use nym_sphinx_acknowledgements::surb_ack::SurbAck;
use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
use nym_sphinx_forwarding::packet::MixPacket;
@@ -57,6 +57,7 @@ impl SphinxPacketProcessor {
}
/// Takes the received framed packet and tries to unwrap it from the sphinx encryption.
#[instrument(level="debug", skip_all)]
fn perform_initial_unwrapping(
&self,
received: FramedSphinxPacket,
@@ -73,6 +74,7 @@ impl SphinxPacketProcessor {
/// Processed received forward hop packet - tries to extract next hop address, sets delay
/// and packs all the data in a way that can be easily sent to the next hop.
#[instrument(level="debug", skip_all)]
fn process_forward_hop(
&self,
packet: SphinxPacket,
@@ -88,6 +90,7 @@ impl SphinxPacketProcessor {
/// Split data extracted from the final hop sphinx packet into a SURBAck and message
/// that should get delivered to a client.
#[instrument(level="debug", skip_all)]
fn split_hop_data_into_ack_and_message(
&self,
mut extracted_data: Vec<u8>,
@@ -105,6 +108,7 @@ impl SphinxPacketProcessor {
/// Tries to extract a SURBAck that could be sent back into the mix network and message
/// that should get delivered to a client from received Sphinx packet.
#[instrument(level="debug", skip_all)]
fn split_into_ack_and_message(
&self,
data: Vec<u8>,
@@ -129,6 +133,7 @@ impl SphinxPacketProcessor {
/// Processed received final hop packet - tries to extract SURBAck out of it (assuming the
/// packet itself is not an ACK) and splits it from the message that should get delivered
/// to the destination.
#[instrument(level="debug", skip_all)]
fn process_final_hop(
&self,
destination: DestinationAddressBytes,
@@ -150,6 +155,7 @@ impl SphinxPacketProcessor {
/// Performs final processing for the unwrapped packet based on whether it was a forward hop
/// or a final hop.
#[instrument(level="debug", skip_all)]
fn perform_final_processing(
&self,
packet: ProcessedPacket,
@@ -167,7 +173,7 @@ impl SphinxPacketProcessor {
}
}
}
#[instrument(level="debug", skip_all, fields(packet_size=?received.packet_size()))]
pub fn process_received(
&self,
received: FramedSphinxPacket,
common/nymsphinx/params/src/packet_sizes.rs
+1
View File
@@ -233,3 +233,4 @@ mod tests {
assert_eq!(iv_size, ACK_IV_SIZE);
}
}
common/nymsphinx/params/src/packet_version.rs
+1
View File
@@ -67,3 +67,4 @@ impl Default for PacketVersion {
PacketVersion::Versioned(CURRENT_PACKET_VERSION_NUMBER)
}
}
mixnode/Cargo.toml
+6
View File
@@ -36,6 +36,12 @@ tokio = { version="1.21.2", features = ["rt-multi-thread", "net", "signal"] }
tokio-util = { version="0.7.3", features = ["codec"] }
toml = "0.5.8"
url = { version = "2.2", features = ["serde"] }
tracing = "0.1.37"
tracing-subscriber = {version = "0.3.16", features = ["registry", "std", "env-filter"]}
tracing-opentelemetry = "0.18.0"
opentelemetry-jaeger = {version = "0.17.0", features = ["collector_client","rt-tokio","isahc_collector_client"]}
opentelemetry = {version = "0.18.0", features = ["rt-tokio"]}
tracing-flame = "0.2.0"
atty = "0.2"
## internal
mixnode/src/main.rs
+30 -1
View File
@@ -7,6 +7,9 @@ extern crate rocket;
use ::nym_config::defaults::setup_env;
use clap::{crate_name, crate_version, Parser, ValueEnum};
use lazy_static::lazy_static;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::{EnvFilter, Registry, filter};
use tracing_flame::FlameLayer;
use nym_bin_common::logging::setup_logging;
use nym_bin_common::{build_information::BinaryBuildInformation, logging::banner};
@@ -62,7 +65,32 @@ impl Cli {
#[tokio::main]
async fn main() {
setup_logging();
//setup_logging();
//let tracer = opentelemetry_jaeger::new_agent_pipeline()
//.with_endpoint("143.42.21.138:6831")
//.with_service_name("nym_mixnode1")
//.with_auto_split_batch(true)
//.install_batch(opentelemetry::runtime::Tokio)
//.expect("Failed to initialize tracer");
//let jaeger_layer = tracing_opentelemetry::layer().with_tracer(tracer);
let hyper_filter = filter::filter_fn(|metadata| {!metadata.target().starts_with("hyper")});
let tokio_filter = filter::filter_fn(|metadata| {!metadata.target().starts_with("tokio")});
//let (flame_layer, _guard) = FlameLayer::with_file("./tracing.folded").unwrap();
let subscriber = Registry::default()
.with(EnvFilter::from_default_env())
.with(hyper_filter)
.with(tokio_filter)
.with(tracing_subscriber::fmt::layer().pretty());
//.with(flame_layer);
//.with(jaeger_layer);
tracing::subscriber::set_global_default(subscriber)
.expect("Failed to set global subscriber");
if atty::is(atty::Stream::Stdout) {
println!("{}", banner(crate_name!(), crate_version!()));
}
@@ -70,6 +98,7 @@ async fn main() {
let args = Cli::parse();
setup_env(args.config_env_file.as_ref());
commands::execute(args).await;
opentelemetry::global::shutdown_tracer_provider();
}
#[cfg(test)]
mixnode/src/node/listener/connection_handler/mod.rs
+13 -6
View File
@@ -7,8 +7,10 @@ use crate::node::listener::connection_handler::packet_processing::{
use crate::node::packet_delayforwarder::PacketDelayForwardSender;
use crate::node::TaskClient;
use futures::StreamExt;
use log::{error, info};
use tracing::{error, info, trace, debug, warn};
use tracing::*;
use nym_sphinx::forwarding::packet::MixPacket;
use nym_sphinx::params::PacketSize;
use nym_sphinx::framing::codec::SphinxCodec;
use nym_sphinx::framing::packet::FramedSphinxPacket;
use nym_sphinx::Delay as SphinxDelay;
@@ -16,6 +18,7 @@ use std::net::SocketAddr;
use tokio::net::TcpStream;
use tokio::time::Instant;
use tokio_util::codec::Framed;
use std::time::{SystemTime,UNIX_EPOCH};
pub(crate) mod packet_processing;
@@ -35,10 +38,12 @@ impl ConnectionHandler {
delay_forwarding_channel,
}
}
#[instrument(level="debug", skip_all, "Sending to packet forwarder", fields(packet_size))]
fn delay_and_forward_packet(&self, mix_packet: MixPacket, delay: Option<SphinxDelay>) {
// determine instant at which packet should get forwarded. this way we minimise effect of
// being stuck in the queue [of the channel] to get inserted into the delay queue
let packet_size = PacketSize::get_type(mix_packet.sphinx_packet().len()).unwrap();
Span::current().record("packet_size", field::debug(packet_size));
let forward_instant = delay.map(|delay| Instant::now() + delay.to_duration());
// if unbounded_send() failed it means that the receiver channel was disconnected
@@ -47,7 +52,7 @@ impl ConnectionHandler {
.unbounded_send((mix_packet, forward_instant))
.expect("the delay-forwarder has died!");
}
#[instrument(level="info", skip_all, "Handling packet",fields(packet_size=?framed_sphinx_packet.packet_size()))]
fn handle_received_packet(&self, framed_sphinx_packet: FramedSphinxPacket) {
//
// TODO: here be replay attack detection - it will require similar key cache to the one in
@@ -69,7 +74,7 @@ impl ConnectionHandler {
},
}
}
#[instrument(level="info", skip_all, "Connection handling", fields(address=%remote))]
pub(crate) async fn handle_connection(
self,
conn: TcpStream,
@@ -83,7 +88,7 @@ impl ConnectionHandler {
tokio::select! {
biased;
_ = shutdown.recv() => {
log::trace!("ConnectionHandler: received shutdown");
trace!("ConnectionHandler: received shutdown");
}
Some(framed_sphinx_packet) = framed_conn.next() => {
match framed_sphinx_packet {
@@ -96,7 +101,9 @@ impl ConnectionHandler {
// in theory we could process multiple sphinx packet from the same connection in parallel,
// but we already handle multiple concurrent connections so if anything, making
// that change would only slow things down
//println!("{:?}_In_{:?}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros(), framed_sphinx_packet.packet_size().size());
self.handle_received_packet(framed_sphinx_packet);
//println!("{:?}_Processed_{:?}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros(), mix_packet.sphinx_packet().len());
}
Err(err) => {
error!(
@@ -113,6 +120,6 @@ impl ConnectionHandler {
"Closing connection from {:?}",
framed_conn.into_inner().peer_addr()
);
log::trace!("ConnectionHandler: Exiting");
trace!("ConnectionHandler: Exiting");
}
}
mixnode/src/node/listener/mod.rs
+7 -6
View File
@@ -2,7 +2,8 @@
// SPDX-License-Identifier: Apache-2.0
use crate::node::listener::connection_handler::ConnectionHandler;
use log::error;
use tracing::*;
use tracing::{error, info, trace, warn};
use std::net::SocketAddr;
use std::process;
use tokio::net::TcpListener;
@@ -21,9 +22,9 @@ impl Listener {
pub(crate) fn new(address: SocketAddr, shutdown: TaskClient) -> Self {
Listener { address, shutdown }
}
#[instrument(level="info", skip_all, "Mixnet Listener")]
async fn run(&mut self, connection_handler: ConnectionHandler) {
log::trace!("Starting Listener");
trace!("Starting Listener");
let listener = match TcpListener::bind(self.address).await {
Ok(listener) => listener,
Err(err) => {
@@ -36,20 +37,20 @@ impl Listener {
tokio::select! {
biased;
_ = self.shutdown.recv() => {
log::trace!("Listener: Received shutdown");
trace!("Listener: Received shutdown");
}
connection = listener.accept() => {
match connection {
Ok((socket, remote_addr)) => {
let handler = connection_handler.clone();
tokio::spawn(handler.handle_connection(socket, remote_addr, self.shutdown.clone()));
tokio::spawn(handler.handle_connection(socket, remote_addr, self.shutdown.clone()).instrument(info_span!("Connection handling", address = %remote_addr)));
}
Err(err) => warn!("Failed to accept incoming connection - {err}"),
}
},
};
}
log::trace!("Listener: Exiting");
trace!("Listener: Exiting");
}
pub(crate) fn start(mut self, connection_handler: ConnectionHandler) -> JoinHandle<()> {
mixnode/src/node/mod.rs
+5 -4
View File
@@ -17,9 +17,10 @@ use crate::node::listener::Listener;
use crate::node::node_description::NodeDescription;
use crate::node::node_statistics::SharedNodeStats;
use crate::node::packet_delayforwarder::{DelayForwarder, PacketDelayForwardSender};
use tracing::*;
use tracing::{info, error, warn};
use crate::OutputFormat;
use colored::Colorize;
use log::{error, info, warn};
use mixnode_common::verloc::{self, AtomicVerlocResult, VerlocMeasurer};
use nym_bin_common::version_checker::parse_version;
use nym_config::NymConfig;
@@ -215,7 +216,7 @@ impl MixNode {
let packet_sender = packet_forwarder.sender();
tokio::spawn(async move { packet_forwarder.run().await });
tokio::spawn(async move { packet_forwarder.run().await }.instrument(info_span!("Packet delay forwarder")));
packet_sender
}
@@ -294,7 +295,7 @@ impl MixNode {
async fn wait_for_interrupt(&self, shutdown: TaskManager) {
let _res = shutdown.catch_interrupt().await;
log::info!("Stopping nym mixnode");
info!("Stopping nym mixnode");
}
pub async fn run(&mut self) {
@@ -304,7 +305,7 @@ impl MixNode {
if duplicate_node_key == self.identity_keypair.public_key().to_base58_string() {
warn!("You seem to have bonded your mixnode before starting it - that's highly unrecommended as in the future it might result in slashing");
} else {
log::error!(
error!(
"Our announce-host is identical to an existing node's announce-host! (its key is {:?})",
duplicate_node_key
);
mixnode/src/node/packet_delayforwarder.rs
+10 -5
View File
@@ -4,10 +4,13 @@
use crate::node::node_statistics::UpdateSender;
use futures::channel::mpsc;
use futures::StreamExt;
use nym_sphinx::params::packet_sizes::PacketSize;
use nym_nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue};
use nym_sphinx::forwarding::packet::MixPacket;
use std::io;
use tokio::time::Instant;
use tracing::*;
use tracing::{trace};
use super::TaskClient;
@@ -55,11 +58,13 @@ where
pub(crate) fn sender(&self) -> PacketDelayForwardSender {
self.packet_sender.clone()
}
#[instrument(level="debug", skip_all, "Forwarding packet", fields(packet_size))]
fn forward_packet(&mut self, packet: MixPacket) {
let next_hop = packet.next_hop();
let packet_mode = packet.packet_mode();
let sphinx_packet = packet.into_sphinx_packet();
let packet_size = PacketSize::get_type(sphinx_packet.len()).unwrap();
Span::current().record("packet_size", field::debug(packet_size));
if let Err(err) =
self.mixnet_client
@@ -87,7 +92,7 @@ where
let delayed_packet = packet.into_inner();
self.forward_packet(delayed_packet)
}
#[instrument(level="debug", skip_all, "Handling packet", fields(packet_size))]
fn handle_new_packet(&mut self, new_packet: (MixPacket, Option<Instant>)) {
// in case of a zero delay packet, don't bother putting it in the delay queue,
// just forward it immediately
@@ -105,7 +110,7 @@ where
}
pub(crate) async fn run(&mut self) {
log::trace!("Starting DelayForwarder");
trace!("Starting DelayForwarder");
loop {
tokio::select! {
delayed = self.delay_queue.next() => {
@@ -117,12 +122,12 @@ where
self.handle_new_packet(new_packet.unwrap())
}
_ = self.shutdown.recv() => {
log::trace!("DelayForwarder: Received shutdown");
trace!("DelayForwarder: Received shutdown");
break;
}
}
}
log::trace!("DelayForwarder: Exiting");
trace!("DelayForwarder: Exiting");
}
}