implement down functionality

Signed-off-by: Brooks Townsend <[email protected]>

Cargo.toml

@@ -27,6 +27,7 @@ ctrlc = { workspace = true }
 dirs = { workspace = true }
 env_logger = { workspace = true }
 envmnt = { workspace = true }
+futures = { workspace = true }
 indicatif = { workspace = true }
 log = { workspace = true }
 nkeys = { workspace = true }

crates/wash-lib/src/start/wasmcloud.rs

@@ -87,8 +87,7 @@ pub async fn ensure_wasmcloud_for_os_arch_pair<P>(
 where
     P: AsRef<Path>,
 {
-    //TODO: check version with 0.63.0
-    // check_version(version)?;
+    check_version(version)?;
     if let Some(dir) = find_wasmcloud_binary(&dir, version).await {
         // wasmCloud already exists, return early
         return Ok(dir);
@@ -267,7 +266,7 @@ fn wasmcloud_url(os: &str, arch: &str, version: &str) -> String {
 }
 
 /// Helper function to ensure the version of wasmCloud is above the minimum
-/// supported version (v0.57.0) that runs mix releases
+/// supported version (v0.63.0) that runs burrito releases
 fn check_version(version: &str) -> Result<()> {
     let version_req = semver::VersionReq::parse(&format!(">={MINIMUM_WASMCLOUD_VERSION}"))?;
     match semver::Version::parse(version.trim_start_matches('v')) {
@@ -414,7 +413,7 @@ mod test {
     }
 
     const NATS_SERVER_VERSION: &str = "v2.8.4";
-    const WASMCLOUD_HOST_VERSION: &str = "v0.62.2-burrito-test";
+    const WASMCLOUD_HOST_VERSION: &str = "v0.63.0";
 
     #[tokio::test]
     async fn can_download_and_start_wasmcloud() -> anyhow::Result<()> {

src/down/mod.rs

@@ -1,21 +1,63 @@
 use std::collections::HashMap;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::process::{Output, Stdio};
-use std::time::Duration;
 
-use anyhow::Result;
+use anyhow::{anyhow, Result};
+use async_nats::Client;
 use clap::Parser;
+use log::warn;
 use serde_json::json;
 use tokio::process::Command;
 use wash_lib::cli::{CommandOutput, OutputKind};
-use wash_lib::start::{find_wasmcloud_binary, nats_pid_path, NATS_SERVER_BINARY, WADM_PID};
+use wash_lib::config::{DEFAULT_NATS_HOST, DEFAULT_NATS_PORT};
+use wash_lib::id::ServerId;
+use wash_lib::start::{nats_pid_path, NATS_SERVER_BINARY, WADM_PID};
 
 use crate::appearance::spinner::Spinner;
 use crate::cfg::cfg_dir;
-use crate::up::{DOWNLOADS_DIR, WASMCLOUD_PID_FILE};
+use crate::up::{
+    DEFAULT_LATTICE_PREFIX, DOWNLOADS_DIR, WASMCLOUD_CTL_CREDSFILE, WASMCLOUD_CTL_HOST,
+    WASMCLOUD_CTL_JWT, WASMCLOUD_CTL_PORT, WASMCLOUD_CTL_SEED, WASMCLOUD_LATTICE_PREFIX,
+};
+use crate::util::nats_client_from_opts;
 
 #[derive(Parser, Debug, Clone)]
-pub(crate) struct DownCommand {}
+pub(crate) struct DownCommand {
+    /// A lattice prefix is a unique identifier for a lattice, and is frequently used within NATS topics to isolate messages from different lattices
+    #[clap(
+            short = 'x',
+            long = "lattice-prefix",
+            default_value = DEFAULT_LATTICE_PREFIX,
+            env = WASMCLOUD_LATTICE_PREFIX,
+        )]
+    pub(crate) lattice_prefix: String,
+
+    /// An IP address or DNS name to use to connect to NATS for Control Interface (CTL) messages, defaults to the value supplied to --nats-host if not supplied
+    #[clap(long = "ctl-host", env = WASMCLOUD_CTL_HOST)]
+    pub(crate) ctl_host: Option<String>,
+
+    /// A port to use to connect to NATS for CTL messages, defaults to the value supplied to --nats-port if not supplied
+    #[clap(long = "ctl-port", env = WASMCLOUD_CTL_PORT)]
+    pub(crate) ctl_port: Option<u16>,
+
+    /// Convenience flag for CTL authentication, internally this parses the JWT and seed from the credsfile
+    #[clap(long = "ctl-credsfile", env = WASMCLOUD_CTL_CREDSFILE)]
+    pub(crate) ctl_credsfile: Option<PathBuf>,
+
+    /// A seed nkey to use to authenticate to NATS for CTL messages
+    #[clap(long = "ctl-seed", env = WASMCLOUD_CTL_SEED, requires = "ctl_jwt")]
+    pub(crate) ctl_seed: Option<String>,
+
+    /// A user JWT to use to authenticate to NATS for CTL messages
+    #[clap(long = "ctl-jwt", env = WASMCLOUD_CTL_JWT, requires = "ctl_seed")]
+    pub(crate) ctl_jwt: Option<String>,
+
+    #[clap(long = "host-id")]
+    pub host_id: Option<ServerId>,
+
+    #[clap(long = "all")]
+    pub all: bool,
+}
 
 pub(crate) async fn handle_command(
     command: DownCommand,
@@ -25,40 +67,41 @@ pub(crate) async fn handle_command(
 }
 
 pub(crate) async fn handle_down(
-    _cmd: DownCommand,
+    cmd: DownCommand,
     output_kind: OutputKind,
 ) -> Result<CommandOutput> {
     let install_dir = cfg_dir()?.join(DOWNLOADS_DIR);
     let sp = Spinner::new(&output_kind)?;
+    sp.update_spinner_message(" Stopping wasmCloud ...".to_string());
 
     let mut out_json = HashMap::new();
     let mut out_text = String::from("");
 
-    //TODO: stop host via nats instead of local
-    let version = tokio::fs::read_to_string(install_dir.join(WASMCLOUD_PID_FILE))
-        .await
-        .map_err(|e| {
-            anyhow::anyhow!("Unable to find wasmcloud pid file for stopping process: {e}")
-        })?;
-    let host_bin = find_wasmcloud_binary(&install_dir, &version)
-        .await
-        .ok_or_else(|| anyhow::anyhow!("Couldn't find path to wasmCloud binary. Is it running?"))?;
-    if host_bin.is_file() {
-        sp.update_spinner_message(" Stopping host ...".to_string());
-        if let Ok(output) = stop_wasmcloud(host_bin).await {
-            if output.stderr.is_empty() && output.stdout.is_empty() {
-                // if there was a host running, 'stop' has no output.
-                // Give it time to stop before stopping nats
-                tokio::time::sleep(Duration::from_secs(6)).await;
-                out_json.insert("host_stopped".to_string(), json!(true));
-                out_text.push_str("✅ wasmCloud host stopped successfully\n");
-            } else {
-                out_json.insert("host_stopped".to_string(), json!(true));
-                out_text.push_str(
-                    "🤔 Host did not appear to be running, assuming it's already stopped\n",
-                );
-            }
+    if let Ok(client) = nats_client_from_opts(
+        &cmd.ctl_host.unwrap_or(DEFAULT_NATS_HOST.to_string()),
+        &cmd.ctl_port
+            .map(|port| port.to_string())
+            .unwrap_or(DEFAULT_NATS_PORT.to_string()),
+        cmd.ctl_jwt,
+        cmd.ctl_seed,
+        cmd.ctl_credsfile,
+    )
+    .await
+    {
+        let (hosts, hosts_remain) =
+            stop_hosts(client, &cmd.lattice_prefix, &cmd.host_id, cmd.all).await?;
+        out_json.insert("hosts_stopped".to_string(), json!(hosts));
+        out_text.push_str("✅ wasmCloud hosts stopped successfully\n");
+        if hosts_remain {
+            out_json.insert("nats_stopped".to_string(), json!(false));
+            out_json.insert("wadm_stopped".to_string(), json!(false));
+            out_text.push_str(
+                "🛁 Exiting without stopping NATS or wadm, there are still hosts running",
+            );
+            return Ok(CommandOutput::new(out_text, out_json));
         }
+    } else {
+        warn!("Couldn't connect to NATS, unable to stop running hosts")
     }
 
     match stop_wadm(&install_dir).await {
@@ -69,7 +112,7 @@ pub(crate) async fn handle_down(
         }
         Err(e) => {
             out_json.insert("wadm_stopped".to_string(), json!(false));
-            out_text.push_str(&format!("❌ wadm did not stop successfully: {e:?}\n"));
+            out_text.push_str(&format!("❌ Could not stop wadm: {e:?}\n"));
         }
     }
 
@@ -94,6 +137,66 @@ pub(crate) async fn handle_down(
     Ok(CommandOutput::new(out_text, out_json))
 }
 
+/// Stop running wasmCloud hosts, returns a vector of host IDs that were stopped and
+/// a boolean indicating whether any hosts remain running
+async fn stop_hosts(
+    nats_client: Client,
+    lattice_prefix: &str,
+    host_id: &Option<ServerId>,
+    all: bool,
+) -> Result<(Vec<String>, bool)> {
+    let client = wasmcloud_control_interface::ClientBuilder::new(nats_client)
+        .lattice_prefix(lattice_prefix)
+        .auction_timeout(std::time::Duration::from_secs(2))
+        .build()
+        .await
+        .map_err(|e| anyhow!(e))?;
+
+    let hosts = client.get_hosts().await.map_err(|e| anyhow!(e))?;
+
+    // If a host ID was supplied, stop only that host
+    if let Some(host_id) = host_id {
+        let host_id_string = host_id.to_string();
+        client.stop_host(&host_id_string, None).await.map_err(|e| {
+            anyhow!(
+                "Could not stop host, ensure a host with that ID is running: {:?}",
+                e
+            )
+        })?;
+
+        Ok((vec![host_id_string], hosts.len() > 1))
+    } else {
+        if hosts.is_empty() {
+            Ok((vec![], false))
+        } else if hosts.len() == 1 {
+            let host_id = &hosts[0].id;
+            client
+                .stop_host(host_id, None)
+                .await
+                .map_err(|e| anyhow!(e))?;
+            Ok((vec![host_id.to_string()], false))
+        } else if all {
+            let host_stops = hosts
+                .iter()
+                .map(|host| async {
+                    let host_id = &host.id;
+                    let _ = client
+                        .stop_host(&host_id, None)
+                        .await
+                        // TODO: deal with this error
+                        .map_err(|e| anyhow!(e));
+                    host_id.to_owned()
+                })
+                .collect::<Vec<_>>();
+            Ok((futures::future::join_all(host_stops).await, false))
+        } else {
+            Err(anyhow!(
+                "More than one host is running, please specify a host ID or use --all\nRunning hosts: {:?}", hosts.into_iter().map(|h| h.id).collect::<Vec<_>>()
+            ))
+        }
+    }
+}
+
 /// Helper function to send wasmCloud the `stop` command and wait for it to clean up
 pub(crate) async fn stop_wasmcloud<P>(bin_path: P) -> Result<Output>
 where

src/up/config.rs

@@ -11,7 +11,7 @@ pub(crate) const DEFAULT_NATS_PORT: &str = "4222";
 // wadm configuration values
 pub(crate) const WADM_VERSION: &str = "v0.4.0-alpha.2";
 // wasmCloud configuration values, https://wasmcloud.dev/reference/host-runtime/host_configure/
-pub(crate) const WASMCLOUD_HOST_VERSION: &str = "v0.62.2-burrito-test";
+pub(crate) const WASMCLOUD_HOST_VERSION: &str = "v0.63.0";
 pub(crate) const WASMCLOUD_DASHBOARD_PORT: &str = "WASMCLOUD_DASHBOARD_PORT";
 pub(crate) const DEFAULT_DASHBOARD_PORT: &str = "4000";
 // NATS isolation configuration variables

src/up/mod.rs

@@ -36,9 +36,7 @@ use crate::util::nats_client_from_opts;
 
 mod config;
 mod credsfile;
-pub use config::DOWNLOADS_DIR;
-pub use config::WASMCLOUD_PID_FILE;
-use config::*;
+pub use config::*;
 
 const LOCALHOST: &str = "127.0.0.1";
 
@@ -425,18 +423,10 @@ pub(crate) async fn handle_up(cmd: UpCommand, output_kind: OutputKind) -> Result
 
         let spinner = Spinner::new(&output_kind)?;
         spinner.update_spinner_message(
-            "CTRL+c received, gracefully stopping wasmCloud, wadm, and NATS...".to_string(),
+            // wasmCloud and NATS both exit immediately when sent SIGINT
+            "CTRL+c received, stopping wasmCloud, wadm, and NATS...".to_string(),
         );
 
-        // Terminate wasmCloud and NATS processes
-        let output = stop_wasmcloud(wasmcloud_executable.clone()).await?;
-        if !output.status.success() {
-            log::warn!("wasmCloud exited with a non-zero exit status, processes may need to be cleaned up manually")
-        }
-        if !cmd.nats_opts.connect_only {
-            stop_nats(&install_dir).await?;
-        }
-
         // remove wadm pidfile, the process is stopped automatically by CTRL+c
         tokio::fs::remove_file(&install_dir.join("wadm.pid")).await?;
 
@@ -558,6 +548,8 @@ async fn run_wasmcloud_interactive(
     if let Some(handle) = handle {
         handle.abort()
     };
+
+    wasmcloud_child.kill().await?;
     Ok(())
 }