Merge branch 'develop' of github.com:near/mpc-recovery into phuong/fe…

…at/secure-p2p

integration-tests/src/env/local.rs

@@ -1,14 +1,13 @@
+use crate::env::{LeaderNodeApi, SignerNodeApi};
+use crate::mpc::{self, NodeProcess};
+use crate::util;
 use aes_gcm::aead::consts::U32;
 use aes_gcm::aead::generic_array::GenericArray;
 use mpc_recovery::firewall::allowed::DelegateActionRelayer;
 use mpc_recovery::logging;
 use mpc_recovery::relayer::NearRpcAndRelayerClient;
 use multi_party_eddsa::protocols::ExpandedKeyPair;
 
-use crate::env::{LeaderNodeApi, SignerNodeApi};
-use crate::mpc::{self, NodeProcess};
-use crate::util;
-
 pub struct SignerNode {
     pub address: String,
     env: String,
@@ -30,7 +29,7 @@ impl SignerNode {
         cipher_key: &GenericArray<u8, U32>,
     ) -> anyhow::Result<Self> {
         let web_port = util::pick_unused_port().await?;
-        let args = mpc_recovery::Cli::StartSign {
+        let cli = mpc_recovery::Cli::StartSign {
             env: ctx.env.clone(),
             node_id,
             web_port,
@@ -43,7 +42,7 @@ impl SignerNode {
         };
 
         let sign_node_id = format!("sign-{node_id}");
-        let process = mpc::spawn(ctx.release, &sign_node_id, args).await?;
+        let process = mpc::spawn(ctx.release, &sign_node_id, cli).await?;
         let address = format!("http://127.0.0.1:{web_port}");
         tracing::info!("Signer node is starting at {}", address);
         util::ping_until_ok(&address, 60).await?;
@@ -88,7 +87,7 @@ impl LeaderNode {
         tracing::info!("Running leader node...");
         let account_creator = &ctx.relayer_ctx.creator_account;
         let web_port = util::pick_unused_port().await?;
-        let args = mpc_recovery::Cli::StartLeader {
+        let cli = mpc_recovery::Cli::StartLeader {
             env: ctx.env.clone(),
             web_port,
             sign_nodes,
@@ -123,7 +122,7 @@ impl LeaderNode {
             logging_options: logging::Options::default(),
         };
 
-        let process = mpc::spawn(ctx.release, "leader", args).await?;
+        let process = mpc::spawn(ctx.release, "leader", cli).await?;
         let address = format!("http://127.0.0.1:{web_port}");
         tracing::info!("Leader node container is starting at {}", address);
         util::ping_until_ok(&address, 60).await?;

mpc-recovery/Cargo.toml

@@ -15,6 +15,7 @@ async-trait = "0.1"
 atty = "0.2"
 axum = "0.6.19"
 axum-extra = "0.7"
+axum-tracing-opentelemetry = "0.14.1"
 base64 = "0.21"
 borsh = "0.10.3"
 chrono = "0.4.24"

mpc-recovery/src/leader_node/mod.rs

@@ -26,15 +26,14 @@ use axum::{
     Extension, Json, Router,
 };
 use axum_extra::extract::WithRejection;
+use axum_tracing_opentelemetry::middleware::{OtelAxumLayer, OtelInResponseLayer};
 use borsh::BorshDeserialize;
 use curv::elliptic::curves::{Ed25519, Point};
-use prometheus::{Encoder, TextEncoder};
-
 use near_fetch::signer::KeyRotatingSigner;
 use near_primitives::delegate_action::{DelegateAction, NonDelegateAction};
 use near_primitives::transaction::{Action, DeleteKeyAction};
 use near_primitives::types::AccountId;
-
+use prometheus::{Encoder, TextEncoder};
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Instant;
@@ -116,7 +115,11 @@ pub async fn run(config: Config) {
         .route("/metrics", get(metrics))
         .route_layer(middleware::from_fn(track_metrics))
         .layer(Extension(state))
-        .layer(cors_layer);
+        .layer(cors_layer)
+        // Include trace context as header into the response
+        .layer(OtelInResponseLayer)
+        // Start OpenTelemetry trace on incoming request
+        .layer(OtelAxumLayer::default());
 
     let addr = SocketAddr::from(([0, 0, 0, 0], port));
     tracing::debug!(?addr, "starting http server");

mpc-recovery/src/logging.rs

@@ -1,3 +1,4 @@
+use opentelemetry::sdk::propagation::TraceContextPropagator;
 use opentelemetry::sdk::trace::{self, RandomIdGenerator, Sampler, Tracer};
 use opentelemetry::sdk::Resource;
 use opentelemetry::KeyValue;
@@ -8,13 +9,13 @@ use std::sync::OnceLock;
 use tracing::subscriber::DefaultGuard;
 use tracing_appender::non_blocking::NonBlocking;
 use tracing_opentelemetry::OpenTelemetryLayer;
-use tracing_subscriber::filter::{Filtered, LevelFilter};
+use tracing_subscriber::filter::Filtered;
 use tracing_subscriber::layer::{Layered, SubscriberExt};
 use tracing_subscriber::registry::LookupSpan;
 use tracing_subscriber::{fmt, reload, EnvFilter, Layer, Registry};
 
 static LOG_LAYER_RELOAD_HANDLE: OnceLock<reload::Handle<EnvFilter, Registry>> = OnceLock::new();
-static OTLP_LAYER_RELOAD_HANDLE: OnceLock<reload::Handle<LevelFilter, LogLayer<Registry>>> =
+static OTLP_LAYER_RELOAD_HANDLE: OnceLock<reload::Handle<EnvFilter, LogLayer<Registry>>> =
     OnceLock::new();
 
 type LogLayer<Inner> = Layered<
@@ -27,7 +28,7 @@ type LogLayer<Inner> = Layered<
 >;
 
 type TracingLayer<Inner> = Layered<
-    Filtered<OpenTelemetryLayer<Inner, Tracer>, reload::Layer<LevelFilter, Inner>, Inner>,
+    Filtered<OpenTelemetryLayer<Inner, Tracer>, reload::Layer<EnvFilter, Inner>, Inner>,
     Inner,
 >;
 
@@ -88,24 +89,24 @@ pub struct Options {
         value_enum,
         default_value = "off"
     )]
-    opentelemetry_level: OpenTelemetryLevel,
+    pub opentelemetry_level: OpenTelemetryLevel,
 
     /// Opentelemetry gRPC collector endpoint.
     #[clap(
         long,
         env("MPC_RECOVERY_OTLP_ENDPOINT"),
         default_value = "http://localhost:4317"
     )]
-    otlp_endpoint: String,
+    pub otlp_endpoint: String,
 
     /// Whether the log needs to be colored.
     #[clap(long, value_enum, default_value = "auto")]
-    color: ColorOutput,
+    pub color: ColorOutput,
 
     /// Enable logging of spans. For instance, this prints timestamps of entering and exiting a span,
     /// together with the span duration and used/idle CPU time.
     #[clap(long)]
-    log_span_events: bool,
+    pub log_span_events: bool,
 }
 
 impl Default for Options {
@@ -191,17 +192,22 @@ async fn add_opentelemetry_layer<S>(
     env: String,
     node_id: String,
     subscriber: S,
-) -> (TracingLayer<S>, reload::Handle<LevelFilter, S>)
+) -> (TracingLayer<S>, reload::Handle<EnvFilter, S>)
 where
     S: tracing::Subscriber + for<'span> LookupSpan<'span> + Send + Sync,
 {
     let filter = match opentelemetry_level {
-        OpenTelemetryLevel::OFF => LevelFilter::OFF,
-        OpenTelemetryLevel::INFO => LevelFilter::INFO,
-        OpenTelemetryLevel::DEBUG => LevelFilter::DEBUG,
-        OpenTelemetryLevel::TRACE => LevelFilter::TRACE,
+        OpenTelemetryLevel::OFF => EnvFilter::new("off"),
+        OpenTelemetryLevel::INFO => EnvFilter::new("info"),
+        OpenTelemetryLevel::DEBUG => EnvFilter::new("debug"),
+        OpenTelemetryLevel::TRACE => EnvFilter::new("trace"),
     };
-    let (filter, handle) = reload::Layer::<LevelFilter, S>::new(filter);
+    // `otel::tracing` should be a level info to emit opentelemetry trace & span
+    // `otel::setup` set to debug to log detected resources, configuration read and infered
+    let filter = filter
+        .add_directive("otel::tracing=trace".parse().unwrap())
+        .add_directive("otel=debug".parse().unwrap());
+    let (filter, handle) = reload::Layer::<EnvFilter, S>::new(filter);
 
     let resource = vec![
         KeyValue::new(SERVICE_NAME, format!("mpc:{}:{}", env, node_id)),
@@ -224,12 +230,18 @@ where
         )
         .install_batch(opentelemetry::runtime::Tokio)
         .unwrap();
+
+    init_propagator();
     let layer = tracing_opentelemetry::layer()
         .with_tracer(tracer)
         .with_filter(filter);
     (subscriber.with(layer), handle)
 }
 
+pub fn init_propagator() {
+    opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new());
+}
+
 fn set_default_otlp_level(options: &Options) {
     // Record the initial tracing level specified as a command-line flag. Use this recorded value to
     // reset opentelemetry filter when the LogConfig file gets deleted.