Add Pool sync logic

This patch adds the logic to sync a Pool 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: I7293f4d29683d143fdc4ad6778d090a476c36e7c

Author: fernandoroyosanchez

ovn_octavia_provider/driver.py

@@ -119,6 +119,26 @@ def _get_listener_request_info(self, listener):
                         'admin_state_up': admin_state_up}
         return request_info
 
+    def _get_pool_request_info(self, pool):
+        self._check_for_supported_protocols(pool.protocol)
+        self._check_for_supported_algorithms(pool.lb_algorithm)
+        admin_state_up = pool.admin_state_up
+        if isinstance(admin_state_up, o_datamodels.UnsetType):
+            admin_state_up = True
+        request_info = {'id': pool.pool_id,
+                        'loadbalancer_id': pool.loadbalancer_id,
+                        'protocol': pool.protocol,
+                        'lb_algorithm': pool.lb_algorithm,
+                        'listener_id': pool.listener_id,
+                        'admin_state_up': admin_state_up}
+        if not isinstance(
+                pool.session_persistence, o_datamodels.UnsetType):
+            self._check_for_supported_session_persistence(
+                pool.session_persistence)
+            request_info['session_persistence'] = pool.session_persistence
+
+        return request_info
+
     def loadbalancer_create(self, loadbalancer):
         request = {'type': ovn_const.REQ_TYPE_LB_CREATE,
                    'info': self._get_loadbalancer_request_info(
@@ -612,7 +632,7 @@ 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 and listener only
+            # TODO(froyo): By now just syncing LB, listener and pool only
             status = self._ovn_helper.lb_create(
                 self._get_loadbalancer_request_info(loadbalancer))
 
@@ -622,6 +642,12 @@ def _ensure_loadbalancer(self, loadbalancer):
                     status_listener = self._ovn_helper.listener_create(
                         self._get_listener_request_info(listener))
                     status[constants.LISTENERS].append(status_listener)
+            if not isinstance(loadbalancer.pools, o_datamodels.UnsetType):
+                status[constants.POOLS] = []
+                for pool in loadbalancer.pools:
+                    status_pool = self._ovn_helper.pool_create(
+                        self._get_pool_request_info(pool))
+                    status[constants.POOLS].append(status_pool)
             self._ovn_helper._update_status_to_octavia(status)
         else:
             # Load Balancer found, check LB and listener/pool/member/hms
@@ -638,6 +664,11 @@ def _ensure_loadbalancer(self, loadbalancer):
                     for listener in loadbalancer.listeners:
                         self._ovn_helper.listener_sync(
                             self._get_listener_request_info(listener), ovn_lb)
+                # Pool
+                if not isinstance(loadbalancer.pools, o_datamodels.UnsetType):
+                    for pool in loadbalancer.pools:
+                        self._ovn_helper.pool_sync(
+                            self._get_pool_request_info(pool), ovn_lb)
                 status = self._ovn_helper._get_current_operating_statuses(
                     ovn_lb)
                 self._ovn_helper._update_status_to_octavia(status)
@@ -659,4 +690,10 @@ def do_sync(self, **lb_filters):
                 for listener in listeners
             ] if listeners else o_datamodels.Unset
 
+            pools = provider_lb.pools or []
+            provider_lb.pools = [
+                o_datamodels.Pool.from_dict(pool)
+                for pool in pools
+            ] if pools else o_datamodels.Unset
+
             self._ensure_loadbalancer(provider_lb)

ovn_octavia_provider/helper.py

@@ -663,6 +663,65 @@ def _update_protocol_if_needed(self, listener, ovn_lb, commands):
                 )
             )
 
