[CONTINT-4562] Add a parameter to continuously stop and recreate containers

Author: L3n41c

CHANGELOG.md

@@ -5,11 +5,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+## Added
+- Add a `max_lifetime` parameter to the `container` generator to make it generate
+  a continuous stream of docker containers deletion and re-creation.
 
 ## [0.25.9]
 ## Added
 - Introduce a `container` generator able to generate an arbitrary number
-  of docker containers
+  of docker containers.
 ## Changed
 - Lading's byte-unit crate is now updated to 5.x. This version of the crate is
   very strict about the difference between MiB, Mb etc.

examples/lading-container.yaml

@@ -19,4 +19,5 @@ generator:
       network_disabled: false
       exposed_ports:
         - 5000/tcp
+      max_lifetime: 20
       number_of_containers: 10

lading/src/generator/container.rs

@@ -12,14 +12,14 @@ use bollard::container::{
 use bollard::image::CreateImageOptions;
 use bollard::secret::ContainerCreateResponse;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use tokio_stream::StreamExt;
 use tracing::{debug, info, warn};
 use uuid::Uuid;
 
 use super::General;
 
-#[derive(Debug, Deserialize, Serialize, PartialEq)]
+#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
 #[serde(deny_unknown_fields)]
 /// Configuration of the container generator.
 pub struct Config {
@@ -37,8 +37,10 @@ pub struct Config {
     pub network_disabled: Option<bool>,
     /// Ports to expose from the container
     pub exposed_ports: Option<Vec<String>>,
+    /// Maximum lifetime of containers before being replaced
+    pub max_lifetime: Option<u64>,
     /// Number of containers to spin up (defaults to 1)
-    pub number_of_containers: Option<u32>,
+    pub number_of_containers: Option<usize>,
 }
 
 /// Errors produced by the `Container` generator.
@@ -55,14 +57,7 @@ pub enum Error {
 #[derive(Debug)]
 /// Represents a container that can be spun up from a configured image.
 pub struct Container {
-    image: String,
-    tag: String,
-    args: Option<Vec<String>>,
-    env: Option<Vec<String>>,
-    labels: Option<HashMap<String, String>>,
-    network_disabled: Option<bool>,
-    exposed_ports: Option<Vec<String>>,
-    number_of_containers: usize,
+    config: Config,
     shutdown: lading_signal::Watcher,
 }
 
@@ -80,50 +75,11 @@ impl Container {
         shutdown: lading_signal::Watcher,
     ) -> Result<Self, Error> {
         Ok(Self {
-            image: config.repository.clone(),
-            tag: config.tag.clone(),
-            args: config.args.clone(),
-            env: config.env.clone(),
-            labels: config.labels.clone(),
-            network_disabled: config.network_disabled,
-            exposed_ports: config.exposed_ports.clone(),
-            number_of_containers: config.number_of_containers.unwrap_or(1) as usize,
+            config: config.clone(),
             shutdown,
         })
     }
 
-    /// Convert the `Container` instance to a `ContainerConfig` for the Docker API.
-    #[must_use]
-    fn to_container_config<'a>(&'a self, full_image: &'a str) -> ContainerConfig<&'a str> {
-        ContainerConfig {
-            image: Some(full_image),
-            tty: Some(true),
-            cmd: self
-                .args
-                .as_ref()
-                .map(|args| args.iter().map(String::as_str).collect()),
-            env: self
-                .env
-                .as_ref()
-                .map(|env| env.iter().map(String::as_str).collect()),
-            labels: self.labels.as_ref().map(|labels| {
-                labels
-                    .iter()
-                    .map(|(key, value)| (key.as_str(), value.as_str()))
-                    .collect()
-            }),
-            network_disabled: self.network_disabled,
-            #[allow(clippy::zero_sized_map_values)]
-            exposed_ports: self.exposed_ports.as_ref().map(|ports| {
-                ports
-                    .iter()
-                    .map(|port| (port.as_str(), HashMap::new()))
-                    .collect()
-            }),
-            ..Default::default()
-        }
-    }
-
     /// Run the `Container` generator.
     ///
     /// # Errors
@@ -138,11 +94,14 @@ impl Container {
     /// 4. Wait for shutdown signal.
     /// 5. On shutdown, stop and remove the container.
     pub async fn spin(self) -> Result<(), Error> {
-        info!("Container generator running: {}:{}", self.image, self.tag);
+        info!(
+            "Container generator running: {}:{}",
+            self.config.repository, self.config.tag
+        );
 
         let docker = Docker::connect_with_local_defaults()?;
 
-        let full_image = format!("{}:{}", self.image, self.tag);
+        let full_image = format!("{}:{}", self.config.repository, self.config.tag);
         debug!("Ensuring image is available: {full_image}");
 
         // Pull the image
@@ -169,46 +128,35 @@ impl Container {
             }
         }
 
-        let mut containers = Vec::with_capacity(self.number_of_containers);
-
-        for _ in 0..self.number_of_containers {
-            let container_name = format!("lading_container_{}", Uuid::new_v4());
-            debug!("Creating container: {container_name}");
-
-            let container = docker
-                .create_container(
-                    Some(CreateContainerOptions {
-                        name: &container_name,
-                        platform: None,
-                    }),
-                    self.to_container_config(&full_image),
-                )
-                .await?;
-
-            debug!("Created container with id: {id}", id = container.id);
-            for warning in &container.warnings {
-                warn!("Container warning: {warning}");
-            }
-
-            containers.push(container);
-        }
-
-        for container in &containers {
-            docker
-                .start_container(&container.id, None::<StartContainerOptions<String>>)
-                .await?;
+        let number_of_containers = self.config.number_of_containers.unwrap_or(1);
+        let mut containers = VecDeque::with_capacity(number_of_containers);
 
-            debug!("Started container: {id}", id = container.id);
+        for _ in 0..number_of_containers {
+            containers.push_back(
+                self.config
+                    .create_and_start_container(&docker, &full_image)
+                    .await?,
+            );
         }
 
         // Wait for shutdown signal
         let shutdown_wait = self.shutdown.recv();
         tokio::pin!(shutdown_wait);
-        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(10));
+        let mut recreate_interval = self.config.max_lifetime.map(|max_lifetime| {
+            tokio::time::interval(tokio::time::Duration::from_millis(
+                1_000 * max_lifetime / number_of_containers as u64,
+            ))
+        });
+        let mut liveness_interval = tokio::time::interval(tokio::time::Duration::from_secs(10));
         loop {
             tokio::select! {
+                // Destroy and replace containers
+                _ = if let Some(ref mut interval) = recreate_interval { interval.tick() } else { std::future::pending().await } => {
+                    stop_and_remove_container(&docker, &containers.pop_front().ok_or(Error::Generic(String::from("No container left")))?).await?;
+                    containers.push_back(self.config.create_and_start_container(&docker, &full_image).await?);
+                }
                 // Check that containers are still running every 10 seconds
-                _ = interval.tick() => {
+                _ = liveness_interval.tick() => {
                     for container in &containers {
                         if let Some(state) = docker.inspect_container(&container.id, None).await?.state {
                             if !state.running.unwrap_or(false) {
@@ -223,39 +171,105 @@ impl Container {
                 () = &mut shutdown_wait => {
                     debug!("shutdown signal received");
                     for container in &containers {
-                        Self::stop_and_remove_container(&docker, container).await?;
+                        stop_and_remove_container(&docker, container).await?;
                     }
 
                     return Ok(())
                 }
             }
         }
     }
+}
 
-    async fn stop_and_remove_container(
-        docker: &Docker,
-        container: &ContainerCreateResponse,
-    ) -> Result<(), Error> {
-        info!("Stopping container: {id}", id = container.id);
-        if let Err(e) = docker
-            .stop_container(&container.id, Some(StopContainerOptions { t: 5 }))
-            .await
-        {
-            warn!("Error stopping container {id}: {e}", id = container.id);
+impl Config {
+    /// Convert the `Container` instance to a `ContainerConfig` for the Docker API.
+    #[must_use]
+    fn to_container_config<'a>(&'a self, full_image: &'a str) -> ContainerConfig<&'a str> {
+        ContainerConfig {
+            image: Some(full_image),
+            tty: Some(true),
+            cmd: self
+                .args
+                .as_ref()
+                .map(|args| args.iter().map(String::as_str).collect()),
+            env: self
+                .env
+                .as_ref()
+                .map(|env| env.iter().map(String::as_str).collect()),
+            labels: self.labels.as_ref().map(|labels| {
+                labels
+                    .iter()
+                    .map(|(key, value)| (key.as_str(), value.as_str()))
+                    .collect()
+            }),
+            network_disabled: self.network_disabled,
+            #[allow(clippy::zero_sized_map_values)]
+            exposed_ports: self.exposed_ports.as_ref().map(|ports| {
+                ports
+                    .iter()
+                    .map(|port| (port.as_str(), HashMap::new()))
+                    .collect()
+            }),
+            ..Default::default()
         }
+    }
 
-        debug!("Removing container: {id}", id = container.id);
-        docker
-            .remove_container(
-                &container.id,
-                Some(RemoveContainerOptions {
-                    force: true,
-                    ..Default::default()
+    async fn create_and_start_container(
+        &self,
+        docker: &Docker,
+        full_image: &str,
+    ) -> Result<ContainerCreateResponse, Error> {
+        let container_name = format!("lading_container_{}", Uuid::new_v4());
+        debug!("Creating container: {container_name}");
+
+        let container = docker
+            .create_container(
+                Some(CreateContainerOptions {
+                    name: &container_name,
+                    platform: None,
                 }),
+                self.to_container_config(full_image),
             )
             .await?;
 
-        debug!("Removed container: {id}", id = container.id);
-        Ok(())
+        debug!("Created container with id: {id}", id = container.id);
+        for warning in &container.warnings {
+            warn!("Container warning: {warning}");
+        }
+
+        docker
+            .start_container(&container.id, None::<StartContainerOptions<String>>)
+            .await?;
+
+        debug!("Started container: {id}", id = container.id);
+
+        Ok(container)
     }
 }
+
+async fn stop_and_remove_container(
+    docker: &Docker,
+    container: &ContainerCreateResponse,
+) -> Result<(), Error> {
+    info!("Stopping container: {id}", id = container.id);
+    if let Err(e) = docker
+        .stop_container(&container.id, Some(StopContainerOptions { t: 0 }))
+        .await
+    {
+        warn!("Error stopping container {id}: {e}", id = container.id);
+    }
+
+    debug!("Removing container: {id}", id = container.id);
+    docker
+        .remove_container(
+            &container.id,
+            Some(RemoveContainerOptions {
+                force: true,
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    debug!("Removed container: {id}", id = container.id);
+    Ok(())
+}