Clean lb crd status upon Load Balancer removal

When a lb transitions to ERROR or the IP on the Service
spec differs from the lb VIP, the lb is released and
the CRD doesn't get updated, causing Not Found expections
when handling the creation of others load balancer
resources. This commit fixes the issue by ensuring the
clean up of the status field happens upon lb release.
Also, it adds protection in case we still get
nonexistent lb on the CRD.

Closes-Bug: 1894758
Change-Id: I484ece6a7b52b51d878f724bd4fad0494eb759d6

kuryr_kubernetes/controller/drivers/lbaasv2.py

@@ -502,6 +502,7 @@ class LBaaSv2Driver(base.LBaaSDriver):
             loadbalancer['provider'] = os_lb.provider
             if os_lb.provisioning_status == 'ERROR':
                 self.release_loadbalancer(loadbalancer)
+                utils.clean_lb_crd_status(loadbalancer['name'])
                 return None
         except (KeyError, StopIteration):
             return None
@@ -537,8 +538,12 @@ class LBaaSv2Driver(base.LBaaSDriver):
         }
 
         # Wait for the loadbalancer to be ACTIVE
-        self._wait_for_provisioning(loadbalancer, _ACTIVATION_TIMEOUT,
+        if not self._wait_for_provisioning(
-                                    _LB_STS_POLL_FAST_INTERVAL)
+                loadbalancer, _ACTIVATION_TIMEOUT,
+                _LB_STS_POLL_FAST_INTERVAL):
+            LOG.debug('Skipping ACLs update. '
+                      'No Load Balancer Provisioned.')
+            return
 
         lbaas = clients.get_loadbalancer_client()
         try:
@@ -665,7 +670,9 @@ class LBaaSv2Driver(base.LBaaSDriver):
                             interval=_LB_STS_POLL_FAST_INTERVAL):
         for remaining in self._provisioning_timer(_ACTIVATION_TIMEOUT,
                                                   interval):
-            self._wait_for_provisioning(loadbalancer, remaining, interval)
+            if not self._wait_for_provisioning(
+                    loadbalancer, remaining, interval):
+                return None
             try:
                 result = self._ensure(create, find, obj, loadbalancer)
                 if result:
@@ -684,7 +691,9 @@ class LBaaSv2Driver(base.LBaaSDriver):
                     delete(*args, **kwargs)
                     return
                 except (os_exc.ConflictException, os_exc.BadRequestException):
-                    self._wait_for_provisioning(loadbalancer, remaining)
+                    if not self._wait_for_provisioning(
+                            loadbalancer, remaining):
+                        return
             except os_exc.NotFoundException:
                 return
 
@@ -695,17 +704,30 @@ class LBaaSv2Driver(base.LBaaSDriver):
         lbaas = clients.get_loadbalancer_client()
 
         for remaining in self._provisioning_timer(timeout, interval):
-            response = lbaas.get_load_balancer(loadbalancer['id'])
+            try:
+                response = lbaas.get_load_balancer(loadbalancer['id'])
+            except os_exc.ResourceNotFound:
+                LOG.debug("Cleaning CRD status for deleted "
+                          "loadbalancer %s", loadbalancer['name'])
+                utils.clean_lb_crd_status(loadbalancer['name'])
+                return None
+
             status = response.provisioning_status
             if status == 'ACTIVE':
                 LOG.debug("Provisioning complete for %(lb)s", {
                     'lb': loadbalancer})
-                return
+                return loadbalancer
             elif status == 'ERROR':
                 LOG.debug("Releasing loadbalancer %s with error status",
                           loadbalancer['id'])
                 self.release_loadbalancer(loadbalancer)
-                break
+                utils.clean_lb_crd_status(loadbalancer['name'])
+                return None
+            elif status == 'DELETED':
+                LOG.debug("Cleaning CRD status for deleted "
+                          "loadbalancer %s", loadbalancer['name'])
+                utils.clean_lb_crd_status(loadbalancer['name'])
+                return None
             else:
                 LOG.debug("Provisioning status %(status)s for %(lb)s, "
                           "%(rem).3gs remaining until timeout",

kuryr_kubernetes/controller/handlers/loadbalancer.py

@@ -229,17 +229,24 @@ class KuryrLoadBalancerHandler(k8s_base.ResourceEventHandler):
                           {'security_groups': lb_sgs})
         except k_exc.K8sResourceNotFound:
             LOG.debug('KuryrLoadBalancer %s not found', svc_name)
+            return None
+        except k_exc.K8sUnprocessableEntity:
+            LOG.debug('KuryrLoadBalancer entity not processable '
+                      'due to missing loadbalancer field.')
+            return None
         except k_exc.K8sClientException:
             LOG.exception('Error syncing KuryrLoadBalancer'
                           ' %s', svc_name)
             raise
-
         return klb_crd
 
     def _add_new_members(self, loadbalancer_crd):
         changed = False
+
         if loadbalancer_crd['status'].get('loadbalancer'):
             loadbalancer_crd = self._sync_lbaas_sgs(loadbalancer_crd)
+        if not loadbalancer_crd:
+            return changed
 
         lsnr_by_id = {l['id']: l for l in loadbalancer_crd['status'].get(
             'listeners', [])}
@@ -328,6 +335,8 @@ class KuryrLoadBalancerHandler(k8s_base.ResourceEventHandler):
                         target_ref_namespace=target_ref['namespace'],
                         target_ref_name=target_ref['name'],
                         listener_port=listener_port)
