Revert all changes, except typing.

Will be adding new function

esrally/mechanic/launcher.py

@@ -55,7 +55,7 @@ def start(self, node_configurations):
     def _start_process(self, binary_path):
         compose_cmd = self._docker_compose(binary_path, "up -d")
 
-        ret = process.run_subprocess_with_logging(compose_cmd).returncode
+        ret = process.run_subprocess_with_logging(compose_cmd)
         if ret != 0:
             msg = f"Docker daemon startup failed with exit code [{ret}]"
             logging.error(msg)

esrally/mechanic/provisioner.py

@@ -363,7 +363,7 @@ def install(self, es_home_path, plugin_url=None):
             self.logger.info("Installing [%s] into [%s]", self.plugin_name, es_home_path)
             install_cmd = '%s install --batch "%s"' % (installer_binary_path, self.plugin_name)
 
-        return_code = process.run_subprocess_with_logging(install_cmd, env=self.env()).returncode
+        return_code = process.run_subprocess_with_logging(install_cmd, env=self.env())
         # see: https://www.elastic.co/guide/en/elasticsearch/plugins/current/_other_command_line_parameters.html
         if return_code == 0:
             self.logger.info("Successfully installed [%s].", self.plugin_name)

esrally/mechanic/supplier.py

@@ -808,7 +808,7 @@ def run(self, command, override_src_dir=None):
         console.info("Creating installable binary from source files")
         self.logger.info("Running build command [%s]", build_cmd)
 
-        if process.run_subprocess(build_cmd).returncode != 0:
+        if process.run_subprocess(build_cmd) != 0:
             msg = f"Executing '{command}' failed. The last 20 lines in the build log file are:\n"
             msg += "=========================================================================================================\n"
             with open(log_file, encoding="utf-8") as f:

esrally/telemetry.py

@@ -351,7 +351,7 @@ def detach_from_node(self, node, running):
             heap_dump_file = os.path.join(self.log_root, f"heap_at_exit_{node.pid}.hprof")
             console.info(f"{self.human_name}: Writing heap dump to [{heap_dump_file}]", logger=self.logger)
             cmd = f"jmap -dump:format=b,file={heap_dump_file} {node.pid}"
-            if process.run_subprocess_with_logging(cmd).returncode:
+            if process.run_subprocess_with_logging(cmd):
                 self.logger.warning("Could not write heap dump to [%s]", heap_dump_file)
 
 

esrally/utils/git.py

@@ -26,7 +26,7 @@ def probed(f):
     def probe(src, *args, **kwargs):
         # Probe for -C
         if not process.exit_status_as_bool(
-            lambda: process.run_subprocess_with_logging(f"git -C {io.escape_path(src)} --version", level=logging.DEBUG).returncode,
+            lambda: process.run_subprocess_with_logging(f"git -C {io.escape_path(src)} --version", level=logging.DEBUG),
             quiet=True,
         ):
             version = process.run_subprocess_with_output("git --version")
@@ -52,14 +52,14 @@ def is_working_copy(src):
 @probed
 def is_branch(src_dir, identifier):
     show_ref_cmd = f"git -C {src_dir} show-ref {identifier}"
-    completed = process.run_subprocess_with_logging(show_ref_cmd)
+    completed_process = process.run_subprocess_with_logging(show_ref_cmd)
 
     # if we get an non-zero exit code, we know that the identifier is not a branch (local or remote)
-    if not process.exit_status_as_bool(lambda: completed.returncode):
+    if not process.exit_status_as_bool(lambda: completed_process.returncode):
         return False
 
     # it's possible the identifier could be a tag, so we explicitly check that here
-    ref = completed.stdout.split("\n")
+    ref = completed_process.stdout.split("\n")
     if "refs/tags" in ref[0]:
         return False
 
@@ -69,32 +69,32 @@ def is_branch(src_dir, identifier):
 def clone(src, *, remote):
     io.ensure_dir(src)
     # Don't swallow subprocess output, user might need to enter credentials...
-    if process.run_subprocess_with_logging("git clone %s %s" % (remote, io.escape_path(src))).returncode:
+    if process.run_subprocess_with_logging("git clone %s %s" % (remote, io.escape_path(src))):
         raise exceptions.SupplyError("Could not clone from [%s] to [%s]" % (remote, src))
 
 
 @probed
 def fetch(src, *, remote):
