Storage Strategy for Kubernetes Backend

This commit will enable users to apply kubernetes specific configuration
to create and mount PVC for the companions. The PVC created is unique
with labels to identify them. The specified storage class should ensure
dynamic PV creation.

Additionally size specifications will now be parsed as ByteSize with
backwards compatibility.

api/Cargo.toml

@@ -15,6 +15,7 @@ path = "src/main.rs"
 async-trait = "0.1"
 base64 = "0.21"
 boa_engine = "0.17"
+bytesize = { version = "1.3", features = ["serde"] }
 chrono = { version = "0.4", features = ["serde"] }
 clap = { version = "3.2", features = ["derive", "cargo"] }
 env_logger = "0.10"

api/README.md

@@ -17,6 +17,14 @@ type = 'Kubernetes'
 # This information is crucial if you run PREvant behind a Traefik instance that enforces the user ot be
 # logged in.
 labelsPath = '/run/podinfo/labels'
+
+[runtime.storageConfig]
+# Size of the storage space that is reserved and mounted to the deployed companion with storage.
+# If unspecified storage is defaulted to 2g.
+storageSize = '10g'
+# Storage class denotes the type of storage to be used for companions deployed with storage.
+# Manually managed storage classes can be specified here. If unspecified default storage class will be used.
+storageClass = 'local-path'
 ```
 
 ## Container Options

api/src/config/container.rs

@@ -23,35 +23,39 @@
  * THE SOFTWARE.
  * =========================LICENSE_END==================================
  */
+use bytesize::ByteSize;
 use serde::{de, Deserialize, Deserializer};
 
 #[derive(Clone, Default, Deserialize)]
 pub struct ContainerConfig {
     #[serde(deserialize_with = "ContainerConfig::parse_from_memory_string")]
-    memory_limit: Option<u64>,
+    memory_limit: Option<ByteSize>,
 }
 
 impl ContainerConfig {
-    fn parse_from_memory_string<'de, D>(deserializer: D) -> Result<Option<u64>, D::Error>
+    fn parse_from_memory_string<'de, D>(deserializer: D) -> Result<Option<ByteSize>, D::Error>
     where
         D: Deserializer<'de>,
     {
         let container_limit = String::deserialize(deserializer)?;
+        match container_limit.parse::<ByteSize>() {
+            Ok(result) => Ok(Some(result)),
+            Err(_) => {
+                let (size, unit) = container_limit.split_at(container_limit.len() - 1);
+                let limit = size.parse::<u64>().map_err(de::Error::custom)?;
 
-        let (size, unit) = container_limit.split_at(container_limit.len() - 1);
-        let limit = size.parse::<u64>().map_err(de::Error::custom)?;
-
-        let exp = match unit.to_lowercase().as_str() {
-            "k" => 1,
-            "m" => 2,
-            "g" => 3,
-            _ => 0,
-        };
-
-        Ok(Some(limit * 1024_u64.pow(exp)))
+                let exp = match unit.to_lowercase().as_str() {
+                    "k" => 1,
+                    "m" => 2,
+                    "g" => 3,
+                    _ => 0,
+                };
+                Ok(Some(ByteSize(limit * 1024_u64.pow(exp))))
+            }
+        }
     }
 
-    pub fn memory_limit(&self) -> Option<u64> {
+    pub fn memory_limit(&self) -> Option<ByteSize> {
         self.memory_limit
     }
 }

api/src/config/runtime.rs

@@ -1,5 +1,3 @@
-use std::path::PathBuf;
-
 /*-
  * ========================LICENSE_START=================================
  * PREvant REST API
@@ -25,6 +23,10 @@ use std::path::PathBuf;
  * THE SOFTWARE.
  * =========================LICENSE_END==================================
  */
+use bytesize::ByteSize;
+use serde::Deserialize;
+use std::path::PathBuf;
+
 #[derive(Clone, Debug, Deserialize, PartialEq)]
 #[serde(tag = "type")]
 pub enum Runtime {
@@ -43,12 +45,18 @@ impl Default for Runtime {
 pub struct KubernetesRuntimeConfig {
     #[serde(default)]
     downward_api: KubernetesDownwardApiConfig,
+    #[serde(default)]
+    storage_config: KubernetesStorageConfig,
 }
 
 impl KubernetesRuntimeConfig {
     pub fn downward_api(&self) -> &KubernetesDownwardApiConfig {
         &self.downward_api
     }
+
+    pub fn storage_config(&self) -> &KubernetesStorageConfig {
+        &self.storage_config
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, PartialEq)]
@@ -71,6 +79,37 @@ impl Default for KubernetesDownwardApiConfig {
     }
 }
 
+#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct KubernetesStorageConfig {
+    #[serde(default = "KubernetesStorageConfig::default_storage_size")]
+    storage_size: ByteSize,
+    storage_class: Option<String>,
+}
+
+impl KubernetesStorageConfig {
+    pub fn storage_size(&self) -> &ByteSize {
+        &self.storage_size
+    }
+
+    pub fn storage_class(&self) -> &Option<String> {
+        &self.storage_class
+    }
+
+    fn default_storage_size() -> ByteSize {
+        ByteSize::gb(2)
+    }
+}
+
+impl Default for KubernetesStorageConfig {
+    fn default() -> Self {
+        Self {
+            storage_size: Self::default_storage_size(),
+            storage_class: None,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -112,7 +151,8 @@ mod tests {
             Runtime::Kubernetes(KubernetesRuntimeConfig {
                 downward_api: KubernetesDownwardApiConfig {
                     labels_path: PathBuf::from("/some/path")
-                }
+                },
+                storage_config: KubernetesStorageConfig::default()
             })
         );
     }
@@ -130,4 +170,27 @@ mod tests {
             &PathBuf::from("/run/podinfo/labels")
         )
     }
+
+    #[test]
+    fn parse_as_kubernetes_storage_config() {
+        let runtime_toml = r#"
+        type = 'Kubernetes'
+        [storageConfig]
+        storageSize = '10g'
+        storageClass = 'local-path'
+        "#;
+
+        let runtime = toml::de::from_str::<Runtime>(runtime_toml).unwrap();
+
+        assert_eq!(
+            runtime,
+            Runtime::Kubernetes(KubernetesRuntimeConfig {
+                downward_api: KubernetesDownwardApiConfig::default(),
+                storage_config: KubernetesStorageConfig {
+                    storage_size: ByteSize::gb(10),
+                    storage_class: Some(String::from("local-path"))
+                }
+            })
+        );
+    }
 }

api/src/infrastructure/docker.rs

@@ -484,8 +484,8 @@ impl DockerInfrastructure {
         options.restart_policy("always", 5);
 
         if let Some(memory_limit) = container_config.memory_limit() {
-            options.memory(memory_limit);
-            options.memory_swap(memory_limit as i64);
+            options.memory(memory_limit.as_u64());
+            options.memory_swap(memory_limit.as_u64() as i64);
         }
 
         options.build()

api/src/infrastructure/kubernetes/infrastructure.rs

@@ -25,11 +25,12 @@
  */
 use super::super::{
     APP_NAME_LABEL, CONTAINER_TYPE_LABEL, IMAGE_LABEL, REPLICATED_ENV_LABEL, SERVICE_NAME_LABEL,
+    STORAGE_TYPE_LABEL,
 };
 use super::payloads::{
     deployment_payload, deployment_replicas_payload, image_pull_secret_payload,
-    ingress_route_payload, middleware_payload, namespace_payload, secrets_payload, service_payload,
-    IngressRoute, Middleware,
+    ingress_route_payload, middleware_payload, namespace_payload, persistent_volume_claim_payload,
+    secrets_payload, service_payload, IngressRoute, Middleware,
 };
 use crate::config::{Config as PREvantConfig, ContainerConfig, Runtime};
 use crate::deployment::deployment_unit::{DeployableService, DeploymentUnit};
@@ -41,20 +42,22 @@ use async_trait::async_trait;
 use chrono::{DateTime, FixedOffset, Utc};
 use failure::Error;
 use futures::future::join_all;
+use k8s_openapi::api::storage::v1::StorageClass;
 use k8s_openapi::api::{
     apps::v1::Deployment as V1Deployment, core::v1::Namespace as V1Namespace,
-    core::v1::Pod as V1Pod, core::v1::Secret as V1Secret, core::v1::Service as V1Service,
+    core::v1::PersistentVolumeClaim, core::v1::Pod as V1Pod, core::v1::Secret as V1Secret,
+    core::v1::Service as V1Service,
 };
 use kube::{
     api::{Api, DeleteParams, ListParams, LogParams, Patch, PatchParams, PostParams},
     client::Client,
     config::Config,
     error::{Error as KubeError, ErrorResponse},
 };
-use log::debug;
+use log::{debug, warn};
 use multimap::MultiMap;
 use secstr::SecUtf8;
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 use std::convert::{From, TryFrom};
 use std::net::IpAddr;
 use std::path::PathBuf;
@@ -88,6 +91,8 @@ pub enum KubernetesInfrastructureError {
         deployment_name
     )]
     MissingImageLabel { deployment_name: String },
+    #[fail(display = "The default storage class is missing in kubernetes.")]
+    MissingDefaultStorageClass,
 }
 
 impl KubernetesInfrastructure {
@@ -327,6 +332,10 @@ impl KubernetesInfrastructure {
                 );
                 Ok(())
             }
+            Err(KubeError::Api(ErrorResponse { code, .. })) if code == 409 => {
+                debug!("Secrets already exists for {}", app_name);
+                Ok(())
+            }
             Err(e) => {
                 error!("Cannot deploy namespace: {}", e);
                 Err(e.into())
@@ -346,6 +355,10 @@ impl KubernetesInfrastructure {
 
         let client = self.client().await?;
 
+        let persistence_volume_map = self
+            .create_persistent_volume_claim(app_name, service)
+            .await?;
+
         match Api::namespaced(client.clone(), app_name)
             .create(
                 &PostParams::default(),
@@ -356,6 +369,7 @@ impl KubernetesInfrastructure {
                     self.config
                         .registry_credentials(&service.image().registry().unwrap_or_default())
                         .is_some(),
+                    &persistence_volume_map,
                 ),
             )
             .await
@@ -387,6 +401,7 @@ impl KubernetesInfrastructure {
                                     &service.image().registry().unwrap_or_default(),
                                 )
                                 .is_some(),
+                            &persistence_volume_map,
                         )),
                     )
                     .await?;
@@ -478,6 +493,111 @@ impl KubernetesInfrastructure {
 
         Ok(service)
     }
+
+    async fn create_persistent_volume_claim<'a>(
+        &self,
+        app_name: &str,
+        service: &'a DeployableService,
+    ) -> Result<Option<HashMap<&'a String, PersistentVolumeClaim>>, KubernetesInfrastructureError>
+    {
+        let client = self.client().await?;
+        let Runtime::Kubernetes(k8s_config) = self.config.runtime_config() else {return Ok(None)};
+
+        let storage_size = k8s_config.storage_config().storage_size();
+        let storage_class = match k8s_config.storage_config().storage_class() {
+            Some(sc) => sc.into(),
+            None => self
+                .fetch_default_storage_class()
+                .await?
+                .metadata
+                .name
+                .ok_or(KubernetesInfrastructureError::UnexpectedError {
+                    internal_message: String::from(
+                        "The default storage class contains an empty name",
+                    ),
+                })?,
+        };
+
+        let mut persistent_volume_map = HashMap::new();
+        let existing_pvc: Api<PersistentVolumeClaim> = Api::namespaced(client.clone(), app_name);
+
+        for declared_volume in service.declared_volumes() {
+            let pvc_list_params = ListParams {
+                label_selector: Some(format!(
+                    "{}={},{}={},{}={}",
+                    APP_NAME_LABEL,
+                    app_name,
+                    SERVICE_NAME_LABEL,
+                    service.service_name(),
+                    STORAGE_TYPE_LABEL,
+                    declared_volume.split('/').last().unwrap_or("default")
+                )),
+                ..Default::default()
+            };
+
+            let fetched_pvc = existing_pvc.list(&pvc_list_params).await?.items;
+
+            if fetched_pvc.is_empty() {
+                match Api::namespaced(client.clone(), app_name)
+                    .create(
+                        &PostParams::default(),
+                        &persistent_volume_claim_payload(
+                            app_name,
+                            service,
+                            storage_size,
+                            &storage_class,
+                            declared_volume,
+                        ),
+                    )
+                    .await
+                {
+                    Ok(pvc) => {
+                        persistent_volume_map.insert(declared_volume, pvc);
+                    }
+                    Err(e) => {
+                        error!("Cannot deploy persistent volume claim: {}", e);
+                        return Err(e.into());
+                    }
+                }
+            } else {
+                if fetched_pvc.len() != 1 {
+                    warn!(
+                        "Found more than 1 Persistent Volume Claim - {:?} for declared image path {} \n Using the first available Persistent Volume Claim - {:?}",
+                        &fetched_pvc.iter().map(|pvc| &pvc.metadata.name),
+                        declared_volume,
+                        fetched_pvc.first().unwrap().metadata.name
+                    );
+                }
+
+                persistent_volume_map
+                    .insert(declared_volume, fetched_pvc.into_iter().next().unwrap());
+            }
+        }
+        Ok(Some(persistent_volume_map))
+    }
+
+    async fn fetch_default_storage_class(
+        &self,
+    ) -> Result<StorageClass, KubernetesInfrastructureError> {
+        let storage_classes: Api<StorageClass> = Api::all(self.client().await?);
+
+        match storage_classes.list(&ListParams::default()).await {
+            Ok(sc) => sc
+                .items
+                .into_iter()
+                .find(|sc| {
+                    sc.metadata.annotations.as_ref().map_or_else(
+                        || false,
+                        |v| {
+                            v.get("storageclass.kubernetes.io/is-default-class")
+                                == Some(&"true".into())
+                        },
+                    )
+                })
+                .ok_or(KubernetesInfrastructureError::MissingDefaultStorageClass),
+            Err(err) => Err(err.into()),
+        }
+    }
 }
 
 #[async_trait]