Run Validations with ThreadPoolExecutor

The validations are sequentially executed through ansible-playbook and
could take some time to complete. This patch adds support of running all
the validations in parallel through a ThreadPoolExecutor in order to
save processing time.

Without this patch, it takes almost ~5min to run all the validations but
only ~1min10 with the ThreadPoolExecutor.

- Add --worker/-w argument to give the maximum number of threads that
  can be used to execute the given validations
- Python 2.7: use futures backport instead, ThreadPoolExecutor is default
- Use six in tripleo_validator.py for python 3 compatibility

Change-Id: Ia805a556bc26700a3eb520ed72e90b37546901b8
Signed-off-by: Gael Chamoulaud <gchamoul@redhat.com>

lower-constraints.txt

@@ -25,6 +25,7 @@ extras==1.0.0
 fasteners==0.7.0
 fixtures==3.0.0
 flake8==2.5.5
+futures==3.0.0
 futurist==1.2.0
 gitdb==0.6.4
 GitPython==1.0.1

requirements.txt

@@ -18,3 +18,4 @@ osc-lib>=1.8.0 # Apache-2.0
 websocket-client>=0.44.0 # LGPLv2+
 tripleo-common>=10.7.0 # Apache-2.0
 cryptography>=2.1 # BSD/Apache-2.0
+futures>=3.0.0;python_version=='2.7' or python_version=='2.6' # BSD

tripleoclient/v1/tripleo_validator.py

@@ -18,8 +18,10 @@ import json
 import logging
 import os
 import pwd
+import six
 import sys
 
+from concurrent.futures import ThreadPoolExecutor
 from osc_lib.command import command
 from osc_lib.i18n import _
 
@@ -212,6 +214,16 @@ class TripleOValidatorRun(command.Command):
             help=_("Execute the validations using Mistral")
         )
 
+        parser.add_argument(
+            '--workers', '-w',
+            metavar='N',
+            dest='workers',
+            default=1,
+            type=int,
+            help=_("The maximum number of threads that can "
+                   "be used to execute the given validations")
+        )
+
         extra_vars_group = parser.add_mutually_exclusive_group(required=False)
 
         extra_vars_group.add_argument(
@@ -302,6 +314,23 @@ class TripleOValidatorRun(command.Command):
                 out.get('validation_name'),
                 oooutils.indent(out.get('stdout'))))
 
+    def _run_ansible(self, logger, plan, workdir, log_path_dir, playbook,
+                     inventory, retries, output_callback, extra_vars,
+                     python_interpreter, gathering_policy):
+        rc, output = oooutils.run_ansible_playbook(
+            logger=logger,
+            plan=plan,
+            workdir=workdir,
+            log_path_dir=log_path_dir,
+            playbook=playbook,
+            inventory=inventory,
+            retries=retries,
+            output_callback=output_callback,
+            extra_vars=extra_vars,
+            python_interpreter=python_interpreter,
+            gathering_policy=gathering_policy)
+        return rc, output
+
     def _run_validator_run(self, parsed_args):
         clients = self.app.client_manager
         LOG = logging.getLogger(__name__ + ".ValidationsRunAnsible")
@@ -343,10 +372,11 @@ class TripleOValidatorRun(command.Command):
 
         failed_val = False
 
-        for playbook in playbooks:
-            try:
-                LOG.debug(_('Running the validations with Ansible'))
-                rc, output = oooutils.run_ansible_playbook(
+        with ThreadPoolExecutor(max_workers=parsed_args.workers) as executor:
+            LOG.debug(_('Running the validations with Ansible'))
+            tasks_exec = {
+                executor.submit(
+                    self._run_ansible,
                     logger=LOG,
                     plan=parsed_args.plan,
                     workdir=constants.ANSIBLE_VALIDATION_DIR,
@@ -357,13 +387,18 @@ class TripleOValidatorRun(command.Command):
                     output_callback='validation_output',
                     extra_vars=extra_vars_input,
                     python_interpreter=python_interpreter,
-                    gathering_policy='explicit')
-                print('[SUCCESS] - {}\n{}'.format(playbook,
-                                                  oooutils.indent(output)))
+                    gathering_policy='explicit'): playbook
+                for playbook in playbooks
+            }
+
+        for tk, pl in six.iteritems(tasks_exec):
+            try:
+                rc, output = tk.result()
+                print('[SUCCESS] - {}\n{}'.format(pl, oooutils.indent(output)))
             except Exception as e:
                 failed_val = True
                 LOG.error('[FAILED] - {}\n{}'.format(
-                    playbook, oooutils.indent(e.args[0])))
+                    pl, oooutils.indent(e.args[0])))
 
         LOG.debug(_('Removing static tripleo ansible inventory file'))
         oooutils.cleanup_tripleo_ansible_inventory_file(