Browse Source

NSX|V3: Add housekeeping jobs

Adding houskeeper for NSX V3 including handling orphaned DHCP server,
logical swithces, firewall sections & logical routers, and handling
mismatched logical ports.

Change-Id: Id5e038a5c713796a83e485343cdc1672d0c1fd24
changes/42/578342/15
Adit Sarfaty 3 years ago
parent
commit
f2589aefb2
  1. 21
      doc/source/housekeeper.rst
  2. 6
      setup.cfg
  3. 12
      vmware_nsx/common/config.py
  4. 1
      vmware_nsx/plugins/common/housekeeper/housekeeper.py
  5. 0
      vmware_nsx/plugins/nsx_v3/housekeeper/__init__.py
  6. 87
      vmware_nsx/plugins/nsx_v3/housekeeper/mismatch_logical_port.py
  7. 77
      vmware_nsx/plugins/nsx_v3/housekeeper/orphaned_dhcp_server.py
  8. 77
      vmware_nsx/plugins/nsx_v3/housekeeper/orphaned_firewall_section.py
  9. 77
      vmware_nsx/plugins/nsx_v3/housekeeper/orphaned_logical_router.py
  10. 77
      vmware_nsx/plugins/nsx_v3/housekeeper/orphaned_logical_switch.py
  11. 13
      vmware_nsx/plugins/nsx_v3/plugin.py
  12. 0
      vmware_nsx/tests/unit/nsx_v3/housekeeper/__init__.py
  13. 91
      vmware_nsx/tests/unit/nsx_v3/housekeeper/test_mismatch_logical_port.py
  14. 83
      vmware_nsx/tests/unit/nsx_v3/housekeeper/test_orphaned_dhcp_server.py
  15. 85
      vmware_nsx/tests/unit/nsx_v3/housekeeper/test_orphaned_logical_router.py
  16. 84
      vmware_nsx/tests/unit/nsx_v3/housekeeper/test_orphaned_logical_switch.py

21
doc/source/housekeeper.rst

@ -20,10 +20,10 @@ Configuration
Housekeeping mechanism uses two configuration parameters:
nsxv.housekeeping_jobs: The housekeeper can be configured which tasks to
nsxv/v3.housekeeping_jobs: The housekeeper can be configured which tasks to
execute and which should be skipped.
nsxv.housekeeping_readonly: Housekeeper may attempt to fix a broken environment
nsxv/v3.housekeeping_readonly: Housekeeper may attempt to fix a broken environment
when this flag is set to False, or otherwise will just warn about
inconsistencies.
@ -71,3 +71,20 @@ When in non-readonly mode, the job will reset the Edge appliance configuration.
lbaas_pending: scans the neutron DB for LBaaS objects which are pending for too
long. Report it, and if in non-readonly mode change its status to ERROR
NSX-v3
~~~~~~
orphaned_logical_router: scans the NSX backend for logical routers which are
missing from the neutron DB. Report it, and if in non-readonly mode delete them.
orphaned_logical_swithces: scans the NSX backend for logical switches which are
missing from the neutron DB. Report it, and if in non-readonly mode delete them.
orphaned_dhcp_server: scans the NSX backend for DHCP servers which are
missing a matching network in the neutron DB. Report it, and if in non-readonly
mode delete them.
orphaned_firewall_section: scans the NSX backend for firewall sections which are
missing a matching security group in the neutron DB. Report it, and if in non-readonly
mode delete them.

6
setup.cfg

@ -85,6 +85,12 @@ vmware_nsx.neutron.nsxv.housekeeper.jobs =
error_dhcp_edge = vmware_nsx.plugins.nsx_v.housekeeper.error_dhcp_edge:ErrorDhcpEdgeJob
error_backup_edge = vmware_nsx.plugins.nsx_v.housekeeper.error_backup_edge:ErrorBackupEdgeJob
lbaas_pending = vmware_nsx.plugins.nsx_v.housekeeper.lbaas_pending:LbaasPendingJob
vmware_nsx.neutron.nsxv3.housekeeper.jobs =
orphaned_dhcp_server = vmware_nsx.plugins.nsx_v3.housekeeper.orphaned_dhcp_server:OrphanedDhcpServerJob
orphaned_logical_switch = vmware_nsx.plugins.nsx_v3.housekeeper.orphaned_logical_switch:OrphanedLogicalSwitchJob
orphaned_logical_router = vmware_nsx.plugins.nsx_v3.housekeeper.orphaned_logical_router:OrphanedLogicalRouterJob
orphaned_firewall_section = vmware_nsx.plugins.nsx_v3.housekeeper.orphaned_firewall_section:OrphanedFirewallSectionJob
mismatch_logical_port = vmware_nsx.plugins.nsx_v3.housekeeper.mismatch_logical_port:MismatchLogicalportJob
[build_sphinx]
source-dir = doc/source