+                    if not member:
+                        continue
                     members = loadbalancer_crd['status'].get('members', [])
                     if members:
                         loadbalancer_crd['status'].get('members', []).append(
@@ -465,6 +474,8 @@ class KuryrLoadBalancerHandler(k8s_base.ResourceEventHandler):
                 continue
             pool = self._drv_lbaas.ensure_pool(loadbalancer_crd['status'][
                 'loadbalancer'], listener)
+            if not pool:
+                continue
             pools = loadbalancer_crd['status'].get('pools', [])
             if pools:
                 loadbalancer_crd['status'].get('pools', []).append(
@@ -670,26 +681,14 @@ class KuryrLoadBalancerHandler(k8s_base.ResourceEventHandler):
             if pub_info:
                 self._drv_service_pub_ip.disassociate_pub_ip(
                     loadbalancer_crd['status']['service_pub_ip_info'])
+                self._drv_service_pub_ip.release_pub_ip(
+                    loadbalancer_crd['status']['service_pub_ip_info'])
 
             self._drv_lbaas.release_loadbalancer(
                 loadbalancer=lb)
-            lb = None
-            loadbalancer_crd['status']['pools'] = []
-            loadbalancer_crd['status']['listeners'] = []
-            loadbalancer_crd['status']['members'] = []
 
-            kubernetes = clients.get_kubernetes_client()
+            lb = {}
-            try:
+            loadbalancer_crd['status'] = {}
-                kubernetes.patch_crd('status', loadbalancer_crd['metadata'][
-                    'selfLink'], loadbalancer_crd['status'])
-            except k_exc.K8sResourceNotFound:
-                LOG.debug('KuryrLoadbalancer CRD not found %s',
-                          loadbalancer_crd)
-            except k_exc.K8sClientException:
-                LOG.exception('Error updating KuryrLoadbalancer CRD %s',
-                              loadbalancer_crd)
-                raise
-            changed = True
 
         if not lb:
             if loadbalancer_crd['spec'].get('ip'):
@@ -706,12 +705,6 @@ class KuryrLoadBalancerHandler(k8s_base.ResourceEventHandler):
                     service_type=loadbalancer_crd['spec'].get('type'),
                     provider=self._lb_provider)
                 loadbalancer_crd['status']['loadbalancer'] = lb
-                changed = True
-            elif loadbalancer_crd['status'].get('service_pub_ip_info'):
-                self._drv_service_pub_ip.release_pub_ip(
-                    loadbalancer_crd['status']['service_pub_ip_info'])
-                loadbalancer_crd['status']['service_pub_ip_info'] = None
-                changed = True
 
             kubernetes = clients.get_kubernetes_client()
             try:
@@ -724,6 +717,7 @@ class KuryrLoadBalancerHandler(k8s_base.ResourceEventHandler):
                 LOG.exception('Error updating KuryrLoadbalancer CRD %s',
                               loadbalancer_crd)
                 raise
+            changed = True
 
         return changed
 

kuryr_kubernetes/exceptions.py

@@ -52,6 +52,12 @@ class K8sNamespaceTerminating(K8sForbidden):
             "Namespace already terminated: %r" % message)
 
 
+class K8sUnprocessableEntity(K8sClientException):
+    def __init__(self, message):
+        super(K8sUnprocessableEntity, self).__init__(
+            "Unprocessable: %r" % message)
+
+
 class InvalidKuryrNetworkAnnotation(Exception):
     pass
 

kuryr_kubernetes/k8s_client.py

@@ -87,6 +87,8 @@ class K8sClient(object):
             if 'because it is being terminated' in response.json()['message']:
                 raise exc.K8sNamespaceTerminating(response.text)
             raise exc.K8sForbidden(response.text)
+        if response.status_code == requests.codes.unprocessable_entity:
+            raise exc.K8sUnprocessableEntity(response.text)
         if not response.ok:
             raise exc.K8sClientException(response.text)
 
@@ -133,10 +135,15 @@ class K8sClient(object):
             data = [{'op': action,
                      'path': f'/{field}/{data}'}]
         else:
-            data = [{'op': action,
+            if data:
-                     'path': f'/{field}/{crd_field}',
+                data = [{'op': action,
-                     'value': value}
+                         'path': f'/{field}/{crd_field}',
-                    for crd_field, value in data.items()]
+                         'value': value}
+                        for crd_field, value in data.items()]
+            else:
+                data = [{'op': action,
+                         'path': f'/{field}',
+                         'value': data}]
 
         LOG.debug("Patch %(path)s: %(data)s", {
             'path': path, 'data': data})

kuryr_kubernetes/tests/unit/controller/drivers/test_lbaasv2.py

@@ -526,7 +526,7 @@ class TestLBaaSv2Driver(test_base.TestCase):
         cls = d_lbaasv2.LBaaSv2Driver
         m_driver = mock.Mock(spec=d_lbaasv2.LBaaSv2Driver)
         loadbalancer = {
-            'name': 'TEST_NAME',
+            'name': 'test_namespace/test_name',
             'project_id': 'TEST_PROJECT',
             'subnet_id': 'D3FA400A-F543-4B91-9CD3-047AF0CE42D1',
             'ip': '1.2.3.4',

kuryr_kubernetes/utils.py

@@ -423,3 +423,18 @@ def get_service_subnet_version():
         LOG.exception("Service subnet %s not found", svc_subnet_id)
         raise
     return svc_subnet.ip_version
+
+
+def clean_lb_crd_status(loadbalancer_name):
+    namespace, name = loadbalancer_name.split('/')
+    k8s = clients.get_kubernetes_client()
+    try:
+        k8s.patch_crd('status', f'{constants.K8S_API_CRD_NAMESPACES}'
+                      f'/{namespace}/kuryrloadbalancers/{name}', {})
+    except exceptions.K8sResourceNotFound:
+        LOG.debug('KuryrLoadbalancer CRD not found %s',
+                  name)
+    except exceptions.K8sClientException:
+        LOG.exception('Error updating KuryrLoadbalancer CRD %s',
+                      name)
+        raise