feat(mgmt): add k8s-less mode

Add a k8s-less client that watches the config directory and uses
the config client to request the config processor to apply a new
config.

Signed-off-by: Fredi Raspall <fredi@githedgehog.com>

Author: Fredi-raspall

Cargo.lock

@@ -158,6 +158,7 @@ fn main() {
     /* start management */
     start_mgmt(MgmtParams {
         grpc_addr,
+        config_dir: args.config_dir().cloned(),
         hostname: get_gw_name().unwrap_or_else(|| unreachable!()).to_owned(),
         processor_params: ConfigProcessorParams {
             router_ctl: setup.router.get_ctl_tx(),

dataplane/src/main.rs

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

mgmt/Cargo.toml

@@ -0,0 +1,69 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Open Network Fabric Authors
+
+use config::{ExternalConfig, GwConfig};
+use futures::TryFutureExt;
+use k8s_less::kubeless_watch_gateway_agent_crd;
+use std::sync::Arc;
+use tracing::{error, info};
+
+use crate::processor::mgmt_client::{ConfigClient, ConfigProcessorError};
+
+#[derive(Debug, thiserror::Error)]
+pub enum K8sLessError {
+    #[error("K8sless exited early")]
+    EarlyTermination,
+    #[error("Watching error: {0}")]
+    WatchError(String),
+}
+
+pub struct K8sLess {
+    pathdir: String,
+    client: ConfigClient,
+}
+
+impl K8sLess {
+    pub fn new(pathdir: &str, client: ConfigClient) -> Self {
+        Self {
+            pathdir: pathdir.to_string(),
+            client,
+        }
+    }
+
+    pub async fn start_config_watch(k8sless: Arc<Self>) -> Result<(), K8sLessError> {
+        info!("Starting config watcher for directory {}", k8sless.pathdir);
+
+        kubeless_watch_gateway_agent_crd(&k8sless.pathdir.clone(), async move |ga| {
+            info!("Attempting to deserialize new gateway CRD ...");
+
+            let external_config = ExternalConfig::try_from(ga);
+            match external_config {
+                Err(e) => error!("Failed to convert K8sGatewayAgent to ExternalConfig: {e}"),
+                Ok(external_config) => {
+                    let genid = external_config.genid;
+                    let applied_genid = match k8sless.client.get_generation().await {
+                        Ok(genid) => genid,
+                        Err(ConfigProcessorError::NoConfigApplied) => 0,
+                        Err(e) => {
+                            error!("Failed to get current config generation: {e}");
+                            return;
+                        }
+                    };
+                    info!("Current configuration is {applied_genid}");
+
+                    let gwconfig = GwConfig::new(external_config);
+
+                    // request the config processor to apply the config and update status on success
+                    match k8sless.client.apply_config(gwconfig).await {
+                        Ok(()) => info!("Config for generation {genid} was successfully applied"),
+                        Err(e) => error!("Failed to apply the config for generation {genid}: {e}"),
+                    }
+                }
+            }
+        })
+        .map_err(K8sLessError::WatchError)
+        .await?;
+
+        Err(K8sLessError::EarlyTermination)
+    }
+}

mgmt/src/processor/k8s_less_client.rs

@@ -2,6 +2,7 @@
 // Copyright Open Network Fabric Authors
 
 use crate::processor::k8s_client::{K8sClient, K8sClientError};
+use crate::processor::k8s_less_client::{K8sLess, K8sLessError};
 use crate::processor::proc::ConfigProcessor;
 
 use std::fmt::Display;
@@ -47,6 +48,9 @@ pub enum LaunchError {
     ProcessorError(std::io::Error),
     #[error("Error starting/waiting for Config Processor task: {0}")]
     ProcessorJoinError(tokio::task::JoinError),
+
+    #[error("Error in k8s-less mode: {0}")]
+    K8LessError(#[from] K8sLessError),
 }
 
 /// Start the gRPC server on TCP
@@ -181,6 +185,7 @@ impl Display for ServerAddress {
 
 pub struct MgmtParams {
     pub grpc_addr: Option<GrpcAddress>,
+    pub config_dir: Option<String>,
     pub hostname: String,
     pub processor_params: ConfigProcessorParams,
 }
@@ -229,7 +234,17 @@ pub fn start_mgmt(
                         Err(LaunchError::PrematureGrpcExit)
                     }
                 })
-            } else {
+            }  else if let Some(config_dir) = &params.config_dir {
+                warn!("Running in k8s-less mode....");
+                rt.block_on(async {
+                    let (processor, client) = ConfigProcessor::new(params.processor_params);
+                    let k8sless = Arc::new(K8sLess::new(config_dir, client));
+                    tokio::spawn(async { processor.run().await });
+                    K8sLess::start_config_watch(k8sless).await
+                })?;
+                Ok(())
+            }
+            else {
                 debug!("Will start watching k8s for configuration changes");
                 rt.block_on(async {
                     let (processor, client) = ConfigProcessor::new(params.processor_params);

mgmt/src/processor/launch.rs

@@ -8,6 +8,7 @@ pub(crate) mod confbuild;
 mod display;
 pub(crate) mod gwconfigdb;
 pub(crate) mod k8s_client;
+pub(crate) mod k8s_less_client;
 pub(crate) mod launch;
 pub(crate) mod mgmt_client;
 pub(crate) mod proc;