merge queue: embarking unstable (a8035d7) and [#7500 + #7459 + #7180 + #7246 + #7052] together

.github/workflows/test-suite.yml

@@ -196,7 +196,7 @@ jobs:
     - name: Run network tests for all known forks
       run: make test-network
       env:
-        TEST_FEATURES: portable,ci_logger
+        TEST_FEATURES: portable
         CI_LOGGER_DIR: ${{ runner.temp }}/network_test_logs
     - name: Upload logs
       uses: actions/upload-artifact@v4

Cargo.toml

@@ -159,7 +159,7 @@ hyper = "1"
 itertools = "0.10"
 libsecp256k1 = "0.7"
 log = "0.4"
-logroller = "0.1.4"
+logroller = "0.1.8"
 lru = "0.12"
 maplit = "1"
 milhouse = "0.5"

beacon_node/beacon_chain/src/test_utils.rs

@@ -69,8 +69,6 @@ use types::{typenum::U4294967296, *};
 pub const HARNESS_GENESIS_TIME: u64 = 1_567_552_690;
 // Environment variable to read if `fork_from_env` feature is enabled.
 pub const FORK_NAME_ENV_VAR: &str = "FORK_NAME";
-// Environment variable to read if `ci_logger` feature is enabled.
-pub const CI_LOGGER_DIR_ENV_VAR: &str = "CI_LOGGER_DIR";
 
 // Pre-computed data column sidecar using a single static blob from:
 // `beacon_node/execution_layer/src/test_utils/fixtures/mainnet/test_blobs_bundle.ssz`

beacon_node/execution_layer/src/engine_api/http.rs

@@ -1242,6 +1242,10 @@ impl HttpJsonRpc {
         } else {
             let engine_version = self.get_client_version_v1().await?;
             *lock = Some(CachedResponse::new(engine_version.clone()));
+            if !engine_version.is_empty() {
+                // reset metric gauge when there's a fresh fetch
+                crate::metrics::reset_execution_layer_info_gauge();
+            }
             Ok(engine_version)
         }
     }

beacon_node/execution_layer/src/lib.rs

@@ -129,8 +129,7 @@ impl<E: EthSpec> TryFrom<BuilderBid<E>> for ProvenancedPayload<BlockProposalCont
                 block_value: builder_bid.value,
                 kzg_commitments: builder_bid.blob_kzg_commitments,
                 blobs_and_proofs: None,
-                // TODO(fulu): update this with builder api returning the requests
-                requests: None,
+                requests: Some(builder_bid.execution_requests),
             },
         };
         Ok(ProvenancedPayload::Builder(
@@ -1555,10 +1554,14 @@ impl<E: EthSpec> ExecutionLayer<E> {
         &self,
         age_limit: Option<Duration>,
     ) -> Result<Vec<ClientVersionV1>, Error> {
-        self.engine()
+        let versions = self
+            .engine()
             .request(|engine| engine.get_engine_version(age_limit))
             .await
-            .map_err(Into::into)
+            .map_err(Into::<Error>::into)?;
+        metrics::expose_execution_layer_info(&versions);
+
+        Ok(versions)
     }
 
     /// Used during block production to determine if the merge has been triggered.

beacon_node/execution_layer/src/metrics.rs

@@ -116,3 +116,29 @@ pub static EXECUTION_LAYER_PAYLOAD_BIDS: LazyLock<Result<IntGaugeVec>> = LazyLoc
         &["source"]
     )
 });
+pub static EXECUTION_LAYER_INFO: LazyLock<Result<IntGaugeVec>> = LazyLock::new(|| {
+    try_create_int_gauge_vec(
+        "execution_layer_info",
+        "The build of the execution layer connected to lighthouse",
+        &["code", "name", "version", "commit"],
+    )
+});
+
+pub fn reset_execution_layer_info_gauge() {
+    let _ = EXECUTION_LAYER_INFO.as_ref().map(|gauge| gauge.reset());
+}
+
+pub fn expose_execution_layer_info(els: &Vec<crate::ClientVersionV1>) {
+    for el in els {
+        set_gauge_vec(
+            &EXECUTION_LAYER_INFO,
+            &[
+                &el.code.to_string(),
+                &el.name,
+                &el.version,
+                &el.commit.to_string(),
+            ],
+            1,
+        );
+    }
+}

beacon_node/lighthouse_network/tests/rpc_tests.rs

@@ -12,7 +12,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::runtime::Runtime;
 use tokio::time::sleep;
