feat(mgmt, dataplane): Add previously unimplmented k8s config processor

Adds a k8s config processor using the k8s-intf watcher client.

The watcher needs the hostname as well, so MgmtParams is augmented to
accept hostname as a configuration parameter.

Signed-off-by: Manish Vachharajani <[email protected]>

Author: mvachhar

Cargo.lock

@@ -16,6 +16,7 @@ ctrlc = { workspace = true, features = ["termination"] }
 dpdk = { workspace = true }
 dyn-iter = { workspace = true }
 futures = { workspace = true }
+hostname = { workspace = true }
 hyper = { workspace = true }
 hyper-util = { workspace = true }
 id = { workspace = true }

dataplane/Cargo.toml

@@ -130,8 +130,15 @@ fn main() {
     let pipeline_factory = setup.pipeline;
 
     /* start management */
+    let hostname = hostname::get()
+        .expect("Failed to get hostname")
+        .to_str()
+        .expect("hostname is not valid unicode")
+        .to_string();
+
     start_mgmt(MgmtParams {
         grpc_addr,
+        hostname,
         processor_params: ConfigProcessorParams {
             router_ctl: setup.router.get_ctl_tx(),
             vpcmapw: setup.vpcmapw,

dataplane/src/main.rs

@@ -23,6 +23,7 @@ concurrency = { workspace = true }
 gateway_config = { workspace = true }
 id = { workspace = true }
 interface-manager = { workspace = true }
+k8s-intf = { workspace = true }
 lpm = { workspace = true }
 nat = { workspace = true }
 net = { workspace = true }

mgmt/Cargo.toml

@@ -3,13 +3,54 @@
 
 use tokio::sync::mpsc::Sender;
 
-use crate::processor::proc::ConfigChannelRequest;
+use config::{ExternalConfig, GwConfig};
+use k8s_intf::client::WatchError;
+use k8s_intf::watch_gateway_agent_crd;
+use tracing::error;
+
+use crate::processor::proc::{ConfigChannelRequest, ConfigRequest, ConfigResponse};
 
 #[derive(Debug, thiserror::Error)]
 pub enum K8sClientError {
-    // Define error variants here
+    #[error("K8s client exited early")]
+    EarlyTermination,
+    #[error("K8s client could not get hostname: {0}")]
+    HostnameError(#[from] std::io::Error),
+    #[error("K8s watch failed: {0}")]
+    WatchError(#[from] WatchError),
 }
 
-pub async fn k8s_start_client(_tx: Sender<ConfigChannelRequest>) -> Result<(), K8sClientError> {
-    unimplemented!()
+pub async fn k8s_start_client(
+    hostname: &str,
+    tx: Sender<ConfigChannelRequest>,
+) -> Result<(), K8sClientError> {
+    watch_gateway_agent_crd(hostname, async move |ga| {
+        let external_config = ExternalConfig::try_from(ga);
+        match external_config {
+            Ok(external_config) => {
+                let gw_config = Box::new(GwConfig::new(external_config));
+
+                let (req, rx) = ConfigChannelRequest::new(ConfigRequest::ApplyConfig(gw_config));
+                let tx_result = tx.send(req).await;
+                if let Err(e) = tx_result {
+                    error!("Failure sending request to config processor: {e}");
+                }
+                match rx.await {
+                    Err(e) => error!("Failure receiving from config processor: {e}"),
+                    Ok(response) => match response {
+                        ConfigResponse::ApplyConfig(Err(e)) => {
+                            error!("Failed to apply config: {e}");
+                        }
+                        ConfigResponse::ApplyConfig(Ok(())) => {}
+                        _ => unreachable!(),
+                    },
+                };
+            }
+            Err(e) => {
+                error!("Failed to convert K8sGatewayAgent to ExternalConfig: {e}");
+            }
+        }
+    })
+    .await?;
+    Err(K8sClientError::EarlyTermination)
 }

mgmt/src/processor/k8s_client.rs

@@ -183,6 +183,7 @@ impl Display for ServerAddress {
 
 pub struct MgmtParams {
     pub grpc_addr: Option<GrpcAddress>,
+    pub hostname: String,
     pub processor_params: ConfigProcessorParams,
 }
 
@@ -233,7 +234,7 @@ pub fn start_mgmt(
                 rt.block_on(async {
                     let (processor, tx) = ConfigProcessor::new(params.processor_params);
                     let processor_handle = tokio::spawn(async { processor.run().await });
-                    let k8s_handle = tokio::spawn(async move { k8s_start_client(tx).await });
+                    let k8s_handle = tokio::spawn(async move { k8s_start_client(params.hostname.as_str(), tx).await });
                     tokio::select! {
                         result = processor_handle => {
                             match result {