chore(starknet_sequencer_node): add node runner id (#3125)

* chore(starknet_sequencer_node): add annotator task

commit-id:24244932

* chore(starknet_sequencer_node): add node runner id

commit-id:865b3a9a

crates/starknet_integration_tests/src/sequencer_manager.rs

@@ -11,7 +11,7 @@ use starknet_api::state::StateNumber;
 use starknet_api::transaction::TransactionHash;
 use starknet_sequencer_infra::test_utils::AvailablePorts;
 use starknet_sequencer_node::config::component_config::ComponentConfig;
-use starknet_sequencer_node::test_utils::node_runner::spawn_run_node;
+use starknet_sequencer_node::test_utils::node_runner::{spawn_run_node, NodeRunner};
 use starknet_types_core::felt::Felt;
 use tokio::task::JoinHandle;
 use tracing::info;
@@ -41,7 +41,10 @@ impl SequencerManager {
         info!("Running sequencers.");
         let sequencer_run_handles = sequencers
             .iter()
-            .map(|sequencer| spawn_run_node(sequencer.node_config_path.clone()))
+            .enumerate()
+            .map(|(i, sequencer)| {
+                spawn_run_node(sequencer.node_config_path.clone(), NodeRunner::new(i))
+            })
             .collect::<Vec<_>>();
 
         let sequencer_manager = Self { sequencers, sequencer_run_handles };

crates/starknet_sequencer_node/src/test_utils/node_runner.rs

@@ -3,38 +3,90 @@ use std::process::Stdio;
 
 use infra_utils::command::create_shell_command;
 use infra_utils::path::resolve_project_relative_path;
+use tokio::io::{AsyncWriteExt, BufReader};
 use tokio::process::Child;
 use tokio::task::{self, JoinHandle};
 use tracing::{error, info, instrument};
 
 pub const NODE_EXECUTABLE_PATH: &str = "target/debug/starknet_sequencer_node";
 
-pub fn spawn_run_node(node_config_path: PathBuf) -> JoinHandle<()> {
+pub struct NodeRunner {
+    description: String,
+}
+
+impl NodeRunner {
+    pub fn new(index: usize) -> Self {
+        Self { description: format! {"Node id {}:", index} }
+    }
+
+    pub fn get_description(&self) -> String {
+        self.description.clone()
+    }
+}
+
+pub fn spawn_run_node(node_config_path: PathBuf, node_runner: NodeRunner) -> JoinHandle<()> {
     task::spawn(async move {
         info!("Running the node from its spawned task.");
-        let _node_run_result = spawn_node_child_process(node_config_path).
-            await. // awaits the completion of spawn_node_child_task.
-            wait(). // runs the node until completion -- should be running indefinitely.
-            await; // awaits the completion of the node.
+        // Obtain both handles, as the processes are terminated when their handles are dropped.
+        let (mut node_handle, _annotator_handle) =
+            spawn_node_child_process(node_config_path, node_runner).await;
+        let _node_run_result = node_handle.
+            wait(). // Runs the node until completion, should be running indefinitely.
+            await; // Awaits the completion of the node.
         panic!("Node stopped unexpectedly.");
     })
 }
 
-#[instrument()]
-async fn spawn_node_child_process(node_config_path: PathBuf) -> Child {
-    // TODO(Tsabary): Capture output to a log file, and present it in case of a failure.
+#[instrument(skip(node_runner))]
+async fn spawn_node_child_process(
+    node_config_path: PathBuf,
+    node_runner: NodeRunner,
+) -> (Child, Child) {
     info!("Getting the node executable.");
     let node_executable = get_node_executable_path();
 
     info!("Running the node from: {}", node_executable);
-    create_shell_command(node_executable.as_str())
+    let mut node_cmd: Child = create_shell_command(node_executable.as_str())
         .arg("--config_file")
         .arg(node_config_path.to_str().unwrap())
         .stderr(Stdio::inherit())
+        .stdout(Stdio::piped())
+        .kill_on_drop(true) // Required for stopping when the handle is dropped.
+        .spawn()
+        .expect("Spawning sequencer node should succeed.");
+
+    let mut annotator_cmd: Child = create_shell_command("awk")
+        .arg("-v")
+        .arg(format!{"prefix={}", node_runner.get_description()})
+        .arg("{print prefix, $0}")
+        .stdin(std::process::Stdio::piped())
+        .stderr(Stdio::inherit())
         .stdout(Stdio::inherit())
-        .kill_on_drop(true) // Required for stopping the node when the handle is dropped.
+        .kill_on_drop(true) // Required for stopping when the handle is dropped.
         .spawn()
-        .expect("Failed to spawn the sequencer node.")
+        .expect("Spawning node output annotation should succeed.");
+
+    // Get the node stdout and the annotator stdin.
+    let node_stdout = node_cmd.stdout.take().expect("Node stdout should be available.");
+    let mut annotator_stdin =
+        annotator_cmd.stdin.take().expect("Annotator stdin should be available.");
+
+    // Spawn a task to connect the node stdout with the annotator stdin.
+    tokio::spawn(async move {
+        // Copy data from node stdout to annotator stdin.
+        if let Err(e) =
+            tokio::io::copy(&mut BufReader::new(node_stdout), &mut annotator_stdin).await
+        {
+            error!("Error while copying from node stdout to annotator stdin: {}", e);
+        }
+
+        // Close the annotator stdin when done.
+        if let Err(e) = annotator_stdin.shutdown().await {
+            error!("Failed to shut down annotator stdin: {}", e);
+        }
+    });
+
+    (node_cmd, annotator_cmd)
 }
 
 pub fn get_node_executable_path() -> String {