Remove the neutron-debug tool

With removal of the neutron client shell code this tool is
no longer usable. It had been marked for deprecation since
the Newton (9.0) cycle and unmaintained.

This code is also breaking the neutron gate pep8 job.

Change-Id: I3c0c93de0b860d9287019b7834cb8337d9668cc0
This commit is contained in:
Brian Haley 2023-05-12 12:42:31 -04:00
parent 47d4ec4e99
commit 01af4b2cda
11 changed files with 6 additions and 1072 deletions

View File

@ -8,6 +8,5 @@ Command-Line Interface Reference
.. toctree::
:maxdepth: 1
neutron-debug
neutron-sanity-check
neutron-status

View File

@ -1,309 +0,0 @@
.. This file is manually generated, unlike many of the other chapters.
=============
neutron-debug
=============
The :command:`neutron-debug` client is an extension to the :command:`neutron`
command-line interface (CLI) for the OpenStack neutron-debug tool.
This chapter documents :command:`neutron-debug` version ``2.3.0``.
For help on a specific :command:`neutron-debug` command, enter:
.. code-block:: console
$ neutron-debug help COMMAND
.. _neutron-debug_usage:
neutron-debug usage
~~~~~~~~~~~~~~~~~~~
.. code-block:: console
usage: neutron-debug [--version] [-v] [-q] [-h] [-r NUM]
[--os-service-type <os-service-type>]
[--os-endpoint-type <os-endpoint-type>]
[--service-type <service-type>]
[--endpoint-type <endpoint-type>]
[--os-auth-strategy <auth-strategy>] [--os-cloud <cloud>]
[--os-auth-url <auth-url>]
[--os-tenant-name <auth-tenant-name> | --os-project-name <auth-project-name>]
[--os-tenant-id <auth-tenant-id> | --os-project-id <auth-project-id>]
[--os-username <auth-username>]
[--os-user-id <auth-user-id>]
[--os-user-domain-id <auth-user-domain-id>]
[--os-user-domain-name <auth-user-domain-name>]
[--os-project-domain-id <auth-project-domain-id>]
[--os-project-domain-name <auth-project-domain-name>]
[--os-cert <certificate>] [--os-cacert <ca-certificate>]
[--os-key <key>] [--os-password <auth-password>]
[--os-region-name <auth-region-name>]
[--os-token <token>] [--http-timeout <seconds>]
[--os-url <url>] [--insecure] [--config-file CONFIG_FILE]
<subcommand> ...
Subcommands
-----------
``probe-create``
Create probe port - create port and interface within a network namespace.
``probe-list``
List all probes.
``probe-clear``
Clear all probes.
``probe-delete``
Delete probe - delete port then delete the namespace.
``probe-exec``
Execute commands in the namespace of the probe.
``ping-all``
``ping-all`` is an all-in-one command to ping all fixed IPs in a specified
network.
.. _neutron-debug_optional:
neutron-debug optional arguments
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
``--version``
Show program's version number and exit
``-v, --verbose, --debug``
Increase verbosity of output and show tracebacks on
errors. You can repeat this option.
``-q, --quiet``
Suppress output except warnings and errors.
``-h, --help``
Show this help message and exit
``-r NUM, --retries NUM``
How many times the request to the Neutron server
should be retried if it fails.
``--os-service-type <os-service-type>``
Defaults to env[OS_NETWORK_SERVICE_TYPE] or network.
``--os-endpoint-type <os-endpoint-type>``
Defaults to ``env[OS_ENDPOINT_TYPE]`` or public.
``--service-type <service-type>``
DEPRECATED! Use --os-service-type.
``--endpoint-type <endpoint-type>``
DEPRECATED! Use --os-endpoint-type.
``--os-auth-strategy <auth-strategy>``
DEPRECATED! Only keystone is supported.
``os-cloud <cloud>``
Defaults to env[OS_CLOUD].
``--os-auth-url <auth-url>``
Authentication URL, defaults to env[OS_AUTH_URL].
``--os-tenant-name <auth-tenant-name>``
Authentication tenant name, defaults to
env[OS_TENANT_NAME].
``--os-project-name <auth-project-name>``
Another way to specify tenant name. This option is
mutually exclusive with --os-tenant-name. Defaults to
env[OS_PROJECT_NAME].
``--os-tenant-id <auth-tenant-id>``
Authentication tenant ID, defaults to
env[OS_TENANT_ID].
``--os-project-id <auth-project-id>``
Another way to specify tenant ID. This option is
mutually exclusive with --os-tenant-id. Defaults to
env[OS_PROJECT_ID].
``--os-username <auth-username>``
Authentication username, defaults to env[OS_USERNAME].
``--os-user-id <auth-user-id>``
Authentication user ID (Env: OS_USER_ID)
``--os-user-domain-id <auth-user-domain-id>``
OpenStack user domain ID. Defaults to
env[OS_USER_DOMAIN_ID].
``--os-user-domain-name <auth-user-domain-name>``
OpenStack user domain name. Defaults to
env[OS_USER_DOMAIN_NAME].
``--os-project-domain-id <auth-project-domain-id>``
Defaults to env[OS_PROJECT_DOMAIN_ID].
``--os-project-domain-name <auth-project-domain-name>``
Defaults to env[OS_PROJECT_DOMAIN_NAME].
``--os-cert <certificate>``
Path of certificate file to use in SSL connection.
This file can optionally be prepended with the private
key. Defaults to env[OS_CERT].
``--os-cacert <ca-certificate>``
Specify a CA bundle file to use in verifying a TLS
(https) server certificate. Defaults to
env[OS_CACERT].
``--os-key <key>``
Path of client key to use in SSL connection. This
option is not necessary if your key is prepended to
your certificate file. Defaults to env[OS_KEY].
``--os-password <auth-password>``
Authentication password, defaults to env[OS_PASSWORD].
``--os-region-name <auth-region-name>``
Authentication region name, defaults to
env[OS_REGION_NAME].
``--os-token <token>``
Authentication token, defaults to env[OS_TOKEN].
``--http-timeout <seconds>``
Timeout in seconds to wait for an HTTP response.
Defaults to env[OS_NETWORK_TIMEOUT] or None if not
specified.
``--os-url <url>``
Defaults to env[OS_URL]
``--insecure``
Explicitly allow neutronclient to perform "insecure"
SSL (https) requests. The server's certificate will
not be verified against any certificate authorities.
This option should be used with caution.
``--config-file CONFIG_FILE``
Config file for interface driver (You may also use l3_agent.ini)
neutron-debug probe-create command
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: console
usage: neutron-debug probe-create NET
Create probe port - create port and interface,
then place it into the created network namespace.
Positional arguments
--------------------
``NET ID``
ID of the network in which the probe will be created.
neutron-debug probe-list command
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: console
usage: neutron-debug probe-list
List probes.
neutron-debug probe-clear command
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: console
usage: neutron-debug probe-clear
Clear all probes.
neutron-debug probe-delete command
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: console
usage: neutron-debug probe-delete <port-id>
Remove a probe.
Positional arguments
--------------------
``<port-id>``
ID of the probe to delete.
neutron-debug probe-exec command
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: console
usage: neutron-debug probe-exec <port-id> <command>
Execute commands in the namespace of the probe
neutron-debug ping-all command
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
.. code-block:: console
usage: neutron-debug ping-all <port-id> --timeout <number>
All-in-one command to ping all fixed IPs in a specified network.
A probe creation is not needed for this command.
A new probe is created automatically.
It will, however, need to be deleted manually when it is no longer needed.
When there are multiple networks, the newly created probe will be attached
to a random network and thus the ping will take place from within that
random network.
Positional arguments
--------------------
``<port-id>``
ID of the port to use.
Optional arguments
------------------
``--timeout <timeout in seconds>``
Optional ping timeout.
neutron-debug example
~~~~~~~~~~~~~~~~~~~~~
.. code-block:: console
usage: neutron-debug create-probe <NET_ID>
Create a probe namespace within the network identified by ``NET_ID``.
The namespace will have the name of qprobe-<UUID of the probe port>
.. note::
For the following examples to function, the security group rules
may need to be modified to allow the SSH (TCP port 22) or ping
(ICMP) traffic into network.
.. code-block:: console
usage: neutron-debug probe-exec <probe ID> "ssh <IP of instance>"
SSH to an instance within the network.
.. code-block:: console
usage: neutron-debug ping-all <network ID>
Ping all instances on this network to verify they are responding.
.. code-block:: console
usage: neutron-debug probe-exec <probe_ID> dhcping <VM_MAC address> -s <IP of DHCP server>
Ping the DHCP server for this network using dhcping to verify it is working.

