Repository: elastic/rally

Issue:
Return both exit code, and output in process module
In https://github.com/elastic/rally/pull/1591 it became clear that it'd be nice to modify an existing function in our [`process` module](https://github.com/elastic/rally/blob/master/esrally/utils/process.py) to return both the stdout/stderr of the subprocess as well as the exit code.

Discussion:
favilo: > LGTM, but #1595 lacks explicit clarification of what's exactly the win. I think we can find it [here](https://github.com/elastic/rally/pull/1591#discussion_r986874797).
> 
> WDYT about using the modified `run_subprocess()` to avoid the double call in [is_branch()](https://github.com/elastic/rally/blob/5e6b1bc869f0406024bd796b825511dc13a6d133/esrally/utils/git.py#L52) as part of this PR? On quick scan that's probably the only improvement in the `git` module we can make as other methods use different commands in a sequence?

What do you think about the latest change?
gbanasiak: > What do you think about the latest change?

Thank you for iterating. https://github.com/elastic/rally/pull/1828/commits/21caafa2f0ffd405ff93fc26d5fed706fd624b57 eliminates duplication but made me realize we're missing the logging now which I think should be kept for troubleshooting purposes. We could add snowflake logging at `git` module level, or add logging option to `run_subprocess()`. Or, rewrite `run_subprocess_with_logging` to be based on `subprocess.run` and return `CompletedProcess`, but this would extend the scope of the PR. Which option do you find the most elegant?




favilo: > > What do you think about the latest change?
> 
> Thank you for iterating. [21caafa](https://github.com/elastic/rally/commit/21caafa2f0ffd405ff93fc26d5fed706fd624b57) eliminates duplication but made me realize we're missing the logging now which I think should be kept for troubleshooting purposes. We could add snowflake logging at `git` module level, or add logging option to `run_subprocess()`. Or, rewrite `run_subprocess_with_logging()` to be based on `subprocess.run()` and return `CompletedProcess`, but this would extend the scope of the PR. Which option do you find the most elegant?

Adding logging option to `run_subprocess()` feels like the least effort solution here. I wouldn't mind having `run_subprocess_with_logging()` return a Popen object which would also have `.returncode` and `.stdout`, but that has a _lot_ more uses, and since we are already logging, do we need to keep track of the output separately?

I could certainly be convinced that the latter one is the correct approach, but maybe not in this small PR?
gbanasiak: IT tests are failing as it turns out `run_subprocess_with_logging()` is also used in https://github.com/elastic/rally-teams, e.g. [here](https://github.com/elastic/rally-teams/blob/9ea8b51e2dec6aca3181ea7394f66c242da8b46f/cars/v1/x_pack/base/config.py#L60C5-L66C61). I wasn't aware of that. This changes the picture considerably. I think we should go for a new method in the `process` module for backward compatibility. The problem is https://github.com/elastic/rally-teams code can be interpreted by a range of Rally versions.
favilo: Huh, I see the IT tests failed, but looking at the logs, I don't actually see any failures noted.

Patch:
diff --git a/esrally/mechanic/supplier.py b/esrally/mechanic/supplier.py
index d40773cb..0fa2c486 100644
--- a/esrally/mechanic/supplier.py
+++ b/esrally/mechanic/supplier.py
@@ -810,7 +810,7 @@ class Builder:
         console.info("Creating installable binary from source files")
         self.logger.info("Running build command [%s]", build_cmd)
 
-        if process.run_subprocess(build_cmd):
+        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:
diff --git a/esrally/utils/git.py b/esrally/utils/git.py
index 7d2da029..c55e3aee 100644
--- a/esrally/utils/git.py
+++ b/esrally/utils/git.py
@@ -26,7 +26,8 @@ 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), quiet=True
+            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")
             if version:
@@ -51,13 +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 = process.run_subprocess_with_logging_and_output(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: process.run_subprocess_with_logging(show_ref_cmd)):
+    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 = process.run_subprocess_with_output(show_ref_cmd)
+    ref = completed_process.stdout.split("\n")
     if "refs/tags" in ref[0]:
         return False
 
diff --git a/esrally/utils/process.py b/esrally/utils/process.py
index 5d9e3bb4..18eb22c4 100644
--- a/esrally/utils/process.py
+++ b/esrally/utils/process.py
@@ -20,15 +20,25 @@ import os
 import shlex
 import subprocess
 import time
+from typing import Callable, Dict, List
 
 import psutil
 
+LogLevel = int
+FileId = int
 
-def run_subprocess(command_line):
+
+def run_subprocess(command_line: str) -> int:
+    """
+    Runs the provided command line in a subprocess.
+
+    :param command_line: The command line of the subprocess to launch.
+    :return: The process' return code
+    """
     return subprocess.call(command_line, shell=True)
 
 
-def run_subprocess_with_output(command_line, env=None):
+def run_subprocess_with_output(command_line: str, env: Dict[str, str] = None) -> List[str]:
     logger = logging.getLogger(__name__)
     logger.debug("Running subprocess [%s] with output.", command_line)
     command_line_args = shlex.split(command_line)
@@ -44,7 +54,7 @@ def run_subprocess_with_output(command_line, env=None):
     return lines
 
 
-def exit_status_as_bool(runnable, quiet=False):
+def exit_status_as_bool(runnable: Callable[[], int], quiet: bool = False) -> bool:
     """
 
     :param runnable: A runnable returning an int as exit status assuming ``0`` is meaning success.
@@ -60,7 +70,14 @@ def exit_status_as_bool(runnable, quiet=False):
         return False
 
 
-def run_subprocess_with_logging(command_line, header=None, level=logging.INFO, stdin=None, env=None, detach=False):
+def run_subprocess_with_logging(
+    command_line: str,
+    header: str = None,
+    level: LogLevel = logging.INFO,
+    stdin: FileId = None,
+    env: Dict[str, str] = None,
+    detach: bool = False,
+) -> int:
     """
     Runs the provided command line in a subprocess. All output will be captured by a logger.
 
@@ -98,7 +115,53 @@ def run_subprocess_with_logging(command_line, header=None, level=logging.INFO, s
     return command_line_process.returncode
 
 
-def is_rally_process(p):
+def run_subprocess_with_logging_and_output(
+    command_line: str,
+    header: str = None,
+    level: LogLevel = logging.INFO,
+    stdin: FileId = None,
+    env: Dict[str, str] = None,
+    detach: bool = False,
+) -> subprocess.CompletedProcess:
+    """
+    Runs the provided command line in a subprocess. All output will be captured by a logger.
+
+    :param command_line: The command line of the subprocess to launch.
+    :param header: An optional header line that should be logged (this will be logged on info level, regardless of the defined log level).
+    :param level: The log level to use for output (default: logging.INFO).
+    :param stdin: The stdout object returned by subprocess.Popen(stdout=PIPE) allowing chaining of shell operations with pipes
+      (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 process exit code as an int.
+    """
+    logger = logging.getLogger(__name__)
+    logger.debug("Running subprocess [%s] with logging.", command_line)
+    command_line_args = shlex.split(command_line)
+    pre_exec = os.setpgrp if detach else None
+    if header is not None:
+        logger.info(header)
+
+    # pylint: disable=subprocess-popen-preexec-fn
+    completed = subprocess.run(
+        command_line_args,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        env=env,
+        check=False,
+        stdin=stdin if stdin else None,
+        preexec_fn=pre_exec,
+    )
+
+    for stdout in completed.stdout.splitlines():
+        logger.log(level=level, msg=stdout)
+
+    logger.debug("Subprocess [%s] finished with return code [%s].", command_line, str(completed.returncode))
+    return completed
+
+
+def is_rally_process(p: psutil.Process) -> bool:
     return (
         p.name() == "esrally"
         or p.name() == "rally"
@@ -110,14 +173,14 @@ def is_rally_process(p):
     )
 
 
-def find_all_other_rally_processes():
+def find_all_other_rally_processes() -> List[psutil.Process]:
     others = []
     for_all_other_processes(is_rally_process, others.append)
     return others
 
 
-def kill_all(predicate):
-    def kill(p):
+def kill_all(predicate: Callable[[psutil.Process], bool]) -> None:
+    def kill(p: psutil.Process):
         logging.getLogger(__name__).info("Killing lingering process with PID [%s] and command line [%s].", p.pid, p.cmdline())
         p.kill()
         # wait until process has terminated, at most 3 seconds. Otherwise we might run into race conditions with actor system
@@ -132,7 +195,7 @@ def kill_all(predicate):
     for_all_other_processes(predicate, kill)
 
 
-def for_all_other_processes(predicate, action):
+def for_all_other_processes(predicate: Callable[[psutil.Process], bool], action: Callable[[psutil.Process], None]) -> None:
     # no harakiri please
     my_pid = os.getpid()
     for p in psutil.process_iter():
@@ -143,8 +206,8 @@ def for_all_other_processes(predicate, action):
             pass
 
 
-def kill_running_rally_instances():
-    def rally_process(p):
+def kill_running_rally_instances() -> None:
+    def rally_process(p: psutil.Process) -> bool:
         return (
             p.name() == "esrally"
             or p.name() == "rally"


Test patch:
diff --git a/tests/utils/process_test.py b/tests/utils/process_test.py
index b54916d7..15c350c8 100644
--- a/tests/utils/process_test.py
+++ b/tests/utils/process_test.py
@@ -198,3 +198,12 @@ class TestProcess:
         assert rally_process_mac.killed
         assert not own_rally_process.killed
         assert not night_rally_process.killed
+
+
+def test_run_subprocess():
+    cmd = "ls . not-a-file"
+    completed_process = process.run_subprocess_with_logging_and_output(cmd)
+
+    assert completed_process.returncode != 0
+    assert completed_process.stdout != ""
+    assert completed_process.stderr is None