openstack/octavia
Code Issues Proposed changes

Merge "Handle race condition deleting security group rule"

Browse Source
This commit is contained in:
Zuul 2017-11-27 23:37:58 +00:00 committed by Gerrit Code Review
commit 62c398c5cb
2 changed files with 25 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -156,7 +156,12 @@ class AllowedAddressPairsDriver(neutron_base.BaseNeutronDriver):
del_ports = set(old_ports) - set(updated_ports)
for rule in rules.get('security_group_rules', []):
if rule.get('port_range_max') in del_ports:
self.neutron_client.delete_security_group_rule(rule.get('id'))
rule_id = rule.get('id')
try:
self.neutron_client.delete_security_group_rule(rule_id)
except neutron_client_exceptions.NotFound:
LOG.info("Security group rule %s not found, will assume "
"it is already deleted.", rule_id)
ethertype = self._get_ethertype_for_ip(load_balancer.vip.ip_address)
for port in add_ports:

19
octavia/tests/unit/network/drivers/neutron/test_allowed_address_pairs.py
View File

@ -675,6 +675,25 @@ class TestAllowedAddressPairsDriver(base.TestCase):
self.driver.update_vip(lb)
delete_rule.assert_called_once_with('ssh-rule')
def test_update_vip_when_security_group_rule_deleted(self):
listeners = []
vip = data_models.Vip(ip_address='10.0.0.2')
lb = data_models.LoadBalancer(id='1', listeners=listeners, vip=vip)
list_sec_grps = self.driver.neutron_client.list_security_groups
list_sec_grps.return_value = {'security_groups': [{'id': 'secgrp-1'}]}
fake_rules = {
'security_group_rules': [
{'id': 'all-egress', 'protocol': None, 'direction': 'egress'},
{'id': 'ssh-rule', 'protocol': 'tcp', 'port_range_max': 22}
]
}
list_rules = self.driver.neutron_client.list_security_group_rules
list_rules.return_value = fake_rules
delete_rule = self.driver.neutron_client.delete_security_group_rule
delete_rule.side_effect = neutron_exceptions.NotFound
self.driver.update_vip(lb)
delete_rule.assert_called_once_with('ssh-rule')
def test_failover_preparation(self):
original_dns_integration_state = self.driver.dns_integration_enabled
self.driver.dns_integration_enabled = False