mds_plugin/util.py

# Copyright (c) 2021, 2024, Oracle and/or its affiliates.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License, version 2.0,
# as published by the Free Software Foundation.
#
# This program is designed to work with certain software (including
# but not limited to OpenSSL) that is licensed under separate terms, as
# designated in a particular file or component or in included license
# documentation.  The authors of MySQL hereby grant you an additional
# permission to link the program and your derivative works with the
# separately licensed software that they have either included with
# the program or referenced in the documentation.
#
# This program is distributed in the hope that it will be useful,  but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
# the GNU General Public License, version 2.0, for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA

"""Sub-Module with utilities for the MySQL Database Service"""

from mysqlsh.plugin_manager import plugin_function
from mds_plugin import core, configuration, mysql_database_service
from os import getenv

# cSpell:ignore SQLDB, Popen, bufsize, dryrun


@plugin_function('mds.util.heatWaveLoadData', shell=True, cli=True, web=True)
def mds_heat_wave_load_data(**kwargs):
    """Loads data to a HeatWave Cluster

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        schemas (list): The list of schemas
        mode (str): The mode to use, "normal"|"dryrun"
        output (str): The output mode to use, "normal"|"compact"|"silent"|"help"
        disable_unsupported_columns (bool): Whether to disable unsupported columns
        optimize_load_parallelism (bool): Whether to optimize parallelism
        enable_memory_check (bool): Whether to enable the memory check
        sql_mode (str): The sql_mode to use
        exclude_list (str): The database object list to exclude
        session (object): The database session to use.
        interactive (bool): Indicates whether to execute in interactive mode
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
       None in interactive mode, the result sets as string otherwise
    """
    schemas = kwargs.get("schemas")
    mode = kwargs.get("mode", "normal")
    output = kwargs.get("output", "normal")
    disable_unsupported_columns = kwargs.get(
        "disable_unsupported_columns", True)
    optimize_load_parallelism = kwargs.get("optimize_load_parallelism", True)
    enable_memory_check = kwargs.get("enable_memory_check", True)
    sql_mode = kwargs.get("sql_mode", "")
    exclude_list = kwargs.get("exclude_list", "")

    session = kwargs.get("session")
    interactive = kwargs.get("interactive", core.get_interactive_default())
    raise_exceptions = kwargs.get("raise_exceptions", not interactive)

    try:
        if not schemas:
            raise ValueError("At least one schema needs to be specified.")

        session = core.get_current_session(session)

        policy = (
            "disable_unsupported_columns" if disable_unsupported_columns else "not_disable_unsupported_columns")
        set_load_parallelism = (
            "TRUE" if optimize_load_parallelism else "FALSE")

        schemasJson = "JSON_ARRAY(" + \
            ', '.join(f'"{s}"' for s in schemas) + ")"
        optionsJson = ("JSON_OBJECT("
                       f'"mode", "{mode}", '
                       f'"output", "{output}", '
                       f'"sql_mode", {sql_mode}, '
                       f'"policy", "{policy}", '
                       f'"set_load_parallelism", {set_load_parallelism}, '
                       f'"auto_enc", JSON_OBJECT("mode", "{"check" if enable_memory_check else "off"}")')
        if exclude_list:
            optionsJson += f', "exclude_list", JSON_ARRAY({exclude_list})'
        optionsJson += ")"

        if interactive:
            print(f"Loading Data to HeatWave Cluster Using Auto Parallel Load.\n")

        sql = f"CALL sys.heatwave_load({schemasJson}, {optionsJson})"
        if interactive:
            print(f"MySQL > {sql}\n")

        res = session.run_sql(sql)

        out_str = ""
        next_result = True
        while next_result:
            rows = res.fetch_all()

            if len(rows) == 0:
                next_result = res.next_result()
                continue

            if interactive:
                print(core.format_result_set(res, rows, addFooter=False))
            else:
                out_str += out_str + "\n"

            next_result = res.next_result()

        if not interactive:
            return out_str

    except Exception as e:
        if raise_exceptions:
            raise
        else:
            print(f"Error: {str(e)}")


