Add Listener sync logic

This patch adds the logic to sync a Listener entity from
the Octavia database, correcting any discrepancies in fields or
creating it if it does not exist in the OVN LB related on
OVN Northbound (NB) database.

Future patches will incrementally add support for syncing the
remaining entities.

Related-Bug: #2045415

Co-authored-by: Fernando Royo <[email protected]>
Co-authored-by: Rico Lin <[email protected]>
Change-Id: Ibf39a73386eed1e00c7ce39e60274e028df0cced

Author: fernandoroyosanchez

ovn_octavia_provider/driver.py

@@ -105,6 +105,20 @@ def _get_loadbalancer_request_info(self, loadbalancer):
                 loadbalancer.additional_vips
         return request_info
 
+    def _get_listener_request_info(self, listener):
+        self._check_for_supported_protocols(listener.protocol)
+        self._check_for_allowed_cidrs(listener.allowed_cidrs)
+        admin_state_up = listener.admin_state_up
+        if isinstance(admin_state_up, o_datamodels.UnsetType):
+            admin_state_up = True
+        request_info = {'id': listener.listener_id,
+                        'protocol': listener.protocol,
+                        'loadbalancer_id': listener.loadbalancer_id,
+                        'protocol_port': listener.protocol_port,
+                        'default_pool_id': listener.default_pool_id,
+                        'admin_state_up': admin_state_up}
+        return request_info
+
     def loadbalancer_create(self, loadbalancer):
         request = {'type': ovn_const.REQ_TYPE_LB_CREATE,
                    'info': self._get_loadbalancer_request_info(
@@ -598,9 +612,16 @@ def _ensure_loadbalancer(self, loadbalancer):
         except idlutils.RowNotFound:
             LOG.debug(f"OVN loadbalancer {loadbalancer.loadbalancer_id} "
                       "not found. Start create process.")
-            # TODO(froyo): By now just syncing LB only
+            # TODO(froyo): By now just syncing LB and listener only
             status = self._ovn_helper.lb_create(
                 self._get_loadbalancer_request_info(loadbalancer))
+
+            if not isinstance(loadbalancer.listeners, o_datamodels.UnsetType):
+                status[constants.LISTENERS] = []
+                for listener in loadbalancer.listeners:
+                    status_listener = self._ovn_helper.listener_create(
+                        self._get_listener_request_info(listener))
+                    status[constants.LISTENERS].append(status_listener)
             self._ovn_helper._update_status_to_octavia(status)
         else:
             # Load Balancer found, check LB and listener/pool/member/hms
@@ -611,6 +632,12 @@ def _ensure_loadbalancer(self, loadbalancer):
                     "found checking other entities related")
                 self._ovn_helper.lb_sync(
                     self._get_loadbalancer_request_info(loadbalancer), ovn_lb)
+                # Listener
+                if not isinstance(loadbalancer.listeners,
+                                  o_datamodels.UnsetType):
+                    for listener in loadbalancer.listeners:
+                        self._ovn_helper.listener_sync(
+                            self._get_listener_request_info(listener), ovn_lb)
                 status = self._ovn_helper._get_current_operating_statuses(
                     ovn_lb)
                 self._ovn_helper._update_status_to_octavia(status)
@@ -625,4 +652,11 @@ def do_sync(self, **lb_filters):
             provider_lb = (
                 self._ovn_helper._octavia_driver_lib.get_loadbalancer(lb.id)
             )
+
+            listeners = provider_lb.listeners or []
+            provider_lb.listeners = [
+                o_datamodels.Listener.from_dict(listener)
+                for listener in listeners
+            ] if listeners else o_datamodels.Unset
+
             self._ensure_loadbalancer(provider_lb)

ovn_octavia_provider/helper.py

@@ -619,6 +619,50 @@ def _sync_lb_to_lr_association(self, ovn_lb, ovn_lr):
                               "router %s: %s", ovn_lb.uuid, ovn_lr.uuid,
                               str(e))
 
