openstack
/
octavia
Code Issues Proposed changes

Handle custom security groups in controller 

Change-Id: I6f525ebd294308a411e384d3628ba8876d429d02
This commit is contained in:
Gregory Thiemonge 2024-04-08 02:37:44 -04:00
parent 9cec5d651f
commit 8f46569fc1
5 changed files with 91 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
octavia/common/constants.py
View File

@ -424,10 +424,12 @@ REQ_READ_TIMEOUT = 'req_read_timeout'
REQUEST_ERRORS = 'request_errors'
REQUEST_ID = 'request_id'
ROLE = 'role'
SECURITY_GROUP_IDS = 'security_group_ids'
SECURITY_GROUPS = 'security_groups'
SECURITY_GROUP_RULES = 'security_group_rules'
SERVER_GROUP_ID = 'server_group_id'
SERVER_PEM = 'server_pem'
SG_ID = 'sg_id'
SNI_CONTAINER_DATA = 'sni_container_data'
SNI_CONTAINERS = 'sni_containers'
SOFT_ANTI_AFFINITY = 'soft-anti-affinity'

5
octavia/controller/worker/v2/flows/load_balancer_flows.py
View File

@ -372,6 +372,11 @@ class LoadBalancerFlows:
update_LB_flow = linear_flow.Flow(constants.UPDATE_LOADBALANCER_FLOW)
update_LB_flow.add(lifecycle_tasks.LoadBalancerToErrorOnRevertTask(
requires=constants.LOADBALANCER))
update_LB_flow.add(network_tasks.UpdateVIPSecurityGroup(
requires=constants.LOADBALANCER_ID,
provides=constants.VIP_SG_ID))
update_LB_flow.add(network_tasks.UpdateAmphoraSecurityGroup(
requires=constants.LOADBALANCER_ID))
update_LB_flow.add(network_tasks.ApplyQos(
requires=(constants.LOADBALANCER, constants.UPDATE_DICT)))
update_LB_flow.add(amphora_driver_tasks.ListenersUpdate(

30
octavia/controller/worker/v2/tasks/network_tasks.py
View File

@ -497,6 +497,20 @@ class UpdateVIPSecurityGroup(BaseNetworkTask):
return sg_id
class UpdateAmphoraSecurityGroup(BaseNetworkTask):
"""Task to update SGs for an Amphora."""
def execute(self, loadbalancer_id):
session = db_apis.get_session()
with session.begin():
db_lb = self.loadbalancer_repo.get(
session, id=loadbalancer_id)
for amp in db_lb.amphorae:
self.network_driver.update_aap_port_sg(db_lb,
amp,
db_lb.vip)
class GetSubnetFromVIP(BaseNetworkTask):
"""Task to plumb a VIP."""
@ -607,7 +621,10 @@ class AllocateVIP(BaseNetworkTask):
LOG.debug('Allocated an additional VIP: subnet=%(subnet)s '
'ip_address=%(ip)s', {'subnet': add_vip.subnet_id,
'ip': add_vip.ip_address})
return (vip.to_dict(),
vip_dict = vip.to_dict()
for vip_sg in vip.sgs:
vip_dict["sgs"].append(vip_sg.sg_id)
return (vip_dict,
[additional_vip.to_dict()
for additional_vip in additional_vips])
@ -963,16 +980,21 @@ class CreateVIPBasePort(BaseNetworkTask):
def execute(self, vip, vip_sg_id, amphora_id, additional_vips):
port_name = constants.AMP_BASE_PORT_PREFIX + amphora_id
fixed_ips = [{constants.SUBNET_ID: vip[constants.SUBNET_ID]}]
sg_id = []
sg_ids = []
# To remove some confusion:
# - vip_sg_id is the ID of the SG created by Octavia for the LB.
# - vip['sgs'] are the ID of the SGs provided by the user for the LB.
if vip_sg_id:
sg_id = [vip_sg_id]
sg_ids = [vip_sg_id]
if vip["sgs"]:
sg_ids += vip["sgs"]
secondary_ips = [vip[constants.IP_ADDRESS]]
for add_vip in additional_vips:
secondary_ips.append(add_vip[constants.IP_ADDRESS])
port = self.network_driver.create_port(
vip[constants.NETWORK_ID], name=port_name, fixed_ips=fixed_ips,
secondary_ips=secondary_ips,
security_group_ids=sg_id,
security_group_ids=sg_ids,
qos_policy_id=vip[constants.QOS_POLICY_ID])
LOG.info('Created port %s with ID %s for amphora %s',
port_name, port.id, amphora_id)

73
octavia/network/drivers/neutron/allowed_address_pairs.py
View File

@ -80,7 +80,7 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
return interface
return None
def _plug_amphora_vip(self, amphora, subnet):
def _plug_amphora_vip(self, amphora, subnet, vip: data_models.Vip):
# We need a vip port owned by Octavia for Act/Stby and failover
try:
port = {
@ -90,6 +90,11 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
constants.ADMIN_STATE_UP: True,
constants.DEVICE_OWNER: constants.OCTAVIA_OWNER,
}
if vip.sgs:
port[constants.SECURITY_GROUP_IDS] = [
vip_sg.sg_id
for vip_sg in vip.sgs]
new_port = self.network_proxy.create_port(**port)
new_port = utils.convert_port_to_model(new_port)
@ -149,7 +154,12 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
net = ipaddress.ip_network(cidr)
return 'IPv6' if net.version == 6 else 'IPv4'
def _update_security_group_rules(self, load_balancer, sec_grp_id):
def _update_security_group_rules(self,
load_balancer: data_models.LoadBalancer,
sec_grp_id):
# Skip adding listener rules if sgs is not None or not empty
skip_listener_rules = load_balancer.vip.sgs
rules = tuple(self.network_proxy.security_group_rules(
security_group_id=sec_grp_id))
@ -160,19 +170,20 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
constants.DELETED]):
continue
protocol = constants.PROTOCOL_TCP.lower()
if listener.protocol == constants.PROTOCOL_UDP:
protocol = constants.PROTOCOL_UDP.lower()
elif listener.protocol == lib_consts.PROTOCOL_SCTP:
protocol = lib_consts.PROTOCOL_SCTP.lower()
if not skip_listener_rules:
protocol = constants.PROTOCOL_TCP.lower()
if listener.protocol == constants.PROTOCOL_UDP:
protocol = constants.PROTOCOL_UDP.lower()
elif listener.protocol == lib_consts.PROTOCOL_SCTP:
protocol = lib_consts.PROTOCOL_SCTP.lower()
if listener.allowed_cidrs:
for ac in listener.allowed_cidrs:
port = (listener.protocol_port, protocol, ac.cidr)
if listener.allowed_cidrs:
for ac in listener.allowed_cidrs:
port = (listener.protocol_port, protocol, ac.cidr)
updated_ports.append(port)
else:
port = (listener.protocol_port, protocol, None)
updated_ports.append(port)
else:
port = (listener.protocol_port, protocol, None)
updated_ports.append(port)
listener_peer_ports.append(listener.peer_port)
@ -262,12 +273,15 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
raise base.PlugVIPException(str(e))
def _add_vip_security_group_to_port(self, load_balancer_id, port_id,
sec_grp_id=None):
sec_grp_id = (sec_grp_id or
self._get_lb_security_group(load_balancer_id).get(
constants.ID))
sec_grp_id=None, vip_sgs=None):
sec_grp_ids = [sec_grp_id or
self._get_lb_security_group(load_balancer_id).get(
constants.ID)]
if vip_sgs:
sec_grp_ids += [vip_sg.sg_id
for vip_sg in vip_sgs]
try:
self._add_security_group_to_port(sec_grp_id, port_id)
self._update_security_groups(sec_grp_ids, port_id)
except base.PortNotFound:
raise
except base.NetworkException as e:
@ -409,15 +423,26 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
self._update_security_group_rules(load_balancer,
sec_grp.get(constants.ID))
self._add_vip_security_group_to_port(load_balancer.id, vip.port_id,
sec_grp.get(constants.ID))
sec_grp.get(constants.ID),
vip_sgs=vip.sgs)
return sec_grp.get(constants.ID)
return None
def update_aap_port_sg(self, load_balancer, amphora, vip):
if self.sec_grp_enabled:
sec_grp = self._get_lb_security_group(load_balancer.id)
if not sec_grp:
return
self._add_vip_security_group_to_port(load_balancer.id,
amphora.vrrp_port_id,
sec_grp.get(constants.ID),
vip_sgs=vip.sgs)
def plug_aap_port(self, load_balancer, vip, amphora, subnet):
interface = self._get_plugged_interface(
amphora.compute_id, subnet.network_id, amphora.lb_network_ip)
if not interface:
interface = self._plug_amphora_vip(amphora, subnet)
interface = self._plug_amphora_vip(amphora, subnet, vip)
aap_address_list = [vip.ip_address]
for add_vip in load_balancer.additional_vips:
@ -426,7 +451,8 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
if self.sec_grp_enabled:
self._add_vip_security_group_to_port(load_balancer.id,
interface.port_id)
interface.port_id,
vip_sgs=vip.sgs)
vrrp_ip = None
for fixed_ip in interface.fixed_ips:
is_correct_subnet = fixed_ip.subnet_id == subnet.id
@ -558,6 +584,11 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
constants.DEVICE_OWNER: constants.OCTAVIA_OWNER,
project_id_key: load_balancer.project_id}
if load_balancer.vip.sgs:
port[constants.SECURITY_GROUP_IDS] = [
vip_sg.sg_id
for vip_sg in load_balancer.vip.sgs]
if fixed_ips:
port[constants.FIXED_IPS] = fixed_ips
try:

