liz/veilid
liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

simplify tracing

Browse Source
This commit is contained in:
John Smith 2023-06-24 22:59:51 -04:00
parent c8fdded5a7
commit 234f048241
38 changed files with 197 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
veilid-core/Cargo.toml
View File

@ -15,6 +15,7 @@ crypto-test = ["enable-crypto-vld0", "enable-crypto-none"]
crypto-test-none = ["enable-crypto-none"]
enable-crypto-vld0 = []
enable-crypto-none = []
verbose-tracing = []
rt-async-std = ["async-std", "async-std-resolver", "async_executors/async_std", "rtnetlink/smol_socket", "veilid-tools/rt-async-std"]
rt-tokio = ["tokio", "tokio-util", "tokio-stream", "trust-dns-resolver/tokio-runtime", "async_executors/tokio_tp", "async_executors/tokio_io", "async_executors/tokio_timer", "rtnetlink/tokio_socket", "veilid-tools/rt-tokio"]
rt-wasm-bindgen = ["veilid-tools/rt-wasm-bindgen", "async_executors/bindgen"]

4
veilid-core/src/network_manager/connection_handle.rs
View File

@ -34,14 +34,14 @@ impl ConnectionHandle {
self.descriptor.clone()
}
#[instrument(level="trace", skip(self, message), fields(message.len = message.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", skip(self, message), fields(message.len = message.len())))]
pub fn send(&self, message: Vec<u8>) -> ConnectionHandleSendResult {
match self.channel.send((Span::current().id(), message)) {
Ok(()) => ConnectionHandleSendResult::Sent,
Err(e) => ConnectionHandleSendResult::NotSent(e.0 .1),
}
}
#[instrument(level="trace", skip(self, message), fields(message.len = message.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", skip(self, message), fields(message.len = message.len())))]
pub async fn send_async(&self, message: Vec<u8>) -> ConnectionHandleSendResult {
match self
.channel

6
veilid-core/src/network_manager/mod.rs
View File

@ -775,7 +775,7 @@ impl NetworkManager {
}
/// Builds an envelope for sending over the network
#[instrument(level = "trace", skip(self, body), err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, body), err))]
fn build_envelope<B: AsRef<[u8]>>(
&self,
dest_node_id: TypedKey,
@ -806,7 +806,7 @@ impl NetworkManager {
/// node_ref is the direct destination to which the envelope will be sent
/// If 'destination_node_ref' is specified, it can be different than the node_ref being sent to
/// which will cause the envelope to be relayed
#[instrument(level = "trace", skip(self, body), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, body), ret, err))]
pub async fn send_envelope<B: AsRef<[u8]>>(
&self,
node_ref: NodeRef,
@ -872,7 +872,7 @@ impl NetworkManager {
// Called when a packet potentially containing an RPC envelope is received by a low-level
// network protocol handler. Processes the envelope, authenticates and decrypts the RPC message
// and passes it to the RPC handler
#[instrument(level = "trace", ret, err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", ret, err, skip(self, data), fields(data.len = data.len())))]
async fn on_recv_envelope(
&self,
data: &mut [u8],

8
veilid-core/src/network_manager/native/mod.rs
View File

@ -358,7 +358,7 @@ impl Network {
// This creates a short-lived connection in the case of connection-oriented protocols
// for the purpose of sending this one message.
// This bypasses the connection table as it is not a 'node to node' connection.
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_unbound_to_dial_info(
&self,
dial_info: DialInfo,
@ -416,7 +416,7 @@ impl Network {
// This creates a short-lived connection in the case of connection-oriented protocols
// for the purpose of sending this one message.
// This bypasses the connection table as it is not a 'node to node' connection.
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_recv_data_unbound_to_dial_info(
&self,
dial_info: DialInfo,
@ -496,7 +496,7 @@ impl Network {
}
}
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_to_existing_connection(
&self,
descriptor: ConnectionDescriptor,
@ -556,7 +556,7 @@ impl Network {
// Send data directly to a dial info, possibly without knowing which node it is going to
// Returns a descriptor for the connection used to send the data
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_to_dial_info(
&self,
dial_info: DialInfo,

13
veilid-core/src/network_manager/native/protocol/tcp.rs
View File

@ -56,11 +56,12 @@ impl RawTcpNetworkConnection {
stream.flush().await.into_network_result()
}
//#[instrument(level="trace", err, skip(self, message), fields(network_result, message.len = message.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, message), fields(network_result, message.len = message.len())))]
pub async fn send(&self, message: Vec<u8>) -> io::Result<NetworkResult<()>> {
let mut stream = self.stream.clone();
let out = Self::send_internal(&mut stream, message).await?;
//tracing::Span::current().record("network_result", &tracing::field::display(&out));
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("network_result", &tracing::field::display(&out));
Ok(out)
}
@ -87,11 +88,15 @@ impl RawTcpNetworkConnection {
Ok(NetworkResult::Value(out))
}
// #[instrument(level = "trace", err, skip(self), fields(network_result))]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", err, skip(self), fields(network_result))
)]
pub async fn recv(&self) -> io::Result<NetworkResult<Vec<u8>>> {
let mut stream = self.stream.clone();
let out = Self::recv_internal(&mut stream).await?;
//tracing::Span::current().record("network_result", &tracing::field::display(&out));
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("network_result", &tracing::field::display(&out));
Ok(out)
}
}

15
veilid-core/src/network_manager/native/protocol/udp.rs
View File

@ -15,7 +15,7 @@ impl RawUdpProtocolHandler {
}
}
// #[instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.len, ret.descriptor))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.len, ret.descriptor)))]
pub async fn recv_message(&self, data: &mut [u8]) -> io::Result<(usize, ConnectionDescriptor)> {
let (message_len, descriptor) = loop {
// Get a packet
@ -49,12 +49,14 @@ impl RawUdpProtocolHandler {
break (message.len(), descriptor);
};
// tracing::Span::current().record("ret.len", &message_len);
// tracing::Span::current().record("ret.descriptor", &format!("{:?}", descriptor).as_str());
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.len", &size);
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.descriptor", &format!("{:?}", descriptor).as_str());
Ok((message_len, descriptor))
}
//#[instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.descriptor))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, data), fields(data.len = data.len(), ret.len, ret.descriptor)))]
pub async fn send_message(
&self,
data: Vec<u8>,
@ -95,7 +97,10 @@ impl RawUdpProtocolHandler {
SocketAddress::from_socket_addr(local_socket_addr),
);
// tracing::Span::current().record("ret.descriptor", &format!("{:?}", descriptor).as_str());
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.len", &len);
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.descriptor", &format!("{:?}", descriptor).as_str());
Ok(NetworkResult::value(descriptor))
}

12
veilid-core/src/network_manager/native/protocol/ws.rs
View File

@ -72,7 +72,7 @@ where
// .map_err(to_io_error_other)
// }
//#[instrument(level = "trace", err, skip(self, message), fields(network_result, message.len = message.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, message), fields(network_result, message.len = message.len())))]
pub async fn send(&self, message: Vec<u8>) -> io::Result<NetworkResult<()>> {
if message.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("received too large WS message");
@ -82,6 +82,7 @@ where
Err(e) => err_to_network_result(e),
};
if !out.is_value() {
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("network_result", &tracing::field::display(&out));
return Ok(out);
}
@ -89,11 +90,13 @@ where
Ok(v) => NetworkResult::value(v),
Err(e) => err_to_network_result(e),
};
//tracing::Span::current().record("network_result", &tracing::field::display(&out));
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("network_result", &tracing::field::display(&out));
Ok(out)
}
// #[instrument(level = "trace", err, skip(self), fields(network_result, ret.len))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self), fields(network_result, ret.len)))]
pub async fn recv(&self) -> io::Result<NetworkResult<Vec<u8>>> {
let out = match self.stream.clone().next().await {
Some(Ok(Message::Binary(v))) => {
@ -120,7 +123,8 @@ where
)),
};
// tracing::Span::current().record("network_result", &tracing::field::display(&out));
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("network_result", &tracing::field::display(&out));
Ok(out)
}
}

5
veilid-core/src/network_manager/network_connection.rs
View File

@ -179,7 +179,7 @@ impl NetworkConnection {
}
}
#[instrument(level="trace", skip(message, stats), fields(message.len = message.len()), ret)]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", skip(message, stats), fields(message.len = message.len()), ret))]
async fn send_internal(
protocol_connection: &ProtocolNetworkConnection,
stats: Arc<Mutex<NetworkConnectionStats>>,
@ -194,7 +194,7 @@ impl NetworkConnection {
Ok(NetworkResult::Value(out))
}
#[instrument(level="trace", skip(stats), fields(ret.len))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", skip(stats), fields(ret.len)))]
async fn recv_internal(
protocol_connection: &ProtocolNetworkConnection,
stats: Arc<Mutex<NetworkConnectionStats>>,
@ -205,6 +205,7 @@ impl NetworkConnection {
let mut stats = stats.lock();
stats.last_message_recv_time.max_assign(Some(ts));
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.len", out.len());
Ok(NetworkResult::Value(out))

10
veilid-core/src/network_manager/send_data.rs
View File

@ -389,7 +389,10 @@ impl NetworkManager {
/// Send a reverse connection signal and wait for the return receipt over it
/// Then send the data across the new connection
/// Only usable for PublicInternet routing domain
#[instrument(level = "trace", skip(self, data), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, data), err)
)]
async fn do_reverse_connect(
&self,
relay_nr: NodeRef,
@ -475,7 +478,10 @@ impl NetworkManager {
/// Send a hole punch signal and do a negotiating ping and wait for the return receipt
/// Then send the data across the new connection
/// Only usable for PublicInternet routing domain
#[instrument(level = "trace", skip(self, data), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, data), err)
)]
async fn do_hole_punch(
&self,
relay_nr: NodeRef,

8
veilid-core/src/network_manager/wasm/mod.rs
View File

@ -79,7 +79,7 @@ impl Network {
/////////////////////////////////////////////////////////////////
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_unbound_to_dial_info(
&self,
dial_info: DialInfo,
@ -119,7 +119,7 @@ impl Network {
// This creates a short-lived connection in the case of connection-oriented protocols
// for the purpose of sending this one message.
// This bypasses the connection table as it is not a 'node to node' connection.
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_recv_data_unbound_to_dial_info(
&self,
dial_info: DialInfo,
@ -167,7 +167,7 @@ impl Network {
}
}
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_to_existing_connection(
&self,
descriptor: ConnectionDescriptor,
@ -212,7 +212,7 @@ impl Network {
Ok(Some(data))
}
#[instrument(level="trace", err, skip(self, data), fields(data.len = data.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level="trace", err, skip(self, data), fields(data.len = data.len())))]
pub async fn send_data_to_dial_info(
&self,
dial_info: DialInfo,

10
veilid-core/src/network_manager/wasm/protocol/ws.rs
View File

@ -64,7 +64,7 @@ impl WebsocketNetworkConnection {
// self.inner.ws_meta.close().await.map_err(to_io).map(drop)
// }
//#[instrument(level = "trace", err, skip(self, message), fields(network_result, message.len = message.len()))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self, message), fields(network_result, message.len = message.len())))]
pub async fn send(&self, message: Vec<u8>) -> io::Result<NetworkResult<()>> {
if message.len() > MAX_MESSAGE_SIZE {
bail_io_error_other!("sending too large WS message");
@ -79,11 +79,12 @@ impl WebsocketNetworkConnection {
.map_err(to_io)
.into_network_result()?;
//tracing::Span::current().record("network_result", &tracing::field::display(&out));
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("network_result", &tracing::field::display(&out));
Ok(out)
}
// #[instrument(level = "trace", err, skip(self), fields(network_result, ret.len))]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", err, skip(self), fields(network_result, ret.len)))]
pub async fn recv(&self) -> io::Result<NetworkResult<Vec<u8>>> {
let out = match SendWrapper::new(self.inner.ws_stream.clone().next()).await {
Some(WsMessage::Binary(v)) => {
@ -103,7 +104,8 @@ impl WebsocketNetworkConnection {
)));
}
};
// tracing::Span::current().record("network_result", &tracing::field::display(&out));
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("network_result", &tracing::field::display(&out));
Ok(out)
}
}

7
veilid-core/src/routing_table/mod.rs
View File

@ -518,7 +518,10 @@ impl RoutingTable {
}
/// Look up the best way for two nodes to reach each other over a specific routing domain
#[instrument(level = "trace", skip(self), ret)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), ret)
)]
pub fn get_contact_method(
&self,
routing_domain: RoutingDomain,
@ -1028,7 +1031,7 @@ impl RoutingTable {
.sort_and_clean_closest_noderefs(node_id, closest_nodes)
}
#[instrument(level = "trace", skip(self), ret)]
#[instrument(level = "trace", skip(self, peers), ret)]
pub fn register_find_node_answer(
&self,
crypto_kind: CryptoKind,

18
veilid-core/src/routing_table/route_spec_store/route_spec_store.rs
View File

@ -533,7 +533,7 @@ impl RouteSpecStore {
}
/// validate data using a private route's key and signature chain
#[instrument(level = "trace", skip(self, data, callback), ret)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, data, callback), ret))]
pub fn with_signature_validated_route<F,R>(
&self,
public_key: &TypedKey,
@ -593,7 +593,7 @@ impl RouteSpecStore {
Some(callback(rssd, rsd))
}
#[instrument(level = "trace", skip(self), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self), ret, err))]
async fn test_allocated_route(&self, private_route_id: RouteId) -> EyreResult<bool> {
// Make loopback route to test with
let dest = {
@ -716,7 +716,7 @@ impl RouteSpecStore {
}
/// Test an allocated route for continuity
#[instrument(level = "trace", skip(self), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self), ret, err))]
pub async fn test_route(&self, id: RouteId) -> EyreResult<bool> {
let is_remote = self.is_route_id_remote(&id);
if is_remote {
@ -1083,7 +1083,7 @@ impl RouteSpecStore {
}
/// Get an allocated route that matches a particular safety spec
#[instrument(level = "trace", skip(self, inner, rti), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, inner, rti), ret, err))]
fn get_route_for_safety_spec_inner(
&self,
inner: &mut RouteSpecStoreInner,
@ -1154,7 +1154,7 @@ impl RouteSpecStore {
}
/// Get a private route to use for the answer to question
#[instrument(level = "trace", skip(self), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self), ret, err))]
pub fn get_private_route_for_safety_spec(
&self,
crypto_kind: CryptoKind,
@ -1263,7 +1263,7 @@ impl RouteSpecStore {
/// Assemble a single private route for publication
/// Returns a PrivateRoute object for an allocated private route key
#[instrument(level = "trace", skip(self), err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self), err))]
pub fn assemble_private_route(
&self,
key: &PublicKey,
@ -1290,7 +1290,7 @@ impl RouteSpecStore {
/// Assemble private route set for publication
/// Returns a vec of PrivateRoute objects for an allocated private route
#[instrument(level = "trace", skip(self), err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self), err))]
pub fn assemble_private_routes(
&self,
id: &RouteId,
@ -1316,7 +1316,7 @@ impl RouteSpecStore {
/// Import a remote private route for compilation
/// It is safe to import the same route more than once and it will return the same route id
/// Returns a route set id
#[instrument(level = "trace", skip(self, blob), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, blob), ret, err))]
pub fn import_remote_private_route(&self, blob: Vec<u8>) -> EyreResult<RouteId> {
let cur_ts = get_aligned_timestamp();
@ -1347,7 +1347,7 @@ impl RouteSpecStore {
}
/// Release a remote private route that is no longer in use
#[instrument(level = "trace", skip(self), ret)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self), ret))]
pub fn release_remote_private_route(&self, id: RouteId) -> bool {
let inner = &mut *self.inner.lock();
inner.cache.remove_remote_private_route(id)

5
veilid-core/src/routing_table/routing_table_inner.rs
View File

@ -207,7 +207,10 @@ impl RoutingTableInner {
true
}
#[instrument(level = "trace", skip(self), ret)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), ret)
)]
pub fn get_contact_method(
&self,
routing_domain: RoutingDomain,

50
veilid-core/src/rpc_processor/mod.rs
View File

@ -525,7 +525,10 @@ impl RPCProcessor {
})
}
#[instrument(level = "trace", skip(self, waitable_reply), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, waitable_reply), err)
)]
async fn wait_for_reply(
&self,
waitable_reply: WaitableReply,
@ -662,7 +665,10 @@ impl RPCProcessor {
/// Produce a byte buffer that represents the wire encoding of the entire
/// unencrypted envelope body for a RPC message. This incorporates
/// wrapping a private and/or safety route if they are specified.
#[instrument(level = "debug", skip(self, operation), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "debug", skip(self, operation), err)
)]
fn render_operation(
&self,
dest: Destination,
@ -790,7 +796,10 @@ impl RPCProcessor {
/// routing table caching when it is okay to do so
/// Also check target's timestamp of our own node info, to see if we should send that
/// And send our timestamp of the target's node info so they can determine if they should update us on their next rpc
#[instrument(level = "trace", skip(self), ret)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), ret)
)]
fn get_sender_peer_info(&self, dest: &Destination) -> SenderPeerInfo {
// Don't do this if the sender is to remain private
// Otherwise we would be attaching the original sender's identity to the final destination,
@ -1091,7 +1100,10 @@ impl RPCProcessor {
/// Issue a question over the network, possibly using an anonymized route
/// Optionally keeps a context to be passed to the answer processor when an answer is received
#[instrument(level = "debug", skip(self, question), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "debug", skip(self, question), err)
)]
async fn question(
&self,
dest: Destination,
@ -1170,7 +1182,10 @@ impl RPCProcessor {
}
/// Issue a statement over the network, possibly using an anonymized route
#[instrument(level = "debug", skip(self, statement), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "debug", skip(self, statement), err)
)]
async fn statement(
&self,
dest: Destination,
@ -1227,7 +1242,10 @@ impl RPCProcessor {
}
/// Issue a reply over the network, possibly using an anonymized route
/// The request must want a response, or this routine fails
#[instrument(level = "debug", skip(self, request, answer), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "debug", skip(self, request, answer), err)
)]
async fn answer(
&self,
request: RPCMessage,
@ -1335,7 +1353,10 @@ impl RPCProcessor {
}
//////////////////////////////////////////////////////////////////////
#[instrument(level = "trace", skip(self, encoded_msg), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, encoded_msg), err)
)]
async fn process_rpc_message(
&self,
encoded_msg: RPCMessageEncoded,
@ -1505,7 +1526,10 @@ impl RPCProcessor {
}
}
#[instrument(level = "trace", skip(self, body), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, body), err)
)]
pub fn enqueue_direct_message(
&self,
envelope: Envelope,
@ -1538,7 +1562,10 @@ impl RPCProcessor {
Ok(())
}
#[instrument(level = "trace", skip(self, body), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, body), err)
)]
fn enqueue_safety_routed_message(
&self,
direct: RPCMessageHeaderDetailDirect,
@ -1569,7 +1596,10 @@ impl RPCProcessor {
Ok(())
}
#[instrument(level = "trace", skip(self, body), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, body), err)
)]
fn enqueue_private_routed_message(
&self,
direct: RPCMessageHeaderDetailDirect,

5
veilid-core/src/rpc_processor/operation_waiter.rs
View File

@ -114,7 +114,10 @@ where
}
/// Complete the app call
#[instrument(level = "trace", skip(self, message), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, message), err)
)]
pub async fn complete_op_waiter(&self, op_id: OperationId, message: T) -> Result<(), RPCError> {
let waiting_op = {
let mut inner = self.inner.lock();

9
veilid-core/src/rpc_processor/rpc_app_call.rs
View File

@ -3,7 +3,10 @@ use super::*;
impl RPCProcessor {
// Sends a high level app request and wait for response
// Can be sent via all methods including relays and routes
#[instrument(level = "trace", skip(self, message), fields(message.len = message.len(), ret.latency, ret.len), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, message), fields(message.len = message.len(), ret.latency, ret.len), err)
)]
pub async fn rpc_call_app_call(
self,
dest: Destination,
@ -38,12 +41,14 @@ impl RPCProcessor {
let a_message = app_call_a.destructure();
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.latency", latency.as_u64());
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.len", a_message.len());
Ok(NetworkResult::value(Answer::new(latency, a_message)))
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_app_call_q(
&self,
msg: RPCMessage,

7
veilid-core/src/rpc_processor/rpc_app_message.rs
View File

@ -3,7 +3,10 @@ use super::*;
impl RPCProcessor {
// Sends a high level app message
// Can be sent via all methods including relays and routes
#[instrument(level = "trace", skip(self, message), fields(message.len = message.len()), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, message), fields(message.len = message.len()), err)
)]
pub async fn rpc_call_app_message(
self,
dest: Destination,
@ -16,7 +19,7 @@ impl RPCProcessor {
self.statement(dest, statement).await
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_app_message(
&self,
msg: RPCMessage,

2
veilid-core/src/rpc_processor/rpc_cancel_tunnel.rs
View File

@ -1,7 +1,7 @@
use super::*;
impl RPCProcessor {
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_cancel_tunnel_q(
&self,
msg: RPCMessage,

2
veilid-core/src/rpc_processor/rpc_complete_tunnel.rs
View File

@ -1,7 +1,7 @@
use super::*;
impl RPCProcessor {
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_complete_tunnel_q(
&self,
msg: RPCMessage,

2
veilid-core/src/rpc_processor/rpc_find_block.rs
View File

@ -1,7 +1,7 @@
use super::*;
impl RPCProcessor {
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_find_block_q(
&self,
msg: RPCMessage,

7
veilid-core/src/rpc_processor/rpc_find_node.rs
View File

@ -7,7 +7,10 @@ impl RPCProcessor {
/// Because this leaks information about the identity of the node itself,
/// replying to this request received over a private route will leak
/// the identity of the node and defeat the private route.
#[instrument(level = "trace", skip(self), err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), err)
)]
pub async fn rpc_call_find_node(
self,
dest: Destination,
@ -67,7 +70,7 @@ impl RPCProcessor {
Ok(NetworkResult::value(Answer::new(latency, peers)))
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_find_node_q(
&self,
msg: RPCMessage,

23
veilid-core/src/rpc_processor/rpc_get_value.rs
View File

@ -15,13 +15,17 @@ impl RPCProcessor {
/// Because this leaks information about the identity of the node itself,
/// replying to this request received over a private route will leak
/// the identity of the node and defeat the private route.
#[instrument(level = "trace", skip(self, last_descriptor),
fields(ret.value.data.len,
ret.value.data.seq,
ret.value.data.writer,
ret.peers.len,
ret.latency
),err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, last_descriptor),
fields(ret.value.data.len,
ret.value.data.seq,
ret.value.data.writer,
ret.peers.len,
ret.latency
),err)
)]
pub async fn rpc_call_get_value(
self,
dest: Destination,
@ -107,12 +111,15 @@ impl RPCProcessor {
return Ok(NetworkResult::invalid_message("non-closer peers returned"));
}
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.latency", latency.as_u64());
#[cfg(feature = "verbose-tracing")]
if let Some(value) = &value {
tracing::Span::current().record("ret.value.data.len", value.value_data().data().len());
tracing::Span::current().record("ret.value.data.seq", value.value_data().seq());
tracing::Span::current().record("ret.value.data.writer", value.value_data().writer().to_string());
}
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.peers.len", peers.len());
Ok(NetworkResult::value(Answer::new(
@ -125,7 +132,7 @@ impl RPCProcessor {
)))
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_get_value_q(
&self,
msg: RPCMessage,

7
veilid-core/src/rpc_processor/rpc_return_receipt.rs
View File

@ -3,7 +3,10 @@ use super::*;
impl RPCProcessor {
// Sends a unidirectional in-band return receipt
// Can be sent via all methods including relays and routes
#[instrument(level = "trace", skip(self, receipt), ret, err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, receipt), ret, err)
)]
pub async fn rpc_call_return_receipt<D: AsRef<[u8]>>(
self,
dest: Destination,
@ -20,7 +23,7 @@ impl RPCProcessor {
Ok(NetworkResult::value(()))
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_return_receipt(
&self,
msg: RPCMessage,

14
veilid-core/src/rpc_processor/rpc_route.rs
View File

@ -1,7 +1,7 @@
use super::*;
impl RPCProcessor {
#[instrument(level = "trace", skip_all, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip_all, err))]
async fn process_route_safety_route_hop(
&self,
routed_operation: RoutedOperation,
@ -52,7 +52,7 @@ impl RPCProcessor {
.await
}
#[instrument(level = "trace", skip_all, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip_all, err))]
async fn process_route_private_route_hop(
&self,
routed_operation: RoutedOperation,
@ -99,7 +99,7 @@ impl RPCProcessor {
/// Note: it is important that we never respond with a safety route to questions that come
/// in without a private route. Giving away a safety route when the node id is known is
/// a privacy violation!
#[instrument(level = "trace", skip_all, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip_all, err))]
fn process_safety_routed_operation(
&self,
detail: RPCMessageHeaderDetailDirect,
@ -141,7 +141,7 @@ impl RPCProcessor {
}
/// Process a routed operation that came in over both a safety route and a private route
#[instrument(level = "trace", skip_all, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip_all, err))]
fn process_private_routed_operation(
&self,
detail: RPCMessageHeaderDetailDirect,
@ -209,7 +209,7 @@ impl RPCProcessor {
Ok(NetworkResult::value(()))
}
#[instrument(level = "trace", skip_all, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip_all, err))]
fn process_routed_operation(
&self,
detail: RPCMessageHeaderDetailDirect,
@ -239,7 +239,7 @@ impl RPCProcessor {
)
}
}
#[instrument(level = "trace", skip_all, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip_all, err))]
pub(crate) async fn process_private_route_first_hop(
&self,
mut routed_operation: RoutedOperation,
@ -360,7 +360,7 @@ impl RPCProcessor {
Ok(NetworkResult::value(route_hop))
}
#[instrument(level = "trace", skip(self, msg), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), ret, err))]
pub(crate) async fn process_route(
&self,
msg: RPCMessage,

13
veilid-core/src/rpc_processor/rpc_set_value.rs
View File

@ -14,7 +14,9 @@ impl RPCProcessor {
/// Because this leaks information about the identity of the node itself,
/// replying to this request received over a private route will leak
/// the identity of the node and defeat the private route.
#[instrument(level = "trace", skip(self, value, descriptor),
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self, value, descriptor),
fields(value.data.len = value.value_data().data().len(),
value.data.seq = value.value_data().seq(),
value.data.writer = value.value_data().writer().to_string(),
@ -24,7 +26,8 @@ impl RPCProcessor {
ret.value.data.writer,
ret.peers.len,
ret.latency
), err)]
), err)
)]
pub async fn rpc_call_set_value(
self,
dest: Destination,
@ -118,13 +121,17 @@ impl RPCProcessor {
return Ok(NetworkResult::invalid_message("non-closer peers returned"));
}
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.latency", latency.as_u64());
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.set", set);
#[cfg(feature = "verbose-tracing")]
if let Some(value) = &value {
tracing::Span::current().record("ret.value.data.len", value.value_data().data().len());
tracing::Span::current().record("ret.value.data.seq", value.value_data().seq());
tracing::Span::current().record("ret.value.data.writer", value.value_data().writer().to_string());
}
#[cfg(feature = "verbose-tracing")]
tracing::Span::current().record("ret.peers.len", peers.len());
Ok(NetworkResult::value(Answer::new(
@ -133,7 +140,7 @@ impl RPCProcessor {
)))
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_set_value_q(
&self,
msg: RPCMessage,

7
veilid-core/src/rpc_processor/rpc_signal.rs
View File

@ -3,7 +3,10 @@ use super::*;
impl RPCProcessor {
// Sends a unidirectional signal to a node
// Can be sent via relays but not routes. For routed 'signal' like capabilities, use AppMessage.
#[instrument(level = "trace", skip(self), ret, err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), ret, err)
)]
pub async fn rpc_call_signal(
self,
dest: Destination,
@ -29,7 +32,7 @@ impl RPCProcessor {
self.statement(dest, statement).await
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_signal(
&self,
msg: RPCMessage,

2
veilid-core/src/rpc_processor/rpc_start_tunnel.rs
View File

@ -1,7 +1,7 @@
use super::*;
impl RPCProcessor {
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_start_tunnel_q(
&self,
msg: RPCMessage,

7
veilid-core/src/rpc_processor/rpc_status.rs
View File

@ -15,7 +15,10 @@ impl RPCProcessor {
// direct -> node status + sender info
// safety -> node status
// private -> nothing
#[instrument(level = "trace", skip(self), ret, err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), ret, err)
)]
pub async fn rpc_call_status(
self,
dest: Destination,
@ -210,7 +213,7 @@ impl RPCProcessor {
Ok(NetworkResult::value(Answer::new(latency, opt_sender_info)))
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_status_q(
&self,
msg: RPCMessage,

2
veilid-core/src/rpc_processor/rpc_supply_block.rs
View File

@ -1,7 +1,7 @@
use super::*;
impl RPCProcessor {
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_supply_block_q(
&self,
msg: RPCMessage,

7
veilid-core/src/rpc_processor/rpc_validate_dial_info.rs
View File

@ -2,7 +2,10 @@ use super::*;
impl RPCProcessor {
// Can only be sent directly, not via relays or routes
#[instrument(level = "trace", skip(self), ret, err)]
#[cfg_attr(
feature = "verbose-tracing",
instrument(level = "trace", skip(self), ret, err)
)]
pub async fn rpc_call_validate_dial_info(
self,
peer: NodeRef,
@ -50,7 +53,7 @@ impl RPCProcessor {
}
}
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_validate_dial_info(
&self,
msg: RPCMessage,

2
veilid-core/src/rpc_processor/rpc_value_changed.rs
View File

@ -1,7 +1,7 @@
use super::*;
impl RPCProcessor {
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), err))]
pub(crate) async fn process_value_changed(
&self,
msg: RPCMessage,

2
veilid-core/src/rpc_processor/rpc_watch_value.rs
View File

@ -1,7 +1,7 @@
use super::*;
impl RPCProcessor {
#[instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err)]
#[cfg_attr(feature="verbose-tracing", instrument(level = "trace", skip(self, msg), fields(msg.operation.op_id), ret, err))]
pub(crate) async fn process_watch_value_q(
&self,
msg: RPCMessage,

1
veilid-core/src/storage_manager/mod.rs
View File

@ -95,6 +95,7 @@ impl StorageManager {
Ok(())
}
#[instrument(level = "debug", skip_all)]
pub async fn terminate(&self) {
debug!("starting storage manager shutdown");

11
veilid-core/src/veilid_api/api.rs
View File

@ -142,7 +142,6 @@ impl VeilidAPI {
}
/// Connect to the network
#[instrument(level = "debug", err, skip_all)]
pub async fn attach(&self) -> VeilidAPIResult<()> {
let attachment_manager = self.attachment_manager()?;
if !attachment_manager.attach().await {
@ -152,7 +151,6 @@ impl VeilidAPI {
}
/// Disconnect from the network
#[instrument(level = "debug", err, skip_all)]
pub async fn detach(&self) -> VeilidAPIResult<()> {
let attachment_manager = self.attachment_manager()?;
if !attachment_manager.detach().await {
@ -164,7 +162,6 @@ impl VeilidAPI {
////////////////////////////////////////////////////////////////
// Routing Context
#[instrument(level = "debug", skip(self))]
pub fn routing_context(&self) -> RoutingContext {
RoutingContext::new(self.clone())
}
@ -175,7 +172,6 @@ impl VeilidAPI {
/// Allocate a new private route set with default cryptography and network options
/// Returns a route id and a publishable 'blob' with the route encrypted with each crypto kind
/// Those nodes importing the blob will have their choice of which crypto kind to use
#[instrument(level = "debug", skip(self))]
pub async fn new_private_route(&self) -> VeilidAPIResult<(RouteId, Vec<u8>)> {
self.new_custom_private_route(
&VALID_CRYPTO_KINDS,
@ -186,7 +182,6 @@ impl VeilidAPI {
}
///
#[instrument(level = "debug", skip(self))]
pub async fn new_custom_private_route(
&self,
crypto_kinds: &[CryptoKind],
@ -238,14 +233,12 @@ impl VeilidAPI {
Ok((route_id, blob))
}
#[instrument(level = "debug", skip(self))]
pub fn import_remote_private_route(&self, blob: Vec<u8>) -> VeilidAPIResult<RouteId> {
let rss = self.routing_table()?.route_spec_store();
rss.import_remote_private_route(blob)
.map_err(|e| VeilidAPIError::invalid_argument(e, "blob", "private route blob"))
}
#[instrument(level = "debug", skip(self))]
pub fn release_private_route(&self, route_id: RouteId) -> VeilidAPIResult<()> {
let rss = self.routing_table()?.route_spec_store();
if !rss.release_route(route_id) {
@ -257,7 +250,6 @@ impl VeilidAPI {
////////////////////////////////////////////////////////////////
// App Calls
#[instrument(level = "debug", skip(self))]
pub async fn app_call_reply(
&self,
call_id: OperationId,
@ -274,7 +266,6 @@ impl VeilidAPI {
// Tunnel Building
#[cfg(feature = "unstable-tunnels")]
#[instrument(level = "debug", err, skip(self))]
pub async fn start_tunnel(
&self,
_endpoint_mode: TunnelMode,
@ -284,7 +275,6 @@ impl VeilidAPI {
}
#[cfg(feature = "unstable-tunnels")]
#[instrument(level = "debug", err, skip(self))]
pub async fn complete_tunnel(
&self,
_endpoint_mode: TunnelMode,
@ -295,7 +285,6 @@ impl VeilidAPI {
}
#[cfg(feature = "unstable-tunnels")]
#[instrument(level = "debug", err, skip(self))]
pub async fn cancel_tunnel(&self, _tunnel_id: TunnelId) -> VeilidAPIResult<bool> {
panic!("unimplemented");
}

2
veilid-core/src/veilid_api/routing_context.rs
View File

@ -137,7 +137,6 @@ impl RoutingContext {
////////////////////////////////////////////////////////////////
// App-level Messaging
#[instrument(level = "debug", err, skip(self))]
pub async fn app_call(&self, target: Target, request: Vec<u8>) -> VeilidAPIResult<Vec<u8>> {
let rpc_processor = self.api.rpc_processor()?;
@ -165,7 +164,6 @@ impl RoutingContext {
Ok(answer.answer)
}
#[instrument(level = "debug", err, skip(self))]
pub async fn app_message(&self, target: Target, message: Vec<u8>) -> VeilidAPIResult<()> {
let rpc_processor = self.api.rpc_processor()?;

2
veilid-flutter/rust/src/dart_ffi.rs
View File

@ -264,7 +264,6 @@ pub extern "C" fn initialize_veilid_core(platform_config: FfiStr) {
}
#[no_mangle]
#[instrument(level = "debug")]
pub extern "C" fn change_log_level(layer: FfiStr, log_level: FfiStr) {
// get layer to change level on
let layer = layer.into_opt_string().unwrap_or("all".to_owned());
@ -1502,7 +1501,6 @@ pub struct VeilidVersion {
}
#[no_mangle]
#[instrument]
pub extern "C" fn veilid_version() -> VeilidVersion {
let (major, minor, patch) = veilid_core::veilid_version();
VeilidVersion {

3
veilid-server/src/client_api.rs
View File

@ -76,7 +76,6 @@ impl ClientApi {
crate::server::shutdown();
}
#[instrument(level = "trace", skip_all)]
fn change_log_level(
&self,
layer: String,
@ -107,7 +106,6 @@ impl ClientApi {
trace!("ClientApi::stop: stopped");
}
#[instrument(level = "trace", skip(self), err)]
async fn handle_incoming(self, bind_addr: SocketAddr) -> std::io::Result<()> {
let listener = TcpListener::bind(bind_addr).await?;
debug!("Client API listening on: {:?}", bind_addr);
@ -402,7 +400,6 @@ impl ClientApi {
awg.done();
}
#[instrument(level = "trace", skip(self))]
pub fn handle_update(&self, veilid_update: veilid_core::VeilidUpdate) {
// serialize update to NDJSON
let veilid_update = serialize_json(json_api::RecvMessage::Update(veilid_update)) + "\n";