Add Member sync logic

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

Author: fernandoroyosanchez

ovn_octavia_provider/driver.py

@@ -139,6 +139,42 @@ def _get_pool_request_info(self, pool):
 
         return request_info
 
+    def _get_member_request_info(self, member, create=True):
+        # Validate monitoring options if present
+        admin_state_up = None
+        if create:
+            self._check_member_monitor_options(member)
+            if self._ip_version_differs(member):
+                raise ovn_exc.IPVersionsMixingNotSupportedError()
+            admin_state_up = member.admin_state_up
+        subnet_id = member.subnet_id
+        if (isinstance(subnet_id, o_datamodels.UnsetType) or not subnet_id):
+            subnet_id, subnet_cidr = self._ovn_helper._get_subnet_from_pool(
+                member.pool_id)
+            if not (subnet_id and
+                    self._ovn_helper._check_ip_in_subnet(member.address,
+                                                         subnet_cidr)):
+                msg = _('Subnet is required, or Loadbalancer associated with '
+                        'Pool must have a subnet, for Member creation '
+                        'with OVN Provider Driver if it is not the same as '
+                        'LB VIP subnet')
+                raise driver_exceptions.UnsupportedOptionError(
+                    user_fault_string=msg,
+                    operator_fault_string=msg)
+
+        if isinstance(admin_state_up, o_datamodels.UnsetType):
+            admin_state_up = True
+        request_info = {'id': member.member_id,
+                        'address': member.address,
+                        'protocol_port': member.protocol_port,
+                        'pool_id': member.pool_id,
+                        'subnet_id': subnet_id}
+
+        if admin_state_up and create:
+            request_info['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(
@@ -648,7 +684,16 @@ def _ensure_loadbalancer(self, loadbalancer):
                     status_pool = self._ovn_helper.pool_create(
                         self._get_pool_request_info(pool))
                     status[constants.POOLS].append(status_pool)
+                    for member in pool.members:
+                        status[constants.MEMBERS] = []
+                        if not member.subnet_id:
+                            member.subnet_id = loadbalancer.vip_subnet_id
+                        status_member = self._ovn_helper.member_create(
+                            self._get_member_request_info(member))
+                        status[constants.MEMBERS].append(status_member)
+
             self._ovn_helper._update_status_to_octavia(status)
+
         else:
             # Load Balancer found, check LB and listener/pool/member/hms
             # related
@@ -667,8 +712,53 @@ def _ensure_loadbalancer(self, loadbalancer):
                 # 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)
+                        pool_info = self._get_pool_request_info(pool)
+                        self._ovn_helper.pool_sync(pool_info, ovn_lb)
+                        ovn_pool_key = self._ovn_helper._get_pool_key(
+                            pool_info[constants.ID],
+                            is_enabled=pool_info[constants.ADMIN_STATE_UP])
+                        member_ids = []
+                        if not isinstance(pool.members,
+                                          o_datamodels.UnsetType):
+                            for member in pool.members:
+                                if not member.subnet_id:
+                                    member.subnet_id = (
+                                        loadbalancer.vip_subnet_id
+                                    )
+                                self._ovn_helper.member_sync(
+                                    self._get_member_request_info(member),
+                                    ovn_lb,
+                                    ovn_pool_key)
+                                member_ids.append(member.member_id)
+
+                            for ovn_mb_info in \
+                                self._ovn_helper._get_members_in_ovn_lb(
+                                    ovn_lb, ovn_pool_key):
+                                # If member ID not in pool member list,
+                                # delete it.
+                                if ovn_mb_info[3] not in member_ids:
+                                    LOG.debug(
+                                        "Start deleting extra member "
+                                        f"{ovn_mb_info[3]} from pool "
+                                        "{pool_info[constants.ID]} in OVN."
+                                    )
+                                    mb_delete_info = {
+                                        'id': ovn_mb_info[3],
+                                        'subnet_id': ovn_mb_info[2],
+                                    }
+                                    self._ovn_helper.member_delete(
+                                        mb_delete_info)
+
+                                    mb_delete_dvr_info = {
+                                        'id': ovn_mb_info[3],
+                                        'address': ovn_mb_info[0],
+                                        'pool_id': pool_info[constants.ID],
+                                        'subnet_id': ovn_mb_info[2],
+                                        'action':
+                                            ovn_const.REQ_INFO_MEMBER_DELETED
+                                    }
+                                    self._ovn_helper.handle_member_dvr(
+                                        mb_delete_dvr_info)
                 status = self._ovn_helper._get_current_operating_statuses(
                     ovn_lb)
                 self._ovn_helper._update_status_to_octavia(status)
@@ -691,9 +781,20 @@ def do_sync(self, **lb_filters):
             ] 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
-
+            provider_pools = []
+            for pool in pools:
+                provider_pool = o_datamodels.Pool.from_dict(pool)
+                # format member provider
+                members = provider_pool.members
+                if not isinstance(members, o_datamodels.UnsetType) and members:
+                    provider_pool.members = [
+                        o_datamodels.Member.from_dict(m)
+                        for m in members]
+                else:
+                    provider_pool.members = o_datamodels.Unset
+                provider_pools.append(provider_pool)
+
+            provider_lb.pools = (
+                provider_pools if provider_pools else o_datamodels.Unset
+            )
             self._ensure_loadbalancer(provider_lb)

ovn_octavia_provider/helper.py

@@ -722,6 +722,42 @@ def _add_persistence_timeout_command(self, commands, ovn_lb,
                                          ('options', options))
             )
 
+    def _update_pool_data(self, member, pool_key, external_ids):
+        """Update pool data with member information."""
+        pool_data = None
+        existing_members = external_ids.get(pool_key, "")
+        member_info = self._get_member_info(member)
+
+        if existing_members:
+            members = existing_members.split(",")
+            if member_info not in members:
+                members.append(member_info)
+                pool_data = {pool_key: ",".join(members)}
+        else:
+            pool_data = {pool_key: member_info}
+
+        return pool_data
+
+    def _add_pool_data_command(self, commands, ovn_lb, pool_data):
+        """Add command to update pool data in LoadBalancer."""
+        if pool_data:
+            commands.append(
+                self.ovn_nbdb_api.db_set('Load_Balancer', ovn_lb.uuid,
+                                         ('external_ids', pool_data))
+            )
+
+    def _get_related_lr(self, member):
+        """Retrieve the logical router related to the member's subnet."""
+        neutron_client = clients.get_neutron_client()
+        try:
+            subnet = neutron_client.get_subnet(member[constants.SUBNET_ID])
+            ls_name = utils.ovn_name(subnet.network_id)
+            ovn_ls = self.ovn_nbdb_api.ls_get(ls_name).execute(
+                check_error=True)
+            return self._find_lr_of_ls(ovn_ls, subnet.gateway_ip)
+        except (idlutils.RowNotFound, openstack.exceptions.ResourceNotFound):
+            return None
+
     def _lb_status(self, loadbalancer, provisioning_status, operating_status):
         """Return status for the LoadBalancer."""
         return {
@@ -2507,6 +2543,73 @@ def _update_external_ids_member_status(self, ovn_lb, member, status=None,
                           " %s delete: %s status: %s", str(member),
                           str(delete), str(status))
 
+    def _get_members_in_ovn_lb(self, ovn_lb, pool_key):
+        existing_members = ovn_lb.external_ids.get(pool_key, None)
+        if existing_members:
+            existing_members = existing_members.split(",")
+            return [
+                self._extract_member_info(
+                    member)[0] for member in existing_members
+            ]
+        else:
+            return []
+
+    def member_sync(self, member, ovn_lb, pool_key):
+        """Sync Member object with an OVN LoadBalancer
+
+        The method performs the following steps:
+        1. Update pool key with member info on OVN Loadbalancer external_ids
+        if needed
+        2. Update OVN LoadBalancer vips
+        3. Update references on LS or LR from the member if needed
+        4. Update OVN Loadbalancer member_status info on external_ids
+
+        :param member: The source member object from Octavia DB
+        :param ovn_lb: The OVN LoadBalancer object that needs to be sync
+        :param pool_key: The pool_key where member is associated
+        """
+        external_ids = copy.deepcopy(ovn_lb.external_ids)
+        pool_data = self._update_pool_data(member, pool_key, external_ids)
+
+        commands = []
+        if pool_data:
+            self._add_pool_data_command(commands, ovn_lb, pool_data)
+            external_ids[pool_key] = pool_data[pool_key]
+
+        try:
+            if member.get(constants.ADMIN_STATE_UP, False):
+                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
+
+        self._update_lb_to_ls_association(
+            ovn_lb, subnet_id=member[constants.SUBNET_ID], associate=True,
+            update_ls_ref=True, is_sync=True)
+
+        # Make sure that all logical switches related to logical router
+        # are associated with the load balancer. This is needed to handle
+        # potential race that happens when lrp and lb are created at the
+        # same time.
+        ovn_lr = self._get_related_lr(member)
+
+        if ovn_lr:
+            self._sync_lb_to_lr_association(ovn_lb, ovn_lr)
+
+        # TODO(froyo): Check if originally status in Octavia is ERROR if
+        # we receive that info from the object
+        self._update_external_ids_member_status(
+            ovn_lb,
+            member[constants.ID],
+            constants.NO_MONITOR)
+
     def _add_member(self, member, ovn_lb, pool_key):
         external_ids = copy.deepcopy(ovn_lb.external_ids)
         existing_members = external_ids[pool_key]