More resilient node shutdown (#860)

With this commit we make the shutdown procedure more resilient by
handling the case that the Elasticsearch process has already terminated
and moving steps during shutdown with the `stop` subcommand so related
errors happen in related places in the code and also as early as
possible.

esrally/mechanic/launcher.py

@@ -16,7 +16,6 @@
 # under the License.
 import logging
 import os
-import signal
 
 import psutil
 
@@ -207,27 +206,38 @@ def stop(self, nodes, metrics_store):
             self.logger.info("Keeping [%d] nodes on this host running.", len(nodes))
         else:
             self.logger.info("Shutting down [%d] nodes on this host.", len(nodes))
+        stopped_nodes = []
         for node in nodes:
-            proc = psutil.Process(pid=node.pid)
             node_name = node.node_name
             telemetry.add_metadata_for_node(metrics_store, node_name, node.host_name)
+            try:
+                es = psutil.Process(pid=node.pid)
+                node.telemetry.detach_from_node(node, running=True)
+            except psutil.NoSuchProcess:
+                self.logger.warning("No process found with PID [%s] for node [%s].", node.pid, node_name)
+                es = None
 
-            node.telemetry.detach_from_node(node, running=True)
             if not self.keep_running:
-                stop_watch = self._clock.stop_watch()
-                stop_watch.start()
-                try:
-                    os.kill(proc.pid, signal.SIGTERM)
-                    proc.wait(10.0)
-                except ProcessLookupError:
-                    self.logger.warning("No process found with PID [%s] for node [%s]", proc.pid, node_name)
-                except psutil.TimeoutExpired:
-                    self.logger.info("kill -KILL node [%s]", node_name)
+                if es:
+                    stop_watch = self._clock.stop_watch()
+                    stop_watch.start()
                     try:
-                        # kill -9
-                        proc.kill()
-                    except ProcessLookupError:
-                        self.logger.warning("No process found with PID [%s] for node [%s]", proc.pid, node_name)
+                        es.terminate()
+                        es.wait(10.0)
+                        stopped_nodes.append(node)
+                    except psutil.NoSuchProcess:
+                        self.logger.warning("No process found with PID [%s] for node [%s].", es.pid, node_name)
+                    except psutil.TimeoutExpired:
+                        self.logger.info("kill -KILL node [%s]", node_name)
+                        try:
+                            # kill -9
+                            es.kill()
+                            stopped_nodes.append(node)
+                        except psutil.NoSuchProcess:
+                            self.logger.warning("No process found with PID [%s] for node [%s].", es.pid, node_name)
+                    self.logger.info("Done shutting down node [%s] in [%.1f] s.", node_name, stop_watch.split_time())
+
                 node.telemetry.detach_from_node(node, running=False)
-                node.telemetry.store_system_metrics(node, metrics_store)
-                self.logger.info("Done shutting down node [%s] in [%.1f] s.", node_name, stop_watch.split_time())
+            # store system metrics in any case (telemetry devices may derive system metrics while the node is running)
+            node.telemetry.store_system_metrics(node, metrics_store)
+            return stopped_nodes

esrally/mechanic/mechanic.py

@@ -101,6 +101,13 @@ def start(cfg):
 def stop(cfg):
     root_path = paths.install_root(cfg)
     node_config = provisioner.load_node_configuration(root_path)
+    if node_config.build_type == "tar":
+        node_launcher = launcher.ProcessLauncher(cfg)
+    elif node_config.build_type == "docker":
+        node_launcher = launcher.DockerLauncher(cfg)
+    else:
+        raise exceptions.SystemSetupError("Unknown build type [{}]".format(node_config.build_type))
+
     nodes, race_id = _load_node_file(root_path)
 
     cls = metrics.metrics_store_class(cfg)
@@ -125,22 +132,16 @@ def stop(cfg):
         challenge_name=current_race.challenge_name
     )
 
-    if node_config.build_type == "tar":
-        node_launcher = launcher.ProcessLauncher(cfg)
-    elif node_config.build_type == "docker":
-        node_launcher = launcher.DockerLauncher(cfg)
-    else:
-        raise exceptions.SystemSetupError("Unknown build type [{}]".format(node_config.build_type))
-
     node_launcher.stop(nodes, metrics_store)