12
vmware_nsx/common/config.py

@ -459,6 +459,18 @@ nsx_v3_opts = [
help=_("When True, port security will be set to False for "
"newly created ENS networks and ports, overriding "
"user settings")),
cfg.ListOpt('housekeeping_jobs',
default=['orphaned_dhcp_server', 'orphaned_logical_switch',
'orphaned_logical_router', 'mismatch_logical_port',
'orphaned_firewall_section'],
help=_("List of the enabled housekeeping jobs")),
cfg.ListOpt('housekeeping_readonly_jobs',
default=[],
help=_("List of housekeeping jobs which are enabled in read "
"only mode")),
cfg.BoolOpt('housekeeping_readonly',
default=True,
help=_("Housekeeping will only warn about breakage.")),
]

1
vmware_nsx/plugins/common/housekeeper/housekeeper.py

@ -53,6 +53,7 @@ class NsxHousekeeper(stevedore.named.NamedExtensionManager):
else:
LOG.info('Housekeeper initialized')
self.results = {}
self.jobs = {}
super(NsxHousekeeper, self).__init__(
hk_ns, hk_jobs, invoke_on_load=True,

0
vmware_nsx/plugins/nsx_v3/housekeeper/__init__.py

87
vmware_nsx/plugins/nsx_v3/housekeeper/mismatch_logical_port.py

@ -0,0 +1,87 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3 import utils as v3_utils
LOG = log.getLogger(__name__)
class MismatchLogicalportJob(base_job.BaseJob):
def __init__(self, global_readonly, readonly_jobs):
super(MismatchLogicalportJob, self).__init__(
global_readonly, readonly_jobs)
def get_project_plugin(self, plugin):
return plugin.get_plugin_by_type(projectpluginmap.NsxPlugins.NSX_T)
def get_name(self):
return 'mismatch_logical_port'
def get_description(self):
return 'Detect mismatched configuration on NSX logical ports'
def run(self, context, readonly=False):
super(MismatchLogicalportJob, self).run(context)
# get all orphaned DHCP servers
mismatch_ports = v3_utils.get_mismatch_logical_ports(
context, self.plugin.nsxlib, self.plugin)
info = ""
if not mismatch_ports:
msg = 'No mismatched logical ports detected.'
info = base_job.housekeeper_info(info, msg)
return {'error_count': 0, 'fixed_count': 0, 'error_info': info}
msg = ("Found %(len)s mismatched logical port%(plural)s:" %
{'len': len(mismatch_ports),
'plural': 's' if len(mismatch_ports) > 1 else ''})
info = base_job.housekeeper_warning(info, msg)
fixed_count = 0
for port_problem in mismatch_ports:
msg = ("Logical port %(nsx_id)s "
"[neutron id: %(id)s] error: %(err)s" %
{'nsx_id': port_problem['nsx_id'],
'id': port_problem['neutron_id'],
'err': port_problem['error']})
if not readonly:
# currently we mitigate only address bindings mismatches
err_type = port_problem['error_type']
if err_type == v3_utils.PORT_ERROR_TYPE_BINDINGS:
# Create missing address bindings on backend
port = port_problem['port']
try:
address_bindings = self.plugin._build_address_bindings(
port)
self.plugin.nsxlib.logical_port.update(
port_problem['nsx_id'], port_problem['neutron_id'],
address_bindings=address_bindings)
except Exception as e:
msg = "%s failed to be fixed: %s" % (msg, e)
else:
fixed_count = fixed_count + 1
msg = "%s was fixed." % msg
else:
msg = "%s cannot be fixed automatically." % msg
info = base_job.housekeeper_warning(info, msg)
return {'error_count': len(mismatch_ports),
'error_info': info,
'fixed_count': fixed_count}

77
vmware_nsx/plugins/nsx_v3/housekeeper/orphaned_dhcp_server.py

@ -0,0 +1,77 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3 import utils as v3_utils
LOG = log.getLogger(__name__)
class OrphanedDhcpServerJob(base_job.BaseJob):
def __init__(self, global_readonly, readonly_jobs):
super(OrphanedDhcpServerJob, self).__init__(
global_readonly, readonly_jobs)
def get_project_plugin(self, plugin):
return plugin.get_plugin_by_type(projectpluginmap.NsxPlugins.NSX_T)
def get_name(self):
return 'orphaned_dhcp_server'
def get_description(self):
return 'Detect orphaned DHCP server'
def run(self, context, readonly=False):
super(OrphanedDhcpServerJob, self).run(context)
# get all orphaned DHCP servers
orphaned_servers = v3_utils.get_orphaned_dhcp_servers(
context, self.plugin, self.plugin.nsxlib)
info = ""
if not orphaned_servers:
msg = 'No orphaned DHCP servers detected.'
info = base_job.housekeeper_info(info, msg)
return {'error_count': 0, 'fixed_count': 0, 'error_info': msg}
msg = ("Found %(len)s orphaned DHCP server%(plural)s:" %
{'len': len(orphaned_servers),
'plural': 's' if len(orphaned_servers) > 1 else ''})
info = base_job.housekeeper_warning(info, msg)
fixed_count = 0
for server in orphaned_servers:
msg = ("DHCP server %(name)s [id: %(id)s] "
"(neutron network: %(net)s)" %
{'name': server['display_name'],
'id': server['id'],
'net': server['neutron_net_id']
if server.get('neutron_net_id') else 'Unknown'})
if not readonly:
success, error = v3_utils.delete_orphaned_dhcp_server(
context, self.plugin.nsxlib, server)
if success:
msg = "%s was removed." % msg
fixed_count = fixed_count + 1
else:
msg = "%s failed to be removed: %s." % (msg, error)
info = base_job.housekeeper_warning(info, msg)
return {'error_count': len(orphaned_servers),
'error_info': info,
'fixed_count': fixed_count}

77
vmware_nsx/plugins/nsx_v3/housekeeper/orphaned_firewall_section.py

@ -0,0 +1,77 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3 import utils as v3_utils
LOG = log.getLogger(__name__)
class OrphanedFirewallSectionJob(base_job.BaseJob):
def __init__(self, global_readonly, readonly_jobs):
super(OrphanedFirewallSectionJob, self).__init__(
global_readonly, readonly_jobs)
def get_project_plugin(self, plugin):
return plugin.get_plugin_by_type(projectpluginmap.NsxPlugins.NSX_T)
def get_name(self):
return 'orphaned_firewall_section'
def get_description(self):
return 'Detect orphaned firewall sections'
def run(self, context, readonly=False):
super(OrphanedFirewallSectionJob, self).run(context)
# get all orphaned firewall sections
orphaned_sections = v3_utils.get_orphaned_firewall_sections(
context, self.plugin.nsxlib)
info = ""
if not orphaned_sections:
msg = 'No orphaned firewall sections detected.'
info = base_job.housekeeper_info(info, msg)
return {'error_count': 0, 'fixed_count': 0, 'error_info': info}
msg = ("Found %(len)s orphaned firewall section%(plural)s:" %
{'len': len(orphaned_sections),
'plural': 's' if len(orphaned_sections) > 1 else ''})
info = base_job.housekeeper_warning(info, msg)
fixed_count = 0
for section in orphaned_sections:
msg = ("Firewall section %(name)s [id: %(id)s] "
"neutron security group: %(sg)s" %
{'name': section['display_name'],
'id': section['id'],
'sg': section['neutron_sg_id'] if section['neutron_sg_id']
else 'Unknown'})
if not readonly:
try:
self.plugin.nsxlib.firewall_section.delete(section['id'])
except Exception as e:
msg = "%s failed to be removed: %s." % (msg, e)
else:
fixed_count = fixed_count + 1
msg = "%s was removed." % msg
info = base_job.housekeeper_warning(info, msg)
return {'error_count': len(orphaned_sections),
'error_info': info,
'fixed_count': fixed_count}

77
vmware_nsx/plugins/nsx_v3/housekeeper/orphaned_logical_router.py

@ -0,0 +1,77 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3 import utils as v3_utils
LOG = log.getLogger(__name__)
class OrphanedLogicalRouterJob(base_job.BaseJob):
def __init__(self, global_readonly, readonly_jobs):
super(OrphanedLogicalRouterJob, self).__init__(
global_readonly, readonly_jobs)
def get_project_plugin(self, plugin):
return plugin.get_plugin_by_type(projectpluginmap.NsxPlugins.NSX_T)
def get_name(self):
return 'orphaned_logical_router'
def get_description(self):
return 'Detect orphaned logical routers'
def run(self, context, readonly=False):
super(OrphanedLogicalRouterJob, self).run(context)
# get all orphaned DHCP servers
orphaned_routers = v3_utils.get_orphaned_routers(
context, self.plugin.nsxlib)
info = ""
if not orphaned_routers:
msg = 'No orphaned logical routers detected.'
info = base_job.housekeeper_info(info, msg)
return {'error_count': 0, 'fixed_count': 0, 'error_info': info}
msg = ("Found %(len)s orphaned logical router%(plural)s:" %
{'len': len(orphaned_routers),
'plural': 's' if len(orphaned_routers) > 1 else ''})
info = base_job.housekeeper_warning(info, msg)
fixed_count = 0
for router in orphaned_routers:
msg = ("Logical router %(name)s [id: %(id)s] "
"(neutron router: %(rtr)s)" %
{'name': router['display_name'],
'id': router['id'],
'rtr': router['neutron_router_id']
if router['neutron_router_id'] else 'Unknown'})
if not readonly:
success, error = v3_utils.delete_orphaned_router(
self.plugin.nsxlib, router['id'])
if success:
fixed_count = fixed_count + 1
msg = "%s was removed." % msg
else:
msg = "%s failed to be removed: %s." % (msg, error)
info = base_job.housekeeper_warning(info, msg)
return {'error_count': len(orphaned_routers),
'error_info': info,
'fixed_count': fixed_count}

77
vmware_nsx/plugins/nsx_v3/housekeeper/orphaned_logical_switch.py

@ -0,0 +1,77 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3 import utils as v3_utils
LOG = log.getLogger(__name__)
class OrphanedLogicalSwitchJob(base_job.BaseJob):
def __init__(self, global_readonly, readonly_jobs):
super(OrphanedLogicalSwitchJob, self).__init__(
global_readonly, readonly_jobs)
def get_project_plugin(self, plugin):
return plugin.get_plugin_by_type(projectpluginmap.NsxPlugins.NSX_T)
def get_name(self):
return 'orphaned_logical_switch'
def get_description(self):
return 'Detect orphaned logical switches'
def run(self, context, readonly=False):
super(OrphanedLogicalSwitchJob, self).run(context)
# get all orphaned DHCP servers
orphaned_swithces = v3_utils.get_orphaned_networks(
context, self.plugin.nsxlib)
info = ""
if not orphaned_swithces:
msg = 'No orphaned logical switches detected.'
info = base_job.housekeeper_info(info, msg)
return {'error_count': 0, 'fixed_count': 0, 'error_info': info}
msg = ("Found %(len)s orphaned logical switch%(plural)s:" %
{'len': len(orphaned_swithces),
'plural': 'es' if len(orphaned_swithces) > 1 else ''})
info = base_job.housekeeper_warning(info, msg)
fixed_count = 0
for switch in orphaned_swithces:
msg = ("Logical switch %(name)s [id: %(id)s] "
"(neutron network: %(net)s)" %
{'name': switch['display_name'],
'id': switch['id'],
'net': switch['neutron_net_id'] if switch['neutron_net_id']
else 'Unknown'})
if not readonly:
try:
self.plugin.nsxlib.logical_switch.delete(switch['id'])
except Exception as e:
msg = "%s failed to be removed: %s." % (msg, e)
else:
fixed_count = fixed_count + 1
msg = "%s was removed." % (msg)
info = base_job.housekeeper_warning(info, msg)
return {'error_count': len(orphaned_swithces),
'error_info': info,
'fixed_count': fixed_count}

13
vmware_nsx/plugins/nsx_v3/plugin.py

@ -104,10 +104,12 @@ from vmware_nsx.db import extended_security_group_rule as extend_sg_rule
from vmware_nsx.db import maclearning as mac_db
from vmware_nsx.dhcp_meta import rpc as nsx_rpc
from vmware_nsx.extensions import advancedserviceproviders as as_providers
from vmware_nsx.extensions import housekeeper as hk_ext
from vmware_nsx.extensions import maclearning as mac_ext
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.extensions import providersecuritygroup as provider_sg
from vmware_nsx.extensions import securitygrouplogging as sg_logging
from vmware_nsx.plugins.common.housekeeper import housekeeper
from vmware_nsx.plugins.common import plugin as nsx_plugin_common
from vmware_nsx.plugins.nsx import utils as tvd_utils
from vmware_nsx.plugins.nsx_v3 import availability_zones as nsx_az
@ -182,7 +184,8 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
vlantransparent_db.Vlantransparent_db_mixin,
mac_db.MacLearningDbMixin,
nsx_com_az.NSXAvailabilityZonesPluginCommon,
l3_attrs_db.ExtraAttributesMixin):
l3_attrs_db.ExtraAttributesMixin,
hk_ext.Housekeeper):
__native_bulk_support = True
__native_pagination_support = True
@ -209,6 +212,7 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
"subnet_allocation",
"security-group-logging",
"provider-security-group",
"housekeeper",
"port-security-groups-filtering"]
@resource_registry.tracked_resources(
@ -456,6 +460,13 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
# Init the FWaaS support
self._init_fwaas()
# Init the house keeper
self.housekeeper = housekeeper.NsxHousekeeper(
hk_ns='vmware_nsx.neutron.nsxv3.housekeeper.jobs',
hk_jobs=cfg.CONF.nsx_v3.housekeeping_jobs,
hk_readonly=cfg.CONF.nsx_v3.housekeeping_readonly,
hk_readonly_jobs=cfg.CONF.nsx_v3.housekeeping_readonly_jobs)
self.init_is_complete = True
def _extend_fault_map(self):

0
vmware_nsx/tests/unit/nsx_v3/housekeeper/__init__.py

91
vmware_nsx/tests/unit/nsx_v3/housekeeper/test_mismatch_logical_port.py

@ -0,0 +1,91 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from neutron.tests import base
from neutron_lib.plugins import constants
from oslo_utils import uuidutils
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3.housekeeper import mismatch_logical_port
from vmware_nsxlib.v3 import exceptions as nsxlib_exc
DUMMY_PORT = {
"resource_type": "LogicalPort",
"id": uuidutils.generate_uuid(),
"display_name": "test",
"tags": [{
"scope": "os-neutron-dport-id",
"tag": uuidutils.generate_uuid()
}, {
"scope": "os-project-id",
"tag": uuidutils.generate_uuid()
}, {
"scope": "os-project-name",
"tag": "admin"
}, {
"scope": "os-api-version",
"tag": "13.0.0.0b3.dev90"
}],
"logical_switch_id": uuidutils.generate_uuid(),
"admin_state": "UP",
"switching_profile_ids": []}
class MismatchLogicalPortTestCaseReadOnly(base.BaseTestCase):
def setUp(self):
def get_plugin_mock(alias=constants.CORE):
if alias in (constants.CORE, constants.L3):
return self.plugin
super(MismatchLogicalPortTestCaseReadOnly, self).setUp()
self.plugin = mock.Mock()
self.plugin.nsxlib = mock.Mock()
self.plugin.nsxlib.switching_profile.find_by_display_name = mock.Mock(
return_value=[{'id': 'Dummy'}])
self.context = mock.Mock()
self.context.session = mock.Mock()
mock.patch('neutron_lib.plugins.directory.get_plugin',
side_effect=get_plugin_mock).start()
self.log = mock.Mock()
base_job.LOG = self.log
self.job = mismatch_logical_port.MismatchLogicalportJob(True, [])
def run_job(self):
self.job.run(self.context, readonly=True)
def test_clean_run(self):
with mock.patch.object(self.plugin, 'get_ports', return_value=[]):
self.run_job()
self.log.warning.assert_not_called()
def test_with_mismatched_ls(self):
with mock.patch.object(
self.plugin, 'get_ports',
return_value=[{'id': uuidutils.generate_uuid()}]),\
mock.patch("vmware_nsx.plugins.nsx_v3.utils.get_port_nsx_id",
return_value=uuidutils.generate_uuid()),\
mock.patch.object(self.plugin.nsxlib.logical_port, 'get',
side_effect=nsxlib_exc.ResourceNotFound):
self.run_job()
self.log.warning.assert_called()
class MismatchLogicalPortTestCaseReadWrite(
MismatchLogicalPortTestCaseReadOnly):
def run_job(self):
self.job.run(self.context, readonly=False)

83
vmware_nsx/tests/unit/nsx_v3/housekeeper/test_orphaned_dhcp_server.py

@ -0,0 +1,83 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from neutron.tests import base
from neutron_lib.plugins import constants
from oslo_utils import uuidutils
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3.housekeeper import orphaned_dhcp_server
DUMMY_DHCP_SERVER = {
"resource_type": "LogicalDhcpServer",
"id": uuidutils.generate_uuid(),
"display_name": "test",
"tags": [{
"scope": "os-neutron-net-id",
"tag": uuidutils.generate_uuid()
}, {
"scope": "os-project-id",
"tag": uuidutils.generate_uuid()
}, {
"scope": "os-project-name",
"tag": "admin"
}, {
"scope": "os-api-version",
"tag": "13.0.0.0b3.dev90"
}],
"attached_logical_port_id": uuidutils.generate_uuid(),
"dhcp_profile_id": uuidutils.generate_uuid()}
class OrphanedDhcpServerTestCaseReadOnly(base.BaseTestCase):
def setUp(self):
def get_plugin_mock(alias=constants.CORE):
if alias in (constants.CORE, constants.L3):
return self.plugin
super(OrphanedDhcpServerTestCaseReadOnly, self).setUp()
self.plugin = mock.Mock()
self.plugin.nsxlib = mock.Mock()
self.context = mock.Mock()
self.context.session = mock.Mock()
mock.patch('neutron_lib.plugins.directory.get_plugin',
side_effect=get_plugin_mock).start()
self.log = mock.Mock()
base_job.LOG = self.log
self.job = orphaned_dhcp_server.OrphanedDhcpServerJob(True, [])
def run_job(self):
self.job.run(self.context, readonly=True)
def test_clean_run(self):
with mock.patch.object(self.plugin.nsxlib.dhcp_server, 'list',
return_value={'results': []}):
self.run_job()
self.log.warning.assert_not_called()
def test_with_orphaned_servers(self):
with mock.patch.object(self.plugin.nsxlib.dhcp_server, 'list',
return_value={'results': [DUMMY_DHCP_SERVER]}),\
mock.patch.object(self.plugin, 'get_network',
side_effect=Exception):
self.run_job()
self.log.warning.assert_called()
class OrphanedDhcpServerTestCaseReadWrite(OrphanedDhcpServerTestCaseReadOnly):
def run_job(self):
self.job.run(self.context, readonly=False)

85
vmware_nsx/tests/unit/nsx_v3/housekeeper/test_orphaned_logical_router.py

@ -0,0 +1,85 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from neutron.tests import base
from neutron_lib.plugins import constants
from oslo_utils import uuidutils
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3.housekeeper import orphaned_logical_router
DUMMY_ROUTER = {
"resource_type": "LogicalRouter",
"id": uuidutils.generate_uuid(),
"display_name": "test",
"tags": [{
"scope": "os-neutron-router-id",
"tag": uuidutils.generate_uuid()
}, {
"scope": "os-project-id",
"tag": uuidutils.generate_uuid()
}, {
"scope": "os-project-name",
"tag": "admin"
}, {
"scope": "os-api-version",
"tag": "13.0.0.0b3.dev90"
}],
"edge_cluster_id": uuidutils.generate_uuid(),
"router_type": "TIER1"}
class OrphanedLogicalRouterTestCaseReadOnly(base.BaseTestCase):
def setUp(self):
def get_plugin_mock(alias=constants.CORE):
if alias in (constants.CORE, constants.L3):
return self.plugin
super(OrphanedLogicalRouterTestCaseReadOnly, self).setUp()
self.plugin = mock.Mock()
self.plugin.nsxlib = mock.Mock()
self.context = mock.Mock()
self.context.session = mock.Mock()
mock.patch('neutron_lib.plugins.directory.get_plugin',
side_effect=get_plugin_mock).start()
self.log = mock.Mock()
base_job.LOG = self.log
self.job = orphaned_logical_router.OrphanedLogicalRouterJob(True, [])
def run_job(self):
self.job.run(self.context, readonly=True)
def test_clean_run(self):
with mock.patch.object(self.plugin.nsxlib.logical_router, 'list',
return_value={'results': []}):
self.run_job()
self.log.warning.assert_not_called()
def test_with_orphaned_ls(self):
with mock.patch.object(self.plugin.nsxlib.logical_router, 'list',
return_value={'results': [DUMMY_ROUTER]}),\
mock.patch("vmware_nsx.db.db.get_neutron_from_nsx_router_id",
return_value=None):
self.run_job()
self.log.warning.assert_called()
class OrphanedLogicalRouterTestCaseReadWrite(
OrphanedLogicalRouterTestCaseReadOnly):
def run_job(self):
self.job.run(self.context, readonly=False)

84
vmware_nsx/tests/unit/nsx_v3/housekeeper/test_orphaned_logical_switch.py

@ -0,0 +1,84 @@
# Copyright 2018 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from neutron.tests import base
from neutron_lib.plugins import constants
from oslo_utils import uuidutils
from vmware_nsx.plugins.common.housekeeper import base_job
from vmware_nsx.plugins.nsx_v3.housekeeper import orphaned_logical_switch
DUMMY_LS = {
"resource_type": "LogicalSwitch",
"id": uuidutils.generate_uuid(),
"display_name": "test",
"tags": [{
"scope": "os-neutron-net-id",
"tag": uuidutils.generate_uuid()
}, {
"scope": "os-project-id",
"tag": uuidutils.generate_uuid()
}, {
"scope": "os-project-name",
"tag": "admin"
}, {
"scope": "os-api-version",
"tag": "13.0.0.0b3.dev90"
}],
"transport_zone_id": uuidutils.generate_uuid(),
"address_bindings": []}
class OrphanedLogicalSwitchTestCaseReadOnly(base.BaseTestCase):
def setUp(self):
def get_plugin_mock(alias=constants.CORE):
if alias in (constants.CORE, constants.L3):
return self.plugin
super(OrphanedLogicalSwitchTestCaseReadOnly, self).setUp()
self.plugin = mock.Mock()
self.plugin.nsxlib = mock.Mock()
self.context = mock.Mock()
self.context.session = mock.Mock()
mock.patch('neutron_lib.plugins.directory.get_plugin',
side_effect=get_plugin_mock).start()
self.log = mock.Mock()
base_job.LOG = self.log
self.job = orphaned_logical_switch.OrphanedLogicalSwitchJob(True, [])
def run_job(self):
self.job.run(self.context, readonly=True)
def test_clean_run(self):
with mock.patch.object(self.plugin.nsxlib.logical_switch, 'list',
return_value={'results': []}):
self.run_job()
self.log.warning.assert_not_called()
def test_with_orphaned_ls(self):
with mock.patch.object(self.plugin.nsxlib.logical_switch, 'list',
return_value={'results': [DUMMY_LS]}),\
mock.patch("vmware_nsx.db.db.get_net_ids", return_value=None):
self.run_job()
self.log.warning.assert_called()
class OrphanedLogicalSwitchTestCaseReadWrite(
OrphanedLogicalSwitchTestCaseReadOnly):
def run_job(self):
self.job.run(self.context, readonly=False)
Loading…
Cancel
Save