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

tracing first part

Browse Source
This commit is contained in:
Simon Wicky
2023-01-31 09:25:15 +00:00
parent e82a669cd3
commit 27bf8b2e00
17 changed files with 101 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+5 -1
View File
@@ -369,7 +369,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
@@ -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");
common/client-libs/mixnet-client/Cargo.toml
+1
View File
@@ -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" }
common/client-libs/mixnet-client/src/forwarder.rs
+1 -1
View File
@@ -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;
common/nymsphinx/params/src/packet_sizes.rs
+8 -4
View File
@@ -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<u8> 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 }),
}
}
}
gateway/Cargo.toml
+6
View File
@@ -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 }
gateway/gateway-requests/Cargo.toml
+1
View File
@@ -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" }
gateway/gateway-requests/src/registration/handshake/state.rs
+1 -1
View File
@@ -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};
gateway/src/commands/run.rs
+4 -4
View File
@@ -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;
}
gateway/src/main.rs
+38 -3
View File
@@ -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 {
gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs
+19 -4
View File
@@ -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<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(
@@ -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<u8>) -> 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<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
@@ -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);
}
}
}
gateway/src/node/client_handling/websocket/connection_handler/coconut.rs
+1 -1
View File
@@ -3,7 +3,7 @@
use std::time::{Duration, SystemTime};
use log::*;
use tracing::*;
use coconut_interface::{Credential, VerificationKey};
use validator_client::{
gateway/src/node/client_handling/websocket/connection_handler/fresh.rs
+9 -1
View File
@@ -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<u8>,
@@ -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<Result<Message, WsError>>
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<Result<Message, WsError>> = 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<AuthenticatedHandler<R, S, St>>
@@ -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
gateway/src/node/client_handling/websocket/connection_handler/mod.rs
+1 -1
View File
@@ -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};
gateway/src/node/client_handling/websocket/listener.rs
+1 -1
View File
@@ -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;
gateway/src/node/mixnet_handling/receiver/connection_handler.rs
+1 -4
View File
@@ -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<St: Storage> ConnectionHandler<St> {
}
}
}
fn try_push_message_to_client(
&mut self,
client_address: DestinationAddressBytes,
@@ -104,7 +103,6 @@ impl<St: Storage> ConnectionHandler<St> {
}
}
}
pub(crate) async fn store_processed_packet_payload(
&self,
client_address: DestinationAddressBytes,
@@ -117,7 +115,6 @@ impl<St: Storage> ConnectionHandler<St> {
self.storage.store_message(client_address, message).await
}
fn forward_ack(&self, forward_ack: Option<MixPacket>, client_address: DestinationAddressBytes) {
if let Some(forward_ack) = forward_ack {
trace!(
gateway/src/node/mixnet_handling/receiver/listener.rs
+2 -2
View File
@@ -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<St>(&mut self, connection_handler: ConnectionHandler<St>)
where
St: Storage + Clone + 'static,
gateway/src/node/mod.rs
+2 -2
View File
@@ -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
}