+    def _build_listener_info(self, listener, external_ids):
+        """Build listener key and listener info."""
+        listener_key = self._get_listener_key(
+            listener.get(constants.ID),
+            is_enabled=listener.get(constants.ADMIN_STATE_UP)
+        )
+        pool_key = ''
+        if listener.get(constants.DEFAULT_POOL_ID):
+            pool_key = self._get_pool_key(
+                listener.get(constants.DEFAULT_POOL_ID))
+        external_ids[listener_key] = self._make_listener_key_value(
+            listener[constants.PROTOCOL_PORT], pool_key
+        )
+        listener_info = {listener_key: external_ids[listener_key]}
+        return listener_key, listener_info
+
+    def _update_listener_key_if_needed(self, listener_key, listener_info,
+                                       ovn_lb, commands):
+        """Update listener key on OVN LoadBalancer if needed."""
+        prev_listener_key_content = ovn_lb.external_ids.get(listener_key, '')
+        if (listener_key not in ovn_lb.external_ids or
+                listener_info.get(listener_key) != prev_listener_key_content):
+            commands.append(
+                self.ovn_nbdb_api.db_set(
+                    'Load_Balancer',
+                    ovn_lb.uuid,
+                    ('external_ids', listener_info)
+                )
+            )
+
+    def _update_protocol_if_needed(self, listener, ovn_lb, commands):
+        """Update protocol on OVN LoadBalancer if needed."""
+        current_protocol = ''
+        if ovn_lb.protocol:
+            current_protocol = ovn_lb.protocol[0].lower()
+        listener_protocol = str(listener.get(constants.PROTOCOL)).lower()
+        if current_protocol != listener_protocol:
+            commands.append(
+                self.ovn_nbdb_api.db_set(
+                    'Load_Balancer', ovn_lb.uuid,
+                    ('protocol', listener_protocol)
+                )
+            )
+
     def _lb_status(self, loadbalancer, provisioning_status, operating_status):
         """Return status for the LoadBalancer."""
         return {
@@ -1868,6 +1912,39 @@ def listener_create(self, listener):
                  constants.PROVISIONING_STATUS: constants.ACTIVE}]}
         return status
 
+    def listener_sync(self, listener, ovn_lb):
+        """Sync Listener object with an OVN LoadBalancer
+
+        The method performs the following steps:
+        1. Update listener key on OVN Loadbalancer external_ids if needed
+        2. Update OVN LoadBalancer protocol from Listener info if needed
+        3. Refresh OVN LoadBalancer vips
+
+        :param listener: The source listener object from Octavia DB
+        :param ovn_lb: The OVN LoadBalancer object that needs to be sync
+        """
+        commands = []
+        external_ids = copy.deepcopy(ovn_lb.external_ids)
+
+        listener_key, listener_info = self._build_listener_info(
+            listener, external_ids)
+        self._update_listener_key_if_needed(
+            listener_key, listener_info, ovn_lb, commands)
+        self._update_protocol_if_needed(listener, ovn_lb, commands)
+
+        try:
+            commands.extend(self._refresh_lb_vips(
+                ovn_lb, external_ids, is_sync=True))
+        except Exception as e:
+            LOG.exception(f"Failed to refresh LB VIPs: {e}")
+            return
+
+        try:
+            self._execute_commands(commands)
+        except Exception as e:
+            LOG.exception(f"Failed to execute commands for listener sync: {e}")
+            return
+
     def listener_delete(self, listener):
         status = {
             constants.LISTENERS: [

ovn_octavia_provider/tests/unit/test_driver.py

@@ -1237,32 +1237,82 @@ def test_health_monitor_delete(self):
 
     @mock.patch.object(ovn_helper.OvnProviderHelper,
                        '_update_status_to_octavia')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'listener_create')
     @mock.patch.object(ovn_helper.OvnProviderHelper, 'lb_create')
     def test_ensure_loadbalancer_lb_not_found(
-            self, mock_lb_create, mock_update_status):
+            self, mock_lb_create, mock_listener_create, mock_update_status):
         self.mock_find_ovn_lbs_with_retry.side_effect = [
             idlutils.RowNotFound]
-        self.driver._ensure_loadbalancer(self.ref_lb0)
+        self.driver._ensure_loadbalancer(self.ref_lb_fully_populated)
         mock_lb_create.assert_called_once_with(
-            self.driver._get_loadbalancer_request_info(self.ref_lb0),
+            self.driver._get_loadbalancer_request_info(
+                self.ref_lb_fully_populated),
         )
+        mock_listener_create.assert_called_once_with(
+            self.driver._get_listener_request_info(
+                self.ref_lb_fully_populated.listeners[0]),
+        )
+
+    @mock.patch.object(ovn_helper.OvnProviderHelper,
+                       '_update_status_to_octavia')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'listener_create')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'lb_create')
+    def test_ensure_loadbalancer_lb_no_listener_not_found(
+            self, mock_lb_create, mock_listener_create, mock_update_status):
+        self.mock_find_ovn_lbs_with_retry.side_effect = [
+            idlutils.RowNotFound]
+        self.ref_lb_fully_populated.listeners = data_models.Unset
+        self.driver._ensure_loadbalancer(self.ref_lb_fully_populated)
+        mock_lb_create.assert_called_once_with(
+            self.driver._get_loadbalancer_request_info(
+                self.ref_lb_fully_populated),
+        )
+        mock_listener_create.assert_not_called()
 
     @mock.patch.object(ovn_helper.OvnProviderHelper,
                        '_update_status_to_octavia')
     @mock.patch.object(ovn_helper.OvnProviderHelper,
                        '_get_current_operating_statuses')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'listener_sync')
     @mock.patch.object(ovn_helper.OvnProviderHelper, 'lb_sync')
     def test_ensure_loadbalancer_lb_found(
-            self, mock_lb_sync, mock_get_status, mock_update_status):
+            self, mock_lb_sync, mock_listener_sync, mock_get_status,
+            mock_update_status):
         self.mock_find_ovn_lbs_with_retry.return_value = [
             self.ovn_lb]
-        self.driver._ensure_loadbalancer(self.ref_lb0)
+        self.driver._ensure_loadbalancer(self.ref_lb_fully_populated)
         mock_lb_sync.assert_called_with(
-            self.driver._get_loadbalancer_request_info(self.ref_lb0),
+            self.driver._get_loadbalancer_request_info(
+                self.ref_lb_fully_populated),
+            self.ovn_lb
+        )
+        mock_listener_sync.assert_called_with(
+            self.driver._get_listener_request_info(
+                self.ref_lb_fully_populated.listeners[0]),
             self.ovn_lb
         )
         mock_get_status.assert_called_with(self.ovn_lb)
 
+    @mock.patch.object(ovn_helper.OvnProviderHelper,
+                       '_update_status_to_octavia')
+    @mock.patch.object(ovn_helper.OvnProviderHelper,
+                       '_get_current_operating_statuses')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'listener_sync')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'lb_sync')
+    def test_ensure_loadbalancer_lb_no_listener_found(
+            self, mock_lb_sync, mock_listener_sync, mock_get_status,
+            mock_update_status):
+        self.mock_find_ovn_lbs_with_retry.return_value = [
+            self.ovn_lb]
+        self.ref_lb_fully_populated.listeners = data_models.Unset
+        self.driver._ensure_loadbalancer(self.ref_lb_fully_populated)
+        mock_lb_sync.assert_called_with(
+            self.driver._get_loadbalancer_request_info(
+                self.ref_lb_fully_populated),
+            self.ovn_lb
+        )
+        mock_listener_sync.assert_not_called()
+
     @mock.patch.object(ovn_helper.OvnProviderHelper, 'get_octavia_lbs')
     @mock.patch.object(clients, 'get_octavia_client')
     def test_do_sync_no_loadbalancers(self, mock_get_octavia_client,
@@ -1276,16 +1326,40 @@ def test_do_sync_no_loadbalancers(self, mock_get_octavia_client,
             self.driver.do_sync(**lb_filters)
             mock_ensure_lb.assert_not_called()
 
+    @mock.patch.object(data_models.Listener, 'from_dict')
     @mock.patch.object(o_driver_lib.DriverLibrary, 'get_loadbalancer')
     @mock.patch.object(ovn_helper.OvnProviderHelper, 'get_octavia_lbs')
     @mock.patch.object(clients, 'get_octavia_client')
     def test_do_sync_with_loadbalancers(self,
                                         mock_get_octavia_client,
                                         mock_get_octavia_lbs,
-                                        mock_get_loadbalancer):
+                                        mock_get_loadbalancer,
+                                        mock_listener_from_dict):
         lb = mock.MagicMock(id=self.ref_lb_fully_sync_populated.name)
         mock_get_octavia_lbs.return_value = [lb]
         mock_get_loadbalancer.return_value = self.ref_lb_fully_sync_populated
+        mock_listener_from_dict.return_value = self.ref_listener
+        lb_filters = {}
+        with mock.patch.object(self.driver, '_ensure_loadbalancer') \
+                as mock_ensure_lb:
+            self.driver.do_sync(**lb_filters)
+            mock_ensure_lb.assert_any_call(
+                self.ref_lb_fully_sync_populated)
+
+    @mock.patch.object(data_models.Listener, 'from_dict')
+    @mock.patch.object(o_driver_lib.DriverLibrary, 'get_loadbalancer')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'get_octavia_lbs')
+    @mock.patch.object(clients, 'get_octavia_client')
+    def test_do_sync_with_loadbalancers_no_listener(
+            self,
+            mock_get_octavia_client,
+            mock_get_octavia_lbs,
+            mock_get_loadbalancer,
+            mock_listener_from_dict):
+        lb = mock.MagicMock(id=self.ref_lb_fully_sync_populated.name)
+        mock_get_octavia_lbs.return_value = [lb, lb]
+        mock_get_loadbalancer.return_value = self.ref_lb_fully_sync_populated
+        self.ref_lb_fully_sync_populated.listeners = data_models.Unset
         lb_filters = {}
         with mock.patch.object(self.driver, '_ensure_loadbalancer') \
                 as mock_ensure_lb:

