feat: change db username and metrics collector (#14)

* feat: improved drop user and change username to hash

* chore: added signal to exit

* feat: updated pg connection and metrics collector

* chore: changed metrics drop user fn name

* chore: adjusted clippy

Author: paulobressan

operator/Cargo.lock

@@ -22,6 +22,10 @@ tracing-subscriber = "0.3.18"
 rand = "0.8.5"
 prometheus = "0.13.3"
 actix-web = "4.4.0"
+bech32 = "0.9.1"
+sha3 = "0.10.8"
+lazy_static = "1.4.0"
+deadpool-postgres = "0.12.1"
 
 [[bin]]
 name = "controller"

operator/Cargo.toml

@@ -0,0 +1,61 @@
+use lazy_static::lazy_static;
+use std::{env, time::Duration};
+
+lazy_static! {
+    static ref CONTROLLER_CONFIG: Config = Config::from_env();
+}
+
+pub fn get_config() -> &'static Config {
+    &CONTROLLER_CONFIG
+}
+
+#[derive(Debug, Clone)]
+pub struct Config {
+    pub db_url_mainnet: String,
+    pub db_url_preprod: String,
+    pub db_url_preview: String,
+
+    pub dcu_per_second_mainnet: f64,
+    pub dcu_per_second_preprod: f64,
+    pub dcu_per_second_preview: f64,
+
+    pub metrics_delay: Duration,
+}
+
+impl Config {
+    pub fn from_env() -> Self {
+        let db_url_mainnet = env::var("DB_URL_MAINNET").expect("DB_URL_MAINNET must be set");
+        let db_url_preprod = env::var("DB_URL_PREPROD").expect("DB_URL_PREPROD must be set");
+        let db_url_preview = env::var("DB_URL_PREVIEW").expect("DB_URL_PREVIEW must be set");
+
+        let metrics_delay = Duration::from_secs(
+            env::var("METRICS_DELAY")
+                .expect("METRICS_DELAY must be set")
+                .parse::<u64>()
+                .expect("METRICS_DELAY must be a number"),
+        );
+
+        let dcu_per_second_mainnet = env::var("DCU_PER_SECOND_MAINNET")
+            .expect("DCU_PER_SECOND_MAINNET must be set")
+            .parse::<f64>()
+            .expect("DCU_PER_SECOND_MAINNET must be a number");
+        let dcu_per_second_preprod = env::var("DCU_PER_SECOND_PREPROD")
+            .expect("DCU_PER_SECOND_PREPROD must be set")
+            .parse::<f64>()
+            .expect("DCU_PER_SECOND_PREPROD must be a number");
+        let dcu_per_second_preview = env::var("DCU_PER_SECOND_PREVIEW")
+            .expect("DCU_PER_SECOND_PREVIEW must be set")
+            .parse::<f64>()
+            .expect("DCU_PER_SECOND_PREVIEW must be a number");
+
+        Self {
+            db_url_mainnet,
+            db_url_preprod,
+            db_url_preview,
+            metrics_delay,
+            dcu_per_second_mainnet,
+            dcu_per_second_preprod,
+            dcu_per_second_preview,
+        }
+    }
+}

operator/src/config.rs

