From dce4d6b34bc9708710e728338e03f96a5e32f97f Mon Sep 17 00:00:00 2001 From: Tommy Verrall Date: Mon, 16 Feb 2026 19:13:11 +0100 Subject: [PATCH] otel: refactor key selection, add environment label, fix clippy --- common/bin-common/src/logging/mod.rs | 11 +- common/nym-lp/src/message.rs | 4 +- docker/localnet/Dockerfile.localnet | 45 ++--- docker/localnet/loadtest.sh | 6 +- nym-node/src/cli/mod.rs | 38 ++-- nym-node/src/logging.rs | 31 +++- nym-node/src/node/mixnet/handler.rs | 165 ++++++++---------- .../src/node/mixnet/packet_forwarding/mod.rs | 2 +- nym-node/src/node/mixnet/shared/final_hop.rs | 9 +- nym-node/src/node/mixnet/shared/mod.rs | 15 +- 10 files changed, 162 insertions(+), 164 deletions(-) diff --git a/common/bin-common/src/logging/mod.rs b/common/bin-common/src/logging/mod.rs index 45f7c5fa85..fe0848cc3e 100644 --- a/common/bin-common/src/logging/mod.rs +++ b/common/bin-common/src/logging/mod.rs @@ -76,11 +76,14 @@ pub fn setup_tracing_logger() { /// or "https://ingest.eu.signoz.cloud:443" for SigNoz Cloud) /// * `ingestion_key` - Optional SigNoz Cloud ingestion key. When provided, it is /// sent as the `signoz-ingestion-key` gRPC metadata header on every export. +/// * `environment` - Deployment environment label (e.g. "sandbox", "mainnet", "canary"). +/// Attached as the `deployment.environment` OTel resource attribute. #[cfg(feature = "otel-otlp")] pub fn init_otel_layer( service_name: &str, endpoint: &str, ingestion_key: Option<&str>, + environment: &str, ) -> Result< ( tracing_opentelemetry::OpenTelemetryLayer, @@ -102,8 +105,8 @@ where // Explicitly configure TLS when the endpoint uses HTTPS if endpoint.starts_with("https://") { - builder = builder - .with_tls_config(tonic::transport::ClientTlsConfig::new().with_native_roots()); + builder = + builder.with_tls_config(tonic::transport::ClientTlsConfig::new().with_native_roots()); } if let Some(key) = ingestion_key { @@ -122,6 +125,10 @@ where .with_resource( opentelemetry_sdk::Resource::builder() .with_service_name(service_name.to_owned()) + .with_attribute(opentelemetry::KeyValue::new( + "deployment.environment", + environment.to_owned(), + )) .build(), ) .build(); diff --git a/common/nym-lp/src/message.rs b/common/nym-lp/src/message.rs index 77f332eec2..aa58476997 100644 --- a/common/nym-lp/src/message.rs +++ b/common/nym-lp/src/message.rs @@ -391,7 +391,9 @@ impl ForwardPacketData { } // SAFETY: we ensured we have sufficient data, and the length is correct for casting #[allow(clippy::unwrap_used)] - let ipv6 = IpAddr::V6(Ipv6Addr::from(<[u8; 16]>::try_from(&bytes[33..49]).unwrap())); + let ipv6 = IpAddr::V6(Ipv6Addr::from( + <[u8; 16]>::try_from(&bytes[33..49]).unwrap(), + )); let port = u16::from_le_bytes([bytes[49], bytes[50]]); (SocketAddr::new(ipv6, port), 51) } else { diff --git a/docker/localnet/Dockerfile.localnet b/docker/localnet/Dockerfile.localnet index 528ea3bf7e..f10f044c92 100644 --- a/docker/localnet/Dockerfile.localnet +++ b/docker/localnet/Dockerfile.localnet @@ -1,8 +1,9 @@ # Multi-stage Dockerfile for Nym localnet -# Stage 1: Build binaries -# Stage 2: Slim runtime with only the final binaries +# Stage 1: Build Rust binaries +# Stage 2: Build wireguard-go (cached separately from Rust builds) +# Stage 3: Slim runtime with only the final binaries -# --- Build stage --- +# --- Rust build stage --- FROM rust:latest AS builder WORKDIR /usr/src/nym @@ -13,12 +14,28 @@ ENV CARGO_BUILD_JOBS=8 RUN cargo build --release --locked -p nym-node --features otel && \ cargo build --release --locked -p nym-network-requester -p nym-socks5-client +# --- wireguard-go build stage (cached independently) --- +FROM debian:trixie-slim AS wireguard-builder + +ARG TARGETARCH + +RUN apt-get update && apt-get install -y --no-install-recommends \ + curl git build-essential ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +RUN curl -fsSL "https://go.dev/dl/go1.23.6.linux-${TARGETARCH}.tar.gz" \ + | tar -C /usr/local -xz + +ENV PATH="/usr/local/go/bin:$PATH" + +RUN git clone https://git.zx2c4.com/wireguard-go && \ + cd wireguard-go && make + # --- Runtime stage --- FROM debian:trixie-slim RUN apt-get update && apt-get install -y --no-install-recommends \ ca-certificates \ - build-essential \ python3 \ python3-pip \ netcat-openbsd \ @@ -26,27 +43,15 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ iproute2 \ net-tools \ wireguard-tools \ - git \ iptables \ - curl \ && rm -rf /var/lib/apt/lists/* -# Install Go and build wireguard-go, then clean up -ARG TARGETARCH -RUN curl -fsSL "https://go.dev/dl/go1.23.6.linux-${TARGETARCH}.tar.gz" \ - | tar -C /usr/local -xz && \ - export PATH="/usr/local/go/bin:$PATH" && \ - git clone https://git.zx2c4.com/wireguard-go && \ - cd wireguard-go && \ - make && \ - cp wireguard-go /usr/local/bin/ && \ - cd .. && \ - rm -rf wireguard-go /usr/local/go && \ - apt-get purge -y --auto-remove build-essential curl - RUN pip3 install --break-system-packages base58 -# Copy only the compiled binaries from the builder stage +# Copy wireguard-go from its dedicated build stage +COPY --from=wireguard-builder /wireguard-go/wireguard-go /usr/local/bin/ + +# Copy only the compiled binaries from the Rust builder stage COPY --from=builder /usr/src/nym/target/release/nym-node /usr/local/bin/ COPY --from=builder /usr/src/nym/target/release/nym-network-requester /usr/local/bin/ COPY --from=builder /usr/src/nym/target/release/nym-socks5-client /usr/local/bin/ diff --git a/docker/localnet/loadtest.sh b/docker/localnet/loadtest.sh index e76b956ad6..fa0eb609e1 100755 --- a/docker/localnet/loadtest.sh +++ b/docker/localnet/loadtest.sh @@ -131,7 +131,11 @@ pick_url() { # Millisecond timestamp (works on both GNU and BSD/macOS date) now_ms() { - python3 -c 'import time; print(int(time.time()*1000))' + if command -v python3 &>/dev/null; then + python3 -c 'import time; print(int(time.time()*1000))' + else + echo $(($(date +%s) * 1000)) + fi } # Worker function: runs requests in a loop until duration expires diff --git a/nym-node/src/cli/mod.rs b/nym-node/src/cli/mod.rs index 175b3022f5..be68f15e14 100644 --- a/nym-node/src/cli/mod.rs +++ b/nym-node/src/cli/mod.rs @@ -59,6 +59,11 @@ pub(crate) struct Cli { #[clap(long, env = "NYMNODE_OTEL_KEY")] pub(crate) otel_key: Option, + /// Deployment environment label attached to all exported traces. + /// Used to distinguish sandbox / mainnet / canary in the OTel backend. + #[clap(long, env = "NYMNODE_OTEL_ENV", default_value = "mainnet")] + pub(crate) otel_env: String, + #[clap(subcommand)] command: Commands, } @@ -71,26 +76,19 @@ impl Cli { // Set up tracing inside the runtime so the OTel batch exporter // can spawn its background tasks on the tokio reactor. - let _otel_guard = runtime.block_on(async { - self.setup_logging() - })?; + let _otel_guard = runtime.block_on(async { self.setup_logging() })?; - let result = runtime.block_on(async { + // `_otel_guard` is dropped at function exit, flushing pending spans via its Drop impl + runtime.block_on(async { match self.command { Commands::BuildInfo(args) => build_info::execute(args)?, - Commands::BondingInformation(args) => { - bonding_information::execute(args).await? - } - Commands::NodeDetails(args) => { - node_details::execute(args).await? - } + Commands::BondingInformation(args) => bonding_information::execute(args).await?, + Commands::NodeDetails(args) => node_details::execute(args).await?, Commands::Run(args) => run::execute(*args).await?, Commands::Migrate(args) => migrate::execute(*args)?, Commands::Sign(args) => sign::execute(args).await?, Commands::TestThroughput(args) => test_throughput::execute(args)?, - Commands::UnsafeResetSphinxKeys(args) => { - reset_sphinx_keys::execute(args).await? - } + Commands::UnsafeResetSphinxKeys(args) => reset_sphinx_keys::execute(args).await?, Commands::Debug(debug) => match debug.command { DebugCommands::ResetProvidersGatewayDbs(args) => { debug::reset_providers_dbs::execute(args).await? @@ -98,17 +96,7 @@ impl Cli { }, } Ok::<(), anyhow::Error>(()) - }); - - // Flush any pending OTel spans before exit - #[cfg(feature = "otel")] - if let Some(guard) = _otel_guard { - if let Err(e) = guard.provider.shutdown() { - eprintln!("OpenTelemetry shutdown error: {e}"); - } - } - - result + }) } #[cfg(feature = "otel")] @@ -121,6 +109,7 @@ impl Cli { endpoint: self.otel_endpoint.clone(), service_name: "nym-node".to_string(), ingestion_key: self.otel_key.clone(), + environment: self.otel_env.clone(), }) } else { None @@ -138,6 +127,7 @@ impl Cli { endpoint: self.otel_endpoint.clone(), service_name: "nym-node".to_string(), ingestion_key: self.otel_key.clone(), + environment: self.otel_env.clone(), }) } else { None diff --git a/nym-node/src/logging.rs b/nym-node/src/logging.rs index c36e1d776c..dd14b1b780 100644 --- a/nym-node/src/logging.rs +++ b/nym-node/src/logging.rs @@ -14,14 +14,25 @@ pub(crate) struct OtelConfig { pub endpoint: String, pub service_name: String, pub ingestion_key: Option, + pub environment: String, } -/// Handle returned when OTel is active so the caller can trigger a graceful shutdown. +/// Handle returned when OTel is active. Flushes pending spans on drop +/// to prevent telemetry loss during panics or early exits. #[cfg(feature = "otel")] pub(crate) struct OtelGuard { pub provider: opentelemetry_sdk::trace::SdkTracerProvider, } +#[cfg(feature = "otel")] +impl Drop for OtelGuard { + fn drop(&mut self) { + if let Err(e) = self.provider.shutdown() { + eprintln!("OpenTelemetry shutdown error in Drop: {e}"); + } + } +} + pub(crate) fn granual_filtered_env() -> anyhow::Result { fn directive_checked(directive: impl Into) -> anyhow::Result { directive.into().parse().map_err(From::from) @@ -43,9 +54,7 @@ pub(crate) fn granual_filtered_env() -> anyhow::Result { /// OTLP exporter layer is added and the returned `OtelGuard` must be used to /// flush pending spans on shutdown. #[cfg(feature = "otel")] -pub(crate) fn setup_tracing_logger( - otel: Option, -) -> anyhow::Result> { +pub(crate) fn setup_tracing_logger(otel: Option) -> anyhow::Result> { let stderr_layer = default_tracing_fmt_layer(std::io::stderr).with_filter(granual_filtered_env()?); @@ -57,7 +66,12 @@ pub(crate) fn setup_tracing_logger( &otel_config.service_name, &otel_config.endpoint, otel_config.ingestion_key.as_deref(), - ).map_err(|e| anyhow::anyhow!("failed to initialise OpenTelemetry: {e}"))?; + &otel_config.environment, + ).map_err(|e| anyhow::anyhow!( + "failed to initialise OpenTelemetry exporter (endpoint: {}, service: {}): {e}", + otel_config.endpoint, + otel_config.service_name, + ))?; tracing_subscriber::registry() .with(console_layer) @@ -80,7 +94,12 @@ pub(crate) fn setup_tracing_logger( &otel_config.service_name, &otel_config.endpoint, otel_config.ingestion_key.as_deref(), - ).map_err(|e| anyhow::anyhow!("failed to initialise OpenTelemetry: {e}"))?; + &otel_config.environment, + ).map_err(|e| anyhow::anyhow!( + "failed to initialise OpenTelemetry exporter (endpoint: {}, service: {}): {e}", + otel_config.endpoint, + otel_config.service_name, + ))?; tracing_subscriber::registry() .with(stderr_layer) diff --git a/nym-node/src/node/mixnet/handler.rs b/nym-node/src/node/mixnet/handler.rs index ae0793cebd..e644a2c5d7 100644 --- a/nym-node/src/node/mixnet/handler.rs +++ b/nym-node/src/node/mixnet/handler.rs @@ -1,6 +1,7 @@ // Copyright 2024 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only +use crate::node::key_rotation::active_keys::SphinxKeyGuard; use crate::node::mixnet::shared::SharedData; use futures::StreamExt; use nym_noise::connection::Connection; @@ -20,7 +21,10 @@ use std::net::SocketAddr; use tokio::net::TcpStream; use tokio::time::Instant; use tokio_util::codec::Framed; -use tracing::{debug, error, instrument, trace, warn, Span}; +use tracing::{Span, debug, error, instrument, trace, warn}; + +/// How often (in packets) the stream-level span updates its packet count. +const SPAN_UPDATE_INTERVAL: u64 = 10_000; struct PendingReplayCheckPackets { // map of rotation id used for packet creation to the packets @@ -140,7 +144,7 @@ impl ConnectionHandler { level = "debug", fields( remote_addr = %self.remote_address, - delay_ms, + delay_ms = tracing::field::Empty, ) )] fn handle_forward_packet(&self, now: Instant, mix_packet: MixPacket, delay: Option) { @@ -155,12 +159,12 @@ impl ConnectionHandler { } let forward_instant = self.create_delay_target(now, delay); - Span::current().record( - "delay_ms", - forward_instant - .map(|i| i.saturating_duration_since(now).as_millis() as u64) - .unwrap_or(0), - ); + if let Some(target) = forward_instant { + Span::current().record( + "delay_ms", + target.saturating_duration_since(now).as_millis() as u64, + ); + } self.shared.forward_mix_packet(mix_packet, forward_instant); } @@ -273,29 +277,59 @@ impl ConnectionHandler { time_threshold && count_threshold } - #[instrument( - name = "mixnode.sphinx_partial_unwrap", - skip(self, packet), - level = "debug", - fields( - key_rotation, - unwrap_result, - ) - )] - fn try_partially_unwrap_packet( + /// Resolve the sphinx key for the given rotation, recording the rotation + /// label on the current tracing span. Returns `ExpiredKey` if the requested + /// odd/even key has already been rotated out. + fn resolve_rotation_key( &self, - packet: FramedNymPacket, - ) -> Result { - let rotation_label = match packet.header().key_rotation { + rotation: SphinxKeyRotation, + ) -> Result { + let rotation_label = match rotation { SphinxKeyRotation::Unknown => "unknown", SphinxKeyRotation::OddRotation => "odd", SphinxKeyRotation::EvenRotation => "even", }; Span::current().record("key_rotation", rotation_label); - let result = match packet.header().key_rotation { + match rotation { + SphinxKeyRotation::Unknown => Ok(self.shared.sphinx_keys.primary()), + SphinxKeyRotation::OddRotation => self.shared.sphinx_keys.odd().ok_or_else(|| { + warn!( + event = "packet.dropped.expired_key", + key_rotation = "odd", + remote_addr = %self.remote_address, + "dropping packet: odd key rotation expired" + ); + PacketProcessingError::ExpiredKey + }), + SphinxKeyRotation::EvenRotation => self.shared.sphinx_keys.even().ok_or_else(|| { + warn!( + event = "packet.dropped.expired_key", + key_rotation = "even", + remote_addr = %self.remote_address, + "dropping packet: even key rotation expired" + ); + PacketProcessingError::ExpiredKey + }), + } + } + + #[instrument( + name = "mixnode.sphinx_partial_unwrap", + skip(self, packet), + level = "debug", + fields(key_rotation, unwrap_result,) + )] + fn try_partially_unwrap_packet( + &self, + packet: FramedNymPacket, + ) -> Result { + let rotation = packet.header().key_rotation; + + let result = match rotation { SphinxKeyRotation::Unknown => { - let primary = self.shared.sphinx_keys.primary(); + // Unknown rotation: try primary, fallback to secondary + let primary = self.resolve_rotation_key(rotation)?; let primary_rotation = primary.rotation_id(); match PartiallyUnwrappedPacket::new(packet, primary.inner().as_ref()) { @@ -314,35 +348,12 @@ impl ConnectionHandler { } } } - SphinxKeyRotation::OddRotation => { - let Some(odd_key) = self.shared.sphinx_keys.odd() else { - warn!( - event = "packet.dropped.expired_key", - key_rotation = "odd", - remote_addr = %self.remote_address, - "dropping packet: odd key rotation expired" - ); - return Err(PacketProcessingError::ExpiredKey); - }; - let odd_rotation = odd_key.rotation_id(); - PartiallyUnwrappedPacket::new(packet, odd_key.inner().as_ref()) + _ => { + let key = self.resolve_rotation_key(rotation)?; + let rotation_id = key.rotation_id(); + PartiallyUnwrappedPacket::new(packet, key.inner().as_ref()) .map_err(|(_, err)| err) - .map(|p| p.with_key_rotation(odd_rotation)) - } - SphinxKeyRotation::EvenRotation => { - let Some(even_key) = self.shared.sphinx_keys.even() else { - warn!( - event = "packet.dropped.expired_key", - key_rotation = "even", - remote_addr = %self.remote_address, - "dropping packet: even key rotation expired" - ); - return Err(PacketProcessingError::ExpiredKey); - }; - let even_rotation = even_key.rotation_id(); - PartiallyUnwrappedPacket::new(packet, even_key.inner().as_ref()) - .map_err(|(_, err)| err) - .map(|p| p.with_key_rotation(even_rotation)) + .map(|p| p.with_key_rotation(rotation_id)) } }; @@ -487,10 +498,7 @@ impl ConnectionHandler { name = "mixnode.replay_check_batch", skip(self), level = "debug", - fields( - batch_size, - mutex_wait_ms, - ) + fields(batch_size, mutex_wait_ms,) )] async fn handle_pending_packets_batch(&mut self, now: Instant) { let batch_size = self.pending_packets.total_count(); @@ -513,10 +521,7 @@ impl ConnectionHandler { self.shared.shutdown_token.cancel(); return; }; - Span::current().record( - "mutex_wait_ms", - mutex_start.elapsed().as_millis() as u64, - ); + Span::current().record("mutex_wait_ms", mutex_start.elapsed().as_millis() as u64); self.handle_post_replay_detection_packets(now, batch, replay_check_results) .await; @@ -532,42 +537,8 @@ impl ConnectionHandler { &self, packet: FramedNymPacket, ) -> Result { - let rotation_label = match packet.header().key_rotation { - SphinxKeyRotation::Unknown => "unknown", - SphinxKeyRotation::OddRotation => "odd", - SphinxKeyRotation::EvenRotation => "even", - }; - Span::current().record("key_rotation", rotation_label); - - match packet.header().key_rotation { - SphinxKeyRotation::Unknown => { - process_framed_packet(packet, self.shared.sphinx_keys.primary().inner().as_ref()) - } - SphinxKeyRotation::OddRotation => { - let Some(odd_key) = self.shared.sphinx_keys.odd() else { - warn!( - event = "packet.dropped.expired_key", - key_rotation = "odd", - remote_addr = %self.remote_address, - "dropping packet: odd key rotation expired" - ); - return Err(PacketProcessingError::ExpiredKey); - }; - process_framed_packet(packet, odd_key.inner().as_ref()) - } - SphinxKeyRotation::EvenRotation => { - let Some(even_key) = self.shared.sphinx_keys.even() else { - warn!( - event = "packet.dropped.expired_key", - key_rotation = "even", - remote_addr = %self.remote_address, - "dropping packet: even key rotation expired" - ); - return Err(PacketProcessingError::ExpiredKey); - }; - process_framed_packet(packet, even_key.inner().as_ref()) - } - } + let key = self.resolve_rotation_key(packet.header().key_rotation)?; + process_framed_packet(packet, key.inner().as_ref()) } async fn handle_received_packet_with_no_replay_detection( @@ -602,7 +573,7 @@ impl ConnectionHandler { level = "debug", fields( remote = %self.remote_address, - noise_handshake_ms, + noise_handshake_ms = tracing::field::Empty, ) )] pub(crate) async fn handle_connection(&mut self, socket: TcpStream) { @@ -663,7 +634,7 @@ impl ConnectionHandler { Some(Ok(packet)) => { self.handle_received_nym_packet(packet).await; packets_processed += 1; - if packets_processed % 10000 == 0 { + if packets_processed.is_multiple_of(SPAN_UPDATE_INTERVAL) { Span::current().record("packets_processed", packets_processed); } } diff --git a/nym-node/src/node/mixnet/packet_forwarding/mod.rs b/nym-node/src/node/mixnet/packet_forwarding/mod.rs index 3a887284f2..ef695a05eb 100644 --- a/nym-node/src/node/mixnet/packet_forwarding/mod.rs +++ b/nym-node/src/node/mixnet/packet_forwarding/mod.rs @@ -155,7 +155,7 @@ impl PacketForwarder { self.handle_new_packet(new_packet.unwrap()); let channel_len = self.packet_sender.len(); let delay_queue_len = self.delay_queue.len(); - if processed % 1000 == 0 { + if processed.is_multiple_of(1000) { match channel_len { n if n > 1000 => error!( event = "forwarder.queue_overload", diff --git a/nym-node/src/node/mixnet/shared/final_hop.rs b/nym-node/src/node/mixnet/shared/final_hop.rs index 0f852707f5..0235617342 100644 --- a/nym-node/src/node/mixnet/shared/final_hop.rs +++ b/nym-node/src/node/mixnet/shared/final_hop.rs @@ -38,11 +38,13 @@ impl SharedFinalHopData { Err(message) } Some(sender_channel) => { + let send_start = Instant::now(); if let Err(unsent) = sender_channel.unbounded_send(vec![message]) { warn!( event = "gateway.push_to_client", client_found = true, send_result = "channel_closed", + send_us = send_start.elapsed().as_micros() as u64, "client {client_address} channel closed, message not delivered" ); // the unwrap here is fine as the original message got returned; @@ -54,6 +56,7 @@ impl SharedFinalHopData { event = "gateway.push_to_client", client_found = true, send_result = "ok", + send_us = send_start.elapsed().as_micros() as u64, "pushed message to client {client_address}" ); Ok(()) @@ -74,14 +77,12 @@ impl SharedFinalHopData { if result.is_ok() { debug!( event = "gateway.disk_store", - store_ms, - "stored message for {client_address} on disk in {store_ms}ms" + store_ms, "stored message for {client_address} on disk in {store_ms}ms" ); } else { warn!( event = "gateway.disk_store_failed", - store_ms, - "failed to store message for {client_address} on disk after {store_ms}ms" + store_ms, "failed to store message for {client_address} on disk after {store_ms}ms" ); } result diff --git a/nym-node/src/node/mixnet/shared/mod.rs b/nym-node/src/node/mixnet/shared/mod.rs index 13a664772c..19b15d9116 100644 --- a/nym-node/src/node/mixnet/shared/mod.rs +++ b/nym-node/src/node/mixnet/shared/mod.rs @@ -190,15 +190,14 @@ impl SharedData { .mixnet_forwarder .forward_packet(PacketToForward::new(packet, delay_until)) .is_err() + && !self.shutdown_token.is_cancelled() { - if !self.shutdown_token.is_cancelled() { - error!( - event = "forwarder.channel_send_failed", - has_delay, - "failed to forward sphinx packet on the channel while the process is not going through the shutdown!" - ); - self.shutdown_token.cancel(); - } + error!( + event = "forwarder.channel_send_failed", + has_delay, + "failed to forward sphinx packet on the channel while the process is not going through the shutdown!" + ); + self.shutdown_token.cancel(); } }