x/whitebox-neutron-tempest-plugin
Code Issues Proposed changes
Files
master
whitebox-neutron-tempest-pl…/whitebox_neutron_tempest_plugin/tests/scenario/test_qos.py
Maor Blaustein 592f6dc961 Workaround Nova bug to fix test_qos_after_live_migration coverage 
This relates to the `test_qos_after_live_migration` failure on CI jobs,
other tests running live migration do pass such as:
`tempest.scenario.test_network_advanced_server_ops.TestNetworkAdvancedServerOps.test_server_connectivity_live_migration`.

If I understand correctly, this is the launchpad related from recent
discussion [1] opened a year ago, and still no assignee from nova.

Test attaches second interface for control on tenant VMs [2],
from reading about this nova bug, this follow-up test change will avoid
the issue by tenant VM reboot before live migration (renews MTU info),
so we'll still have full coverage of QoS after VM live migrate tested.

(Reproduced original issue and verified its fix on a podified setup,
both cold/live migration pass with change).

Notice that reverting this workaround will verify/add coverage for
Nova launchpad when fix available [1].

Resolves-Bug: OSPRH-19571
Related-Bug: #2080531

[1]
live migration fails if multiple networks exist on instance
https://bugs.launchpad.net/nova/+bug/2080531

[2]
957966: Control interface on tenant VMs for all QoS bandwidth tests | https://review.opendev.org/c/x/whitebox-neutron-tempest-plugin/+/957966

Signed-off-by: Maor Blaustein <mblue@redhat.com>
Change-Id: I51774b46d6918ca8abc4283f91ca759fbf2e86e2
2025-09-01 20:56:24 +03:00

1714 lines
77 KiB
Python
Raw Permalink Blame History

# Copyright 2024 Red Hat, 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 json
import random
import time
import netaddr
import testtools
from neutron_lib.api.definitions import qos as qos_apidef
from neutron_lib import constants
from neutron_tempest_plugin.common import ssh
from neutron_tempest_plugin import exceptions as neutron_exceptions
from neutron_tempest_plugin.scenario import test_qos
from oslo_log import log
from tempest.common import waiters
from tempest import config
from tempest import exceptions as tempest_exceptions
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions
from whitebox_neutron_tempest_plugin.common import tcpdump_capture as capture
from whitebox_neutron_tempest_plugin.common import utils
from whitebox_neutron_tempest_plugin.tests.scenario import base
try:
from tempest.lib.common import api_microversion_fixture
from tempest.lib.common import api_version_utils
except ImportError:
pass
CONF = config.CONF
WB_CONF = CONF.whitebox_neutron_plugin_options
LOG = log.getLogger(__name__)
class QosBaseTest(test_qos.QoSTestMixin, base.TrafficFlowTest):
required_extensions = [qos_apidef.ALIAS]
credentials = ['primary', 'admin']
dscp_mark_net = 56
dscp_mark_net_new = 22
dscp_mark_port = 48
dscp_mark_fip = 36
dscp_mark_fip_new = 38
dscp_mark_gw = 10
dscp_mark_gw_new = 12
MIN_KBPS_NO_BWLIMIT = 10000
IPERF_PORT = 4321
bwlimit_kbps_net = 1000
bwlimit_kbps_net_new = 500
bwlimit_kbps_port = 250
sriov_test = False
@classmethod
def skip_checks(cls):
super(QosBaseTest, cls).skip_checks()
advanced_image_available = (
CONF.neutron_plugin_options.advanced_image_ref or
CONF.neutron_plugin_options.default_image_is_advanced)
if not advanced_image_available:
skip_reason = "This test require advanced tools for this test"
raise cls.skipException(skip_reason)
@classmethod
def setup_clients(cls):
super(QosBaseTest, cls).setup_clients()
cls.admin_client = cls.os_admin.network_client
cls.qos_bw_limit_rule_client = \
cls.os_admin.qos_limit_bandwidth_rules_client
cls.qos_bw_limit_rule_client_primary = \
cls.os_primary.qos_limit_bandwidth_rules_client
cls.qos_min_bw_rules_client = \
cls.os_admin.qos_minimum_bandwidth_rules_client
cls.qos_min_bw_rules_client_primary = \
cls.os_primary.qos_minimum_bandwidth_rules_client
@classmethod
def resource_setup(cls):
super(QosBaseTest, cls).resource_setup()
if not hasattr(cls, 'nodes'):
raise cls.skipException(
"Nodes info not available. Test won't be able to connect to "
"nodes.")
msg = "Required QoS config is not set"
if cls.is_devstack:
for node in cls.nodes:
if node['is_controller'] is False:
continue
cls.check_service_setting(
host=node, service='',
config_files=[cls.neutron_conf],
param='service_plugins', value='qos')
cls.check_service_setting(
host=node, service='',
config_files=[WB_CONF.ml2_plugin_config], section='ml2',
param='extension_drivers', value='qos')
else:
config_files = cls.get_configs_of_service('neutron')
cls.check_service_setting(
{'client': cls.proxy_host_client}, service='neutron',
config_files=config_files, section='DEFAULT',
param='service_plugins', value='qos', msg=msg)
cls.check_service_setting(
{'client': cls.proxy_host_client}, service='neutron',
config_files=config_files, section='ml2',
param='extension_drivers', value='qos', msg=msg)
if CONF.neutron_plugin_options.default_image_is_advanced:
cls.flavor_ref = CONF.compute.flavor_ref
cls.image_ref = CONF.compute.image_ref
cls.username = CONF.validation.image_ssh_user
else:
cls.flavor_ref = (
CONF.neutron_plugin_options.advanced_image_flavor_ref)
cls.image_ref = CONF.neutron_plugin_options.advanced_image_ref
cls.username = CONF.neutron_plugin_options.advanced_image_ssh_user
# Reduce initial rate for ovs virt envs to avoid instabilities
if not cls.has_sriov_support and not cls.has_ovn_support:
cls.MIN_KBPS_NO_BWLIMIT = 600
cls.bwlimit_kbps_net = 400
cls.bwlimit_kbps_net_new = 250
cls.bwlimit_kbps_port = 125
cls.router = cls.create_router_by_client()
cls.keypair = cls.create_keypair()
cls.secgroup = cls.os_primary.network_client.create_security_group(
name=data_utils.rand_name('secgroup_qos'))
cls.security_groups.append(cls.secgroup['security_group'])
cls.create_loginable_secgroup_rule(
secgroup_id=cls.secgroup['security_group']['id'])
cls.create_pingable_secgroup_rule(
secgroup_id=cls.secgroup['security_group']['id'])
# create security group rules for icmpv6
rulesets = [{'protocol': constants.PROTO_NAME_IPV6_ICMP,
'ethertype': 'IPv6',
'direction': 'ingress'}]
# create security group rules for the bw limit tests both udp and tcp
# ports need to be accessible and both for IPv4 and IPv6
for proto in (constants.PROTO_NAME_TCP, constants.PROTO_NAME_UDP):
for ipversion in ('IPV4', 'IPV6'):
rulesets.append({'protocol': proto,
'ethertype': ipversion,
'port_range_min': cls.IPERF_PORT,
'port_range_max': cls.IPERF_PORT,
'direction': 'ingress'})
cls.create_secgroup_rules(
rulesets, cls.secgroup['security_group']['id'])
# Use separate resources not affected by QoS bw limits, in order to
# to control tenant VMs such as ones used for client/server iperf
ctl_ptn = data_utils.rand_name('tenant-ctl-%s')
cls.ctl_secgroup = cls.create_security_group(
name=ctl_ptn % 'secgroup')
cls.security_groups.append(cls.ctl_secgroup)
cls.create_loginable_secgroup_rule(
secgroup_id=cls.ctl_secgroup['id'])
cls.create_pingable_secgroup_rule(
secgroup_id=cls.ctl_secgroup['id'])
cls.ctl_network = cls.create_network(name=ctl_ptn % 'network')
cls.ctl_subnet = cls.create_subnet(
network=cls.ctl_network, name=ctl_ptn % 'subnet')
cls.ctl_router = cls.create_router_by_client()
cls.create_router_interface(
cls.ctl_router['id'], cls.ctl_subnet['id'])
def _ensure_ctl_fip_for_vm(self, vm):
"""Create control FIP through additional interface/port on tenant VM
(if doesn't exist already).
Mostly needed for bandwidth limit tests.
"""
if 'ctl_fip' in vm:
return
port = self.create_port(
self.ctl_network,
qos_policy_id=None,
security_groups=[self.ctl_secgroup['id']])
self.create_interface(vm['id'], port['id'])
waiters.wait_for_interface_status(
self.os_primary.interfaces_client, vm['id'],
port['id'], constants.PORT_STATUS_ACTIVE)
utils.configure_interface_up(vm['ssh_client'], port)
vm['ctl_fip'] = self.create_floatingip(port=port)
vm['ctl_ssh'] = ssh.Client(
vm['ctl_fip']['floating_ip_address'],
self.username,
pkey=self.keypair['private_key'])
# NOTE(mblue): consider explicit symmetric and strict (1) setting
# for test nic, custom image defaults to off (0), so
# `max(all, per-iface)` set to 1, also first nic is set default route
# force tenant symmetric route for control interface
self.tenant_sym_path_for_port(vm['ssh_client'], port)
# TODO(mblue): fix when new NTP changes available for rdo gate,
# since NTP commit 2d663554 missing.
self.check_connectivity(host=None, ssh_client=vm['ctl_ssh'])
@staticmethod
def _get_iperf_proto_param(protocol):
if protocol == constants.PROTO_NAME_TCP:
params = ''
if WB_CONF.window_size:
params = '-w {}'.format(WB_CONF.window_size)
return params
elif protocol == constants.PROTO_NAME_UDP:
return '-u'
else:
raise ValueError('Unsupported protocol %s' % protocol)
def _test_egress_bw(
self, ssh_client, ssh_server, client_ip, server_ip, protocol,
maxbitrate, timeout=6):
utils.kill_iperf_process(ssh_server)
utils.kill_iperf_process(ssh_client)
iperf_server_filename = utils.get_temp_file(ssh_server)
server_cmd = (
f'iperf3 -s -B {server_ip} -p {self.IPERF_PORT} -J --logfile '
f'{iperf_server_filename} -D')
LOG.debug('Run iperf3 command on server: %s', server_cmd)
ssh_server.exec_command(server_cmd)
time.sleep(0.1)
protocol_param = self._get_iperf_proto_param(protocol)
client_cmd = (
f'iperf3 -c {server_ip} -B {client_ip} -p {self.IPERF_PORT} '
f'{protocol_param} -t {timeout} -b {maxbitrate}')
LOG.debug('Run iperf3 command on client: %s', client_cmd)
ssh_client.exec_command(client_cmd)
time.sleep(0.1)
return json.loads(utils.cat_remote_file(
ssh_server, iperf_server_filename))
def _test_ingress_bw(
self, ssh_client, ssh_server, client_ip, server_ip, protocol,
maxbitrate, timeout=6):
utils.kill_iperf_process(ssh_server)
utils.kill_iperf_process(ssh_client)
server_cmd = f'iperf3 -s -B {server_ip} -p {self.IPERF_PORT} -D'
LOG.debug('Run iperf3 command on server: %s', server_cmd)
ssh_server.exec_command(server_cmd)
time.sleep(0.1)
iperf_client_filename = utils.get_temp_file(ssh_client)
protocol_param = self._get_iperf_proto_param(protocol)
client_cmd = (
f'iperf3 -c {server_ip} -B {client_ip} -p {self.IPERF_PORT} '
f'{protocol_param} -t {timeout} -b {maxbitrate} '
f'-R -J > {iperf_client_filename}')
LOG.debug('Run iperf3 command on client: %s', client_cmd)
ssh_client.exec_command(client_cmd)
time.sleep(0.1)
return json.loads(utils.cat_remote_file(
ssh_client, iperf_client_filename))
def _calculate_bw(self, perf_measures):
# For rocky images, final interval is ignored
# TODO(eolivare): provide link to iperf/rocky bug
intervals_end = (len(perf_measures['intervals'])
if self.username != "rocky"
else len(perf_measures['intervals']) - 1)
intervals = perf_measures['intervals'][:intervals_end]
# First 3 for ovs envs, 2 for ovn sriov envs and 1 for normal ovn
# intervals are removed because BW measured during it is not
# limited - it takes ~2-4 seconds to traffic shaping algorithm to apply
# several tests in parallel increase this application time
# bw limit properly (buffer is empty when traffic starts being sent)
if not self.has_ovn_support:
intervals_start = 3
elif self.sriov_test:
intervals_start = 2
else:
intervals_start = 1
if len(intervals) > intervals_start + 1:
intervals = intervals[intervals_start:]
bits_received = sum([interval['sum']['bytes'] * 8
for interval in intervals])
totaltime = sum([interval['sum']['seconds'] for interval in intervals])
# bw in bits per second
return bits_received / totaltime
def _skip_if_iperf3_not_installed(self, ssh_client):
try:
ssh_client.execute_script('PATH=$PATH:/usr/sbin which iperf3')
except neutron_exceptions.SSHScriptFailed:
raise self.skipException(
"iperf3 is not available on VM instance")
def _get_tenant_port_by_type(self, vm_id, _type, ignore_port_ids=None):
# bw tests: iperf3 tenant VMs assumed to only have control/test ports,
# special cases may use ignore_port_ids.
assert _type in ('control', 'test')
vm_ports = self.admin_client.list_ports(device_id=vm_id)['ports']
for port in vm_ports:
if ignore_port_ids and port['id'] in ignore_port_ids:
LOG.debug('Ignored port %s (found in %s ignore list)',
port['id'], ignore_port_ids)
continue
for fixed_ip in port['fixed_ips']:
if _type == 'control':
found = fixed_ip['subnet_id'] == self.ctl_subnet['id']
else:
found = fixed_ip['subnet_id'] != self.ctl_subnet['id']
if found:
LOG.debug(
"%s port %s (related ip %s) found for tenant vm %s",
_type, port['id'], fixed_ip['ip_address'], vm_id)
return port
LOG.warning("%s port not found for tenant vm %s (ignore list %s)",
_type, vm_id, ignore_port_ids)
# retry only when noticed measuring issue, as reported in BZ#2274465
@utils.retry_on_assert_fail(
max_retries=2,
assert_regex='not .* than')
def _validate_bw_limit(self, client, server, egress=True, ipv6=False,
bw_limit=None, protocol=constants.PROTO_NAME_TCP,
ignore_port_ids=None):
# control port/fip, per server/client tenant vms of iperf
self._ensure_ctl_fip_for_vm(client)
self._ensure_ctl_fip_for_vm(server)
# NOTE(mblue): iperf3 needs the soon to be overwhelmed test port,
# while control port remains for ssh (and any non bw limited traffic)
# ignore list currently for a single OVN test using 3 ports on vm
server_port = self._get_tenant_port_by_type(
server['id'], 'test', ignore_port_ids=ignore_port_ids)
server_ips = []
for fixed_ip in server_port['fixed_ips']:
if netaddr.valid_ipv6(fixed_ip['ip_address']) and not ipv6:
continue
server_ips.append(fixed_ip['ip_address'])
self.assertGreater(len(server_ips), 0)
client_ip = self._get_tenant_port_by_type(
client['id'], 'test', ignore_port_ids=ignore_port_ids
)['fixed_ips'][0]['ip_address']
if egress:
test_bw_method = self._test_egress_bw
direction = 'egress'
else:
test_bw_method = self._test_ingress_bw
direction = 'ingress'
# NOTE: with traffic shaping, if the bitrate exceeds the BW limit, the
# excees traffic is buffered and sent once the iperf3 client has ended
# sending traffic, but this could lead to an iperf error:
# "control socket has closed unexpectedly"
# due to this we limit the bitrate to 1.5 * bw_limit
maxbitrate = (bw_limit * 1.5 if bw_limit is not None
else self.MIN_KBPS_NO_BWLIMIT * 2000)
# TODO(mblue): all server ips tested, but not all client ips, nested
# loop to test all server/client ip possibilities would add test
# time, consider if needed (ex: ipv6 to ipv6 not tested, etc).
# egress: send payload from client to server
# ingress: download payload from server to client
for server_ip in server_ips:
perf_measures = test_bw_method(
client['ctl_ssh'], server['ctl_ssh'],
client_ip, server_ip, protocol, maxbitrate=maxbitrate)
LOG.debug('perf_measures = %s', perf_measures)
# verify bw limit
measured_bw = self._calculate_bw(perf_measures)
LOG.debug(
'%s %s / server_ip = %s , client_ip = %s / measured_bw = %f',
direction, protocol, server_ip, client_ip, measured_bw)
if bw_limit is None:
LOG.debug('no %s bw_limit configured', direction)
self.assertGreater(measured_bw,
self.MIN_KBPS_NO_BWLIMIT * 1000)
else:
LOG.debug('%s bw_limit = %s', direction, bw_limit)
# a 20% of upper deviation is allowed
self.assertLess(measured_bw, bw_limit * 1.2)
# a 20% of lower deviation is allowed
self.assertGreater(measured_bw, bw_limit * 0.8)
def _validate_traffic_bw_limit(self, client, server, egress=True,
ipv6=False, fip_qos=False, gw_qos=False):
"""Validate that bw limit is applied to the traffic between client and
server VMs.
Precedence summary for max-bw limit:
- only port: then max-bw port
- port + gw: minimum max-bw
- port + gw/fip: minimum max-bw of port or FIP
(GW is discarded because FIP has precedence)
Scenario:
1. First make sure that bw between VMs is not limited.
2. Create a QoS policy, attach to the network where client VM
is connected to.
3. Add a bw limit rule to the policy.
4. Send traffic between 2 VMs and make sure that now bw limit
is applied.
5. Update the bw limit rule with a different value and make
sure that bw is limited accordingly.
6. Create another QoS policy, apply to the port of client VM
and add a bw limit rule with another value.
7. Send traffic between 2 VMs and make sure that now bw is
limited according to the new value (since rule for port has higher
priority).
8. Create another QoS policy applied to GW,
verify together with port QoS when lower/higher.
9. Delete bandwidth rule from the router gateway QoS policy
and make sure that traffic is limited for port value - step 12
(unless FIP checked, then rule kept to verify when all 3 used).
10. Create another QoS policy applied to FIP, verify
FIP and port lower/higher (if enabled also GW higher),
also verify when GW lower than FIP/port - lowest between FIP/port
prioritized (priority - A: port > FIP > GW, B: FIP > port > GW).
11. Delete bandwidth rule from FIP QoS policy,
and make sure that traffic is limited for GW value.
12. Delete bandwidth rule from GW QoS policy,
and make sure that traffic is limited for port value again.
13. Delete bandwidth limit rule from the port QoS policy and make
sure that traffic is not limited.
14. Verify a QoS policy attached to a port cannot be deleted.
"""
# Prerequisite: install iperf3
self._skip_if_iperf3_not_installed(client['ssh_client'])
self._skip_if_iperf3_not_installed(server['ssh_client'])
# First, let's make sure that bw is not limited
self._validate_bw_limit(client, server, egress)
if self.sriov_test:
# TODO(eolivare): investigate why this delay between iperf tests is
# needed when SRIOV ports are used
time.sleep(5)
self._validate_bw_limit(client, server, egress,
protocol=constants.PROTO_NAME_UDP)
direction = 'egress' if egress else 'ingress'
# Create new QoS policy and attach to the src test network
net_bwlimit_policy_id = self._create_qos_policy()
src_port = self._get_tenant_port_by_type(client['id'], 'test')
self.admin_client.update_network(
src_port['network_id'], qos_policy_id=net_bwlimit_policy_id)
self.addCleanup(self.admin_client.update_network,
src_port['network_id'], qos_policy_id=None)
max_kbps = (self.bwlimit_kbps_net
if src_port['binding:vnic_type'] == 'normal'
else self.bwlimit_kbps_net * 1000)
rule_data = {
'max_kbps': max_kbps,
'max_burst_kbps': max_kbps * 0.8,
'direction': direction}
net_rule_id = self._create_qos_bw_limit_rule(
net_bwlimit_policy_id, rule_data)['id']
# Validate the bw limit - using UDP
self._validate_bw_limit(
client, server, egress, bw_limit=max_kbps * 1000,
protocol=constants.PROTO_NAME_UDP)
# Update the bw limit rule and verify that new bw limit is applied
max_kbps = (self.bwlimit_kbps_net_new
if src_port['binding:vnic_type'] == 'normal'
else self.bwlimit_kbps_net_new * 1000)
rule_update_data = {
'max_kbps': max_kbps,
'max_burst_kbps': max_kbps * 0.8}
self.qos_bw_limit_rule_client.update_limit_bandwidth_rule(
qos_policy_id=net_bwlimit_policy_id, rule_id=net_rule_id,
**rule_update_data)
# Validate the bw limit - using TCP
self._validate_bw_limit(client, server, egress,
bw_limit=max_kbps * 1000)
# Create a new QoS policy and attach to the port of src server
bwlimit_policy_id_new = self._create_qos_policy()
max_kbps = (self.bwlimit_kbps_port
if src_port['binding:vnic_type'] == 'normal'
else self.bwlimit_kbps_port * 1000)
rule_data = {
'max_kbps': max_kbps,
'max_burst_kbps': max_kbps * 0.8,
'direction': direction}
port_rule_id = self._create_qos_bw_limit_rule(
bwlimit_policy_id_new, rule_data)['id']
self.client.update_port(
src_port['id'], qos_policy_id=bwlimit_policy_id_new)
self.addCleanup(self.admin_client.update_port,
src_port['id'], qos_policy_id=None)
# Verify that bw limit applied corresponds with the port QoS policy
# Validate the bw limit - using UDP
self._validate_bw_limit(
client, server, egress, bw_limit=max_kbps * 1000,
protocol=constants.PROTO_NAME_UDP)
# NOTE(mblue): only enabled in north-south topology option when
# routed through gateways.
if gw_qos:
# apply policy for router gateway (no rule in policy yet)
gw_bw_policy_id = self._create_qos_policy()
self.router['external_gateway_info'][
'qos_policy_id'] = gw_bw_policy_id
self.router = self.admin_client.update_router(
self.router['id'],
external_gateway_info=self.router['external_gateway_info']
)['router']
del self.router['external_gateway_info']['qos_policy_id']
self.addCleanup(
self.admin_client.update_router,
self.router['id'],
external_gateway_info=self.router['external_gateway_info'])
# gw bw limit greater than port bw limit
gw_max_kbps = max_kbps * 20
rule_data = {
'max_kbps': gw_max_kbps,
'max_burst_kbps': gw_max_kbps * 0.8,
'direction': direction}
gw_port_rule_id = self._create_qos_bw_limit_rule(
gw_bw_policy_id, rule_data)['id']
# port bw limit is lower than gw bw limit, so max_kbps
# will be the measured bw
self._validate_bw_limit(
client, server, egress, bw_limit=max_kbps * 1000,
protocol=constants.PROTO_NAME_UDP)
# gw bw limit lower than port bw limit
gw_max_kbps = max_kbps // 2
rule_update_data = {
'max_kbps': gw_max_kbps,
'max_burst_kbps': gw_max_kbps * 0.8}
self.qos_bw_limit_rule_client.update_limit_bandwidth_rule(
qos_policy_id=gw_bw_policy_id, rule_id=gw_port_rule_id,
**rule_update_data)
# For rocky images, running iperf tests with low BW limits using
# TCP does not work well, wo UDP is used instead
# TODO(eolivare): provide link to iperf/rocky bug
protocol = (constants.PROTO_NAME_TCP
if self.username != "rocky"
else constants.PROTO_NAME_UDP)
# gw bw limit is lower than port bw limit, so gw_max_kbps
# will be the measured bw
self._validate_bw_limit(client, server, egress,
bw_limit=gw_max_kbps * 1000,
protocol=protocol)
# delete bw limit rule associated to gw qos policy,
# port bw limit applies again
# (unless FIP checked, then keep GW rule to also test
# FIP/GW precedence when each is lower).
# NOTE(mblue): port validation happens anyways at end of method.
if not fip_qos:
self.qos_bw_limit_rule_client.delete_limit_bandwidth_rule(
gw_bw_policy_id, gw_port_rule_id)
self._validate_bw_limit(client, server, egress,
bw_limit=max_kbps * 1000,
protocol=protocol)
else:
# gw bw limit greater than port/fip bw limit
gw_max_kbps = max_kbps * 20
rule_update_data = {
'max_kbps': gw_max_kbps,
'max_burst_kbps': gw_max_kbps * 0.8,
'direction': direction}
self.qos_bw_limit_rule_client.update_limit_bandwidth_rule(
qos_policy_id=gw_bw_policy_id, rule_id=gw_port_rule_id,
**rule_update_data)
# Create a new Qos Policy and attach to the FIP of src server
# This only applies to south-north tests because the traffic from the
# src server to the dst server goes through the src FIP
if fip_qos:
fip_qos_pol_id = self._create_qos_policy()
# fip bw limit greater than port bw limit
fip_max_kbps = max_kbps * 10
rule_data = {
'max_kbps': fip_max_kbps,
'max_burst_kbps': fip_max_kbps * 0.8,
'direction': direction}
fip_port_rule_id = self._create_qos_bw_limit_rule(
fip_qos_pol_id, rule_data)['id']
src_fip_id = self.client.list_floatingips(
fixed_ip_address=src_port['fixed_ips'][0]['ip_address'])[
'floatingips'][0]['id']
self.client.update_floatingip(
src_fip_id, qos_policy_id=fip_qos_pol_id)
self.addCleanup(self.admin_client.update_floatingip,
src_fip_id, qos_policy_id=None)
# port bw limit is lower than fip (sometimes also gw) bw limit,
# so max_kbps will be the measured bw
self._validate_bw_limit(
client, server, egress, bw_limit=max_kbps * 1000,
protocol=constants.PROTO_NAME_UDP)
# only if gw_qos True, test when gw lower than fip and port:
# expected - fip prioritzied over gw (LP#2110018),
# and port limit applied before reaching this datapath part.
if gw_qos:
gw_max_kbps = max_kbps // 2
rule_update_data = {
'max_kbps': gw_max_kbps,
'max_burst_kbps': gw_max_kbps * 0.8,
'direction': direction}
self.qos_bw_limit_rule_client.update_limit_bandwidth_rule(
qos_policy_id=gw_bw_policy_id, rule_id=gw_port_rule_id,
**rule_update_data)
# port limit expected
self._validate_bw_limit(client, server, egress,
bw_limit=max_kbps * 1000,
protocol=protocol)
# update fip bw limit lower than port bw limit
# (also lower than gw limit if exists)
fip_max_kbps = max_kbps // 3
rule_update_data = {
'max_kbps': fip_max_kbps,
'max_burst_kbps': fip_max_kbps * 0.8}
self.qos_bw_limit_rule_client.update_limit_bandwidth_rule(
qos_policy_id=fip_qos_pol_id, rule_id=fip_port_rule_id,
**rule_update_data)
# For rocky images, running iperf tests with low BW limits using
# TCP does not work well, wo UDP is used instead
# TODO(eolivare): provide link to iperf/rocky bug
protocol = (constants.PROTO_NAME_TCP
if self.username != "rocky"
else constants.PROTO_NAME_UDP)
# fip bw limit is lower than port/gw bw limit, so fip_max_kbps
# will be the measured bw
self._validate_bw_limit(client, server, egress,
bw_limit=fip_max_kbps * 1000,
protocol=protocol)
# delete bw limit rule associated to fip qos policy,
# so either port/gw bw limit applies again
self.qos_bw_limit_rule_client.delete_limit_bandwidth_rule(
fip_qos_pol_id, fip_port_rule_id)
# only for GW: test lower GW prioritized when with port (no FIP),
# delete GW rule for next checks
if gw_qos:
self._validate_bw_limit(client, server, egress,
bw_limit=gw_max_kbps * 1000,
protocol=protocol)
self.qos_bw_limit_rule_client.delete_limit_bandwidth_rule(
gw_bw_policy_id, gw_port_rule_id)
# validate port limit applies again
self._validate_bw_limit(client, server, egress,
bw_limit=max_kbps * 1000,
protocol=protocol)
# Delete bw limit rule from the port QoS policy and validate that
# bw is not limited anymore
self.qos_bw_limit_rule_client.delete_limit_bandwidth_rule(
bwlimit_policy_id_new, port_rule_id)
# Validate the bw limit - using TCP
self._validate_bw_limit(client, server, egress)
# Verify a QoS policy attached to a port cannot be deleted
try:
self.admin_client.delete_qos_policy(bwlimit_policy_id_new)
except exceptions.Conflict:
LOG.debug('QoS Policy %s cannot be deleted because '
'it is attached to a port', bwlimit_policy_id_new)
else:
raise RuntimeError('Deletion of QoS Policy %s should '
'have been rejected' % bwlimit_policy_id_new)
# remove QoS policies from port and network
self.admin_client.update_port(src_port['id'], qos_policy_id=None)
self.admin_client.update_network(
src_port['network_id'], qos_policy_id=None)
def _get_dscp_filters(
self, src_port, dst_ip, mark, outer=False, tunnel_type=None,
segmentation_id=None):
# IPv4 tcpdump filters explanation:
# ip[1] refers to the byte 1 (the TOS byte) of the IP header
# 0xfc = 11111100 is the mask to get only DSCP value from the ToS
# As DSCP mark is most significant 6 bits we do right shift (>>)
# twice in order to divide by 4 and compare with the decimal value
# See details at http://darenmatthews.com/blog/?p=1199
filters_outer = '(ip and (ip[1] & 0xfc) >> 2 == {})'.format(mark)
# IPv6 tcpdump filters explanation:
# ip6[0:2] refers to the bytes 0 and 1 of the IPv6 header,
# containing Version (bits 0 to 3), Traffic Class (bits 4 to 11)
# and part of the Flow Label
# 0x0fc0 = 0000111111000000 is the mask to get only DSCP value from
# the Traffic Class
# We need to do right shift 6 positions (>>) to obtain the DSCP
# value
filters_outer_v6 = (
'(ip6 and (ip6[0:2] & 0x0fc0) >> 6 == {})'.format(mark))
if not outer:
dst_host_filter = 'dst host {}'.format(dst_ip)
if netaddr.valid_ipv6(dst_ip):
filters_outer = filters_outer_v6
return '"{} and {}"'.format(filters_outer, dst_host_filter)
supported_tunnel_types = ('vxlan', 'geneve')
if tunnel_type not in supported_tunnel_types:
raise self.skipException(
"Tunnel type '{}' is not supported by the test. Supported "
"types: {}".format(tunnel_type, supported_tunnel_types))
mac_pattern = '0x' + src_port['mac_address'].replace(':', '')[4:]
# scenario index, which is used in the lists below
# 0 - outer ipv4, inner ipv4
# 1 - outer ipv4, inner ipv6
# 2 - outer ipv6, inner ipv4
# 3 - outer ipv6, inner ipv6
dscp_length = [1, 2, 1, 2]
tos_pattern = ['0xfc', '0xfc0', '0xfc', '0xfc0']
tos_shift = [2, 6, 2, 6]
icmp_value = ['1', '0x3a', '1', '0x3a']
dscp_offset = {'vxlan': [51, 50, 71, 70],
'geneve': [59, 58, 79, 78]}
icmp_offset = {'vxlan': [59, 56, 79, 76],
'geneve': [67, 64, 87, 84]}
if self.is_ipv6 is True:
ip = 'ip6'
# this offset covers difference between ipv4 and ipv6 header length
extra_offset = 20
scenario = 3 if netaddr.valid_ipv6(dst_ip) else 2
filters_outer = filters_outer_v6
else:
ip = 'ip'
extra_offset = 0
scenario = 1 if netaddr.valid_ipv6(dst_ip) else 0
vni_offset = 31 + extra_offset
if tunnel_type == 'vxlan':
tunnel_port = 4789
tunnel_type_offset = 28 + extra_offset
tunnel_type_value = '0x0800'
mac_offset = 44 + extra_offset
vnet_id = segmentation_id
if tunnel_type == 'geneve':
tunnel_port = 6081
tunnel_type_offset = 30 + extra_offset
tunnel_type_value = '0x6558'
mac_offset = 52 + extra_offset
vnet_id = self.get_datapath_tunnel_key(
"external_id:name2=" + self.networks[0]["name"])
port_filter = 'port {}'.format(tunnel_port)
tunnel_type_filter = '{}[{}:2] = {}'.format(
ip, tunnel_type_offset, tunnel_type_value)
vni_filter = '({}[{}:4] & 0x00FFFFFF) == {}'.format(
ip, vni_offset, vnet_id)
mac_filter = '{}[{}:4] = {}'.format(ip, mac_offset, mac_pattern)
inner_dscp_filter = ('({}[{}:{}] & {}) >> {} = {}'.format(
ip, dscp_offset[tunnel_type][scenario], dscp_length[scenario],
tos_pattern[scenario], tos_shift[scenario], mark))
icmp_filter = '{}[{}:1] = {}'.format(
ip, icmp_offset[tunnel_type][scenario], icmp_value[scenario])
filters_inner = ('{} and {} and {} and {} and {} and {}'.format(
port_filter, tunnel_type_filter, vni_filter,
mac_filter, inner_dscp_filter, icmp_filter))
return '"{} and {}"'.format(filters_outer, filters_inner)
def _validate_traffic_marked(
self, mark, src_server, dst_server, ipv6=False, outer=False):
dst_port = self._get_tenant_port_by_type(dst_server['id'], 'test')
dst_ips = []
for fixed_ip in dst_port['fixed_ips']:
if netaddr.valid_ipv6(fixed_ip['ip_address']) and not ipv6:
continue
dst_ips.append(fixed_ip['ip_address'])
self.assertGreater(len(dst_ips), 0)
src_port = self._get_tenant_port_by_type(src_server['id'], 'test')
if outer:
interface = WB_CONF.node_tenant_interface
dst_server['host'] = self.get_host_for_server(
dst_server['id']).split('.')[0]
for node in self.nodes:
if node['name'] == dst_server['host']:
dst_node_ssh_client = node['client']
break
network = self.os_admin.network_client.show_network(
src_port['network_id'])['network']
segmentation_id = network['provider:segmentation_id']
tunnel_type = network['provider:network_type']
capture_host_client = dst_node_ssh_client
else:
interface = utils.get_default_interface(dst_server['ssh_client'])
segmentation_id = None
tunnel_type = None
capture_host_client = dst_server['ssh_client']
for dst_ip in dst_ips:
filters = self._get_dscp_filters(
src_port, dst_ip, mark, outer, tunnel_type, segmentation_id)
remote_capture = capture.TcpdumpCapture(
capture_host_client, interface, filters)
self.useFixture(remote_capture)
time.sleep(10)
self.check_remote_connectivity(
src_server['ssh_client'], dst_ip, ping_count=1)
time.sleep(5)
remote_capture.stop()
msg = 'Not found packets with expected DSCP value'
if outer:
msg += ' set in inner and outer headers'
self.assertFalse(remote_capture.is_empty(), msg)
def _validate_traffic_dscp_marking(
self, src_server, dst_server, ipv6=False, fip_qos=False,
gw_qos=False):
"""Validate that traffic between servers has a dscp mark.
Scenario:
1. First make sure that traffic between servers is not marked.
2. Create a QoS policy, attach to the network where src_server
is connected to.
3. Add a DSCP marking rule to the policy.
4. Send traffic between 2 servers and make sure that now packets
are marked with corresponding DSCP mark.
5. Update the DSCP marking rule to use different mark and make
sure that traffic is marked with a new mark.
6. Create another QoS policy, apply to the port of src_server
and add a DSCP marking rule with another mark.
7. Send traffic between 2 servers and make sure that now packets
are marked with a new DSCP mark (since rule for port has higher
priority).
8. Create another QoS policy applied to router gateway routing to
src_server, verify it takes precedence over port policy.
9. Delete DSCP marking rule from the router gateway QoS policy
and make sure that traffic is marked for port value again.
10. Create another QoS policy applied to FIP of src_server, verify
it takes precedence over port policy.
11. Delete DSCP marking rule from the FIP QoS policy
and make sure that traffic is marked for port value again.
12. Delete DSCP marking rule from the port QoS policy and make sure
that traffic is not marked.
13. Verify a QoS policy attached to a port cannot be deleted.
"""
# First, let's make sure that traffic is not marked
self._validate_traffic_marked(0, src_server, dst_server)
# Create new QoS policy and attach to the src network
net_dscp_policy_id = self._create_qos_policy()
src_port = self.client.list_ports(
device_id=src_server['id'])['ports'][0]
self.admin_client.update_network(
src_port['network_id'], qos_policy_id=net_dscp_policy_id)
self.addCleanup(self.admin_client.update_network,
src_port['network_id'], qos_policy_id=None)
net_rule_id = self.admin_client.create_dscp_marking_rule(
net_dscp_policy_id, self.dscp_mark_net)[
'dscp_marking_rule']['id']
# Validate that traffic that reach destination host is marked
self._validate_traffic_marked(
self.dscp_mark_net, src_server, dst_server)
# Update the DSCP marking rule and verify that traffic is marked
# with a new value
self.admin_client.update_dscp_marking_rule(
net_dscp_policy_id, net_rule_id,
dscp_mark=self.dscp_mark_net_new)
self._validate_traffic_marked(
self.dscp_mark_net_new, src_server, dst_server)
# Create a new QoS policy and attach to the port of src server
dscp_policy_id_new = self._create_qos_policy()
port_rule_id = self.admin_client.create_dscp_marking_rule(
dscp_policy_id_new, self.dscp_mark_port)[
'dscp_marking_rule']['id']
self.client.update_port(
src_port['id'], qos_policy_id=dscp_policy_id_new)
self.addCleanup(self.admin_client.update_port,
src_port['id'], qos_policy_id=None)
# Verify that traffic now is marked with a new value
self._validate_traffic_marked(
self.dscp_mark_port, src_server, dst_server)
# Test QoS policy enforced on router gateway (north-south topology)
if gw_qos:
gw_dscp_policy_id = self._create_qos_policy()
self.router['external_gateway_info'][
'qos_policy_id'] = gw_dscp_policy_id
self.router = self.admin_client.update_router(
self.router['id'],
external_gateway_info=self.router['external_gateway_info']
)['router']
del self.router['external_gateway_info']['qos_policy_id']
self.addCleanup(
self.admin_client.update_router,
self.router['id'],
external_gateway_info=self.router['external_gateway_info'])
gw_rule_id = self.admin_client.create_dscp_marking_rule(
gw_dscp_policy_id, self.dscp_mark_gw)[
'dscp_marking_rule']['id']
# Verify that traffic is marked with a value from gw qos policy
self._validate_traffic_marked(
self.dscp_mark_gw, src_server, dst_server)
# update dscp mark associated with gw qos policy
self.admin_client.update_dscp_marking_rule(
gw_dscp_policy_id, gw_rule_id,
dscp_mark=self.dscp_mark_gw_new)
self._validate_traffic_marked(
self.dscp_mark_gw_new, src_server, dst_server)
# delete dscp mark associated gw qos policy
# port dscp rule applies
self.admin_client.delete_dscp_marking_rule(
gw_dscp_policy_id, gw_rule_id)
self._validate_traffic_marked(
self.dscp_mark_port, src_server, dst_server)
# Create a new QoS Policy and attach to the FIP of src server
# This only applies to south-north tests because the traffic from the
# src server to the dst server goes through the src FIP
if fip_qos:
fip_qos_pol_id = self._create_qos_policy()
# dscp mark value applied to fip is self.dscp_mark_fip
fip_port_rule_id = self.admin_client.create_dscp_marking_rule(
fip_qos_pol_id, self.dscp_mark_fip)['dscp_marking_rule']['id']
src_fip_id = self.client.list_floatingips(
fixed_ip_address=src_port['fixed_ips'][0]['ip_address'])[
'floatingips'][0]['id']
self.client.update_floatingip(
src_fip_id, qos_policy_id=fip_qos_pol_id)
# Verify that traffic is marked with a value from fip qos policy
self._validate_traffic_marked(
self.dscp_mark_fip, src_server, dst_server)
# update dscp mark associated fip qos policy
self.admin_client.update_dscp_marking_rule(
fip_qos_pol_id, fip_port_rule_id,
dscp_mark=self.dscp_mark_fip_new)
self._validate_traffic_marked(
self.dscp_mark_fip_new, src_server, dst_server)
# delete dscp mark associated fip qos policy
# port dscp rule applies
self.admin_client.delete_dscp_marking_rule(
fip_qos_pol_id, fip_port_rule_id)
self._validate_traffic_marked(
self.dscp_mark_port, src_server, dst_server)
# Delete DSCP rule from the port QoS policy and validate that traffic
# is not marked
self.admin_client.delete_dscp_marking_rule(
dscp_policy_id_new, port_rule_id)
self._validate_traffic_marked(0, src_server, dst_server)
# Verify a QoS policy attached to a port cannot be deleted
try:
self.admin_client.delete_qos_policy(dscp_policy_id_new)
except exceptions.Conflict:
LOG.debug('QoS Policy %s cannot be deleted because '
'it is attached to a port', dscp_policy_id_new)
else:
raise RuntimeError('Deletion of QoS Policy %s should '
'have been rejected' % dscp_policy_id_new)
def _test_both_bwlimit_dscp(
self, vms, bwlimit_kbps, dscp_mark, network_id):
# Prerequisite: install iperf3
for role in 'sender', 'receiver':
self._skip_if_iperf3_not_installed(vms[role]['ssh_client'])
# First, let's make sure that bw is not limited
self._validate_bw_limit(client=vms['sender'], server=vms['receiver'])
# Also, make sure that traffic is not marked
self._validate_traffic_marked(0, vms['sender'], vms['receiver'])
# configure qos policies
policy_id = self._create_qos_policy()
self.admin_client.update_network(network_id, qos_policy_id=policy_id)
self.addCleanup(
self.admin_client.update_network, network_id, qos_policy_id=None)
rule_data = {
'max_kbps': bwlimit_kbps,
'max_burst_kbps': bwlimit_kbps * 0.8,
'direction': 'egress'}
self._create_qos_bw_limit_rule(
policy_id, rule_data)['id']
self.admin_client.create_dscp_marking_rule(
policy_id, self.dscp_mark_net)['dscp_marking_rule']['id']
# Make sure that bw limit applied properly before action (using TCP)
self._validate_bw_limit(
bw_limit=bwlimit_kbps * 1000, client=vms['sender'],
server=vms['receiver'])
# Validate that traffic that reach receiver host is marked
self._validate_traffic_marked(
self.dscp_mark_net, vms['sender'], vms['receiver'])
def _test_qos_after_migration(
self, src_server, dst_server, migration_method):
vms = {'sender': src_server, 'receiver': dst_server}
port = self._get_tenant_port_by_type(
vms['sender']['id'], 'test')
max_kbps = (self.bwlimit_kbps_net
if port['binding:vnic_type'] == 'normal'
else self.bwlimit_kbps_net * 1000)
dscp_mark = self.dscp_mark_net
# validate both bwlimit and dscp mark
self._test_both_bwlimit_dscp(
vms, max_kbps, dscp_mark, port['network_id'])
block_migration = (CONF.compute_feature_enabled.
block_migration_for_live_migration)
vm_host = {
'sender': self.get_host_for_server(
vms['sender']['id']).split('.')[0],
'receiver': self.get_host_for_server(
vms['receiver']['id']).split('.')[0]}
for role in 'sender', 'receiver':
if migration_method == 'cold-migration':
self.os_admin.servers_client.migrate_server(vms[role]['id'])
waiters.wait_for_server_status(self.os_admin.servers_client,
vms[role]['id'],
'VERIFY_RESIZE')
# confirm migration
self.os_admin.servers_client.confirm_resize_server(
vms[role]['id'])
elif migration_method == 'live-migration':
# TODO(mblue): reboot WA for nova LP#2080531, revert when fixed
self.os_admin.servers_client.reboot_server(
vms[role]['id'], type='SOFT')
self.wait_for_server_active(vms[role])
block_migration = (CONF.compute_feature_enabled.
block_migration_for_live_migration)
self.os_admin.servers_client.live_migrate_server(
vms[role]['id'], host=None,
block_migration=block_migration, disk_over_commit=False)
else:
raise RuntimeError('Unsupported migration method %s'
% migration_method)
self.wait_for_server_active(vms[role])
vm_new_host = self.get_host_for_server(
vms[role]['id']).split('.')[0]
self.assertNotEqual(vm_host[role], vm_new_host,
'%s vm did not migrate' % role)
vm_host[role] = vm_new_host
LOG.debug("Validating server '%s'", role)
# Validate that traffic that reach destination host is still marked
self._validate_traffic_marked(
dscp_mark, vms['sender'], vms['receiver'])
# Make sure that bw limit still works after migration, but only if
# sender VM and receiver VM are in different hosts.
if vm_host['sender'] != vm_host['receiver']:
# NOTE(mblue): enable and re-add symmetric and strict routing
# for control ip address/nic after cold migration.
# TODO(mblue): reboot WA for nova LP#2080531, revert when fixed
# to apply `tenant_sym_path_for_port` only on cold migration.
for _vm in (vms['sender'], vms['receiver']):
ctl_port = self._get_tenant_port_by_type(
_vm['id'], 'control')
utils.configure_interface_up(
_vm['ssh_client'], ctl_port)
self.tenant_sym_path_for_port(
_vm['ssh_client'], ctl_port)
self._validate_bw_limit(
bw_limit=max_kbps * 1000, client=vms['sender'],
server=vms['receiver'])
def _check_dscp_inheritance(self):
src_server, dst_server = self._create_vms_by_topology(ipv6=True)
policy_id = self._create_qos_policy()
src_port = self.client.list_ports(
device_id=src_server['id'])['ports'][0]
self.admin_client.create_dscp_marking_rule(
policy_id, self.dscp_mark_port)
self.client.update_port(
src_port['id'], qos_policy_id=policy_id)
self.addCleanup(self.admin_client.update_port,
src_port['id'], qos_policy_id=None)
self._validate_traffic_marked(
self.dscp_mark_port, src_server, dst_server, ipv6=True, outer=True)
class QosTestCommon(QosBaseTest):
@decorators.idempotent_id('db036021-ae2b-4149-b342-a5619aa606e4')
def test_dscp_marking_tenant_network(self):
src_server, dst_server = self._create_vms_by_topology()
self._validate_traffic_dscp_marking(src_server, dst_server, ipv6=True)
@decorators.idempotent_id('8cf45ea9-bdbf-478e-8810-792083e1c467')
def test_dscp_marking_external_network(self):
src_server, dst_server = self._create_vms_by_topology(
topology='external')
self._validate_traffic_dscp_marking(src_server, dst_server)
# Since QoS is applied to egress traffic than south->north
# test direction is more interesting than north->south
@decorators.idempotent_id('886d6c11-7c19-4c01-a5ac-74779bb2f364')
def test_dscp_marking_south_north(self):
# For south>north the order of returned by
# self._create_vms_by_topology should be reverse
dst_server, src_server = self._create_vms_by_topology(
topology='north-south')
if self.has_ovn_support:
self._validate_traffic_dscp_marking(
src_server, dst_server, fip_qos=True, gw_qos=True)
else:
self._validate_traffic_dscp_marking(src_server, dst_server)
@decorators.idempotent_id('d075dd6f-0101-4b1a-990c-a001313d3914')
def test_dscp_marking_east_west(self):
src_server, dst_server = self._create_vms_by_topology(
topology='east-west')
self._validate_traffic_dscp_marking(src_server, dst_server, ipv6=True)
@decorators.idempotent_id('161dc56f-a7e8-426e-9b06-3807e0c45692')
def test_bw_limit_tenant_network(self):
client, server = self._create_vms_by_topology()
self._validate_traffic_bw_limit(client, server, egress=True, ipv6=True)
self._validate_traffic_bw_limit(
client, server, egress=False, ipv6=True)
@decorators.idempotent_id('653e92b4-1f19-4f85-93f7-c14d0e900f44')
def test_bw_limit_south_north(self):
# For south>north the order of returned by
# self._create_vms_by_topology should be reverse
server, client = self._create_vms_by_topology(topology='north-south')
self._validate_traffic_bw_limit(
client, server, egress=True, fip_qos=True, gw_qos=True)
self._validate_traffic_bw_limit(
client, server, egress=False, fip_qos=True, gw_qos=True)
@decorators.idempotent_id('fc833d46-d18f-4edf-b082-5f5fe909fb79')
def test_bw_limit_east_west(self):
client, server = self._create_vms_by_topology(topology='east-west')
self._validate_traffic_bw_limit(client, server, egress=True, ipv6=True)
self._validate_traffic_bw_limit(
client, server, egress=False, ipv6=True)
@decorators.idempotent_id('121bd2dd-6d41-4658-8a05-6fdd444381fa')
@testtools.skipUnless(CONF.compute_feature_enabled.live_migration,
'Live migration is not available.')
@decorators.attr(type='slow')
def test_qos_after_live_migration(self):
if self.is_setup_single_compute():
raise self.skipException('Test needs more than 1 compute')
src_server, dst_server = (
self._create_vms_by_topology())
self._test_qos_after_migration(
src_server, dst_server, migration_method='live-migration')
@decorators.idempotent_id('f6e1e433-33b1-4352-bdf3-92dbe9e3f2ee')
@testtools.skipUnless(CONF.compute_feature_enabled.cold_migration,
'Cold migration is not available.')
@decorators.attr(type='slow')
def test_qos_after_cold_migration(self):
if self.is_setup_single_compute():
raise self.skipException('Test needs more than 1 compute')
src_server, dst_server = self._create_vms_by_topology()
self._test_qos_after_migration(
src_server, dst_server, migration_method='cold-migration')
@decorators.idempotent_id('5cbf643a-b4ed-453c-a073-d468a95a9d78')
def test_default_qos_policy(self):
default_policy = self.admin_client.create_qos_policy(
name='default-qos-policy',
shared=True,
is_default=True)['policy']
self.qos_policies.append(default_policy)
# create network using admin_client (same client used to create qos
# policies and rules)
network1 = self.admin_client.create_network(
name='default-qos-network-1')['network']
self.addCleanup(
self.admin_client.delete_network, network1['id'])
# check default qos policy was assigned at network creation
self.assertEqual(default_policy['id'], network1['qos_policy_id'])
# qos policy is not default anymore
self.admin_client.update_qos_policy(default_policy['id'],
is_default=False)
# create network using admin_client (same client used to create qos
# policies and rules
network2 = self.admin_client.create_network(
name='default-qos-network-2')['network']
self.addCleanup(
self.admin_client.delete_network, network2['id'])
# check no default qos policy was assigned at network creation
self.assertIsNone(network2['qos_policy_id'])
def _get_user_policy_file(self):
cmd = ("{} crudini --get /etc/neutron/neutron.conf "
"oslo_policy policy_file ".format(self.neutron_container))
policy_file = self.run_on_master_controller(cmd).rstrip()
self.assertNotEqual(
'', policy_file,
'Failed to determine a valid policy file path')
return policy_file
def _dump_user_policy_file(self, node):
cmd = ("{} oslopolicy-policy-generator "
"--namespace neutron --output-file {}".format(
self.neutron_container, self.user_policy_file))
node['client'].exec_command(cmd)
node['client'].exec_command("{0} cp -f {1} {1}.bak".format(
self.neutron_container, self.user_policy_file))
def _restore_user_policy_files(self):
for node in self.nodes:
if node['type'] != 'controller':
continue
node['client'].exec_command(
"{0} bash -c 'cp -f {1}.bak {1} || true'".format(
self.neutron_container, self.user_policy_file))
def _set_user_policy(self, node, policy, value):
cmd = (r"{} sed -i 's/\(\"{}\":\ \).*/\1\"{}\"/' {}".format(
self.neutron_container, policy, value, self.user_policy_file))
LOG.debug("cmd = %s", cmd)
return node['client'].exec_command(cmd).rstrip()
class QosTestDscpInheritanceOvn(QosBaseTest, base.BaseTempestTestCaseOvn):
@testtools.skipUnless(
'geneve' in config.CONF.neutron_plugin_options.available_type_drivers,
"No 'geneve' in neutron_plugin_options.available_type_drivers")
@decorators.idempotent_id('55f9a614-3c80-4dc6-be4b-de9b18583317')
def test_dscp_inheritance_geneve(self):
cmd = "sudo ovs-vsctl get open . external_ids:ovn-encap-tos || true"
msg = "external_ids:ovn-encap-tos is not set to 'inherit'"
if self.is_devstack:
result = self.run_on_master_controller(cmd)
if 'inherit' not in result:
raise self.skipException(msg)
else:
for node in self.nodes:
result = node['client'].exec_command(cmd)
if 'inherit' not in result:
raise self.skipException(msg)
self._check_dscp_inheritance()
class QosTestExternalNetwork(QosBaseTest):
"""These tests should not be executed in parallel with other tests because
they perform changes in the external network, which is shared with many
other tests and could affect them
"""
def setUp(self):
super(QosTestExternalNetwork, self).setUp()
self.ensure_external_network_is_shared()
@decorators.idempotent_id('c018b96b-47eb-4b5f-b750-e0c857fd86a9')
def test_dscp_bwlimit_external_network(self):
if self.is_setup_single_compute():
raise self.skipException('Test needs more than 1 compute')
max_kbps = self.bwlimit_kbps_net
dscp_mark = self.dscp_mark_net
# create sender and receiver VMs connected to the external network
vms = {}
vms['sender'], vms['receiver'] = self._create_vms_by_topology(
topology='external')
# validate both bwlimit and dscp mark from sender to receiver via
# external network
self._test_both_bwlimit_dscp(
vms, max_kbps, dscp_mark, CONF.network.public_network_id)
class QosTestSriovBaseTest(QosBaseTest):
def skip_if_no_qos_extension_sriov_agent(self):
l3_agents = self.os_admin.network_client.list_agents(
binary='neutron-sriov-nic-agent')['agents']
if not l3_agents:
raise self.skipTest('No sriov agent found')
for agent in l3_agents:
if 'qos' in agent['configurations'].get('extensions', []):
return
raise self.skipTest('No sriov agent with qos extension enabled')
def setUp(self):
super(QosTestSriovBaseTest, self).setUp()
self.skip_if_no_qos_extension_sriov_agent()
class QosTestSriovBwLimitTest(QosTestSriovBaseTest):
MIN_KBPS_NO_BWLIMIT = 1000000
sriov_test = True
@decorators.idempotent_id('4aeb240a-b350-4dd4-b3b2-8aadf9ad8e6d')
def test_bw_limit_sriov_direct(self):
client, server = self._create_vms_by_topology(topology='external',
port_type='direct')
# at this moment, SRIOV ports only support egress bw limit
self._validate_traffic_bw_limit(client, server, egress=True)
@decorators.idempotent_id('9a38b196-2492-4bcb-a914-5e1958f6bbb9')
def test_bw_limit_sriov_macvtap(self):
client, server = self._create_vms_by_topology(topology='external',
port_type='macvtap')
# at this moment, SRIOV ports only support egress bw limit
self._validate_traffic_bw_limit(client, server, egress=True)
class QosTestSriovMinBwPlacementEnforcementTest(QosTestSriovBaseTest):
# Nova rejects to boot VM with port which has resource_request field, below
# microversion 2.72
compute_min_microversion = '2.72'
compute_max_microversion = 'latest'
INVENTORY_KEYS = {'egress': 'NET_BW_EGR_KILOBIT_PER_SEC',
'ingress': 'NET_BW_IGR_KILOBIT_PER_SEC'}
@classmethod
def skip_checks(cls):
super(QosTestSriovMinBwPlacementEnforcementTest, cls).skip_checks()
api_version_utils.check_skip_with_microversion(
cls.compute_min_microversion, cls.compute_max_microversion,
CONF.compute.min_microversion, CONF.compute.max_microversion)
@classmethod
def resource_setup(cls):
super(QosTestSriovMinBwPlacementEnforcementTest, cls).resource_setup()
cls.compute_request_microversion = (
api_version_utils.select_request_microversion(
cls.compute_min_microversion,
CONF.compute.min_microversion))
def setUp(self):
super(QosTestSriovMinBwPlacementEnforcementTest, self).setUp()
self.useFixture(api_microversion_fixture.APIMicroversionFixture(
compute_microversion=self.compute_request_microversion))
@classmethod
def setup_clients(cls):
super(QosTestSriovMinBwPlacementEnforcementTest, cls).setup_clients()
try:
cls.resource_providers_client = \
cls.os_admin.placement.ResourceProvidersClient()
except AttributeError:
LOG.info('placement ResourceProvidersClient is not supported - '
'a configured bandwidth should be provided via '
'tempest configuration')
cls.resource_providers_client = None
def _test_minbw_placement_enforcement_sriov(self, direction):
def _check_rp_allocations(configured_min_bws, server_ids):
if not self.resource_providers_client:
LOG.debug('Resource Provider allocations cannot be verified')
return
for rp_id in self.nic_rp_ids:
rp_allocations = (
self.resource_providers_client.show_resource_provider(
rp_id + '/allocations'))['allocations']
for uuid, resources in rp_allocations.items():
self.assertIn(uuid, server_ids)
LOG.debug('Server %s found in RP allocations', uuid)
server_ids.remove(uuid)
for d, configured_min_bw in configured_min_bws.items():
self.assertEqual(
configured_min_bw,
resources['resources'][self.INVENTORY_KEYS[d]])
LOG.debug(
'Verified allocated %s bandwidth (kbps) = %d',
d, configured_min_bw)
self.assertEqual(0, len(server_ids)) # all servers found
def _bw_per_nic(directions):
if not self.resource_providers_client:
LOG.debug('BW and number of NICs need to be configured')
nic_bws = {}
if 'egress' in directions:
nic_bws['egress'] = WB_CONF.minbw_placement_nic_kbps_egress
if 'ingress' in directions:
nic_bws['ingress'] = (WB_CONF.
minbw_placement_nic_kbps_ingress)
return (
nic_bws,
WB_CONF.minbw_placement_num_nics)
# obtain list of resource providers and filter those
# corresponding with NICs
# then, obtain the bw supported per NIC
rp_list = self.resource_providers_client.list_resource_providers()[
'resource_providers']
# nic_rp_ids will contain the RPs corresponding to NIC resources
self.nic_rp_ids = []
nic_bws_list = []
for rp in rp_list:
if ':NIC Switch agent:' in rp['name']:
self.nic_rp_ids.append(rp['uuid'])
inventories = (
self.resource_providers_client.
list_resource_provider_inventories(
rp['uuid'])['inventories'])
nic_bws = {}
for d in directions:
nic_bws[d] = inventories[
self.INVENTORY_KEYS[d]]['total']
nic_bws_list.append(nic_bws)
for bw in nic_bws_list[1:]:
if nic_bws_list[0] != bw:
self.skipTest(
'Test only supported if all NICs are configured with '
'identical BW values '
'(ingress and egress values can be different)')
return nic_bws_list[0], len(nic_bws_list)
if direction in ('egress', 'ingress'):
directions = [direction]
elif direction == 'both':
directions = ['egress', 'ingress']
else:
raise ValueError('direction value not supported: %s' %
direction)
nic_bws, num_nics = _bw_per_nic(directions)
# the number of VMs supported equals the number of NICs,
# considering each VM is created with a port
# with min_bw between 60% and 100% of nic_bw
max_number_vms = num_nics
# create qos policy that will be applied to the provider network
minbw_policy_id = self._create_qos_policy()
configured_min_bws = {}
for d in directions:
configured_min_bws[d] = int(nic_bws[d] * random.uniform(0.6, 1.0))
self.qos_min_bw_rules_client.create_minimum_bandwidth_rule(
qos_policy_id=minbw_policy_id,
**{'direction': d,
'min_kbps': configured_min_bws[d]})
# creation of max_number_vms VMs should be successful
server_ids = []
for i in range(max_number_vms):
port_type = random.choice(('direct', 'macvtap'))
server_id = self._create_server_for_topology(
network_id=CONF.network.public_network_id,
port_type=port_type,
port_qos_policy_id=minbw_policy_id)['id']
server_ids.append(server_id)
_check_rp_allocations(configured_min_bws, server_ids)
self.assertRaises(
tempest_exceptions.BuildErrorException,
self._create_server_for_topology,
network_id=CONF.network.public_network_id,
port_type=random.choice(('direct', 'macvtap')),
port_qos_policy_id=minbw_policy_id)
@decorators.idempotent_id('6c63d5b8-d642-4a7c-9b07-c0755979b6a8')
def test_minbw_placement_enforcement_sriov_egress(self):
self._test_minbw_placement_enforcement_sriov('egress')
@decorators.idempotent_id('c1791f96-9eb4-4119-b598-5610c6043ef1')
def test_minbw_placement_enforcement_sriov_ingress(self):
self._test_minbw_placement_enforcement_sriov('ingress')
@decorators.idempotent_id('e212fab8-9a2d-4753-ba5d-53a299b0af36')
def test_minbw_placement_enforcement_sriov_both(self):
self._test_minbw_placement_enforcement_sriov('both')
class QosTestOvn(base.BaseTempestTestCaseOvn, QosBaseTest):
MAX_KBPS = 1000
MAX_BURST_KBPS = int(0.8 * MAX_KBPS)
DSCP_MARK_OPTION = QosBaseTest.dscp_mark_net
def _create_qos_policy_bw_and_dscp(self):
# Creates QoS policy with bw and dscp rules,
# rules values are according to class constants.
# Returns:
# str:uuid of QoS policy
policy_id = self._create_qos_policy()
rule_data = {
'max_kbps': self.MAX_KBPS,
'max_burst_kbps': self.MAX_BURST_KBPS}
self._create_qos_bw_limit_rule(
policy_id, rule_data)['id']
self.admin_client.create_dscp_marking_rule(
policy_id,
self.DSCP_MARK_OPTION)
LOG.debug("Created QoS policy and rules")
return policy_id
def _validate_bw_limit_nbdb_qos(self, qos_settings):
bandwidth_settings = ''
for line in qos_settings.splitlines():
if line.startswith('bandwidth'):
bandwidth_settings = line
break
self.assertIn(f'rate={self.MAX_KBPS:.0f}', bandwidth_settings,
'Bandwidth options are not set as expected')
self.assertIn(f'burst={self.MAX_BURST_KBPS:.0f}', bandwidth_settings,
'Bandwidth options are not set as expected')
LOG.debug('BW limit options found')
def _validate_dscp_nbdb_qos(self, qos_settings):
dscp_settings = ''
for line in qos_settings.splitlines():
if line.startswith('action'):
dscp_settings = line
break
self.assertIn(f'dscp={self.DSCP_MARK_OPTION}', dscp_settings,
'DSCP options are not set as expected')
LOG.debug('DSCP options found')
def _validate_qos_rules_nbdb(
self, port_id=None, fip_id=None, expected_empty=False):
# Validates QoS bw and dscp rules with constant values in OVN NBDB.
# Parameters:
# port_id (str): port uuid that has QoS policy attached.
# fip_id (str): floating ip uuid that has QoS policy attached.
# expected_empty(bool): True if no qos rules are expected
self.assertTrue(port_id or fip_id,
'At least one of the input params is required')
if port_id:
cmd = r'{} find qos match="inport\ \=\=\ \"{}\""'.format(
self.nbctl, port_id)
qos_settings = self.run_on_master_controller(cmd).rstrip()
if expected_empty:
self.assertFalse(qos_settings,
'QoS is not supposed to be applied on this '
'port in OVN NBDB')
LOG.debug('Success: no QoS policies found, as expected')
else:
self._validate_dscp_nbdb_qos(qos_settings)
self._validate_bw_limit_nbdb_qos(qos_settings)
if fip_id:
cmd = r'%s find qos external_ids={"neutron\:fip_id"="%s"}' % (
self.nbctl, fip_id)
qos_settings = self.run_on_master_controller(cmd).rstrip()
if expected_empty:
self.assertFalse(qos_settings,
'QoS is not supposed to be applied on this '
'FIP in OVN NBDB')
LOG.debug('Success: no QoS policies found, as expected')
else:
self._validate_dscp_nbdb_qos(qos_settings)
self._validate_bw_limit_nbdb_qos(qos_settings)
@decorators.idempotent_id('08b74ece-d7f2-4a80-9a1e-5fb7ec928a9b')
def test_attach_qos_port_to_vm_with_another_port(self):
# TODO(eolivare): this module should also test BW limit on VMs running
# on a single compute, checking tenant network type is different from
# vlan - this could be done reusing the existing tests or adding new
# ones, not sure yet.
if self.is_setup_single_compute():
raise self.skipException('Test needs more than 1 compute')
# create resources
network_qos = self.create_network()
network_no_qos = self.create_network()
subnet_qos = self.create_subnet(network_qos, cidr="10.10.1.0/24")
subnet_no_qos = self.create_subnet(network_no_qos, cidr="10.10.2.0/24")
secgroup = self.secgroup['security_group']
self.create_router_interface(self.router['id'], subnet_qos['id'])
self.create_router_interface(self.router['id'], subnet_no_qos['id'])
policy_id = self._create_qos_policy_bw_and_dscp()
# create port with QoS policy, another port without policy
port_qos = self.create_port(
network_qos,
qos_policy_id=policy_id,
security_groups=[secgroup['id']])
port_no_qos = self.create_port(
network_no_qos,
qos_policy_id=None,
security_groups=[secgroup['id']])
# launch server with policy port, then attach to non policy port
# (so default route on tenant vm test port first, as other bw tests)
port_qos, fip_qos, server = self._create_server(
port=port_qos).values()
# other server to validate QoS policy port later
scheduler_hints = {'different_host': server['id']}
other_fip, other_server = tuple(self._create_server(
network=network_qos,
security_groups=[{'name': secgroup['name']}],
scheduler_hints=scheduler_hints
).values())[1:]
server['ssh_client'] = ssh.Client(
fip_qos['floating_ip_address'],
self.username,
pkey=self.keypair['private_key'])
other_server['ssh_client'] = ssh.Client(
other_fip['floating_ip_address'],
self.username,
pkey=self.keypair['private_key'])
server['ssh_client'].test_connection_auth()
self.create_interface(server['id'], port_no_qos['id'])
waiters.wait_for_interface_status(
self.os_primary.interfaces_client, server['id'],
port_no_qos['id'], constants.PORT_STATUS_ACTIVE)
# configure ip and activate QoS port interface from server CLI
utils.configure_interface_up(server['ssh_client'], port_no_qos)
# validate connectivity with QoS port using another VM on QoS subnet
self.check_remote_connectivity(
other_server['ssh_client'],
port_qos['fixed_ips'][0]['ip_address'],
ping_count=1)
# validate only one port from both has QoS applied in OVN NBDB
self._validate_qos_rules_nbdb(port_id=port_qos['id'])
self._validate_qos_rules_nbdb(port_id=port_no_qos['id'],
expected_empty=True)
# validate dscp rules applied on traffic
self._validate_traffic_marked(
self.DSCP_MARK_OPTION,
src_server=server, dst_server=other_server)
# validate bw limit rules applied on traffic
# (optionally if iperf3 installed on advanced image)
is_iperf_installed = False
try:
for ssh_client in other_server['ssh_client'], server['ssh_client']:
self._skip_if_iperf3_not_installed(ssh_client)
is_iperf_installed = True
except self.skipException:
LOG.debug("iperf3 not found on VM, skipped QoS traffic validation")
if is_iperf_installed:
self._validate_bw_limit(
client=server, server=other_server,
bw_limit=self.MAX_KBPS * 1000,
ignore_port_ids=(port_no_qos['id'],))
@decorators.idempotent_id('ba85bd87-f4f6-45a8-a2bd-97acb804b6f9')
def test_create_network_qos_policy_before_creating_vm(self):
# create QoS policy and rules first, then other resources
policy_id = self._create_qos_policy_bw_and_dscp()
network = self.create_network()
LOG.debug("Created network '%s'", network['name'])
subnet = self.create_subnet(network)
self.create_router_interface(self.router['id'], subnet['id'])
# attach QoS policy to network before server is launched
self.admin_client.update_network(
network['id'], qos_policy_id=policy_id)
LOG.debug("Attached QoS policy to network '%s'", network['name'])
port, fip, server = tuple(self._create_server(
create_floating_ip=True, network=network).values())
# attach a qos policy to the fip
fip_policy_id = self._create_qos_policy_bw_and_dscp()
self.client.update_floatingip(fip['id'], qos_policy_id=fip_policy_id)
# validate QoS object appears in NBDB, disappears when server and fip
# are deleted
self._validate_qos_rules_nbdb(port_id=port['id'], fip_id=fip['id'])
self.os_primary.servers_client.delete_server(server['id'])
waiters.wait_for_server_termination(self.os_primary.servers_client,
server['id'])
self.client.delete_floatingip(fip['id'])
self._validate_qos_rules_nbdb(
port_id=port['id'], fip_id=fip['id'], expected_empty=True)
View Git Blame Copy Permalink