From 27bf8b2e00e97dd270ae5fbfc94ba9d63a582d54 Mon Sep 17 00:00:00 2001 From: Simon Wicky Date: Tue, 31 Jan 2023 09:25:15 +0000 Subject: [PATCH] tracing first part --- .../real_traffic_stream.rs | 6 ++- common/client-libs/mixnet-client/Cargo.toml | 1 + .../mixnet-client/src/forwarder.rs | 2 +- common/nymsphinx/params/src/packet_sizes.rs | 12 ++++-- gateway/Cargo.toml | 6 +++ gateway/gateway-requests/Cargo.toml | 1 + .../src/registration/handshake/state.rs | 2 +- gateway/src/commands/run.rs | 8 ++-- gateway/src/main.rs | 41 +++++++++++++++++-- .../connection_handler/authenticated.rs | 23 +++++++++-- .../websocket/connection_handler/coconut.rs | 2 +- .../websocket/connection_handler/fresh.rs | 10 ++++- .../websocket/connection_handler/mod.rs | 2 +- .../client_handling/websocket/listener.rs | 2 +- .../receiver/connection_handler.rs | 5 +-- .../node/mixnet_handling/receiver/listener.rs | 4 +- gateway/src/node/mod.rs | 4 +- 17 files changed, 101 insertions(+), 30 deletions(-) diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index 9bf2415d43..c3f5f7069d 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -369,7 +369,7 @@ where fn poll_poisson(&mut self, cx: &mut Context<'_>) -> Poll> { // 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 @@ -544,6 +544,9 @@ where } pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::TaskClient) { + self.run_test().await; + return; + println!("START LINE"); debug!("Started OutQueueControl with graceful shutdown support"); #[cfg(not(target_arch = "wasm32"))] @@ -614,6 +617,7 @@ where } info!("10sec warmup done"); + return; sleep(Duration::new(10, 0)); info!("10 seconds cooldown elapsed"); info!("Resetting delay"); diff --git a/common/client-libs/mixnet-client/Cargo.toml b/common/client-libs/mixnet-client/Cargo.toml index 5267ecb366..4115243b0d 100644 --- a/common/client-libs/mixnet-client/Cargo.toml +++ b/common/client-libs/mixnet-client/Cargo.toml @@ -11,6 +11,7 @@ futures = "0.3" log = "0.4.8" tokio = { version = "1.21.2", features = ["time", "net", "rt"] } tokio-util = { version = "0.7.3", features = ["codec"] } +tracing = "0.1.37" # internal nymsphinx = {path = "../../nymsphinx" } diff --git a/common/client-libs/mixnet-client/src/forwarder.rs b/common/client-libs/mixnet-client/src/forwarder.rs index 1aa68be8da..e5215e95e8 100644 --- a/common/client-libs/mixnet-client/src/forwarder.rs +++ b/common/client-libs/mixnet-client/src/forwarder.rs @@ -4,7 +4,7 @@ use crate::client::{Client, Config, SendWithoutResponse}; use futures::channel::mpsc; use futures::StreamExt; -use log::*; +use tracing::*; use nymsphinx::forwarding::packet::MixPacket; use std::time::Duration; diff --git a/common/nymsphinx/params/src/packet_sizes.rs b/common/nymsphinx/params/src/packet_sizes.rs index f7264699c4..e87e7b695e 100644 --- a/common/nymsphinx/params/src/packet_sizes.rs +++ b/common/nymsphinx/params/src/packet_sizes.rs @@ -32,8 +32,10 @@ const EXTENDED_PACKET_SIZE_200: usize = HEADER_SIZE + PAYLOAD_OVERHEAD_SIZE + 20 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; +#[derive(Debug, Error)] +pub enum InvalidPacketSize { + #[error("{received} is not a valid packet size tag")] + UnknownPacketTag { received: u8 }, #[error("{received} is not a valid extended packet size variant")] UnknownExtendedPacketVariant { received: String }, @@ -93,7 +95,9 @@ impl FromStr for PacketSize { "extended200" => Ok(Self::ExtendedPacket200), "extended250" => Ok(Self::ExtendedPacket250), "extended500" => Ok(Self::ExtendedPacket500), - _ => Err(InvalidPacketSize), + s => Err(InvalidPacketSize::UnknownExtendedPacketVariant { + received: s.to_string(), + }), } } } @@ -118,7 +122,7 @@ impl TryFrom for PacketSize { _ 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), + v => Err(InvalidPacketSize::UnknownPacketTag { received: v }), } } } diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index a1761a7e17..cbd3f3df53 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -49,6 +49,12 @@ tokio-stream = { version = "0.1.9", features = ["fs"] } tokio-tungstenite = "0.14" tokio-util = { version = "0.7.3", 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 coconut-interface = { path = "../common/coconut-interface", optional = true } diff --git a/gateway/gateway-requests/Cargo.toml b/gateway/gateway-requests/Cargo.toml index 9501755306..5368836c29 100644 --- a/gateway/gateway-requests/Cargo.toml +++ b/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" crypto = { path = "../../common/crypto" } pemstore = { path = "../../common/pemstore" } diff --git a/gateway/gateway-requests/src/registration/handshake/state.rs b/gateway/gateway-requests/src/registration/handshake/state.rs index 1560b5fd9f..e121773c43 100644 --- a/gateway/gateway-requests/src/registration/handshake/state.rs +++ b/gateway/gateway-requests/src/registration/handshake/state.rs @@ -12,7 +12,7 @@ use crypto::{ symmetric::stream_cipher, }; use futures::{Sink, SinkExt, Stream, StreamExt}; -use log::*; +use tracing::*; use nymsphinx::params::{GatewayEncryptionAlgorithm, GatewaySharedKeyHkdfAlgorithm}; use rand::{CryptoRng, RngCore}; use std::convert::{TryFrom, TryInto}; diff --git a/gateway/src/commands/run.rs b/gateway/src/commands/run.rs index e91f260d1b..7cf4befb65 100644 --- a/gateway/src/commands/run.rs +++ b/gateway/src/commands/run.rs @@ -7,7 +7,7 @@ use crate::{ }; use clap::Args; use config::NymConfig; -use log::*; +use tracing::*; #[derive(Args, Clone)] pub struct Run { @@ -105,7 +105,7 @@ fn special_addresses() -> Vec<&'static str> { pub async fn execute(args: &Run) { println!("Starting gateway {}...", args.id); - + event!(Level::INFO, "Loading config"); let mut config = match Config::load_from_file(Some(&args.id)) { Ok(cfg) => cfg, Err(err) => { @@ -128,12 +128,12 @@ pub async fn execute(args: &Run) { 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; println!( "\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(); - gateway.run().await; + gateway.run().instrument(info_span!("Gateway run")).await; } diff --git a/gateway/src/main.rs b/gateway/src/main.rs index bfcfd0e563..aec53d1b60 100644 --- a/gateway/src/main.rs +++ b/gateway/src/main.rs @@ -6,6 +6,16 @@ use logging::setup_logging; use network_defaults::setup_env; use once_cell::sync::OnceCell; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::{EnvFilter, Registry, filter}; +use tracing::*; +use opentelemetry::trace::{TraceError}; +use tracing_flame::FlameLayer; +use std::time::Duration; +use std::thread::sleep; + + + mod commands; mod config; mod node; @@ -29,8 +39,30 @@ struct Cli { } #[tokio::main] -async fn main() { - setup_logging(); +async fn main() -> Result<(), TraceError>{ + let tracer = opentelemetry_jaeger::new_agent_pipeline() + .with_endpoint("143.42.21.138:6831") + .with_service_name("nym_gateway") + .install_simple() + .expect("Failed to initialize tracer"); + + let jaeger_layer = tracing_opentelemetry::layer().with_tracer(tracer); + let filter_layer = filter::filter_fn(|metadata| {metadata.target().starts_with("nym_gateway")}); + + let (flame_layer, _guard) = FlameLayer::with_file("./tracing.folded").unwrap(); + + let subscriber = Registry::default() + .with(EnvFilter::from_default_env()) + .with(filter_layer) + .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(); println!("{}", banner()); LONG_VERSION .set(long_version()) @@ -38,7 +70,10 @@ async fn main() { let args = Cli::parse(); setup_env(args.config_env_file.clone()); - commands::execute(args).await; + commands::execute(args).instrument( info_span!("Executing run command")).await; + opentelemetry::global::shutdown_tracer_provider(); + println!("Exiting"); + Ok(()) } fn banner() -> String { diff --git a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs index 40ca0d7715..f486b5b784 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs @@ -9,7 +9,8 @@ use futures::StreamExt; use gateway_requests::iv::IVConversionError; use gateway_requests::types::{BinaryRequest, ServerResponse}; use gateway_requests::{ClientControlRequest, GatewayRequestsError}; -use log::*; +use self::log; +use tracing::*; use nymsphinx::forwarding::packet::MixPacket; use rand::{CryptoRng, Rng}; use std::convert::TryFrom; @@ -180,6 +181,7 @@ where /// # Arguments /// /// * `mix_packet`: packet received from the client that should get forwarded into the network. + //SW#[instrument(level="info", 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."); @@ -328,13 +330,14 @@ where /// # Arguments /// /// * `mix_packet`: packet received from the client that should get forwarded into the network. + //SW#[instrument(level="info", skip_all)] async fn handle_forward_sphinx( &self, mix_packet: MixPacket, ) -> Result { 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( @@ -342,7 +345,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 { @@ -355,6 +358,7 @@ where /// # Arguments /// /// * `bin_msg`: raw message to handle. + //SW#[instrument(level="info", skip_all)] async fn handle_binary(&self, bin_msg: Vec) -> Message { // this function decrypts the request and checks the MAC match BinaryRequest::try_from_encrypted_tagged_bytes(bin_msg, &self.client.shared_keys) { @@ -376,6 +380,7 @@ where /// # Arguments /// /// * `raw_request`: raw message to handle. + //SW#[instrument(level="info", 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(), @@ -398,6 +403,7 @@ where /// # Arguments /// /// * `raw_request`: raw received websocket message. + //SW#[instrument(level="info", skip_all)] async fn handle_request(&mut self, raw_request: Message) -> Option { // 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 @@ -412,6 +418,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. + //SW#[instrument(level="info", skip_all, name="Serving requests")] pub(crate) async fn listen_for_requests(mut self) where S: AsyncRead + AsyncWrite + Unpin, @@ -422,11 +429,14 @@ where loop { tokio::select! { socket_msg = self.inner.read_websocket_message() => { + //let span = info_span!("Processing client request"); + //let guard = span.enter(); + 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; } }; @@ -436,6 +446,7 @@ where } if let Some(response) = self.handle_request(socket_msg).await { + //debug!(parent: &span, "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.", @@ -443,13 +454,17 @@ where break; } } + //drop(guard); }, 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); } } } diff --git a/gateway/src/node/client_handling/websocket/connection_handler/coconut.rs b/gateway/src/node/client_handling/websocket/connection_handler/coconut.rs index ba53fe4dbd..4cbd44e75a 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/coconut.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/coconut.rs @@ -3,7 +3,7 @@ use std::time::{Duration, SystemTime}; -use log::*; +use tracing::*; use coconut_interface::{Credential, VerificationKey}; use validator_client::{ diff --git a/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs b/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs index 9dd99eb23b..c5437b8ec0 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs @@ -19,7 +19,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 nymsphinx::DestinationAddressBytes; use rand::{CryptoRng, Rng}; @@ -124,6 +124,7 @@ where /// Attempts to perform websocket handshake with the remote and upgrades the raw TCP socket /// to the framed WebSocket. + //SW#[instrument(level="info", skip_all)] pub(crate) async fn perform_websocket_handshake(&mut self) -> Result<(), WsError> where S: AsyncRead + AsyncWrite + Unpin, @@ -147,6 +148,7 @@ where /// # Arguments /// /// * `init_msg`: a client handshake init message which should contain its identity public key as well as an ephemeral key. + //SW#[instrument(level="info", skip_all)] async fn perform_registration_handshake( &mut self, init_msg: Vec, @@ -170,6 +172,7 @@ where } /// Attempts to read websocket message from the associated socket. + ////SW#[instrument(level="info", skip_all)] pub(crate) async fn read_websocket_message(&mut self) -> Option> where S: AsyncRead + AsyncWrite + Unpin, @@ -185,6 +188,7 @@ where /// # Arguments /// /// * `msg`: WebSocket message to write back to the client. + //SW#[instrument(level="info", skip_all)] pub(crate) async fn send_websocket_message(&mut self, msg: Message) -> Result<(), WsError> where S: AsyncRead + AsyncWrite + Unpin, @@ -205,6 +209,7 @@ where /// /// * `shared_keys`: keys derived between the client and gateway. /// * `packets`: unwrapped packets that are to be pushed back to the client. + //SW#[instrument(level="info", skip_all)] pub(crate) async fn push_packets_to_client( &mut self, shared_keys: SharedKeys, @@ -215,6 +220,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> = packets .into_iter() .map(|received_message| { @@ -574,6 +580,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 + //SW#[instrument(level="info", skip_all)] pub(crate) async fn perform_initial_authentication( mut self, ) -> Option> @@ -610,6 +617,7 @@ where } } Ok(auth_result) => { + debug!("Confirming auth"); if let Err(err) = self .send_websocket_message(auth_result.server_response.into()) .await diff --git a/gateway/src/node/client_handling/websocket/connection_handler/mod.rs b/gateway/src/node/client_handling/websocket/connection_handler/mod.rs index 27b8366563..2ac4eb4c2f 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/mod.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/mod.rs @@ -3,7 +3,7 @@ use gateway_requests::registration::handshake::SharedKeys; use gateway_requests::ServerResponse; -use log::{trace, warn}; +use tracing::*; use nymsphinx::DestinationAddressBytes; use rand::{CryptoRng, Rng}; use tokio::io::{AsyncRead, AsyncWrite}; diff --git a/gateway/src/node/client_handling/websocket/listener.rs b/gateway/src/node/client_handling/websocket/listener.rs index 580ec61750..ee6c4e478b 100644 --- a/gateway/src/node/client_handling/websocket/listener.rs +++ b/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::FreshHandler; use crate::node::storage::Storage; use crypto::asymmetric::identity; -use log::*; +use tracing::*; use mixnet_client::forwarder::MixForwardingSender; use rand::rngs::OsRng; use std::net::SocketAddr; diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs index 5db66bd7cc..89d18eb35c 100644 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ b/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 nymsphinx::forwarding::packet::MixPacket; @@ -85,7 +85,6 @@ impl ConnectionHandler { } } } - fn try_push_message_to_client( &mut self, client_address: DestinationAddressBytes, @@ -104,7 +103,6 @@ impl ConnectionHandler { } } } - pub(crate) async fn store_processed_packet_payload( &self, client_address: DestinationAddressBytes, @@ -117,7 +115,6 @@ impl ConnectionHandler { self.storage.store_message(client_address, message).await } - fn forward_ack(&self, forward_ack: Option, client_address: DestinationAddressBytes) { if let Some(forward_ack) = forward_ack { trace!( diff --git a/gateway/src/node/mixnet_handling/receiver/listener.rs b/gateway/src/node/mixnet_handling/receiver/listener.rs index d301dc692b..f09f7cf534 100644 --- a/gateway/src/node/mixnet_handling/receiver/listener.rs +++ b/gateway/src/node/mixnet_handling/receiver/listener.rs @@ -3,7 +3,7 @@ use crate::node::mixnet_handling::receiver::connection_handler::ConnectionHandler; use crate::node::storage::Storage; -use log::*; +use tracing::*; use std::net::SocketAddr; use std::process; use tokio::task::JoinHandle; @@ -17,7 +17,7 @@ impl Listener { pub(crate) fn new(address: SocketAddr) -> Self { Listener { address } } - + //SW#[instrument(level="info", skip_all, name="Mixnet Listener")] pub(crate) async fn run(&mut self, connection_handler: ConnectionHandler) where St: Storage + Clone + 'static, diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 7500302787..0bca8de913 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -10,7 +10,7 @@ use crate::node::mixnet_handling::receiver::connection_handler::ConnectionHandle use crate::node::statistics::collector::GatewayStatisticsCollector; use crate::node::storage::Storage; use crypto::asymmetric::{encryption, identity}; -use log::*; +use tracing::*; use mixnet_client::forwarder::{MixForwardingSender, PacketForwarder}; #[cfg(feature = "coconut")] use network_defaults::NymNetworkDetails; @@ -213,7 +213,7 @@ where self.config.get_maximum_connection_buffer_size(), self.config.get_use_legacy_sphinx_framing(), ); - + //let span = info_span!("PacketForwarder Run"); tokio::spawn(async move { packet_forwarder.run().await }); packet_sender }