From 5281895d5b7b3eba91f2a928041035a99a3001d5 Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Tue, 28 Mar 2023 12:39:06 +0000 Subject: [PATCH] mixnode instrumentation --- common/client-libs/mixnet-client/Cargo.toml | 1 + .../client-libs/mixnet-client/src/client.rs | 8 +- common/mixnode-common/Cargo.toml | 1 + .../src/packet_processor/processor.rs | 15 ++-- common/nymsphinx/params/src/packet_sizes.rs | 86 ++++++++++++++++++- common/nymsphinx/params/src/packet_version.rs | 11 +++ mixnode/Cargo.toml | 6 ++ mixnode/src/main.rs | 31 ++++++- .../node/listener/connection_handler/mod.rs | 19 ++-- mixnode/src/node/listener/mod.rs | 13 +-- mixnode/src/node/mod.rs | 17 ++-- mixnode/src/node/packet_delayforwarder.rs | 15 ++-- 12 files changed, 188 insertions(+), 35 deletions(-) diff --git a/common/client-libs/mixnet-client/Cargo.toml b/common/client-libs/mixnet-client/Cargo.toml index 5267ecb366..1d8e6abd1e 100644 --- a/common/client-libs/mixnet-client/Cargo.toml +++ b/common/client-libs/mixnet-client/Cargo.toml @@ -8,6 +8,7 @@ edition = "2021" [dependencies] futures = "0.3" +tracing = "0.1.37" log = "0.4.8" tokio = { version = "1.21.2", features = ["time", "net", "rt"] } tokio-util = { version = "0.7.3", features = ["codec"] } diff --git a/common/client-libs/mixnet-client/src/client.rs b/common/client-libs/mixnet-client/src/client.rs index 600268b69b..7cc02cc15b 100644 --- a/common/client-libs/mixnet-client/src/client.rs +++ b/common/client-libs/mixnet-client/src/client.rs @@ -3,10 +3,11 @@ use futures::channel::mpsc; use futures::StreamExt; -use log::*; +use tracing::*; use nymsphinx::framing::codec::SphinxCodec; use nymsphinx::framing::packet::FramedSphinxPacket; use nymsphinx::params::PacketMode; +use nymsphinx::params::packet_sizes::PacketSize; use nymsphinx::{addressing::nodes::NymNodeRoutingAddress, SphinxPacket}; use std::collections::HashMap; use std::io; @@ -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( diff --git a/common/mixnode-common/Cargo.toml b/common/mixnode-common/Cargo.toml index 53676023af..d7b41990e3 100644 --- a/common/mixnode-common/Cargo.toml +++ b/common/mixnode-common/Cargo.toml @@ -16,6 +16,7 @@ serde = { version = "1.0", features = ["derive"] } tokio = { version = "1.21.2", features = ["time", "macros", "rt", "net", "io-util"] } tokio-util = { version = "0.7.3", features = ["codec"] } url = "2.2" +tracing = "0.1.37" crypto = { path = "../crypto" } nymsphinx-acknowledgements = { path = "../nymsphinx/acknowledgements" } diff --git a/common/mixnode-common/src/packet_processor/processor.rs b/common/mixnode-common/src/packet_processor/processor.rs index aaadfe0171..7a925f2def 100644 --- a/common/mixnode-common/src/packet_processor/processor.rs +++ b/common/mixnode-common/src/packet_processor/processor.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::packet_processor::error::MixProcessingError; -use log::*; +use tracing::*; use nymsphinx_acknowledgements::surb_ack::SurbAck; use nymsphinx_addressing::nodes::NymNodeRoutingAddress; use nymsphinx_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, @@ -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, @@ -116,10 +120,7 @@ impl SphinxPacketProcessor { trace!("received an ack packet!"); Ok((None, data)) } - PacketSize::RegularPacket - | PacketSize::ExtendedPacket8 - | PacketSize::ExtendedPacket16 - | PacketSize::ExtendedPacket32 => { + _ => { trace!("received a normal packet!"); let (ack_data, message) = self.split_hop_data_into_ack_and_message(data)?; let (ack_first_hop, ack_packet) = SurbAck::try_recover_first_hop_packet(&ack_data)?; @@ -132,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, @@ -153,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, @@ -170,7 +173,7 @@ impl SphinxPacketProcessor { } } } - + #[instrument(level="debug", skip_all, fields(packet_size=?received.packet_size()))] pub fn process_received( &self, received: FramedSphinxPacket, diff --git a/common/nymsphinx/params/src/packet_sizes.rs b/common/nymsphinx/params/src/packet_sizes.rs index aa573ffe11..3b2bbf2754 100644 --- a/common/nymsphinx/params/src/packet_sizes.rs +++ b/common/nymsphinx/params/src/packet_sizes.rs @@ -20,6 +20,17 @@ const EXTENDED_PACKET_SIZE_8: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 8 * const EXTENDED_PACKET_SIZE_16: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 16 * 1024; const EXTENDED_PACKET_SIZE_32: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 32 * 1024; +const EXTENDED_PACKET_SIZE_10: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 10 * 1024; +const EXTENDED_PACKET_SIZE_15: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 15 * 1024; +const EXTENDED_PACKET_SIZE_20: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 20 * 1024; +const EXTENDED_PACKET_SIZE_25: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 25 * 1024; +const EXTENDED_PACKET_SIZE_50: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 50 * 1024; +const EXTENDED_PACKET_SIZE_100: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 100 * 1024; +const EXTENDED_PACKET_SIZE_150: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 150 * 1024; +const EXTENDED_PACKET_SIZE_200: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 200 * 1024; +const EXTENDED_PACKET_SIZE_250: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 250 * 1024; +const EXTENDED_PACKET_SIZE_500: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 500 * 1024; + #[derive(Debug)] pub struct InvalidPacketSize; @@ -43,6 +54,18 @@ pub enum PacketSize { // for example for streaming fast and furious in compressed XviD quality ExtendedPacket16 = 5, + + ExtendedPacket10 = 6, + ExtendedPacket15 = 7, + ExtendedPacket20 = 8, + + ExtendedPacket25 = 9, + ExtendedPacket50 = 10, + ExtendedPacket100 = 11, + ExtendedPacket150 = 12, + ExtendedPacket200 = 13, + ExtendedPacket250 = 14, + ExtendedPacket500 = 15, } impl FromStr for PacketSize { @@ -55,6 +78,16 @@ impl FromStr for PacketSize { "extended8" => Ok(Self::ExtendedPacket8), "extended16" => Ok(Self::ExtendedPacket16), "extended32" => Ok(Self::ExtendedPacket32), + "extended10" => Ok(Self::ExtendedPacket10), + "extended15" => Ok(Self::ExtendedPacket15), + "extended20" => Ok(Self::ExtendedPacket20), + "extended25" => Ok(Self::ExtendedPacket25), + "extended50" => Ok(Self::ExtendedPacket50), + "extended100" => Ok(Self::ExtendedPacket100), + "extended150" => Ok(Self::ExtendedPacket150), + "extended200" => Ok(Self::ExtendedPacket200), + "extended250" => Ok(Self::ExtendedPacket250), + "extended500" => Ok(Self::ExtendedPacket500), _ => Err(InvalidPacketSize), } } @@ -70,6 +103,16 @@ impl TryFrom for PacketSize { _ if value == (PacketSize::ExtendedPacket8 as u8) => Ok(Self::ExtendedPacket8), _ if value == (PacketSize::ExtendedPacket16 as u8) => Ok(Self::ExtendedPacket16), _ if value == (PacketSize::ExtendedPacket32 as u8) => Ok(Self::ExtendedPacket32), + _ if value == (PacketSize::ExtendedPacket10 as u8) => Ok(Self::ExtendedPacket10), + _ if value == (PacketSize::ExtendedPacket15 as u8) => Ok(Self::ExtendedPacket15), + _ if value == (PacketSize::ExtendedPacket20 as u8) => Ok(Self::ExtendedPacket20), + _ if value == (PacketSize::ExtendedPacket25 as u8) => Ok(Self::ExtendedPacket25), + _ if value == (PacketSize::ExtendedPacket50 as u8) => Ok(Self::ExtendedPacket50), + _ if value == (PacketSize::ExtendedPacket100 as u8) => Ok(Self::ExtendedPacket100), + _ if value == (PacketSize::ExtendedPacket150 as u8) => Ok(Self::ExtendedPacket150), + _ if value == (PacketSize::ExtendedPacket200 as u8) => Ok(Self::ExtendedPacket200), + _ if value == (PacketSize::ExtendedPacket250 as u8) => Ok(Self::ExtendedPacket250), + _ if value == (PacketSize::ExtendedPacket500 as u8) => Ok(Self::ExtendedPacket500), _ => Err(InvalidPacketSize), } } @@ -83,6 +126,16 @@ impl PacketSize { PacketSize::ExtendedPacket8 => EXTENDED_PACKET_SIZE_8, PacketSize::ExtendedPacket16 => EXTENDED_PACKET_SIZE_16, PacketSize::ExtendedPacket32 => EXTENDED_PACKET_SIZE_32, + PacketSize::ExtendedPacket10 => EXTENDED_PACKET_SIZE_10, + PacketSize::ExtendedPacket15 => EXTENDED_PACKET_SIZE_15, + PacketSize::ExtendedPacket20 => EXTENDED_PACKET_SIZE_20, + PacketSize::ExtendedPacket25 => EXTENDED_PACKET_SIZE_25, + PacketSize::ExtendedPacket50 => EXTENDED_PACKET_SIZE_50, + PacketSize::ExtendedPacket100 => EXTENDED_PACKET_SIZE_100, + PacketSize::ExtendedPacket150 => EXTENDED_PACKET_SIZE_150, + PacketSize::ExtendedPacket200 => EXTENDED_PACKET_SIZE_200, + PacketSize::ExtendedPacket250 => EXTENDED_PACKET_SIZE_250, + PacketSize::ExtendedPacket500 => EXTENDED_PACKET_SIZE_500, } } @@ -105,6 +158,26 @@ impl PacketSize { Ok(PacketSize::ExtendedPacket16) } else if PacketSize::ExtendedPacket32.size() == size { Ok(PacketSize::ExtendedPacket32) + } else if PacketSize::ExtendedPacket10.size() == size { + Ok(PacketSize::ExtendedPacket10) + } else if PacketSize::ExtendedPacket15.size() == size { + Ok(PacketSize::ExtendedPacket15) + } else if PacketSize::ExtendedPacket20.size() == size { + Ok(PacketSize::ExtendedPacket20) + } else if PacketSize::ExtendedPacket25.size() == size { + Ok(PacketSize::ExtendedPacket25) + } else if PacketSize::ExtendedPacket50.size() == size { + Ok(PacketSize::ExtendedPacket50) + } else if PacketSize::ExtendedPacket100.size() == size { + Ok(PacketSize::ExtendedPacket100) + } else if PacketSize::ExtendedPacket150.size() == size { + Ok(PacketSize::ExtendedPacket150) + } else if PacketSize::ExtendedPacket200.size() == size { + Ok(PacketSize::ExtendedPacket200) + } else if PacketSize::ExtendedPacket250.size() == size { + Ok(PacketSize::ExtendedPacket250) + } else if PacketSize::ExtendedPacket500.size() == size { + Ok(PacketSize::ExtendedPacket500) } else { Err(InvalidPacketSize) } @@ -115,7 +188,17 @@ impl PacketSize { PacketSize::RegularPacket | PacketSize::AckPacket => false, PacketSize::ExtendedPacket8 | PacketSize::ExtendedPacket16 - | PacketSize::ExtendedPacket32 => true, + | PacketSize::ExtendedPacket32 + | PacketSize::ExtendedPacket10 + | PacketSize::ExtendedPacket15 + | PacketSize::ExtendedPacket20 + | PacketSize::ExtendedPacket25 + | PacketSize::ExtendedPacket50 + | PacketSize::ExtendedPacket100 + | PacketSize::ExtendedPacket150 + | PacketSize::ExtendedPacket200 + | PacketSize::ExtendedPacket250 + | PacketSize::ExtendedPacket500 => true, } } @@ -133,3 +216,4 @@ impl Default for PacketSize { PacketSize::RegularPacket } } + diff --git a/common/nymsphinx/params/src/packet_version.rs b/common/nymsphinx/params/src/packet_version.rs index 001f46c84d..12f1666811 100644 --- a/common/nymsphinx/params/src/packet_version.rs +++ b/common/nymsphinx/params/src/packet_version.rs @@ -47,6 +47,16 @@ impl From for PacketVersion { n if n == PacketSize::ExtendedPacket8 as u8 => PacketVersion::Legacy, n if n == PacketSize::ExtendedPacket16 as u8 => PacketVersion::Legacy, n if n == PacketSize::ExtendedPacket32 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket10 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket15 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket20 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket25 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket50 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket100 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket150 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket200 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket250 as u8 => PacketVersion::Legacy, + n if n == PacketSize::ExtendedPacket500 as u8 => PacketVersion::Legacy, n => PacketVersion::Versioned(n), } } @@ -57,3 +67,4 @@ impl Default for PacketVersion { PacketVersion::Versioned(CURRENT_PACKET_VERSION_NUMBER) } } + diff --git a/mixnode/Cargo.toml b/mixnode/Cargo.toml index 776b7594f3..08714e3a04 100644 --- a/mixnode/Cargo.toml +++ b/mixnode/Cargo.toml @@ -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" ## internal config = { path="../common/config" } diff --git a/mixnode/src/main.rs b/mixnode/src/main.rs index ac0b99c1bf..f332b678bd 100644 --- a/mixnode/src/main.rs +++ b/mixnode/src/main.rs @@ -8,6 +8,9 @@ use ::config::defaults::setup_env; use clap::{crate_version, Parser}; use lazy_static::lazy_static; use logging::setup_logging; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::{EnvFilter, Registry, filter}; +use tracing_flame::FlameLayer; mod commands; mod config; @@ -35,12 +38,38 @@ struct 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"); + println!("{}", banner()); let args = Cli::parse(); setup_env(args.config_env_file.clone()); commands::execute(args).await; + opentelemetry::global::shutdown_tracer_provider(); } fn banner() -> String { diff --git a/mixnode/src/node/listener/connection_handler/mod.rs b/mixnode/src/node/listener/connection_handler/mod.rs index c9baaecacf..d8ab263279 100644 --- a/mixnode/src/node/listener/connection_handler/mod.rs +++ b/mixnode/src/node/listener/connection_handler/mod.rs @@ -7,8 +7,10 @@ use crate::node::listener::connection_handler::packet_processing::{ use crate::node::packet_delayforwarder::PacketDelayForwardSender; use crate::node::ShutdownListener; use futures::StreamExt; -use log::{error, info}; +use tracing::{error, info, trace, debug, warn}; +use tracing::*; use nymsphinx::forwarding::packet::MixPacket; +use nymsphinx::params::PacketSize; use nymsphinx::framing::codec::SphinxCodec; use nymsphinx::framing::packet::FramedSphinxPacket; use nymsphinx::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) { // 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, @@ -91,7 +96,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!( @@ -103,7 +110,7 @@ impl ConnectionHandler { } }, _ = shutdown.recv() => { - log::trace!("ConnectionHandler: received shutdown"); + trace!("ConnectionHandler: received shutdown"); } } } @@ -112,6 +119,6 @@ impl ConnectionHandler { "Closing connection from {:?}", framed_conn.into_inner().peer_addr() ); - log::trace!("ConnectionHandler: Exiting"); + trace!("ConnectionHandler: Exiting"); } } diff --git a/mixnode/src/node/listener/mod.rs b/mixnode/src/node/listener/mod.rs index a1542fba79..cb991bcf2a 100644 --- a/mixnode/src/node/listener/mod.rs +++ b/mixnode/src/node/listener/mod.rs @@ -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: ShutdownListener) -> 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) => { @@ -38,17 +39,17 @@ impl Listener { 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), } }, _ = self.shutdown.recv() => { - log::trace!("Listener: Received shutdown"); + trace!("Listener: Received shutdown"); } }; } - log::trace!("Listener: Exiting"); + trace!("Listener: Exiting"); } pub(crate) fn start(mut self, connection_handler: ConnectionHandler) -> JoinHandle<()> { diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 0bca9b31b0..6dbdf83407 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -19,7 +19,8 @@ use crate::node::node_statistics::SharedNodeStats; use crate::node::packet_delayforwarder::{DelayForwarder, PacketDelayForwardSender}; use ::crypto::asymmetric::{encryption, identity}; use config::NymConfig; -use log::{error, info, warn}; +use tracing::*; +use tracing::{info, error, warn}; use mixnode_common::verloc::{self, AtomicVerlocResult, VerlocMeasurer}; use rand::seq::SliceRandom; use rand::thread_rng; @@ -211,7 +212,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 } @@ -255,13 +256,13 @@ impl MixNode { atomic_verloc_results } - fn random_api_client(&self) -> validator_client::ApiClient { + fn random_api_client(&self) -> validator_client::client::ApiClient { let endpoints = self.config.get_validator_api_endpoints(); let validator_api = endpoints .choose(&mut thread_rng()) .expect("The list of validator apis is empty"); - validator_client::ApiClient::new(validator_api.clone()) + validator_client::client::ApiClient::new(validator_api.clone()) } // TODO: ask DH whether this function still makes sense in ^0.10 @@ -291,13 +292,13 @@ impl MixNode { async fn wait_for_interrupt(&self, mut shutdown: ShutdownNotifier) { wait_for_signal().await; - log::info!("Sending shutdown"); + info!("Sending shutdown"); shutdown.signal_shutdown().ok(); - log::info!("Waiting for tasks to finish... (Press ctrl-c to force)"); + info!("Waiting for tasks to finish... (Press ctrl-c to force)"); shutdown.wait_for_shutdown().await; - log::info!("Stopping nym mixnode"); + info!("Stopping nym mixnode"); } pub async fn run(&mut self) { @@ -307,7 +308,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 ); diff --git a/mixnode/src/node/packet_delayforwarder.rs b/mixnode/src/node/packet_delayforwarder.rs index 81e0303afd..74a1ebfb95 100644 --- a/mixnode/src/node/packet_delayforwarder.rs +++ b/mixnode/src/node/packet_delayforwarder.rs @@ -4,10 +4,13 @@ use crate::node::node_statistics::UpdateSender; use futures::channel::mpsc; use futures::StreamExt; +use nymsphinx::params::packet_sizes::PacketSize; use nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue}; use nymsphinx::forwarding::packet::MixPacket; use std::io; use tokio::time::Instant; +use tracing::*; +use tracing::{trace}; use super::ShutdownListener; @@ -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)) { // 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"); } }