Added deployment generation

infinyon · simlay · Aug 24, 2021 · Aug 24, 2021 · Aug 25, 2021 · Aug 25, 2021
commit 04f7748f6ef21ade131b5cd37807569a73b0c996
diff --git a/src/sc/src/k8/controllers/manged_connector_deployment.rs b/src/sc/src/k8/controllers/manged_connector_deployment.rs
@@ -26,6 +26,11 @@ use crate::k8::objects::spg_service::SpgServiceSpec;
 //use crate::k8::objects::deployment::DeploymentStatus;
 use crate::k8::objects::managed_connector_deployment::ManagedConnectorDeploymentSpec;
 
+use crate::stores::k8::K8MetaItem;
+use fluvio_controlplane_metadata::store::MetadataStoreObject;
+use crate::k8::objects::managed_connector_deployment::K8DeploymentSpec;
+
+
 
 /// Update Statefulset and Service from SPG
 pub struct ManagedConnectorDeploymentController {
@@ -66,62 +71,32 @@ impl ManagedConnectorDeploymentController {
         }
     }
 
-    #[instrument(skip(self), name = "SpgStatefulSetController")]
+    #[instrument(skip(self), name = "ManagedConnectorDeploymentController")]
     async fn inner_loop(&mut self) -> Result<(), ClientError> {
         todo!();
         use tokio::select;
 
         let mut connector_listener = self.connectors.change_listener();
-        let mut config_listener = self.configs.change_listener();
 
-        self.sync_connectors_to_deployment(&mut connector_listener).await?;
-        self.sync_with_config(&mut config_listener).await?;
+        self.sync_connectors_to_deployments(&mut connector_listener).await?;
 
         loop {
             select! {
                 _ = connector_listener.listen() => {
                     debug!("detected connector changes");
-                    self.sync_connectors_to_deployment(&mut connector_listener).await?;
-                },
-
-                _ = config_listener.listen() => {
-                    debug!("detected config changes");
-                    self.sync_with_config(&mut config_listener).await?;
-                }
-
-            }
-        }
-        /*
-
-        let mut spg_listener = self.groups.change_listener();
-        let mut config_listener = self.configs.change_listener();
-
-        self.sync_spgs_to_statefulset(&mut spg_listener).await?;
-        self.sync_with_config(&mut config_listener).await?;
-
-        loop {
-            select! {
-                _ = spg_listener.listen() => {
-                    debug!("detected spg changes");
-                    self.sync_spgs_to_statefulset(&mut spg_listener).await?;
+                    self.sync_connectors_to_deployments(&mut connector_listener).await?;
                 },
-
-                _ = config_listener.listen() => {
-                    debug!("detected config changes");
-                    self.sync_with_config(&mut config_listener).await?;
-                }
-
             }
         }
-        */
     }
 
