from neutron.db import models_v2
from neutron.db import netmtu_db
from neutron.db.quota import driver # noqa
+from neutron.db import securitygroups_db
from neutron.db import securitygroups_rpc_base as sg_db_rpc
from neutron.db import vlantransparent_db
from neutron.extensions import allowedaddresspairs as addr_pair
from neutron.plugins.ml2 import managers
from neutron.plugins.ml2 import models
from neutron.plugins.ml2 import rpc
+from neutron.quota import resource_registry
LOG = log.getLogger(__name__)
self._aliases = aliases
return self._aliases
+ @resource_registry.tracked_resources(
+ network=models_v2.Network,
+ port=models_v2.Port,
+ subnet=models_v2.Subnet,
+ subnetpool=models_v2.SubnetPool,
+ security_group=securitygroups_db.SecurityGroup,
+ security_group_rule=securitygroups_db.SecurityGroupRule)
def __init__(self):
# First load drivers, then initialize DB, then initialize drivers
self.type_manager = managers.TypeManager()
from oslo_db import api as oslo_db_api
from oslo_db import exception as oslo_db_exception
from oslo_log import log
+from oslo_utils import excutils
from sqlalchemy import event
from neutron.db import api as db_api
@lockutils.synchronized('dirty_tenants')
def _db_event_handler(self, mapper, _conn, target):
- tenant_id = target.get('tenant_id')
- if not tenant_id:
- # NOTE: This is an unexpected error condition. Log anomaly but do
- # not raise as this might have unexpected effects on other
- # operations
- LOG.error(_LE("Model class %s does not have tenant_id attribute"),
- target)
- return
+ try:
+ tenant_id = target['tenant_id']
+ except AttributeError:
+ with excutils.save_and_reraise_exception():
+ LOG.error(_LE("Model class %s does not have a tenant_id "
+ "attribute"), target)
self._dirty_tenants.add(tenant_id)
# Retry the operation if a duplicate entry exception is raised. This
from neutron.db import l3_hamode_db
from neutron.db import l3_hascheduler_db
from neutron.plugins.common import constants
+from neutron.quota import resource_registry
class L3RouterPlugin(common_db_mixin.CommonDbMixin,
"extraroute", "l3_agent_scheduler",
"l3-ha"]
+ @resource_registry.tracked_resources(router=l3_db.Router,
+ floatingip=l3_db.FloatingIP)
def __init__(self):
self.setup_rpc()
self.router_scheduler = importutils.import_object(
self.callbacks = dhcp_rpc.DhcpRpcCallback()
self.log_p = mock.patch('neutron.api.rpc.handlers.dhcp_rpc.LOG')
self.log = self.log_p.start()
+ set_dirty_p = mock.patch('neutron.quota.resource_registry.'
+ 'set_resources_dirty')
+ self.mock_set_dirty = set_dirty_p.start()
def test_get_active_networks(self):
plugin_retval = [dict(id='a'), dict(id='b')]
'22', '22',
remote_ip_prefix,
None,
- None,
- ethertype)
+ ethertype=ethertype)
res = self._create_security_group_rule(self.fmt, rule)
self.assertEqual(res.status_int, webob.exc.HTTPBadRequest.code)
'22', '22',
remote_ip_prefix,
None,
- None,
- ethertype)
+ ethertype=ethertype)
res = self._create_security_group_rule(self.fmt, rule)
self.assertEqual(res.status_int, 201)
res_sg = self.deserialize(self.fmt, res)