-use tracing::{debug, error, warn};
+use tracing::{debug, error, info_span, warn, Instrument};
 use types::{
     BeaconBlock, BeaconBlockAltair, BeaconBlockBase, BeaconBlockBellatrix, BlobSidecar, ChainSpec,
     EmptyBlock, Epoch, EthSpec, FixedBytesExtended, ForkName, Hash256, MinimalEthSpec,
@@ -117,7 +117,8 @@ fn test_tcp_status_rpc() {
                     _ => {}
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // build the receiver future
         let receiver_future = async {
@@ -141,7 +142,8 @@ fn test_tcp_status_rpc() {
                     _ => {} // Ignore other events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
 
         tokio::select! {
             _ = sender_future => {}
@@ -245,7 +247,8 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
                     _ => {} // Ignore other behaviour events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // build the receiver future
         let receiver_future = async {
@@ -286,7 +289,8 @@ fn test_tcp_blocks_by_range_chunked_rpc() {
                     _ => {} // Ignore other events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
 
         tokio::select! {
             _ = sender_future => {}
@@ -373,7 +377,8 @@ fn test_blobs_by_range_chunked_rpc() {
                     _ => {} // Ignore other behaviour events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // build the receiver future
         let receiver_future = async {
@@ -407,7 +412,8 @@ fn test_blobs_by_range_chunked_rpc() {
                     _ => {} // Ignore other events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
 
         tokio::select! {
             _ = sender_future => {}
@@ -479,7 +485,8 @@ fn test_tcp_blocks_by_range_over_limit() {
                     _ => {} // Ignore other behaviour events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // build the receiver future
         let receiver_future = async {
@@ -512,7 +519,8 @@ fn test_tcp_blocks_by_range_over_limit() {
                     _ => {} // Ignore other events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
 
         tokio::select! {
             _ = sender_future => {}
@@ -601,7 +609,8 @@ fn test_tcp_blocks_by_range_chunked_rpc_terminates_correctly() {
                     _ => {} // Ignore other behaviour events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // determine messages to send (PeerId, RequestId). If some, indicates we still need to send
         // messages
@@ -648,7 +657,8 @@ fn test_tcp_blocks_by_range_chunked_rpc_terminates_correctly() {
                     }
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
 
         tokio::select! {
             _ = sender_future => {}
@@ -734,7 +744,8 @@ fn test_tcp_blocks_by_range_single_empty_rpc() {
                     _ => {} // Ignore other behaviour events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // build the receiver future
         let receiver_future = async {
@@ -767,7 +778,8 @@ fn test_tcp_blocks_by_range_single_empty_rpc() {
                     _ => {} // Ignore other events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
         tokio::select! {
             _ = sender_future => {}
             _ = receiver_future => {}
@@ -877,7 +889,8 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
                     _ => {} // Ignore other behaviour events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // build the receiver future
         let receiver_future = async {
@@ -916,7 +929,8 @@ fn test_tcp_blocks_by_root_chunked_rpc() {
                     _ => {} // Ignore other events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
         tokio::select! {
             _ = sender_future => {}
             _ = receiver_future => {}
@@ -1015,7 +1029,8 @@ fn test_tcp_blocks_by_root_chunked_rpc_terminates_correctly() {
                     _ => {} // Ignore other behaviour events
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // determine messages to send (PeerId, RequestId). If some, indicates we still need to send
         // messages
@@ -1062,7 +1077,8 @@ fn test_tcp_blocks_by_root_chunked_rpc_terminates_correctly() {
                     }
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
 
         tokio::select! {
             _ = sender_future => {}
@@ -1115,7 +1131,8 @@ fn goodbye_test(log_level: &str, enable_logging: bool, protocol: Protocol) {
                     _ => {} // Ignore other RPC messages
                 }
             }
-        };
+        }
+        .instrument(info_span!("Sender"));
 
         // build the receiver future
         let receiver_future = async {
@@ -1125,7 +1142,8 @@ fn goodbye_test(log_level: &str, enable_logging: bool, protocol: Protocol) {
                     return;
                 }
             }
-        };
+        }
+        .instrument(info_span!("Receiver"));
 
         let total_future = futures::future::join(sender_future, receiver_future);
 

beacon_node/network/Cargo.toml

@@ -57,4 +57,3 @@ disable-backfill = []
 fork_from_env = ["beacon_chain/fork_from_env"]
 portable = ["beacon_chain/portable"]
 test_logger = []
-ci_logger = []

beacon_node/network/src/sync/tests/lookups.rs

@@ -107,6 +107,8 @@ impl TestRig {
         // deterministic seed
         let rng = ChaCha20Rng::from_seed([0u8; 32]);
 
+        init_tracing();
+
         TestRig {
             beacon_processor_rx,
             beacon_processor_rx_queue: vec![],

beacon_node/network/src/sync/tests/mod.rs

@@ -9,9 +9,14 @@ use beacon_processor::WorkEvent;
 use lighthouse_network::NetworkGlobals;
 use rand_chacha::ChaCha20Rng;
 use slot_clock::ManualSlotClock;
-use std::sync::Arc;
+use std::fs::OpenOptions;
+use std::io::Write;
+use std::sync::{Arc, Once};
 use store::MemoryStore;
 use tokio::sync::mpsc;
+use tracing_subscriber::fmt::MakeWriter;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
 use types::{ChainSpec, ForkName, MinimalEthSpec as E};
 
 mod lookups;
@@ -65,3 +70,55 @@ struct TestRig {
     fork_name: ForkName,
     spec: Arc<ChainSpec>,
 }
+
+// Environment variable to read if `fork_from_env` feature is enabled.
+pub const FORK_NAME_ENV_VAR: &str = "FORK_NAME";
+// Environment variable specifying the log output directory in CI.
+pub const CI_LOGGER_DIR_ENV_VAR: &str = "CI_LOGGER_DIR";
+
+static INIT_TRACING: Once = Once::new();
+
+pub fn init_tracing() {
+    INIT_TRACING.call_once(|| {
+        if std::env::var(CI_LOGGER_DIR_ENV_VAR).is_ok() {
+            // Enable logging to log files for each test and each fork.
+            tracing_subscriber::registry()
+                .with(
+                    tracing_subscriber::fmt::layer()
+                        .with_ansi(false)
+                        .with_writer(CILogWriter),
+                )
+                .init();
+        }
+    });
+}
+
+// CILogWriter writes logs to separate files for each test and each fork.
+struct CILogWriter;
+
+impl<'a> MakeWriter<'a> for CILogWriter {
+    type Writer = Box<dyn Write + Send>;
+
+    // fmt::Layer calls this method each time an event is recorded.
+    fn make_writer(&'a self) -> Self::Writer {
+        let log_dir = std::env::var(CI_LOGGER_DIR_ENV_VAR).unwrap();
+        let fork_name = std::env::var(FORK_NAME_ENV_VAR)
+            .map(|s| format!("{s}_"))
+            .unwrap_or_default();
+
+        // The current test name can be got via the thread name.
+        let test_name = std::thread::current()
+            .name()
+            .unwrap_or("unnamed")
+            .replace(|c: char| !c.is_alphanumeric(), "_");
+
+        let file_path = format!("{log_dir}/{fork_name}{test_name}.log");
+        let file = OpenOptions::new()
+            .append(true)
+            .create(true)
+            .open(&file_path)
+            .expect("failed to open a log file");
+
+        Box::new(file)
+    }
+}

common/logging/src/tracing_libp2p_discv5_logging_layer.rs

@@ -59,6 +59,7 @@ impl tracing_core::field::Visit for LogMessageExtractor {
 pub fn create_libp2p_discv5_tracing_layer(
     base_tracing_log_path: Option<PathBuf>,
     max_log_size: u64,
+    file_mode: u32,
 ) -> Option<Libp2pDiscv5TracingLayer> {
     if let Some(mut tracing_log_path) = base_tracing_log_path {
         // Ensure that `tracing_log_path` only contains directories.
@@ -75,12 +76,14 @@ pub fn create_libp2p_discv5_tracing_layer(
         let libp2p_writer =
             LogRollerBuilder::new(tracing_log_path.clone(), PathBuf::from("libp2p.log"))
                 .rotation(Rotation::SizeBased(RotationSize::MB(max_log_size)))
-                .max_keep_files(1);
+                .max_keep_files(1)
+                .file_mode(file_mode);
 
         let discv5_writer =
             LogRollerBuilder::new(tracing_log_path.clone(), PathBuf::from("discv5.log"))
                 .rotation(Rotation::SizeBased(RotationSize::MB(max_log_size)))
-                .max_keep_files(1);
+                .max_keep_files(1)
+                .file_mode(file_mode);
 
         let libp2p_writer = match libp2p_writer.build() {
             Ok(writer) => writer,