@@ -1,3 +1,4 @@
+use bech32::ToBase32;
 use futures::StreamExt;
 use kube::{
     api::{Patch, PatchParams},
@@ -13,28 +14,14 @@ use rand::distributions::{Alphanumeric, DistString};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
+use sha3::{Digest, Sha3_256};
 use std::{sync::Arc, time::Duration};
 use tracing::error;
 
-use crate::{postgres::Postgres, Config, Error, Metrics, Network, State};
+use crate::{postgres::Postgres, Error, Network, State};
 
 pub static DB_SYNC_PORT_FINALIZER: &str = "dbsyncports.demeter.run";
 
-struct Context {
-    pub client: Client,
-    pub metrics: Metrics,
-    pub config: Config,
-}
-impl Context {
-    pub fn new(client: Client, metrics: Metrics, config: Config) -> Self {
-        Self {
-            client,
-            metrics,
-            config,
-        }
-    }
-}
-
 #[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
 #[kube(
     kind = "DbSyncPort",
@@ -64,22 +51,19 @@ impl DbSyncPort {
             .unwrap_or(false)
     }
 
-    async fn reconcile(&self, ctx: Arc<Context>, pg: &mut Postgres) -> Result<Action, Error> {
-        let client = ctx.client.clone();
+    async fn reconcile(&self, state: Arc<State>, pg: &Postgres) -> Result<Action, Error> {
+        let client = state.kube_client.clone();
         let ns = self.namespace().unwrap();
         let name = self.name_any();
         let crds: Api<DbSyncPort> = Api::namespaced(client, &ns);
 
-        let username = format!("{name}.{ns}");
+        let username = gen_username_hash(&format!("{name}.{ns}")).await?;
         let password = Alphanumeric.sample_string(&mut rand::thread_rng(), 16);
 
         if !self.was_executed() {
-            match pg.user_already_exists(&username).await? {
-                true => pg.user_enable(&username, &password).await?,
-                false => pg.user_create(&username, &password).await?,
-            };
+            pg.create_user(&username, &password).await?;
 
-            let new_status = Patch::Apply(json!({
+            let status = Patch::Apply(json!({
                 "apiVersion": "demeter.run/v1alpha1",
                 "kind": "DbSyncPort",
                 "status": DbSyncPortStatus {
@@ -89,61 +73,77 @@ impl DbSyncPort {
             }));
 
             let ps = PatchParams::apply("cntrlr").force();
-            crds.patch_status(&name, &ps, &new_status)
+            crds.patch_status(&name, &ps, &status)
                 .await
                 .map_err(Error::KubeError)?;
 
-            ctx.metrics.count_user_created(&username);
+            state.metrics.count_user_created(&ns, &self.spec.network);
         };
 
-        Ok(Action::requeue(Duration::from_secs(5 * 60)))
+        Ok(Action::await_change())
     }
 
-    async fn cleanup(&self, ctx: Arc<Context>, pg: &mut Postgres) -> Result<Action, Error> {
-        let username = self.status.as_ref().unwrap().username.clone();
-        pg.user_disable(&username).await?;
-        ctx.metrics.count_user_deactivated(&username);
+    async fn cleanup(&self, state: Arc<State>, pg: &Postgres) -> Result<Action, Error> {
+        if self.was_executed() {
+            let ns = self.namespace().unwrap();
+            let username = self.status.as_ref().unwrap().username.clone();
+            pg.drop_user(&username).await?;
+            state
+                .metrics
+                .count_user_droped(&ns, &self.spec.network);
+        }
+
         Ok(Action::await_change())
     }
 }
 
-async fn reconcile(crd: Arc<DbSyncPort>, ctx: Arc<Context>) -> Result<Action, Error> {
-    let url = match crd.spec.network {
-        Network::Mainnet => &ctx.config.db_url_mainnet,
-        Network::Preprod => &ctx.config.db_url_preprod,
-        Network::Preview => &ctx.config.db_url_preview,
-    };
-
+async fn reconcile(crd: Arc<DbSyncPort>, state: Arc<State>) -> Result<Action, Error> {
     let ns = crd.namespace().unwrap();
-    let crds: Api<DbSyncPort> = Api::namespaced(ctx.client.clone(), &ns);
+    let crds: Api<DbSyncPort> = Api::namespaced(state.kube_client.clone(), &ns);
 
-    let mut postgres = Postgres::new(url).await?;
+    let postgres = state.get_pg_by_network(&crd.spec.network);
 
     finalizer(&crds, DB_SYNC_PORT_FINALIZER, crd, |event| async {
         match event {
-            Event::Apply(crd) => crd.reconcile(ctx.clone(), &mut postgres).await,
-            Event::Cleanup(crd) => crd.cleanup(ctx.clone(), &mut postgres).await,
+            Event::Apply(crd) => crd.reconcile(state.clone(), &postgres).await,
+            Event::Cleanup(crd) => crd.cleanup(state.clone(), &postgres).await,
         }
     })
     .await
     .map_err(|e| Error::FinalizerError(Box::new(e)))
 }
 
-fn error_policy(crd: Arc<DbSyncPort>, err: &Error, ctx: Arc<Context>) -> Action {
+fn error_policy(crd: Arc<DbSyncPort>, err: &Error, state: Arc<State>) -> Action {
     error!("reconcile failed: {:?}", err);
-    ctx.metrics.reconcile_failure(&crd, err);
+    state.metrics.reconcile_failure(&crd, err);
     Action::requeue(Duration::from_secs(5))
 }
 
-pub async fn run(state: Arc<State>, config: Config) -> Result<(), Error> {
+async fn gen_username_hash(username: &str) -> Result<String, Error> {
+    let mut hasher = Sha3_256::new();
+    hasher.update(username);
+    let sha256_hash = hasher.finalize();
+
+    let bech32_hash = bech32::encode(
+        "dmtr_dbsync",
+        sha256_hash.to_base32(),
+        bech32::Variant::Bech32,
+    )?;
+
+    let bech32_truncated: String = bech32_hash.chars().take(32).collect();
+
+    Ok(bech32_truncated)
+}
+
+pub async fn run(state: Arc<State>) -> Result<(), Error> {
     let client = Client::try_default().await?;
     let crds = Api::<DbSyncPort>::all(client.clone());
 
-    let ctx = Context::new(client, state.metrics.clone(), config);
+    // let ctx = Context::new(client, state.clone());
 
     Controller::new(crds, WatcherConfig::default().any_semantic())
         .shutdown_on_signal()
-        .run(reconcile, error_policy, Arc::new(ctx))
+        .run(reconcile, error_policy, state)
         .for_each(|_| futures::future::ready(()))
         .await;