-    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src)} fetch --prune --tags {remote}").returncode:
+    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src)} fetch --prune --tags {remote}"):
         raise exceptions.SupplyError("Could not fetch source tree from [%s]" % remote)
 
 
 @probed
 def checkout(src_dir, *, branch):
-    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src_dir)} checkout {branch}").returncode:
+    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src_dir)} checkout {branch}"):
         raise exceptions.SupplyError("Could not checkout [%s]. Do you have uncommitted changes?" % branch)
 
 
 @probed
 def checkout_branch(src_dir, remote, branch):
-    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src_dir)} checkout {remote}/{branch}").returncode:
+    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src_dir)} checkout {remote}/{branch}"):
         raise exceptions.SupplyError("Could not checkout [%s]. Do you have uncommitted changes?" % branch)
 
 
 @probed
 def rebase(src_dir, *, remote, branch):
     checkout(src_dir, branch=branch)
-    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src_dir)} rebase {remote}/{branch}").returncode:
+    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src_dir)} rebase {remote}/{branch}"):
         raise exceptions.SupplyError("Could not rebase on branch [%s]" % branch)
 
 
@@ -110,13 +110,13 @@ def pull_ts(src_dir, ts, *, remote, branch):
     clean_src = io.escape_path(src_dir)
     rev_list_command = f'git -C {clean_src} rev-list -n 1 --before="{ts}" --date=iso8601 {remote}/{branch}'
     revision = process.run_subprocess_with_output(rev_list_command)[0].strip()
-    if process.run_subprocess_with_logging(f"git -C {clean_src} checkout {revision}").returncode:
+    if process.run_subprocess_with_logging(f"git -C {clean_src} checkout {revision}"):
         raise exceptions.SupplyError("Could not checkout source tree for timestamped revision [%s]" % ts)
 
 
 @probed
 def checkout_revision(src_dir, *, revision):
-    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src_dir)} checkout {revision}").returncode:
+    if process.run_subprocess_with_logging(f"git -C {io.escape_path(src_dir)} checkout {revision}"):
         raise exceptions.SupplyError("Could not checkout source tree for revision [%s]" % revision)
 
 

esrally/utils/jvm.py

@@ -34,7 +34,7 @@ def supports_option(java_home, option):
     :param option: The JVM option or combination of JVM options (separated by spaces) to check.
     :return: True iff the provided ``option`` is supported on this JVM.
     """
-    return process.exit_status_as_bool(lambda: process.run_subprocess_with_logging(f"{_java(java_home)} {option} -version").returncode)
+    return process.exit_status_as_bool(lambda: process.run_subprocess_with_logging(f"{_java(java_home)} {option} -version"))
 
 
 def system_property(java_home, system_property_name):

esrally/utils/process.py

@@ -25,14 +25,14 @@
 import psutil
 
 
-def run_subprocess(command_line: str) -> subprocess.CompletedProcess:
+def run_subprocess(command_line: str) -> int:
     """
     Runs the provided command line in a subprocess. All output will be returned in the `CompletedProcess.stdout` field.
 
     :param command_line: The command line of the subprocess to launch.
-    :return: The `CompletedProcess` object for the subprocess. `.returncode` contains the process' return code
+    :return: The process' return code
     """
-    return subprocess.run(command_line, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False)
+    return subprocess.call(command_line, shell=True)
 
 
 def run_subprocess_with_output(command_line: str, env: Dict[str, str] = None) -> List[str]:
@@ -78,7 +78,7 @@ def run_subprocess_with_logging(
     stdin: FileId = None,
     env: Dict[str, str] = None,
     detach: bool = False,
-) -> subprocess.CompletedProcess:
+) -> int:
     """
     Runs the provided command line in a subprocess. All output will be captured by a logger.
 
