SERVER-29522 better error codes when running concurrent tests in resmoke

2 files changed, 152 insertions, 70 deletions

diff --git a/buildscripts/resmokelib/testing/job.py b/buildscripts/resmokelib/testing/job.py
index 4d17152b17d..4507a07a0dd 100644
--- a/buildscripts/resmokelib/testing/job.py
+++ b/buildscripts/resmokelib/testing/job.py
@@ -94,7 +94,7 @@ class Job(object):
         Calls the before/after test hooks and executes 'test'.
         """
 
-        test.configure(self.fixture, config.NUM_CLIENTS_PER_FIXTURE)
+        test.configure(self.fixture)
         self._run_hooks_before_tests(test)
 
         test(self.report)
diff --git a/buildscripts/resmokelib/testing/testcases/jstest.py b/buildscripts/resmokelib/testing/testcases/jstest.py
index c1c93925936..aeda0f71e92 100644
--- a/buildscripts/resmokelib/testing/testcases/jstest.py
+++ b/buildscripts/resmokelib/testing/testcases/jstest.py
@@ -14,54 +14,44 @@ from . import interface
 from ... import config
 from ... import core
 from ... import utils
+from ...utils import registry
 
 
-class JSTestCase(interface.TestCase):
+class _SingleJSTestCase(interface.TestCase):
     """
     A jstest to execute.
     """
 
-    REGISTERED_NAME = "js_test"
-
-    # A wrapper for the thread class that lets us propagate exceptions.
-    class ExceptionThread(threading.Thread):
-        def __init__(self, my_target, my_args):
-            threading.Thread.__init__(self, target=my_target, args=my_args)
-            self.err = None
-
-        def run(self):
-            try:
-                threading.Thread.run(self)
-            except:
-                self.err = sys.exc_info()[1]
-            else:
-                self.err = None
-
-        def _get_exception(self):
-            return self.err
-
-    DEFAULT_CLIENT_NUM = 1
+    REGISTERED_NAME = registry.LEAVE_UNREGISTERED
 
     def __init__(self,
                  logger,
                  js_filename,
                  shell_executable=None,
-                 shell_options=None,
-                 test_kind="JSTest"):
-        """Initializes the JSTestCase with the JS file to run."""
+                 shell_options=None):
+        """
+        Initializes the _SingleJSTestCase with the JS file to run.
+        """
 
-        interface.TestCase.__init__(self, logger, test_kind, js_filename)
+        interface.TestCase.__init__(self, logger, "JSTest", js_filename)
 
         # Command line options override the YAML configuration.
         self.shell_executable = utils.default_if_none(config.MONGO_EXECUTABLE, shell_executable)
 
         self.js_filename = js_filename
         self.shell_options = utils.default_if_none(shell_options, {}).copy()
-        self.num_clients = JSTestCase.DEFAULT_CLIENT_NUM
 
-    def configure(self, fixture, num_clients=DEFAULT_CLIENT_NUM, *args, **kwargs):
+    def configure(self, fixture, *args, **kwargs):
         interface.TestCase.configure(self, fixture, *args, **kwargs)
 
+    def configure_shell(self):
+        """
+        Sets up the global variables for the shell, and data/ directory for the mongod.
+
+        configure_shell() only needs to be called once per test. Therefore if creating multiple
+        _SingleJSTestCase instances to be run in parallel, only call configure_shell() on one of
+        them.
+        """
         global_vars = self.shell_options.get("global_vars", {}).copy()
         data_dir = self._get_data_dir(global_vars)
 
@@ -86,16 +76,14 @@ class JSTestCase(interface.TestCase):
             global_vars["MongoRunner.mongoShellPath"] = self.shell_executable
 
         test_data = global_vars.get("TestData", {}).copy()
-        test_data["minPort"] = core.network.PortAllocator.min_test_port(fixture.job_num)
-        test_data["maxPort"] = core.network.PortAllocator.max_test_port(fixture.job_num)
+        test_data["minPort"] = core.network.PortAllocator.min_test_port(self.fixture.job_num)
+        test_data["maxPort"] = core.network.PortAllocator.max_test_port(self.fixture.job_num)
 
         global_vars["TestData"] = test_data
         self.shell_options["global_vars"] = global_vars
 
         shutil.rmtree(data_dir, ignore_errors=True)
 
-        self.num_clients = num_clients
-
         try:
             os.makedirs(data_dir)
         except os.error:
@@ -130,34 +118,78 @@ class JSTestCase(interface.TestCase):
                             config.MONGO_RUNNER_SUBDIR)
 
     def run_test(self):
-        threads = []
         try:
-            # Don't thread if there is only one client.
-            if self.num_clients == 1:
-                shell = self._make_process(self.logger)
-                self._execute(shell)
-            else:
-                # If there are multiple clients, make a new thread for each client.
-                for i in xrange(self.num_clients):
-                    t = self.ExceptionThread(my_target=self._run_test_in_thread, my_args=[i])
-                    t.start()
-                    threads.append(t)
+            shell = self._make_process()
+            self._execute(shell)
         except self.failureException:
             raise
         except:
             self.logger.exception("Encountered an error running jstest %s.", self.basename())
             raise
-        finally:
-            for t in threads:
-                t.join()
-            for t in threads:
-                if t._get_exception() is not None:
-                    raise t._get_exception()
-
-    def _make_process(self, logger=None, thread_id=0):
-        # Since _make_process() is called by each thread, we make a shallow copy of the mongo shell
-        # options to avoid modifying the shared options for the JSTestCase.
-        shell_options = self.shell_options.copy()
+
+    def _make_process(self):
+        return core.programs.mongo_shell_program(
+            self.logger,
+            executable=self.shell_executable,
+            filename=self.js_filename,
+            connection_string=self.fixture.get_driver_connection_url(),
+            **self.shell_options)
+
+
+class JSTestCase(interface.TestCase):
+    """
+    A wrapper for several copies of a SingleJSTest to execute.
+    """
+
+    REGISTERED_NAME = "js_test"
+
+    class ThreadWithException(threading.Thread):
+        """
+        A wrapper for the thread class that lets us propagate exceptions.
+        """
+
+        def __init__(self, *args, **kwargs):
+            threading.Thread.__init__(self, *args, **kwargs)
+            self.exc_info = None
+
+        def run(self):
+            try:
+                threading.Thread.run(self)
+            except:
+                self.exc_info = sys.exc_info()
+
+    def __init__(self,
+                 logger,
+                 js_filename,
+                 shell_executable=None,
+                 shell_options=None,
+                 test_kind="JSTest"):
+        """
+        Initializes the JSTestCase with the JS file to run.
+        """
+
+        interface.TestCase.__init__(self, logger, test_kind, js_filename)
+
+        self.num_clients = config.NUM_CLIENTS_PER_FIXTURE
+        self.test_case_template = _SingleJSTestCase(logger, js_filename, shell_executable,
+                                                    shell_options)
+
+    def configure(self, fixture, *args, **kwargs):
+        interface.TestCase.configure(self, fixture, *args, **kwargs)
+        self.test_case_template.configure(fixture, *args, **kwargs)
+        self.test_case_template.configure_shell()
+
+    def _make_process(self):
+        # This function should only be called by interface.py's as_command().
+        return self.test_case_template._make_process()
+
+    def _get_shell_options_for_thread(self, thread_id):
+        """
+        Get shell_options with an initialized TestData object for given thread.
+        """
+
+        # We give each _SingleJSTestCase its own copy of the shell_options.
+        shell_options = self.test_case_template.shell_options.copy()
         global_vars = shell_options["global_vars"].copy()
         test_data = global_vars["TestData"].copy()
 
@@ -172,20 +204,70 @@ class JSTestCase(interface.TestCase):
         global_vars["TestData"] = test_data
         shell_options["global_vars"] = global_vars
 
-        # If logger is none, it means that it's not running in a thread and thus logger should be
-        # set to self.logger.
-        logger = utils.default_if_none(logger, self.logger)
+        return shell_options
 
-        return core.programs.mongo_shell_program(
-            logger,
-            executable=self.shell_executable,
-            filename=self.js_filename,
-            connection_string=self.fixture.get_driver_connection_url(),
-            **shell_options)
-
-    def _run_test_in_thread(self, thread_id):
-        # Make a logger for each thread. When this method gets called self.logger has been
-        # overridden with a TestLogger instance by the TestReport in the startTest() method.
-        logger = self.logger.new_test_thread_logger(self.test_kind, str(thread_id))
-        shell = self._make_process(logger, thread_id)
-        self._execute(shell)
+    def _create_test_case_for_thread(self, logger, thread_id):
+        """
+        Create and configure a _SingleJSTestCase to be run in a separate thread.
+        """
+
+        shell_options = self._get_shell_options_for_thread(thread_id)
+        test_case = _SingleJSTestCase(logger,
+                                      self.test_case_template.js_filename,
+                                      self.test_case_template.shell_executable,
+                                      shell_options)
+
+        test_case.configure(self.fixture)
+        return test_case
+
+    def _run_single_copy(self):
+        test_case = self._create_test_case_for_thread(self.logger, thread_id=0)
+        try:
+            test_case.run_test()
+            # If there was an exception, it will be logged in test_case's run_test function.
+        finally:
+            self.return_code = test_case.return_code
+
+    def _run_multiple_copies(self):
+        threads = []
+        test_cases = []
+        try:
+            # If there are multiple clients, make a new thread for each client.
+            for thread_id in xrange(self.num_clients):
+                logger = self.logger.new_test_thread_logger(self.test_kind, str(thread_id))
+                test_case = self._create_test_case_for_thread(logger, thread_id)
+                test_cases.append(test_case)
+
+                thread = self.ThreadWithException(target=test_case.run_test)
+                threads.append(thread)
+                thread.start()
+        except:
+            self.logger.exception("Encountered an error starting threads for jstest %s.",
+                                  self.basename())
+            raise
+        finally:
+            for thread in threads:
+                thread.join()
+
+            # Go through each test's return code and store the first nonzero one if it exists.
+            return_code = 0
+            for test_case in test_cases:
+                if test_case.return_code != 0:
+                    return_code = test_case.return_code
+                    break
+            self.return_code = return_code
+
+            for (thread_id, thread) in enumerate(threads):
+                if thread.exc_info is not None:
+                    if not isinstance(thread.exc_info[1], self.failureException):
+                        self.logger.error(
+                            "Encountered an error inside thread %d running jstest %s.",
+                            thread_id, self.basename(),
+                            exc_info=thread.exc_info)
+                    raise thread.exc_info
+
+    def run_test(self):
+        if self.num_clients == 1:
+            self._run_single_copy()
+        else:
+            self._run_multiple_copies()