View File

@ -1,38 +0,0 @@
Debug Helper Script for Neutron
- Configure
export NEUTRON_TEST_CONFIG_FILE=/etc/neutron/debug.ini
or
export NEUTRON_TEST_CONFIG_FILE=/etc/neutron/l3_agent.ini
you can also specify config file by --config-file option
- Usage
neutron-debug commands
probe-create <net-id>
Create probe port - create port and interface, then plug it in.
This commands returns a port id of a probe port. A probe port is a port which is used to test.
The port id is probe id.
We can have multiple probe probes in a network, in order to check connectivity between ports.
neutron-debug probe-exec probe_id_1 'nc -l 192.168.100.3 22'
neutron-debug probe-exec probe_id_2 'nc -vz 192.168.100.4 22'
Note: You should use a user and a tenant who has permission to
modify network and subnet if you want to probe. For example, you need to be admin user if you
want to probe external network.
probe-delete <port-id> Delete probe - delete port then unplug
probe-exec <port-id> 'command' Exec commands on the namespace of the probe
`probe-exec <port-id>` 'interactive command' Exec interactive command (eg, ssh)
probe-list List probes
probe-clear Clear All probes
ping-all --id <network_id> --timeout 1 (optional)
ping-all is all-in-one command to ping all fixed ip's in all network or a specified network.
In the command probe is automatically created if needed.
neutron-debug extends the shell of neutronclient, so you can use all the commands of neutron

View File

@ -1,130 +0,0 @@
# Copyright 2012, Nachi Ueno, NTT MCL, 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 cliff import lister
from neutronclient.common import utils
from neutronclient.neutron import v2_0 as client
from neutronclient.neutron.v2_0 import port
from neutron._i18n import _
class ProbeCommand(client.NeutronCommand):
def get_debug_agent(self):
return self.app.debug_agent
class CreateProbe(ProbeCommand):
"""Create probe port and interface, then plug it in."""
def get_parser(self, prog_name):
parser = super(CreateProbe, self).get_parser(prog_name)
parser.add_argument(
'id', metavar='network_id',
help=_('ID of network to probe'))
parser.add_argument(
'--device-owner',
default='network', choices=['network', 'compute'],
help=_('Owner type of the device: network/compute'))
return parser
def take_action(self, parsed_args):
debug_agent = self.get_debug_agent()
probe_port = debug_agent.create_probe(parsed_args.id,
parsed_args.device_owner)
self.log.info(_('Probe created : %s '), probe_port.id)
class DeleteProbe(ProbeCommand):
"""Delete probe - delete port then uplug."""
def get_parser(self, prog_name):
parser = super(DeleteProbe, self).get_parser(prog_name)
parser.add_argument(
'id', metavar='port_id',
help=_('ID of probe port to delete'))
return parser
def take_action(self, parsed_args):
debug_agent = self.get_debug_agent()
debug_agent.delete_probe(parsed_args.id)
self.log.info(_('Probe %s deleted'), parsed_args.id)
class ListProbe(ProbeCommand, lister.Lister):
"""List probes."""
_formatters = {'fixed_ips': port._format_fixed_ips, }
def take_action(self, parsed_args):
debug_agent = self.get_debug_agent()
info = debug_agent.list_probes()
columns = sorted(info[0].keys()) if info else []
return (columns, (utils.get_item_properties(
s, columns, formatters=self._formatters, )
for s in info), )
class ClearProbe(ProbeCommand):
"""Clear All probes."""
def take_action(self, parsed_args):
debug_agent = self.get_debug_agent()
cleared_probes_count = debug_agent.clear_probes()
self.log.info('%d probe(s) deleted', cleared_probes_count)
class ExecProbe(ProbeCommand):
"""Exec commands on the namespace of the probe."""
def get_parser(self, prog_name):
parser = super(ExecProbe, self).get_parser(prog_name)
parser.add_argument(
'id', metavar='port_id',
help=_('ID of probe port to execute command'))
parser.add_argument(
'command', metavar='command',
nargs='?',
default=None,
help=_('Command to execute'))
return parser
def take_action(self, parsed_args):
debug_agent = self.get_debug_agent()
result = debug_agent.exec_command(parsed_args.id, parsed_args.command)
self.app.stdout.write(result + '\n')
class PingAll(ProbeCommand):
"""Ping all fixed_ip."""
def get_parser(self, prog_name):
parser = super(PingAll, self).get_parser(prog_name)
parser.add_argument(
'--timeout', metavar='<timeout>',
default=10,
help=_('Ping timeout'))
parser.add_argument(
'--id', metavar='network_id',
default=None,
help=_('ID of network'))
return parser
def take_action(self, parsed_args):
debug_agent = self.get_debug_agent()
result = debug_agent.ping_all(parsed_args.id,
timeout=parsed_args.timeout)
self.app.stdout.write(result + '\n')

View File

@ -1,176 +0,0 @@
# Copyright 2012, Nachi Ueno, NTT MCL, 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 shlex
import socket
import netaddr
from neutron_lib.api.definitions import portbindings
from neutron_lib import constants
from oslo_log import log as logging
from neutron.agent.linux import dhcp
from neutron.agent.linux import ip_lib
LOG = logging.getLogger(__name__)
DEVICE_OWNER_NETWORK_PROBE = constants.DEVICE_OWNER_NETWORK_PREFIX + 'probe'
DEVICE_OWNER_COMPUTE_PROBE = constants.DEVICE_OWNER_COMPUTE_PREFIX + 'probe'
class NeutronDebugAgent(object):
def __init__(self, conf, client, driver):
self.conf = conf
self.client = client
self.driver = driver
def _get_namespace(self, port):
return "qprobe-%s" % port.id
def create_probe(self, network_id, device_owner='network'):
network = self._get_network(network_id)
port = self._create_port(network, device_owner)
interface_name = self.driver.get_device_name(port)
namespace = self._get_namespace(port)
if ip_lib.device_exists(interface_name, namespace=namespace):
LOG.debug('Reusing existing device: %s.', interface_name)
else:
self.driver.plug(network.id,
port.id,
interface_name,
port.mac_address,
namespace=namespace)
ip_cidrs = []
for fixed_ip in port.fixed_ips:
subnet = fixed_ip.subnet
net = netaddr.IPNetwork(subnet.cidr)
ip_cidr = '%s/%s' % (fixed_ip.ip_address, net.prefixlen)
ip_cidrs.append(ip_cidr)
self.driver.init_l3(interface_name, ip_cidrs, namespace=namespace)
return port
def _get_subnet(self, subnet_id):
subnet_dict = self.client.show_subnet(subnet_id)['subnet']
return dhcp.DictModel(subnet_dict)
def _get_network(self, network_id):
network_dict = self.client.show_network(network_id)['network']
network = dhcp.DictModel(network_dict)
# pylint: disable=assigning-non-slot
network.external = network_dict.get('router:external')
obj_subnet = [self._get_subnet(s_id) for s_id in network.subnets]
network.subnets = obj_subnet
return network
def clear_probes(self):
"""Returns number of deleted probes"""
ports = self.client.list_ports(
device_id=socket.gethostname(),
device_owner=[DEVICE_OWNER_NETWORK_PROBE,
DEVICE_OWNER_COMPUTE_PROBE])
info = ports['ports']
for port in info:
self.delete_probe(port['id'])
return len(info)
def delete_probe(self, port_id):
port = dhcp.DictModel(self.client.show_port(port_id)['port'])
namespace = self._get_namespace(port)
if ip_lib.network_namespace_exists(namespace):
self.driver.unplug(self.driver.get_device_name(port),
namespace=namespace)
try:
ip_lib.delete_network_namespace(namespace)
except Exception:
LOG.warning('Failed to delete namespace %s', namespace)
else:
self.driver.unplug(self.driver.get_device_name(port))
self.client.delete_port(port.id)
def list_probes(self):
ports = self.client.list_ports(
device_owner=[DEVICE_OWNER_NETWORK_PROBE,
DEVICE_OWNER_COMPUTE_PROBE])
info = ports['ports']
for port in info:
port['device_name'] = self.driver.get_device_name(
dhcp.DictModel(port))
return info
def exec_command(self, port_id, command=None):
port = dhcp.DictModel(self.client.show_port(port_id)['port'])
ip = ip_lib.IPWrapper()
namespace = self._get_namespace(port)
if not command:
return "sudo ip netns exec %s" % self._get_namespace(port)
namespace = ip.ensure_namespace(namespace)
# NOTE(ralonsoh): this is going to be called from inside the
# "neutron-debug" shell command; privsep is not configured.
return namespace.netns.execute(shlex.split(command))
def ensure_probe(self, network_id):
ports = self.client.list_ports(network_id=network_id,
device_id=socket.gethostname(),
device_owner=DEVICE_OWNER_NETWORK_PROBE)
info = ports.get('ports', [])
if info:
return dhcp.DictModel(info[0])
else:
return self.create_probe(network_id)
def ping_all(self, network_id=None, timeout=1):
if network_id:
ports = self.client.list_ports(network_id=network_id)['ports']
else:
ports = self.client.list_ports()['ports']
result = ""
for port in ports:
probe = self.ensure_probe(port['network_id'])
if port['device_owner'] == DEVICE_OWNER_NETWORK_PROBE:
continue
for fixed_ip in port['fixed_ips']:
address = fixed_ip['ip_address']
subnet = self._get_subnet(fixed_ip['subnet_id'])
if subnet.ip_version == 4:
ping_command = 'ping'
else:
ping_command = 'ping6'
result += self.exec_command(probe.id,
'%s -c 1 -w %s %s' % (ping_command,
timeout,
address))
return result
def _create_port(self, network, device_owner):
host = self.conf.host
body = {'port': {'admin_state_up': True,
'network_id': network.id,
'device_id': '%s' % socket.gethostname(),
'device_owner': '%s:probe' % device_owner,
'tenant_id': network.tenant_id,
portbindings.HOST_ID: host,
'fixed_ips': [dict(subnet_id=s.id)
for s in network.subnets]}}
port_dict = self.client.create_port(body)['port']
port = dhcp.DictModel(port_dict)
# pylint: disable=assigning-non-slot
port.network = network
for fixed_ip in port.fixed_ips:
fixed_ip.subnet = self._get_subnet(fixed_ip.subnet_id)
return port