+    def _prepare_external_ids(self, pool, ovn_lb):
+        """Prepare the updated external_ids for the LoadBalancer."""
+        external_ids = copy.deepcopy(ovn_lb.external_ids)
+        pool_key = self._get_pool_key(
+            pool[constants.ID],
+            is_enabled=pool[constants.ADMIN_STATE_UP])
+        external_ids[pool_key] = ''
+
+        if pool[constants.LISTENER_ID]:
+            self._update_listener_association(
+                pool, ovn_lb, external_ids, pool_key)
+
+        return external_ids
+
+    def _update_listener_association(self, pool, ovn_lb, external_ids,
+                                     pool_key):
+        """Update the listener association in external_ids."""
+        listener_key = self._get_listener_key(pool[constants.LISTENER_ID])
+        if listener_key in ovn_lb.external_ids:
+            pool_key_enable = self._get_pool_key(pool[constants.ID],
+                                                 is_enabled=True)
+            pool_key_disable = self._get_pool_key(pool[constants.ID],
+                                                  is_enabled=False)
+
+            if pool[constants.ID] in external_ids[listener_key]:
+                # Remove existing pool keys before adding the updated key
+                external_ids[listener_key] = (
+                    external_ids[listener_key]
+                    .replace(pool_key_disable, '')
+                    .replace(pool_key_enable, '')
+                )
+
+            external_ids[listener_key] += str(pool_key)
+
+    def _extract_persistence_timeout(self, pool):
+        """Extract persistence timeout value from the pool, if available."""
+        if pool.get(constants.SESSION_PERSISTENCE):
+            return pool[constants.SESSION_PERSISTENCE].get(
+                constants.PERSISTENCE_TIMEOUT, '360')
+        return None
+
+    def _add_external_ids_command(self, commands, ovn_lb, external_ids):
+        """Add a command to update the external_ids of the LoadBalancer."""
+        commands.append(
+            self.ovn_nbdb_api.db_set('Load_Balancer', ovn_lb.uuid,
+                                     ('external_ids', external_ids))
+        )
+
+    def _add_persistence_timeout_command(self, commands, ovn_lb,
+                                         persistence_timeout):
+        """Add command to update persistence timeout in LoadBalancer."""
+        options = copy.deepcopy(ovn_lb.options)
+        options[ovn_const.AFFINITY_TIMEOUT] = str(persistence_timeout)
+        if ovn_lb.options != options:
+            commands.append(
+                self.ovn_nbdb_api.db_set('Load_Balancer', ovn_lb.uuid,
+                                         ('options', options))
+            )
+
     def _lb_status(self, loadbalancer, provisioning_status, operating_status):
         """Return status for the LoadBalancer."""
         return {
@@ -2115,7 +2174,11 @@ def pool_create(self, pool):
         external_ids[pool_key] = ''
         if pool[constants.LISTENER_ID]:
             listener_key = self._get_listener_key(pool[constants.LISTENER_ID])
-            if listener_key in ovn_lb.external_ids:
+            # NOTE(froyo): checking is not already when ovn-db-sync-tool is
+            # triggered, because listener_create could be added already if
+            # pool is considered as default one
+            if listener_key in ovn_lb.external_ids and \
+                    str(pool_key) not in external_ids[listener_key]:
                 external_ids[listener_key] = str(
                     external_ids[listener_key]) + str(pool_key)
         persistence_timeout = None
@@ -2174,6 +2237,29 @@ def pool_create(self, pool):
 
         return status
 
+    def pool_sync(self, pool, ovn_lb):
+        """Sync Pool object with an OVN LoadBalancer
+
+        The method performs the following steps:
+        1. Update pool key on OVN Loadbalancer external_ids if needed
+        2. Update OVN LoadBalancer options from Pool info
+
+        :param pool: The source pool object from Octavia DB
+        :param ovn_lb: The OVN LoadBalancer object that needs to be sync
+        """
+        external_ids = self._prepare_external_ids(pool, ovn_lb)
+        persistence_timeout = self._extract_persistence_timeout(pool)
+
+        try:
+            commands = []
+            self._add_external_ids_command(commands, ovn_lb, external_ids)
+            if persistence_timeout:
+                self._add_persistence_timeout_command(commands, ovn_lb,
+                                                      persistence_timeout)
+            self._execute_commands(commands)
+        except Exception as e:
+            LOG.exception(f"Failed to execute commands for pool sync: {e}")
+
     def pool_delete(self, pool):
         status = {
             constants.POOLS: [

ovn_octavia_provider/tests/unit/test_driver.py

@@ -1237,10 +1237,12 @@ def test_health_monitor_delete(self):
 
     @mock.patch.object(ovn_helper.OvnProviderHelper,
                        '_update_status_to_octavia')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'pool_create')
     @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_listener_create, mock_update_status):
+            self, mock_lb_create, mock_listener_create, mock_pool_create,
+            mock_update_status):
         self.mock_find_ovn_lbs_with_retry.side_effect = [
             idlutils.RowNotFound]
         self.driver._ensure_loadbalancer(self.ref_lb_fully_populated)
@@ -1252,16 +1254,21 @@ def test_ensure_loadbalancer_lb_not_found(
             self.driver._get_listener_request_info(
                 self.ref_lb_fully_populated.listeners[0]),
         )
