feat(mgmt): Add task to periodically patch status for gateway

mvachhar · mvachhar · commit 60788747babe · 2025-12-15T13:02:41.000-07:00
Adds a tokio task to periodically update the status directly
with the k8s API.

Also patches the status whenever a new config generation is
applied.

Also fixes an issue where the same config generation was
repeatedly applied.

Signed-off-by: Manish Vachharajani &lt;manish@githedgehog.com&gt;
diff --git a/mgmt/src/processor/k8s_client.rs b/mgmt/src/processor/k8s_client.rs
@@ -1,56 +1,229 @@
 // SPDX-License-Identifier: Apache-2.0
 // Copyright Open Network Fabric Authors
 
+use std::time::SystemTime;
+
+use chrono::{TimeZone, Utc};
+use config::converters::k8s::ToK8sConversionError;
 use tokio::sync::mpsc::Sender;
 
-use config::{ExternalConfig, GwConfig};
-use k8s_intf::client::WatchError;
-use k8s_intf::watch_gateway_agent_crd;
-use tracing::error;
+use config::converters::k8s::status::dataplane_status::DataplaneStatusForK8sConversion;
+use config::{ExternalConfig, GwConfig, internal::status::DataplaneStatus};
+use k8s_intf::client::{
+    ReplaceStatusError, WatchError, replace_gateway_status, watch_gateway_agent_crd,
+};
+use k8s_intf::gateway_agent_crd::GatewayAgentStatus;
+use tracing::{debug, error};
 
 use crate::processor::proc::{ConfigChannelRequest, ConfigRequest, ConfigResponse};
 
 #[derive(Debug, thiserror::Error)]
 pub enum K8sClientError {
     #[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),
+    #[error("Failed to convert dataplane status to k8s format: {0}")]
+    StatusConversionError(#[from] ToK8sConversionError),
+    #[error("Failed to patch k8s gateway status: {0}")]
+    ReplaceStatusError(#[from] ReplaceStatusError),
 }
 
-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}");
+async fn get_dataplane_status(
+    tx: &Sender<ConfigChannelRequest>,
+) -> Result<DataplaneStatus, MgmtStatusError> {
+    let (req, rx) = ConfigChannelRequest::new(ConfigRequest::GetDataplaneStatus);
+    tx.send(req).await.map_err(|_| {
+        MgmtStatusError::FetchStatusError("Failure relaying status fetch request".to_string())
+    })?;
+    let response = rx.await.map_err(|_| {
+        MgmtStatusError::FetchStatusError(
+            "Failure receiving status from config processor".to_string(),
+        )
+    })?;
+
+    match response {
+        ConfigResponse::GetDataplaneStatus(status) => Ok(*status),
+        _ => unreachable!(),
+    }
+}
+
+async fn get_current_config(
+    tx: &Sender<ConfigChannelRequest>,
+) -> Result<GwConfig, MgmtStatusError> {
+    let (req, rx) = ConfigChannelRequest::new(ConfigRequest::GetCurrentConfig);
+    tx.send(req).await.map_err(|_| {
+        MgmtStatusError::FetchStatusError("Failure relaying get config request".to_string())
+    })?;
+    let response = rx.await.map_err(|_| {
+        MgmtStatusError::FetchStatusError("Failure receiving config from processor".to_string())
+    })?;
+    match response {
+        ConfigResponse::GetCurrentConfig(opt_config) => {
+            opt_config.ok_or(MgmtStatusError::NoConfigApplied)
+        }
+        _ => unreachable!(),
+    }
+}
+
+fn to_datetime(opt_time: Option<&SystemTime>) -> chrono::DateTime<Utc> {
+    match opt_time {
+        Some(time) => chrono::DateTime::<Utc>::from(*time),
+        None => Utc.timestamp_opt(0, 0).unwrap(),
+    }
+}
+
+async fn update_gateway_status(hostname: &str, tx: &Sender<ConfigChannelRequest>) -> () {
+    let status = get_dataplane_status(tx).await;
+
+    let status = match status {
+        Ok(status) => status,
+        Err(err) => {
+            error!(
+                "Failed to fetch dataplane status, skipping status update: {}",
+                err
+            );
+            return;
+        }
+    };
+
+    let (last_applied_gen, last_applied_time) = match get_current_config(tx).await {
+        Ok(config) => (config.genid(), to_datetime(config.meta.apply_t.as_ref())),
+        Err(e) => {
+            error!("Failed to get current config, skipping status update: {e}");
+            return;
+        }
+    };
+
+    let k8s_status = match GatewayAgentStatus::try_from(&DataplaneStatusForK8sConversion {
+        last_applied_gen: Some(last_applied_gen),
+        last_applied_time: Some(&last_applied_time),
+        last_collected_time: Some(&chrono::Utc::now()),
+        status: Some(&status),
+    }) {
+        Ok(status) => status,
+        Err(err) => {
+            error!("Failed to convert status to GatewayAgentStatus: {err}");
+            return;
+        }
+    };
+
+    match replace_gateway_status(hostname, &k8s_status).await {
+        Ok(()) => (),
+        Err(err) => {
+            error!("Failed to update gateway status: {err}");
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+enum MgmtStatusError {
+    #[error("Failed to fetch dataplane status: {0}")]
+    FetchStatusError(String),
+    #[error("No config is currently applied")]
+    NoConfigApplied,
+}
+
+pub struct K8sClient {
+    hostname: String,
+}
+
+impl K8sClient {
+    pub fn new(hostname: &str) -> Self {
+        Self {
+            hostname: hostname.to_string(),
+        }
+    }
+
+    pub async fn init(&self) -> Result<(), K8sClientError> {
+        // Reset the config generation and applied time in K8s
+        replace_gateway_status(
+            &self.hostname,
+            &GatewayAgentStatus {
+                agent_version: None,
+                last_applied_gen: Some(0),
+                last_applied_time: Some(
+                    Utc.timestamp_opt(0, 0)
+                        .unwrap()
+                        .to_rfc3339_opts(chrono::SecondsFormat::Nanos, true),
+                ),
+                state: None,
+            },
+        )
+        .await?;
+        Ok(())
+    }
+
+    pub async fn k8s_start_config_watch(
+        &self,
+        tx: Sender<ConfigChannelRequest>,
+    ) -> Result<(), K8sClientError> {
+        // Clone this here so that the closure does not try to borrow self
+        // and cause K8sClient to not be Send for 'static but only a specific
+        // lifetime
+        let hostname = self.hostname.clone();
+        watch_gateway_agent_crd(&hostname.clone(), async move |ga| {
+            let external_config = ExternalConfig::try_from(ga);
+            match external_config {
+                Ok(external_config) => {
+                    let genid = external_config.genid;
+                    let current_genid = match get_current_config(&tx).await {
+                        Ok(config) => config.genid(),
+                        Err(e) => match e {
+                            MgmtStatusError::NoConfigApplied => 0,
+                            _ => {
+                                error!("Failed to get current config generation: {e}");
+                                return;
+                            }
                         }
-                        ConfigResponse::ApplyConfig(Ok(())) => {}
-                        _ => unreachable!(),
-                    },
-                };
-            }
-            Err(e) => {
-                error!("Failed to convert K8sGatewayAgent to ExternalConfig: {e}");
+                    };
+                    if current_genid == genid {
+                        debug!("Not applying config, configuration generation unchanged (old={current_genid}, new={genid})");
+                        return;
+                    }
+
+                    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(())) => {
+                                update_gateway_status(&hostname, &tx).await;
+                            }
+                            _ => unreachable!(),
+                        },
+                    };
+                }
+                Err(e) => {
+                    error!("Failed to convert K8sGatewayAgent to ExternalConfig: {e}");
+                }
             }
+        })
+        .await?;
+        Err(K8sClientError::EarlyTermination)
+    }
+
+    pub async fn k8s_start_status_update(
+        &self,
+        tx: Sender<ConfigChannelRequest>,
+        status_update_interval: &std::time::Duration,
+    ) -> Result<(), K8sClientError> {
+        // Clone this here so that the closure does not try to borrow self
+        // and cause K8sClient to not be Send for 'static but only a specific
+        // lifetime
+        let hostname = self.hostname.clone();
+        loop {
+            update_gateway_status(&hostname, &tx).await;
+            tokio::time::sleep(*status_update_interval).await;
         }
-    })
-    .await?;
-    Err(K8sClientError::EarlyTermination)
+    }
 }
