remove test client

clients/client-core/src/client/real_messages_control/real_traffic_stream.rs

@@ -333,7 +333,7 @@ where
     fn poll_poisson(&mut self, cx: &mut Context<'_>) -> Poll<Option<StreamMessage>> {
         // The average delay could change depending on if backpressure in the downstream channel
         // (mix_tx) was detected.
-        self.adjust_current_average_message_sending_delay();
+        //self.adjust_current_average_message_sending_delay();
         let avg_delay = self.current_average_message_sending_delay();
 
         // Start by checking if we have any incoming messages about closed connections

clients/client-core/src/config/mod.rs

@@ -697,6 +697,17 @@ pub enum ExtendedPacketSize {
     Extended8,
     Extended16,
     Extended32,
+    Extended10,
+    Extended15,
+    Extended20,
+    Extended25,
+    Extended50,
+    Extended100,
+    Extended150,
+    Extended200,
+    Extended250,
+    Extended500,
+
 }
 
 impl Default for DebugConfig {
@@ -734,6 +745,17 @@ impl From<ExtendedPacketSize> for PacketSize {
             ExtendedPacketSize::Extended8 => PacketSize::ExtendedPacket8,
             ExtendedPacketSize::Extended16 => PacketSize::ExtendedPacket16,
             ExtendedPacketSize::Extended32 => PacketSize::ExtendedPacket32,
+            ExtendedPacketSize::Extended10 => PacketSize::ExtendedPacket10,
+            ExtendedPacketSize::Extended15 => PacketSize::ExtendedPacket15,
+            ExtendedPacketSize::Extended20 => PacketSize::ExtendedPacket20,
+            ExtendedPacketSize::Extended25 => PacketSize::ExtendedPacket25,
+            ExtendedPacketSize::Extended50 => PacketSize::ExtendedPacket50,
+            ExtendedPacketSize::Extended100 => PacketSize::ExtendedPacket100,
+            ExtendedPacketSize::Extended150 => PacketSize::ExtendedPacket150,
+            ExtendedPacketSize::Extended200 => PacketSize::ExtendedPacket200,
+            ExtendedPacketSize::Extended250 => PacketSize::ExtendedPacket250,
+            ExtendedPacketSize::Extended500 => PacketSize::ExtendedPacket500,
         }
     }
 }
+

common/client-libs/gateway-client/src/packet_router.rs

@@ -74,7 +74,7 @@ impl PacketRouter {
                 received_messages.push(received_packet);
             } else {
                 // this can happen if other clients are not padding their messages
-                warn!("Received message of unexpected size. Probably from an outdated client... len: {}", received_packet.len());
+                //warn!("Received message of unexpected size. Probably from an outdated client... len: {}", received_packet.len());
                 received_messages.push(received_packet);
             }
         }

common/client-libs/mixnet-client/Cargo.toml

@@ -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

@@ -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/client-libs/mixnet-client/src/forwarder.rs

@@ -4,8 +4,8 @@
 use crate::client::{Client, Config, SendWithoutResponse};
 use futures::channel::mpsc;
 use futures::StreamExt;
-use log::*;
 use nym_sphinx::forwarding::packet::MixPacket;
+use tracing::*;
 use std::time::Duration;
 
 pub type MixForwardingSender = mpsc::UnboundedSender<MixPacket>;
@@ -53,10 +53,10 @@ impl PacketForwarder {
             tokio::select! {
                 biased;
                 _ = self.shutdown.recv() => {
-                    log::trace!("PacketForwarder: Received shutdown");
+                    trace!("PacketForwarder: Received shutdown");
                 }
                 Some(mix_packet) = self.packet_receiver.next() => {
-                     trace!("Going to forward packet to {:?}", mix_packet.next_hop());
+                    trace!("Going to forward packet to {:?}", mix_packet.next_hop());
 
                     let next_hop = mix_packet.next_hop();
                     let packet_mode = mix_packet.packet_mode();

common/mixnode-common/Cargo.toml

@@ -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

@@ -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>,
@@ -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,

common/nymsphinx/params/src/packet_sizes.rs

@@ -20,6 +20,16 @@ const ACK_PACKET_SIZE: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + ACK_IV_SIZE
 const EXTENDED_PACKET_SIZE_8: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 8 * 1024;
 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, Error)]
 pub enum InvalidPacketSize {
@@ -51,6 +61,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 {
@@ -63,6 +85,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),
             s => Err(InvalidPacketSize::UnknownExtendedPacketVariant {
                 received: s.to_string(),
             }),
@@ -80,6 +112,16 @@ impl TryFrom<u8> 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),
             v => Err(InvalidPacketSize::UnknownPacketTag { received: v }),
         }
     }
