Merge pull request #24 from CoreDB-io/ahendel/updateEventTypes

add types for events

reconciler/src/lib.rs

@@ -1,6 +1,8 @@
 mod ingress_route_tcp_crd;
 mod pg_cluster_crd;
+pub mod types;
 
+use base64::{engine::general_purpose, Engine as _};
 use ingress_route_tcp_crd::IngressRouteTCP;
 use k8s_openapi::api::core::v1::{Namespace, Secret};
 use kube::api::{DeleteParams, ListParams, Patch, PatchParams};
@@ -12,8 +14,6 @@ use serde_json::{from_str, to_string, Value};
 use std::fmt::Debug;
 use thiserror::Error;
 
-use base64::{engine::general_purpose, Engine as _};
-
 #[derive(Error, Debug)]
 pub enum Error {
     #[error("Kube Error: {0}")]
@@ -22,17 +22,12 @@ pub enum Error {
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
 
-pub async fn generate_spec(body: Value) -> Value {
-    let name: String = serde_json::from_value(body["resource_name"].clone()).unwrap();
-    let cpu: String = serde_json::from_value(body["cpu"].clone()).unwrap();
-    let memory: String = serde_json::from_value(body["memory"].clone()).unwrap();
-    let storage: String = serde_json::from_value(body["storage"].clone()).unwrap();
-
+pub async fn generate_spec(event_body: &types::EventBody) -> Value {
     let spec = serde_json::json!({
         "apiVersion": "postgres-operator.crunchydata.com/v1beta1",
         "kind": "PostgresCluster",
         "metadata": {
-            "name": format!("{}", name),
+            "name": format!("{}", event_body.resource_name),
         },
         "spec": {
             "image": "registry.developers.crunchydata.com/crunchydata/crunchy-postgres:ubi8-14.6-2".to_owned(),
@@ -42,16 +37,16 @@ pub async fn generate_spec(body: Value) -> Value {
                     "name": "instance1",
                     "dataVolumeClaimSpec": {
                         "accessModes": ["ReadWriteOnce"],
-                        "resources": {"requests": {"storage": format!("{}", storage)}},
+                        "resources": {"requests": {"storage": format!("{}", event_body.storage)}},
                     },
                     "resources": {
                         "limits": {
-                            "cpu": format!("{}", cpu),
-                            "memory": format!("{}", memory),
+                            "cpu": format!("{}", event_body.cpu),
+                            "memory": format!("{}", event_body.memory),
                         },
                         "requests": {
-                            "cpu": format!("{}", cpu),
-                            "memory": format!("{}", memory),
+                            "cpu": format!("{}", event_body.cpu),
+                            "memory": format!("{}", event_body.memory),
                         },
                     },
                 },

reconciler/src/main.rs

@@ -8,6 +8,8 @@ use reconciler::{
 use std::env;
 use std::{thread, time};
 
+use reconciler::types::EventBody;
+
 #[tokio::main]
 async fn run() -> Result<(), Box<dyn std::error::Error>> {
     // // Read connection info from environment variable
@@ -45,10 +47,12 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
                 info!("Doing nothing for now")
             }
             Some("Create") | Some("Update") => {
+                let crud_event_body =
+                    serde_json::from_value::<EventBody>(read_msg.message["body"].clone())
+                        .expect("error parsing body");
                 // create namespace if it does not exist
-                let namespace: String =
-                    serde_json::from_value(read_msg.message["body"]["resource_name"].clone())
-                        .unwrap();
+                let namespace: String = crud_event_body.resource_name.clone();
+
                 create_namespace(client.clone(), namespace.clone())
                     .await
                     .expect("error creating namespace");
@@ -59,7 +63,7 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
                     .expect("error creating IngressRouteTCP");
 
                 // generate PostgresCluster spec based on values in body
-                let spec = generate_spec(read_msg.message["body"].clone()).await;
+                let spec = generate_spec(&crud_event_body).await;
 
                 // create or update PostgresCluster
                 create_or_update(client.clone(), namespace.clone(), spec)

reconciler/src/types.rs

@@ -0,0 +1,19 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct CRUDevent {
+    pub data_plane_id: String,
+    pub event_id: String,
+    pub message_type: String,
+    pub body: EventBody,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct EventBody {
+    pub resource_type: String,
+    pub resource_name: String,
+    pub extensions: Option<Vec<String>>,
+    pub storage: String,
+    pub memory: String,
+    pub cpu: String,
+}