-    async fn sync_with_config(
+    /// svc has been changed, update spu
+    async fn sync_connectors_to_deployments(
         &mut self,
-        listener: &mut K8ChangeListener<ScK8Config>,
+        listener: &mut K8ChangeListener<ManagedConnectorSpec>,
     ) -> Result<(), ClientError> {
         if !listener.has_change() {
-            trace!("no config change, skipping");
+            trace!("no managed connector change, skipping");
             return Ok(());
         }
 
@@ -130,113 +105,97 @@ impl ManagedConnectorDeploymentController {
         let (updates, deletes) = changes.parts();
 
         debug!(
-            "received config changes updates: {},deletes: {},epoch: {}",
+            "received service changes updates: {},deletes: {},epoch: {}",
             updates.len(),
             deletes.len(),
             epoch,
         );
 
-        for config in updates.into_iter() {
-            for group_item in self.connectors.store().clone_values().await {
-                /*
-                let spu_group = SpuGroupObj::new(group_item);
+        for mc_item in updates.into_iter() {
 
-                self.sync_spg_to_statefulset(spu_group, &config.spec)
-                    .await?
-                */
-            }
+            self.sync_connector_to_deployment(mc_item).await?
         }
-        /*
-        */
 
         Ok(())
     }
 
-    /// svc has been changed, update spu
-    async fn sync_connectors_to_deployment(
+    #[instrument(skip(self, managed_connector))]
+    async fn sync_connector_to_deployment(
         &mut self,
-        listener: &mut K8ChangeListener<ManagedConnectorSpec>,
+        managed_connector: MetadataStoreObject<ManagedConnectorSpec, K8MetaItem>,
     ) -> Result<(), ClientError> {
-        /*
-        if !listener.has_change() {
-            trace!("no spg change, skipping");
-            return Ok(());
-        }
-
-        let changes = listener.sync_changes().await;
-        let epoch = changes.epoch;
-        let (updates, deletes) = changes.parts();
-
-        debug!(
-            "received service changes updates: {},deletes: {},epoch: {}",
-            updates.len(),
-            deletes.len(),
-            epoch,
+        let key = managed_connector.key();
+        let k8_deployment_spec = Self::generate_k8_deployment_spec(
+            &managed_connector.spec(),
+            &self.namespace,
+            key,
         );
+        trace!(?k8_deployment_spec);
+        use crate::stores::MetadataStoreObject;
+        use crate::stores::actions::WSAction;
 
-        if let Some(config_obj) = self.configs.store().value("fluvio").await {
-            let config = config_obj.inner_owned().spec;
-            for group_item in updates.into_iter() {
-                let spu_group = SpuGroupObj::new(group_item);
+        let deployment_action =
+            WSAction::Apply(
+                MetadataStoreObject::with_spec(key, k8_deployment_spec.into())
+                .with_context(managed_connector.ctx().create_child()),
+            );
 
-                self.sync_spg_to_statefulset(spu_group, &config).await?
-            }
-        } else {
-            error!("config map is not loaded, skipping");
-        }
-        */
+        debug!(?deployment_action, "applying statefulset");
+
+        self.deployment
+            .wait_action(&key, deployment_action)
+            .await?;
 
         Ok(())
     }
+    const DEFAULT_CONNECTOR_NAME : &'static str = "fluvio-connector";
+    pub fn generate_k8_deployment_spec(
+        mc_spec: &ManagedConnectorSpec,
+        namespace: &str,
+        name: &str,
+    ) -> K8DeploymentSpec {
+        use k8_types::{
+            TemplateSpec,
+            TemplateMeta,
+            core::pod::{
+                PodSpec,
+                ContainerSpec,
+            },
+            LabelProvider,
+        };
 
-    #[instrument(skip(self, spu_k8_config, spu_group))]
-    async fn sync_spg_to_statefulset(
-        &mut self,
-        spu_group: SpuGroupObj,
-        spu_k8_config: &ScK8Config,
-    ) -> Result<(), ClientError> {
+        let image = format!("infinyon/fluvio-connector-{}", mc_spec.name);
         /*
-        let spg_name = spu_group.key();
-
-        // ensure we don't have conflict with existing spu group
-        if let Some(conflict_id) = spu_group.is_conflict_with(self.spus.store()).await {
-            warn!(conflict_id, "connector is in conflict with existing id");
-            let status = SpuGroupStatus::invalid(format!("conflict with: {}", conflict_id));
-
-            self.connectors
-                .update_status(spg_name.to_owned(), status)
-                .await?;
-        } else {
-            // if we pass this stage, then we reserved id.
-            if !spu_group.is_already_valid() {
-                debug!("not valid");
-                let status = SpuGroupStatus::reserved();
-                self.groups
-                    .update_status(spg_name.to_owned(), status)
-                    .await?;
-                return Ok(());
-            }
-
-            debug!("continue");
-            let (spg_service_key, spg_service_action) = spu_group.as_service();
-
-            /*
-            trace!("spg_service_actions: {:#?}", spg_service_action);
-            self.spg_services
-                .wait_action(&spg_service_key, spg_service_action)
-                .await?;
-            */
-
-            let (stateful_key, stateful_action) =
-                spu_group.as_statefulset(&self.namespace, spu_k8_config, self.tls.as_ref());
-
-            debug!(?stateful_action, "applying statefulset");
-            self.statefulsets
-                .wait_action(&stateful_key, stateful_action)
-                .await?;
-        }
+        let env = todo!();
+        let args = todo!();
         */
 
-        Ok(())
+        let template = TemplateSpec {
+            metadata: Some(
+                TemplateMeta::default()
+                    .set_labels(vec![("app", Self::DEFAULT_CONNECTOR_NAME)]),
+            ),
+            spec: PodSpec {
+                termination_grace_period_seconds: Some(10),
+                containers: vec![ContainerSpec {
+                    name: Self::DEFAULT_CONNECTOR_NAME.to_owned(),
+                    image: Some(image.clone()),
+                    /*
+                    env,
+                    args,
+                    */
+                    ..Default::default()
+                }],
+                //security_context: spu_k8_config.pod_security_context.clone(),
+                //node_selector: Some(spu_pod_config.node_selector.clone()),
+                ..Default::default()
+            },
+        };
+        K8DeploymentSpec {
+            template,
+            ..Default::default()
+        }
     }
+
 }
+
diff --git a/src/sc/src/k8/objects/managed_connector_deployment.rs b/src/sc/src/k8/objects/managed_connector_deployment.rs
@@ -22,6 +22,20 @@ impl Spec for ManagedConnectorDeploymentSpec {
     type Owner = ManagedConnectorSpec;
 }
 
+/*
+ * TODO: Maybe make this more generic.
+#[derive(Deserialize, Serialize, Debug, Default, Clone, PartialEq)]
+#[serde(transparent)]
+pub struct DeploymentSpec<T: Spec>(K8DeploymentSpec, T);
+
+impl<T: Spec> Spec for DeploymentSpec<T> {
+    const LABEL: &'static str = "Deployment";
+    type IndexKey = String;
+    type Status = DeploymentStatus;
+    type Owner = T;
+}
+*/
+
 impl From<K8DeploymentSpec> for ManagedConnectorDeploymentSpec {
     fn from(k8: K8DeploymentSpec) -> Self {
         Self(k8)