@plugin_function('mds.util.createComputeInstanceForEndpoint')
def create_compute_instance_for_endpoint(**kwargs):
    """Returns a public compute instance

    If the instance does not yet exists in the compartment, create it

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        instance_name (str): The name of the compute instance
        db_system_name (str): The new name of the DB System.
        db_system_id (str): The OCID of the db_system
        private_key_file_path (str): The file path to an SSH private key
        subnet_id (str): The OCID of the subnet to use
        public_ip (bool): If set to false, no public IP will be assigned
        shape (str): The name of the shape to use
        cpu_count (int): The number of OCPUs
        memory_size (int): The amount of memory
        dns_a_record_notification (bool): Whether to print a message to setup the DNS A record for this instance
        domain_name (str): The domain name of the compute instance
        compartment_id (str): The OCID of the compartment
        config (object): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        interactive (bool): Indicates whether to execute in interactive mode
        raise_exceptions (bool): If set to true exceptions are raised
        return_formatted (bool): If set to true, a list object is returned
        return_python_object (bool): Used for internal plugin calls

    Returns:
       None
    """
    instance_name = kwargs.get("instance_name", "MDSJumpHost")
    db_system_name = kwargs.get("db_system_name")
    db_system_id = kwargs.get("db_system_id")
    private_key_file_path = kwargs.get(
        "private_key_file_path", "~/.ssh/id_rsa")
    public_ip = kwargs.get("public_ip", True)
    subnet_id = kwargs.get("subnet_id")
    shape = kwargs.get("shape", "VM.Standard.E4.Flex")
    cpu_count = kwargs.get("cpu_count", 1)
    memory_size = kwargs.get("memory_size", 16)
    dns_a_record_notification = kwargs.get("dns_a_record_notification", False)
    domain_name = kwargs.get("domain_name")

    compartment_id = kwargs.get("compartment_id")
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")

    interactive = kwargs.get("interactive", core.get_interactive_default())
    raise_exceptions = kwargs.get("raise_exceptions", not interactive)
    return_formatted = kwargs.get("return_formatted", interactive)
    return_python_object = kwargs.get("return_python_object", False)

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)

        import oci.mysql
        from pathlib import Path
        import os.path
        from mds_plugin import compartment, compute, network
        import time

        db_system = mysql_database_service.get_db_system(
            db_system_name=db_system_name, db_system_id=db_system_id,
            compartment_id=compartment_id, config=config,
            interactive=interactive, raise_exceptions=True,
            return_python_object=True)
        if db_system is None:
            raise ValueError("DB System not specified or found.")

        # Get compartment and public subnet_id from MDS
        if not compartment_id:
            compartment_id = db_system.compartment_id
        if not subnet_id:
            mds_subnet = network.get_subnet(
                subnet_id=db_system.subnet_id,
                config=config, interactive=False)
            subnet = network.get_subnet(
                network_id=mds_subnet.vcn_id,
                public_subnet=public_ip,
                config=config,
                interactive=False)
            if subnet:
                subnet_id = subnet.id

        if not subnet_id:
            if public_ip:
                raise ValueError(
                    'The network used by the MDS instance does not have public subnet.'
                    'Please add a public subnet first')
            else:
                raise ValueError('No subnet specified.')

        # Try to get the Compute Instance with the given name
        mds_jump_host = compute.get_instance(
            instance_name=instance_name,
            compartment_id=compartment_id,
            config=config, interactive=False,
            raise_exceptions=True,
            return_python_object=True)

        # If it already exists, return it
        if mds_jump_host:
            return mds_jump_host

        # if interactive:
        #     # If there is no MySQL DBSystemProxy instance yet, ask the user
        #     print(f"In order to perform the requested operation for the MySQL "
        #           f"DB System\na compute instance named 'MDSJumpHost' "
        #           f"needs to be created.\n")
        #     prompt = core.prompt(
        #         "Do you want to create a new compute instance to be used as "
        #         "bastion host? [YES/no]: ",
        #         {'defaultValue': 'yes'}).strip().lower()
        #     if prompt != "yes":
        #         print("Operation cancelled.\n")
        #         return

        if interactive:
            print(f"Creating Compute Instance '{instance_name}'...")

        new_jump_host = compute.create_instance(
            instance_name=instance_name,
            shape=shape,
            cpu_count=cpu_count, memory_size=memory_size,
            operating_system="Oracle Linux",
            operating_system_version="9",
            use_latest_image=True,
            subnet_id=subnet_id,
            public_subnet=public_ip,
            init_script_file_path=os.path.join(
                os.path.join(Path(__file__).parent.absolute(), "internal"), "init_router_script.sh"),
            interactive=False,
            return_python_object=True)
        if new_jump_host is None:
            print("Compute instance could not be created.")
            return

        # Initialize the identity client
        compute_client = core.get_oci_compute_client(config=config)

        print(f"Waiting for Compute Instance '{instance_name}' to become "
              "available.\nThis can take up to 5 minutes or more.", end="")

        # Wait until the lifecycle_state == RUNNING, 5 minutes max
        try:
            cycles = 0
            while cycles < 60:
                mds_jump_host = compute_client.get_instance(
                    new_jump_host.id).data
                if mds_jump_host.lifecycle_state == "RUNNING":
                    break
                else:
                    time.sleep(5)
                    print(".", end="")
                cycles += 1
            print("")
        except oci.exceptions.ServiceError as e:
            print(f'Could not fetch the compute instances state.\n'
                  f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
            return
        if mds_jump_host.lifecycle_state != "RUNNING":
            print(f"Compute Instance '{instance_name}' did not become available "
                  f"within 5 minutes. Please check the state manually.")
            return None if interactive else mds_jump_host

        if interactive:
            print(f"Compute Instance '{instance_name}' became available.")

        # Get the public IP of the instance
        public_ip = compute.get_instance_public_ip(
            instance_id=mds_jump_host.id, compartment_id=compartment_id,
            config=config, private_ip_fallback=True)
        if public_ip is None or public_ip == "":
            raise Exception(
                f"The public IP of the {instance_name} instance could not be "
                "fetched.")

        if dns_a_record_notification and interactive:
            print("\nATTENTION: Please create a DNS A record using the following values.\n"
                  f"Domain: {domain_name}\n"
                  f"Destination TCP/IP address: {public_ip}")

            answered = False
            while not answered:
                try:
                    result = core.prompt(
                        f"Please click OK once the DNS A record has been created. [OK/Cancel]: ",
                        {"defaultValue": "OK"})
                    answered = True
                except:
                    print(
                        "Please select OK or Cancel on the confirmation notification.")
                    pass

            if result != "OK":
                raise Exception(
                    "Endpoint creation cancelled. Please delete the compute instance that has been created.")

        if interactive:
            print("\nPerforming base configuration.\n"
                  f"Connecting to {instance_name} instance at {public_ip}...",
                  end="")

        setup_complete = False
        connected = False
        cycles = 0
        while not setup_complete and cycles < 10:
            cycles += 1
            try:
                with compute.SshConnection(
                        username="opc", host=public_ip,
                        private_key_file_path=private_key_file_path) as conn:

                    connected = True
                    if interactive:
                        print(f"\nConnected to {instance_name} instance at "
                              f"{public_ip}.")

                    # Get MySQL Router configuration from remote instance
                    output = ""
                    output = conn.execute('mysqlsh --js -e "mds.info()"').strip()
                    if "MySQL Shell MDS Plugin" not in output:
                        # If the config is not available yet, give the instance
                        # time to complete setup
                        if interactive:
                            print(f"Waiting for {instance_name} setup to be "
                                  f"completed.\nThis can take up to 2 minutes.",
                                  end="")
                        try:
                            i = 0
                            while ("MySQL Shell MDS Plugin" not in output
                                   and i < 25):
                                output = conn.execute(
                                    'mysqlsh --js -e "mds.info()"').strip()
                                if "MySQL Shell MDS Plugin" not in output:
                                    time.sleep(5)
                                    if interactive:
                                        print(".", end="")
                                i += 1
                        except:
                            pass

                        if interactive:
                            print("")

                    if "MySQL Shell MDS Plugin" not in output:
                        raise Exception(
                            f"\nCould not finish the '{instance_name}' setup "
                            f"at {public_ip}.\n")
                    else:
                        setup_complete = True

            except Exception as e:
                if cycles < 10 and not connected:
                    time.sleep(5)
                    if interactive:
                        print(".", end="")
                else:
                    raise Exception(
                        f"Could not connect to compute instance "
                        f"'{instance_name}' at {public_ip}.\n"
                        f"ERROR: {str(e)}")

        if interactive:
            print(f"Compute Instance '{instance_name}' successfully created.")

        return core.return_oci_object(
            oci_object=new_jump_host,
            return_formatted=return_formatted,
            return_python_object=return_python_object,
            format_function=compute.format_instance_listing)
    except Exception as e:
        if raise_exceptions:
            raise
        print(f"ERROR: {str(e)}")


@plugin_function('mds.util.createEndpoint', shell=True, cli=True, web=True)
def add_public_endpoint(**kwargs):
    """Creates a public endpoint using MySQL Router on a compute instance

    If no id is given, it will prompt the user for the id.

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        instance_name (str): Name of the compute instance
        db_system_name (str): The new name of the DB System.
        db_system_id (str): The OCID of the db_system
        private_key_file_path (str): The file path to an SSH private key
        shape (str): The name of the shape to use
        cpu_count (int): The number of OCPUs
        memory_size (int): The amount of memory
        mysql_user_name (str): The MySQL user name to use for bootstrapping
        public_ip (bool): If set to true, a public IP will be assigned to the compute instance
        domain_name (str): The domain name of the compute instance
        port_forwarding (bool): Whether port forwarding of MySQL ports should be enabled
        mrs (bool): Whether the MySQL REST Service (MRS) should be enabled
        ssl_cert (bool): Whether SSL Certificates should be managed
        jwt_secret (str): The JWT secret for MRS
        compartment_id (str): The OCID of the compartment
        config (object): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        interactive (bool): Indicates whether to execute in interactive mode
        raise_exceptions (bool): If set to true exceptions are raised
        return_formatted (bool): If set to true, a list object is returned

    Returns:
       None
    """

    # cSpell:ignore OCPU
    instance_name = kwargs.get("instance_name")
    db_system_name = kwargs.get("db_system_name")
    db_system_id = kwargs.get("db_system_id")
    private_key_file_path = kwargs.get(
        "private_key_file_path", "~/.ssh/id_rsa")
    shape = kwargs.get("shape", "VM.Standard.E4.Flex")
    cpu_count = kwargs.get("cpu_count", 1)
    memory_size = kwargs.get("memory_size", 16)

    mysql_user_name = kwargs.get("mysql_user_name", "dba")
    mysql_user_password = core.prompt(
        f"Please enter the password for {mysql_user_name}",
        {"type": "password"})

    public_ip = kwargs.get("public_ip", True)
    domain_name = kwargs.get("domain_name")

    port_forwarding = kwargs.get("port_forwarding", True)
    mrs = kwargs.get("mrs", True)
    ssl_cert = kwargs.get("ssl_cert", False)
    jwt_secret = kwargs.get("jwt_secret")

    compartment_id = kwargs.get("compartment_id")
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")

    interactive = kwargs.get("interactive", core.get_interactive_default())
    raise_exceptions = kwargs.get("raise_exceptions", not interactive)
    return_formatted = kwargs.get("return_formatted", interactive)

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        db_system_id = configuration.get_current_db_system_id(
            db_system_id=db_system_id, config=config)

        from mds_plugin import compute
        import configparser
        import io
        import time
        import hashlib

        db_system = mysql_database_service.get_db_system(
            db_system_name=db_system_name, db_system_id=db_system_id,
            compartment_id=compartment_id, config=config,
            interactive=interactive,
            return_python_object=True)
        if db_system is None:
            raise ValueError("No DB System selected."
                             "Cancelling operation")

        if not jwt_secret:
            md5 = hashlib.md5()
            md5.update(db_system.id.encode())
            jwt_secret = md5.hexdigest()

        # Get the first active endpoint
        endpoints = [e for e in db_system.endpoints if e.status == 'ACTIVE']
        if len(endpoints) < 1:
            raise Exception(
                "This MySQL DB System has no active endpoints assigned.")
        endpoint = endpoints[0]

        # Get the compute instance (let the user create it
        # if it does not exist yet)
        jump_host = create_compute_instance_for_endpoint(
            instance_name=instance_name,
            private_key_file_path=private_key_file_path,
            db_system_id=db_system_id,
            shape=shape, memory_size=memory_size,
            cpu_count=cpu_count,
            dns_a_record_notification=ssl_cert,
            domain_name=domain_name,
            compartment_id=db_system.compartment_id, config=config,
            interactive=interactive,
            return_python_object=True)
        if not jump_host:
            raise Exception(f"Compute instance {instance_name} not available."
                            "Operation cancelled.")

        # Get the public IP of the instance
        public_ip = compute.get_instance_public_ip(
            instance_id=jump_host.id, compartment_id=db_system.compartment_id,
            config=config)
        if not public_ip:
            raise Exception(f"The public IP of the instance {instance_name} "
                            "could not be fetched.")

        sec_lists = compute.get_instance_vcn_security_lists(
            instance_id=jump_host.id,
            compartment_id=db_system.compartment_id,
            config=config, interactive=interactive,
            raise_exceptions=raise_exceptions,
            return_python_object=True)
        if sec_lists is None:
            raise Exception(
                "The network security lists could not be fetched.")

        # Allow traffic on port 80 in order to fetch the certs
        if mrs and public_ip:
            compute.add_ingress_port_to_security_lists(
                security_lists=sec_lists, port=80,
                description="MRS HTTP via Router",
                compartment_id=compartment_id, config=config,
                interactive=interactive,
                raise_exceptions=raise_exceptions)

        # Open an SSH connection to the instance
        if interactive:
            print("\nBootstrapping the MySQL Router.\n"
                  f"Connecting to {instance_name} instance at {public_ip}...")
        try:
            with compute.SshConnection(
                    username="opc", host=public_ip,
                    private_key_file_path=private_key_file_path) as conn:

                if interactive:
                    print(f"Connected to {instance_name} instance at "
                          f"{public_ip}.")

                # Get MySQL Router configuration from remote instance
                output = ""
                output = conn.execute(
                    'test -f /etc/mysqlrouter/'
                    'mysqlrouter.conf && echo "Available"').strip()
                if output != "Available":
                    # If the config is not available yet, give the instance time
                    # to complete setup
                    if interactive:
                        print(
                            f"Waiting for MySQL Router Configuration to become "
                            f"available.\nThis can take up to 2 minutes.",
                            end="")
                    try:
                        i = 0
                        while output != "Available" and i < 25:
                            output = conn.execute(
                                'test -f /etc/mysqlrouter/'
                                'mysqlrouter.conf && echo "Available"').strip()
                            if output != "Available":
                                time.sleep(5)
                                if interactive:
                                    print(".", end="")
                            i += 1
                    except:
                        pass

                if output == "":
                    raise Exception(
                        "Could not fetch MySQL Router configuration from remote instance.")

                if interactive:
                    print("Bootstrapping MySQL Router against "
                          f"{mysql_user_name}@{endpoint.ip_address}:{endpoint.port} "
                          f"using JWT secret {jwt_secret} ...")

                (success, output) = conn.executeAndSendOnStdin(
                    f"sudo mysqlrouter_bootstrap {mysql_user_name}@{endpoint.ip_address}:{endpoint.port} "
                    f"-u mysqlrouter "
                    f"--mrs --mrs-global-secret {jwt_secret} " if mrs else ""
                    "--https-port 8446 "
                    "--conf-set-option=http_server.ssl=0 "
                    "--conf-set-option=http_server.port=8446 ",
                    mysql_user_password)

                if not success:
                    if output:
                        print(output)
                    raise Exception("Bootstrap operation failed.")

                # Manually fix the MySQL Router config till bootstrap allows to disable SSL
                # Load config to in-memory stream
                conn.execute(
                    "sudo cp /etc/mysqlrouter/mysqlrouter.conf /home/opc/mysqlrouter.conf")
                conn.execute("sudo chown opc:opc /home/opc/mysqlrouter.conf")
                with io.BytesIO() as router_config_stream:
                    # Get the remote config file
                    try:
                        conn.get_remote_file_as_file_object(
                            "/home/opc/mysqlrouter.conf",
                            router_config_stream)
                    except Exception as e:
                        raise Exception("Could not get router config file. "
                                        f"{str(e)}")

                    # If there was an error, print it
                    last_error = conn.get_last_error()
                    if last_error != "":
                        raise Exception(f"Could not read router config file. "
                                        f"{last_error}")

                    # Load CLI config file
                    router_config = configparser.ConfigParser()
                    router_config.read_string(
                        router_config_stream.getvalue().decode("utf-8"))
                    router_config_stream.close()

                # # # Ensure that there is a section with the name of
                # # # "routing:classic"
                # # if "routing:classic" not in router_config.sections():
                # #     router_config["routing:classic"] = {}

                # # cnf = router_config["routing:classic"]
                # # cnf["routing_strategy"] = "round-robin"
                # # cnf["bind_address"] = "0.0.0.0"
                # # cnf["bind_port"] = "6446"
                # # cnf["destinations"] = f"{endpoint.ip_address}:{endpoint.port}"

                # # Ensure that there is a section with the name of "http_server"
                # if "http_server" not in router_config.sections():
                #     router_config["http_server"] = {}

                cnf = router_config["http_server"]
                cnf["port"] = "8446"
                cnf["ssl"] = "0"
                # cnf["ssl_cert"] = ""
                # cnf["ssl_key"] = ""
                # cnf["static_folder"] = "/var/run/mysqlrouter/www/"

                # # # Ensure that there is a section with the name of "routing:x"
                # # if "routing:x" not in router_config.sections():
                # #     router_config["routing:x"] = {}

                # # cnf = router_config["routing:x"]
                # # cnf["routing_strategy"] = "round-robin"
                # # cnf["bind_address"] = "0.0.0.0"
                # # cnf["bind_port"] = "6447"
                # # cnf["destinations"] = f"{endpoint.ip_address}:{endpoint.port_x}"

                # # # cSpell:ignore mrds SQLR
                # # # Ensure that there is a section with the name of
                # # # "mysql_rest_service"
                # # if "mysql_rest_service" not in router_config.sections():
                # #     router_config["mysql_rest_service"] = {}

                # # cnf = router_config["mysql_rest_service"]
                # # cnf["mysql_user"] = "dba"
                # # cnf["mysql_password"] = "MySQLR0cks!"
                # # cnf["mysql_read_only_route"] = "classic"
                # # cnf["mysql_read_write_route"] = "classic"

                if interactive:
                    print("Writing updated MySQL Router configuration file...")

                # Write config to in-memory stream
                with io.BytesIO() as router_config_bytes_stream:
                    with io.StringIO() as router_config_stream:
                        router_config.write(router_config_stream)

                        # Seek to the beginning of the text stream
                        router_config_bytes_stream.write(
                            router_config_stream.getvalue().encode("utf-8"))
                        router_config_bytes_stream.seek(0)

                        # Write out new config file to remote instance
                        try:
                            conn.put_local_file_object(
                                router_config_bytes_stream,
                                "/home/opc/mysqlrouter.conf")
                        except Exception as e:
                            raise Exception(
                                "Could not upload router config file. "
                                f"{str(e)}")

                # Move config to final place and fix privileges
                conn.execute(
                    "sudo cp /home/opc/mysqlrouter.conf /etc/mysqlrouter/mysqlrouter.conf")
                conn.execute(
                    "sudo chown mysqlrouter:mysqlrouter /etc/mysqlrouter/mysqlrouter.conf")
                conn.execute("sudo rm /home/opc/mysqlrouter.conf")

                # # If there was an error, print it
                # last_error = conn.get_last_error()
                # if last_error != "":
                #     raise Exception(
                #         "Could not upload router config file. "
                #         f"ERROR: {last_error}")

                # Install the SSL certificate
                if ssl_cert:
                    # Restart NGINX
                    conn.execute("sudo systemctl restart nginx.service")

                    ssl_cert_created = False
                    while not ssl_cert_created and domain_name != "":
                        # Check for one minute if the domain name points to the instance's ip address
                        if interactive:
                            print(
                                f"\nCreating the SSL certificate for {domain_name} ...")

                        try:
                            i = 0
                            while not ssl_cert_created and i < 5:
                                # cSpell:ignore certbot certonly webroot
                                # conn.execute(
                                #     f"sudo certbot certonly --webroot -w /usr/share/nginx/html -d {domain_name} --agree-tos "
                                #     "--register-unsafely-without-email --key-type rsa")
                                output = conn.execute(f"sudo /home/opc/.acme.sh/acme.sh --issue -d {domain_name} "
                                                      "--webroot /usr/share/nginx/html "
                                                      "--force --server letsencrypt --home /home/opc/.acme.sh")
                                last_error = conn.get_last_error()
                                if last_error != "":
                                    if i == 0 and interactive:
                                        print("\nATTENTION: Please create a DNS A record using the following values.\n"
                                              f"Domain: {domain_name}\n"
                                              f"Destination TCP/IP address: {public_ip}")
                                        print(
                                            f"\nWaiting for DNS A record creation ...", end="")
                                    time.sleep(10)
                                    if interactive:
                                        print(".", end="")
                                else:
                                    ssl_cert_created = True
                                i += 1
                        except:
                            pass

                        if not ssl_cert_created:
                            if interactive:
                                print(f"Failed to create the SSL certificate for {domain_name}. {output} {last_error}\n"
                                      "Please correct the domain name or leave empty to cancel.")

                            try:
                                domain_name = core.prompt(
                                    f"Domain Name for {public_ip}: ")
                                if not domain_name:
                                    if interactive:
                                        print(
                                            "Skipping SSL certificate generation.")
                                    domain_name = ""
                                    continue
                            except:
                                domain_name = ""
                                continue

                    if ssl_cert_created:
                        # Install Certificate
                        conn.execute("sudo mkdir -p /etc/pki/nginx/private")
                        # cSpell:ignore reloadcmd
                        output = conn.execute(f"sudo /home/opc/.acme.sh/acme.sh --install-cert -d {domain_name} "
                                              "--key-file /etc/pki/nginx/private/key.pem "
                                              "--fullchain-file /etc/pki/nginx/cert.pem "
                                              '--reloadcmd "sudo systemctl restart nginx.service" '
                                              "--force --home /home/opc/.acme.sh")
                        last_error = conn.get_last_error()
                        if last_error != "":
                            raise Exception(
                                f"Failed to install SSL certificate. {output} {last_error}")

                if interactive:
                    print("Writing web server configuration...")

                # cSpell:ignore letsencrypt fullchain privkey
                if domain_name != "":
                    ssl_cert_path = "/etc/pki/nginx/cert.pem"
                    ssl_cert_key_path = "/etc/pki/nginx/private/key.pem"
                    # ssl_cert_path = f"/etc/letsencrypt/live/{domain_name}/fullchain.pem"
                    # ssl_cert_key_path = f"/etc/letsencrypt/live/{domain_name}/privkey.pem"
                else:
                    ssl_cert_path = "/var/lib/mysqlrouter/router-cert.pem"
                    ssl_cert_key_path = "/var/lib/mysqlrouter/router-key.pem"

                nginx_config = f"""server {{
    listen 443 ssl http2;

    {f"server_name {domain_name};" if domain_name != "" else ""}

    ssl_certificate {ssl_cert_path};
    ssl_certificate_key {ssl_cert_key_path};

    # Allow large attachments
    client_max_body_size 128M;

    location / {{
        proxy_pass http://127.0.0.1:8446;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }}
}}
"""

                # Write config to in-memory stream
                with io.BytesIO() as nginx_config_bytes_stream:
                    with io.StringIO() as nginx_config_stream:
                        nginx_config_stream.write(nginx_config)

                        # Seek to the beginning of the text stream
                        nginx_config_bytes_stream.write(
                            nginx_config_stream.getvalue().encode("utf-8"))
                        nginx_config_bytes_stream.seek(0)

                        # Write out new config file to remote instance
                        try:
                            conn.put_local_file_object(
                                nginx_config_bytes_stream,
                                "/home/opc/mrs.nginx.conf")
                        except Exception as e:
                            raise Exception(
                                "Could not upload router config file. "
                                f"{str(e)}")

                # Move config to final place and fix privileges
                conn.execute(
                    "sudo cp /home/opc/mrs.nginx.conf /etc/nginx/conf.d/mrs.nginx.conf")
                conn.execute(
                    "sudo chown root:root /etc/nginx/conf.d/mrs.nginx.conf")
                conn.execute("sudo rm /home/opc/mrs.nginx.conf")

                # If there was an error, print it
                last_error = conn.get_last_error()
                if last_error != "":
                    raise Exception(
                        "Could not upload web server config file. "
                        f"ERROR: {last_error}")

                if interactive:
                    print("Opening Firewall ports...")

                # Open MySQL Router ports on the firewall
                if port_forwarding:
                    conn.execute(
                        "sudo firewall-cmd --zone=public --permanent --add-port=6446-6449/tcp")
                # If mrs was requested but no public_ip, open the port for a Load Balancer
                if mrs and not public_ip:
                    conn.execute(
                        "sudo firewall-cmd --zone=public --permanent --add-port=8446/tcp")
                conn.execute("sudo firewall-cmd --reload")

                # If mrs was requested but no public_ip, open the port for a Load Balancer
                # cSpell:ignore semanage mysqld
                if mrs and not public_ip:
                    conn.execute(
                        "sudo semanage port -a -t mysqld_port_t -p tcp 8446")

                if interactive:
                    print("Restarting MySQL Router...")

                # Restart mysqlrouter.service
                conn.execute("sudo systemctl restart mysqlrouter.service")

                if interactive:
                    print("Restarting web server...")

                # Restart NGINX
                conn.execute("sudo systemctl restart nginx.service")

            # Add ingress rules for MySQL ports to security list
            compute.add_ingress_port_to_security_lists(
                security_lists=sec_lists, port=6446,
                description="Classic MySQL Protocol RW via Router",
                compartment_id=compartment_id, config=config,
                interactive=interactive,
                raise_exceptions=raise_exceptions)
            compute.add_ingress_port_to_security_lists(
                security_lists=sec_lists, port=6447,
                description="Classic MySQL Protocol RO via Router",
                compartment_id=compartment_id, config=config,
                interactive=interactive,
                raise_exceptions=raise_exceptions)
            compute.add_ingress_port_to_security_lists(
                security_lists=sec_lists, port=6448,
                description="MySQL X Protocol RW via Router",
                compartment_id=compartment_id, config=config,
                interactive=interactive,
                raise_exceptions=raise_exceptions)
            compute.add_ingress_port_to_security_lists(
                security_lists=sec_lists, port=6449,
                description="MySQL X Protocol RO via Router",
                compartment_id=compartment_id, config=config,
                interactive=interactive,
                raise_exceptions=raise_exceptions)
            if mrs and not public_ip:
                compute.add_ingress_port_to_security_lists(
                    security_lists=sec_lists, port=8446,
                    description="MRS HTTP via Router",
                    compartment_id=compartment_id, config=config,
                    interactive=interactive,
                    raise_exceptions=raise_exceptions)
            elif mrs:
                compute.add_ingress_port_to_security_lists(
                    security_lists=sec_lists, port=443,
                    description="MRS HTTPS via Router",
                    compartment_id=compartment_id, config=config,
                    interactive=interactive,
                    raise_exceptions=raise_exceptions)
                compute.add_ingress_port_to_security_lists(
                    security_lists=sec_lists, port=80,
                    description="MRS HTTP via Router",
                    compartment_id=compartment_id, config=config,
                    interactive=interactive,
                    raise_exceptions=raise_exceptions)

            if interactive:
                endpoint_address = domain_name if domain_name != "" else public_ip
                print("\nNew endpoint successfully created.\n")
                if port_forwarding:
                    print(f"    Classic MySQL Protocol: {endpoint_address}:6446\n"
                          f"    MySQL X Protocol: {endpoint_address}:6448\n")
                if mrs:
                    print(
                        f"    MySQL REST Service HTTPS: https://{endpoint_address}/\n")
                if port_forwarding:
                    print(
                        f"Example:\n    mysqlsh mysql://{mysql_user_name}@{endpoint_address}:6446")

            if not return_formatted:
                return {
                    "ip": public_ip,
                    "domainName": domain_name,
                    "port": 6446,
                    "port_x": 6447,
                    "rest_http": 443
                }
        except Exception as e:
            raise Exception(
                f"Could not configure the compute instance '{instance_name}' "
                f"at {public_ip}.\n{str(e)}")
    except Exception as e:
        if raise_exceptions:
            raise
        print(f"ERROR: {str(e)}")


def create_bucket_import_pars(object_name_prefix, bucket_name, db_system_name,
                              compartment_id, config):
    """Creates the PARs needed for an import from a bucket

    Args:
        object_name_prefix (str): The prefix used for the object names
        bucket_name (str): The name of the bucket
        db_system_name (str): The name of the MySQL DB System
        compartment_id (str): The OCID of the compartment
        config (object): An OCI config object or None.
    """
    from mds_plugin import object_store

    par = object_store.create_bucket_object_par(
        bucket_object_name=f'{object_name_prefix}@.manifest.json',
        bucket_name=bucket_name,
        access_type="r",
        compartment_id=compartment_id,
        config=config,
        return_object=True)
    if par is None:
        print(f"Could not create PAR for manifest file. Operation cancelled.\n")

    progress_par = object_store.create_bucket_object_par(
        bucket_object_name=f'{object_name_prefix}@.{db_system_name}.progress',
        bucket_name=bucket_name,
        access_type="rw",
        compartment_id=compartment_id,
        config=config,
        return_object=True)
    if progress_par is None:
        print(f"Could not create PAR for progress file. Operation cancelled.\n")

    return par, progress_par


@plugin_function('mds.util.importDumpFromBucket')
def import_from_bucket(**kwargs):
    """Imports a dump from a bucket on a DB System

    Args:
        **kwargs: Additional options

    Keyword Args:
        bucket_name (str): The name of the bucket
        object_name_prefix (str): The prefix used for the object names
        db_system_name (str): The name of the MySQL DB System
        db_system_id (str): The OCID of the db_system
        db_system_ip (str): The IP to use for the import, overwriting the
                one from the db_system if given
        db_system_port (str): The Port to use for the import, overwriting
                the one from the db_system if given
        admin_username (str): The name of the administrator user account
        admin_password (str): The password of the administrator account
        private_key_file_path (str): The file path to an SSH private key
        perform_cleanup (bool): Whether the PARs and bucket should be deleted
        compartment_id (str): The OCID of the compartment
        config (object): An OCI config object or None.
        interactive (bool): Whether user input is considered
    Returns:
       None
    """

    bucket_name = kwargs.get('bucket_name')
    db_system_name = kwargs.get('db_system_name')
    private_key_file_path = kwargs.get(
        "private_key_file_path", "~/.ssh/id_rsa")
    object_name_prefix = kwargs.get("object_name_prefix", "")
    admin_username = kwargs.get('admin_username')
    admin_password = kwargs.get('admin_password')
    db_system_id = kwargs.get("db_system_id")
    db_system_ip = kwargs.get("db_system_ip")
    db_system_port = kwargs.get("db_system_port")
    perform_cleanup = kwargs.get("perform_cleanup")
    compartment_id = kwargs.get("compartment_id")
    config = kwargs.get("config")
    interactive = kwargs.get("interactive", True)

    # Get the active config, compartment and db_system
    try:
        config = configuration.get_current_config(config=config)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        db_system_id = configuration.get_current_db_system_id(
            db_system_id=db_system_id, config=config)
        bucket_name = configuration.get_current_bucket_name(
            bucket_name=bucket_name, config=config)
    except ValueError as e:
        if not interactive:
            raise
        print(f"ERROR: {str(e)}")
        return

    from mds_plugin import compute, user, object_store
    import datetime
    import time
    import base64
    import json
    import mysqlsh

    # If no explicit IP is given, get DB System
    if not db_system_ip:
        db_system = mysql_database_service.get_db_system(
            db_system_name=db_system_name, db_system_id=db_system_id,
            compartment_id=compartment_id, config=config,
            interactive=interactive)
        if db_system is None:
            print("No MySQL DB System given. Operation cancelled.\n")
            return

        # Get the first active endpoint
        endpoints = [e for e in db_system.endpoints if e.status == 'ACTIVE']
        if len(endpoints) < 1:
            print("ERROR: This MySQL DB System has no active endpoints "
                  "assigned.")
            return
        endpoint = endpoints[0]

        db_system_name = db_system.display_name
        db_system_ip = endpoint.ip_address
        db_system_port = endpoint.port

    if not db_system_port:
        db_system_port = '3306'

    if not db_system_name:
        db_system_name = 'DbSystem'

    # Get ObjectStorage namespace
    os_namespace = object_store.get_object_store_namespace(config=config)

    # Get bucket
    bucket = object_store.get_bucket(
        bucket_name=bucket_name,
        compartment_id=compartment_id,
        config=config,
        ignore_current=False,
        interactive=interactive)
    if bucket is None:
        print("No bucket given. Operation cancelled.\n")
        return

    # Check if manifest is available on this bucket
    obj = object_store.get_bucket_object(
        name=object_name_prefix + '@.manifest.json',
        bucket_name=bucket.name,
        compartment_id=None,
        config=None,
        no_error_on_not_found=False)
    if obj is None:
        print(f"Manifest file '{object_name_prefix}@.manifest.json' not "
              f"found on bucket '{bucket.name}'. Operation cancelled.\n")
        return

    # Get an admin_username
    if interactive and not admin_username:
        admin_username = mysqlsh.globals.shell.prompt(
            "MySQL Administrator account name [admin]: ",
            {'defaultValue': 'admin'}).strip()
    if not admin_username:
        print("Operation cancelled.")
        return

    # Get an admin_password
    if interactive and admin_password is None:
        admin_password = mysqlsh.globals.shell.prompt(
            "MySQL Administrator account password: ",
            {'defaultValue': '', 'type': 'password'}).strip()
    if not admin_password:
        print("Operation cancelled.")
        return

    # Create PAR for manifest and progress files
    par, progress_par = create_bucket_import_pars(
        object_name_prefix=object_name_prefix,
        bucket_name=bucket.name,
        db_system_name=db_system_name,
        compartment_id=compartment_id,
        config=config)
    if par is None or progress_par is None:
        print("Could not create pre-authenticated requests. "
              "Operation cancelled.")
        return

    # Build URLs
    par_url_prefix = object_store.get_par_url_prefix(config=config)
    par_url = par_url_prefix + par.access_uri
    progress_par_url = par_url_prefix + progress_par.access_uri

    try:
        # Get the 'MySQLDBBastionHost' compute instance (let the user create it
        # if it does not exist yet)
        mds_proxy = create_compute_instance_for_endpoint(
            private_key_file_path=private_key_file_path,
            compartment_id=compartment_id, config=config,
            interactive=False)
        if mds_proxy is None:
            print("Could not get a compute instance. Operation cancelled.")
            return

        # Get the public IP of the instance
        public_ip = compute.get_instance_public_ip(
            instance_id=mds_proxy.id, compartment_id=compartment_id,
            config=config, private_ip_fallback=True)
        if public_ip is None or public_ip == "":
            print("The public IP of the MySQLDBBastionHost instance could not "
                  "be fetched.")
            return

        # Open an SSH connection to the instance
        print(f"Connecting to MySQLDBBastionHost instance at {public_ip}...")
        try:
            success = None
            with compute.SshConnection(
                    username="opc", host=public_ip,
                    private_key_file_path=private_key_file_path) as conn:

                print(f"Connected to MySQLDBBastionHost instance at "
                      f"{public_ip}.\n"
                      f"Starting import...")

                # Build command string to execute
                cmd = ['mysqlsh',
                       (f'{admin_username}@'
                        f'{endpoint.ip_address}:{endpoint.port}'),
                       '--passwords-from-stdin',
                       '--save-passwords=never',
                       '--',
                       'util',
                       'load-dump',
                       f'{par_url}',
                       f'--osBucketName={bucket.name}',
                       f'--osNamespace={os_namespace}',
                       f'--progress-file={progress_par_url}',
                       '--loadUsers=true',
                       '--showProgress=true',
                       '--ignoreVersion=true']
                cmd = " ".join(cmd)

                # Open channel
                chan = conn.client.get_transport().open_session()
                try:
                    chan.settimeout(timeout=None)
                    chan.set_combine_stderr(combine=True)

                    # Execute shell and call import function
                    chan.exec_command(cmd)

                    # Send password to stdin
                    chan.sendall(f"{admin_password}\n".encode('utf-8'))
                    chan.shutdown_write()

                    def print_lines(buffer, no_line_break=False):
                        for line in buffer:
                            if "Please provide the password for" \
                                    in line:
                                line = line.replace(
                                    "Please provide the password for",
                                    "Importing dump on")
                            if 'Worker' in line:
                                line += ''
                            if line.strip() != "":
                                if no_line_break:
                                    print(line, end='\r')
                                else:
                                    print(f'>> {line}')

                    # While the command didn't return an exit code yet
                    read_buffer_size = 1024
                    buffer = ""
                    while not chan.exit_status_ready():
                        # Update every 0.1 seconds
                        time.sleep(0.1)

                        if chan.recv_ready():
                            buffer += chan.recv(
                                read_buffer_size).decode('utf-8')
                            # Check if a line break was in the buffer
                            if "\n" in buffer:
                                end = buffer.rfind("\n")+1
                                ready = buffer[:end]
                                buffer = buffer[end:]
                                # Print the lines that are ready
                                print_lines(ready.split("\n"))
                            elif "\r" in buffer:
                                end = buffer.rfind("\r")+1
                                ready = buffer[:end]
                                buffer = buffer[end:]
                                # Print the lines that are ready
                                print_lines(
                                    ready.split("\r"), no_line_break=True)

                    # Ensure we gobble up all remaining data
                    while True:
                        try:
                            output = chan.recv(
                                read_buffer_size).decode('utf-8')
                            if not output and not chan.recv_ready():
                                break
                            else:
                                buffer += output
                                # Check if a line break was in the buffer
                                if "\n" in buffer:
                                    end = buffer.rfind("\n")+1
                                    ready = buffer[:end]
                                    buffer = buffer[end:]
                                    # Print the lines that are ready
                                    print_lines(ready.split("\n"))
                        except Exception:
                            continue

                    # If there is still something in the buffer, print it
                    if buffer:
                        print(buffer)

                    success = chan.recv_exit_status() == 0
                finally:
                    chan.close()

            if success is not None:
                print(f"Connection to MySQLDBBastionHost instance closed.\n"
                      f"Import completed "
                      f"{'successfully' if success else 'with errors'}.")

        except Exception as e:
            print("Could not execute the import on compute instance "
                  f"'MySQLDBBastionHost' at {public_ip}.\nERROR: {str(e)}")
    finally:
        # Delete the PARs created especially for this import
        object_store.delete_par(
            bucket_name=bucket.name,
            par_id=par.id,
            compartment_id=compartment_id,
            config=config,
            interactive=False)
        object_store.delete_par(
            bucket_name=bucket.name,
            par_id=progress_par.id,
            compartment_id=compartment_id,
            config=config,
            interactive=False)

        if perform_cleanup:
            print("Performing cleanup...")

            if object_name_prefix:
                # Delete all PARs created by dump
                object_store.delete_par(
                    bucket_name=bucket.name,
                    name=f'shell-dump-{object_name_prefix}*',
                    compartment_id=compartment_id,
                    config=config,
                    interactive=False)

                # Delete all bucket objects
                object_store.delete_bucket_object(
                    name=f"{object_name_prefix}*",
                    bucket_name=bucket.name,
                    compartment_id=compartment_id,
                    config=config,
                    interactive=False)
            else:
                # If no object_name_prefix was used, also delete the bucket
                object_store.delete_bucket(
                    bucket_name=bucket.name,
                    compartment_id=compartment_id,
                    config=config,
                    interactive=False)

    print("Operation completed.")


@plugin_function('mds.util.importDumpFromLocalDir')
def import_from_local_dir(local_dump_dir=None, db_system_name=None,
                          options=None):
    """Imports a local dump on a DB System

    Args:
        local_dump_dir (str): The directory that holds the local dump
        db_system_name (str): The new name of the DB System.
        options (dict): A dict with various options
            object_name_prefix (str): The prefix used for the object names
            db_system_id (str): The OCID of the db_system
            admin_username (str): The name of the administrator user account
            admin_password (str): The password of the administrator user account
            private_key_file_path (str): The file path to an SSH private key
            compartment_id (str): The OCID of the compartment
            config (object): An OCI config object or None.
            interactive (bool): Whether user input is considered

    Returns:
       None
    """
    if options is None:
        options = {}

    object_name_prefix = options.get("object_name_prefix")
    db_system_id = options.get("db_system_id")
    admin_username = options.get("admin_username")
    admin_password = options.get("admin_password")
    private_key_file_path = options.get(
        "private_key_file_path", "~/.ssh/id_rsa")
    compartment_id = options.get("compartment_id")
    config = options.get("config")
    interactive = options.get("interactive", True)

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(config=config)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)

        import oci.mysql
        from pathlib import Path
        from os import listdir
        import os.path
        import datetime
        import mysqlsh
        import object_store

        if interactive:
            print("Preparing for data import from a local directory...\n")

        if interactive and local_dump_dir is None:
            local_dump_dir = mysqlsh.globals.shell.prompt(
                "Please specify the directory path that contains the dump: ",
                {'defaultValue': ''}).strip()
            if local_dump_dir == "":
                print("Operation cancelled.")
                return
        elif local_dump_dir is None:
            print("No directory path given. Operation cancelled.")
            return

        # Get an admin_username
        if interactive and admin_username is None:
            admin_username = mysqlsh.globals.shell.prompt(
                "MySQL Administrator account name [admin]: ",
                {'defaultValue': 'admin'}).strip()
            if admin_username == "":
                print("Operation cancelled.")
                return
        elif admin_username is None:
            admin_username = "admin"

        # Get an admin_password
        if interactive and admin_password is None:
            admin_password = mysqlsh.globals.shell.prompt(
                "MySQL Administrator account password: ",
                {'defaultValue': '', 'type': 'password'}).strip()
            if admin_password == "":
                print("Operation cancelled.")
                return
        elif admin_password is None:
            raise ValueError("The argument admin_password must be set.")

        # Get DB System
        db_system = mysql_database_service.get_db_system(
            db_system_name=db_system_name, db_system_id=db_system_id,
            compartment_id=compartment_id, config=config)
        if db_system is None:
            print("Operation cancelled.\n")
            return

        options["db_system_id"] = db_system.id

        # Make sure there is a 'MySQLDBBastionHost' compute instance (let the user
        # create it if it does not exist yet)
        mds_proxy = create_compute_instance_for_endpoint(
            private_key_file_path=private_key_file_path,
            compartment_id=db_system.compartment_id, config=config,
            interactive=False)
        if mds_proxy is None:
            print("Operation cancelled.")
            return

        # Take all alphanumeric chars from the DB System display_name
        # to create the bucket_name
        bucket_name = (
            f"{''.join(e for e in db_system.display_name if e.isalnum())}"
            f"_import_{datetime.datetime.now():%Y%m%d%H%M%S}")

        print(f"\nCreating bucket {bucket_name}...")

        bucket = object_store.create_bucket(
            bucket_name=bucket_name, compartment_id=compartment_id,
            config=config, return_object=True)
        if bucket is None:
            print("Cancelling operation")
            return

        # Upload the files from the given directory to the bucket
        file_count = object_store.create_bucket_objects_from_local_dir(
            local_dir_path=local_dump_dir, bucket_name=bucket.name,
            object_name_prefix=object_name_prefix,
            compartment_id=compartment_id, config=config, interactive=False)
        if file_count is None:
            print("Cancelling operation")
            return

        # Start the import from the bucket
        import_from_bucket(
            bucket_name=bucket.name,
            options=options)

        # if not object_store.delete_bucket_object(
        #         bucket_object_name="*", bucket_name=bucket.name,
        #         compartment_id=compartment_id, config=config,
        #         interactive=False):
        #     print("Could not delete files from buckets.")

        if not object_store.delete_bucket(
                bucket_name=bucket.name,
                compartment_id=compartment_id, config=config,
                interactive=False):
            print("Could not delete the bucket.")

        if interactive:
            print("Operation completed.")
    except oci.exceptions.ServiceError as e:
        if not interactive:
            raise
        print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
        return
    except (ValueError, oci.exceptions.ClientError) as e:
        if not interactive:
            raise
        print(f'ERROR: {e}')
        return


