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

instrumentation

Browse Source
This commit is contained in:
Simon Wicky
2023-03-28 13:24:45 +00:00
parent 921f01a789
commit 009b2131f6
12 changed files with 1538 additions and 981 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1476 -931
View File
File diff suppressed because it is too large Load Diff
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+1 -1
View File
@@ -604,7 +604,7 @@ where
let mut now = Instant::now();
while let Some(next_message) = self.next().await {
let packet = MixPacket::try_from_bytes(&dummy_packet.clone()).unwrap();
self.on_message(next_message, Some(packet)).await;
self.on_message(next_message, None).await;
if now.elapsed().as_secs() > 10 {
break;
}
common/client-libs/mixnet-client/src/client.rs
+6 -2
View File
@@ -3,10 +3,11 @@
use futures::channel::mpsc;
use futures::StreamExt;
use log::*;
use tracing::*;
use nymsphinx::framing::codec::SphinxCodec;
use nymsphinx::framing::packet::FramedSphinxPacket;
use nymsphinx::params::PacketMode;
use nymsphinx::params::packet_sizes::PacketSize;
use nymsphinx::{addressing::nodes::NymNodeRoutingAddress, SphinxPacket};
use std::collections::HashMap;
use std::io;
@@ -197,6 +198,7 @@ impl Client {
}
impl SendWithoutResponse for Client {
#[instrument(level="info", skip(self, packet), "Sending packet to mixnet", fields(packet_size))]
fn send_without_response(
&mut self,
address: NymNodeRoutingAddress,
@@ -204,13 +206,15 @@ impl SendWithoutResponse for Client {
packet_mode: PacketMode,
) -> io::Result<()> {
trace!("Sending packet to {:?}", address);
let packet_size = PacketSize::get_type(packet.len()).unwrap();
Span::current().record("packet_size", field::debug(packet_size));
let framed_packet =
FramedSphinxPacket::new(packet, packet_mode, self.config.use_legacy_version);
if let Some(sender) = self.conn_new.get_mut(&address) {
if let Err(err) = sender.channel.try_send(framed_packet) {
if err.is_full() {
debug!("Connection to {} seems to not be able to handle all the traffic - dropping the current packet", address);
info!("Connection to {} seems to not be able to handle all the traffic - dropping the current packet", address);
// it's not a 'big' error, but we did not manage to send the packet
// if the queue is full, we can't really do anything but to drop the packet
Err(io::Error::new(
common/client-libs/mixnet-client/src/forwarder.rs
+2 -2
View File
@@ -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();
gateway/src/commands/run.rs
+2 -2
View File
@@ -134,6 +134,6 @@ pub async fn execute(args: Run, output: OutputFormat) -> Result<(), Box<dyn Erro
"\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)?;
gateway.run().instrument(info_span!("Gateway run")).await;
//.instrument(info_span!("Gateway run"))
gateway.run().await
}
gateway/src/main.rs
+18 -13
View File
@@ -5,7 +5,6 @@ use build_information::BinaryBuildInformation;
use clap::{crate_name, crate_version, Parser, ValueEnum};
use colored::Colorize;
use lazy_static::lazy_static;
use log::error;
use logging::setup_logging;
use network_defaults::setup_env;
use std::error::Error;
@@ -13,7 +12,6 @@ use std::error::Error;
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;
@@ -77,33 +75,36 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
let tracer = opentelemetry_jaeger::new_agent_pipeline()
.with_endpoint("143.42.21.138:6831")
.with_service_name("nym_gateway")
.install_simple()
.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 filter_layer = filter::filter_fn(|metadata| {metadata.target().starts_with("nym_gateway")});
//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(filter_layer)
.with(hyper_filter)
.with(tokio_filter)
.with(tracing_subscriber::fmt::layer().pretty())
.with(flame_layer)
.with(jaeger_layer);
.with(flame_layer);
//.with(jaeger_layer)
tracing::subscriber::set_global_default(subscriber)
.expect("Failed to set global subscriber");
// tracing::subscriber::set_global_default(subscriber)
// .expect("Failed to set global subscriber");
//tracing_subscriber::fmt::init();
//setup_logging();
setup_logging();
if atty::is(atty::Stream::Stdout) {
println!("{}", logging::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();
@@ -111,7 +112,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
+11 -11
View File
@@ -138,6 +138,7 @@ where
}
/// Explicitly removes handle from the global store.
#[instrument(level="debug", skip_all)]
fn disconnect(self) {
self.inner
.active_clients_store
@@ -186,7 +187,7 @@ where
/// # Arguments
///
/// * `mix_packet`: packet received from the client that should get forwarded into the network.
//SW#[instrument(level="info", skip_all)]
#[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.");
@@ -351,7 +352,7 @@ where
/// # Arguments
///
/// * `mix_packet`: packet received from the client that should get forwarded into the network.
//SW#[instrument(level="info", skip_all)]
#[instrument(level="debug", skip_all)]
async fn handle_forward_sphinx(
&self,
mix_packet: MixPacket,
@@ -379,7 +380,7 @@ where
/// # Arguments
///
/// * `bin_msg`: raw message to handle.
//SW#[instrument(level="info", skip_all)]
#[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) {
@@ -401,7 +402,7 @@ where
/// # Arguments
///
/// * `raw_request`: raw message to handle.
//SW#[instrument(level="info", skip_all)]
#[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(),
@@ -424,7 +425,7 @@ where
/// # Arguments
///
/// * `raw_request`: raw received websocket message.
//SW#[instrument(level="info", skip_all)]
//#[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
@@ -439,7 +440,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")]
#[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,
@@ -450,11 +451,10 @@ 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() => {
//let span = info_span!("Processing client request");
//let guard = span.enter();
debug!("Handling client request");
let socket_msg = match socket_msg {
None => break,
@@ -470,7 +470,7 @@ where
}
if let Some(response) = self.handle_request(socket_msg).await {
//debug!(parent: &span, "Sending response to client request");
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.",
@@ -478,7 +478,7 @@ where
break;
}
}
//drop(guard);
},
mix_messages = self.mix_receiver.next() => {
//let span = info_span!("Processing mixnet message");
gateway/src/node/client_handling/websocket/connection_handler/fresh.rs
+6 -6
View File
@@ -124,7 +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)]
#[instrument(level="debug", skip_all)]
pub(crate) async fn perform_websocket_handshake(&mut self) -> Result<(), WsError>
where
S: AsyncRead + AsyncWrite + Unpin,
@@ -148,7 +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)]
#[instrument(level="debug", skip_all)]
async fn perform_registration_handshake(
&mut self,
init_msg: Vec<u8>,
@@ -172,7 +172,7 @@ where
}
/// Attempts to read websocket message from the associated socket.
////SW#[instrument(level="info", skip_all)]
#[instrument(level="debug", skip_all)]
pub(crate) async fn read_websocket_message(&mut self) -> Option<Result<Message, WsError>>
where
S: AsyncRead + AsyncWrite + Unpin,
@@ -188,7 +188,7 @@ where
/// # Arguments
///
/// * `msg`: WebSocket message to write back to the client.
//SW#[instrument(level="info", skip_all)]
#[instrument(level="debug", skip_all)]
pub(crate) async fn send_websocket_message(&mut self, msg: Message) -> Result<(), WsError>
where
S: AsyncRead + AsyncWrite + Unpin,
@@ -209,7 +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)]
#[instrument(level="debug", skip_all, fields(packets = packets.len()))]
pub(crate) async fn push_packets_to_client(
&mut self,
shared_keys: SharedKeys,
@@ -580,7 +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)]
#[instrument(level="debug", skip_all)]
pub(crate) async fn perform_initial_authentication(
mut self,
) -> Option<AuthenticatedHandler<R, S, St>>
gateway/src/node/client_handling/websocket/listener.rs
+3 -3
View File
@@ -65,7 +65,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 {
@@ -85,7 +85,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}"),
}
@@ -108,6 +108,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
+6 -3
View File
@@ -86,6 +86,7 @@ impl<St: Storage> ConnectionHandler<St> {
}
}
}
#[instrument(level="debug", skip_all)]
fn try_push_message_to_client(
&mut self,
client_address: DestinationAddressBytes,
@@ -104,6 +105,7 @@ impl<St: Storage> ConnectionHandler<St> {
}
}
}
#[instrument(level="debug", skip_all)]
pub(crate) async fn store_processed_packet_payload(
&self,
client_address: DestinationAddressBytes,
@@ -116,6 +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!(
@@ -127,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;
@@ -151,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
@@ -183,7 +186,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
+5 -4
View File
@@ -19,7 +19,7 @@ impl Listener {
pub(crate) fn new(address: SocketAddr, shutdown: TaskClient) -> Self {
Listener { address, shutdown }
}
//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,
@@ -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
+2 -3
View File
@@ -225,8 +225,7 @@ where
self.config.get_use_legacy_sphinx_framing(),
shutdown,
);
//let span = info_span!("PacketForwarder Run");
tokio::spawn(async move { packet_forwarder.run().await });
tokio::spawn(async move { packet_forwarder.run().await }.instrument(info_span!("Packet Forwarder")));
packet_sender
}
@@ -235,7 +234,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
}