Improve Octavia ovn provider scenario test

So far Octavia ovn-provider did not allow the end user to curl the LB
VIP.

Now that it is allowed, this patch improves the Octavia ovn-provider
scenario test by creating the Octavia resources using ports 80/8080 and
cURLing them (instead of only SSHing them to generate traffic).

Change-Id: I32670197f4600f09d07b36810fd081f1512b77bc

tobiko/openstack/octavia/_validators.py

@@ -59,12 +59,13 @@ def check_members_balanced(ip_address: str,
     for attempt in tobiko.retry(count=members_count * requests_count,
                                 interval=interval):
         try:
-            content = curl.execute_curl(hostname=ip_address,
-                                        scheme=protocol,
-                                        port=port,
-                                        path='id',
-                                        connect_timeout=connect_timeout,
-                                        ssh_client=ssh_client).strip()
+            content = curl.execute_curl(
+                hostname=ip_address,
+                scheme='HTTP' if protocol == 'TCP' else protocol,
+                port=port,
+                path='id',
+                connect_timeout=connect_timeout,
+                ssh_client=ssh_client).strip()
         except sh.ShellCommandFailed as ex:
             if ex.exit_status == 28:
                 raise octavia.TrafficTimeoutError(
@@ -93,6 +94,8 @@ def check_members_balanced(ip_address: str,
 
     # assert that 'members_count' servers replied
     missing_members_count = members_count - len(replies)
+    LOG.debug(f'Members count from pool {pool_id} is {members_count}')
+    LOG.debug(f'len(replies) is {len(replies)}')
     test_case.assertEqual(0, missing_members_count,
                           f'Missing replies from {missing_members_count} "'
                           '"members.')

tobiko/openstack/stacks/_octavia.py

@@ -298,11 +298,9 @@ class OVNIPv6LoadBalancerStack(OVNIPv4LoadBalancerStack):
 class TcpSourceIpPortOvnIpv4Listener(HttpRoundRobinAmphoraIpv4Listener):
     loadbalancer = tobiko.required_fixture(OVNIPv4LoadBalancerStack)
     lb_protocol = 'TCP'
-    lb_port = 22
-    has_monitor = False
+    hm_type = 'TCP'
     lb_algorithm = 'SOURCE_IP_PORT'
     pool_protocol = 'TCP'
-    application_port = 22
 
 
 class TcpSourceIpPortOvnIpv6Listener(TcpSourceIpPortOvnIpv4Listener):

tobiko/tests/scenario/octavia/test_traffic.py

@@ -14,10 +14,6 @@
 #    under the License.
 from __future__ import absolute_import
 
-import collections
-import json
-import typing
-
 import testtools
 from oslo_log import log
 
@@ -27,7 +23,6 @@ from tobiko.openstack import neutron
 from tobiko.openstack import octavia
 from tobiko.openstack import stacks
 from tobiko.shell import sh
-from tobiko.shell import ssh
 
 
 LOG = log.getLogger(__name__)
@@ -64,25 +59,42 @@ class OctaviaBasicTrafficScenarioTest(testtools.TestCase):
         self.listener_stack.wait_for_members_to_be_reachable()
 
     def test_round_robin_traffic(self):
-        # For 5 minutes we ignore specific exceptions as we know
-        # that Octavia resources are being provisioned
-        for attempt in tobiko.retry(timeout=300.):
-            try:
-                octavia.check_members_balanced(
-                    pool_id=self.listener_stack.pool_id,
-                    ip_address=self.loadbalancer_stack.floating_ip_address,
-                    lb_algorithm=self.listener_stack.lb_algorithm,
-                    protocol=self.listener_stack.lb_protocol,
-                    port=self.listener_stack.lb_port)
-                break
-            except (octavia.RoundRobinException,
-                    octavia.TrafficTimeoutError,
-                    sh.ShellCommandFailed):
+        _test_traffic(
+            pool_id=self.listener_stack.pool_id,
+            ip_address=self.loadbalancer_stack.floating_ip_address,
+            lb_algorithm=self.listener_stack.lb_algorithm,
+            protocol=self.listener_stack.lb_protocol,
+            port=self.listener_stack.lb_port)
+
+
+def _test_traffic(pool_id: str, ip_address: str, lb_algorithm: str,
+                  protocol: str, port: int):
+    # For 5 minutes we ignore specific exceptions as we know
+    # that Octavia resources are being provisioned
+    for attempt in tobiko.retry(timeout=300.):
+        try:
+            octavia.check_members_balanced(pool_id=pool_id,
+                                           ip_address=ip_address,
+                                           lb_algorithm=lb_algorithm,
+                                           protocol=protocol,
+                                           port=port)
+            break
+
+        # TODO oschwart: change the following without duplicating code
+        except octavia.RoundRobinException:
+            if lb_algorithm == 'ROUND_ROBIN':
                 LOG.exception(f"Traffic didn't reach all members after "
                               f"#{attempt.number} attempts and "
                               f"{attempt.elapsed_time} seconds")
-                if attempt.is_last:
-                    raise
+            if attempt.is_last:
+                raise
+        except (octavia.TrafficTimeoutError,
+                sh.ShellCommandFailed):
+            LOG.exception(f"Traffic didn't reach all members after "
+                          f"#{attempt.number} attempts and "
+                          f"{attempt.elapsed_time} seconds")
+            if attempt.is_last:
+                raise
 
 
 @neutron.skip_unless_is_ovn()
@@ -93,7 +105,7 @@ class OctaviaOVNProviderTrafficTest(testtools.TestCase):
     Create an OVN provider load balancer with 2 members that run a server
     application,
     Create a client that is connected to the load balancer VIP port via FIP,
-    Generate network traffic from the client to the load balanacer via ssh.
+    Generate TCP network traffic from the client to the load balancer FIP.
     """
     loadbalancer_stack = tobiko.required_fixture(
         stacks.OVNIPv4LoadBalancerStack)
@@ -113,69 +125,12 @@ class OctaviaOVNProviderTrafficTest(testtools.TestCase):
 
         self.loadbalancer_stack.wait_for_octavia_service()
 
-    def test_ssh_traffic(self):
-        """SSH every member server to get its hostname using a load balancer
+    def test_source_ip_port_traffic(self):
+        """Send traffic to the load balancer FIP to test source ip port
         """
-        username: typing.Optional[str] = None
-        password: typing.Optional[str] = None
-        missing_replies = set()
-
-        for member_server in [self.listener_stack.server_stack,
-                              self.listener_stack.other_server_stack]:
-            ssh_client = member_server.ssh_client
-            hostname = sh.get_hostname(ssh_client=ssh_client)
-            missing_replies.add(hostname)
-            if username is None:
-                username = member_server.username
-            else:
-                self.assertEqual(username,
-                                 member_server.username,
-                                 "Not all member servers have the same "
-                                 "username to login with")
-            if password is None:
-                password = member_server.password
-            else:
-                self.assertEqual(password, member_server.password,
-                                 "Not all member servers have the same "
-                                 "password to login with")
-
-        # Get SSH client to the load balancer virtual IP
-        ssh_client = ssh.ssh_client(
-            host=self.loadbalancer_stack.floating_ip_address,
-            port=self.listener_stack.lb_port,
-            username=username,
-            password=password)
-
-        replies = []
-        for attempt in tobiko.retry(timeout=120.):
-            LOG.debug(f"SSH to member server by using the load balancer "
-                      f"(login='{ssh_client.login}', attempt={attempt})...")
-
-            with ssh_client:  # disconnect after every loop
-                hostname = sh.ssh_hostname(ssh_client=ssh_client)
-            try:
-                missing_replies.remove(hostname)
-            except KeyError:
-                self.assertIn(hostname, replies,
-                              f"Unexpected hostname reached: {hostname}")
-            replies.append(hostname)
-            if missing_replies:
-                LOG.debug('Reached member server(s):\n'
-                          f'{pretty_replies(replies)}')
-                if attempt.is_last:
-                    self.fail('Unreached member server(s): {missing_replies}')
-                else:
-                    LOG.debug('Waiting for reaching remaining server(s)... '
-                              f'{missing_replies}')
-            else:
-                LOG.debug('All member servers reached:\n'
-                          f'{pretty_replies(replies)}')
-                break
-        else:
-            raise RuntimeError('Broken retry loop')
-
-
-def pretty_replies(replies: typing.Iterable[str]):
-    return json.dumps(collections.Counter(replies),
-                      indent=4,
-                      sort_keys=True)
+        _test_traffic(
+            pool_id=self.listener_stack.pool_id,
+            ip_address=self.loadbalancer_stack.floating_ip_address,
+            lb_algorithm=self.listener_stack.lb_algorithm,
+            protocol=self.listener_stack.lb_protocol,
+            port=self.listener_stack.lb_port)