diff --git a/neutron/services/metering/drivers/iptables/iptables_driver.py b/neutron/services/metering/drivers/iptables/iptables_driver.py index 5c2c080f094..b6570e0bcbd 100644 --- a/neutron/services/metering/drivers/iptables/iptables_driver.py +++ b/neutron/services/metering/drivers/iptables/iptables_driver.py @@ -135,7 +135,11 @@ class IptablesMeteringDriver(abstract_driver.MeteringAbstractDriver): def _process_metering_label_rules(self, rm, rules, label_chain, rules_chain): im = rm.iptables_manager - ext_dev = self.get_external_device_name(rm.router['gw_port_id']) + ex_gw_port = rm.router.get('gw_port_id') + if not ex_gw_port: + return + + ext_dev = self.get_external_device_name(ex_gw_port) if not ext_dev: return diff --git a/neutron/tests/unit/services/metering/drivers/test_iptables.py b/neutron/tests/unit/services/metering/drivers/test_iptables.py index a33a18b2880..0b71413cdc0 100644 --- a/neutron/tests/unit/services/metering/drivers/test_iptables.py +++ b/neutron/tests/unit/services/metering/drivers/test_iptables.py @@ -166,6 +166,36 @@ class IptablesDriverTestCase(base.BaseTestCase): self.v4filter_inst.assert_has_calls(calls) + def test_process_metering_label_rules_with_no_gateway_router(self): + routers = copy.deepcopy(TEST_ROUTERS) + for router in routers: + router['gw_port_id'] = None + + self.metering.add_metering_label(None, routers) + + calls = [mock.call.add_chain('neutron-meter-l-c5df2fe5-c60', + wrap=False), + mock.call.add_chain('neutron-meter-r-c5df2fe5-c60', + wrap=False), + mock.call.add_rule('neutron-meter-FORWARD', '-j ' + 'neutron-meter-r-c5df2fe5-c60', + wrap=False), + mock.call.add_rule('neutron-meter-l-c5df2fe5-c60', + '', + wrap=False), + mock.call.add_chain('neutron-meter-l-eeef45da-c60', + wrap=False), + mock.call.add_chain('neutron-meter-r-eeef45da-c60', + wrap=False), + mock.call.add_rule('neutron-meter-FORWARD', '-j ' + 'neutron-meter-r-eeef45da-c60', + wrap=False), + mock.call.add_rule('neutron-meter-l-eeef45da-c60', + '', + wrap=False)] + + self.v4filter_inst.assert_has_calls(calls, any_order=False) + def test_add_metering_label_with_rules(self): routers = copy.deepcopy(TEST_ROUTERS) routers[1]['_metering_labels'][0]['rules'][0].update({