From 32f1ebebf989cb2e15aed241ff57f9c769ef138d Mon Sep 17 00:00:00 2001 From: Floriane TUERNAL SABOTINOV Date: Fri, 17 Oct 2025 11:27:45 +0200 Subject: [PATCH] refactor init tracer and use of tracing_opentelemetry span for async tracing --- Cargo.lock | 1 + .../bin-common/src/opentelemetry/context.rs | 18 +- common/bin-common/src/opentelemetry/mod.rs | 211 ++++++++++--- common/gateway-requests/Cargo.toml | 2 + .../src/types/text_request/mod.rs | 11 +- .../connection_handler/authenticated.rs | 26 +- .../websocket/connection_handler/fresh.rs | 50 +-- .../websocket/connection_handler/mod.rs | 24 +- .../client_handling/websocket/listener.rs | 4 +- nym-node/src/cli/commands/run/mod.rs | 4 +- nym-node/src/cli/mod.rs | 109 +++++-- nym-node/src/node/mixnet/handler.rs | 9 +- nym-node/src/node/mixnet/listener.rs | 6 +- nym-node/src/node/mixnet/shared/mod.rs | 5 +- nym-node/src/node/mod.rs | 15 +- sdk/rust/nym-sdk/examples/surb_reply.rs | 291 ++++++++++++++---- sdk/rust/nym-sdk/src/mixnet/traits.rs | 16 +- 17 files changed, 597 insertions(+), 205 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fdd308eb6d..681bc81951 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5811,6 +5811,7 @@ dependencies = [ "time", "tokio", "tracing", + "tracing-opentelemetry", "tungstenite 0.20.1", "wasmtimer", "zeroize", diff --git a/common/bin-common/src/opentelemetry/context.rs b/common/bin-common/src/opentelemetry/context.rs index 023ffc2159..a529074cd1 100644 --- a/common/bin-common/src/opentelemetry/context.rs +++ b/common/bin-common/src/opentelemetry/context.rs @@ -5,6 +5,7 @@ use opentelemetry_sdk::{propagation::TraceContextPropagator, trace::IdGenerator} use tracing_opentelemetry::OpenTelemetrySpanExt; use std::collections::HashMap; use std::fmt::Display; +use tracing::instrument; /// Make a Carrier for context propagation pub struct ContextCarrier { @@ -93,6 +94,7 @@ pub struct ManualContextPropagator { } impl ManualContextPropagator { + #[instrument(skip_all, level = "debug")] pub fn new(name: &str, context: HashMap) -> Self { let carrier = ContextCarrier::new_with_data(context); let trace_id = match carrier.extract_trace_id() { @@ -101,7 +103,6 @@ impl ManualContextPropagator { }; let root_span_builder = new_span_context_with_id(trace_id.clone()); - let _guard = root_span_builder.clone().attach(); let root_span = tracing::info_span!("trace_root", name = %name, trace_id = %trace_id); root_span.set_parent(root_span_builder); @@ -112,9 +113,9 @@ impl ManualContextPropagator { } } + #[instrument(skip_all, level = "debug")] pub fn new_from_tid(name: &str, trace_id: TraceId) -> Self { let root_span_builder = new_span_context_with_id(trace_id.clone()); - let _guard = root_span_builder.clone().attach(); let root_span = tracing::info_span!("trace_root", name = %name, trace_id = %trace_id); root_span.set_parent(root_span_builder); @@ -125,8 +126,13 @@ impl ManualContextPropagator { } } + pub fn root_span(&self) -> &tracing::Span { + &self.root_span + } + } +#[instrument(skip_all, level = "debug")] pub fn new_span_context_with_id(trace_id: TraceId) -> Context { let id_gen = opentelemetry_sdk::trace::RandomIdGenerator::default(); let span_id = id_gen.new_span_id(); @@ -139,4 +145,12 @@ pub fn new_span_context_with_id(trace_id: TraceId) -> Context { ); Context::current().with_remote_span_context(span_context) +} + +#[instrument(skip_all, level = "debug")] +pub fn extract_trace_id_from_tracing_cx() -> TraceId { + let cx = tracing::Span::current().context(); + let binding = cx.span(); + let trace_id = binding.span_context().trace_id(); + trace_id } \ No newline at end of file diff --git a/common/bin-common/src/opentelemetry/mod.rs b/common/bin-common/src/opentelemetry/mod.rs index 500b40bbf4..2f1aeeb94a 100644 --- a/common/bin-common/src/opentelemetry/mod.rs +++ b/common/bin-common/src/opentelemetry/mod.rs @@ -53,23 +53,30 @@ pub(crate) fn granual_filtered_env() -> Result Result<(), TracingError> { +pub fn setup_tracing_logger(service_name: String) -> Result { if tracing::dispatcher::has_been_set() { // It shouldn't be - this is really checking that it is torn down between async command executions return Err(TracingError::TracingLoggerAlreadyInitialised); } - let key = - std::env::var("SIGNOZ_INGESTION_KEY".to_string()).expect("SIGNOZ_INGESTION_KEY not set"); + // define ingestion points + let endpoint = std::env::var("SIGNOZ_ENDPOINT").expect("SIGNOZ_ENDPOINT not set"); + let key = std::env::var("SIGNOZ_INGESTION_KEY").expect("SIGNOZ_INGESTION_KEY not set"); let mut metadata = MetadataMap::new(); metadata.insert( "signoz-ingestion-key", key.parse().expect("Could not parse signoz ingestion key"), ); - let tracer_provider = init_tracer_provider(metadata.clone(), service_name.clone())?; - let meter_provider = init_meter_provider(metadata.clone(), service_name.clone())?; - let tracer = tracer_provider.tracer("tracing-otel-subscriber"); + // Build resources + let resource = build_resource(&service_name); + + // Initialize tracer and meter providers + let tracer_provider = init_tracer_provider(&endpoint, metadata.clone(), resource.clone())?; + let meter_provider = init_meter_provider(&endpoint, metadata.clone(), resource.clone())?; + + // Bridge tracing and opentelemetry + let tracer = tracer_provider.tracer("otel-subscriber"); let fmt_layer = fmt::layer() .json() .with_writer(std::io::stderr) @@ -78,36 +85,26 @@ pub fn setup_tracing_logger(service_name: String) -> Result<(), TracingError> { .with_current_span(true) .event_format(trace_id_format::TraceIdFormat); - cfg_if::cfg_if! {if #[cfg(feature = "tokio-console")] { - // instrument tokio console subscriber needs RUSTFLAGS="--cfg tokio_unstable" at build time - let console_layer = console_subscriber::spawn(); + let registry = tracing_subscriber::registry() + .with(fmt_layer) + .with(granual_filtered_env()?) + .with(tracing_subscriber::filter::LevelFilter::from_level(Level::INFO)) + .with(MetricsLayer::new(meter_provider.clone())) + .with(OpenTelemetryLayer::new(tracer)); - tracing_subscriber::registry() - .with(console_layer) - .with(fmt_layer) - .with(granual_filtered_env()?) - .with(tracing_subscriber::filter::LevelFilter::from_level(Level::INFO)) - .with(MetricsLayer::new(meter_provider)) - .with(OpenTelemetryLayer::new(tracer)) - .try_init() - .map_err(|e| TracingError::TracingTryInitError(e))?; - } else { - tracing_subscriber::registry() - .with(fmt_layer) - .with(granual_filtered_env()?) - .with(tracing_subscriber::filter::LevelFilter::from_level(Level::INFO)) - .with(MetricsLayer::new(meter_provider)) - .with(OpenTelemetryLayer::new(tracer)) - .try_init() - .map_err(|e| TracingError::TracingTryInitError(e))?; - }} + registry.try_init().map_err(TracingError::TracingTryInitError)?; - Ok(()) + global::set_tracer_provider(tracer_provider.clone()); + global::set_meter_provider(meter_provider.clone()); + + info!("Tracing initialized with service name: {}", service_name); + + Ok(TracerProviderGuard(Some(tracer_provider))) } -fn resource(service_name: String) -> Resource { +fn build_resource(service_name: &str) -> Resource { Resource::builder() - .with_service_name(service_name) + .with_service_name(service_name.to_string()) .with_schema_url( [ KeyValue::new(SERVICE_VERSION, env!("CARGO_PKG_VERSION")), @@ -118,14 +115,15 @@ fn resource(service_name: String) -> Resource { .build() } -fn init_tracer_provider(metadata: MetadataMap, service_name: String) -> Result { - let endpoint = std::env::var("SIGNOZ_ENDPOINT".to_string()).expect("SIGNOZ_ENDPOINT not set"); - info!("SIGNOZ_ENDPOINT = {}", endpoint); - +fn init_tracer_provider( + endpoint: &str, + metadata: MetadataMap, + resource: Resource, +) -> Result { let mut exporter_builder = opentelemetry_otlp::SpanExporter::builder() .with_tonic() .with_metadata(metadata) - .with_endpoint(&endpoint); + .with_endpoint(endpoint); if endpoint.starts_with("https://") { exporter_builder = @@ -139,7 +137,7 @@ fn init_tracer_provider(metadata: MetadataMap, service_name: String) -> Result Result Result { - let endpoint = std::env::var("SIGNOZ_ENDPOINT".to_string()).expect("SIGNOZ_ENDPOINT not set"); - +fn init_meter_provider( + endpoint: &str, + metadata: MetadataMap, + resource: Resource, +) -> Result { let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder() .with_tonic() .with_metadata(metadata) - .with_endpoint(&endpoint) + .with_endpoint(endpoint) .with_temporality(opentelemetry_sdk::metrics::Temporality::default()); if endpoint.starts_with("https://") { @@ -170,7 +170,7 @@ fn init_meter_provider(metadata: MetadataMap, service_name: String) -> Result Result Result<(), TracingError> { +// if tracing::dispatcher::has_been_set() { +// // It shouldn't be - this is really checking that it is torn down between async command executions +// return Err(TracingError::TracingLoggerAlreadyInitialised); +// } + +// let key = +// std::env::var("SIGNOZ_INGESTION_KEY".to_string()).expect("SIGNOZ_INGESTION_KEY not set"); +// let mut metadata = MetadataMap::new(); +// metadata.insert( +// "signoz-ingestion-key", +// key.parse().expect("Could not parse signoz ingestion key"), +// ); + +// let tracer_provider = init_tracer_provider(metadata.clone(), service_name.clone())?; +// let meter_provider = init_meter_provider(metadata.clone(), service_name.clone())?; +// let tracer = tracer_provider.tracer("tracing-otel-subscriber"); +// let fmt_layer = fmt::layer() +// .json() +// .with_writer(std::io::stderr) +// .with_span_events(FmtSpan::NEW | FmtSpan::CLOSE) +// .with_span_list(false) +// .with_current_span(true) +// .event_format(trace_id_format::TraceIdFormat); + +// cfg_if::cfg_if! {if #[cfg(feature = "tokio-console")] { +// // instrument tokio console subscriber needs RUSTFLAGS="--cfg tokio_unstable" at build time +// let console_layer = console_subscriber::spawn(); + +// tracing_subscriber::registry() +// .with(console_layer) +// .with(fmt_layer) +// .with(granual_filtered_env()?) +// .with(tracing_subscriber::filter::LevelFilter::from_level(Level::INFO)) +// .with(MetricsLayer::new(meter_provider)) +// .with(OpenTelemetryLayer::new(tracer)) +// .try_init() +// .map_err(|e| TracingError::TracingTryInitError(e))?; +// } else { +// tracing_subscriber::registry() +// .with(fmt_layer) +// .with(granual_filtered_env()?) +// .with(tracing_subscriber::filter::LevelFilter::from_level(Level::INFO)) +// .with(MetricsLayer::new(meter_provider)) +// .with(OpenTelemetryLayer::new(tracer)) +// .try_init() +// .map_err(|e| TracingError::TracingTryInitError(e))?; +// }} + +// Ok(()) +// } + +// fn resource(service_name: String) -> Resource { +// Resource::builder() +// .with_service_name(service_name) +// .with_schema_url( +// [ +// KeyValue::new(SERVICE_VERSION, env!("CARGO_PKG_VERSION")), +// KeyValue::new(DEPLOYMENT_ENVIRONMENT_NAME, "develop"), +// ], +// SCHEMA_URL, +// ) +// .build() +// } + +// fn init_tracer_provider(metadata: MetadataMap, service_name: String) -> Result { +// let endpoint = std::env::var("SIGNOZ_ENDPOINT".to_string()).expect("SIGNOZ_ENDPOINT not set"); +// info!("SIGNOZ_ENDPOINT = {}", endpoint); + +// let mut exporter_builder = opentelemetry_otlp::SpanExporter::builder() +// .with_tonic() +// .with_metadata(metadata) +// .with_endpoint(&endpoint); + +// if endpoint.starts_with("https://") { +// exporter_builder = +// exporter_builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots()); +// } + +// let exporter = exporter_builder.build()?; + +// let tracer = SdkTracerProvider::builder() +// .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased( +// 1.0, +// )))) +// .with_id_generator(Compact13BytesIdGenerator) +// .with_resource(resource(service_name)) +// .with_batch_exporter(exporter) +// .build(); + +// global::set_tracer_provider(tracer.clone()); +// Ok(tracer) +// } + +// fn init_meter_provider(metadata: MetadataMap, service_name: String) -> Result { +// let endpoint = std::env::var("SIGNOZ_ENDPOINT".to_string()).expect("SIGNOZ_ENDPOINT not set"); + +// let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder() +// .with_tonic() +// .with_metadata(metadata) +// .with_endpoint(&endpoint) +// .with_temporality(opentelemetry_sdk::metrics::Temporality::default()); + +// if endpoint.starts_with("https://") { +// exporter_builder = exporter_builder.with_tls_config(ClientTlsConfig::new().with_enabled_roots()); +// } + +// let exporter = exporter_builder.build()?; + +// let reader = PeriodicReader::builder(exporter) +// .with_interval(std::time::Duration::from_secs(30)) +// .build(); + +// let stdout_reader = +// PeriodicReader::builder(opentelemetry_stdout::MetricExporter::default()).build(); + +// let meter_provider = MeterProviderBuilder::default() +// .with_resource(resource(service_name)) +// .with_reader(reader) +// .with_reader(stdout_reader) +// .build(); + +// global::set_meter_provider(meter_provider.clone()); + +// Ok(meter_provider) +// } diff --git a/common/gateway-requests/Cargo.toml b/common/gateway-requests/Cargo.toml index 682a4a485f..d3b3f4618e 100644 --- a/common/gateway-requests/Cargo.toml +++ b/common/gateway-requests/Cargo.toml @@ -36,6 +36,7 @@ nym-credentials-interface = { path = "../credentials-interface" } opentelemetry = { workspace = true, features = ["trace"], optional = true } opentelemetry_sdk = { workspace = true, optional = true } +tracing-opentelemetry = { workspace = true, optional = true } tracing = { workspace = true, features = ["std", "attributes", "tracing-attributes"] } [target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio] @@ -62,4 +63,5 @@ otel = [ "nym-bin-common/otel", "opentelemetry", "opentelemetry_sdk", + "tracing-opentelemetry", ] diff --git a/common/gateway-requests/src/types/text_request/mod.rs b/common/gateway-requests/src/types/text_request/mod.rs index 08f7af6df4..d3051af2d4 100644 --- a/common/gateway-requests/src/types/text_request/mod.rs +++ b/common/gateway-requests/src/types/text_request/mod.rs @@ -164,11 +164,18 @@ impl ClientControlRequest { #[cfg(feature = "otel")] let context_carrier = { - let otel_context = opentelemetry::Context::current(); + use nym_bin_common::opentelemetry::context::extract_trace_id_from_tracing_cx; + let trace_id = extract_trace_id_from_tracing_cx(); + tracing::info!("[DEBUG] Trace id in new_authenticate_v2: {:?}", trace_id); + + use tracing_opentelemetry::OpenTelemetrySpanExt; + + let current_span = tracing::Span::current(); + let otel_context = current_span.context(); ContextCarrier::new_with_current_context(otel_context).into_map() }; #[cfg(not(feature = "otel"))] - let context_carrier = None; + let context_carrier: HashMap = HashMap::new(); Ok(ClientControlRequest::AuthenticateV2(Box::new( AuthenticateRequest::new( 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 ee37ccdfb1..fc9fa0c7aa 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs @@ -152,7 +152,7 @@ pub(crate) struct AuthenticatedHandler { is_active_request_receiver: IsActiveRequestReceiver, is_active_ping_pending_reply: Option<(u64, IsActiveResultSender)>, #[cfg(feature = "otel")] - otel_propagator: Option, + pub otel_propagator: Option, } // explicitly remove handle from the global store upon being dropped @@ -337,6 +337,7 @@ impl AuthenticatedHandler { let remaining_bandwidth = self .bandwidth_storage_manager .try_use_bandwidth(required_bandwidth) + .in_current_span() .await?; self.forward_packet(mix_packet); @@ -374,7 +375,7 @@ impl AuthenticatedHandler { // currently only a single type exists BinaryRequest::ForwardSphinx { packet } | BinaryRequest::ForwardSphinxV2 { packet } => { - self.handle_forward_sphinx(packet).await.into_ws_message() + self.handle_forward_sphinx(packet).in_current_span().await.into_ws_message() } _ => RequestHandlingError::UnknownBinaryRequest.into_error_message(), }, @@ -391,6 +392,7 @@ impl AuthenticatedHandler { .shared_state() .storage() .handle_forget_me(self.client.address) + .in_current_span() .await?; } Ok(SensitiveServerResponse::ForgetMeAck {}.encrypt(&self.client.shared_keys)?) @@ -451,9 +453,9 @@ impl AuthenticatedHandler { hkdf_salt, derived_key_digest, } => self.handle_key_upgrade(hkdf_salt, derived_key_digest).await, - ClientRequest::ForgetMe { client, stats } => self.handle_forget_me(client, stats).await, + ClientRequest::ForgetMe { client, stats } => self.handle_forget_me(client, stats).in_current_span().await, ClientRequest::RememberMe { session_type } => { - self.handle_remember_me(session_type).await + self.handle_remember_me(session_type).in_current_span().await } _ => Err(RequestHandlingError::UnknownEncryptedTextRequest), } @@ -486,10 +488,10 @@ impl AuthenticatedHandler { match request { ClientControlRequest::EncryptedRequest { ciphertext, nonce } => { - self.handle_encrypted_text_request(ciphertext, nonce).await + self.handle_encrypted_text_request(ciphertext, nonce).in_current_span().await } ClientControlRequest::EcashCredential { enc_credential, iv } => { - self.handle_ecash_bandwidth(enc_credential, iv).await + self.handle_ecash_bandwidth(enc_credential, iv).in_current_span().await } ClientControlRequest::BandwidthCredential { .. } => { Err(RequestHandlingError::IllegalRequest { @@ -578,10 +580,10 @@ impl AuthenticatedHandler { // them and let's test that claim. If that's not the case, just copy code from // desktop nym-client websocket as I've manually handled everything there match raw_request { - Message::Binary(bin_msg) => Some(self.handle_binary(bin_msg).await), - Message::Text(text_msg) => Some(self.handle_text(text_msg).await), + Message::Binary(bin_msg) => Some(self.handle_binary(bin_msg).in_current_span().await), + Message::Text(text_msg) => Some(self.handle_text(text_msg).in_current_span().await), Message::Pong(msg) => { - self.handle_pong(msg).await; + self.handle_pong(msg).in_current_span().await; None } _ => None, @@ -675,7 +677,7 @@ impl AuthenticatedHandler { ping_timeout = None.into(); self.handle_ping_timeout().await; }, - socket_msg = self.inner.read_websocket_message() => { + socket_msg = self.inner.read_websocket_message().in_current_span() => { #[cfg(feature = "otel")] let span = { let span = match &self.otel_propagator { @@ -700,7 +702,7 @@ impl AuthenticatedHandler { break; } - if let Some(response) = self.handle_request(socket_msg).await { + if let Some(response) = self.handle_request(socket_msg).in_current_span().await { if let Err(err) = self.inner.send_websocket_message(response).await { debug!( "Failed to send message over websocket: {err}. Assuming the connection is dead.", @@ -709,7 +711,7 @@ impl AuthenticatedHandler { } } }, - mix_messages = self.mix_receiver.next() => { + mix_messages = self.mix_receiver.next().in_current_span() => { #[cfg(feature = "otel")] let span = { let span = match &self.otel_propagator { 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 272c91e930..09372a4369 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs @@ -47,7 +47,7 @@ use tokio::time::timeout; use tokio_tungstenite::tungstenite::{protocol::Message, Error as WsError}; #[cfg(feature = "otel")] use tracing::info_span; -use tracing::{debug, error, info, instrument, warn}; +use tracing::{debug, error, info, instrument, Instrument, warn}; @@ -171,6 +171,7 @@ impl FreshHandler { /// Attempts to perform websocket handshake with the remote and upgrades the raw TCP socket /// to the framed WebSocket. + #[instrument(skip_all)] pub(crate) async fn perform_websocket_handshake(&mut self) -> Result<(), WsError> where S: AsyncRead + AsyncWrite + Unpin, @@ -180,7 +181,7 @@ impl FreshHandler { SocketStream::RawTcp(conn) => { // TODO: perhaps in the future, rather than panic here (and uncleanly shut tcp stream) // return a result with an error? - let ws_stream = Box::new(tokio_tungstenite::accept_async(conn).await?); + let ws_stream = Box::new(tokio_tungstenite::accept_async(conn).in_current_span().await?); SocketStream::UpgradedWebSocket(ws_stream) } other => other, @@ -194,6 +195,7 @@ impl FreshHandler { /// # Arguments /// /// * `init_msg`: a client handshake init message which should contain its identity public key as well as an ephemeral key. + #[instrument(skip_all)] async fn perform_registration_handshake( &mut self, init_msg: Vec, @@ -212,6 +214,7 @@ impl FreshHandler { init_msg, self.shutdown.clone(), ) + .in_current_span() .await } _ => unreachable!(), @@ -225,7 +228,7 @@ impl FreshHandler { S: AsyncRead + AsyncWrite + Unpin, { match self.socket_connection { - SocketStream::UpgradedWebSocket(ref mut ws_stream) => ws_stream.next().await, + SocketStream::UpgradedWebSocket(ref mut ws_stream) => ws_stream.next().in_current_span().await, _ => panic!("impossible state - websocket handshake was somehow reverted"), } } @@ -247,7 +250,7 @@ impl FreshHandler { // TODO: more closely investigate difference between `Sink::send` and `Sink::send_all` // it got something to do with batching and flushing - it might be important if it // turns out somehow we've got a bottleneck here - SocketStream::UpgradedWebSocket(ref mut ws_stream) => ws_stream.send(msg.into()).await, + SocketStream::UpgradedWebSocket(ref mut ws_stream) => ws_stream.send(msg.into()).in_current_span().await, _ => panic!("impossible state - websocket handshake was somehow reverted"), } } @@ -261,6 +264,7 @@ impl FreshHandler { S: AsyncRead + AsyncWrite + Unpin + Send, { self.send_websocket_message(ServerResponse::new_error(err.to_string())) + .in_current_span() .await } @@ -713,18 +717,20 @@ impl FreshHandler { self.shared_state .storage .update_last_used_authentication_timestamp(client_id, session_request_start) + .in_current_span() .await?; // push any old stored messages to the client // (this will be removed soon) self.push_stored_messages_to_client(address, &shared_key.key) + .in_current_span() .await?; // finally check and retrieve client's bandwidth let available_bandwidth = self.get_registered_available_bandwidth(client_id).await?; let bandwidth_remaining = if available_bandwidth.expired() { - self.shared_state.storage.reset_bandwidth(client_id).await?; + self.shared_state.storage.reset_bandwidth(client_id).in_current_span().await?; 0 } else { available_bandwidth.bytes @@ -826,8 +832,8 @@ impl FreshHandler { return Err(InitialAuthenticationError::DuplicateConnection); } - let shared_keys = self.perform_registration_handshake(init_data).await?; - let client_id = self.register_client(remote_address, &shared_keys).await?; + let shared_keys = self.perform_registration_handshake(init_data).in_current_span().await?; + let client_id = self.register_client(remote_address, &shared_keys).in_current_span().await?; debug!(client_id = %client_id, "managed to finalize client registration"); @@ -877,15 +883,11 @@ impl FreshHandler { S: AsyncRead + AsyncWrite + Unpin + Send, R: CryptoRng + RngCore + Send, { - #[cfg(not(feature = "otel"))] - info!("[DEBUG] received initial client request no otel"); - #[cfg(feature = "otel")] - info!("[DEBUG] received initial client request with otel"); // extract and set up opentelemetry context if provided #[cfg(feature = "otel")] let (context_propagator, otel_ctx) = if let ClientControlRequest::AuthenticateV2(ref auth_req) = request { if let Some(otel_context) = &auth_req.otel_context { - warn!("OpenTelemetry context provided in the request: {otel_context:?}"); + warn!("=== OpenTelemetry context provided in the request: {otel_context:?} ==="); (Some(ManualContextPropagator::new("handling_initial_client_request_with_otel", otel_context.clone())), Some(otel_context.clone())) } else { warn!("No OpenTelemetry context provided in the request"); @@ -915,18 +917,19 @@ impl FreshHandler { otel_context: _, } => { self.handle_legacy_authenticate(protocol_version, address, enc_address, iv) + .in_current_span() .await } #[cfg(feature = "otel")] - ClientControlRequest::AuthenticateV2(req) => self.handle_authenticate_v2(req, otel_ctx).await, + ClientControlRequest::AuthenticateV2(req) => self.handle_authenticate_v2(req, otel_ctx).in_current_span().await, #[cfg(not(feature = "otel"))] - ClientControlRequest::AuthenticateV2(req) => self.handle_authenticate_v2(req).await, + ClientControlRequest::AuthenticateV2(req) => self.handle_authenticate_v2(req).in_current_span().await, ClientControlRequest::RegisterHandshakeInitRequest { protocol_version, data, - } => self.handle_register(protocol_version, data).await, + } => self.handle_register(protocol_version, data).in_current_span().await, ClientControlRequest::SupportedProtocol { .. } => { - self.handle_reply_supported_protocol_request().await; + self.handle_reply_supported_protocol_request().in_current_span().await; return Ok(None); } _ => { @@ -944,7 +947,7 @@ impl FreshHandler { } other => debug!("authentication failure: {other}"), } - self.send_and_forget_error_response(&err).await; + self.send_and_forget_error_response(&err).in_current_span().await; return Err(err); } }; @@ -979,21 +982,21 @@ impl FreshHandler { R: CryptoRng + RngCore + Send, { loop { - let req = self.wait_for_initial_message().await; + let req = self.wait_for_initial_message().in_current_span().await; let initial_request = match req { Ok(req) => req, Err(err) => { - self.send_and_forget_error_response(err).await; + self.send_and_forget_error_response(err).in_current_span().await; return None; } }; // see if we managed to register the client through this request - let maybe_auth_res = match self.handle_initial_client_request(initial_request).await { + let maybe_auth_res = match self.handle_initial_client_request(initial_request).in_current_span().await { Ok(maybe_auth_res) => maybe_auth_res, Err(err) => { debug!("initial client request handling error: {err}"); - self.send_and_forget_error_response(err).await; + self.send_and_forget_error_response(err).in_current_span().await; return None; } }; @@ -1015,6 +1018,7 @@ impl FreshHandler { mix_receiver, is_active_request_receiver, ) + .in_current_span() .await .inspect_err(|err| error!("failed to upgrade client handler: {err}")) .ok(); @@ -1030,7 +1034,7 @@ impl FreshHandler { where S: AsyncRead + AsyncWrite + Unpin + Send, { - let msg = match timeout(INITIAL_MESSAGE_TIMEOUT, self.read_websocket_message()).await { + let msg = match timeout(INITIAL_MESSAGE_TIMEOUT, self.read_websocket_message()).in_current_span().await { Ok(Some(Ok(msg))) => msg, Ok(Some(Err(source))) => { debug!("failed to obtain message from websocket stream! stopping connection handler: {source}"); @@ -1086,7 +1090,7 @@ impl FreshHandler { _ = shutdown.cancelled() => { tracing::trace!("received cancellation") } - _ = super::handle_connection(self) => { + _ = super::handle_connection(self).instrument(tracing::debug_span!("connection_handler", remote = %remote)) => { tracing::debug!("finished connection handler for {remote}") } } 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 01aee5b2c2..c562eaa94a 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/mod.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/mod.rs @@ -13,7 +13,7 @@ use std::time::Duration; use time::OffsetDateTime; use tokio::io::{AsyncRead, AsyncWrite}; use tokio_tungstenite::WebSocketStream; -use tracing::{debug, instrument, trace, warn}; +use tracing::{debug, instrument, Instrument, trace, warn}; pub(crate) use self::authenticated::AuthenticatedHandler; pub(crate) use self::fresh::FreshHandler; @@ -108,7 +108,7 @@ where { match tokio::time::timeout( WEBSOCKET_HANDSHAKE_TIMEOUT, - handle.perform_websocket_handshake(), + handle.perform_websocket_handshake().in_current_span(), ) .await { @@ -125,8 +125,24 @@ where trace!("managed to perform websocket handshake!"); - if let Some(auth_handle) = handle.handle_until_authenticated_or_failure().await { - auth_handle.listen_for_requests().await + if let Some(auth_handle) = handle.handle_until_authenticated_or_failure().in_current_span().await { + #[cfg(feature = "otel")] + { + let from_client_span = { + let parent = match auth_handle.otel_propagator.as_ref() { + Some(propagator) => propagator.root_span(), + None => &tracing::Span::current(), // fallback to current span if no propagator + }; + tracing::info_span!(parent: parent, "listening for requests") + }; + auth_handle.listen_for_requests() + .instrument(from_client_span) + .await + } + #[cfg(not(feature = "otel"))] + { + auth_handle.listen_for_requests().await; + } } trace!("the handler is done!"); diff --git a/gateway/src/node/client_handling/websocket/listener.rs b/gateway/src/node/client_handling/websocket/listener.rs index 180b488bd6..0f9c8ba826 100644 --- a/gateway/src/node/client_handling/websocket/listener.rs +++ b/gateway/src/node/client_handling/websocket/listener.rs @@ -92,7 +92,7 @@ impl Listener { let metrics_ref = handle.shared_state.metrics.clone(); // 4.1. handle all client requests until connection gets terminated - handle.start_handling().await; + handle.start_handling().in_current_span().await; // 4.2. decrement the connection counter metrics_ref.network.disconnected_ingress_websocket_client(); @@ -125,7 +125,7 @@ impl Listener { trace!("client_handling::Listener: received shutdown"); break } - connection = tcp_listener.accept() => { + connection = tcp_listener.accept().in_current_span() => { self.try_handle_accepted_connection(connection) } } diff --git a/nym-node/src/cli/commands/run/mod.rs b/nym-node/src/cli/commands/run/mod.rs index 3112e9c9c6..713fedccd9 100644 --- a/nym-node/src/cli/commands/run/mod.rs +++ b/nym-node/src/cli/commands/run/mod.rs @@ -8,7 +8,7 @@ use crate::node::mixnet::packet_forwarding::global::is_global_ip; use crate::node::NymNode; use std::fs; use std::net::IpAddr; -use tracing::{debug, info, instrument, trace, warn}; +use tracing::{debug, info, instrument, Instrument, trace, warn}; mod args; @@ -125,5 +125,5 @@ pub(crate) async fn execute(mut args: Args) -> Result<(), NymNodeError> { return Ok(()); } - nym_node.run().await + nym_node.run().in_current_span().await } diff --git a/nym-node/src/cli/mod.rs b/nym-node/src/cli/mod.rs index 2bc5b32e21..86dc1b22ed 100644 --- a/nym-node/src/cli/mod.rs +++ b/nym-node/src/cli/mod.rs @@ -17,7 +17,9 @@ use nym_bin_common::logging::error::TracingError; #[cfg(feature = "otel")] use nym_bin_common::opentelemetry::setup_tracing_logger; #[cfg(feature = "otel")] -use opentelemetry::{Context, global, trace::{TraceContextExt, Tracer}}; +use opentelemetry::{global, trace::{TraceContextExt, Tracer}}; +#[cfg(feature = "otel")] +use tracing::Instrument; use std::future::Future; use std::sync::OnceLock; use tracing::instrument; @@ -92,48 +94,103 @@ impl Cli { // SigNoz/OTEL run in async context Commands::BondingInformation(args) => Self::execute_async(async move { #[cfg(feature = "otel")] - setup_tracing_logger("nym-node".to_string()) - .map_err(TracingError::from)?; + { + let _guard = setup_tracing_logger("nym-node".to_string()) + .map_err(TracingError::from)?; + let main_span = tracing::info_span!("startup", service = "nym-node"); + async { + bonding_information::execute(args).in_current_span().await?; + Ok::<(), anyhow::Error>(()) + } + .instrument(main_span) + .await + } #[cfg(not(feature = "otel"))] - setup_no_otel_logger().expect("failed to initialize logging"); - bonding_information::execute(args).await?; - Ok::<(), anyhow::Error>(()) + { + setup_no_otel_logger().expect("failed to initialize logging"); + bonding_information::execute(args).await?; + Ok::<(), anyhow::Error>(()) + } })??, Commands::NodeDetails(args) => Self::execute_async(async move { #[cfg(feature = "otel")] - setup_tracing_logger("nym-node".to_string()) - .map_err(TracingError::from)?; + { + let _guard = setup_tracing_logger("nym-node".to_string()) + .map_err(TracingError::from)?; + let main_span = tracing::info_span!("startup", service = "nym-node"); + async { + node_details::execute(args).in_current_span().await?; + Ok::<(), anyhow::Error>(()) + } + .instrument(main_span) + .await + } #[cfg(not(feature = "otel"))] - setup_no_otel_logger().expect("failed to initialize logging"); - node_details::execute(args).await?; - Ok::<(), anyhow::Error>(()) + { + setup_no_otel_logger().expect("failed to initialize logging"); + node_details::execute(args).await?; + Ok::<(), anyhow::Error>(()) + } })??, Commands::Run(args) => Self::execute_async(async move { #[cfg(feature = "otel")] - setup_tracing_logger("nym-node".to_string()) - .map_err(TracingError::from)?; + { + let _guard = setup_tracing_logger("nym-node".to_string()) + .map_err(TracingError::from)?; + let main_span = tracing::info_span!("startup", service = "nym-node"); + async { + run::execute(*args).in_current_span().await?; + Ok::<(), anyhow::Error>(()) + } + .instrument(main_span) + .await + } #[cfg(not(feature = "otel"))] - setup_no_otel_logger().expect("failed to initialize logging"); - run::execute(*args).await?; - Ok::<(), anyhow::Error>(()) + { + setup_no_otel_logger().expect("failed to initialize logging"); + run::execute(*args).await?; + Ok::<(), anyhow::Error>(()) + } })??, Commands::Sign(args) => Self::execute_async(async move { #[cfg(feature = "otel")] - setup_tracing_logger("nym-node".to_string()) - .map_err(TracingError::from)?; + { + let _guard = setup_tracing_logger("nym-node".to_string()) + .map_err(TracingError::from)?; + let main_span = tracing::info_span!("startup", service = "nym-node"); + async { + sign::execute(args).in_current_span().in_current_span().await?; + Ok::<(), anyhow::Error>(()) + } + .instrument(main_span) + .await + } #[cfg(not(feature = "otel"))] - setup_no_otel_logger().expect("failed to initialize logging"); - sign::execute(args).await?; - Ok::<(), anyhow::Error>(()) + { + setup_no_otel_logger().expect("failed to initialize logging"); + sign::execute(args).await?; + Ok::<(), anyhow::Error>(()) + } })??, Commands::UnsafeResetSphinxKeys(args) => Self::execute_async(async move { #[cfg(feature = "otel")] - setup_tracing_logger("nym-node".to_string()) - .map_err(TracingError::from)?; + { + let _guard = setup_tracing_logger("nym-node".to_string()) + .map_err(TracingError::from)?; + let main_span = tracing::info_span!("startup", service = "nym-node"); + async { + reset_sphinx_keys::execute(args).in_current_span().await?; + Ok::<(), anyhow::Error>(()) + } + .instrument(main_span) + .await + } #[cfg(not(feature = "otel"))] - setup_no_otel_logger().expect("failed to initialize logging"); - reset_sphinx_keys::execute(args).await?; - Ok::<(), anyhow::Error>(()) + { + setup_no_otel_logger().expect("failed to initialize logging"); + reset_sphinx_keys::execute(args).await?; + Ok::<(), anyhow::Error>(()) + } })??, } Ok(()) diff --git a/nym-node/src/node/mixnet/handler.rs b/nym-node/src/node/mixnet/handler.rs index 6a14277c1d..d31159e95f 100644 --- a/nym-node/src/node/mixnet/handler.rs +++ b/nym-node/src/node/mixnet/handler.rs @@ -20,7 +20,7 @@ use std::net::SocketAddr; use tokio::net::TcpStream; use tokio::time::Instant; use tokio_util::codec::Framed; -use tracing::{debug, error, instrument, trace, warn}; +use tracing::{debug, error, instrument, trace, warn, Instrument}; struct PendingReplayCheckPackets { // map of rotation id used for packet creation to the packets @@ -446,6 +446,7 @@ impl ConnectionHandler { // if it's a sphinx packet attempt to do pre-processing and replay detection if packet.is_sphinx() && !self.shared.replay_protection_filter.disabled() { self.handle_received_packet_with_replay_detection(now, packet) + .in_current_span() .await; } else { // otherwise just skip that whole procedure and go straight to payload unwrapping @@ -463,7 +464,7 @@ impl ConnectionHandler { ) )] pub(crate) async fn handle_connection(&mut self, socket: TcpStream) { - let noise_stream = match upgrade_noise_responder(socket, &self.shared.noise_config).await { + let noise_stream = match upgrade_noise_responder(socket, &self.shared.noise_config).in_current_span().await { Ok(noise_stream) => noise_stream, Err(err) => { error!( @@ -478,9 +479,11 @@ impl ConnectionHandler { self.remote_address ); self.handle_stream(Framed::new(noise_stream, NymCodec)) + .in_current_span() .await } + #[instrument(skip_all)] pub(crate) async fn handle_stream( &mut self, mut mixnet_connection: Framed, NymCodec>, @@ -494,7 +497,7 @@ impl ConnectionHandler { } maybe_framed_nym_packet = mixnet_connection.next() => { match maybe_framed_nym_packet { - Some(Ok(packet)) => self.handle_received_nym_packet(packet).await, + Some(Ok(packet)) => self.handle_received_nym_packet(packet).in_current_span().await, Some(Err(err)) => { debug!("connection got corrupted with: {err}"); return diff --git a/nym-node/src/node/mixnet/listener.rs b/nym-node/src/node/mixnet/listener.rs index 7fdca155a7..3545f78c03 100644 --- a/nym-node/src/node/mixnet/listener.rs +++ b/nym-node/src/node/mixnet/listener.rs @@ -4,7 +4,7 @@ use crate::node::mixnet::SharedData; use nym_task::ShutdownToken; use std::net::SocketAddr; -use tracing::{debug, error, info, trace}; +use tracing::{debug, error, info, instrument, Instrument, trace}; pub(crate) struct Listener { bind_address: SocketAddr, @@ -18,7 +18,7 @@ impl Listener { shared_data, } } - + #[instrument(skip_all, level = "debug")] pub(crate) async fn run(&mut self, shutdown: ShutdownToken) { info!("attempting to run mixnet listener on {}", self.bind_address); @@ -38,7 +38,7 @@ impl Listener { trace!("mixnet listener: received shutdown"); break } - connection = tcp_listener.accept() => { + connection = tcp_listener.accept().in_current_span() => { self.shared_data.try_handle_connection(connection); } } diff --git a/nym-node/src/node/mixnet/shared/mod.rs b/nym-node/src/node/mixnet/shared/mod.rs index 67dea46a7a..ff2b3507bd 100644 --- a/nym-node/src/node/mixnet/shared/mod.rs +++ b/nym-node/src/node/mixnet/shared/mod.rs @@ -23,7 +23,7 @@ use std::time::Duration; use tokio::net::TcpStream; use tokio::task::JoinHandle; use tokio::time::Instant; -use tracing::{debug, error}; +use tracing::{debug, error, instrument, Instrument}; pub(crate) mod final_hop; @@ -164,6 +164,7 @@ impl SharedData { } } + #[instrument(skip_all)] pub(super) fn try_handle_connection( &self, accepted: io::Result<(TcpStream, SocketAddr)>, @@ -173,7 +174,7 @@ impl SharedData { debug!("accepted incoming mixnet connection from: {remote_addr}"); let mut handler = ConnectionHandler::new(self, remote_addr); let join_handle = - tokio::spawn(async move { handler.handle_connection(socket).await }); + tokio::spawn(async move { handler.handle_connection(socket).in_current_span().await }); self.log_connected_clients(); Some(join_handle) } diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index 5ecdf80862..16612adc44 100644 --- a/nym-node/src/node/mod.rs +++ b/nym-node/src/node/mod.rs @@ -66,7 +66,7 @@ use std::ops::Deref; use std::path::Path; use std::sync::Arc; use tokio::sync::mpsc; -use tracing::{debug, info, instrument, trace}; +use tracing::{debug, info, instrument, Instrument, trace}; use zeroize::Zeroizing; pub mod bonding_information; @@ -638,7 +638,7 @@ impl NymNode { .build_websocket_listener(active_clients_store.clone()) .await?; self.shutdown_tracker() - .try_spawn_named(async move { websocket.run().await }, "EntryWebsocket"); + .try_spawn_named(async move { websocket.run().in_current_span().await }, "EntryWebsocket"); } else { info!("node not running in entry mode: the websocket will remain closed"); } @@ -681,7 +681,9 @@ impl NymNode { let authenticator = gateway_tasks_builder .build_wireguard_authenticator(topology_provider) .await?; - let started_authenticator = authenticator.start_service_provider().await?; + let started_authenticator = authenticator.start_service_provider() + .in_current_span() + .await?; active_clients_store.insert_embedded(started_authenticator.handle); info!( @@ -691,6 +693,7 @@ impl NymNode { gateway_tasks_builder .try_start_wireguard() + .in_current_span() .await .map_err(NymNodeError::GatewayTasksStartupFailure)?; } else { @@ -1119,7 +1122,7 @@ impl NymNode { let shutdown_token = self.shutdown_token(); self.shutdown_tracker().try_spawn_named( - async move { mixnet_listener.run(shutdown_token).await }, + async move { mixnet_listener.run(shutdown_token).in_current_span().await }, "MixnetListener", ); @@ -1159,7 +1162,7 @@ impl NymNode { ); debug!("config: {:#?}", self.config); - let http_server = self.build_http_server().await?; + let http_server = self.build_http_server().in_current_span().await?; let bind_address = self.config.http.bind_address; let server_shutdown = self.shutdown_manager.clone_shutdown_token(); @@ -1198,6 +1201,7 @@ impl NymNode { network_refresher.routing_filter(), noise_config, ) + .in_current_span() .await?; let metrics_sender = self.setup_metrics_backend( @@ -1211,6 +1215,7 @@ impl NymNode { active_clients_store, mix_packet_sender, ) + .in_current_span() .await?; self.setup_key_rotation(nym_apis_client, bloomfilters_manager) diff --git a/sdk/rust/nym-sdk/examples/surb_reply.rs b/sdk/rust/nym-sdk/examples/surb_reply.rs index 2e4cc4bd5a..c0c1f388ab 100644 --- a/sdk/rust/nym-sdk/examples/surb_reply.rs +++ b/sdk/rust/nym-sdk/examples/surb_reply.rs @@ -6,34 +6,106 @@ use nym_sdk::DebugConfig; use opentelemetry::trace::{TraceContextExt, Tracer}; #[cfg(feature = "otel")] use opentelemetry::{global, Context}; +#[cfg(feature = "otel")] +use tracing_opentelemetry::OpenTelemetrySpanExt; use tracing::warn; use tracing::instrument; +#[cfg(feature = "otel")] +use tracing::Instrument; #[tokio::main] #[instrument(name = "sdk-example-surb-reply", skip_all)] async fn main() { // Setup OpenTelemetry tracing #[cfg(feature = "otel")] - let _guard = { - nym_bin_common::opentelemetry::setup_tracing_logger("sdk-example-surb-reply".to_string()).unwrap(); + { + let _guard = nym_bin_common::opentelemetry::setup_tracing_logger("sdk-example-surb-reply".to_string()).unwrap(); + let main_span = tracing::info_span!("startup", service = "sdk-example-surb-reply"); + async { + let tracing_context = tracing::Span::current(); + tracing::info!("Current tracing context: {:?}", tracing_context); + let cx = tracing::Span::current().context(); + let sc = cx.span(); + let spcx = sc.span_context(); + tracing::info!("Current OTEL context: {:?}, trace_id: {:?}", cx, spcx.trace_id()); - let tracer = global::tracer("sdk-example-surb-reply"); - let span = tracer.start("client-root-span"); - let cx = Context::current_with_span(span); - let _guard = cx.clone().attach(); + // Ignore performance requirements for the sake of the example + let mut debug_config = DebugConfig::default(); + debug_config.cover_traffic.disable_loop_cover_traffic_stream = true; + debug_config + .traffic + .disable_main_poisson_packet_distribution = true; - let trace_id = cx.span().span_context().trace_id(); - warn!("Main TRACE_ID: {:?}", trace_id); + debug_config.topology.minimum_mixnode_performance = 0; + debug_config.topology.minimum_gateway_performance = 0; - let context = Context::current(); - println!("Current OTEL context: {:?}", context); - - _guard - }; + // Create a mixnet client which connect to a specific node + let client_builder = MixnetClientBuilder::new_ephemeral(); + let mixnet_client = client_builder + .debug_config(debug_config) + .request_gateway("FtR9Mb9y9EViYU3at6Qf7MzNHaMw8gofMicwqoscMBMP".to_string()) + .build() + .unwrap(); + + // Now we connect to the mixnet, using keys now stored in the paths provided. + // let mut client = client.connect_to_mixnet().await.unwrap(); + + let mut client = mixnet_client.connect_to_mixnet().await.unwrap(); + + // Be able to get our client address + let our_address = client.nym_address(); + println!("\nOur client nym address is: {our_address}"); + + // Send a message through the mixnet to ourselves using our nym address + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); + + // we're going to parse the sender_tag (AnonymousSenderTag) from the incoming message and use it to 'reply' to ourselves instead of our Nym address. + // we know there will be a sender_tag since the sdk sends SURBs along with messages by default. + println!("Waiting for message\n"); + + // get the actual message - discard the empty vec sent along with a potential SURB topup request + let mut message: Vec = Vec::new(); + while let Some(new_message) = client.wait_for_messages().await { + if new_message.is_empty() { + continue; + } + message = new_message; + break; + } + + let mut parsed = String::new(); + if let Some(r) = message.first() { + parsed = String::from_utf8(r.message.clone()).unwrap(); + } + // parse sender_tag: we will use this to reply to sender without needing their Nym address + let return_recipient: AnonymousSenderTag = message[0].sender_tag.unwrap(); + println!( + "\nReceived the following message: {parsed} \nfrom sender with surb bucket {return_recipient}" + ); + + // reply to self with it: note we use `send_str_reply` instead of `send_str` + println!("Replying with using SURBs"); + client + .send_reply(return_recipient, "hi an0n!") + .await + .unwrap(); + + println!("Waiting for message (once you see it, ctrl-c to exit)\n"); + client + .on_messages(|msg| println!("\nReceived: {}", String::from_utf8_lossy(&msg.message))) + .await; + }.instrument(main_span).await; + } + + // due to the way instrumentation works in async contexts, the totality of the async code we want to trace needs to be under the same block + // This is ugly and unfortunate but cannot think of another way around it right now. #[cfg(not(feature = "otel"))] - nym_bin_common::logging::setup_no_otel_logger().expect("failed to initialize logging"); - - // Ignore performance requirements for the sake of the example + { + nym_bin_common::logging::setup_no_otel_logger().expect("failed to initialize logging"); + let mut debug_config = DebugConfig::default(); debug_config.cover_traffic.disable_loop_cover_traffic_stream = true; debug_config @@ -43,62 +115,145 @@ async fn main() { debug_config.topology.minimum_mixnode_performance = 0; debug_config.topology.minimum_gateway_performance = 0; - // Create a mixnet client which connect to a specific node - let client_builder = MixnetClientBuilder::new_ephemeral(); - let mixnet_client = client_builder - .debug_config(debug_config) - .request_gateway("FtR9Mb9y9EViYU3at6Qf7MzNHaMw8gofMicwqoscMBMP".to_string()) - .build() - .unwrap(); + let client_builder = MixnetClientBuilder::new_ephemeral(); + let mixnet_client = client_builder + .debug_config(debug_config) + .request_gateway("FtR9Mb9y9EViYU3at6Qf7MzNHaMw8gofMicwqoscMBMP".to_string()) + .build() + .unwrap(); - // Now we connect to the mixnet, using keys now stored in the paths provided. - // let mut client = client.connect_to_mixnet().await.unwrap(); + let mut client = mixnet_client.connect_to_mixnet().await.unwrap(); - let mut client = mixnet_client.connect_to_mixnet().await.unwrap(); + let our_address = client.nym_address(); + println!("\nOur client nym address is: {our_address}"); - // Be able to get our client address - let our_address = client.nym_address(); - println!("\nOur client nym address is: {our_address}"); + client + .send_plain_message(*our_address, "hello there") + .await + .unwrap(); + + println!("Waiting for message\n"); - // Send a message through the mixnet to ourselves using our nym address - client - .send_plain_message(*our_address, "hello there") - .await - .unwrap(); - - // we're going to parse the sender_tag (AnonymousSenderTag) from the incoming message and use it to 'reply' to ourselves instead of our Nym address. - // we know there will be a sender_tag since the sdk sends SURBs along with messages by default. - println!("Waiting for message\n"); - - // get the actual message - discard the empty vec sent along with a potential SURB topup request - let mut message: Vec = Vec::new(); - while let Some(new_message) = client.wait_for_messages().await { - if new_message.is_empty() { - continue; + let mut message: Vec = Vec::new(); + while let Some(new_message) = client.wait_for_messages().await { + if new_message.is_empty() { + continue; + } + message = new_message; + break; } - message = new_message; - break; + + let mut parsed = String::new(); + if let Some(r) = message.first() { + parsed = String::from_utf8(r.message.clone()).unwrap(); + } + let return_recipient: AnonymousSenderTag = message[0].sender_tag.unwrap(); + println!( + "\nReceived the following message: {parsed} \nfrom sender with surb bucket {return_recipient}" + ); + + println!("Replying with using SURBs"); + client + .send_reply(return_recipient, "hi an0n!") + .await + .unwrap(); + + println!("Waiting for message (once you see it, ctrl-c to exit)\n"); + client + .on_messages(|msg| println!("\nReceived: {}", String::from_utf8_lossy(&msg.message))) + .await; } - - let mut parsed = String::new(); - if let Some(r) = message.first() { - parsed = String::from_utf8(r.message.clone()).unwrap(); - } - // parse sender_tag: we will use this to reply to sender without needing their Nym address - let return_recipient: AnonymousSenderTag = message[0].sender_tag.unwrap(); - println!( - "\nReceived the following message: {parsed} \nfrom sender with surb bucket {return_recipient}" - ); - - // reply to self with it: note we use `send_str_reply` instead of `send_str` - println!("Replying with using SURBs"); - client - .send_reply(return_recipient, "hi an0n!") - .await - .unwrap(); - - println!("Waiting for message (once you see it, ctrl-c to exit)\n"); - client - .on_messages(|msg| println!("\nReceived: {}", String::from_utf8_lossy(&msg.message))) - .await; } + + +// #[tokio::main] +// #[instrument(name = "sdk-example-surb-reply", skip_all)] +// async fn main() { +// // Setup OpenTelemetry tracing +// #[cfg(feature = "otel")] +// let _guard = nym_bin_common::opentelemetry::setup_tracing_logger("sdk-example-surb-reply".to_string()).unwrap(); +// #[cfg(feature = "otel")] +// let main_span = tracing::info_span!("startup", service = "sdk-example-surb-reply"); +// #[cfg(feature = "otel")] +// let _main_span_guard = main_span.enter(); +// #[cfg(feature = "otel")] +// let tracing_context = tracing::Span::current(); +// #[cfg(feature = "otel")] +// tracing::info!("Current tracing context: {:?}", tracing_context); +// #[cfg(feature = "otel")] +// let otel_context = Context::current(); +// #[cfg(feature = "otel")] +// tracing::info!("Current OTEL context: {:?}", otel_context); + +// #[cfg(not(feature = "otel"))] +// nym_bin_common::logging::setup_no_otel_logger().expect("failed to initialize logging"); + +// // Ignore performance requirements for the sake of the example +// let mut debug_config = DebugConfig::default(); +// debug_config.cover_traffic.disable_loop_cover_traffic_stream = true; +// debug_config +// .traffic +// .disable_main_poisson_packet_distribution = true; + +// debug_config.topology.minimum_mixnode_performance = 0; +// debug_config.topology.minimum_gateway_performance = 0; + +// // Create a mixnet client which connect to a specific node +// let client_builder = MixnetClientBuilder::new_ephemeral(); +// let mixnet_client = client_builder +// .debug_config(debug_config) +// .request_gateway("FtR9Mb9y9EViYU3at6Qf7MzNHaMw8gofMicwqoscMBMP".to_string()) +// .build() +// .unwrap(); + +// // Now we connect to the mixnet, using keys now stored in the paths provided. +// // let mut client = client.connect_to_mixnet().await.unwrap(); + +// let mut client = mixnet_client.connect_to_mixnet().await.unwrap(); + +// // Be able to get our client address +// let our_address = client.nym_address(); +// println!("\nOur client nym address is: {our_address}"); + +// // Send a message through the mixnet to ourselves using our nym address +// client +// .send_plain_message(*our_address, "hello there") +// .await +// .unwrap(); + +// // we're going to parse the sender_tag (AnonymousSenderTag) from the incoming message and use it to 'reply' to ourselves instead of our Nym address. +// // we know there will be a sender_tag since the sdk sends SURBs along with messages by default. +// println!("Waiting for message\n"); + +// // get the actual message - discard the empty vec sent along with a potential SURB topup request +// let mut message: Vec = Vec::new(); +// while let Some(new_message) = client.wait_for_messages().await { +// if new_message.is_empty() { +// continue; +// } +// message = new_message; +// break; +// } + +// let mut parsed = String::new(); +// if let Some(r) = message.first() { +// parsed = String::from_utf8(r.message.clone()).unwrap(); +// } +// // parse sender_tag: we will use this to reply to sender without needing their Nym address +// let return_recipient: AnonymousSenderTag = message[0].sender_tag.unwrap(); +// println!( +// "\nReceived the following message: {parsed} \nfrom sender with surb bucket {return_recipient}" +// ); + +// // reply to self with it: note we use `send_str_reply` instead of `send_str` +// println!("Replying with using SURBs"); +// client +// .send_reply(return_recipient, "hi an0n!") +// .await +// .unwrap(); + +// println!("Waiting for message (once you see it, ctrl-c to exit)\n"); +// client +// .on_messages(|msg| println!("\nReceived: {}", String::from_utf8_lossy(&msg.message))) +// .await; +// } diff --git a/sdk/rust/nym-sdk/src/mixnet/traits.rs b/sdk/rust/nym-sdk/src/mixnet/traits.rs index af40b37300..7c52d0f662 100644 --- a/sdk/rust/nym-sdk/src/mixnet/traits.rs +++ b/sdk/rust/nym-sdk/src/mixnet/traits.rs @@ -5,15 +5,13 @@ use crate::mixnet::{AnonymousSenderTag, IncludedSurbs, Recipient}; use crate::Result; use async_trait::async_trait; use nym_client_core::client::inbound_messages::InputMessage; -#[cfg(feature = "otel" -)] -use nym_bin_common::opentelemetry::compact_id_generator::compress_trace_id; +#[cfg(feature = "otel")] +use nym_bin_common::opentelemetry::{ + compact_id_generator::compress_trace_id, + context::extract_trace_id_from_tracing_cx, +}; use nym_sphinx::params::PacketType; use nym_task::connections::TransmissionLane; -#[cfg(feature = "otel")] -use opentelemetry::trace::TraceContextExt; -#[cfg(feature = "otel")] -use opentelemetry::Context; use tracing::instrument; // defined to guarantee common interface regardless of whether you're using the full client @@ -83,8 +81,8 @@ pub trait MixnetMessageSender { // in the 12 bytes format #[cfg(feature = "otel")] let trace_id = { - let context = Context::current(); - let trace_id = context.span().span_context().trace_id(); + let trace_id = extract_trace_id_from_tracing_cx(); + tracing::info!("[DEBUG] Trace id in send_message: {:?}", trace_id); Some(compress_trace_id(&trace_id)) }; #[cfg(not(feature = "otel"))]