Repository: tangle-network/gadget (language: rust)

Issue:
[TASK] Create `gadget-executor` Gadget for executing shell commands as a service
## Overview
By creating a Gadget that executes shell commands, the Gadget will become capable of encompassing a variety of other services: Docker containers, Eigenlayer AVS, etc. 

### Command Execution
- [x] Execute a command and get the output of the command
- [x] Execute _multiple_ commands sequentially, potentially building upon the output of each executed command
- [x] Launch a continuous service like a Docker container, where a binary may live

### Monitoring/Communication between `gadget-executor` and its children
- [x] Executor can restart children upon restarting; executor _picks back up_
- [x] Should deal with loss of communication from child; is it dead?

### gadget-executor
For now, the gadget-executor is located in the [protocols repo](https://github.com/webb-tools/protocols) as it follows the same overarching structure as each of the protocols within that repository. As it furthers in development and use, it may become necessary for it to be moved to its own separate repository or into gadget. 


Discussion / hints:

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 08241482d..28a153b35 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -8,7 +8,7 @@ members = [
     "protocol-macros",
     "shell-manager",
     "shell-sdk",
-    "sdk",
+    "sdk", "executor",
 ]
 
 [workspace.package]
diff --git a/executor/Cargo.toml b/executor/Cargo.toml
new file mode 100644
index 000000000..8317672cc
--- /dev/null
+++ b/executor/Cargo.toml
@@ -0,0 +1,20 @@
+[package]
+name = "gadget-executor"
+version = "0.1.0"
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+
+[dependencies]
+tokio = "1.38.0"
+futures = "0.3.30"
+serde_json = "1.0.117"
+gadget-common = { workspace = true, features = ["default"] }
+gadget-io = { workspace = true, features = ["default"] }
+failure = "0.1.8"
+nix = { version = "0.29.0", features = ["process", "signal"] }
+serde = { version = "1.0.203", features = ["derive"] }
+itertools = "0.13.0"
+sysinfo = { version = "0.30.12" }
\ No newline at end of file
diff --git a/executor/src/lib.rs b/executor/src/lib.rs
new file mode 100644
index 000000000..c110f7300
--- /dev/null
+++ b/executor/src/lib.rs
@@ -0,0 +1,99 @@
+use crate::process::manager::GadgetProcessManager;
+use crate::process::types::GadgetInstructionData;
+use crate::process::types::{CommandOrSequence, ProcessOutput};
+use crate::process::utils::*;
+use std::error::Error;
+
+mod process;
+
+// TODO: Gadget Setup following V2 - yet to be implemented
+
+pub async fn run_executor(instructions: &str) {
+    // TODO: New commands should be received real-time from connection - will follow new V2 structure
+    let mut manager = GadgetProcessManager::new();
+    let instructions: GadgetInstructionData =
+        serde_json::from_str(instructions).expect("JSON was not well-formatted");
+
+    // Execute all the required commands
+    for to_execute in instructions.commands {
+        let identifier = to_execute.name;
+        let commands = to_execute.command;
+        match commands {
+            CommandOrSequence::Command(cmd) => {
+                println!("Executing single command {:?}: {:?}", identifier, cmd);
+                match manager.run(identifier, &cmd).await {
+                    Ok(_identifier) => {
+                        // Logs/Metrics
+                    }
+                    Err(_error) => {
+                        // Error logging
+                    }
+                }
+            }
+            CommandOrSequence::Sequence(cmds) => {
+                println!(
+                    "Executing sequence of commands {:?}: {:?}",
+                    identifier, cmds
+                );
+                // TODO: Utilizing the output of commands for new commands
+            }
+        }
+    }
+
+    // Receive input from all running processes
+    let mut ended = Vec::new();
+    for (service, process) in &mut manager.children {
+        println!("LOG : Process {}", service.clone());
+        let status = process.status().unwrap();
+        println!("\tSTATUS: {:?}", status);
+        let output = process.read().await;
+        println!("\n{} read result:\n\t {:?}\n", service, output);
+        if let ProcessOutput::Exhausted(_) = output {
+            println!("{} ended, removing from Manager", service);
+            ended.push(service.clone());
+        }
+    }
+    manager.children.retain(|k, _| !ended.contains(k));
+
+    println!(
+        " ENDED EXECUTOR WITH GADGET PROCESS MANAGER:\n{:?}\n",
+        manager
+    );
+}
+
+pub async fn run_tangle_validator() -> Result<(), Box<dyn Error>> {
+    let mut manager = GadgetProcessManager::new();
+
+    // let install = manager.run(&*"binary_install", &*"wget https://github.com/webb-tools/tangle/releases/download/v1.0.0/tangle-default-linux-amd64".to_string()).await?;
+    // let make_executable = manager.run(&*"make_executable", &*"chmod +x tangle-default-linux-amd64".to_string()).await?;
+
+    // let start_validator = manager.run("tangle_validator".to_string(), "./tangle-default-linux-amd64").await?;
+    let ping_test = manager
+        .run("ping_test".to_string(), "ping -c 5 localhost")
+        .await?;
+    manager.focus_service_to_completion(ping_test).await?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_executor() {
+        let the_file = r#"{
+            "commands": [
+                ["ping_google", "ping -c 5 google.com"],
+                ["ping_local", "ping -c 5 localhost"],
+                ["sequence_test", ["ping -c 5 localhost", "ls", "clear"]]
+            ]
+        }"#;
+        run_executor(the_file).await;
+    }
+
+    #[tokio::test]
+    async fn test_validator() {
+        run_tangle_validator().await.unwrap();
+    }
+}
diff --git a/executor/src/process/manager.rs b/executor/src/process/manager.rs
new file mode 100644
index 000000000..38f8a6acb
--- /dev/null
+++ b/executor/src/process/manager.rs
@@ -0,0 +1,156 @@
+use crate::process::types::{GadgetProcess, ProcessOutput, Status};
+use crate::process::utils::*;
+use crate::{craft_child_process, run_command, OS_COMMAND};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::error::Error;
+use sysinfo::{Pid, System};
+use tokio::fs::File;
+use tokio::io::AsyncWriteExt;
+
+/// Manager for gadget-executor process. The processes are recorded to be controlled by their Service name.
+/// This Manager can be reconstructed from a file to recover a gadget-executor.
+#[derive(Serialize, Deserialize, Debug)]
+pub struct GadgetProcessManager {
+    /// Hashmap that contains all the children spawned by this Manager. Keys are the names of each Service.
+    pub children: HashMap<String, GadgetProcess>,
+}
+
+impl GadgetProcessManager {
+    pub(crate) fn new() -> GadgetProcessManager {
+        GadgetProcessManager {
+            children: HashMap::new(),
+        }
+    }
+
+    /// Load the state of previously running processes to recover gadget-executor
+    #[allow(dead_code)]
+    pub(crate) async fn new_from_saved(file: &str) -> Result<GadgetProcessManager, Box<dyn Error>> {
+        let file = std::fs::File::open(file).unwrap();
+        let mut new_manager: GadgetProcessManager = serde_json::from_reader(file).unwrap();
+
+        // Restarts processes that were previously running
+        new_manager.restart_dead().await?;
+
+        Ok(new_manager)
+    }
+
+    /// Store the state of the current processes
+    #[allow(dead_code)]
+    pub(crate) async fn save_state(&self) -> Result<String, Box<dyn Error>> {
+        let serialized_data = serde_json::to_string(self)?;
+        let mut file = File::create("./savestate.json").await?;
+        file.write_all(serialized_data.clone().as_bytes()).await?;
+        Ok(serialized_data)
+    }
+
+    /// Runs the given command and stores it using the identifier as the key. Returns the identifier used
+    pub(crate) async fn run(
+        &mut self,
+        identifier: String,
+        command: &str,
+    ) -> Result<String, Box<dyn Error>> {
+        let gadget_process = run_command!(command)?;
+        self.children.insert(identifier.clone(), gadget_process);
+        Ok(identifier)
+    }
+
+    pub(crate) async fn focus_service_to_completion(
+        &mut self,
+        service: String,
+    ) -> Result<(), Box<dyn Error>> {
+        let process = self
+            .children
+            .get_mut(&service)
+            .ok_or(format!("Failed to focus on {service}, it does not exist"))?;
+        loop {
+            match process.read().await {
+                ProcessOutput::Output(output) => {
+                    println!("{output:?}");
+                    continue;
+                }
+                ProcessOutput::Exhausted(output) => {
+                    println!("{output:?}");
+                    break;
+                }
+                ProcessOutput::Waiting => {
+                    continue;
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Removes processes that are no longer running from the manager. Returns a Vector of the names of processes removed
+    #[allow(dead_code)]
+    pub(crate) async fn remove_dead(&mut self) -> Result<Vec<String>, Box<dyn Error>> {
+        let dead_processes = Vec::new();
+        let mut to_remove = Vec::new();
+        let s = System::new_all();
+
+        // Find dead processes and gather them for return & removal
+        for (key, value) in self.children.iter() {
+            let current_pid = value.pid;
+            if let Some(process) = s.process(Pid::from_u32(current_pid)) {
+                if process.name() == value.process_name {
+                    // Still running
+                    continue;
+                } else {
+                    // No longer running, some unknown process is now utilizing this PID
+                    to_remove.push(key.clone());
+                }
+            }
+        }
+        self.children.retain(|k, _| !to_remove.contains(k));
+
+        // TODO: If dead children are `supposed` to be running, we should start them up again instead of just removing them
+
+        Ok(dead_processes)
+    }
+
+    /// Finds all dead processes that still exist in map and starts them again. This function
+    /// is used to restart all processes after loading a Manager from a file.
+    #[allow(dead_code)]
+    pub(crate) async fn restart_dead(&mut self) -> Result<(), Box<dyn Error>> {
+        let mut restarted_processes = Vec::new();
+        let mut to_remove = Vec::new();
+        // Find dead processes and restart them
+        for (key, value) in self.children.iter_mut() {
+            match value.status() {
+                Ok(status) => {
+                    match status {
+                        Status::Active | Status::Sleeping => {
+                            // TODO: Metrics + Logs for these living processes
+                            // Check if this process is still running what is expected
+                            // If it is still correctly running, we just move along
+                            continue;
+                        }
+                        Status::Inactive | Status::Dead => {
+                            // Dead, should be restarted
+                        }
+                        Status::Unknown(code) => {
+                            println!("LOG : {} yielded {}", key.clone(), code);
+                        }
+                    }
+                }
+                Err(err) => {
+                    // TODO: Log error
+                    // Error generally means it died and no longer exists - restart it
+                    println!(
+                        "LOG : {} yielded {} while attempting to restart dead processes",
+                        key.clone(),
+                        err
+                    );
+                }
+            }
+            restarted_processes.push((key.clone(), value.restart_process().await?));
+            to_remove.push(key.clone());
+        }
+        self.children.retain(|k, _| !to_remove.contains(k));
+        for (service, restarted) in restarted_processes {
+            self.children.insert(service.clone(), restarted);
+        }
+
+        Ok(())
+    }
+}
diff --git a/executor/src/process/mod.rs b/executor/src/process/mod.rs
new file mode 100644
index 000000000..74717cb4e
--- /dev/null
+++ b/executor/src/process/mod.rs
@@ -0,0 +1,3 @@
+pub(crate) mod manager;
+pub(crate) mod types;
+pub(crate) mod utils;
diff --git a/executor/src/process/types.rs b/executor/src/process/types.rs
new file mode 100644
index 000000000..5bd0d6d59
--- /dev/null
+++ b/executor/src/process/types.rs
@@ -0,0 +1,235 @@
+use crate::process::utils::*;
+use crate::{craft_child_process, run_command, OS_COMMAND};
+use failure::format_err;
+use nix::libc::pid_t;
+use nix::sys::signal;
+use nix::sys::signal::Signal;
+use serde::{Deserialize, Serialize};
+use std::error::Error;
+use std::time::Duration;
+use sysinfo::ProcessStatus::{
+    Dead, Idle, LockBlocked, Parked, Run, Sleep, Stop, Tracing, UninterruptibleDiskSleep, Unknown,
+    Wakekill, Waking, Zombie,
+};
+use sysinfo::{Pid, ProcessStatus, System};
+use tokio::io::{BufReader, Lines};
+pub use tokio::process::Child;
+use tokio::time::timeout;
+
+/// A Process spawned by gadget-executor, running some service or command(s)
+#[derive(Serialize, Deserialize, Debug)]
+pub(crate) struct GadgetProcess {
+    /// The command executed
+    pub(crate) command: String,
+    /// The name of the process itself
+    pub(crate) process_name: String,
+    /// Process ID
+    pub(crate) pid: u32,
+    /// History of output from process for reviewing/tracking progress
+    pub(crate) output: Vec<String>,
+    /// Stream for output from child process
+    #[serde(skip_serializing, skip_deserializing)]
+    pub(crate) stream: Option<Lines<BufReader<tokio::process::ChildStdout>>>,
+}
+
+impl GadgetProcess {
+    pub fn new(
+        command: String,
+        pid: Option<u32>,
+        output: Vec<String>,
+        stream: Lines<BufReader<tokio::process::ChildStdout>>,
+    ) -> Result<GadgetProcess, Box<dyn Error>> {
+        let s = System::new_all();
+        let pid = pid.ok_or("No PID found")?;
+        let process_name = s
+            .process(Pid::from_u32(pid))
+            .ok_or(format!("Process {pid} doesn't exist"))?
+            .name()
+            .to_string();
+        Ok(GadgetProcess {
+            command,
+            process_name,
+            pid,
+            output,
+            stream: Some(stream),
+        })
+    }
+
+    /// Read output from this GadgetProcess, returning None if there was no output
+    pub async fn read(&mut self) -> ProcessOutput {
+        let messages = Vec::new();
+        if let Some(stream) = &mut self.stream {
+            // Read lines until we time out, meaning we are still waiting for output - continue for now
+            loop {
+                let read_result = timeout(Duration::from_millis(500), stream.next_line()).await;
+                match read_result {
+                    Ok(output) => {
+                        if output.is_err() {
+                            // TODO: Error logging
+                            println!("{} encountered read error", self.process_name.clone());
+                        } else {
+                            let inbound_message = output.unwrap();
+                            match inbound_message {
+                                None => {
+                                    // Stream is completed - process is finished
+                                    println!(
+                                        "{} : STREAM COMPLETED - ENDING",
+                                        self.process_name.clone()
+                                    );
+                                    // TODO: Log
+                                    return ProcessOutput::Exhausted(messages);
+                                }
+                                Some(streamed_output) => {
+                                    // We received output from child process
+                                    // TODO: Log
+                                    println!(
+                                        "{} : MESSAGE LOG : {}",
+                                        self.process_name.clone(),
+                                        streamed_output.clone()
+                                    );
+                                    self.output.push(streamed_output);
+                                }
+                            }
+                        }
+                    }
+                    Err(timeout) => {
+                        // TODO: Log
+                        println!(
+                            "{} read attempt timed out after {}, continuing...",
+                            self.process_name.clone(),
+                            timeout
+                        );
+                        break;
+                    }
+                }
+            }
+
+            if messages.is_empty() {
+                ProcessOutput::Waiting
+            } else {
+                ProcessOutput::Output(messages)
+            }
+        } else {
+            // TODO: Error logging
+            println!("{} encountered read error", self.process_name.clone());
+            ProcessOutput::Waiting
+        }
+    }
+
+    /// Restart a GadgetProcess, killing the previously running process if it exists. Returns the new GadgetProcess
+    pub(crate) async fn restart_process(&mut self) -> Result<GadgetProcess, Box<dyn Error>> {
+        // Kill current process running this command
+        let s = System::new_all();
+        match s.process(Pid::from_u32(self.pid)) {
+            Some(process) => {
+                if process.name() == self.process_name {
+                    self.kill()?;
+                }
+            }
+            None => {
+                // No need to worry, the previously running process died
+                // TODO: Log
+                println!(
+                    "LOG : Process restart attempt found no process for PID {:?}",
+                    self.pid
+                );
+            }
+        }
+        run_command!(&self.command.clone())
+    }
+
+    /// Checks the status of this GadgetProcess
+    pub(crate) fn status(&self) -> Result<Status, Box<dyn Error>> {
+        let s = System::new_all();
+        match s.process(Pid::from_u32(self.pid)) {
+            Some(process) => Ok(Status::from(process.status())),
+            None => {
+                // If it isn't found, then the process died
+                Ok(Status::Dead)
+            }
+        }
+    }
+
+    /// Gets process name by PID
+    #[allow(dead_code)]
+    pub(crate) fn get_name(&self) -> Result<String, Box<dyn Error>> {
+        let s = System::new_all();
+        let name = s
+            .process(Pid::from_u32(self.pid))
+            .ok_or(format!("Process {} doesn't exist", self.pid))?
+            .name();
+        Ok(name.into())
+    }
+
+    /// Terminates the process depicted by this GadgetProcess - will fail if the PID is now being reused
+    pub(crate) fn kill(&self) -> Result<(), Box<dyn Error>> {
+        let running_process = self.get_name()?;
+        if running_process == self.process_name {
+            Ok(signal::kill(
+                nix::unistd::Pid::from_raw(self.pid as pid_t),
+                Signal::SIGTERM,
+            )?)
+        } else {
+            Err(Box::from(format_err!(
+                "Expected {} and found {} running instead - process termination aborted",
+                self.process_name,
+                running_process
+            )))
+        }
+    }
+}
+
+#[derive(Debug)]
+#[allow(dead_code)]
+pub(crate) enum ProcessOutput {
+    /// Normal collection of output lines from a given process
+    Output(Vec<String>),
+    /// Output lines from a given process that successfully completed
+    Exhausted(Vec<String>),
+    /// No output - still waiting for next output from process
+    Waiting,
+}
+
+#[derive(Debug)]
+pub(crate) enum Status {
+    /// Process is running or able to run
+    Active,
+    /// Stopped process
+    Inactive,
+    /// Sleeping process, either waiting for resources or a signal
+    Sleeping,
+    /// Zombie process
+    Dead,
+    /// Other or invalid status - if this occurs, something is likely wrong
+    Unknown(String),
+}
+
+impl From<ProcessStatus> for Status {
+    fn from(value: ProcessStatus) -> Status {
+        match value {
+            Run | Waking => Status::Active,
+            Sleep | UninterruptibleDiskSleep | Parked | LockBlocked | Wakekill => Status::Sleeping,
+            Stop | Tracing | Idle => Status::Inactive,
+            Dead | Zombie => Status::Dead,
+            Unknown(code) => Status::Unknown(format!("Unknown with code {code}")),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub(crate) struct GadgetInstructionData {
+    pub(crate) commands: Vec<CommandData>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub(crate) struct CommandData {
+    pub(crate) name: String,
+    pub(crate) command: CommandOrSequence,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(untagged)]
+pub(crate) enum CommandOrSequence {
+    Command(String),
+    Sequence(Vec<String>),
+}
diff --git a/executor/src/process/utils.rs b/executor/src/process/utils.rs
new file mode 100644
index 000000000..3fec9d911
--- /dev/null
+++ b/executor/src/process/utils.rs
@@ -0,0 +1,81 @@
+pub use futures::{FutureExt, StreamExt};
+pub use std::process::Stdio;
+pub use tokio::io::Lines;
+pub use tokio::io::{AsyncBufReadExt, BufReader};
+pub use tokio::process::{Child, Command};
+
+#[cfg(target_family = "unix")]
+pub(crate) static OS_COMMAND: (&str, &str) = ("sh", "-c");
+#[cfg(target_family = "windows")]
+pub(crate) static OS_COMMAND: (&str, &str) = ("cmd", "/C");
+
+#[macro_export]
+macro_rules! craft_child_process {
+    ($cmd:expr) => {{
+        let (cmd, arg) = OS_COMMAND;
+        Command::new(cmd).args(vec![arg, $cmd])
+            .stdout(Stdio::piped())
+            .spawn()
+            .expect(&format!("Failed to execute: {:?} - Input should be a valid command", $cmd))
+    }};
+    ($cmd:expr, $($args:expr),*) => {{
+        let (cmd, arg) = OS_COMMAND;
+        let command = Command::new(cmd).args(vec![arg, $cmd]);
+        $(
+            command.args(String::try_from($args).unwrap_or(format!("Argument {:?} is causing an error", $args)).split_whitespace().map(str::to_string).collect::<Vec<String>>());
+        )*
+        command
+            .stdout(Stdio::piped())
+            .spawn()
+            .expect(&format!("Failed to execute: {} {:?} - Input should be a valid command", $cmd, &[$($args),*]));
+    }};
+}
+
+pub(crate) fn create_stream(mut child: Child) -> Lines<BufReader<tokio::process::ChildStdout>> {
+    // Create stream to read output
+    let stdout = child
+        .stdout
+        .take()
+        .expect("Failed to take stdout handle from child...");
+    let reader = BufReader::new(stdout).lines();
+
+    // Run in Tokio runtime to ensure it stays alive
+    tokio::spawn(async move {
+        let status = child
+            .wait()
+            .await
+            .expect("Child process encountered an error...");
+        println!("Child process ended with status: {}", status)
+    });
+
+    // Return stream
+    reader
+}
+
+#[macro_export]
+macro_rules! run_command {
+    ($cmd:expr) => {{
+        // Spawn child running process
+        let child: Child = craft_child_process!($cmd);
+        let pid = child.id().clone();
+        let stream = create_stream(child);
+        GadgetProcess::new(
+            $cmd.to_string(),
+            pid,
+            Vec::new(),
+            stream,
+        )
+    }};
+    ($cmd:expr, $($args:expr),*) => {{
+        // Spawn child running process
+        let child = craft_child_process!($cmd,$($args),*);
+        let pid = child.id().clone();
+        let stream = create_stream(child);
+        GadgetProcess::new(
+            $cmd.to_string(),
+            pid,
+            Vec::new(),
+            stream,
+        )
+    }};
+}


Test patch: