arif/zaza-openstack-tests
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add L3 data plane testing NDR tests

Browse Source 
In order to actually test L3 for advertised routes, a simple ping test
is necessary to perform from a unit that receives dynamic routes.
This commit is contained in:
Dmitrii Shcherbakov
2023-06-14 21:32:11 +04:00
parent 77aeb7300f
commit 8fae640826
2 changed files with 103 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zaza/openstack/charm_tests/dragent/test.py
-104
View File
@@ -1,104 +0,0 @@
#!/usr/bin/env python3
# Copyright 2018 Canonical Ltd.
#
# 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.
"""Run BGP tests."""
import argparse
import logging
import sys
import tenacity
from zaza import model
from zaza.openstack.utilities import (
cli as cli_utils,
juju as juju_utils,
openstack as openstack_utils,
)
def test_bgp_routes(peer_application_name="osci-frr", keystone_session=None):
"""Test BGP routes.
:param peer_application_name: String name of BGP peer application
:type peer_application_name: string
:param keystone_session: Keystone session object for overcloud
:type keystone_session: keystoneauth1.session.Session object
:raises: AssertionError if expected BGP routes are not found
:returns: None
:rtype: None
"""
# If a session has not been provided, acquire one
if not keystone_session:
keystone_session = openstack_utils.get_overcloud_keystone_session()
# Get authenticated clients
neutron_client = openstack_utils.get_neutron_session_client(
keystone_session)
# Get the peer unit
peer_unit = model.get_units(peer_application_name)[0].entity_id
# Get expected advertised routes
private_cidr = neutron_client.list_subnets(
name="private_subnet")["subnets"][0]["cidr"]
floating_ip_cidr = "{}/32".format(
neutron_client.list_floatingips()
["floatingips"][0]["floating_ip_address"])
# This test may run immediately after configuration. It may take time for
# routes to propogate via BGP. Do a binary backoff.
@tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60),
reraise=True, stop=tenacity.stop_after_attempt(10))
def _assert_cidr_in_peer_routing_table(peer_unit, cidr):
logging.debug("Checking for {} on BGP peer {}"
.format(cidr, peer_unit))
# Run show ip route bgp on BGP peer
routes = juju_utils.remote_run(
peer_unit, remote_cmd='vtysh -c "show ip route bgp"')
logging.info(routes)
assert cidr in routes, (
"CIDR, {}, not found in BGP peer's routing table: {}"
.format(cidr, routes))
_assert_cidr_in_peer_routing_table(peer_unit, private_cidr)
logging.info("Private subnet CIDR, {}, found in routing table"
.format(private_cidr))
_assert_cidr_in_peer_routing_table(peer_unit, floating_ip_cidr)
logging.info("Floating IP CIDR, {}, found in routing table"
.format(floating_ip_cidr))
def run_from_cli():
"""Run test for BGP routes from CLI.
:returns: None
:rtype: None
"""
cli_utils.setup_logging()
parser = argparse.ArgumentParser()
parser.add_argument("--peer-application", "-a",
help="BGP Peer application name. Default: osci-frr",
default="osci-frr")
options = parser.parse_args()
peer_application_name = cli_utils.parse_arg(options,
"peer_application")
test_bgp_routes(peer_application_name)
if __name__ == "__main__":
sys.exit(run_from_cli())
zaza/openstack/charm_tests/dragent/tests.py
+103 -6
View File
@@ -16,25 +16,122 @@
"""Define class of BGP tests."""
import logging
import tenacity
import unittest
import zaza.openstack.charm_tests.neutron.tests as neutron_tests
from zaza.openstack.utilities import cli as cli_utils
from zaza.openstack.charm_tests.dragent import test
from zaza import model
from zaza.openstack.utilities import (
cli as cli_utils,
juju as juju_utils,
)
class DRAgentTest(unittest.TestCase):
class DRAgentTest(neutron_tests.NeutronNetworkingBase):
"""Class to encapsulate BPG tests."""
BGP_PEER_APPLICATION = 'osci-frr'
def setUp(self):
"""Run setup actions specific to the class."""
super().setUp()
self._peer_unit = model.get_units(
self.BGP_PEER_APPLICATION)[0].entity_id
@classmethod
def setUpClass(cls):
"""Run setup for BGP tests."""
"""Run setup actions specific to the class."""
super().setUpClass()
cli_utils.setup_logging()
@staticmethod
@tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60),
reraise=True, stop=tenacity.stop_after_attempt(10))
def _assert_cidr_in_peer_routing_table(peer_unit, cidr):
logging.debug("Checking for {} on BGP peer {}"
.format(cidr, peer_unit))
# Run show ip route bgp on BGP peer
routes = juju_utils.remote_run(
peer_unit, remote_cmd='vtysh -c "show ip route bgp"')
logging.info(routes)
assert cidr in routes, (
"CIDR, {}, not found in BGP peer's routing table: {}"
.format(cidr, routes))
@staticmethod
@tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60),
reraise=True, stop=tenacity.stop_after_attempt(10))
def _assert_ip_reachable_via_peer(peer_unit, address):
logging.debug(f"Checking if {peer_unit} can reach {address} using "
f"routes adverised by NDR")
# Ping with -w specified will return an exit code if there is no
# response after the number of seconds specified. This is to ignore the
# first ping that may not arrive due to an ARP resolution.
juju_utils.remote_run(peer_unit, fatal=True,
remote_cmd=f'ping -w4 {address}')
def test_bgp_routes(self):
"""Run bgp tests."""
test.test_bgp_routes(peer_application_name=self.BGP_PEER_APPLICATION)
"""Test BGP routes.
A test that checks only the control plane of Neutron Dynamic Routing.
:raises: AssertionError if expected BGP routes are not found
:returns: None
:rtype: None
"""
# Get expected advertised routes
private_cidr = self.project_subnet['cidr']
floating_ip_cidr = "{}/32".format(
self.neutron_client.list_floatingips()
["floatingips"][0]["floating_ip_address"])
# This test may run immediately after configuration.
# It may take time for routes to propagate via BGP. Do a
# binary backoff.
self._assert_cidr_in_peer_routing_table(self._peer_unit, private_cidr)
logging.info("Private subnet CIDR, {}, found in routing table"
.format(private_cidr))
self._assert_cidr_in_peer_routing_table(self._peer_unit,
floating_ip_cidr)
logging.info("Floating IP CIDR, {}, found in routing table"
.format(floating_ip_cidr))
def test_instance_connectivity(self):
"""Test connectivity to instances via dynamic routes.
Make sure that with routes advertised via NDR it is actually possible
to reach instances from a unit that gets those routes programmed into
its routing table.
"""
# Get an instance but do not perform connectivity checks as the machine
# running those tests does not have the dynamic routes advertised to
# the peer unit by NDR.
fip_instance = self.launch_guest('fip-instance', instance_key='jammy',
perform_connectivity_check=False)
@tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, max=60),
reraise=True, stop=tenacity.stop_after_attempt(10))
def get_fip(nova_client, instance_id):
"""Try to get a FIP from an instance.
Instance FIPs may not be immediately accessible from the Nova
object after the instance creation so a retry logic is necessary.
"""
# The reason for looking up an instance object again is that the
# Nova client does not refresh address information after
instance = nova_client.servers.find(id=instance_id)
fips = neutron_tests.floating_ips_from_instance(instance)
if not fips:
raise tenacity.TryAgain
return fips[0]
fip = get_fip(self.nova_client, fip_instance.id)
# First check that the FIP is present in the peer unit's routing table.
self._assert_cidr_in_peer_routing_table(self._peer_unit, f'{fip}/32')
# Once it is, check if it is actually reachable.
self._assert_ip_reachable_via_peer(self._peer_unit, fip)
if __name__ == "__main__":