diff --git a/mgmt/src/processor/launch.rs b/mgmt/src/processor/launch.rs
@@ -1,8 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // Copyright Open Network Fabric Authors
 
-use crate::processor::k8s_client::K8sClientError;
-use crate::processor::k8s_client::k8s_start_client;
+use crate::processor::k8s_client::{K8sClient, K8sClientError};
 use crate::processor::proc::ConfigChannelRequest;
 use crate::processor::proc::ConfigProcessor;
 
@@ -19,7 +18,10 @@ use tokio::sync::mpsc::Sender;
 use tokio_stream::Stream;
 use tonic::transport::Server;
 
+use futures::future::OptionFuture;
+
 use args::GrpcAddress;
+use concurrency::sync::Arc;
 use tracing::{debug, error, info, warn};
 
 use crate::grpc::server::create_config_service;
@@ -187,6 +189,8 @@ pub struct MgmtParams {
     pub processor_params: ConfigProcessorParams,
 }
 
+const STATUS_UPDATE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(15);
+
 /// Start the mgmt service with either type of socket
 pub fn start_mgmt(
     params: MgmtParams,
@@ -232,29 +236,56 @@ pub fn start_mgmt(
             } else {
                 debug!("Will start watching k8s for configuration changes");
                 rt.block_on(async {
+                    let k8s_client = Arc::new(K8sClient::new(params.hostname.as_str()));
                     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(params.hostname.as_str(), tx).await });
-                    tokio::select! {
-                        result = processor_handle => {
-                            match result {
-                                Ok(_) => {
-                                    error!("Configuration processor task exited unexpectedly");
-                                    Err(LaunchError::PrematureProcessorExit)?
+                    let tx1 = tx.clone();
+                    let k8s_client1 = k8s_client.clone();
+
+                    k8s_client.init().await.map_err(|e| {
+                        error!("Failed to initialize k8s state: {}", e);
+                        LaunchError::K8sClientError(e)
+                    })?;
+                    let mut processor_handle = Some(tokio::spawn(async { processor.run().await }));
+                    let mut k8s_config_handle = Some(tokio::spawn(async move { k8s_client.k8s_start_config_watch(tx).await }));
+                    let mut k8s_status_handle = Some(tokio::spawn(async move {
+                        k8s_client1.k8s_start_status_update(tx1, &STATUS_UPDATE_INTERVAL).await
+                    }));
+                    loop {
+                        tokio::select! {
+                            Some(result) = OptionFuture::from(processor_handle.as_mut()) => {
+                                match result {
+                                    Ok(_) => {
+                                        error!("Configuration processor task exited unexpectedly");
+                                        Err(LaunchError::PrematureProcessorExit)?
+                                    }
+                                    Err(e) => { Err::<(), LaunchError>(LaunchError::ProcessorJoinError(e)) }
                                 }
-                                Err(e) => { Err::<(), LaunchError>(LaunchError::ProcessorJoinError(e)) }
                             }
-                        }
-                        result = k8s_handle => {
-                            match result {
-                                Ok(result) => { result.inspect_err(|e| error!("K8s client task failed: {e}")).map_err(LaunchError::K8sClientError)?;
-                                    error!("Kubernetes client task exited unexpectedly");
-                                    Err(LaunchError::PrematureK8sClientExit)?
+                            Some(result) = OptionFuture::from(k8s_config_handle.as_mut()) => {
+                                match result {
+                                    Ok(result) => { result.inspect_err(|e| error!("K8s config watch task failed: {e}")).map_err(LaunchError::K8sClientError)?;
+                                        error!("Kubernetes config watch task exited unexpectedly");
+                                        Err(LaunchError::PrematureK8sClientExit)?
+                                    }
+                                    Err(e) => { Err(LaunchError::K8sClientJoinError(e))? }
                                 }
-                                Err(e) => { Err(LaunchError::K8sClientJoinError(e))? }
                             }
+                            Some(result) = OptionFuture::from(k8s_status_handle.as_mut()) => {
+                                k8s_status_handle = None;
+                                match result {
+                                    Ok(result) => { result.inspect_err(|e| error!("K8s status update task failed: {e}")).map_err(LaunchError::K8sClientError)?;
+                                        error!("Kubernetes status update task exited unexpectedly");
+                                        Err(LaunchError::PrematureK8sClientExit)?
+                                    }
+                                    Err(e) => { Err(LaunchError::K8sClientJoinError(e))? }
+                                }
+                            }
+                        }?;
+
+                        if processor_handle.is_none() && k8s_config_handle.is_none() && k8s_status_handle.is_none() {
+                            break;
                         }
-                    }?;
+                    }
                     Ok(())
                 })
             }
