ovn-tester: Allow tests to manually end iterations.

Until this point, when running a qps_test, the ovn_context code would
automatically take care of marking an iteration as started or ended.

This is a problem, though, for tests that wait until all iterations have
completed to determine if ports came up within the configured time
limit. The issue is that the ovn_context will mark the iteration as
completed, potentially logging the iteration as successful. However,
after this has been logged, the test could mark the iteration as failed
if it turns out the port did not come up within the configured time
limit.

The solution here is to allow for tests to override the default behavior
by letting them mark the iteration as completed. To do this, the
qps_test() function now accepts a kwargs parameter, and setting
end_iteration=False will allow for the individual test to mark
iterations as complete instead of having it done automatically by the
ovn_context code.

For convenience, a new all_iterations_completed() method is added to
ovn_context so the tests don't have to keep track of iterations to mark
them as complete.

Signed-off-by: Mark Michelson <[email protected]>

ovn-tester/ovn_context.py

@@ -2,6 +2,7 @@
 import ovn_stats
 import time
 import asyncio
+from collections import Counter
 
 log = logging.getLogger(__name__)
 
@@ -49,7 +50,8 @@ def iteration_completed(self, iteration):
         asynchronously since task starts and ends can overlap.'''
         iteration.end()
         duration = iteration.end_time - iteration.start_time
-        ovn_stats.add(ITERATION_STAT_NAME, duration, iteration.failed)
+        ovn_stats.add(ITERATION_STAT_NAME, duration, iteration.failed,
+                      iteration)
         self.iterations = {task_name: it for task_name, it in
                            self.iterations.items() if it != iteration}
         if self.test:
@@ -58,6 +60,18 @@ def iteration_completed(self, iteration):
                 f'Context {self.test_name}, Iteration {iteration.num}, '
                 f'Result: {"FAILURE" if iteration.failed else "SUCCESS"}')
 
+    def all_iterations_completed(self):
+        if self.iteration_singleton is not None:
+            # Weird, but you do you man.
+            self.iteration_completed(self.iteration_singleton)
+            return
+
+        # self.iterations may have the same iteration value for multiple
+        # keys. We need to first get the unique list of iterations.
+        iter_list = Counter(self.iterations.values())
+        for iteration in iter_list:
+            self.iteration_completed(iteration)
+
     def create_task(self, coro, iteration=None):
         '''Create a task to run in this context.'''
         if iteration is None:
@@ -66,23 +80,24 @@ def create_task(self, coro, iteration=None):
         self.iterations[task.get_name()] = iteration
         return task
 
-    async def qps_test(self, qps, coro, *args):
+    async def qps_test(self, qps, coro, *args, **kwargs):
         tasks = []
         for i in range(self.max_iterations):
             iteration = ContextIteration(i, self)
             tasks.append(self.create_task(
-                self.qps_task(iteration, coro, *args), iteration)
+                self.qps_task(iteration, coro, *args, **kwargs), iteration)
             )
             # Use i+1 so that we don't sleep on task 0 and so that
             # we sleep after 20 iterations instead of 21.
             if (i + 1) % qps == 0:
                 await asyncio.sleep(1)
         await asyncio.gather(*tasks)
 
-    async def qps_task(self, iteration, coro, *args):
+    async def qps_task(self, iteration, coro, *args, **kwargs):
         await self.iteration_started(iteration)
         await coro(*args)
-        self.iteration_completed(iteration)
+        if kwargs.get('end_iteration', True):
+            self.iteration_completed(iteration)
 
 
 def get_current_iteration():

ovn-tester/ovn_tester.py

@@ -242,8 +242,10 @@ async def run(self, ovn, bringup_cfg):
             await ovn.create_cluster_join_switch("ls-join")
             await ovn.create_cluster_load_balancer("lb-cluster")
             await ctx.qps_test(bringup_cfg.queries_per_second,
-                               self.provisioner, ovn, bringup_cfg)
+                               self.provisioner, ovn, bringup_cfg,
+                               end_iteration=False)
             await ovn.wait_for_ports_up(self.port_iters)
+            ctx.all_iterations_completed()
 
 
 async def main(global_cfg, cluster_cfg, brex_cfg, bringup_cfg):

ovn-tester/tests/cluster_density.py

@@ -58,8 +58,9 @@ async def run(self, ovn, global_cfg):
                      (self.config.n_runs - self.config.n_startup) //
                      self.batch_ratio, test=self) as ctx:
             await ctx.qps_test(self.config.queries_per_second,
-                               self.tester, ovn, all_ns)
+                               self.tester, ovn, all_ns, end_iteration=False)
             await ovn.wait_for_ports_up(self.test_port_iters)
+            ctx.all_iterations_completed()
 
         if not global_cfg.cleanup:
             return

ovn-tester/tests/density_heavy.py

@@ -80,8 +80,10 @@ async def run(self, ovn, global_cfg):
                      (self.config.n_pods - self.config.n_startup) //
                      self.config.batch, test=self) as ctx:
             await ctx.qps_test(self.config.queries_per_second,
-                               self.provisioner, ns, ovn)
+                               self.provisioner, ns, ovn,
+                               end_iteration=False)
             await ovn.wait_for_ports_up(self.test_port_iters)
+            ctx.all_iterations_completed()
 
         if not global_cfg.cleanup:
             return

ovn-tester/tests/density_light.py

@@ -34,8 +34,10 @@ async def run(self, ovn, global_cfg):
         n_iterations = self.config.n_pods - self.config.n_startup
         with Context('density_light', n_iterations, test=self) as ctx:
             await ctx.qps_test(self.config.queries_per_second,
-                               self.provisioner, ns, ovn)
+                               self.provisioner, ns, ovn,
+                               end_iteration=False)
             await ovn.wait_for_ports_up(self.test_port_iters)
+            ctx.all_iterations_completed()
 
         if not global_cfg.cleanup:
             return