worker creates k8s objects

api/Cargo.toml

@@ -17,7 +17,12 @@ bytes = { workspace = true }
 config = { workspace = true, features = ["yaml"] }
 config-types = { path = "../config-types" }
 k8s-openapi = { workspace = true, features = ["latest"] }
-kube = { workspace = true, features = ["runtime", "derive"] }
+kube = { workspace = true, features = [
+    "runtime",
+    "derive",
+    "client",
+    "rustls-tls",
+] }
 secrecy = { workspace = true, features = ["serde", "alloc"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, features = ["std"] }

api/src/k8s_client.rs

@@ -0,0 +1,150 @@
+use k8s_openapi::api::core::v1::{ConfigMap, Pod, Secret};
+use serde_json::json;
+use thiserror::Error;
+use tracing::*;
+
+use kube::{
+    api::{Api, PostParams, ResourceExt},
+    Client,
+};
+
+#[derive(Debug, Error)]
+pub enum K8sError {
+    #[error["serde_json error: {0}"]]
+    Serde(#[from] serde_json::error::Error),
+
+    #[error["kube error: {0}"]]
+    Kube(#[from] kube::Error),
+}
+
+pub async fn create_bq_service_account_key_secret(
+    bq_service_account_key: &str,
+) -> Result<(), K8sError> {
+    info!("creating BQ service account key secret");
+
+    let secret: Secret = serde_json::from_value(json!({
+      "apiVersion": "v1",
+      "kind": "Secret",
+      "metadata": {
+        "name": "bq-service-account-key"
+      },
+      "type": "Opaque",
+      "stringData": {
+        "service-account-key": bq_service_account_key,
+      }
+    }))?;
+
+    let client = Client::try_default().await?;
+    let secrets: Api<Secret> = Api::default_namespaced(client);
+
+    let pp = PostParams::default();
+    match secrets.create(&pp, &secret).await {
+        Ok(o) => {
+            let name = o.name_any();
+            assert_eq!(secret.name_any(), name);
+            info!("created Secret {}", name);
+        }
+        Err(kube::Error::Api(ae)) => {
+            error!("Error: {ae}");
+            assert_eq!(ae.code, 409);
+        } // if you skipped delete, for instance
+        Err(e) => return Err(e.into()), // any other case is probably bad
+    }
+
+    Ok(())
+}
+
+pub async fn create_config_map(base_config: &str, prod_config: &str) -> Result<(), K8sError> {
+    info!("creating config map");
+
+    let cm: ConfigMap = serde_json::from_value(json!({
+      "kind": "ConfigMap",
+      "apiVersion": "v1",
+      "metadata": {
+        "name": "replicator-config"
+      },
+      "data": {
+        "base.yaml": base_config,
+        "prod.yaml": prod_config,
+      }
+    }))?;
+
+    let client = Client::try_default().await?;
+    let config_maps: Api<ConfigMap> = Api::default_namespaced(client);
+
+    let pp = PostParams::default();
+    match config_maps.create(&pp, &cm).await {
+        Ok(o) => {
+            let name = o.name_any();
+            assert_eq!(cm.name_any(), name);
+            info!("created ConfigMap {}", name);
+        }
+        Err(kube::Error::Api(ae)) => {
+            error!("Error: {ae}");
+            assert_eq!(ae.code, 409);
+        } // if you skipped delete, for instance
+        Err(e) => return Err(e.into()), // any other case is probably bad
+    }
+    Ok(())
+}
+
+pub async fn create_pod() -> Result<(), K8sError> {
+    info!("creating Pod instance replicator");
+
+    let client = Client::try_default().await?;
+    let pods: Api<Pod> = Api::default_namespaced(client);
+
+    let p: Pod = serde_json::from_value(json!({
+        "apiVersion": "v1",
+        "kind": "Pod",
+        "metadata": { "name": "replicator" },
+        "spec": {
+            "volumes": [
+              {
+                "name": "config-file",
+                "configMap": {
+                  "name": "replicator-config"
+                }
+              }
+            ],
+            "containers": [{
+              "name": "replicator",
+              "image": "ramsup/replicator:0.0.7",
+              "env": [
+                {
+                  "name": "APP_ENVIRONMENT",
+                  "value": "prod"
+                },
+                {
+                  "name": "APP_SINK__BIGQUERY__SERVICE_ACCOUNT_KEY",
+                  "valueFrom": {
+                    "secretKeyRef": {
+                      "name": "bq-service-account-key",
+                      "key": "service-account-key"
+                    }
+                  }
+                }
+              ],
+              "volumeMounts": [{
+                "name": "config-file",
+                "mountPath": "/app/configuration"
+              }]
+            }],
+        }
+    }))?;
+
+    let pp = PostParams::default();
+    match pods.create(&pp, &p).await {
+        Ok(o) => {
+            let name = o.name_any();
+            assert_eq!(p.name_any(), name);
+            info!("created Pod {}", name);
+        }
+        Err(kube::Error::Api(ae)) => {
+            error!("Error: {ae}");
+            assert_eq!(ae.code, 409);
+        } // if you skipped delete, for instance
+        Err(e) => return Err(e.into()), // any other case is probably bad
+    }
+    Ok(())
+}

api/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod configuration;
+pub mod k8s_client;
 pub mod queue;
 pub mod startup;
 pub mod telemetry;

api/src/queue.rs

@@ -1,4 +1,3 @@
-use config_types::Settings;
 use sqlx::{PgPool, Postgres, Transaction};
 
 pub async fn enqueue_task(
@@ -29,12 +28,6 @@ pub struct Task {
     pub data: serde_json::Value,
 }
 
-#[allow(clippy::large_enum_variant)]
-pub enum Request {
-    CreateOrUpdate(Settings),
-    Delete(String),
-}
-
 pub async fn dequeue_task(pool: &PgPool) -> Result<Option<(PgTransaction, Task)>, anyhow::Error> {
     let mut txn = pool.begin().await?;
 

api/src/worker.rs

@@ -1,11 +1,13 @@
 use std::time::Duration;
 
+use config_types::SinkSettings;
 use sqlx::PgPool;
-use tracing::debug;
+use tracing::{debug, info};
 use tracing_log::log::error;
 
 use crate::{
     configuration::Settings,
+    k8s_client::{create_bq_service_account_key_secret, create_config_map, create_pod},
     queue::{delete_task, dequeue_task},
     startup::get_connection_pool,
 };
@@ -26,21 +28,61 @@ async fn worker_loop(pool: PgPool, poll_duration: Duration) -> Result<(), anyhow
                 debug!("successfully executed task");
             }
             Err(e) => {
-                error!("error while executing task: {e}");
+                error!("error while executing task: {e:#?}");
             }
         }
         tokio::time::sleep(poll_duration).await;
     }
 }
 
+#[allow(clippy::large_enum_variant)]
+#[derive(serde::Serialize, serde::Deserialize)]
+pub enum Request {
+    CreateOrUpdate {
+        project_ref: String,
+        settings: config_types::Settings,
+    },
+    Delete {
+        project_ref: String,
+    },
+}
+
 pub async fn try_execute_task(pool: &PgPool) -> Result<ExecutionOutcome, anyhow::Error> {
     let task = dequeue_task(pool).await?;
-    if task.is_none() {
+    let Some((transaction, task)) = task else {
         return Ok(ExecutionOutcome::EmptyQueue);
+    };
+
+    let request = serde_json::from_value::<Request>(task.data)?;
+
+    match request {
+        Request::CreateOrUpdate {
+            project_ref,
+            settings,
+        } => {
+            info!(
+                "creating or updating k8s objects for project ref: {}",
+                project_ref
+            );
+
+            let SinkSettings::BigQuery {
+                project_id: _,
+                dataset_id: _,
+                service_account_key,
+            } = &settings.sink;
+            create_bq_service_account_key_secret(service_account_key).await?;
+            let base_config = "";
+            let prod_config = serde_json::to_string(&settings)?;
+            create_config_map(base_config, &prod_config).await?;
+            create_pod().await?;
+        }
+        Request::Delete { project_ref } => {
+            info!("deleting project ref: {}", project_ref);
+        }
     }
-    let (transaction, task) = task.unwrap();
-    //TODO: perform task here
+
     delete_task(transaction, task.id).await?;
+
     Ok(ExecutionOutcome::TaskCompleted)
 }