feat(controlplane): gateway resource cleanup (#443)

use a finalizer to clean up created resources

fixes #440

Signed-off-by: Harald Gutmann <harald@gutmann.one>

Author: hargut

controlplane/src/consts.rs

@@ -11,6 +11,9 @@ pub const BLIXT_DATAPLANE_COMPONENT_LABEL: &str = "dataplane";
 // The finalizer used for Blixt dataplane cleanup.
 pub const DATAPLANE_FINALIZER: &str = "blixt.gateway.networking.k8s.io/dataplane";
 
+// The finalizer used for Blixt controlplane cleanup.
+pub const CONTROLPLANE_FINALIZER: &str = "blixt.gateway.networking.k8s.io/controlplane";
+
 // Controller name for the Blixt GatewayClass.
 pub const GATEWAY_CLASS_CONTROLLER_NAME: &str = "gateway.networking.k8s.io/blixt";
 

controlplane/src/gateway_controller.rs

@@ -20,13 +20,6 @@ use std::{
     time::{Duration, Instant},
 };
 
-use crate::{
-    consts::{GATEWAY_CLASS_CONTROLLER_NAME, GATEWAY_SERVICE_LABEL},
-    *,
-};
-use gateway_utils::*;
-use route_utils::set_condition;
-
 use chrono::Utc;
 use futures::StreamExt;
 use gateway_api::apis::standard::gateways::{Gateway, GatewayStatus};
@@ -39,33 +32,31 @@ use k8s_openapi::apimachinery::pkg::apis::meta::v1 as metav1;
 use kube::{
     Resource, ResourceExt,
     api::{Api, ListParams, Patch, PatchParams},
-    runtime::{Controller, controller::Action, watcher::Config},
+    runtime::{
+        Controller, controller::Action, finalizer, finalizer::Error as FinalizerError,
+        finalizer::Event, watcher::Config,
+    },
 };
 use tracing::{debug, error, info, warn};
 
+use crate::{
+    Context, Error, NamespaceName, Result,
+    consts::{CONTROLPLANE_FINALIZER, GATEWAY_CLASS_CONTROLLER_NAME, GATEWAY_SERVICE_LABEL},
+    gateway_utils::{
+        create_endpoint_if_not_exists, create_loadbalancer_service, delete_endpoint,
+        delete_loadbalancer_service, get_accepted_condition, get_ingress_ip_len, get_service_key,
+        patch_status, set_gateway_status_addresses, set_listener_status,
+        update_service_for_gateway,
+    },
+    gatewayclass_utils,
+    route_utils::set_condition,
+};
+
 pub async fn reconcile(gateway: Arc<Gateway>, ctx: Arc<Context>) -> Result<Action> {
     let start = Instant::now();
     let client = ctx.client.clone();
 
-    let name = gateway
-        .metadata
-        .name
-        .clone()
-        .ok_or(Error::InvalidConfigError("invalid name".to_string()))?;
-
-    let ns = gateway
-        .metadata
-        .namespace
-        .clone()
-        .ok_or(Error::InvalidConfigError("invalid namespace".to_string()))?;
-
-    let gateway_api: Api<Gateway> = Api::namespaced(client.clone(), &ns);
-    let mut gw = Gateway {
-        metadata: gateway.metadata.clone(),
-        spec: gateway.spec.clone(),
-        status: gateway.status.clone(),
-    };
-
+    // check gateway_class
     let gateway_class_api = Api::<GatewayClass>::all(client.clone());
     let gateway_class = gateway_class_api
         .get(gateway.spec.gateway_class_name.as_str())
@@ -90,6 +81,47 @@ pub async fn reconcile(gateway: Arc<Gateway>, ctx: Arc<Context>) -> Result<Actio
         return Ok(Action::await_change());
     }
 
+    let gateway_id = gateway.metadata.namespaced_name()?;
+    let gateway_api: Api<Gateway> = Api::namespaced(ctx.client.clone(), &gateway_id.namespace);
+    finalizer(
+        &gateway_api,
+        CONTROLPLANE_FINALIZER,
+        gateway,
+        |event| async {
+            match event {
+                Event::Apply(gateway) => configure_gateway(gateway, ctx.clone(), start).await,
+                Event::Cleanup(gateway) => cleanup_gateway(gateway, &ctx).await,
+            }
+        },
+    )
+    .await
+    .map_err(|e| match e {
+        FinalizerError::ApplyFailed(e) | FinalizerError::CleanupFailed(e) => e,
+        FinalizerError::AddFinalizer(e) | FinalizerError::RemoveFinalizer(e) => e.into(),
+        FinalizerError::UnnamedObject => Error::MissingResourceName,
+        FinalizerError::InvalidFinalizer => {
+            Error::InvalidConfigError("InvalidFinalizer".to_string())
+        }
+    })
+}
+
+pub async fn configure_gateway(
+    gateway: Arc<Gateway>,
+    ctx: Arc<Context>,
+    start: Instant,
+) -> Result<Action> {
+    let gateway_id = gateway.metadata.namespaced_name()?;
+    let ns = gateway_id.namespace.clone();
+    let name = gateway_id.name.clone();
+    let client = ctx.client.clone();
+
+    let gateway_api: Api<Gateway> = Api::namespaced(ctx.client.clone(), &gateway_id.namespace);
+    let mut gw = Gateway {
+        metadata: gateway.metadata.clone(),
+        spec: gateway.spec.clone(),
+        status: gateway.status.clone(),
+    };
+
     set_listener_status(&mut gw)?;
     let accepted_cond = get_accepted_condition(&gw);
     set_condition(&mut gw, accepted_cond.clone());
@@ -154,7 +186,7 @@ pub async fn reconcile(gateway: Arc<Gateway>, ctx: Arc<Context>) -> Result<Actio
         }
     } else {
         info!("creating loadbalancer service");
-        service = create_svc_for_gateway(ctx.clone(), gateway.as_ref()).await?;
+        service = create_loadbalancer_service(ctx.clone(), gateway.as_ref()).await?;
     }
 
     // invalid_lb_condition is a Condition that signifies that the Loadbalancer service is invalid.