10
octavia/network/drivers/neutron/base.py
View File

@ -83,14 +83,16 @@ class BaseNeutronDriver(base.AbstractNetworkDriver):
port_id=port.id,
load_balancer=load_balancer,
load_balancer_id=load_balancer.id,
octavia_owned=octavia_owned)
octavia_owned=octavia_owned,
sgs=load_balancer.vip.sgs)
else:
primary_vip = data_models.Vip(ip_address=None, subnet_id=None,
network_id=port.network_id,
port_id=port.id,
load_balancer=load_balancer,
load_balancer_id=load_balancer.id,
octavia_owned=octavia_owned)
octavia_owned=octavia_owned,
sgs=load_balancer.vip.sgs)
additional_vips = [
data_models.AdditionalVip(
ip_address=add_fixed_ip.ip_address,
@ -123,11 +125,11 @@ class BaseNeutronDriver(base.AbstractNetworkDriver):
self.network_proxy.update_port(port_id,
allowed_address_pairs=aap)
def _add_security_group_to_port(self, sec_grp_id, port_id):
def _update_security_groups(self, sec_grp_ids, port_id):
# Note: Neutron accepts the SG even if it already exists
try:
self.network_proxy.update_port(
port_id, security_groups=[sec_grp_id])
port_id, security_groups=sec_grp_ids)
except os_exceptions.NotFoundException as e:
raise base.PortNotFound(str(e))
except Exception as e: