drivers_orchestration.py

"""
Run mongo-orchestration and launch a deployment.

Use '--help' for more information.
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import re
import shlex
import shutil
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path, PureWindowsPath

import psutil

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from mongodb_runner import start_mongodb_runner

# Get global values.
HERE = Path(__file__).absolute().parent
EVG_PATH = HERE.parent
DRIVERS_TOOLS = EVG_PATH.parent
LOGGER = logging.getLogger("drivers_orchestration")
logging.basicConfig(level=logging.INFO, format="%(levelname)-8s %(message)s")
PLATFORM = sys.platform.lower()
CRYPT_NAME_MAP = {
    "win32": "mongo_crypt_v1.dll",
    "darwin": "mongo_crypt_v1.dylib",
    "linux": "mongo_crypt_v1.so",
}

# Top level files
URI_TXT = DRIVERS_TOOLS / "uri.txt"
MO_EXPANSION_SH = Path("mo-expansion.sh")
MO_EXPANSION_YML = Path("mo-expansion.yml")


def get_options():
    command = sys.argv[1]
    if command == "run":
        description = __doc__
    else:
        description = f"{sys.argv[1].capitalize()} mongo-orchestration"

    parser = argparse.ArgumentParser(
        description=description, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--verbose", "-v", action="store_true", help="Whether to log at the DEBUG level"
    )
    parser.add_argument(
        "--quiet", "-q", action="store_true", help="Whether to log at the WARNING level"
    )

    if command == "run":
        parser.add_argument(
            "--version",
            default="latest",
            help='The version to download. Use "latest" to download '
            "the newest available version (including release candidates).",
        )
        parser.add_argument(
            "--topology",
            choices=["standalone", "replica_set", "sharded_cluster"],
            help="The topology of the server deployment (defaults to standalone unless another flag like load_balancer is set)",
        )
        parser.add_argument(
            "--auth", action="store_true", help="Whether to add authentication"
        )
        parser.add_argument(
            "--ssl", action="store_true", help="Whether to add TLS configuration"
        )
        parser.add_argument(
            "--local-atlas",
            action="store_true",
            help="Whether to use mongodb-atlas-local to start the server",
        )
        parser.add_argument(
            "--mongodb-runner",
            action="store_true",
            help="Whether to use mongodb-runner to start the server",
        )
        parser.add_argument(
            "--orchestration-file", help="The name of the orchestration config file"
        )

    other_group = parser.add_argument_group("Other options")
    if command == "run":
        other_group.add_argument(
            "--load-balancer",
            action="store_true",
            help="Whether to use a load balancer",
        )
        other_group.add_argument(
            "--auth-aws", action="store_true", help="Whether to use MONGODB-AWS auth"
        )
        other_group.add_argument(
            "--skip-crypt-shared",
            action="store_true",
            help="Whether to skip installing crypt_shared lib",
        )
        other_group.add_argument(
            "--install-legacy-shell",
            action="store_true",
            help="Whether to install the legacy shell",
        )
        other_group.add_argument(
            "--disable-test-commands",
            action="store_true",
            help="Whether to disable test commands",
        )
        other_group.add_argument(
            "--storage-engine",
            choices=["", "mmapv1", "wiredtiger", "inmemory"],
            help="The storage engine to use",
        )
        other_group.add_argument(
            "--require-api-version",
            action="store_true",
            help="Whether to set requireApiVersion",
        )
        other_group.add_argument(
            "--existing-binaries-dir",
            help="A directory containing existing mongodb binaries to use instead of downloading new ones",
        )
        other_group.add_argument(
            "--tls-pem-key-file",
            help="A .pem file that contains the TLS certificate and key for the server",
        )
        other_group.add_argument(
            "--tls-ca-file",
            help="A .pem file that contains the root certificate chain for the server",
        )
        other_group.add_argument(
            "--tls-allow-invalid-certificates",
            action="store_true",
            help="Whether to pass --tlsAllowInvalidCertificates to mongod",
        )
        other_group.add_argument(
            "--arch",
            help="the architecture.  if unspecified, the arch will be inferred.",
        )

    other_group.add_argument(
        "--mongo-orchestration-home", help="The path to mongo-orchestration home"
    )

    if command in ["start", "run"]:
        other_group.add_argument(
            "--mongodb-binaries", help="The path to store the MongoDB binaries"
        )
        other_group.add_argument(
            "--tls-cert-key-file",
            help="A .pem to be used as the tlsCertificateKeyFile option in mongo-orchestration",
        )

    # Get the options, and then allow environment variable overrides.
    opts = parser.parse_args(sys.argv[2:])
    for key in vars(opts).keys():
        env_var = key.upper()
        if env_var == "VERSION":
            env_var = "MONGODB_VERSION"
        if env_var in os.environ:
            if env_var == "AUTH":
                opts.auth = os.environ.get("AUTH") == "auth"
            elif env_var == "SSL":
                ssl_opt = os.environ.get("SSL", "")
                opts.ssl = ssl_opt and ssl_opt.lower() != "nossl"
            elif isinstance(getattr(opts, key), bool):
                if os.environ[env_var]:
                    setattr(opts, key, True)
            else:
                setattr(opts, key, os.environ[env_var])

    if opts.mongo_orchestration_home is None:
        opts.mongo_orchestration_home = DRIVERS_TOOLS / ".evergreen/orchestration"
    if command in ["start", "run"]:
        if opts.mongodb_binaries is None:
            opts.mongodb_binaries = DRIVERS_TOOLS / "mongodb/bin"
    if command == "run":
        if not opts.topology and opts.load_balancer:
            opts.topology = "sharded_cluster"
        if opts.auth_aws:
            opts.auth = True
            opts.orchestration_file = "auth-aws.json"
        if opts.topology == "standalone" or not opts.topology:
            opts.topology = "server"
        if not opts.version:
            opts.version = "latest"

    if opts.verbose:
        LOGGER.setLevel(logging.DEBUG)
    elif opts.quiet:
        LOGGER.setLevel(logging.WARNING)
    return opts, command


def get_docker_cmd():
    """Get the appropriate docker command."""
    docker = shutil.which("podman") or shutil.which("docker")
    if not docker:
        return None
    docker = PureWindowsPath(docker).as_posix()
    if "podman" in docker:
        docker = f"sudo {docker}"
    return docker


def handle_docker_config(data):
    """Modify config to when running in a docker container."""
    items = []

    # Gather all the items that have process settings.
    def traverse(root):
        if isinstance(root, list):
            [traverse(i) for i in root]
            return
        if "ipv6" in root:
            items.append(root)
            return
        for key, value in root.items():
            if key == "routers":
                continue
            if isinstance(value, (dict, list)):
                traverse(value)

    traverse(data)

    # Docker does not enable ipv6 by default.
    # https://docs.docker.com/config/daemon/ipv6/
    # We also need to use 0.0.0.0 instead of 127.0.0.1
    for item in items:
        item["ipv6"] = False
        item["bind_ip"] = "0.0.0.0,::1"
        item["dbpath"] = f"/tmp/mongo-{item['port']}"
        os.makedirs(item["dbpath"], exist_ok=True)

    if "routers" in data:
        for router in data["routers"]:
            router["ipv6"] = False
            router["bind_ip"] = "0.0.0.0,::1"
            router["logpath"] = f"/tmp/mongodb-{item['port']}.log"


def normalize_path(path: Path | str) -> str:
    if PLATFORM != "win32":
        return str(path)
    path = Path(path).as_posix()
    return re.sub("/cygdrive/(.*?)(/)", r"\1://", path, count=1)


def run_command(cmd: str, exit_on_error=True, **kwargs):
    LOGGER.debug(f"Running command {cmd}...")
    try:
        proc = subprocess.run(
            shlex.split(cmd),
            check=True,
            encoding="utf-8",
            stderr=subprocess.STDOUT,
            stdout=subprocess.PIPE,
            **kwargs,
        )
        LOGGER.info(proc.stdout)
    except subprocess.CalledProcessError as e:
        LOGGER.error(e.output)
        LOGGER.error(str(e))
        if exit_on_error:
            sys.exit(e.returncode)
    LOGGER.debug(f"Running command {cmd}... done.")


def start_atlas(opts):
    mo_home = Path(opts.mongo_orchestration_home)
    image = f"mongodb/mongodb-atlas-local:{opts.version}"
    docker = get_docker_cmd()
    stop(opts)
    # If we're on evergreen, we need to log into docker and use the pull-through cache.
    if "CI" in os.environ and "GITHUB_ACTION" not in os.environ:
        LOGGER.info("Logging in to docker...")
        run_command("bash setup.sh", cwd=EVG_PATH / "docker")
        LOGGER.info("Logging in to ECR... done.")
        image = f"901841024863.dkr.ecr.us-east-1.amazonaws.com/dockerhub/{image}"
    cmd = f"{docker} run --rm -d --name mongodb_atlas_local -p 27017:27017"
    if opts.auth:
        cmd += " -e MONGODB_INITDB_ROOT_USERNAME=bob"
        cmd += " -e MONGODB_INITDB_ROOT_PASSWORD=pwd123"
    if "podman" in docker:
        cmd += " --health-cmd '/usr/local/bin/runner healthcheck'"
    cmd += f" -P {image}"
    LOGGER.info("Starting local atlas...")
    LOGGER.debug("Using command: '%s'", cmd)
    container_id = subprocess.check_output(cmd, shell=True, encoding="utf-8").strip()
    (mo_home / "container_id.txt").write_text(container_id)
    # Wait for container to become healthy.
    LOGGER.info("Waiting for container to be healthy...")
    if "podman" in docker:
        run_command(f"{docker} healthcheck run {container_id}", exit_on_error=False)
    cmd = f"{docker} inspect -f '{{{{.State.Health.Status}}}}' {container_id}"
    tries = 0
    while 1:
        resp = subprocess.check_output(shlex.split(cmd), encoding="utf-8").strip()
        if resp == "healthy":
            break
        if tries >= 60:
            LOGGER.error("Timed out waiting for container to become healthy")
            sys.exit(1)
        time.sleep(1)
        tries += 1

    LOGGER.info("Waiting for container to be healthy... done.")
    uri = "mongodb://127.0.0.1:27017?directConnection=true"
    if opts.auth:
        uri = "mongodb://bob:pwd123@127.0.0.1:27017?directConnection=true"
    mongosh = Path(opts.mongodb_binaries) / "mongosh"
    run_command(f"{mongosh} {uri} --eval 'db.runCommand({{ping:1}})'")
    LOGGER.info("Starting local atlas... done.")
    return uri


def get_orchestration_data(opts):
    # Handle orchestration file - explicit or implicit.
    orchestration_file = opts.orchestration_file
    if not orchestration_file:
        fname = "basic"
        if opts.auth:
            fname = "auth"
        if opts.ssl:
            fname += "-ssl"
        if opts.load_balancer:
            fname += "-load-balancer"
        elif opts.disable_test_commands:
            fname = "disableTestCommands"
        elif opts.storage_engine:
            fname = opts.storage_engine
        orchestration_file = f"{fname}.json"

    # Get the orchestration config data.
    topology = opts.topology
    mo_home = Path(opts.mongo_orchestration_home)
    orch_path = mo_home / f"configs/{topology}s/{orchestration_file}"
    LOGGER.info(f"Using orchestration file: {orch_path}")
    text = orch_path.read_text()

    # Handle overriding the tls configuration in the file.
    if opts.tls_pem_key_file or opts.tls_ca_file:
        if not (opts.tls_pem_key_file and opts.tls_ca_file):
            raise ValueError("You must supply both tls-pem-key-file and tls-ca-file")
        base = "ABSOLUTE_PATH_REPLACEMENT_TOKEN/.evergreen/x509gen"
        text = text.replace(f"{base}/server.pem", normalize_path(opts.tls_pem_key_file))
        text = text.replace(f"{base}/ca.pem", normalize_path(opts.tls_ca_file))
    text = text.replace(
        "ABSOLUTE_PATH_REPLACEMENT_TOKEN", normalize_path(DRIVERS_TOOLS)
    )
    data = json.loads(text)

    if opts.require_api_version:
        if opts.topology == "replica_set":
            raise ValueError(
                "requireApiVersion is not supported with replica_sets, see SERVER-97010"
            )
        data["requireApiVersion"] = "1"

    if opts.tls_allow_invalid_certificates:
        if "sslParams" not in data:
            raise ValueError(
                "--tls-allow-invalid-certificates requires TLS to be configured, but no sslParams found in orchestration data"
            )
        data["sslParams"]["tlsAllowInvalidCertificates"] = True

    # If running on Docker, update the orchestration file to be docker-friendly.
    if os.environ.get("DOCKER_RUNNING"):
        handle_docker_config(data)

    return data


def clean_run(opts):
    mdb_binaries = Path(opts.mongodb_binaries)
    mdb_binaries_str = normalize_path(mdb_binaries)
    shutil.rmtree(mdb_binaries_str, ignore_errors=True)

    mongodb_dir = DRIVERS_TOOLS / "mongodb"
    if mongodb_dir.exists():
        shutil.rmtree(normalize_path(mongodb_dir), ignore_errors=True)

    for path in [URI_TXT, MO_EXPANSION_SH, MO_EXPANSION_YML]:
        path.unlink(missing_ok=True)

    crypt_path = DRIVERS_TOOLS / CRYPT_NAME_MAP[PLATFORM]
    crypt_path.unlink(missing_ok=True)


def run(opts):
    # Deferred import so we can run as a script without the cli installed.
    from mongodl import main as mongodl
    from mongosh_dl import main as mongosh_dl

    LOGGER.info("Running orchestration...")
    stop(opts)
    clean_run(opts)

    # NOTE: in general, we need to normalize paths to account for cygwin/Windows.
    mdb_binaries = Path(opts.mongodb_binaries)
    mdb_binaries_str = normalize_path(mdb_binaries)

    # The evergreen directory to path.
    os.environ["PATH"] = f"{EVG_PATH}:{os.environ['PATH']}"

    dl_start = datetime.now()

    version = opts.version
    cache_dir = DRIVERS_TOOLS / ".local/cache"
    cache_dir_str = normalize_path(cache_dir)
    default_args = f"--out {mdb_binaries_str} --cache-dir {cache_dir_str} --retries 5"
    if opts.quiet:
        default_args += " -q"
    elif opts.verbose:
        default_args += " -v"

    if opts.arch:
        default_args += f" --arch={opts.arch}"

    if not opts.local_atlas:
        # Download the archive.
        args = f"{default_args} --version {version}"
        args += " --strip-path-components 2 --component archive"
        if not opts.existing_binaries_dir:
            LOGGER.info(f"Downloading mongodb {version} to {mdb_binaries}...")
            mongodl(shlex.split(args))
            LOGGER.info(f"Downloading mongodb {version} to {mdb_binaries}... done.")
        else:
            LOGGER.info(
                f"Using existing mongod binaries dir: {opts.existing_binaries_dir}"
            )
            shutil.copytree(opts.existing_binaries_dir, mdb_binaries)

        run_command(f"{mdb_binaries_str}/mongod --version")

    # Download legacy shell.
    if opts.install_legacy_shell:
        args = f"{default_args} --version 5.0"
        args += " --strip-path-components 2 --component shell"
        LOGGER.INFO("Downloading legacy shell...")
        mongodl(shlex.split(args))
        LOGGER.INFO("Downloading legacy shell... done.")

    # Download crypt shared.
    if not opts.skip_crypt_shared:
        # Get the download URL for crypt_shared.
        # We download crypt_shared to DRIVERS_TOOLS so that it is on a different
        # path location than the other binaries, which is required for
        # https://github.com/mongodb/specifications/blob/master/source/client-side-encryption/tests/README.md#via-bypassautoencryption
        args = default_args + (
            f" --version {version} --strip-path-components 1 --component crypt_shared"
        )
        LOGGER.info("Downloading crypt_shared...")
        mongodl(shlex.split(args))
        LOGGER.info("Downloading crypt_shared... done.")
        crypt_shared_path = mdb_binaries / CRYPT_NAME_MAP[PLATFORM]
        if crypt_shared_path.exists():
            shutil.move(crypt_shared_path, DRIVERS_TOOLS)
            crypt_shared_path = DRIVERS_TOOLS / crypt_shared_path.name
        else:
            raise RuntimeError(
                f"Could not find expected crypt_shared_path: {crypt_shared_path}"
            )
        crypt_text = f'CRYPT_SHARED_LIB_PATH: "{normalize_path(crypt_shared_path)}"'
        MO_EXPANSION_YML.write_text(crypt_text)
        MO_EXPANSION_SH.write_text(crypt_text.replace(": ", "="))

    # Download mongosh
    args = f"--out {mdb_binaries_str} --strip-path-components 2 --retries 5"
    if opts.verbose:
        args += " -v"
    elif opts.quiet:
        args += " -q"
    LOGGER.info("Downloading mongosh...")
    mongosh_dl(shlex.split(args))
    LOGGER.info("Downloading mongosh... done.")

    dl_end = datetime.now()
    mo_start = datetime.now()

    data = get_orchestration_data(opts)

    if opts.mongodb_runner and version in ("3.6", "4.0"):
        LOGGER.warning(
            "mongodb-runner does not support MongoDB < 4.2, using mongo-orchestration"
        )
        opts.mongodb_runner = False

    if opts.local_atlas:
        uri = start_atlas(opts)
    elif opts.mongodb_runner:
        uri = start_mongodb_runner(opts, data)
    else:
        mo_home = Path(opts.mongo_orchestration_home)

        # Write the config file.
        orch_file = Path(mo_home / "config.json")
        orch_file.write_text(json.dumps(data, indent=2))

        # Start the orchestration.
        start(opts)

        # Configure the server.
        LOGGER.info("Starting deployment...")
        url = f"http://localhost:8889/v1/{opts.topology}s"
        req = urllib.request.Request(
            url, data=json.dumps(data).encode("utf-8"), method="POST"
        )
        try:
            resp = urllib.request.urlopen(req)
        except urllib.error.HTTPError as e:
            stop(opts)
            LOGGER.error("out.log: %s", (mo_home / "out.log").read_text())
            LOGGER.error("server.log: %s", (mo_home / "server.log").read_text())
            raise e
        resp = json.loads(resp.read().decode("utf-8"))
        LOGGER.debug(resp)
        LOGGER.info("Starting deployment... done.")
        uri = resp.get("mongodb_auth_uri", resp["mongodb_uri"])

    # Handle the cluster uri.
    MO_EXPANSION_YML.touch()
    MO_EXPANSION_YML.write_text(
        MO_EXPANSION_YML.read_text() + f'\nMONGODB_URI: "{uri}"'
    )
    MO_EXPANSION_SH.touch()
    MO_EXPANSION_SH.write_text(MO_EXPANSION_SH.read_text() + f'\nMONGODB_URI="{uri}"')
    URI_TXT.write_text(uri)
    LOGGER.info(f"Cluster URI: {uri}")

    # Write the results file.
    mo_end = datetime.now()
    data = dict(
        results=[
            dict(
                status="PASS",
                test_file="Orchestration",
                start=int(mo_start.timestamp()),
                end=int(mo_end.timestamp()),
                elapsed=(mo_end - mo_start).total_seconds(),
            ),
            dict(
                status="PASS",
                test_file="Download MongoDB",
                start=int(dl_start.timestamp()),
                end=int(dl_end.timestamp()),
                elapsed=(dl_end - dl_start).total_seconds(),
            ),
        ]
    )
    Path(DRIVERS_TOOLS / "results.json").write_text(json.dumps(data, indent=2))

    LOGGER.info("Running orchestration... done.")


def clean_start(opts):
    mo_home = Path(opts.mongo_orchestration_home)
    for fname in [
        "out.log",
        "server.log",
        "orchestration.config",
        "config.json",
        "server.pid",
    ]:
        if (mo_home / fname).exists():
            try:
                (mo_home / fname).unlink()
            except PermissionError:
                pass


def start(opts):
    # Start mongo-orchestration

    # Stop a running server.
    mo_home = Path(opts.mongo_orchestration_home)
    if (mo_home / "server.pid").exists():
        stop(opts)

    # Clean up previous files.
    clean_start(opts)

    # Set up the mongo orchestration config.
    os.makedirs(mo_home / "lib", exist_ok=True)
    mo_config = mo_home / "orchestration.config"
    mdb_binaries = Path(opts.mongodb_binaries)
    config = dict(releases=dict(default=normalize_path(mdb_binaries)))
    mo_config.write_text(json.dumps(config, indent=2))
    mo_config_str = normalize_path(mo_config)
    sys_executable = normalize_path(sys.executable)
    command = f"{sys_executable} -m mongo_orchestration.server"

    # Handle Windows-specific concerns.
    if PLATFORM == "win32":
        # Copy default client certificate.
        src = DRIVERS_TOOLS / ".evergreen/x509gen/client.pem"
        dst = mo_home / "lib/client.pem"
        try:
            shutil.copy2(src, dst)
        except (shutil.SameFileError, PermissionError):
            pass

        # We need to use the CLI executable, and add it to our path.
        os.environ["PATH"] = (
            f"{Path(sys_executable).parent}{os.pathsep}{os.environ['PATH']}"
        )
        command = "mongo-orchestration -s wsgiref"

    # Override the client cert file if applicable.
    env = os.environ.copy()
    if opts.tls_cert_key_file:
        env["MONGO_ORCHESTRATION_CLIENT_CERT"] = normalize_path(opts.tls_cert_key_file)

    mo_start = datetime.now()

    # Start the process.
    args = f"{command} start -e default -f {mo_config_str}"
    args += " --socket-timeout-ms=60000 --bind=127.0.0.1 --enable-majority-read-concern"

    LOGGER.info("Starting mongo-orchestration...")
    output_file = mo_home / "out.log"
    server_file = mo_home / "server.log"

    # NOTE: we need to use a separate file id for stdout and close it so Evergreen does not hang.
    output_fid = output_file.open("w")
    try:
        subprocess.run(
            shlex.split(args),
            check=True,
            stderr=subprocess.STDOUT,
            stdout=output_fid,
            env=env,
        )
    except subprocess.CalledProcessError:
        LOGGER.error("Orchestration failed!")
        LOGGER.error(f"server.log:\n{server_file.read_text().strip()}")
        raise
    finally:
        output_fid.close()
        LOGGER.info(f"out.log:\n{output_file.read_text().strip()}")

    # Wait for the server to be available.
    attempt = 0
    while True:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            try:
                s.connect(("localhost", 8889))
                break
            except ConnectionRefusedError:
                if (datetime.now() - mo_start).seconds > 120:
                    stop(opts)
                    LOGGER.error("Orchestration failed!")
                    LOGGER.error(f"server.log: {server_file.read_text()}")
                    raise TimeoutError("Server failed to start") from None
        attempt += 1
        time.sleep(attempt * 1000)

    LOGGER.info("Starting mongo-orchestration... done.")


def shutdown_proc(proc: psutil.Process) -> None:
    try:
        proc.terminate()
        try:
            proc.wait(2)  # Wait up to 2 seconds.
        except psutil.TimeoutExpired:
            proc.kill()
    except Exception as e:
        LOGGER.exception(e)


def shutdown_docker(docker: str, container_id: str) -> None:
    if "podman" in docker:
        cmd = f"{docker} rm -f {container_id}"
    else:
        cmd = f"{docker} kill {container_id}"
    run_command(cmd, exit_on_error=False)


def stop(opts):
    mo_home = Path(opts.mongo_orchestration_home)
    pid_file = mo_home / "server.pid"
    out_log = mo_home / "out.log"
    container_file = mo_home / "container_id.txt"
    docker = get_docker_cmd()

    # First try to shut down using pid file.
    if pid_file.exists():
        pid = int(pid_file.read_text().strip())
        pid_file.unlink(missing_ok=True)
        if psutil.pid_exists(pid):
            LOGGER.info("Stopping mongo-orchestration using pid file...")
            shutdown_proc(psutil.Process(pid))
            LOGGER.info("Stopping mongo-orchestration using pid file... done.")

    # Next try and use the output.log file as a serialized json file.
    if out_log.exists():
        try:
            data = json.loads(out_log.read_text())
        except Exception:
            data = None
        if data:
            LOGGER.info("Stopping mongodb-runner cluster...")
            all_servers = data["serialized"]["servers"]
            for shard in data["serialized"].get("shards", []):
                all_servers.extend(shard["servers"])
            for server in all_servers:
                pid = server["pid"]
                if psutil.pid_exists(pid):
                    shutdown_proc(psutil.Process(pid))
                if Path(server["dbPath"]).exists():
                    shutil.rmtree(server["dbPath"])
            LOGGER.info("Stopping mongodb-runner cluster... done.")
            out_log.unlink()

    # Next try using a docker container file.
    if docker is not None and container_file.exists():
        LOGGER.info("Stopping mongodb_atlas_local using container file...")
        shutdown_docker(docker, container_file.read_text())
        container_file.unlink()
        LOGGER.info("Stopping mongodb_atlas_local using container file ... done.")

    all_procs = list(psutil.process_iter())

    # Next look for mongo-orchestration by command line arguments.
    for proc in all_procs:
        try:
            cmdline = proc.cmdline()
        except (psutil.AccessDenied, psutil.NoSuchProcess):
            continue
        found = False
        for item in cmdline:
            if "mongo_orchestration.server" in item or "mongo-orchestration" in item:
                found = True
                break
        if not found:
            continue
        LOGGER.info("Stopping mongo-orchestration by process info...")
        shutdown_proc(proc)
        LOGGER.info("Stopping mongo-orchestration by process info... done.")

    # Next look for running docker images.
    if docker:
        cmd = f"{docker} ps --format '{{{{.Image}}}}\t{{{{.ID}}}}'"
        try:
            response = subprocess.check_output(
                shlex.split(cmd), encoding="utf-8"
            ).strip()
        except (subprocess.CalledProcessError, FileNotFoundError) as e:
            LOGGER.exception(e)
            response = ""
        for line in response.splitlines():
            image, container_id = line.split("\t")
            if image in ["mongodb/mongodb-atlas-local", "mongo"]:
                LOGGER.info(f"Stopping {image} by image name...")
                shutdown_docker(docker, container_id)
                LOGGER.info(f"Stopping {image} by image name... done.")

    # Finally, look for any processes that are named mongod or mongos.
    for proc in all_procs:
        try:
            name = proc.name()
        except psutil.NoSuchProcess:
            continue
        if name in ["mongod", "mongos", "mongod.exe", "mongos.exe"]:
            LOGGER.info(f"Stopping {name} by process name...")
            shutdown_proc(proc)
            LOGGER.info(f"Stopping {name} by process name... done.")


def main():
    opts, command = get_options()
    if command == "run":
        run(opts)
    elif command == "start":
        start(opts)
    elif command == "stop":
        stop(opts)
    elif command == "clean":
        clean_run(opts)
        clean_start(opts)


if __name__ == "__main__":
    main()