Merge "Add Listener sync logic"

Author: Zuul

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

@@ -618,6 +618,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 {
@@ -1867,6 +1911,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'],