pvc storage

Author: jw3

crd.yml

@@ -33,3 +33,5 @@ spec:
                   type: string
                 workers:
                   type: integer
+                worker_storage:
+                  type: integer

src/bin/cli.rs

@@ -4,6 +4,7 @@ use kube::{Api, Client};
 use kube::api::PostParams;
 
 use example_citus_operator::crd::{CitusCluster, CitusClusterSpec};
+use example_citus_operator::storage;
 
 #[derive(Clone, Debug, Parser)]
 struct Opts {
@@ -28,12 +29,20 @@ struct CreateOpts {
     /// Number of workers
     #[clap(short, long, default_value = "1")]
     workers: usize,
+
+    /// Storage volume size for each worker, in GB
+    #[clap(long, default_value = "1")]
+    worker_storage: usize,
 }
 
 #[derive(Clone, Debug, Parser)]
 struct DeleteOpts {
     /// Name of the cluster
     name: String,
+
+    /// Delete associated persistent storage
+    #[clap(long)]
+    purge: bool,
 }
 
 #[tokio::main]
@@ -55,6 +64,7 @@ async fn main() {
                         },
                         spec: CitusClusterSpec {
                             workers: c.workers as i32,
+                            worker_storage: c.worker_storage,
                         },
                     },
                 )
@@ -66,6 +76,12 @@ async fn main() {
                 .delete(&c.name, &Default::default())
                 .await
                 .expect("delete");
+
+            if c.purge {
+                storage::delete_storage(client.clone(), &c.name, &opts.namespace)
+                    .await
+                    .expect("purge");
+            }
         }
     }
 }

src/bin/main.rs

@@ -68,7 +68,14 @@ async fn reconcile(cc: Arc<CitusCluster>, context: Arc<ContextData>) -> Result<A
     match determine_action(&cc) {
         ClusterAction::Create => {
             cluster::add_finalizer(client.clone(), &name, &namespace).await?;
-            cluster::deploy(client, &name, cc.spec.workers, &namespace).await?;
+            cluster::deploy(
+                client,
+                &name,
+                cc.spec.workers,
+                cc.spec.worker_storage,
+                &namespace,
+            )
+            .await?;
             Ok(Action::requeue(Duration::from_secs(10)))
         }
         ClusterAction::Delete => {

src/cluster.rs

@@ -12,10 +12,12 @@ pub async fn deploy(
     client: Client,
     name: &str,
     num_workers: i32,
+    worker_storage: usize,
     namespace: &str,
 ) -> Result<CitusDeployment, Error> {
     let master = master::deploy(client.clone(), name, namespace).await?;
-    let workers = workers::deploy(client.clone(), name, num_workers, namespace).await?;
+    let workers =
+        workers::deploy(client.clone(), name, num_workers, worker_storage, namespace).await?;
     jobs::register_workers(client.clone(), name, num_workers, namespace).await?;
 
     master::expose(client.clone(), name, namespace).await?;

src/crd.rs

@@ -4,13 +4,14 @@ use serde::{Deserialize, Serialize};
 
 #[derive(CustomResource, Serialize, Deserialize, Debug, PartialEq, Clone, JsonSchema)]
 #[kube(
-group = "jw3.xyz",
-version = "v1alpha1",
-kind = "CitusCluster",
-plural = "citusclusters",
-derive = "PartialEq",
-namespaced
+    group = "jw3.xyz",
+    version = "v1alpha1",
+    kind = "CitusCluster",
+    plural = "citusclusters",
+    derive = "PartialEq",
+    namespaced
 )]
 pub struct CitusClusterSpec {
     pub workers: i32,
+    pub worker_storage: usize,
 }

src/jobs.rs

@@ -22,7 +22,7 @@ pub async fn register_workers(
                 spec: Some(PodSpec {
                     restart_policy: Some("OnFailure".to_owned()),
                     containers: vec![Container {
-                        name: format!("{name}-init-worker-"),
+                        name: format!("{name}-init-worker"),
                         image: Some("citusdata/citus:12.1".to_owned()),
                         command: Some(vec![
                             "bash".to_owned(),

src/lib.rs

@@ -2,4 +2,5 @@ pub mod cluster;
 pub mod crd;
 pub mod jobs;
 pub mod master;
+pub mod storage;
 pub mod workers;

src/storage.rs

@@ -0,0 +1,35 @@
+use std::collections::BTreeMap;
+
+use futures::TryFutureExt;
+use k8s_openapi::api::core::v1::{
+    PersistentVolumeClaim, PersistentVolumeClaimSpec, ResourceRequirements,
+};
+use k8s_openapi::apimachinery::pkg::api::resource::Quantity;
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+use kube::{Api, Client, Error};
+
+pub async fn delete_storage(client: Client, name: &str, namespace: &str) -> Result<(), Error> {
+    let api: Api<PersistentVolumeClaim> = Api::namespaced(client.clone(), namespace);
+    api.delete(name, &Default::default()).map_ok(|_| ()).await
+}
+
+pub fn volume_claim_template(name: &str, gi: usize) -> PersistentVolumeClaim {
+    let mut worker_labels: BTreeMap<String, Quantity> = BTreeMap::new();
+    worker_labels.insert("storage".to_owned(), Quantity(format!("{gi}Gi")));
+
+    PersistentVolumeClaim {
+        metadata: ObjectMeta {
+            name: Some(name.to_owned()),
+            ..Default::default()
+        },
+        spec: Some(PersistentVolumeClaimSpec {
+            access_modes: Some(vec!["ReadWriteOnce".to_owned()]),
+            resources: Some(ResourceRequirements {
+                requests: Some(worker_labels),
+                ..Default::default()
+            }),
+            ..Default::default()
+        }),
+        ..Default::default()
+    }
+}

src/workers.rs

@@ -3,16 +3,20 @@ use std::collections::BTreeMap;
 use k8s_openapi::api::apps::v1::{StatefulSet, StatefulSetSpec};
 use k8s_openapi::api::core::v1::{
     Container, ContainerPort, EnvVar, PodSpec, PodTemplateSpec, Service, ServicePort, ServiceSpec,
+    VolumeMount,
 };
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{LabelSelector, ObjectMeta};
 use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;
 use kube::{Api, Client, Error};
 use kube::api::{DeleteParams, PostParams};
 
+use crate::storage;
+
 pub async fn deploy(
     client: Client,
     name: &str,
     cnt: i32,
+    storage: usize,
     namespace: &str,
 ) -> Result<StatefulSet, Error> {
     let mut worker_labels: BTreeMap<String, String> = BTreeMap::new();
@@ -51,6 +55,11 @@ pub async fn deploy(
                             value: Some("yourpassword".to_owned()),
                             ..EnvVar::default()
                         }]),
+                        volume_mounts: Some(vec![VolumeMount {
+                            mount_path: "/var/lib/postgresql/data".to_owned(),
+                            name: name.to_owned(),
+                            ..Default::default()
+                        }]),
                         ..Container::default()
                     }],
                     ..PodSpec::default()
@@ -60,6 +69,7 @@ pub async fn deploy(
                     ..ObjectMeta::default()
                 }),
             },
+            volume_claim_templates: Some(vec![storage::volume_claim_template(name, storage)]),
             ..StatefulSetSpec::default()
         }),
         ..StatefulSet::default()