View File

@ -1,92 +0,0 @@
# Copyright 2012, Nachi Ueno, NTT MCL, 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 sys
from oslo_config import cfg
from oslo_utils import importutils
from neutron._i18n import _
from neutron.agent.common import utils
from neutron.common import config as common_config
from neutron.conf.agent import common as config
from neutron.conf.plugins.ml2.drivers import ovs_conf
from neutron.debug import debug_agent
from neutronclient.common import exceptions as exc
from neutronclient import shell
COMMAND_V2 = {
'probe-create': importutils.import_class(
'neutron.debug.commands.CreateProbe'),
'probe-delete': importutils.import_class(
'neutron.debug.commands.DeleteProbe'),
'probe-list': importutils.import_class(
'neutron.debug.commands.ListProbe'),
'probe-clear': importutils.import_class(
'neutron.debug.commands.ClearProbe'),
'probe-exec': importutils.import_class(
'neutron.debug.commands.ExecProbe'),
'ping-all': importutils.import_class(
'neutron.debug.commands.PingAll'),
# TODO(nati) ping, netcat , nmap, bench
}
COMMANDS = {'2.0': COMMAND_V2}
class NeutronDebugShell(shell.NeutronShell):
def __init__(self, api_version):
super(NeutronDebugShell, self).__init__(api_version)
for k, v in COMMANDS[api_version].items():
self.command_manager.add_command(k, v)
def build_option_parser(self, description, version):
parser = super(NeutronDebugShell, self).build_option_parser(
description, version)
default = (
shell.env('NEUTRON_TEST_CONFIG_FILE') or
shell.env('QUANTUM_TEST_CONFIG_FILE')
)
parser.add_argument(
'--config-file',
default=default,
help=_('Config file for interface driver '
'(You may also use l3_agent.ini)'))
return parser
def initialize_app(self, argv):
super(NeutronDebugShell, self).initialize_app(argv)
if not self.options.config_file:
raise exc.CommandError(
_("You must provide a config file for bridge -"
" either --config-file or env[NEUTRON_TEST_CONFIG_FILE]"))
client = self.client_manager.neutron
config.register_interface_opts()
config.register_interface_driver_opts_helper(cfg.CONF)
ovs_conf.register_ovs_opts(cfg.CONF)
cfg.CONF(['--config-file', self.options.config_file])
config.setup_logging()
driver = utils.load_interface_driver(cfg.CONF)
self.debug_agent = debug_agent.NeutronDebugAgent(cfg.CONF,
client,
driver)
self.log.warning('This tool is deprecated and will be removed '
'in the future to be replaced with a more '
'powerful troubleshooting toolkit.')
def main(argv=None):
common_config.register_common_config_options()
return NeutronDebugShell(shell.NEUTRON_API_VERSION).run(
argv or sys.argv[1:])