@plugin_function('mds.util.dumpToBucket')
def dump_to_bucket(**kwargs):
    """Imports a dump on a DB System using a PAR

    Args:
        **kwargs: Additional options

    Keyword Args:
        bucket_name (str): The name of the bucket
        create_bucket_if_not_exists (bool): Will create the bucket if set
        object_name_prefix (str): The prefix used for the object names
        connection_uri (str): The URI to the MySQL Server
        connection_password (str): The password of the MySQL
            user account
        compartment_id (str): The OCID of the compartment
        config (object): An OCI config object or None.
        config_file_path (str): The path to the config file.
        interactive (bool): Enables interactive input.
        return_true_on_success (bool): Whether to return true on success

    Returns:
       None or true on success if return_true_on_success is set to true
    """
    # Handle kwargs
    bucket_name = kwargs.get("bucket_name")
    create_bucket_if_not_exists = kwargs.get(
        "create_bucket_if_not_exists", False)
    object_name_prefix = kwargs.get("object_name_prefix")
    connection_uri = kwargs.get("connection_uri")
    connection_password = kwargs.get("connection_password")
    # If no config_file_path is given, first check the MYSQLSH_OCI_CONFIG_FILE env_var and only then fall back to
    # default
    config_file_path = kwargs.get(
        "config_file_path", getenv("MYSQLSH_OCI_CONFIG_FILE"))
    if config_file_path is None:
        config_file_path = "~/.oci/config"
    compartment_id = kwargs.get("compartment_id")
    config = kwargs.get("config")
    interactive = kwargs.get("interactive", True)
    return_true_on_success = kwargs.get("return_true_on_success")

    # Get the active config, compartment and db_system
    try:
        config = configuration.get_current_config(config=config)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        bucket_name = configuration.get_current_bucket_name(
            bucket_name=bucket_name, config=config)

        from mds_plugin import object_store
        import os.path
        import json
        import base64
        import subprocess
        from pathlib import Path
        import time
        import mysqlsh
        import oci.exceptions

        # If no connection_uri was given, try to use the one from the current
        # connection if this one uses the classic protocol
        if connection_uri is None:
            session = mysqlsh.globals.shell.get_session()
            if session is not None:
                if "mysql://" in session.get_uri():
                    connection_uri = session.get_uri()

        # If no connection_uri was given but interactive is True, ask the user
        if connection_uri is None and interactive:
            print("Please enter the MySQL Server connection URI for the MySQL "
                  "Server serving as data source.\nTo get more information about "
                  "the URI format type '\\? connection' in the MySQL Shell.\n"
                  "Example: admin@localhost:3306\n")
            connection_uri = mysqlsh.globals.shell.prompt(
                "MySQL Server connection URI: ")
            if connection_uri == "":
                print("Operation Cancelled.\n")
                return

        # Get an admin_password
        if connection_password is None:
            connection_password = mysqlsh.globals.shell.prompt(
                f"Password for {connection_uri}: ",
                {'defaultValue': '', 'type': 'password'}).strip()
            if connection_password == "":
                print("Operation cancelled.")
                return

        # Get object_name_prefix
        if interactive and object_name_prefix is None:
            print("\nIf needed, a prefix for the dumped object names can be set.\n"
                  "Default is to use no prefix.")
            object_name_prefix = mysqlsh.globals.shell.prompt(
                "\nPlease enter the object name prefix []: ",
                {'defaultValue': ''}).strip()
        elif object_name_prefix is None:
            object_name_prefix = ""

        # Get bucket or create one
        bucket_created = False
        try:
            bucket = object_store.get_bucket(
                bucket_name=bucket_name, compartment_id=compartment_id,
                config=config, interactive=False)
        except oci.exceptions.ServiceError as e:
            if create_bucket_if_not_exists:
                bucket = object_store.create_bucket(
                    bucket_name=bucket_name, compartment_id=compartment_id,
                    config=config, interactive=False, return_object=True)
            else:
                if not interactive:
                    raise
                print(f"The bucket {bucket_name} does not exist.")
                return

        os_namespace = object_store.get_object_store_namespace(config)

        # Convert Unix path to Windows
        config_file_path = os.path.abspath(
            os.path.expanduser(config_file_path))

        # Get the current profile
        profile_name = configuration.get_current_profile()

    except oci.exceptions.ServiceError as e:
        if not interactive:
            raise
        print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
        return
    except (ValueError, oci.exceptions.ClientError) as e:
        if not interactive:
            raise
        print(f'ERROR: {e}')
        return

    # cSpell:ignore ocimds, innodb
    # Setup command
    cmd = [f'mysqlsh',
           (f'{connection_uri}'),
           f'--passwords-from-stdin',
           f'--save-passwords=never',
           # f'--json',
           f'--',
           f'util',
           f'dump-instance',
           f'{object_name_prefix}',
           f'--osBucketName={bucket.name}',
           f'--osNamespace={os_namespace}',
           f'--ociConfigFile={config_file_path}',
           f'--ociProfile={profile_name}',
           '--showProgress=true',
           '--consistent=true',
           '--ociParManifest=true',
           '--ocimds=true']
    #    ('--compatibility=[force_innodb, strip_definers, '
    #    'strip_restricted_grants, strip_tablespaces]')]

    # Workaround till shell supports array arguments in command line API mode
    cmd = ('mysqlsh '
           f'{connection_uri} '
           f'--passwords-from-stdin '
           f'--save-passwords=never '
           f'-e "util.dumpInstance(\'{object_name_prefix}\', '
           '{'
           f'osBucketName: \'{bucket.name}\', '
           f'osNamespace: \'{os_namespace}\', '
           f'ociConfigFile: \'{config_file_path}\', '
           f'ociProfile: \'{profile_name}\', '
           'showProgress: true, '
           'consistent: true, '
           'ociParManifest: true, '
           'ocimds: true, '
           'compatibility: [\'force_innodb\', \'strip_definers\', '
           '\'strip_restricted_grants\', \'strip_tablespaces\']'
           '})"'
           )

    # Run command
    print(f'Starting dump from {connection_uri} to bucket {bucket.name} ...')
    try:
        with subprocess.Popen(
                cmd, shell=True,
                stdout=subprocess.PIPE,
                stdin=subprocess.PIPE, stderr=subprocess.STDOUT,
                universal_newlines=True, bufsize=1) as proc:
            try:
                # Give the process some startup time
                time.sleep(0.2)

                # Provide the password to stdin
                proc.stdin.write(connection_password + "\n")
                proc.stdin.flush()

                # Mirror the output TODO: fix hang on Windows
                for line in iter(proc.stdout.readline, ''):
                    if "Please provide the password for" in line:
                        line = line.replace(
                            "Please provide the password for ",
                            "Performing dump from ")

                    if line.strip() != "":
                        print(f'<< {line}', end='')

                # for line in iter(proc.stderr.readline, ''):
                #     print(f'<< ERROR: {line}', end='')

                proc.stdout.close()
                # proc.stderr.close()
                proc.stdin.close()
                return_code = proc.wait()
                if return_code:
                    # If a bucket has been created, delete it
                    if bucket_created:
                        object_store.delete_bucket(
                            bucket_name=bucket_name,
                            compartment_id=compartment_id, config=config,
                            interactive=False)
                    print("Dump has failed.")
                    if return_true_on_success:
                        return False
                else:
                    print("Dump has finished.")
                    if return_true_on_success:
                        return True
            except subprocess.TimeoutExpired as e:
                proc.kill()

    except OSError as e:
        if not interactive:
            raise e
        print(f"Error when starting shell import. {str(e)}")
    except ValueError as e:
        if not interactive:
            raise
        print(f"Invalid arguments passed to the shell import. {str(e)}")