feat: Implement asleep feature

bootstrap/stage1/crd.tf

@@ -50,6 +50,10 @@ resource "kubernetes_manifest" "customresourcedefinition_hydradoomnodes_hydra_do
               "properties" = {
                 "spec" = {
                   "properties" = {
+                    "asleep" = {
+                      "nullable" = true
+                      "type"     = "boolean"
+                    }
                     "commitInputs" = {
                       "items" = {
                         "type" = "string"

bootstrap/stage1/efs.tf

@@ -0,0 +1,36 @@
+resource "kubernetes_storage_class" "efs_storage_class" {
+  metadata {
+    name = "efs-sc"
+  }
+  storage_provisioner = "efs.csi.aws.com"
+  parameters = {
+    provisioningMode      = "efs-ap"
+    fileSystemId          = var.efs_fs_id
+    directoryPerms        = "777"
+    basePath              = "/hydra-node-persistance"
+    subPathPattern        = "$${.PVC.name}"
+    ensureUniqueDirectory = "true"
+  }
+}
+
+resource "kubernetes_persistent_volume" "efs_pv" {
+  metadata {
+    name = "hydra-doom-persistence"
+  }
+
+  spec {
+    capacity = {
+      storage = "100Gi"
+    }
+    volume_mode                      = "Filesystem"
+    access_modes                     = ["ReadWriteMany"]
+    persistent_volume_reclaim_policy = "Retain"
+    storage_class_name               = "efs-cs"
+    persistent_volume_source {
+      csi {
+        driver        = "efs.csi.aws.com"
+        volume_handle = var.efs_fs_id
+      }
+    }
+  }
+}

bootstrap/stage1/main.tf

@@ -1 +1,9 @@
-# Add Cardano Node and Ingress support.
+variable "efs_fs_id" {
+  type        = string
+  description = "ID of EFS resource to use as persistance."
+}
+
+variable "efs_ap" {
+  type        = string
+  description = "Access point of corresponding EFS."
+}

src/controller.rs

@@ -1,7 +1,7 @@
 use anyhow::bail;
 use k8s_openapi::api::{
-    apps::v1::Deployment,
-    core::v1::{ConfigMap, Service},
+    apps::v1::StatefulSet,
+    core::v1::{ConfigMap, PersistentVolumeClaim, Service},
     networking::v1::Ingress,
 };
 use kube::{
@@ -23,6 +23,7 @@ pub enum HydraDoomNodeState {
     Online,
     HeadIsInitializing,
     HeadIsOpen,
+    Sleeping,
 }
 impl From<f64> for HydraDoomNodeState {
     fn from(value: f64) -> Self {
@@ -41,6 +42,7 @@ impl From<HydraDoomNodeState> for String {
             HydraDoomNodeState::Online => "Online".to_string(),
             HydraDoomNodeState::HeadIsInitializing => "HeadIsInitializing".to_string(),
             HydraDoomNodeState::HeadIsOpen => "HeadIsOpen".to_string(),
+            HydraDoomNodeState::Sleeping => "Sleeping".to_string(),
         }
     }
 }
@@ -62,10 +64,14 @@ pub struct K8sConstants {
     pub state_metric: String,
     pub transactions_metric: String,
     pub dmtrctl_image: String,
+    pub storage_class_name: String,
+    pub persistence_dir_storage_request: String,
 }
 impl Default for K8sConstants {
     fn default() -> Self {
         Self {
+            persistence_dir_storage_request: "5Gi".to_string(),
+            storage_class_name: "efs-sc".to_string(),
             config_dir: "/etc/config".to_string(),
             secret_dir: "/var/secret".to_string(),
             socket_dir: "/ipc".to_string(),
@@ -128,13 +134,14 @@ impl K8sContext {
     pub async fn patch(&self, crd: &HydraDoomNode) -> anyhow::Result<()> {
         info!("Running patch");
         match tokio::join!(
-            self.patch_deployment(crd),
+            self.patch_sts(crd),
             self.patch_service(crd),
             self.patch_ingress(crd),
             self.patch_configmap(crd),
+            self.patch_pvc(crd),
             self.patch_crd(crd)
         ) {
-            (Ok(_), Ok(_), Ok(_), Ok(_), Ok(_)) => (),
+            (Ok(_), Ok(_), Ok(_), Ok(_), Ok(_), Ok(_)) => (),
             _ => bail!("Failed to apply patch for components."),
         };
 
@@ -143,12 +150,13 @@ impl K8sContext {
 
     pub async fn delete(&self, crd: &HydraDoomNode) -> anyhow::Result<()> {
         match tokio::join!(
-            self.remove_deployment(crd),
+            self.remove_sts(crd),
             self.remove_service(crd),
             self.remove_ingress(crd),
-            self.remove_configmap(crd)
+            self.remove_configmap(crd),
+            self.remove_pvc(crd)
         ) {
-            (Ok(_), Ok(_), Ok(_), Ok(_)) => Ok(()),
+            (Ok(_), Ok(_), Ok(_), Ok(_), Ok(_)) => Ok(()),
             _ => bail!("Failed to remove resources"),
         }
     }
@@ -173,6 +181,35 @@ impl K8sContext {
         })
     }
 
+    async fn patch_pvc(&self, crd: &HydraDoomNode) -> anyhow::Result<PersistentVolumeClaim> {
+        let api: Api<PersistentVolumeClaim> =
+            Api::namespaced(self.client.clone(), &crd.namespace().unwrap());
+
+        // Create or patch the configmap
+        api.patch(
+            &crd.internal_name(),
+            &PatchParams::apply("hydra-doom-pod-controller"),
+            &Patch::Apply(&crd.configmap(&self.config, &self.constants)),
+        )
+        .await
+        .map_err(|err| {
+            error!(err = err.to_string(), "Failed to create pvc.");
+            err.into()
+        })
+    }
+
+    async fn remove_pvc(&self, crd: &HydraDoomNode) -> anyhow::Result<()> {
+        let api: Api<PersistentVolumeClaim> =
+            Api::namespaced(self.client.clone(), &crd.namespace().unwrap());
+        match api
+            .delete(&crd.internal_name(), &DeleteParams::default())
+            .await
+        {
+            Ok(_) => Ok(()),
+            Err(e) => Err(e.into()),
+        }
+    }
+
     async fn patch_configmap(&self, crd: &HydraDoomNode) -> anyhow::Result<ConfigMap> {
         let api: Api<ConfigMap> = Api::namespaced(self.client.clone(), &crd.namespace().unwrap());
 
@@ -200,30 +237,27 @@ impl K8sContext {
         }
     }
 
-    async fn patch_deployment(&self, crd: &HydraDoomNode) -> anyhow::Result<Deployment> {
-        let deployments: Api<Deployment> =
-            Api::namespaced(self.client.clone(), &crd.namespace().unwrap());
+    async fn patch_sts(&self, crd: &HydraDoomNode) -> anyhow::Result<StatefulSet> {
+        let api: Api<StatefulSet> = Api::namespaced(self.client.clone(), &crd.namespace().unwrap());
 
-        // Create or patch the deployment
-        deployments
-            .patch(
-                &crd.internal_name(),
-                &PatchParams::apply("hydra-doom-pod-controller"),
-                &Patch::Apply(&crd.deployment(&self.config, &self.constants)),
-            )
-            .await
-            .map_err(|err| {
-                error!(err = err.to_string(), "Failed to create deployment.");
-                err.into()
-            })
+        // Create or patch the sts
+        api.patch(
+            &crd.internal_name(),
+            &PatchParams::apply("hydra-doom-pod-controller"),
+            &Patch::Apply(&crd.sts(&self.config, &self.constants)),
+        )
+        .await
+        .map_err(|err| {
+            error!(err = err.to_string(), "Failed to create sts.");
+            err.into()
+        })
     }
 
-    async fn remove_deployment(&self, crd: &HydraDoomNode) -> anyhow::Result<()> {
-        let deployments: Api<Deployment> =
-            Api::namespaced(self.client.clone(), &crd.namespace().unwrap());
+    async fn remove_sts(&self, crd: &HydraDoomNode) -> anyhow::Result<()> {
+        let api: Api<StatefulSet> = Api::namespaced(self.client.clone(), &crd.namespace().unwrap());
         let dp = DeleteParams::default();
 
-        match deployments.delete(&crd.internal_name(), &dp).await {
+        match api.delete(&crd.internal_name(), &dp).await {
             Ok(_) => Ok(()),
             Err(e) => Err(e.into()),
         }
@@ -287,6 +321,19 @@ impl K8sContext {
             self.constants.metrics_port,
             self.constants.metrics_endpoint
         );
+
+        if crd.spec.asleep.unwrap_or(false) {
+            return HydraDoomNodeStatus {
+                state: HydraDoomNodeState::Sleeping.into(),
+                transactions: 0,
+                local_url: format!("ws://{}:{}", crd.internal_host(), self.constants.port),
+                external_url: format!(
+                    "ws://{}:{}",
+                    crd.external_host(&self.config, &self.constants),
+                    self.config.external_port
+                ),
+            };
+        }
         let default = HydraDoomNodeStatus::offline(crd, &self.config, &self.constants);
 
         match reqwest::get(&url).await {

src/custom_resource.rs

@@ -1,14 +1,18 @@
-use k8s_openapi::api::{
-    apps::v1::{Deployment, DeploymentSpec},
-    core::v1::{
-        ConfigMap, ConfigMapVolumeSource, Container, ContainerPort, EmptyDirVolumeSource, PodSpec,
-        PodTemplateSpec, SecretVolumeSource, Service, ServicePort, ServiceSpec, Volume,
-        VolumeMount,
-    },
-    networking::v1::{
-        HTTPIngressPath, HTTPIngressRuleValue, Ingress, IngressBackend, IngressRule,
-        IngressServiceBackend, IngressSpec, ServiceBackendPort,
+use k8s_openapi::{
+    api::{
+        apps::v1::{StatefulSet, StatefulSetSpec},
+        core::v1::{
+            ConfigMap, ConfigMapVolumeSource, Container, ContainerPort, EmptyDirVolumeSource,
+            PersistentVolumeClaim, PersistentVolumeClaimSpec, PersistentVolumeClaimVolumeSource,
+            PodSpec, PodTemplateSpec, SecretVolumeSource, Service, ServicePort, ServiceSpec,
+            Volume, VolumeMount, VolumeResourceRequirements,
+        },
+        networking::v1::{
+            HTTPIngressPath, HTTPIngressRuleValue, Ingress, IngressBackend, IngressRule,
+            IngressServiceBackend, IngressSpec, ServiceBackendPort,
+        },
     },
+    apimachinery::pkg::api::resource::Quantity,
 };
 use kube::{api::ObjectMeta, CustomResource, ResourceExt};
 use schemars::JsonSchema;
@@ -45,6 +49,7 @@ pub struct HydraDoomNodeSpec {
     pub seed_input: String,
     pub commit_inputs: Vec<String>,
     pub start_chain_from: Option<String>,
+    pub asleep: Option<bool>,
 }
 
 #[derive(Deserialize, Serialize, Clone, Default, Debug, JsonSchema)]
@@ -121,11 +126,35 @@ impl HydraDoomNode {
         }
     }
 
-    pub fn deployment(&self, config: &Config, constants: &K8sConstants) -> Deployment {
+    pub fn pvc(&self, _config: &Config, constants: &K8sConstants) -> PersistentVolumeClaim {
+        let name = self.internal_name();
+
+        PersistentVolumeClaim {
+            metadata: ObjectMeta {
+                name: Some(name.clone()),
+                ..Default::default()
+            },
+            spec: Some(PersistentVolumeClaimSpec {
+                access_modes: Some(vec!["ReadWriteMany".to_string()]),
+                storage_class_name: Some(constants.storage_class_name.clone()),
+                resources: Some(VolumeResourceRequirements {
+                    requests: Some(BTreeMap::from([(
+                        "storage".to_string(),
+                        Quantity(constants.persistence_dir_storage_request.clone()),
+                    )])),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            }),
+            ..Default::default()
+        }
+    }
+
+    pub fn sts(&self, config: &Config, constants: &K8sConstants) -> StatefulSet {
         let name = self.internal_name();
         let labels = self.internal_labels();
 
-        // Common deployment parts:
+        // Common sts parts:
         let main_container_common_args = vec![
             "--host".to_string(),
             "0.0.0.0".to_string(),
@@ -204,6 +233,11 @@ impl HydraDoomNode {
                         mount_path: constants.secret_dir.clone(),
                         ..Default::default()
                     },
+                    VolumeMount {
+                        name: "persistence".to_string(),
+                        mount_path: constants.persistence_dir.clone(),
+                        ..Default::default()
+                    },
                     VolumeMount {
                         name: "ipc".to_string(),
                         mount_path: constants.socket_dir.clone(),
@@ -303,13 +337,18 @@ impl HydraDoomNode {
             })
         }
 
-        Deployment {
+        StatefulSet {
             metadata: ObjectMeta {
                 name: Some(name.clone()),
                 ..Default::default()
             },
-            spec: Some(DeploymentSpec {
-                replicas: Some(1),
+            spec: Some(StatefulSetSpec {
+                service_name: "hydra-doom-node".to_string(),
+                replicas: Some(if self.spec.asleep.unwrap_or(false) {
+                    0
+                } else {
+                    1
+                }),
                 selector: k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector {
                     match_labels: Some(labels.clone()),
                     ..Default::default()
@@ -366,6 +405,14 @@ impl HydraDoomNode {
                                 }),
                                 ..Default::default()
                             },
+                            Volume {
+                                name: "persistence".to_string(),
+                                persistent_volume_claim: Some(PersistentVolumeClaimVolumeSource {
+                                    claim_name: self.name_any(),
+                                    ..Default::default()
+                                }),
+                                ..Default::default()
+                            },
                             Volume {
                                 name: "ipc".to_string(),
                                 empty_dir: Some(EmptyDirVolumeSource::default()),