View File

@ -1,325 +0,0 @@
# Copyright 2012, Nachi Ueno, NTT MCL, 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 socket
from unittest import mock
from neutron_lib.api.definitions import portbindings
from neutron_lib import constants
from oslo_config import cfg
from neutron.agent.linux import interface
from neutron.common import config as common_config
from neutron.conf.agent import common as config
from neutron.debug import commands
from neutron.debug import debug_agent
from neutron.tests import base
class MyApp(object):
def __init__(self, _stdout):
self.stdout = _stdout
class TestDebugCommands(base.BaseTestCase):
def setUp(self):
super(TestDebugCommands, self).setUp()
config.register_interface_opts()
common_config.init([])
config.register_interface_driver_opts_helper(cfg.CONF)
device_exists_p = mock.patch(
'neutron.agent.linux.ip_lib.device_exists', return_value=False)
device_exists_p.start()
namespace_e_p = mock.patch(
'neutron.agent.linux.ip_lib.network_namespace_exists')
namespace_e_p.start()
namespace_d_p = mock.patch(
'neutron.agent.linux.ip_lib.delete_network_namespace')
namespace_d_p.start()
ensure_namespace_p = mock.patch(
'neutron.agent.linux.ip_lib.IPWrapper.ensure_namespace')
ensure_namespace_p.start()
dvr_cls_p = mock.patch('neutron.agent.linux.interface.NullDriver')
driver_cls = dvr_cls_p.start()
mock_driver = mock.MagicMock()
mock_driver.DEV_NAME_LEN = (
interface.LinuxInterfaceDriver.DEV_NAME_LEN)
mock_driver.get_device_name.return_value = 'tap12345678-12'
driver_cls.return_value = mock_driver
self.driver = mock_driver
client_cls_p = mock.patch('neutronclient.v2_0.client.Client')
client_cls = client_cls_p.start()
client_inst = mock.Mock()
client_cls.return_value = client_inst
fake_network = {'network': {'id': 'fake_net',
'tenant_id': 'fake_tenant',
'subnets': ['fake_subnet']}}
fake_port = {'port':
{'id': 'fake_port',
'device_owner': 'fake_device',
'mac_address': 'aa:bb:cc:dd:ee:ffa',
'network_id': 'fake_net',
'fixed_ips':
[{'subnet_id': 'fake_subnet', 'ip_address': '10.0.0.3'}]
}}
fake_ports = {'ports': [fake_port['port']]}
self.fake_ports = fake_ports
allocation_pools = [{'start': '10.0.0.2',
'end': '10.0.0.254'}]
fake_subnet_v4 = {'subnet': {'name': 'fake_subnet_v4',
'id': 'fake_subnet',
'network_id': 'fake_net',
'gateway_ip': '10.0.0.1',
'dns_nameservers': ['10.0.0.2'],
'host_routes': [],
'cidr': '10.0.0.0/24',
'allocation_pools': allocation_pools,
'enable_dhcp': True,
'ip_version': constants.IP_VERSION_4}}
client_inst.list_ports.return_value = fake_ports
client_inst.create_port.return_value = fake_port
client_inst.show_port.return_value = fake_port
client_inst.show_network.return_value = fake_network
client_inst.show_subnet.return_value = fake_subnet_v4
self.client = client_inst
mock_std = mock.Mock()
self.app = MyApp(mock_std)
self.app.debug_agent = debug_agent.NeutronDebugAgent(cfg.CONF,
client_inst,
mock_driver)
def _test_create_probe(self, device_owner):
cmd = commands.CreateProbe(self.app, None)
cmd_parser = cmd.get_parser('create_probe')
if device_owner == debug_agent.DEVICE_OWNER_COMPUTE_PROBE:
args = ['fake_net', '--device-owner', 'compute']
else:
args = ['fake_net']
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
fake_port = {'port':
{'device_owner': device_owner,
'admin_state_up': True,
'network_id': 'fake_net',
'tenant_id': 'fake_tenant',
portbindings.HOST_ID: cfg.CONF.host,
'fixed_ips': [{'subnet_id': 'fake_subnet'}],
'device_id': socket.gethostname()}}
namespace = 'qprobe-fake_port'
self.client.assert_has_calls([mock.call.show_network('fake_net'),
mock.call.show_subnet('fake_subnet'),
mock.call.create_port(fake_port),
mock.call.show_subnet('fake_subnet')])
self.driver.assert_has_calls([mock.call.get_device_name(mock.ANY),
mock.call.plug('fake_net',
'fake_port',
'tap12345678-12',
'aa:bb:cc:dd:ee:ffa',
namespace=namespace),
mock.call.init_l3('tap12345678-12',
['10.0.0.3/24'],
namespace=namespace
)])
def test_create_network_probe(self):
self._test_create_probe(debug_agent.DEVICE_OWNER_NETWORK_PROBE)
def test_create_nova_probe(self):
self._test_create_probe(debug_agent.DEVICE_OWNER_COMPUTE_PROBE)
def _test_create_probe_external(self, device_owner):
fake_network = {'network': {'id': 'fake_net',
'tenant_id': 'fake_tenant',
'router:external': True,
'subnets': ['fake_subnet']}}
self.client.show_network.return_value = fake_network
cmd = commands.CreateProbe(self.app, None)
cmd_parser = cmd.get_parser('create_probe')
if device_owner == debug_agent.DEVICE_OWNER_COMPUTE_PROBE:
args = ['fake_net', '--device-owner', 'compute']
else:
args = ['fake_net']
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
fake_port = {'port':
{'device_owner': device_owner,
'admin_state_up': True,
'network_id': 'fake_net',
'tenant_id': 'fake_tenant',
portbindings.HOST_ID: cfg.CONF.host,
'fixed_ips': [{'subnet_id': 'fake_subnet'}],
'device_id': socket.gethostname()}}
namespace = 'qprobe-fake_port'
self.client.assert_has_calls([mock.call.show_network('fake_net'),
mock.call.show_subnet('fake_subnet'),
mock.call.create_port(fake_port),
mock.call.show_subnet('fake_subnet')])
self.driver.assert_has_calls([mock.call.get_device_name(mock.ANY),
mock.call.plug('fake_net',
'fake_port',
'tap12345678-12',
'aa:bb:cc:dd:ee:ffa',
namespace=namespace),
mock.call.init_l3('tap12345678-12',
['10.0.0.3/24'],
namespace=namespace
)])
def test_create_network_probe_external(self):
self._test_create_probe_external(
debug_agent.DEVICE_OWNER_NETWORK_PROBE)
def test_create_nova_probe_external(self):
self._test_create_probe_external(
debug_agent.DEVICE_OWNER_COMPUTE_PROBE)
def test_delete_probe(self):
cmd = commands.DeleteProbe(self.app, None)
cmd_parser = cmd.get_parser('delete_probe')
args = ['fake_port']
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
namespace = 'qprobe-fake_port'
self.client.assert_has_calls([mock.call.show_port('fake_port'),
mock.call.delete_port('fake_port')])
self.driver.assert_has_calls([mock.call.get_device_name(mock.ANY),
mock.call.unplug('tap12345678-12',
namespace=namespace)])
def test_delete_probe_external(self):
fake_network = {'network': {'id': 'fake_net',
'tenant_id': 'fake_tenant',
'router:external': True,
'subnets': ['fake_subnet']}}
self.client.show_network.return_value = fake_network
cmd = commands.DeleteProbe(self.app, None)
cmd_parser = cmd.get_parser('delete_probe')
args = ['fake_port']
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
namespace = 'qprobe-fake_port'
self.client.assert_has_calls([mock.call.show_port('fake_port'),
mock.call.delete_port('fake_port')])
self.driver.assert_has_calls([mock.call.get_device_name(mock.ANY),
mock.call.unplug('tap12345678-12',
namespace=namespace)])
def test_list_probe(self):
cmd = commands.ListProbe(self.app, None)
cmd_parser = cmd.get_parser('list_probe')
args = []
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
self.client.assert_has_calls(
[mock.call.list_ports(
device_owner=[debug_agent.DEVICE_OWNER_NETWORK_PROBE,
debug_agent.DEVICE_OWNER_COMPUTE_PROBE])])
def test_exec_command(self):
cmd = commands.ExecProbe(self.app, None)
cmd_parser = cmd.get_parser('exec_command')
args = ['fake_port', 'fake_command']
parsed_args = cmd_parser.parse_args(args)
with mock.patch('neutron.agent.linux.ip_lib.IpNetnsCommand') as ns:
cmd.run(parsed_args)
ns.assert_has_calls([mock.call.execute(mock.ANY)])
self.client.assert_has_calls([mock.call.show_port('fake_port')])
def test_clear_probe(self):
cmd = commands.ClearProbe(self.app, None)
cmd_parser = cmd.get_parser('clear_probe')
args = []
parsed_args = cmd_parser.parse_args(args)
cmd.run(parsed_args)
namespace = 'qprobe-fake_port'
self.client.assert_has_calls(
[mock.call.list_ports(
device_id=socket.gethostname(),
device_owner=[debug_agent.DEVICE_OWNER_NETWORK_PROBE,
debug_agent.DEVICE_OWNER_COMPUTE_PROBE]),
mock.call.show_port('fake_port'),
mock.call.delete_port('fake_port')])
self.driver.assert_has_calls([mock.call.get_device_name(mock.ANY),
mock.call.unplug('tap12345678-12',
namespace=namespace)])
def test_ping_all_with_ensure_port(self):
fake_ports = self.fake_ports
def fake_port_list(network_id=None, device_owner=None, device_id=None):
if network_id:
# In order to test ensure_port, return []
return {'ports': []}
return fake_ports
self.client.list_ports.side_effect = fake_port_list
cmd = commands.PingAll(self.app, None)
cmd_parser = cmd.get_parser('ping_all')
args = []
parsed_args = cmd_parser.parse_args(args)
namespace = 'qprobe-fake_port'
with mock.patch('neutron.agent.linux.ip_lib.IpNetnsCommand') as ns:
cmd.run(parsed_args)
ns.assert_has_calls([mock.call.execute(mock.ANY)])
fake_port = {'port':
{'device_owner': debug_agent.DEVICE_OWNER_NETWORK_PROBE,
'admin_state_up': True,
'network_id': 'fake_net',
'tenant_id': 'fake_tenant',
portbindings.HOST_ID: cfg.CONF.host,
'fixed_ips': [{'subnet_id': 'fake_subnet'}],
'device_id': socket.gethostname()}}
expected = [mock.call.show_network('fake_net'),
mock.call.show_subnet('fake_subnet'),
mock.call.create_port(fake_port),
mock.call.show_subnet('fake_subnet')]
self.client.assert_has_calls(expected)
self.driver.assert_has_calls([mock.call.init_l3('tap12345678-12',
['10.0.0.3/24'],
namespace=namespace
)])
def test_ping_all(self):
cmd = commands.PingAll(self.app, None)
cmd_parser = cmd.get_parser('ping_all')
args = []
parsed_args = cmd_parser.parse_args(args)
with mock.patch('neutron.agent.linux.ip_lib.IpNetnsCommand') as ns:
cmd.run(parsed_args)
ns.assert_has_calls([mock.call.execute(mock.ANY)])
expected = [mock.call.list_ports(),
mock.call.list_ports(
network_id='fake_net',
device_owner=debug_agent.DEVICE_OWNER_NETWORK_PROBE,
device_id=socket.gethostname()),
mock.call.show_subnet('fake_subnet'),
mock.call.show_port('fake_port')]
self.client.assert_has_calls(expected)
def test_ping_all_v6(self):
fake_subnet_v6 = {'subnet': {'name': 'fake_v6',
'ip_version': constants.IP_VERSION_6}}
self.client.show_subnet.return_value = fake_subnet_v6
cmd = commands.PingAll(self.app, None)
cmd_parser = cmd.get_parser('ping_all')
args = []
parsed_args = cmd_parser.parse_args(args)
with mock.patch('neutron.agent.linux.ip_lib.IpNetnsCommand') as ns:
cmd.run(parsed_args)
ns.assert_has_calls([mock.call.execute(mock.ANY)])
self.client.assert_has_calls([mock.call.list_ports()])

View File

@ -0,0 +1,6 @@
---
deprecations:
- |
The tool neutron-debug is now removed. With removal of the neutron client
shell code this tool is no longer usable. It had been marked for
deprecation since the Newton (9.0) cycle and unmaintained.

View File

@ -35,7 +35,6 @@ wsgi_scripts =
neutron-api = neutron.cmd.eventlet.server:main_api_eventlet
console_scripts =
neutron-db-manage = neutron.db.migration.cli:main
neutron-debug = neutron.debug.shell:main
neutron-dhcp-agent = neutron.cmd.eventlet.agents.dhcp:main
neutron-keepalived-state-change = neutron.cmd.keepalived_state_change:main
neutron-ipset-cleanup = neutron.cmd.ipset_cleanup:main