@@ -89,7 +89,7 @@ def run_subprocess_with_logging(
       (default: None).
     :param env: Use specific environment variables (default: None).
     :param detach: Whether to detach this process from its parent process (default: False).
-    :return: The `CompletedProcess` object for the subprocess. `.returncode` contains the process' return code
+    :return: The process exit code as an int.
     """
     logger = logging.getLogger(__name__)
     logger.debug("Running subprocess [%s] with logging.", command_line)
@@ -99,22 +99,21 @@ def run_subprocess_with_logging(
         logger.info(header)
 
     # pylint: disable=subprocess-popen-preexec-fn
-    completed = subprocess.run(
+    with subprocess.Popen(
         command_line_args,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
-        text=True,
+        universal_newlines=True,
         env=env,
         stdin=stdin if stdin else None,
-        check=False,
         preexec_fn=pre_exec,
-    )
-
-    for line in completed.stdout.splitlines():
-        logger.log(level=level, msg=line)
+    ) as command_line_process:
+        stdout, _ = command_line_process.communicate()
+        if stdout:
+            logger.log(level=level, msg=stdout)
 
-    logger.debug("Subprocess [%s] finished with return code [%s].", command_line, str(completed.returncode))
-    return completed
+    logger.debug("Subprocess [%s] finished with return code [%s].", command_line, str(command_line_process.returncode))
+    return command_line_process.returncode
 
 
 def is_rally_process(p: psutil.Process) -> bool:

it/__init__.py

@@ -138,9 +138,9 @@ def wait_until_port_is_free(port_number=39200, timeout=120):
 
 
 def check_prerequisites():
-    if process.run_subprocess_with_logging("docker ps").returncode != 0:
+    if process.run_subprocess_with_logging("docker ps") != 0:
         raise AssertionError("Docker must be installed and the daemon must be up and running to run integration tests.")
-    if process.run_subprocess_with_logging("docker-compose --help").returncode != 0:
+    if process.run_subprocess_with_logging("docker-compose --help") != 0:
         raise AssertionError("Docker Compose is required to run integration tests.")
 
 
@@ -253,7 +253,7 @@ def build_docker_image():
         f"-f {ROOT_DIR}/docker/Dockerfiles/Dockerfile-dev {ROOT_DIR}"
     )
 
-    if process.run_subprocess_with_logging(command, env=env_variables).returncode != 0:
+    if process.run_subprocess_with_logging(command, env=env_variables) != 0:
         raise AssertionError("It was not possible to build the docker image from Dockerfile-dev")
 
 

it/docker_dev_image_test.py

@@ -63,9 +63,9 @@ def run_docker_compose_up(test_command):
 
     return process.run_subprocess_with_logging(
         f"docker-compose -f {it.ROOT_DIR}/docker/docker-compose-tests.yml up --abort-on-container-exit", env=env_variables
-    ).returncode
+    )
 
 
 def run_docker_compose_down():
-    if process.run_subprocess_with_logging(f"docker-compose -f {it.ROOT_DIR}/docker/docker-compose-tests.yml down -v").returncode != 0:
+    if process.run_subprocess_with_logging(f"docker-compose -f {it.ROOT_DIR}/docker/docker-compose-tests.yml down -v") != 0:
         raise AssertionError("Failed to stop running containers from docker-compose-tests.yml")

tests/mechanic/launcher_test.py

@@ -346,7 +346,7 @@ class TestDockerLauncher:
     @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):
-        run_subprocess_with_logging.return_value.returncode = 0
+        run_subprocess_with_logging.return_value = 0
         # Docker container id (from docker-compose ps), Docker container id (from docker ps --filter ...)
         run_subprocess_with_output.side_effect = [["de604d0d"], ["de604d0d"]]
         cfg = config.Config()
@@ -385,7 +385,7 @@ def test_starts_container_successfully(self, run_subprocess_with_output, run_sub
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     @mock.patch("esrally.utils.process.run_subprocess_with_output")
     def test_container_not_started(self, run_subprocess_with_output, run_subprocess_with_logging, sleep):
-        run_subprocess_with_logging.return_value.returncode = 0
+        run_subprocess_with_logging.return_value = 0
         # Docker container id (from docker-compose ps), but NO Docker container id (from docker ps --filter...) twice
         run_subprocess_with_output.side_effect = [["de604d0d"], [], []]
         cfg = config.Config()

tests/mechanic/provisioner_test.py

@@ -433,7 +433,7 @@ def test_invokes_hook_no_java_home(self):
 class TestPluginInstaller:
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     def test_install_plugin_successfully(self, installer_subprocess):
-        installer_subprocess.return_value.returncode = 0
+        installer_subprocess.return_value = 0
 
         plugin = team.PluginDescriptor(
             name="unit-test-plugin",
@@ -453,7 +453,7 @@ def test_install_plugin_successfully(self, installer_subprocess):
 
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     def test_install_plugin_with_bundled_jdk(self, installer_subprocess):
-        installer_subprocess.return_value.returncode = 0
+        installer_subprocess.return_value = 0
 
         plugin = team.PluginDescriptor(
             name="unit-test-plugin",
@@ -479,7 +479,7 @@ def test_install_plugin_with_bundled_jdk(self, installer_subprocess):
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     def test_install_unknown_plugin(self, installer_subprocess):
         # unknown plugin
-        installer_subprocess.return_value.returncode = 64
+        installer_subprocess.return_value = 64
 
         plugin = team.PluginDescriptor(name="unknown")
         installer = provisioner.PluginInstaller(plugin, java_home="/usr/local/javas/java8", hook_handler_class=NoopHookHandler)
@@ -496,7 +496,7 @@ def test_install_unknown_plugin(self, installer_subprocess):
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     def test_install_plugin_with_io_error(self, installer_subprocess):
         # I/O error
-        installer_subprocess.return_value.returncode = 74
+        installer_subprocess.return_value = 74
 
         plugin = team.PluginDescriptor(name="simple")
         installer = provisioner.PluginInstaller(plugin, java_home="/usr/local/javas/java8", hook_handler_class=NoopHookHandler)
@@ -513,7 +513,7 @@ def test_install_plugin_with_io_error(self, installer_subprocess):
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     def test_install_plugin_with_unknown_error(self, installer_subprocess):
         # some other error
-        installer_subprocess.return_value.returncode = 12987
+        installer_subprocess.return_value = 12987
 
         plugin = team.PluginDescriptor(name="simple")
         installer = provisioner.PluginInstaller(plugin, java_home="/usr/local/javas/java8", hook_handler_class=NoopHookHandler)

tests/mechanic/supplier_test.py

@@ -182,7 +182,7 @@ class TestBuilder:
     @mock.patch("esrally.utils.jvm.resolve_path")
     def test_build_on_jdk_8(self, jvm_resolve_path, mock_run_subprocess):
         jvm_resolve_path.return_value = (8, "/opt/jdk8")
-        mock_run_subprocess.return_value = mock.Mock(returncode=0)
+        mock_run_subprocess.return_value = False
 
         b = supplier.Builder(src_dir="/src", build_jdk=8, log_dir="logs")
         b.build(["./gradlew clean", "./gradlew assemble"])
@@ -200,7 +200,7 @@ def test_build_on_jdk_8(self, jvm_resolve_path, mock_run_subprocess):
     @mock.patch("esrally.utils.jvm.resolve_path")
     def test_build_on_jdk_10(self, jvm_resolve_path, mock_run_subprocess):
         jvm_resolve_path.return_value = (10, "/opt/jdk10")
-        mock_run_subprocess.return_value = mock.Mock(returncode=0)
+        mock_run_subprocess.return_value = False
 
         b = supplier.Builder(src_dir="/src", build_jdk=8, log_dir="logs")
         b.build(["./gradlew clean", "./gradlew assemble"])

tests/telemetry_test.py

@@ -418,7 +418,7 @@ def test_can_override_options_for_java_9_or_above(self):
 class TestHeapdump:
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     def test_generates_heap_dump(self, run_subprocess_with_logging):
-        run_subprocess_with_logging.return_value.returncode = 0
+        run_subprocess_with_logging.return_value = 0
         heapdump = telemetry.Heapdump("/var/log")
         t = telemetry.Telemetry(enabled_devices=[heapdump.command], devices=[heapdump])
         node = cluster.Node(pid="1234", binary_path="/bin", host_name="localhost", node_name="rally0", telemetry=t)

tests/utils/git_test.py

@@ -133,7 +133,7 @@ def test_list_tags(self):
     @mock.patch("esrally.utils.process.run_subprocess_with_logging")
     def test_git_version_too_old(self, run_subprocess_with_logging, run_subprocess):
         # any non-zero return value will do
-        run_subprocess_with_logging.return_value.returncode = 64
+        run_subprocess_with_logging.return_value = 64
         run_subprocess.return_value = "1.0.0"
 
         with pytest.raises(exceptions.SystemSetupError) as exc: