Merge "Add Member sync logic"

Author: Zuul

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

@@ -721,6 +721,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 {
@@ -2506,6 +2542,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]