feat(rest-test): new rest test project

These are the initial bits of a new test project which leverages the new
control plane to test mayastor.
It has already revealed a few issues.

Cargo.toml

@@ -20,5 +20,6 @@ members = [
 	"control-plane/rest",
 	"control-plane/operators",
 	"control-plane/macros",
-	"control-plane/deployer"
+	"control-plane/deployer",
+	"control-plane/tests"
 ]

composer/src/lib.rs

@@ -113,11 +113,11 @@ impl Binary {
         let path = std::path::PathBuf::from(std::env!("CARGO_MANIFEST_DIR"));
         let srcdir = path.parent().unwrap().to_string_lossy();
 
-        Self::new(format!("{}/target/debug/{}", srcdir, name), vec![])
+        Self::new(&format!("{}/target/debug/{}", srcdir, name), vec![])
     }
     /// Setup nix shell binary from path and arguments
     pub fn from_nix(name: &str) -> Self {
-        Self::new(Self::which(name).expect("binary should exist"), vec![])
+        Self::new(name, vec![])
     }
     /// Add single argument
     /// Only one argument can be passed per use. So instead of:
@@ -171,11 +171,17 @@ impl Binary {
     }
     fn which(name: &str) -> std::io::Result<String> {
         let output = std::process::Command::new("which").arg(name).output()?;
+        if !output.status.success() {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::NotFound,
+                name,
+            ));
+        }
         Ok(String::from_utf8_lossy(&output.stdout).trim().into())
     }