@@ -220,6 +252,15 @@ pub async fn reconcile(gateway: Arc<Gateway>, ctx: Arc<Context>) -> Result<Actio
     Ok(Action::requeue(Duration::from_secs(60)))
 }
 
+async fn cleanup_gateway(gateway: Arc<Gateway>, ctx: &Context) -> Result<Action> {
+    let gateway_id = gateway.metadata.namespaced_name()?;
+
+    delete_endpoint(ctx.client.clone(), &gateway_id).await?;
+    delete_loadbalancer_service(ctx.client.clone(), &gateway_id).await?;
+
+    Ok(Action::await_change())
+}
+
 pub async fn controller(ctx: Context) -> Result<()> {
     let gateway = Api::<Gateway>::all(ctx.client.clone());
     gateway

controlplane/src/gateway_utils.rs

@@ -48,6 +48,7 @@ use k8s_openapi::api::core::v1::{
 use k8s_openapi::apimachinery::pkg::apis::meta::v1 as metav1;
 
 use chrono::Utc;
+use kube::api::DeleteParams;
 use serde_json::json;
 use tracing::*;
 
@@ -149,6 +150,19 @@ pub async fn create_endpoint_if_not_exists(
     Ok(())
 }
 
+// Deletes the Endpoint for the provided Gateway.
+pub(crate) async fn delete_endpoint(k8s_client: Client, gateway_id: &NamespacedName) -> Result<()> {
+    let endpoint_name = format!("service-for-gateway-{}", gateway_id.name.clone());
+    info!("Deleting Endpoint {endpoint_name} for gateway {gateway_id}.");
+
+    let endpoints_api: Api<Endpoints> = Api::namespaced(k8s_client.clone(), &gateway_id.namespace);
+    endpoints_api
+        .delete(&endpoint_name, &DeleteParams::default())
+        .await
+        .map_err(Error::KubeError)
+        .map(|_| ())
+}
+
 // Returns true if the provided error is a not found error.
 pub fn check_if_not_found_err(error: kube::Error) -> bool {
     if let kube::Error::Api(response) = error
@@ -170,11 +184,11 @@ pub fn get_ingress_ip_len(svc_status: &ServiceStatus) -> usize {
 }
 
 // Creates a LoadBalancer Service for the provided Gateway.
-pub async fn create_svc_for_gateway(ctx: Arc<Context>, gateway: &Gateway) -> Result<Service> {
+pub async fn create_loadbalancer_service(ctx: Arc<Context>, gateway: &Gateway) -> Result<Service> {
     let mut svc_meta = ObjectMeta::default();
-    let ns = gateway.namespace().unwrap_or("default".to_string());
-    svc_meta.namespace = Some(ns.clone());
-    svc_meta.generate_name = Some(format!("service-for-gateway-{}-", gateway.name_any()));
+    let gateway_id = gateway.metadata.namespaced_name()?;
+    svc_meta.namespace = Some(gateway_id.namespace.clone());
+    svc_meta.name = Some(format!("service-for-gateway-{}", gateway_id.name));
 
     let mut labels = BTreeMap::new();
     labels.insert(GATEWAY_SERVICE_LABEL.to_string(), gateway.name_any());
@@ -187,7 +201,7 @@ pub async fn create_svc_for_gateway(ctx: Arc<Context>, gateway: &Gateway) -> Res
     };
     update_service_for_gateway(gateway, &mut svc)?;
 
-    let svc_api: Api<Service> = Api::namespaced(ctx.client.clone(), ns.as_str());
+    let svc_api: Api<Service> = Api::namespaced(ctx.client.clone(), &gateway_id.namespace);
     let service = svc_api
         .create(&PostParams::default(), &svc)
         .await
@@ -196,6 +210,22 @@ pub async fn create_svc_for_gateway(ctx: Arc<Context>, gateway: &Gateway) -> Res
     Ok(service)
 }
 
+// Deletes a LoadBalancer Service for the provided Gateway.
+pub(crate) async fn delete_loadbalancer_service(
+    k8s_client: Client,
+    gateway_id: &NamespacedName,
+) -> Result<()> {
+    let service_name = format!("service-for-gateway-{}", &gateway_id.name);
+    info!("Deleting Service {service_name} of type LoadBalancer for Gateway {gateway_id}",);
+
+    let service_api: Api<Service> = Api::namespaced(k8s_client, &gateway_id.namespace);
+    service_api
+        .delete(&service_name, &DeleteParams::default())
+        .await
+        .map_err(Error::KubeError)
+        .map(|_| ())
+}
+
 // Updates the provided Service to match the desired state according to the provided Gateway.
 // Returns true if Service was modified.
 pub fn update_service_for_gateway(gateway: &Gateway, svc: &mut Service) -> Result<bool> {

controlplane/src/lib.rs

@@ -25,14 +25,17 @@ mod tcproute_controller;
 mod traits;
 mod udproute_controller;
 
+use std::fmt::{Debug, Display, Formatter};
+
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+use kube::Client;
+use thiserror::Error;
+
 pub use gateway_controller::controller as gateway_controller;
 pub use gatewayclass_controller::controller as gatewayclass_controller;
 pub use tcproute_controller::controller as tcproute_controller;
 pub use udproute_controller::controller as udproute_controller;
 
-use kube::Client;
-use thiserror::Error;
-
 // Context for our reconciler
 #[derive(Clone)]
 pub struct Context {
@@ -43,7 +46,7 @@ pub struct Context {
 #[derive(Error, Debug)]
 pub enum Error {
     #[error("kube error: {0}")]
-    KubeError(#[source] kube::Error),
+    KubeError(#[from] kube::Error),
     #[error("invalid configuration: `{0}`")]
     InvalidConfigError(String),
     #[error("error reconciling loadbalancer service: `{0}`")]
@@ -52,11 +55,55 @@ pub enum Error {
     CRDNotFoundError(#[source] kube::Error),
     #[error("dataplane error: {0}")]
     DataplaneError(String),
+    #[error("missing resource namespace")]
+    MissingResourceNamespace,
+    #[error("missing resource name")]
+    MissingResourceName,
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
 
+#[derive(Clone, Hash, Eq, PartialEq)]
 pub struct NamespacedName {
     pub name: String,
     pub namespace: String,
 }
+
+impl Display for NamespacedName {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.namespace.as_str())?;
+        f.write_str("/")?;
+        f.write_str(self.name.as_str())
+    }
+}
+
+impl Debug for NamespacedName {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        Display::fmt(self, f)
+    }
+}
+
+pub trait NamespaceName {
+    fn namespace(&self) -> std::result::Result<&str, Error>;
+    fn name(&self) -> std::result::Result<&str, Error>;
+    fn namespaced_name(&self) -> std::result::Result<NamespacedName, Error>;
+}
+
+impl NamespaceName for ObjectMeta {
+    fn namespace(&self) -> std::result::Result<&str, Error> {
+        self.namespace
+            .as_deref()
+            .ok_or(Error::MissingResourceNamespace)
+    }
+
+    fn name(&self) -> std::result::Result<&str, Error> {
+        self.name.as_deref().ok_or(Error::MissingResourceName)
+    }
+
+    fn namespaced_name(&self) -> std::result::Result<NamespacedName, Error> {
+        Ok(NamespacedName {
+            name: self.name()?.to_string(),
+            namespace: self.namespace()?.to_string(),
+        })
+    }
+}