+        mock_pool_create.assert_called_once_with(
+            self.driver._get_pool_request_info(
+                self.ref_lb_fully_populated.pools[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(
+    def test_ensure_loadbalancer_lb_no_listener_no_pool_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.ref_lb_fully_populated.pools = 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(
@@ -1273,11 +1280,12 @@ def test_ensure_loadbalancer_lb_no_listener_not_found(
                        '_update_status_to_octavia')
     @mock.patch.object(ovn_helper.OvnProviderHelper,
                        '_get_current_operating_statuses')
+    @mock.patch.object(ovn_helper.OvnProviderHelper, 'pool_sync')
     @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_listener_sync, mock_get_status,
-            mock_update_status):
+            self, mock_lb_sync, mock_listener_sync, mock_pool_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_lb_fully_populated)
@@ -1327,17 +1335,20 @@ def test_do_sync_no_loadbalancers(self, mock_get_octavia_client,
             mock_ensure_lb.assert_not_called()
 
     @mock.patch.object(data_models.Listener, 'from_dict')
+    @mock.patch.object(data_models.Pool, '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_pool_from_dict,
                                         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_pool_from_dict.return_value = self.ref_pool_with_hm
         mock_listener_from_dict.return_value = self.ref_listener
         lb_filters = {}
         with mock.patch.object(self.driver, '_ensure_loadbalancer') \
@@ -1350,7 +1361,7 @@ def test_do_sync_with_loadbalancers(self,
     @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(
+    def test_do_sync_with_loadbalancers_no_listener_no_pool(
             self,
             mock_get_octavia_client,
             mock_get_octavia_lbs,
@@ -1360,6 +1371,7 @@ def test_do_sync_with_loadbalancers_no_listener(
         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
+        self.ref_lb_fully_sync_populated.pools = 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

@@ -2209,6 +2209,63 @@ def test_pool_delete_ovn_lb_empty_lb_not_empty(self, lb_empty):
         self.helper.ovn_nbdb_api.lb_del.assert_called_once_with(
             self.ovn_lb.uuid)
 
+    def test_pool_sync_exception(self):
+        self.helper.ovn_nbdb_api.db_set.side_effect = [
+            RuntimeError("ERROR_MSG"), RuntimeError("ERROR_MSG")]
+        with mock.patch.object(ovn_helper, 'LOG') as m_l:
+            self.assertIsNone(self.helper.pool_sync(self.pool, self.ovn_lb))
+            m_l.exception.assert_called_once_with(
+                'Failed to execute commands for pool sync: ERROR_MSG')
+
+    def test_pool_sync(self):
+        self.helper.pool_sync(self.pool, self.ovn_lb)
+        listener_key_value = (f"80:pool_{self.pool_id}:"
+                              f"{ovn_const.DISABLED_RESOURCE_SUFFIX}")
+        expected_calls = [mock.call(
+            'Load_Balancer', self.ovn_lb.uuid,
+            ('external_ids', {
+                ovn_const.LB_EXT_IDS_VIP_KEY: mock.ANY,
+                ovn_const.LB_EXT_IDS_VIP_FIP_KEY: '123.123.123.123',
+                ovn_const.LB_EXT_IDS_VIP_PORT_ID_KEY: mock.ANY,
+                'enabled': True,
+                f"pool_{self.pool_id}": mock.ANY,
+                f"listener_{self.listener_id}": listener_key_value,
+                ovn_const.OVN_MEMBER_STATUS_KEY: '{"%s": "%s"}'
+                % (self.member_id, constants.NO_MONITOR),
+                f"pool_{self.pool_id}:{ovn_const.DISABLED_RESOURCE_SUFFIX}": ''
+            }))
+        ]
+        self.helper.ovn_nbdb_api.db_set.assert_has_calls(
+            expected_calls)
+
+    def test_pool_sync_press_key(self):
+        self.pool[constants.SESSION_PERSISTENCE] = {
+            constants.PERSISTENCE_TIMEOUT: '360'
+        }
+        self.ovn_lb.options = {'a': 1}
+        self.helper.pool_sync(self.pool, self.ovn_lb)
+        listener_key_value = (f"80:pool_{self.pool_id}:"
+                              f"{ovn_const.DISABLED_RESOURCE_SUFFIX}")
+        expected_calls = [mock.call(
+            'Load_Balancer', self.ovn_lb.uuid,
+            ('external_ids', {
+                ovn_const.LB_EXT_IDS_VIP_KEY: mock.ANY,
+                ovn_const.LB_EXT_IDS_VIP_FIP_KEY: '123.123.123.123',
+                ovn_const.LB_EXT_IDS_VIP_PORT_ID_KEY: mock.ANY,
+                'enabled': True,
+                f"pool_{self.pool_id}": mock.ANY,
+                f"listener_{self.listener_id}": listener_key_value,
+                ovn_const.OVN_MEMBER_STATUS_KEY: '{"%s": "%s"}'
+                % (self.member_id, constants.NO_MONITOR),
+                f"pool_{self.pool_id}:{ovn_const.DISABLED_RESOURCE_SUFFIX}": ''
+            })),
+            mock.call(
+                'Load_Balancer', self.ovn_lb.uuid,
+                ('options', {'a': 1, 'affinity_timeout': '360'}))
+        ]
+        self.helper.ovn_nbdb_api.db_set.assert_has_calls(
+            expected_calls)
+
     @mock.patch('ovn_octavia_provider.common.clients.get_neutron_client')
     def test_member_create_disabled(self, net_cli):
         net_cli.return_value.show_subnet.side_effect = [idlutils.RowNotFound]