-    fn new(path: String, args: Vec<String>) -> Self {
+    fn new(path: &str, args: Vec<String>) -> Self {
         Self {
-            path,
+            path: Self::which(path).expect("Binary path should exist!"),
             arguments: args,
             ..Default::default()
         }
@@ -310,6 +316,10 @@ pub struct Builder {
     containers: Vec<ContainerSpec>,
     /// the network for the tests used
     network: String,
+    /// reuse existing containers
+    reuse: bool,
+    /// allow cleaning up on a panic (if clean is true)
+    allow_clean_on_panic: bool,
     /// delete the container and network when dropped
     clean: bool,
     /// destroy existing containers if any
@@ -343,6 +353,8 @@ impl Builder {
             name: TEST_NET_NAME.to_string(),
             containers: Default::default(),
             network: "10.1.0.0/16".to_string(),
+            reuse: false,
+            allow_clean_on_panic: true,
             clean: true,
             prune: true,
             autorun: true,
@@ -420,8 +432,7 @@ impl Builder {
     }
 
     /// add a generic container which runs a local binary
-    pub fn add_container_bin(self, name: &str, mut bin: Binary) -> Builder {
-        bin.setup_nats(&self.name);
+    pub fn add_container_bin(self, name: &str, bin: Binary) -> Builder {
         self.add_container_spec(ContainerSpec::from_binary(name, bin))
     }
 
@@ -430,12 +441,25 @@ impl Builder {
         self.add_container_spec(ContainerSpec::from_binary(name, image))
     }
 
+    /// attempt to reuse and restart containers instead of starting new ones
+    pub fn with_reuse(mut self, reuse: bool) -> Builder {
+        self.reuse = reuse;
+        self.prune = !reuse;
+        self
+    }
+
     /// clean on drop?
     pub fn with_clean(mut self, enable: bool) -> Builder {
         self.clean = enable;
         self
     }
 
+    /// allow clean on panic if clean is set
+    pub fn with_clean_on_panic(mut self, enable: bool) -> Builder {
+        self.allow_clean_on_panic = enable;
+        self
+    }
+
     /// prune containers and networks on start
     pub fn with_prune(mut self, enable: bool) -> Builder {
         self.prune = enable;
@@ -463,14 +487,16 @@ impl Builder {
     }
 
     /// setup tracing for the cargo test code with `filter`
+    /// ignore when called multiple times
     pub fn with_tracing(self, filter: &str) -> Self {
-        if let Ok(filter) =
+        let builder = if let Ok(filter) =
             tracing_subscriber::EnvFilter::try_from_default_env()
         {
-            tracing_subscriber::fmt().with_env_filter(filter).init();
+            tracing_subscriber::fmt().with_env_filter(filter)
         } else {
-            tracing_subscriber::fmt().with_env_filter(filter).init();
-        }
+            tracing_subscriber::fmt().with_env_filter(filter)
+        };
+        builder.try_init().ok();
         self
     }
 
@@ -517,6 +543,8 @@ impl Builder {
             containers: Default::default(),
             ipam,
             label_prefix: "io.mayastor.test".to_string(),
+            reuse: self.reuse,
+            allow_clean_on_panic: self.allow_clean_on_panic,
             clean: self.clean,
             prune: self.prune,
             image: self.image,
@@ -526,14 +554,37 @@ impl Builder {
         compose.network_id =
             compose.network_create().await.map_err(|e| e.to_string())?;
 
-        // containers are created where the IPs are ordinal
-        for (i, spec) in self.containers.iter().enumerate() {
-            compose
-                .create_container(
-                    spec,
-                    &net.nth((i + 2) as u32).unwrap().to_string(),
-                )
-                .await?;
+        if self.reuse {
+            let containers =
+                compose.list_network_containers(&self.name).await?;
+
+            for container in containers {
+                let networks = container
+                    .network_settings
+                    .unwrap_or_default()
+                    .networks
+                    .unwrap_or_default();
+                if let Some(n) = container.names.unwrap_or_default().first() {
+                    if let Some(endpoint) = networks.get(&self.name) {
+                        if let Some(ip) = endpoint.ip_address.clone() {
+                            compose.containers.insert(
+                                n[1 ..].into(),
+                                (container.id.unwrap_or_default(), ip.parse()?),
+                            );
+                        }
+                    }
+                }
+            }
+        } else {
+            // containers are created where the IPs are ordinal
+            for (i, spec) in self.containers.iter().enumerate() {
+                compose
+                    .create_container(
+                        spec,
+                        &net.nth((i + 2) as u32).unwrap().to_string(),
+                    )
+                    .await?;
+            }
         }
 
         Ok(compose)
@@ -569,6 +620,10 @@ pub struct ComposeTest {
     /// prefix for labels set on containers and networks
     ///   $prefix.name = $name will be created automatically
     label_prefix: String,
+    /// reuse existing containers
+    reuse: bool,
+    /// allow cleaning up on a panic (if clean is set)
+    allow_clean_on_panic: bool,
     /// automatically clean up the things we have created for this test
     clean: bool,
     /// remove existing containers upon creation
@@ -591,10 +646,10 @@ impl Drop for ComposeTest {
             });
         }
 
-        if self.clean {
+        if self.clean && (!thread::panicking() || self.allow_clean_on_panic) {
             self.containers.keys().for_each(|c| {
                 std::process::Command::new("docker")
-                    .args(&["stop", c])
+                    .args(&["kill", c])
                     .output()
                     .unwrap();
                 std::process::Command::new("docker")
@@ -723,7 +778,7 @@ impl ComposeTest {
     }
 
     /// list containers
-    pub async fn list_containers(
+    pub async fn list_cluster_containers(
         &self,
     ) -> Result<Vec<ContainerSummaryInner>, Error> {
         self.docker
@@ -974,9 +1029,11 @@ impl ComposeTest {
                 ),
             },
         )?;
-        self.docker
-            .start_container::<&str>(id.0.as_str(), None)
-            .await?;
+        if !self.reuse {
+            self.docker
+                .start_container::<&str>(id.0.as_str(), None)
+                .await?;
+        }
 
         Ok(())
     }
@@ -1010,11 +1067,16 @@ impl ComposeTest {
 
     /// restart the container
     pub async fn restart(&self, name: &str) -> Result<(), Error> {
-        let id = self.containers.get(name).unwrap();
+        let (id, _) = self.containers.get(name).unwrap();
+        self.restart_id(id.as_str()).await
+    }
+
+    /// restart the container id
+    pub async fn restart_id(&self, id: &str) -> Result<(), Error> {
         if let Err(e) = self
             .docker
             .restart_container(
-                id.0.as_str(),
+                id,
                 Some(RestartContainerOptions {
                     t: 3,
                 }),
@@ -1115,6 +1177,22 @@ impl ComposeTest {
         result
     }
 
+    /// restart all the containers part of the network
+    /// returns the last error, if any or Ok
+    pub async fn restart_network_containers(&self) -> Result<(), Error> {
+        let mut result = Ok(());
+        let containers = self.list_network_containers(&self.name).await?;
+        for container in containers {
+            if let Some(id) = container.id {
+                if let Err(e) = self.restart_id(&id).await {
+                    println!("Failed to restart container id {:?}", id);
+                    result = Err(e);
+                }
+            }
+        }
+        result
+    }
+
     /// inspect the given container
     pub async fn inspect(
         &self,

control-plane/deployer/bin/src/deployer.rs

@@ -6,5 +6,5 @@ async fn main() -> Result<(), Error> {
     let cli_args = CliArgs::from_args();
     println!("Using options: {:?}", &cli_args);
 
-    cli_args.act().await
+    cli_args.execute().await
 }

control-plane/deployer/src/infra/mod.rs

@@ -235,6 +235,20 @@ pub fn build_error(name: &str, status: Option<i32>) -> Result<(), Error> {
 }
 
 impl Components {
+    pub async fn start_wait(
+        &self,
+        cfg: &ComposeTest,
+        timeout: std::time::Duration,
+    ) -> Result<(), Error> {
+        match tokio::time::timeout(timeout, self.start_wait_inner(cfg)).await {
+            Ok(result) => result,
+            Err(_) => {
+                let error = format!("Time out of {:?} expired", timeout);
+                Err(std::io::Error::new(std::io::ErrorKind::TimedOut, error)
+                    .into())
+            }
+        }
+    }
     pub async fn start(&self, cfg: &ComposeTest) -> Result<(), Error> {
         let mut last_done = None;
         for component in &self.0 {
@@ -254,6 +268,30 @@ impl Components {
         }
         Ok(())
     }
+    async fn start_wait_inner(&self, cfg: &ComposeTest) -> Result<(), Error> {
+        let mut last_done = None;
+        for component in &self.0 {
+            if let Some(last_done) = last_done {
+                if component.boot_order() == last_done {
+                    continue;
+                }
+            }
+            let components = self
+                .0
+                .iter()
+                .filter(|c| c.boot_order() == component.boot_order())
+                .collect::<Vec<&Component>>();
+
+            for component in &components {
+                component.start(&self.1, &cfg).await?;
+            }
+            for component in &components {
+                component.wait_on(&self.1, &cfg).await?;
+            }
+            last_done = Some(component.boot_order());
+        }
+        Ok(())
+    }
 }
 
 #[macro_export]
@@ -313,7 +351,7 @@ macro_rules! impl_component {
                 match tokio::time::timeout(timeout, self.wait_on_inner(cfg)).await {
                     Ok(result) => result,
                     Err(_) => {
-                        let error = format!("Timed out of {:?} expired", timeout);
+                        let error = format!("Time out of {:?} expired", timeout);
                         Err(std::io::Error::new(std::io::ErrorKind::TimedOut, error).into())
                     }
                 }

control-plane/deployer/src/lib.rs

@@ -150,13 +150,13 @@ impl StartOptions {
 
 impl CliArgs {
     /// Act upon the requested action
-    pub async fn act(&self) -> Result<(), Error> {
-        self.action.act().await
+    pub async fn execute(&self) -> Result<(), Error> {
+        self.action.execute().await
     }
 }
 
 impl Action {
-    async fn act(&self) -> Result<(), Error> {
+    async fn execute(&self) -> Result<(), Error> {
         match self {
             Action::Start(options) => options.start(self).await,
             Action::Stop(options) => options.stop(self).await,

control-plane/mbus-api/Cargo.toml

@@ -23,6 +23,7 @@ tracing-futures = "0.2.4"
 tracing-subscriber = "0.2.0"
 paperclip = { version = "0.5.0", features = ["actix3"] }
 percent-encoding = "2.1.0"
+uuid = { version = "0.7", features = ["v4"] }
 
 [dev-dependencies]
 composer = { path = "../../composer" }