From 89329ab3eeaca85a865f34be193d7134bb7cb9dd Mon Sep 17 00:00:00 2001 From: Floriane TUERNAL SABOTINOV Date: Fri, 29 Aug 2025 13:23:44 +0200 Subject: [PATCH] add context propagation --- Cargo.lock | 9 +++ gateway/Cargo.toml | 3 + .../websocket/connection_handler/fresh.rs | 59 ++++++++++++++++++- sdk/rust/nym-sdk/Cargo.toml | 14 ++++- sdk/rust/nym-sdk/examples/surb_reply.rs | 30 +++++++++- 5 files changed, 109 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b280f59850..49d629ba73 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5711,6 +5711,8 @@ dependencies = [ "nym-wireguard", "nym-wireguard-private-metadata-server", "nym-wireguard-types", + "opentelemetry", + "opentelemetry_sdk", "rand 0.8.5", "serde", "sha2 0.10.9", @@ -5721,6 +5723,7 @@ dependencies = [ "tokio-tungstenite", "tokio-util", "tracing", + "tracing-opentelemetry", "url", "zeroize", ] @@ -6733,6 +6736,10 @@ dependencies = [ "nym-task", "nym-topology", "nym-validator-client", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry-semantic-conventions", + "opentelemetry_sdk", "parking_lot", "rand 0.8.5", "reqwest 0.12.24", @@ -6746,6 +6753,8 @@ dependencies = [ "tokio-util", "toml 0.8.23", "tracing", + "tracing-core", + "tracing-opentelemetry", "tracing-subscriber", "url", "uuid", diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index 1e4a5be417..2a7607195b 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -82,6 +82,9 @@ nym-service-provider-requests-common = { path = "../common/service-provider-requ defguard_wireguard_rs = { workspace = true } +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true +tracing-opentelemetry.workspace = true [dev-dependencies] nym-gateway-storage = { path = "../common/gateway-storage", features = ["mock"] } 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 e14c9a9a7c..78d0426cc1 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs @@ -33,6 +33,8 @@ use nym_gateway_storage::traits::SharedKeyGatewayStorage; use nym_node_metrics::events::MetricsEvent; use nym_sphinx::DestinationAddressBytes; use nym_task::ShutdownToken; +use opentelemetry::trace::TraceContextExt; +use opentelemetry_sdk::trace::{IdGenerator, RandomIdGenerator}; use rand::CryptoRng; use std::net::SocketAddr; use std::time::Duration; @@ -41,7 +43,8 @@ use time::OffsetDateTime; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::time::timeout; use tokio_tungstenite::tungstenite::{protocol::Message, Error as WsError}; -use tracing::*; +use tracing::{debug, error, info, info_span, instrument, warn}; +use tracing_opentelemetry::OpenTelemetrySpanExt; #[derive(Debug, Error)] pub(crate) enum InitialAuthenticationError { @@ -848,7 +851,7 @@ impl FreshHandler { debug!("failed to reply with protocol version: {err}") } } -#[instrument(skip_all)] + #[instrument(skip_all)] pub(crate) async fn handle_initial_client_request( &mut self, request: ClientControlRequest, @@ -857,6 +860,58 @@ impl FreshHandler { S: AsyncRead + AsyncWrite + Unpin + Send, R: CryptoRng + RngCore + Send, { + let span = if let ClientControlRequest::AuthenticateV2(ref auth_req) = request { + if let Some(ref trace_id) = auth_req.debug_trace_id { + warn!("RAW TRACE ID: {trace_id:?}"); + let trace_id = opentelemetry::trace::TraceId::from_hex(&trace_id) + .expect("Invalid trace ID format"); + warn!("🫂TraceID: {trace_id}🫂"); + + // We don't need to try and preserve the SpanID, just the TraceID (right?) so + // just making a new SpanID for the moment + let id_generator = RandomIdGenerator::default(); + let span_id = id_generator.new_span_id(); + + let span_context = opentelemetry::trace::SpanContext::new( + trace_id, + span_id, + opentelemetry::trace::TraceFlags::SAMPLED, + true, // is_remote = true since this comes from another service + Default::default(), + ); + + let remote_context = + opentelemetry::Context::current().with_remote_span_context(span_context); + + let _context_guard = remote_context.clone().attach(); + let span = info_span!( + "authenticate_v2", + trace_id = %trace_id + ); + span.set_parent(remote_context.clone()); + + Some(span) + } else { + warn!("AuthenticateV2 request but no trace_id provided"); + None + } + } else { + warn!("Not an AuthenticateV2 request"); + None + }; + + // Probably a nicer way to do this but for now just match + let _guard = match &span { + Some(s) => { + warn!("ENTERED SPAN"); + Some(s.enter()) + } + None => { + warn!("COULDN'T ENTER SPAN"); + None + } + }; + // we can handle stateless client requests without prior authentication, like `ClientControlRequest::SupportedProtocol` let auth_result = match request { ClientControlRequest::Authenticate { diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index 523e5801c5..545ee92ee1 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/sdk/rust/nym-sdk/Cargo.toml @@ -71,10 +71,20 @@ tokio-util.workspace = true uuid = { workspace = true, features = ["v4", "serde"] } bincode = { workspace = true } serde = { workspace = true, features = ["derive"] } -tracing.workspace = true -tracing-subscriber = { workspace = true, features = ["env-filter"] } +# tracing.workspace = true +# tracing-subscriber = { workspace = true, features = ["env-filter"] } dirs.workspace = true +opentelemetry = { workspace = true } +opentelemetry_sdk = { workspace = true } + +tracing = { workspace = true, features = ["std"] } +tracing-subscriber = { workspace = true, features = ["registry", "std"] } +tracing-core = { workspace = true } +tracing-opentelemetry = { workspace = true } +opentelemetry-otlp.workspace = true +opentelemetry-semantic-conventions.workspace = true + [dev-dependencies] anyhow = { workspace = true } dotenvy = { workspace = true } diff --git a/sdk/rust/nym-sdk/examples/surb_reply.rs b/sdk/rust/nym-sdk/examples/surb_reply.rs index f0c3858ccd..8b8c1a6c4f 100644 --- a/sdk/rust/nym-sdk/examples/surb_reply.rs +++ b/sdk/rust/nym-sdk/examples/surb_reply.rs @@ -2,14 +2,40 @@ use nym_sdk::mixnet::{ AnonymousSenderTag, MixnetClientBuilder, MixnetMessageSender, ReconstructedMessage, // StoragePaths, }; -use tracing::instrument; +use opentelemetry::trace::TraceContextExt; +use opentelemetry::trace::Tracer; +use opentelemetry::Context; +use opentelemetry::global; +use tracing::warn; +use tracing::{instrument, info_span}; // use std::path::PathBuf; // use tempfile::TempDir; #[tokio::main] #[instrument(name = "sdk-example-surb-reply", skip_all)] async fn main() { - 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 tracer = global::tracer("sdk-example-surb-reply"); + let span = tracer.start("test_span"); + let cx = Context::current_with_span(span); + let _guard = cx.clone().attach(); + + let trace_id = cx.span().span_context().trace_id(); + warn!("Main trace ID: {}", trace_id); + + let span = info_span!( + "surb_reply_example session", + trace_id = %trace_id.to_string() + ); + let _enter = span.enter(); + + let otel_context = opentelemetry::Context::current(); + warn!("OTEL CONTEXT: {:?}", otel_context); + let span = otel_context.span(); + let context = span.span_context(); + let trace_id = context.trace_id(); + warn!("TRACE_ID: {:?}", trace_id); // // Specify some config options // let config_dir: PathBuf = TempDir::new().unwrap().path().to_path_buf();