feat: introduce cmd to stop all running nodes

resources/ansible/stop_nodes.yml

@@ -0,0 +1,9 @@
+---
+- name: ensure all nodes are stopped using the node manager
+  hosts: all
+  become: True
+  vars:
+    interval: "{{ interval }}"
+  tasks:
+    - name: stop all nodes
+      ansible.builtin.command: safenode-manager stop --interval {{ interval }}

src/ansible/extra_vars.rs

@@ -18,7 +18,7 @@ const NODE_MANAGER_S3_BUCKET_URL: &str = "https://sn-node-manager.s3.eu-west-2.a
 const RPC_CLIENT_BUCKET_URL: &str = "https://sn-node-rpc-client.s3.eu-west-2.amazonaws.com";
 const AUTONOMI_S3_BUCKET_URL: &str = "https://autonomi-cli.s3.eu-west-2.amazonaws.com";
 
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct ExtraVarsDocBuilder {
     map: serde_json::Map<String, Value>,
 }

src/ansible/mod.rs

@@ -135,6 +135,11 @@ pub enum AnsiblePlaybook {
     StartUploaders,
     /// This playbook will stop the faucet for the environment.
     StopFaucet,
+    /// The stop nodes playbook will use the node manager to stop any node services on any
+    /// machines it runs against.
+    ///
+    /// Use in combination with `AnsibleInventoryType::Genesis` or `AnsibleInventoryType::Nodes`.
+    StopNodes,
     /// This playbook will stop the Telegraf service running on each machine.
     ///
     /// It can be necessary for running upgrades, since Telegraf will run `safenode-manager
@@ -185,6 +190,7 @@ impl AnsiblePlaybook {
             AnsiblePlaybook::StartUploaders => "start_uploaders.yml".to_string(),
             AnsiblePlaybook::Status => "node_status.yml".to_string(),
             AnsiblePlaybook::StopFaucet => "stop_faucet.yml".to_string(),
+            AnsiblePlaybook::StopNodes => "stop_nodes.yml".to_string(),
             AnsiblePlaybook::StopTelegraf => "stop_telegraf.yml".to_string(),
             AnsiblePlaybook::StopUploaders => "stop_uploaders.yml".to_string(),
             AnsiblePlaybook::UpgradeNodeManager => "upgrade_node_manager.yml".to_string(),

src/ansible/provisioning.rs

@@ -504,7 +504,7 @@ impl AnsibleProvisioner {
         extra_vars.add_variable("interval", &interval.as_millis().to_string());
 
         if let Some(custom_inventory) = custom_inventory {
-            println!("Running the start telegraf playbook with a custom inventory");
+            println!("Running the start nodes playbook with a custom inventory");
             generate_custom_environment_inventory(
                 &custom_inventory,
                 environment_name,
@@ -567,6 +567,41 @@ impl AnsibleProvisioner {
         Ok(())
     }
 
+    pub fn stop_nodes(
+        &self,
+        environment_name: &str,
+        interval: Duration,
+        custom_inventory: Option<Vec<VirtualMachine>>,
+    ) -> Result<()> {
+        let mut extra_vars = ExtraVarsDocBuilder::default();
+        extra_vars.add_variable("interval", &interval.as_millis().to_string());
+
+        if let Some(custom_inventory) = custom_inventory {
+            println!("Running the stop nodes playbook with a custom inventory");
+            generate_custom_environment_inventory(
+                &custom_inventory,
+                environment_name,
+                &self.ansible_runner.working_directory_path.join("inventory"),
+            )?;
+            self.ansible_runner.run_playbook(
+                AnsiblePlaybook::StopNodes,
+                AnsibleInventoryType::Custom,
+                Some(extra_vars.build()),
+            )?;
+            return Ok(());
+        }
+
+        for node_inv_type in AnsibleInventoryType::iter_node_type() {
+            self.ansible_runner.run_playbook(
+                AnsiblePlaybook::StopNodes,
+                node_inv_type,
+                Some(extra_vars.build()),
+            )?;
+        }
+
+        Ok(())
+    }
+
     pub fn stop_telegraf(
         &self,
         environment_name: &str,

src/error.rs

@@ -125,6 +125,8 @@ pub enum Error {
     NoFaucetError,
     #[error("This deployment does not have any uploaders. It may be a bootstrap deployment.")]
     NoUploadersError,
+    #[error("The node count for the provided custom vms are not equal")]
+    NodeCountMismatch,
     #[error("Could not obtain a multiaddr from the node inventory")]
     NodeAddressNotFound,
     #[error("Failed to upload {0} to S3 bucket {1}")]

src/inventory.rs

@@ -533,6 +533,12 @@ pub struct NodeVirtualMachine {
     pub safenodemand_endpoint: Option<SocketAddr>,
 }
 
+impl NodeVirtualMachine {
+    pub fn node_count(&self) -> usize {
+        self.node_multiaddr.len()
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
 pub struct VirtualMachine {
     pub id: u64,
@@ -669,20 +675,12 @@ impl DeploymentInventory {
         list
     }
 
-    pub fn node_vm_list(&self) -> Vec<VirtualMachine> {
+    pub fn node_vm_list(&self) -> Vec<NodeVirtualMachine> {
         let mut list = Vec::new();
-        list.extend(
-            self.bootstrap_node_vms
-                .iter()
-                .map(|node_vm| node_vm.vm.clone()),
-        );
-        list.extend(self.genesis_vm.iter().map(|node_vm| node_vm.vm.clone()));
-        list.extend(self.node_vms.iter().map(|node_vm| node_vm.vm.clone()));
-        list.extend(
-            self.private_node_vms
-                .iter()
-                .map(|node_vm| node_vm.vm.clone()),
-        );
+        list.extend(self.bootstrap_node_vms.iter().cloned());
+        list.extend(self.genesis_vm.iter().cloned());
+        list.extend(self.node_vms.iter().cloned());
+        list.extend(self.private_node_vms.iter().cloned());
 
         list
     }
@@ -737,31 +735,31 @@ impl DeploymentInventory {
 
     pub fn bootstrap_node_count(&self) -> usize {
         if let Some(first_vm) = self.bootstrap_node_vms.first() {
-            first_vm.node_multiaddr.len()
+            first_vm.node_count()
         } else {
             0
         }
     }
 
     pub fn genesis_node_count(&self) -> usize {
         if let Some(genesis_vm) = &self.genesis_vm {
-            genesis_vm.node_multiaddr.len()
+            genesis_vm.node_count()
         } else {
             0
         }
     }
 
     pub fn node_count(&self) -> usize {
         if let Some(first_vm) = self.node_vms.first() {
-            first_vm.node_multiaddr.len()
+            first_vm.node_count()
         } else {
             0
         }
     }
 
     pub fn private_node_count(&self) -> usize {
         if let Some(first_vm) = self.private_node_vms.first() {
-            first_vm.node_multiaddr.len()
+            first_vm.node_count()
         } else {
             0
         }

src/lib.rs

@@ -683,6 +683,16 @@ impl TestnetDeployer {
         Ok(())
     }
 
+    pub fn stop(
+        &self,
+        interval: Duration,
+        custom_inventory: Option<Vec<VirtualMachine>>,
+    ) -> Result<()> {
+        self.ansible_provisioner
+            .stop_nodes(&self.environment_name, interval, custom_inventory)?;
+        Ok(())
+    }
+
     pub fn stop_telegraf(&self, custom_inventory: Option<Vec<VirtualMachine>>) -> Result<()> {
         self.ansible_provisioner
             .stop_telegraf(&self.environment_name, custom_inventory)?;

src/main.rs

@@ -611,6 +611,27 @@ enum Commands {
         #[clap(long, value_parser = parse_provider, verbatim_doc_comment, default_value_t = CloudProvider::DigitalOcean)]
         provider: CloudProvider,
     },
+    /// Stop all nodes in an environment.
+    #[clap(name = "stop")]
+    Stop {
+        /// Provide a list of VM names to use as a custom inventory.
+        ///
+        /// This will stop nodes on a particular subset of VMs.
+        #[clap(name = "custom-inventory", long, use_value_delimiter = true)]
+        custom_inventory: Option<Vec<String>>,
+        /// Maximum number of forks Ansible will use to execute tasks on target hosts.
+        #[clap(long, default_value_t = 50)]
+        forks: usize,
+        /// The interval between each node stop in milliseconds.
+        #[clap(long, value_parser = |t: &str| -> Result<Duration> { Ok(t.parse().map(Duration::from_millis)?)}, default_value = "2000")]
+        interval: Duration,
+        /// The name of the environment.
+        #[arg(short = 'n', long)]
+        name: String,
+        /// The cloud provider for the environment.
+        #[clap(long, value_parser = parse_provider, verbatim_doc_comment, default_value_t = CloudProvider::DigitalOcean)]
+        provider: CloudProvider,
+    },
     /// Stop the Telegraf service on all machines in the environment.
     ///
     /// This may be necessary for performing upgrades.
@@ -2076,6 +2097,38 @@ async fn main() -> Result<()> {
             testnet_deployer.status()?;
             Ok(())
         }
+        Commands::Stop {
+            custom_inventory,
+            forks,
+            interval,
+            name,
+            provider,
+        } => {
+            let testnet_deployer = TestnetDeployBuilder::default()
+                .ansible_forks(forks)
+                .environment_name(&name)
+                .provider(provider)
+                .build()?;
+
+            let inventory_service = DeploymentInventoryService::from(&testnet_deployer);
+            let inventory = inventory_service
+                .generate_or_retrieve_inventory(&name, true, None)
+                .await?;
+            if inventory.is_empty() {
+                return Err(eyre!("The {name} environment does not exist"));
+            }
+
+            let custom_inventory = if let Some(custom_inventory) = custom_inventory {
+                let custom_vms = get_custom_inventory(&inventory, &custom_inventory)?;
+                Some(custom_vms)
+            } else {
+                None
+            };
+
+            testnet_deployer.stop(interval, custom_inventory)?;
+
+            Ok(())
+        }
         Commands::StopTelegraf {
             custom_inventory,
             forks,