@@ -93,6 +135,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,
         }
     }
 
@@ -115,6 +167,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::UnknownPacketSize { received: size })
         }
@@ -125,7 +197,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,
         }
     }
 

common/nymsphinx/params/src/packet_version.rs

@@ -47,6 +47,16 @@ impl From<u8> 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),
         }
     }

gateway/Cargo.toml

@@ -50,6 +50,12 @@ tokio-stream = { version = "0.1.11", features = ["fs"] }
 tokio-tungstenite = "0.14"
 tokio-util = { version = "0.7.4", features = ["codec"] }
 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
 nym-coconut-interface = { path = "../common/coconut-interface" }

gateway/gateway-requests/Cargo.toml

@@ -18,6 +18,7 @@ rand = { version = "0.7.3", features = ["wasm-bindgen"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 thiserror = "1.0"
+tracing = "0.1.29"
 
 nym-crypto = { path = "../../common/crypto"  }
 nym-pemstore = { path = "../../common/pemstore" }

gateway/gateway-requests/src/registration/handshake/state.rs

@@ -6,7 +6,6 @@ use crate::registration::handshake::shared_key::{SharedKeySize, SharedKeys};
 use crate::registration::handshake::WsItem;
 use crate::types;
 use futures::{Sink, SinkExt, Stream, StreamExt};
-use log::*;
 use nym_crypto::{
     asymmetric::{encryption, identity},
     generic_array::typenum::Unsigned,
@@ -14,6 +13,7 @@ use nym_crypto::{
     symmetric::stream_cipher,
 };
 use nym_sphinx::params::{GatewayEncryptionAlgorithm, GatewaySharedKeyHkdfAlgorithm};
+use tracing::*;
 use rand::{CryptoRng, RngCore};
 use std::convert::{TryFrom, TryInto};
 use tungstenite::Message as WsMessage;

gateway/src/commands/run.rs

@@ -7,6 +7,8 @@ use crate::{
     OutputFormat,
 };
 use clap::Args;
+use config::NymConfig;
+use tracing::*;
 use std::error::Error;
 use std::net::IpAddr;
 use std::path::PathBuf;
@@ -115,19 +117,19 @@ fn special_addresses() -> Vec<&'static str> {
 pub async fn execute(args: Run, output: OutputFormat) -> Result<(), Box<dyn Error + Send + Sync>> {
     let id = args.id.clone();
     println!("Starting gateway {id}...");
-
+    event!(Level::INFO, "Loading config");
     let config = build_config(id, args)?;
     ensure_config_version_compatibility(&config)?;
 
     if special_addresses().contains(&&*config.get_listening_address().to_string()) {
         show_binding_warning(config.get_listening_address().to_string());
     }
-
+    event!(Level::INFO, "Creating Gateway node");
     let mut gateway = crate::node::create_gateway(config).await;
     eprintln!(
         "\nTo bond your gateway you will need to install the Nym wallet, go to https://nymtech.net/get-involved and select the Download button.\n\
          Select the correct version and install it to your machine. You will need to provide the following: \n ");
     gateway.print_node_details(output)?;
-
+//.instrument(info_span!("Gateway run"))
     gateway.run().await
 }

gateway/src/main.rs

@@ -4,11 +4,18 @@
 use clap::{crate_name, crate_version, Parser, ValueEnum};
 use colored::Colorize;
 use lazy_static::lazy_static;
-use log::error;
 use nym_bin_common::logging::setup_logging;
 use nym_bin_common::{build_information::BinaryBuildInformation, logging::banner};
 use nym_network_defaults::setup_env;
 use std::error::Error;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::{EnvFilter, Registry, filter};
+use tracing::*;
+use tracing_flame::FlameLayer;
+use std::time::Duration;
+use std::thread::sleep;
+
+
 
 mod commands;
 mod config;
@@ -64,14 +71,39 @@ impl Cli {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
-    setup_logging();
+    let tracer = opentelemetry_jaeger::new_agent_pipeline()
+        .with_endpoint("143.42.21.138:6831")
+        .with_service_name("nym_gateway")
+        .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");
+    //tracing_subscriber::fmt::init();
+    //setup_logging();
     if atty::is(atty::Stream::Stdout) {
         println!("{}", banner(crate_name!(), crate_version!()));
     }
 
     let args = Cli::parse();
     setup_env(args.config_env_file.as_ref());
-
+//.instrument(info_span!("Execute Run"))
     commands::execute(args).await.map_err(|err| {
         if atty::is(atty::Stream::Stdout) {
             let error_message = format!("{err}").red();
@@ -79,7 +111,11 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
             error!("Exiting...");
         }
         err
-    })
+    });
+    //sleep(Duration::from_secs(5));
+    //opentelemetry::global::shutdown_tracer_provider();
+    Ok(())
+
 }
 
 #[cfg(test)]

gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs

@@ -9,8 +9,8 @@ use futures::StreamExt;
 use gateway_requests::iv::IVConversionError;
 use gateway_requests::types::{BinaryRequest, ServerResponse};
 use gateway_requests::{ClientControlRequest, GatewayRequestsError};
-use log::*;
 use nym_sphinx::forwarding::packet::MixPacket;
+use tracing::*;
 use rand::{CryptoRng, Rng};
 use std::convert::TryFrom;
 use std::process;
@@ -132,6 +132,7 @@ where
     }
 
     /// Explicitly removes handle from the global store.
+    #[instrument(level="debug", skip_all)]
     fn disconnect(self) {
         self.inner
             .active_clients_store
@@ -180,6 +181,7 @@ where
     /// # Arguments
     ///
     /// * `mix_packet`: packet received from the client that should get forwarded into the network.
+    #[instrument(level="debug", skip_all)]
     fn forward_packet(&self, mix_packet: MixPacket) {
         if let Err(err) = self.inner.outbound_mix_sender.unbounded_send(mix_packet) {
             error!("We failed to forward requested mix packet - {err}. Presumably our mix forwarder has crashed. We cannot continue.");
@@ -279,13 +281,14 @@ where
     /// # Arguments
     ///
     /// * `mix_packet`: packet received from the client that should get forwarded into the network.
+    #[instrument(level="debug", skip_all)]
     async fn handle_forward_sphinx(
         &self,
         mix_packet: MixPacket,
     ) -> Result<ServerResponse, RequestHandlingError> {
         let consumed_bandwidth = mix_packet.sphinx_packet().len() as i64;
 
-        let available_bandwidth = self.get_available_bandwidth().await?;
+        let available_bandwidth = self.get_available_bandwidth().instrument(trace_span!("Get available bandwidth")).await?;
 
         if available_bandwidth < consumed_bandwidth {
             return Ok(ServerResponse::new_error(
@@ -293,7 +296,7 @@ where
             ));
         }
 
-        self.consume_bandwidth(consumed_bandwidth).await?;
+        self.consume_bandwidth(consumed_bandwidth).instrument(trace_span!("Consume bandwidth")).await?;
         self.forward_packet(mix_packet);
 
         Ok(ServerResponse::Send {
@@ -306,6 +309,7 @@ where
     /// # Arguments
     ///
     /// * `bin_msg`: raw message to handle.
+    #[instrument(level="debug", skip_all)]
     async fn handle_binary(&self, bin_msg: Vec<u8>) -> Message {
         // this function decrypts the request and checks the MAC
         match BinaryRequest::try_from_encrypted_tagged_bytes(bin_msg, &self.client.shared_keys) {
@@ -327,6 +331,7 @@ where
     /// # Arguments
     ///
     /// * `raw_request`: raw message to handle.
+    #[instrument(level="debug", skip_all)]
     async fn handle_text(&mut self, raw_request: String) -> Message {
         match ClientControlRequest::try_from(raw_request) {
             Err(e) => RequestHandlingError::InvalidTextRequest(e).into_error_message(),
@@ -349,6 +354,7 @@ where
     /// # Arguments
     ///
     /// * `raw_request`: raw received websocket message.
+    //#[instrument(level="info", skip_all)]
     async fn handle_request(&mut self, raw_request: Message) -> Option<Message> {
         // apparently tungstenite auto-handles ping/pong/close messages so for now let's ignore
         // them and let's test that claim. If that's not the case, just copy code from
@@ -363,6 +369,7 @@ where
     /// Simultaneously listens for incoming client requests, which realistically should only be
     /// binary requests to forward sphinx packets or increase bandwidth
     /// and for sphinx packets received from the mix network that should be sent back to the client.
+    #[instrument(level="info", skip_all, name="Serving requests")]
     pub(crate) async fn listen_for_requests(mut self, mut shutdown: TaskClient)
     where
         S: AsyncRead + AsyncWrite + Unpin,
@@ -373,14 +380,16 @@ where
         while !shutdown.is_shutdown() {
             tokio::select! {
                 _ = shutdown.recv() => {
-                    log::trace!("client_handling::AuthenticatedHandler: received shutdown");
+                    trace!("client_handling::AuthenticatedHandler: received shutdown");
                 }
                 socket_msg = self.inner.read_websocket_message() => {
+
+                    debug!("Handling client request");
                     let socket_msg = match socket_msg {
                         None => break,
                         Some(Ok(socket_msg)) => socket_msg,
                         Some(Err(err)) => {
-                            error!("failed to obtain message from websocket stream! stopping connection handler: {err}");
+                            log::error!("failed to obtain message from websocket stream! stopping connection handler: {err}");
                             break;
                         }
                     };
@@ -390,6 +399,7 @@ where
                     }
 
                     if let Some(response) = self.handle_request(socket_msg).await {
+                        trace!("Sending response to client request");
                         if let Err(err) = self.inner.send_websocket_message(response).await {
                             warn!(
                                 "Failed to send message over websocket: {err}. Assuming the connection is dead.",
@@ -397,13 +407,17 @@ where
                             break;
                         }
                     }
+
                 },
                 mix_messages = self.mix_receiver.next() => {
+                    //let span = info_span!("Processing mixnet message");
+                    //let guard = span.enter();
                     let mix_messages = mix_messages.expect("sender was unexpectedly closed! this shouldn't have ever happened!");
                     if let Err(err) = self.inner.push_packets_to_client(self.client.shared_keys, mix_messages).await {
                         warn!("failed to send the unwrapped sphinx packets back to the client - {err}, assuming the connection is dead");
                         break;
                     }
+                    //drop(guard);
                 }
             }
         }

gateway/src/node/client_handling/websocket/connection_handler/coconut.rs

@@ -1,8 +1,9 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use log::*;
 use nym_coconut_interface::Credential;
+use coconut_interface::Credential;
+use tracing::*;
 use std::time::{Duration, SystemTime};
 use validator_client::nyxd::traits::DkgQueryClient;
 use validator_client::{

gateway/src/node/client_handling/websocket/connection_handler/fresh.rs

@@ -17,7 +17,7 @@ use gateway_requests::registration::handshake::error::HandshakeError;
 use gateway_requests::registration::handshake::{gateway_handshake, SharedKeys};
 use gateway_requests::types::{ClientControlRequest, ServerResponse};
 use gateway_requests::{BinaryResponse, PROTOCOL_VERSION};
-use log::*;
+use tracing::*;
 use mixnet_client::forwarder::MixForwardingSender;
 use nym_crypto::asymmetric::identity;
 use nym_sphinx::DestinationAddressBytes;
@@ -111,6 +111,7 @@ where
 
     /// Attempts to perform websocket handshake with the remote and upgrades the raw TCP socket
     /// to the framed WebSocket.
+    #[instrument(level="debug", skip_all)]
     pub(crate) async fn perform_websocket_handshake(&mut self) -> Result<(), WsError>
     where
         S: AsyncRead + AsyncWrite + Unpin,
@@ -134,6 +135,7 @@ where
     /// # Arguments
     ///
     /// * `init_msg`: a client handshake init message which should contain its identity public key as well as an ephemeral key.
+    #[instrument(level="debug", skip_all)]
     async fn perform_registration_handshake(
         &mut self,
         init_msg: Vec<u8>,
@@ -157,6 +159,7 @@ where
     }
 
     /// Attempts to read websocket message from the associated socket.
+    #[instrument(level="debug", skip_all)]
     pub(crate) async fn read_websocket_message(&mut self) -> Option<Result<Message, WsError>>
     where
         S: AsyncRead + AsyncWrite + Unpin,
@@ -172,6 +175,7 @@ where
     /// # Arguments
     ///
     /// * `msg`: WebSocket message to write back to the client.
+    #[instrument(level="debug", skip_all)]
     pub(crate) async fn send_websocket_message(&mut self, msg: Message) -> Result<(), WsError>
     where
         S: AsyncRead + AsyncWrite + Unpin,
@@ -192,6 +196,7 @@ where
     ///
     /// * `shared_keys`: keys derived between the client and gateway.
     /// * `packets`: unwrapped packets that are to be pushed back to the client.
+    #[instrument(level="debug", skip_all, fields(packets = packets.len()))]
     pub(crate) async fn push_packets_to_client(
         &mut self,
         shared_keys: SharedKeys,
@@ -202,6 +207,7 @@ where
     {
         // note: into_ws_message encrypts the requests and adds a MAC on it. Perhaps it should
         // be more explicit in the naming?
+        debug!("Pushing {} packets", packets.len());
         let messages: Vec<Result<Message, WsError>> = packets
             .into_iter()
             .map(|received_message| {
@@ -561,6 +567,7 @@ where
     /// result in client getting registered or authenticated. All other requests, such as forwarding
     /// sphinx packets considered an error and terminate the connection.
     // TODO: somehow cleanup this method
+    #[instrument(level="debug", skip_all)]
     pub(crate) async fn perform_initial_authentication(
         mut self,
     ) -> Option<AuthenticatedHandler<R, S, St>>
@@ -597,6 +604,7 @@ where
                             }
                         }
                         Ok(auth_result) => {
+                            debug!("Confirming auth");
                             if let Err(err) = self
                                 .send_websocket_message(auth_result.server_response.into())
                                 .await

gateway/src/node/client_handling/websocket/connection_handler/mod.rs

@@ -4,9 +4,9 @@
 use crate::node::storage::Storage;
 use gateway_requests::registration::handshake::SharedKeys;
 use gateway_requests::ServerResponse;
-use log::{trace, warn};
 use nym_sphinx::DestinationAddressBytes;
 use nym_task::TaskClient;
+use tracing::*;
 use rand::{CryptoRng, Rng};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_tungstenite::WebSocketStream;

gateway/src/node/client_handling/websocket/listener.rs

@@ -5,7 +5,7 @@ use crate::node::client_handling::active_clients::ActiveClientsStore;
 use crate::node::client_handling::websocket::connection_handler::coconut::CoconutVerifier;
 use crate::node::client_handling::websocket::connection_handler::FreshHandler;
 use crate::node::storage::Storage;
-use log::*;
+use tracing::*;
 use mixnet_client::forwarder::MixForwardingSender;
 use nym_crypto::asymmetric::identity;
 use rand::rngs::OsRng;
@@ -60,7 +60,7 @@ impl Listener {
             tokio::select! {
                 biased;
                 _ = shutdown.recv() => {
-                    log::trace!("client_handling::Listener: received shutdown");
+                    trace!("client_handling::Listener: received shutdown");
                 }
                 connection = tcp_listener.accept() => {
                     match connection {
@@ -79,7 +79,7 @@ impl Listener {
                                 Arc::clone(&self.coconut_verifier),
                             );
                             let shutdown = shutdown.clone();
-                            tokio::spawn(async move { handle.start_handling(shutdown).await });
+                            tokio::spawn(async move { handle.start_handling(shutdown).await }.instrument(info_span!("Connection handling", address = %remote_addr)));
                         }
                         Err(err) => warn!("failed to get client: {err}"),
                     }
@@ -102,6 +102,6 @@ impl Listener {
         tokio::spawn(async move {
             self.run(outbound_mix_sender, storage, active_clients_store, shutdown)
                 .await
-        })
+        }.instrument(info_span!("Client Listener")))
     }
 }

gateway/src/node/mixnet_handling/receiver/connection_handler.rs

@@ -7,7 +7,7 @@ use crate::node::mixnet_handling::receiver::packet_processing::PacketProcessor;
 use crate::node::storage::error::StorageError;
 use crate::node::storage::Storage;
 use futures::StreamExt;
-use log::*;
+use tracing::*;
 use mixnet_client::forwarder::MixForwardingSender;
 use mixnode_common::packet_processor::processor::ProcessedFinalHop;
 use nym_sphinx::forwarding::packet::MixPacket;
@@ -86,7 +86,7 @@ impl<St: Storage> ConnectionHandler<St> {
             }
         }
     }
-
+    #[instrument(level="debug", skip_all)]
     fn try_push_message_to_client(
         &mut self,
         client_address: DestinationAddressBytes,
@@ -105,7 +105,7 @@ impl<St: Storage> ConnectionHandler<St> {
             }
         }
     }
-
+    #[instrument(level="debug", skip_all)]
     pub(crate) async fn store_processed_packet_payload(
         &self,
         client_address: DestinationAddressBytes,
@@ -118,7 +118,7 @@ impl<St: Storage> ConnectionHandler<St> {
 
         self.storage.store_message(client_address, message).await
     }
-
+    #[instrument(level="debug", skip_all)]
     fn forward_ack(&self, forward_ack: Option<MixPacket>, client_address: DestinationAddressBytes) {
         if let Some(forward_ack) = forward_ack {
             trace!(
@@ -130,7 +130,7 @@ impl<St: Storage> ConnectionHandler<St> {
             self.ack_sender.unbounded_send(forward_ack).unwrap();
         }
     }
-
+    #[instrument(level="debug", skip_all)]
     async fn handle_processed_packet(&mut self, processed_final_hop: ProcessedFinalHop) {
         let client_address = processed_final_hop.destination;
         let message = processed_final_hop.message;
@@ -154,7 +154,7 @@ impl<St: Storage> ConnectionHandler<St> {
         // received ack back into the network
         self.forward_ack(forward_ack, client_address);
     }
-
+    //#[instrument(level="info", skip_all)]
     async fn handle_received_packet(&mut self, framed_sphinx_packet: FramedSphinxPacket) {
         //
         // TODO: here be replay attack detection - it will require similar key cache to the one in
@@ -187,7 +187,7 @@ impl<St: Storage> ConnectionHandler<St> {
             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 {

gateway/src/node/mixnet_handling/receiver/listener.rs

@@ -3,8 +3,8 @@
 
 use crate::node::mixnet_handling::receiver::connection_handler::ConnectionHandler;
 use crate::node::storage::Storage;
-use log::*;
 use nym_task::TaskClient;
+use tracing::*;
 use std::net::SocketAddr;
 use std::process;
 use tokio::task::JoinHandle;
@@ -37,13 +37,14 @@ impl Listener {
             tokio::select! {
                 biased;
                 _ = self.shutdown.recv() => {
-                    log::trace!("mixnet_handling::Listener: Received shutdown");
+                    trace!("mixnet_handling::Listener: Received shutdown");
                 }
                 connection = tcp_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!("Mixnet connection handling", address = %remote_addr)));
                         }
                         Err(err) => warn!("failed to get client: {err}"),
                     }
@@ -58,6 +59,6 @@ impl Listener {
     {
         info!("Running mix listener on {:?}", self.address.to_string());
 
-        tokio::spawn(async move { self.run(connection_handler).await })
+        tokio::spawn(async move { self.run(connection_handler).await }.instrument(info_span!("Mixnet Listener")))
     }
 }

gateway/src/node/mod.rs

@@ -14,7 +14,7 @@ use crate::node::statistics::collector::GatewayStatisticsCollector;
 use crate::node::storage::Storage;
 use crate::{commands::sign::load_identity_keys, OutputFormat};
 use colored::Colorize;
-use log::*;
+use tracing::*;
 use mixnet_client::forwarder::{MixForwardingSender, PacketForwarder};
 use nym_crypto::asymmetric::{encryption, identity};
 use nym_network_defaults::NymNetworkDetails;
@@ -220,8 +220,7 @@ where
             self.config.get_use_legacy_sphinx_framing(),
             shutdown,
         );
-
+        tokio::spawn(async move { packet_forwarder.run().await }.instrument(info_span!("Packet Forwarder")));
-        tokio::spawn(async move { packet_forwarder.run().await });
         packet_sender
     }
 
@@ -230,7 +229,7 @@ where
         shutdown: TaskManager,
     ) -> Result<(), Box<dyn Error + Send + Sync>> {
         let res = shutdown.catch_interrupt().await;
-        log::info!("Stopping nym gateway");
+        info!("Stopping nym gateway");
         res
     }
 

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"
 atty = "0.2"
 
 ## internal

mixnode/src/main.rs

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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");
     }
 }
 

nym-api/src/epoch_operations/mod.rs

@@ -127,8 +127,8 @@ impl RewardedSetUpdater {
         }
 
         // Reward all the nodes in the still current, soon to be previous rewarded set
-        log::info!("Rewarding the current rewarded set...");
+        //log::info!("Rewarding the current rewarded set...");
-        self.reward_current_rewarded_set(interval).await?;
+        //self.reward_current_rewarded_set(interval).await?;
 
         // note: those operations don't really have to be atomic, so it's fine to send them
         // as separate transactions

nym-api/src/nym_contract_cache/routes.rs

@@ -68,7 +68,7 @@ pub async fn get_rewarded_set_detailed(
 #[openapi(tag = "contract-cache")]
 #[get("/mixnodes/active")]
 pub async fn get_active_set(cache: &State<NymContractCache>) -> Json<Vec<MixNodeDetails>> {
-    Json(cache.active_set().await.value)
+    Json(cache.mixnodes_filtered().await)
 }
 
 // DEPRECATED: this endpoint now lives in `node_status_api`. Once all consumers are updated,