ovn_octavia_provider/tests/unit/test_helper.py

@@ -1886,6 +1886,67 @@ def test_listener_delete_ovn_lb_empty_ovn_lb_not_found(self, lb_empty):
             mock.call('Load_Balancer', self.ovn_lb.uuid,
                       ('vips', {}))])
 
+    @mock.patch.object(ovn_helper.OvnProviderHelper, '_refresh_lb_vips')
+    def test_listener_sync_listener_same_in_externals_ids(self, refresh_vips):
+        self.listener['admin_state_up'] = True
+        listener_key = 'listener_%s' % self.listener_id
+        self.ovn_lb.external_ids[listener_key] = f"80:pool_{self.pool_id}"
+        self.helper.listener_sync(self.listener, self.ovn_lb)
+        refresh_vips.assert_called_once_with(
+            self.ovn_lb, self.ovn_lb.external_ids, is_sync=True)
+        self.helper.ovn_nbdb_api.db_set.assert_not_called()
+
+    @mock.patch.object(ovn_helper.OvnProviderHelper, '_refresh_lb_vips')
+    def test_listener_sync_listener_diff_in_externals_ids(self, refresh_vips):
+        self.listener['admin_state_up'] = True
+        listener_key = 'listener_%s' % self.listener_id
+        external_ids = copy.deepcopy(self.ovn_lb.external_ids)
+        self.ovn_lb.external_ids[listener_key] = ''
+        self.helper.listener_sync(self.listener, self.ovn_lb)
+        refresh_vips.assert_called_once_with(
+            self.ovn_lb, external_ids, is_sync=True)
+        expected_calls = [
+            mock.call('Load_Balancer', self.ovn_lb.uuid, ('external_ids', {
+                f"listener_{self.listener_id}": f"80:pool_{self.pool_id}"}))
+        ]
+        self.helper.ovn_nbdb_api.db_set.assert_has_calls(
+            expected_calls)
+
+    @mock.patch.object(ovn_helper.OvnProviderHelper, '_execute_commands')
+    def test_listener_sync_exception(self, execute_commands):
+        execute_commands.side_effect = [RuntimeError('a fail')]
+        self.ovn_lb.external_ids.pop('listener_%s' % self.listener_id)
+        self.listener['admin_state_up'] = True
+        with mock.patch.object(ovn_helper, 'LOG') as m_l:
+            self.assertIsNone(self.helper.listener_sync(
+                self.listener, self.ovn_lb))
+            m_l.exception.assert_called_once_with(
+                'Failed to execute commands for listener sync: a fail')
+
+    @mock.patch.object(ovn_helper.OvnProviderHelper, '_refresh_lb_vips')
+    def test_listener_sync_refresh_vips_exception(self, refresh_lb_vips):
+        refresh_lb_vips.side_effect = [RuntimeError('a fail')]
+        self.ovn_lb.external_ids.pop('listener_%s' % self.listener_id)
+        self.listener['admin_state_up'] = True
+        with mock.patch.object(ovn_helper, 'LOG') as m_l:
+            self.assertIsNone(self.helper.listener_sync(
+                self.listener, self.ovn_lb))
+            m_l.exception.assert_called_once_with(
+                'Failed to refresh LB VIPs: a fail')
+
+    @mock.patch.object(ovn_helper.OvnProviderHelper, '_refresh_lb_vips')
+    def test_listener_sync_listener_not_in_externals_ids(
+            self, refresh_vips):
+        self.ovn_lb.external_ids.pop('listener_%s' % self.listener_id)
+        self.listener['admin_state_up'] = True
+        self.helper.listener_sync(self.listener, self.ovn_lb)
+        expected_calls = [
+            mock.call('Load_Balancer', self.ovn_lb.uuid, ('external_ids', {
+                f"listener_{self.listener_id}": f"80:pool_{self.pool_id}"}))
+        ]
+        self.helper.ovn_nbdb_api.db_set.assert_has_calls(
+            expected_calls)
+
     def test_pool_create(self):
         status = self.helper.pool_create(self.pool)
         self.assertEqual(status['loadbalancers'][0]['provisioning_status'],