+    _delete_node_file(root_path)
+
     metrics_store.flush(refresh=True)
     for node in nodes:
         results = metrics.calculate_system_results(metrics_store, node.node_name)
         current_race.add_results(results)
         metrics.results_store(cfg).store_results(current_race)
 
     metrics_store.close()
-    _delete_node_file(root_path)
 
     # TODO: Do we need to expose this as a separate command as well?
     provisioner.cleanup(preserve=cfg.opts("mechanic", "preserve.install"),

tests/mechanic/launcher_test.py

@@ -120,16 +120,23 @@ def wait(self):
 class MockProcess:
     def __init__(self, pid):
         self.pid = pid
-        self.killed = False
 
     def name(self):
         return "p{pid}".format(pid=self.pid)
 
     def wait(self, timeout=None):
         raise psutil.TimeoutExpired(timeout)
 
+    def terminate(self):
+        pass
+
     def kill(self):
-        self.killed = True
+        pass
+
+
+class TerminatedProcess:
+    def __init__(self, pid):
+        raise psutil.NoSuchProcess(pid)
 
 
 def get_metrics_store(cfg):
@@ -146,15 +153,14 @@ def get_metrics_store(cfg):
 
 
 class ProcessLauncherTests(TestCase):
-    @mock.patch('os.kill')
     @mock.patch('subprocess.Popen', new=MockPopen)
     @mock.patch('esrally.mechanic.java_resolver.java_home', return_value=(12, "/java_home/"))
     @mock.patch('esrally.utils.jvm.supports_option', return_value=True)
     @mock.patch('esrally.utils.io.get_size')
     @mock.patch('os.chdir')
     @mock.patch('esrally.mechanic.launcher.wait_for_pidfile', return_value=MOCK_PID_VALUE)
     @mock.patch('psutil.Process', new=MockProcess)
-    def test_daemon_start_stop(self, wait_for_pidfile, chdir, get_size, supports, java_home, kill):
+    def test_daemon_start_stop(self, wait_for_pidfile, chdir, get_size, supports, java_home):
         cfg = config.Config()
         cfg.add(config.Scope.application, "node", "root.dir", "test")
         cfg.add(config.Scope.application, "mechanic", "keep.running", False)
@@ -173,10 +179,37 @@ def test_daemon_start_stop(self, wait_for_pidfile, chdir, get_size, supports, ja
         self.assertEqual(len(nodes), 1)
         self.assertEqual(nodes[0].pid, MOCK_PID_VALUE)
 
-        proc_launcher.stop(nodes, ms)
-        self.assertTrue(kill.called)
+        stopped_nodes = proc_launcher.stop(nodes, ms)
+        # all nodes should be stopped
+        self.assertEqual(nodes, stopped_nodes)
 
-    def test_env_options_order(self):
+    @mock.patch('psutil.Process', new=TerminatedProcess)
+    def test_daemon_stop_with_already_terminated_process(self):
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "node", "root.dir", "test")
+        cfg.add(config.Scope.application, "mechanic", "keep.running", False)
+        cfg.add(config.Scope.application, "telemetry", "devices", [])
+        cfg.add(config.Scope.application, "telemetry", "params", None)
+        cfg.add(config.Scope.application, "system", "env.name", "test")
+
+        ms = get_metrics_store(cfg)
+        proc_launcher = launcher.ProcessLauncher(cfg)
+
+        nodes = [
+            cluster.Node(pid=-1,
+                         binary_path="/bin",
+                         host_name="localhost",
+                         node_name="rally-0",
+                         telemetry=telemetry.Telemetry())
+        ]
+
+        stopped_nodes = proc_launcher.stop(nodes, ms)
+        # no nodes should have been stopped (they were already stopped)
+        self.assertEqual([], stopped_nodes)
+
+    # flight recorder shows a warning for several seconds before continuing
+    @mock.patch("esrally.time.sleep")
+    def test_env_options_order(self, sleep):
         cfg = config.Config()
         cfg.add(config.Scope.application, "mechanic", "keep.running", False)
         cfg.add(config.Scope.application, "system", "env.name", "test")
@@ -233,7 +266,6 @@ def __init__(self, stop_watch):
         def stop_watch(self):
             return self._stop_watch
 
-
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     @mock.patch("esrally.utils.process.run_subprocess_with_output")
     def test_starts_container_successfully(self, run_subprocess_with_output, run_subprocess_with_logging):