# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

import argparse
import cluster as CLUSTER
import database
import dateutil.parser
import utils
import os
import os.path
import prettytable
import shutil
import sys
import time

LOG = utils.get_logger()


def wait_service(need_alive, wait_timeout, cluster, fe_ids, be_ids):
    if wait_timeout == 0:
        return
    if wait_timeout == -1:
        wait_timeout = 1000000000
    expire_ts = time.time() + wait_timeout
    while True:
        db_mgr = database.get_db_mgr(cluster.name,
                                     cluster.get_all_node_net_infos(), False)
        failed_frontends = []
        for id in fe_ids:
            fe = cluster.get_node(CLUSTER.Node.TYPE_FE, id)
            fe_state = db_mgr.get_fe(id)
            fe_alive = fe_state and fe_state.alive
            if fe_alive and need_alive:
                # if need alive, check port available,
                # if need dead, don't check port available because it take some time for the disconnect socket
                fe_alive = utils.is_socket_avail(
                    fe.get_ip(), fe.meta["ports"]["query_port"])
            if fe_alive != need_alive:
                failed_frontends.append(id)
        failed_backends = []
        for id in be_ids:
            be = cluster.get_node(CLUSTER.Node.TYPE_BE, id)
            be_state = db_mgr.get_be(id)
            be_alive = be_state and be_state.alive
            if be_alive and need_alive:
                be_alive = utils.is_socket_avail(
                    be.get_ip(), be.meta["ports"]["webserver_port"])
            if be_alive != need_alive:
                failed_backends.append(id)
        if not failed_frontends and not failed_backends:
            break
        if time.time() >= expire_ts:
            err = ""
            failed_status = "dead" if need_alive else "alive"
            if failed_frontends:
                err += failed_status + " fe: " + str(failed_frontends) + ". "
            if failed_backends:
                err += failed_status + " be: " + str(failed_backends) + ". "
            raise Exception(err)
        time.sleep(1)


# return for_all, related_nodes, related_node_num
def get_ids_related_nodes(cluster,
                          fe_ids,
                          be_ids,
                          ms_ids,
                          recycle_ids,
                          fdb_ids,
                          ignore_not_exists=False):
    if fe_ids is None and be_ids is None and ms_ids is None and recycle_ids is None and fdb_ids is None:
        return True, None, cluster.get_all_nodes_num()

    def get_ids_related_nodes_with_type(node_type, ids):
        if ids is None:
            return []
        if not ids:
            return cluster.get_all_nodes(node_type)
        else:
            nodes = []
            for id in ids:
                try:
                    nodes.append(cluster.get_node(node_type, id))
                except Exception as e:
                    if ignore_not_exists:
                        LOG.warning(
                            utils.render_yellow(
                                "Not found {} with id {}".format(
                                    node_type, id)))
                    else:
                        raise e
            return nodes

    type_ids = [
        (CLUSTER.Node.TYPE_FE, fe_ids),
        (CLUSTER.Node.TYPE_BE, be_ids),
        (CLUSTER.Node.TYPE_MS, ms_ids),
        (CLUSTER.Node.TYPE_RECYCLE, recycle_ids),
        (CLUSTER.Node.TYPE_FDB, fdb_ids),
    ]

    nodes = []
    for node_type, ids in type_ids:
        nodes.extend(get_ids_related_nodes_with_type(node_type, ids))

    return len(nodes) == cluster.get_all_nodes_num(), nodes, len(nodes)


class Command(object):

    def __init__(self, name):
        self.name = name

    def print_use_time(self):
        return True

    def add_parser(self, args_parsers):
        raise Exception("No implemented")

    def run(self, args):
        raise Exception("No implemented")

    def _add_parser_common_args(self, parser):
        parser.add_argument("-v",
                            "--verbose",
                            default=False,
                            action=self._get_parser_bool_action(True),
                            help="verbose logging.")
        parser.add_argument("--output-json",
                            default=False,
                            action=self._get_parser_bool_action(True),
                            help="output as json, and don't print log.")

    def _add_parser_ids_args(self, parser):
        group = parser.add_argument_group("for existing nodes",
                                          "apply to the existing nodes.")
        group.add_argument("--fe-id", nargs="*", type=int, help="Specify up fe ids, support multiple ids, " \
                "if specific --fe-id but not specific ids, apply to all fe. Example: '--fe-id 2 3' will select fe-2 and fe-3.")
        group.add_argument("--be-id", nargs="*", type=int, help="Specify up be ids, support multiple ids, " \
                "if specific --be-id but not specific ids, apply to all be. Example: '--be-id' will select all backends.")
        group.add_argument(
            "--ms-id",
            nargs="*",
            type=int,
            help=
            "Specify up ms ids, support multiple ids. Only use in cloud cluster."
        )
        group.add_argument(
            "--recycle-id",
            nargs="*",
            type=int,
            help=
            "Specify up recycle ids, support multiple ids. Only use in cloud cluster."
        )
        group.add_argument(
            "--fdb-id",
            nargs="*",
            type=int,
            help=
            "Specify up fdb ids, support multiple ids. Only use in cloud cluster."
        )

    def _get_parser_bool_action(self, is_store_true):
        if self._support_boolean_action():
            return argparse.BooleanOptionalAction
        else:
            return "store_true" if is_store_true else "store_false"

    def _support_boolean_action(self):
        return sys.version_info.major == 3 and sys.version_info.minor >= 9

    def _print_table(self, header, datas):
        if utils.is_log_stdout():
            table = prettytable.PrettyTable(
                [utils.render_green(field) for field in header])
            for row in datas:
                table.add_row(row)
            print(table)
            return ""
        else:
            datas.insert(0, header)
            return datas


class SimpleCommand(Command):

    def __init__(self, command, help, options=[]):
        super().__init__(command)
        self.command = command
        self.help = help
        self.options = options

    def add_parser(self, args_parsers):
        help = self.help + " If none of --fe-id, --be-id, --ms-id, --recycle-id, --fdb-id is specific, "\
                "then apply to all containers."
        parser = args_parsers.add_parser(self.command, help=help)
        parser.add_argument("NAME", help="Specify cluster name.")
        self._add_parser_ids_args(parser)
        self._add_parser_common_args(parser)
        return parser

    def run(self, args):
        cluster = CLUSTER.Cluster.load(args.NAME)
        for_all, related_nodes, related_node_num = get_ids_related_nodes(
            cluster, args.fe_id, args.be_id, args.ms_id, args.recycle_id,
            args.fdb_id)
        utils.exec_docker_compose_command(cluster.get_compose_file(),
                                          self.command,
                                          options=self.options,
                                          nodes=related_nodes)
        show_cmd = self.command[0].upper() + self.command[1:]

        if for_all:
            related_nodes = cluster.get_all_nodes()

        LOG.info(
            utils.render_green("{} succ, total related node num {}".format(
                show_cmd, related_node_num)))

        return cluster, related_nodes


class StartBaseCommand(SimpleCommand):

    def add_parser(self, args_parsers):
        parser = super().add_parser(args_parsers)
        parser.add_argument(
            "--wait-timeout",
            type=int,
            default=0,
            help=
            "Specify wait seconds for fe/be ready for service: 0 not wait (default), "\
            "> 0 max wait seconds, -1 wait unlimited."
        )
        return parser

    def run(self, args):
        cluster, related_nodes = super().run(args)
        fe_ids = [node.id for node in related_nodes if node.is_fe()]
        be_ids = [node.id for node in related_nodes if node.is_be()]
        if not cluster.is_host_network():
            wait_service(True, args.wait_timeout, cluster, fe_ids, be_ids)
        return cluster, related_nodes


class StartCommand(StartBaseCommand):

    def __init__(self, command):
        super().__init__(command, "Start the doris containers. "),


class RestartCommand(StartBaseCommand):

    def __init__(self, command):
        super().__init__(command, "Restart the doris containers. ",
                         ["-t", "1"]),


class StopCommand(SimpleCommand):

    def __init__(self, command):
        super().__init__(command, "Stop the doris containers. ", ["-t", "1"]),

    def add_parser(self, args_parsers):
        parser = super().add_parser(args_parsers)
        parser.add_argument(
            "--wait-timeout",
            type=int,
            default=0,
            help=
            "Specify wait seconds for fe/be close for service: 0 not wait (default), "\
            "> 0 max wait seconds, -1 wait unlimited."
        )
        return parser

    def run(self, args):
        cluster, related_nodes = super().run(args)
        fe_ids = [node.id for node in related_nodes if node.is_fe()]
        be_ids = [node.id for node in related_nodes if node.is_be()]
        if not cluster.is_host_network():
            wait_service(False, args.wait_timeout, cluster, fe_ids, be_ids)
        return cluster, related_nodes


class UpCommand(Command):

    def add_parser(self, args_parsers):
        parser = args_parsers.add_parser("up", help="Create and upgrade doris containers, "\
                "or add new containers. " \
                "If none of --add-fe-num, --add-be-num, --add-ms-num, --add-recycle-num, "\
                "--fe-id, --be-id, --ms-id, --recycle-id, --fdb-id is specific, " \
                "then apply to all containers.")
        parser.add_argument("NAME", default="", help="Specific cluster name.")
        parser.add_argument("IMAGE",
                            default="",
                            nargs="?",
                            help="Specify docker image.")

        self._add_parser_common_args(parser)
        parser.add_argument(
            "--cloud",
            default=False,
            action=self._get_parser_bool_action(True),
            help=
            "Create cloud cluster, default is false. Only use when creating new cluster."
        )
        parser.add_argument(
            "--root",
            default=False,
            action=self._get_parser_bool_action(True),
            help=
            "Run cluster as root user, default is false, it will run as host user."
        )

        parser.add_argument(
            "--wait-timeout",
            type=int,
            default=0,
            help=
            "Specify wait seconds for fe/be ready for service: 0 not wait (default), "\
            "> 0 max wait seconds, -1 wait unlimited."
        )

        group1 = parser.add_argument_group("add new nodes",
                                           "add cluster nodes.")
        group1.add_argument(
            "--add-fe-num",
            type=int,
            help=
            "Specify add fe num, default: 3 for a new cluster, 0 for a existing cluster."
        )
        group1.add_argument(
            "--add-be-num",
            type=int,
            help=
            "Specify add be num, default: 3 for a new cluster, 0 for a existing cluster."
        )
        group1.add_argument(
            "--add-ms-num",
            type=int,
            help=
            "Specify add ms num, default: 1 for a new cloud cluster, 0 for a existing cluster. Only use in cloud cluster"
        )
        group1.add_argument(
            "--add-recycle-num",
            type=int,
            help=
            "Specify add recycle num, default: 1 for a new cloud cluster, 0 for a existing cluster. Only use in cloud cluster"
        )
        group1.add_argument("--fe-config",
                            nargs="*",
                            type=str,
                            help="Specify fe configs for fe.conf. "\
                                "Example: --fe-config \"enable_debug_points = true\" \"sys_log_level = ERROR\".")
        group1.add_argument("--be-config",
                            nargs="*",
                            type=str,
                            help="Specify be configs for be.conf. "\
                                    "Example: --be-config \"enable_debug_points = true\" \"enable_auth = true\".")
        group1.add_argument("--ms-config",
                            nargs="*",
                            type=str,
                            help="Specify ms configs for doris_cloud.conf. "\
                                    "Example: --ms-config \"log_level = warn\".")
        group1.add_argument("--recycle-config",
                            nargs="*",
                            type=str,
                            help="Specify recycle configs for doris_cloud.conf. "\
                                    "Example: --recycle-config \"log_level = warn\".")

        group1.add_argument(
            "--fe-follower",
            default=False,
            action=self._get_parser_bool_action(True),
            help=
            "The new added fe is follower but not observer. Only support in cloud mode."
        )
        group1.add_argument("--be-disks",
                            nargs="*",
                            default=None,
                            type=str,
                            help="Specify each be disks, each group is \"disk_type=disk_num[,disk_capactity]\", "\
                                    "disk_type is HDD or SSD, disk_capactity is capactity limit in gb. default: HDD=1. "\
                                  "Example: --be-disks \"HDD=1\", \"SSD=1,10\", \"SSD=2,100\""\
                                  "means each be has 1 HDD without capactity limit, 1 SSD with 10GB capactity limit, "\
                                  "2 SSD with 100GB capactity limit")
        group1.add_argument(
            "--be-cluster",
            type=str,
            help=
            "be cluster name, if not specific, will use compute_cluster. Only use in cloud cluster."
        )

        self._add_parser_ids_args(parser)

        group2 = parser.add_mutually_exclusive_group()
        if self._support_boolean_action():
            group2.add_argument(
                "--start",
                default=True,
                action=self._get_parser_bool_action(False),
                help="Start containers, default is true. If specific --no-start, "\
                "will create or update config image only but not start containers.")
        else:
            group2.add_argument(
                "--no-start",
                dest='start',
                default=True,
                action=self._get_parser_bool_action(False),
                help=
                "Create or update config image only and don't start containers."
            )
        group2.add_argument("--force-recreate",
                           default=False,
                           action=self._get_parser_bool_action(True),
                           help="Recreate containers even if their configuration " \
                                "and image haven't changed. ")

        parser.add_argument(
            "--extra-hosts",
            nargs="*",
            type=str,
            help=
            "Add custom host-to-IP mappings (host:ip). For example: --extra-hosts myhost1:192.168.10.1 myhost2:192.168.10.2 . Only use when creating new cluster."
        )

        parser.add_argument(
            "--env",
            nargs="*",
            type=str,
            help=
            "Add environment variables. For example: --env KEY1=VALUE1 KEY2=VALUE2. Only use when creating new cluster."
        )
        parser.add_argument(
            "--cloud-config",
            nargs="*",
            type=str,
            help=
            "Override cloud store config values. For example: --cloud-config DORIS_CLOUD_AK=xxx DORIS_CLOUD_BUCKET=yyy. Only use when creating new cloud cluster."
        )

        parser.add_argument("--coverage-dir",
                            default="",
                            help="Set code coverage output directory")

        parser.add_argument("--sql-mode-node-mgr",
                            default=False,
                            action=self._get_parser_bool_action(True),
                            help="Manager fe be via sql instead of http")

        parser.add_argument(
            "--remote-master-fe",
            type=str,
            help=
            "Specify remote master fe address with ip:query_port, and all the container use host network. " \
                "Only use when creating new cluster."
        )

        parser.add_argument(
            "--local-network-ip",
            type=str,
            help= "Specify local network ip, no need specify, will auto chose a proper ip. "\
                "Only use when creating new cluster and specify --remote-master-fe."
        )

        parser.add_argument(
            "--external-ms",
            type=str,
            help=
            "Use external meta service cluster (specify cluster name). " \
            "This cluster will not create its own MS/FDB/Recycler, but use the specified cluster's services. " \
            "The external cluster must be a cloud cluster with MS/FDB already running. " \
            "Example: --external-ms shared-meta. Only use when creating new cloud cluster."
        )

        parser.add_argument(
            "--instance-id",
            type=str,
            help=
            "Specify instance ID for cloud mode. If not specified, will auto-generate 'default_instance_id'. " \
            "When using external MS with multiple clusters, each cluster should have a unique instance ID. " \
            "Example: --instance-id prod_instance_1"
        )

        parser.add_argument(
            "--cluster-snapshot",
            type=str,
            help=
            "Cluster snapshot JSON content for FE-1 first startup in cloud mode only. " \
            "The JSON will be written to FE conf/cluster_snapshot.json and passed to start_fe.sh " \
            "with --cluster_snapshot parameter. Only effective on first startup. " \
            "Example: --cluster-snapshot '{\"instance_id\":\"instance_id_xxx\"}'"
        )

        if self._support_boolean_action():
            parser.add_argument(
                "--be-metaservice-endpoint",
                default=True,
                action=self._get_parser_bool_action(False),
                help=
                "Do not set BE meta service endpoint in conf. Default is False."
            )
        else:
            parser.add_argument(
                "--no-be-metaservice-endpoint",
                dest='be_metaservice_endpoint',
                default=True,
                action=self._get_parser_bool_action(False),
                help=
                "Do not set BE meta service endpoint in conf. Default is False."
            )

        # if default==False, use this style to parser, like --be-cluster-id
        parser.add_argument(
            "--be-cluster-id",
            default=False,
            action=self._get_parser_bool_action(True),
            help="Do not set BE cluster ID in conf. Default is False.")

        parser.add_argument(
            "--fdb-version",
            type=str,
            default="7.3.69",
            help="fdb image version. Only use in cloud cluster.")
        parser.add_argument(
            "--fdb-image",
            type=str,
            default=os.environ.get("DORIS_FDB_IMAGE", ""),
            help="Override fdb image. Only use in cloud cluster.")

        parser.add_argument(
            "--tde-ak",
            type=str,
            default="",
            help="tde ak")

        parser.add_argument(
            "--tde-sk",
            type=str,
            default="",
            help="tde sk")

        # if default==True, use this style to parser, like --detach
        if self._support_boolean_action():
            parser.add_argument(
                "--detach",
                default=True,
                action=self._get_parser_bool_action(False),
                help="Detached mode: Run containers in the background. If specific --no-detach, "\
                "will run containers in frontend. ")
        else:
            parser.add_argument("--no-detach",
                                dest='detach',
                                default=True,
                                action=self._get_parser_bool_action(False),
                                help="Run containers in frontend. ")

        if self._support_boolean_action():
            parser.add_argument(
                "--reg-be",
                default=True,
                action=self._get_parser_bool_action(False),
                help="Register be to meta server in cloud mode, use for multi clusters test. If specific --no-reg-be, "\
                "will not register be to meta server. ")
        else:
            parser.add_argument(
                "--no-reg-be",
                dest='reg_be',
                default=True,
                action=self._get_parser_bool_action(False),
                help=
                "Don't register be to meta server in cloud mode, use for multi clusters test"
            )

    def run(self, args):
        if not args.NAME:
            raise Exception("Need specific not empty cluster name")
        for_all = True
        add_fdb_num = 0
        is_new_cluster = False
        try:
            cluster = CLUSTER.Cluster.load(args.NAME)

            if not cluster.is_cloud:
                args.add_ms_num = None
                args.add_recycle_num = None
                args.ms_id = None
                args.recycle_id = None
                args.fdb_id = None

            if args.fe_id != None or args.be_id != None \
                or args.ms_id != None or args.recycle_id != None or args.fdb_id != None \
                or args.add_fe_num or args.add_be_num \
                or args.add_ms_num or args.add_recycle_num:
                for_all = False
        except:
            # a new cluster
            is_new_cluster = True
            if not args.IMAGE:
                raise Exception("New cluster must specific image") from None
            if args.fe_id != None:
                args.fe_id = None
                LOG.warning(
                    utils.render_yellow("Ignore --fe-id for new cluster"))
            if args.be_id != None:
                args.be_id = None
                LOG.warning(
                    utils.render_yellow("Ignore --be-id for new cluster"))

            args.fdb_id = None
            args.ms_id = None
            args.recycle_id = None

            if args.add_fe_num is None:
                args.add_fe_num = 0 if args.remote_master_fe else 3
            if args.add_be_num is None:
                args.add_be_num = 3

            cloud_store_config = {}
            if args.cloud:
                external_ms_cluster = getattr(args, 'external_ms', None)
                if external_ms_cluster:
                    # Using the MS nodes from external cluster, no need to add FDB/MS/Recycler
                    self._validate_external_ms_cluster(external_ms_cluster)
                    add_fdb_num = 0
                    args.add_ms_num = 0
                    args.add_recycle_num = 0
                    LOG.info(f"Using external MS cluster: {external_ms_cluster}")
                else:
                    add_fdb_num = 1
                    if not args.add_ms_num:
                        args.add_ms_num = 1
                    if not args.add_recycle_num:
                        args.add_recycle_num = 1
                    external_ms_cluster = None

                if not args.be_cluster:
                    args.be_cluster = "compute_cluster"
                cloud_store_config = self._merge_cloud_store_config(
                    self._get_cloud_store_config(), args.cloud_config
                )
            else:
                args.add_ms_num = 0
                args.add_recycle_num = 0
                external_ms_cluster = None

            if args.remote_master_fe:
                if not args.local_network_ip:
                    args.local_network_ip = utils.get_local_ip()
                parts = args.remote_master_fe.split(":")
                if len(parts) != 2:
                    raise Exception(
                        f"invalid --remote-master-fe-addr {args.remote_master_fe}, should be 'ip:query_port'"
                    )
                if not parts[0]:
                    args.remote_master_fe = args.local_network_ip + ":" + parts[
                        1]
                if args.cloud:
                    args.sql_mode_node_mgr = True

            instance_id = getattr(args, 'instance_id', None)
            cluster_snapshot = getattr(args, 'cluster_snapshot', '')
            enable_storage_vault = CLUSTER.is_true(
                CLUSTER.get_env_value(args.env, "ENABLE_STORAGE_VAULT"))

            cluster = CLUSTER.Cluster.new(
                args.NAME, args.IMAGE, args.cloud, args.root, args.fe_config,
                args.be_config, args.ms_config, args.recycle_config,
                args.remote_master_fe, args.local_network_ip, args.fe_follower,
                args.be_disks if args.be_disks is not None else ["HDD=1"], args.be_cluster, args.reg_be, args.extra_hosts, args.env,
                args.coverage_dir, cloud_store_config, args.sql_mode_node_mgr,
                args.be_metaservice_endpoint, args.be_cluster_id, args.tde_ak, args.tde_sk,
                external_ms_cluster, instance_id, cluster_snapshot,
                enable_storage_vault)
            LOG.info("Create new cluster {} succ, cluster path is {}".format(
                args.NAME, cluster.get_path()))

        if args.be_cluster and cluster.is_cloud:
            cluster.be_cluster = args.be_cluster

        if cluster.is_cloud:
            cluster.fe_follower = args.fe_follower

        _, related_nodes, _ = get_ids_related_nodes(cluster, args.fe_id,
                                                    args.be_id, args.ms_id,
                                                    args.recycle_id,
                                                    args.fdb_id)
        add_fe_ids = []
        add_be_ids = []
        add_ms_ids = []
        add_recycle_ids = []
        add_fdb_ids = []

        add_type_nums = [
            (CLUSTER.Node.TYPE_FDB, add_fdb_num, add_fdb_ids),
            (CLUSTER.Node.TYPE_MS, args.add_ms_num, add_ms_ids),
            (CLUSTER.Node.TYPE_RECYCLE, args.add_recycle_num, add_recycle_ids),
            (CLUSTER.Node.TYPE_FE, args.add_fe_num, add_fe_ids),
            (CLUSTER.Node.TYPE_BE, args.add_be_num, add_be_ids),
        ]

        if not related_nodes:
            related_nodes = []

        def do_add_node(node_type, add_num, add_ids):
            if not add_num:
                return
            for i in range(add_num):
                node = cluster.add(node_type)
                related_nodes.append(node)
                add_ids.append(node.id)

        # If --be-disks is explicitly provided for an existing cluster,
        # temporarily override cluster.be_disks so newly added BEs use
        # the specified disk config instead of the original cluster config.
        saved_be_disks = cluster.be_disks
        if args.be_disks is not None:
            cluster.be_disks = args.be_disks

        for node_type, add_num, add_ids in add_type_nums:
            do_add_node(node_type, add_num, add_ids)

        # Restore original be_disks to avoid side effects
        cluster.be_disks = saved_be_disks

        if args.IMAGE:
            for node in related_nodes:
                node.set_image(args.IMAGE)
            if for_all:
                cluster.set_image(args.IMAGE)

        fdb_image = args.fdb_image or "foundationdb/foundationdb:{}".format(
            args.fdb_version)
        for node in cluster.get_all_nodes(CLUSTER.Node.TYPE_FDB):
            node.set_image(fdb_image)

        cluster.save()

        options = []
        if not args.start:
            options.append("--no-start")
        else:
            options += ["--remove-orphans"]
            if args.detach:
                options.append("-d")
            if args.force_recreate:
                options.append("--force-recreate")

        related_node_num = len(related_nodes)
        if for_all:
            related_node_num = cluster.get_all_nodes_num()
            related_nodes = None

        output_real_time = args.start and not args.detach
        utils.exec_docker_compose_command(cluster.get_compose_file(),
                                          "up",
                                          options,
                                          related_nodes,
                                          output_real_time=output_real_time)

        if not args.start:
            LOG.info(
                utils.render_green(
                    "Not up cluster cause specific --no-start, related node num {}"
                    .format(related_node_num)))
        else:
            LOG.info("Using SQL mode for node management ? {}".format(
                cluster.sql_mode_node_mgr))

            if cluster.remote_master_fe:
                if is_new_cluster:
                    with open(CLUSTER.get_master_fe_addr_path(cluster.name),
                              "w") as f:
                        f.write(cluster.remote_master_fe)
                    if cluster.is_cloud:
                        cloud_config = "\n".join([
                            f"meta_service_endpoint = {cluster.get_meta_server_addr()}",
                            "deploy_mode = cloud",
                            f"cluster_id = {CLUSTER.CLUSTER_ID}",
                        ])
                        # write add conf to remote_master_fe_add.conf, remote fe can send ssh to get this content.
                        with open(
                                os.path.join(
                                    CLUSTER.get_status_path(cluster.name),
                                    "remote_master_fe_add.conf"), "w") as f:
                            f.write(cloud_config)
                        ans = input(
                            utils.render_red(
                                f"\nAdd remote fe {cluster.remote_master_fe} fe.conf with follow config: "
                            ) + "\n\n" + f"{cloud_config}\n\nConfirm ?  y/n: ")
                        if ans != 'y':
                            LOG.info(
                                "Up cluster failed due to not confirm write the above config."
                            )
                            return

                        LOG.info("Waiting connect to remote FE...")
                        expire_ts = time.time() + 3600 * 5
                        parts = cluster.remote_master_fe.split(":")
                        fe_ip = parts[0]
                        fe_port = int(parts[1])
                        ready = False
                        while expire_ts > time.time():
                            if utils.is_socket_avail(fe_ip, fe_port):
                                ready = True
                                break
                        if not ready:
                            raise Exception(
                                "Cannot connect to remote master fe: " +
                                cluster.remote_master_fe)

                        LOG.info("After connect to remote FE...")
            else:
                # Wait for FE master to be elected
                LOG.info("Waiting for FE master to be elected...")
                expire_ts = time.time() + 30
                while expire_ts > time.time():
                    ready = False
                    db_mgr = database.get_db_mgr(
                        args.NAME, cluster.get_all_node_net_infos(), False)
                    for id in add_fe_ids:
                        fe_state = db_mgr.get_fe(id)
                        if fe_state is not None and fe_state.alive:
                            ready = True
                            break
                    if ready:
                        break
                    LOG.info("there is no fe ready")
                    time.sleep(1)
                LOG.info("after Waiting for FE master to be elected...")
            if cluster.is_cloud and cluster.sql_mode_node_mgr:
                db_mgr = database.get_db_mgr(args.NAME,
                                             cluster.get_all_node_net_infos(),
                                             False)
                master_fe_endpoint = CLUSTER.get_master_fe_endpoint(
                    cluster.name, True)
                # Add FEs except master_fe
                for fe in cluster.get_all_nodes(CLUSTER.Node.TYPE_FE):
                    fe_querypoint = f"{fe.get_ip()}:{fe.meta['ports']['query_port']}"
                    fe_endpoint = f"{fe.get_ip()}:{fe.meta['ports']['edit_log_port']}"
                    if fe_querypoint != master_fe_endpoint:
                        try:
                            db_mgr.add_fe(
                                fe_endpoint, "FOLLOWER"
                                if cluster.fe_follower else "OBSERVER")
                            LOG.info(f"Added FE {fe_endpoint} successfully.")
                        except Exception as e:
                            LOG.error(
                                f"Failed to add FE {fe_endpoint}: {str(e)}")

                # Add BEs
                for be in cluster.get_all_nodes(CLUSTER.Node.TYPE_BE):
                    be_endpoint = f"{be.get_ip()}:{be.meta['ports']['heartbeat_service_port']}"
                    try:
                        db_mgr.add_be(be_endpoint)
                        LOG.info(f"Added BE {be_endpoint} successfully.")
                    except Exception as e:
                        LOG.error(f"Failed to add BE {be_endpoint}: {str(e)}")
                if is_new_cluster:
                    db_mgr.create_default_storage_vault(cluster.cloud_store_config)

            if not cluster.is_host_network():
                wait_service(True, args.wait_timeout, cluster, add_fe_ids,
                             add_be_ids)
            LOG.info(
                utils.render_green(
                    "Up cluster {} succ, related node num {}".format(
                        args.NAME, related_node_num)))

        ls_cmd = "python docker/runtime/doris-compose/doris-compose.py ls " + cluster.name
        LOG.info("Inspect command: " + utils.render_green(ls_cmd) + "\n")
        LOG.info(
            "Master fe query address: " +
            utils.render_green(CLUSTER.get_master_fe_endpoint(cluster.name)) +
            "\n")
        return {
            "fe": {
                "add_list": add_fe_ids,
            },
            "be": {
                "add_list": add_be_ids,
            },
            "ms": {
                "add_list": add_ms_ids,
            },
            "recycle": {
                "add_list": add_recycle_ids,
            },
            "fdb": {
                "add_list": add_fdb_ids,
            },
        }

    def _validate_external_ms_cluster(self, external_ms_cluster_name):
        # 1. Is the external cluster exist?
        try:
            external_cluster = CLUSTER.Cluster.load(external_ms_cluster_name)
        except Exception as e:
            raise Exception(
                f"External MS cluster '{external_ms_cluster_name}' not found. "
                f"Please create it first with: "
                f"python doris-compose.py up {external_ms_cluster_name} <image> --cloud --add-fe-num 0 --add-be-num 0"
            ) from e

        # 2. Is the external cluster a cloud cluster?
        if not external_cluster.is_cloud:
            raise Exception(
                f"External MS cluster '{external_ms_cluster_name}' is not a cloud cluster. "
                f"Only cloud clusters can be used as external MS."
            )

        # 3. Does the external cluster have MS and FDB nodes?
        ms_group = external_cluster.get_group(CLUSTER.Node.TYPE_MS)
        fdb_group = external_cluster.get_group(CLUSTER.Node.TYPE_FDB)

        if ms_group.get_node_num() == 0:
            raise Exception(
                f"External MS cluster '{external_ms_cluster_name}' has no MS nodes. "
                f"Please add MS nodes first."
            )

        if fdb_group.get_node_num() == 0:
            raise Exception(
                f"External MS cluster '{external_ms_cluster_name}' has no FDB nodes. "
                f"Please add FDB nodes first."
            )

        # 4. Are the MS and FDB containers running?
        containers = utils.get_doris_running_containers(external_ms_cluster_name)

        ms_running = False
        fdb_running = False
        for container_name in containers.keys():
            _, node_type, _ = utils.parse_service_name(container_name)
            if node_type == CLUSTER.Node.TYPE_MS:
                ms_running = True
            elif node_type == CLUSTER.Node.TYPE_FDB:
                fdb_running = True

        if not ms_running:
            raise Exception(
                f"External MS cluster '{external_ms_cluster_name}' MS node is not running. "
                f"Please start it with: python doris-compose.py start {external_ms_cluster_name}"
            )

        if not fdb_running:
            raise Exception(
                f"External MS cluster '{external_ms_cluster_name}' FDB node is not running. "
                f"Please start it with: python doris-compose.py start {external_ms_cluster_name}"
            )

        LOG.info(utils.render_green(
            f"✓ External MS cluster '{external_ms_cluster_name}' validation passed: "
            f"MS={external_cluster.get_meta_server_addr()}, "
            f"FDB={external_cluster.get_fdb_cluster()}"
        ))

    def _get_cloud_store_config(self):
        example_cfg_file = os.path.join(CLUSTER.LOCAL_RESOURCE_PATH,
                                        "cloud.ini.example")
        if not CLUSTER.CLOUD_CFG_FILE:
            raise Exception("Cloud cluster need S3 store, specific its config in a file.\n"     \
                "A example file is " + example_cfg_file + ".\n"   \
                "Then setting the env variable  `export DORIS_CLOUD_CFG_FILE=<cfg-file-path>`.")

        if not os.path.exists(CLUSTER.CLOUD_CFG_FILE):
            raise Exception("Cloud store config file '" +
                            CLUSTER.CLOUD_CFG_FILE + "' not exists.")

        config = {}
        with open(example_cfg_file, "r") as f:
            for line in f.readlines():
                if line.startswith('#'):
                    continue
                pos = line.find('=')
                if pos <= 0:
                    continue
                key = line[0:pos].strip()
                if key:
                    config[key] = ""

        with open(CLUSTER.CLOUD_CFG_FILE, "r") as f:
            for line in f.readlines():
                if line.startswith('#'):
                    continue
                pos = line.find('=')
                if pos <= 0:
                    continue
                key = line[0:pos].strip()
                if key and config.get(key, None) != None:
                    config[key] = line[line.find('=') + 1:].strip()

        for key, value in config.items():
            if not value:
                raise Exception(
                    "Should provide none empty property '{}' in file {}".
                    format(key, CLUSTER.CLOUD_CFG_FILE))
        return config

    @staticmethod
    def _merge_cloud_store_config(base_config, overrides):
        if not overrides:
            return base_config

        merged = dict(base_config)
        for item in overrides:
            pos = item.find('=')
            if pos <= 0:
                raise Exception(
                    "cloud config override '{}' error format, should be like 'name=value'".
                    format(item)
                )
            key = item[:pos].strip()
            value = item[pos + 1:].strip()
            if not key or not value:
                raise Exception(
                    "cloud config override '{}' error format, should be like 'name=value'".
                    format(item)
                )
            if key not in merged:
                raise Exception(
                    "Unknown cloud config override '{}', available keys: {}".
                    format(key, ", ".join(sorted(merged.keys())))
                )
            merged[key] = value
        return merged


class DownCommand(Command):

    def add_parser(self, args_parsers):
        parser = args_parsers.add_parser("down",
                                     help="Down doris containers, networks. "\
                                           "It will also remove node from DB. " \
                                           "If none of --fe-id, --be-id, --ms-id, --recycle-id, --fdb-id is specific, "\
                                           "then apply to all containers.")
        parser.add_argument("NAME", help="Specify cluster name")
        self._add_parser_ids_args(parser)
        self._add_parser_common_args(parser)
        parser.add_argument(
            "--clean",
            default=False,
            action=self._get_parser_bool_action(True),
            help=
            "Clean container related files, include expose data, config and logs."
        )
        parser.add_argument(
            "--drop-force",
            default=None,
            action=self._get_parser_bool_action(True),
            help="Drop doris node force. For be, if specific --drop-force, "\
                    "it will send dropp to fe, otherwise send decommission to fe.")

    def run(self, args):
        cluster_name = args.NAME
        cluster = None
        stop_grace = False

        try:
            cluster = CLUSTER.Cluster.load(cluster_name)
            for_all, related_nodes, related_node_num = get_ids_related_nodes(
                cluster,
                args.fe_id,
                args.be_id,
                args.ms_id,
                args.recycle_id,
                args.fdb_id,
                ignore_not_exists=True)
            stop_grace = cluster.coverage_dir
        except Exception as e:
            for_all = not args.fe_id and not args.be_id and not args.ms_id and not args.recycle_id
            related_nodes = []
            related_node_num = 0
            if not for_all:
                raise e

        LOG.info("down cluster " + args.NAME + " for all " +
                 str(for_all).lower())

        if for_all:
            compose_file = CLUSTER.get_compose_file(cluster_name)
            if os.path.exists(compose_file):
                try:
                    options = ["-v", "--remove-orphans"]
                    if not stop_grace:
                        options.extend(["-t", "1"])
                    utils.exec_docker_compose_command(compose_file,
                                                      "down",
                                                      options=options)
                except Exception as e:
                    LOG.warning("down cluster has exception: " + str(e))
            try:
                utils.remove_docker_network(cluster_name)
            except Exception as e:
                LOG.warning("remove network has exception: " + str(e))
            if args.clean:
                cluster_path = CLUSTER.get_cluster_path(cluster_name)
                if os.path.exists(cluster_path):
                    utils.enable_dir_with_rw_perm(cluster_path)
                    shutil.rmtree(cluster_path)
                LOG.info(
                    utils.render_yellow(
                        "Clean cluster data cause has specific --clean"))
        else:
            db_mgr = database.get_db_mgr(cluster.name,
                                         cluster.get_all_node_net_infos())

            for node in related_nodes:
                if node.is_fe():
                    fe_endpoint = "{}:{}".format(
                        node.get_ip(), node.meta["ports"]["edit_log_port"])
                    db_mgr.drop_fe(fe_endpoint)
                elif node.is_be():
                    be_endpoint = "{}:{}".format(
                        node.get_ip(),
                        node.meta["ports"]["heartbeat_service_port"])
                    if args.drop_force:
                        db_mgr.drop_be(be_endpoint)
                    else:
                        db_mgr.decommission_be(be_endpoint)
                else:
                    raise Exception("Unknown node type: {}".format(
                        node.node_type()))

                #utils.exec_docker_compose_command(cluster.get_compose_file(),
                #                                  "stop",
                #                                  nodes=[node])
                utils.exec_docker_compose_command(cluster.get_compose_file(),
                                                  "rm", ["-s", "-v", "-f"],
                                                  nodes=[node])
                if args.clean:
                    utils.enable_dir_with_rw_perm(node.get_path())
                    shutil.rmtree(node.get_path())
                    register_file = "{}/{}-{}-register".format(
                        CLUSTER.get_status_path(cluster.name),
                        node.node_type(), node.id)
                    if os.path.exists(register_file):
                        os.remove(register_file)
                    LOG.info(
                        utils.render_yellow(
                            "Clean {} with id {} data cause has specific --clean"
                            .format(node.node_type(), node.id)))

                cluster.remove(node.node_type(), node.id)
                cluster.save()

        LOG.info(
            utils.render_green(
                "Down cluster {} succ, related node num {}".format(
                    cluster_name, related_node_num)))

        return "down cluster succ"


class ListNode(object):

    def __init__(self):
        self.node_type = ""
        self.id = 0
        self.backend_id = ""
        self.cluster_name = ""
        self.ip = ""
        self.status = ""
        self.container_id = ""
        self.image = ""
        self.created = ""
        self.alive = ""
        self.is_master = ""
        self.tablet_num = ""
        self.last_heartbeat = ""
        self.err_msg = ""
        self.query_port = ""
        self.http_port = ""
        self.heartbeat_port = ""
        self.edit_log_port = ""
        self.heartbeat_port = ""

    def info(self, detail):
        result = [
            self.cluster_name, "{}-{}".format(self.node_type, self.id),
            self.ip, self.status, self.container_id, self.image, self.created,
            self.alive, self.is_master, self.backend_id, self.tablet_num,
            self.last_heartbeat, self.err_msg
        ]
        if detail:
            node_path = CLUSTER.get_node_path(self.cluster_name,
                                              self.node_type, self.id)
            result += [
                self.query_port, self.http_port, node_path, self.edit_log_port,
                self.heartbeat_port
            ]
        return result

    def update_db_info(self, cluster, db_mgr):
        try:
            node = cluster.get_node(self.node_type, self.id, True)
        except:
            node = None
        ports = node.meta["ports"] if node else {}
        if self.node_type == CLUSTER.Node.TYPE_FE:
            fe = db_mgr.get_fe(self.id)
            if fe:
                self.alive = str(fe.alive).lower()
                self.is_master = str(fe.is_master).lower()
                self.last_heartbeat = fe.last_heartbeat
                self.err_msg = fe.err_msg
                self.query_port = fe.query_port
                self.http_port = fe.http_port
                self.edit_log_port = fe.edit_log_port
        elif self.node_type == CLUSTER.Node.TYPE_BE:
            self.backend_id = -1
            be = db_mgr.get_be(self.id)
            if be:
                self.alive = str(be.alive).lower()
                self.backend_id = be.backend_id
                self.tablet_num = be.tablet_num
                self.last_heartbeat = be.last_heartbeat
                self.err_msg = be.err_msg
                self.http_port = be.http_port
                self.heartbeat_port = be.heartbeat_service_port
        elif self.node_type == CLUSTER.Node.TYPE_MS or self.node_type == CLUSTER.Node.TYPE_RECYCLE:
            if ports:
                self.http_port = ports.get("brpc_listen_port", -1)
        if node and node.meta.get("is_remote", False):
            self.ip = node.get_ip()
            self.container_id = "<remote>"
            self.image = "<remote>"


class GenConfCommand(Command):

    def print_use_time(self):
        return False

    def add_parser(self, args_parsers):
        parser = args_parsers.add_parser(
            "config",
            help="Generate regression-conf-custom.groovy for regression test.")
        parser.add_argument("NAME", default="", help="Specific cluster name.")
        parser.add_argument("DORIS_ROOT_PATH", default="", help="Specify doris or selectdb root path, "\
                "i.e. the parent directory of regression-test.")
        parser.add_argument("--connect-follow-fe",
                            default=False,
                            action=self._get_parser_bool_action(True),
                            help="Connect to follow fe.")
        parser.add_argument("-q",
                            "--quiet",
                            default=False,
                            action=self._get_parser_bool_action(True),
                            help="write config quiet, no need confirm.")

        return parser

    def run(self, args):
        base_conf = '''
jdbcUrl = "jdbc:mysql://{fe_ip}:{query_port}/?useLocalSessionState=true&allowLoadLocalInfile=true"
targetJdbcUrl = "jdbc:mysql://{fe_ip}:{query_port}/?useLocalSessionState=true&allowLoadLocalInfile=true"
feSourceThriftAddress = "{fe_ip}:{rpc_port}"
feTargetThriftAddress = "{fe_ip}:{rpc_port}"
syncerAddress = "{fe_ip}:9190"
feHttpAddress = "{fe_ip}:{http_port}"
'''

        cloud_conf = '''
feCloudHttpAddress = "{fe_ip}:18030"
metaServiceHttpAddress = "{ms_endpoint}"
metaServiceToken = "greedisgood9999"
recycleServiceHttpAddress = "{recycle_endpoint}"
instanceId = "12345678"
multiClusterInstance = "12345678"
multiClusterBes = "{multi_cluster_bes}"
cloudUniqueId= "{fe_cloud_unique_id}"
'''
        cluster = CLUSTER.Cluster.load(args.NAME)
        master_fe_ip_ep = CLUSTER.get_master_fe_endpoint(args.NAME)
        if not master_fe_ip_ep:
            print("Not found cluster with name {} in directory {}".format(
                args.NAME, CLUSTER.LOCAL_DORIS_PATH))
            return

        db_mgr = database.get_db_mgr(args.NAME,
                                     cluster.get_all_node_net_infos(), False)
        fe_ip = ""
        rpc_port = 0
        query_port = 0
        http_port = 0
        for fe in db_mgr.get_all_fe().values():
            if not fe.alive:
                continue
            if fe.is_master == (not args.connect_follow_fe):
                fe_ip = fe.ip
                rpc_port = fe.rpc_port
                query_port = fe.query_port
                http_port = fe.http_port
                break
        if not fe_ip:
            role = "follower" if args.connect_follow_fe else "master"
            raise Exception(f"Not found an alive {role} fe")

        relative_custom_file_path = "regression-test/conf/regression-conf-custom.groovy"
        regression_conf_custom = os.path.join(args.DORIS_ROOT_PATH,
                                              relative_custom_file_path)
        if not args.quiet:
            ans = input(
                "\nwrite file {} ?  y/n: ".format(regression_conf_custom))
            if ans != 'y':
                print("\nNo write regression custom file.")
                return

        annotation_start = "//---------- Start auto generate by doris-compose.py---------"
        annotation_end = "//---------- End auto generate by doris-compose.py---------"

        old_contents = []
        if os.path.exists(regression_conf_custom):
            with open(regression_conf_custom, "r") as f:
                old_contents = f.readlines()
        with open(regression_conf_custom, "w") as f:
            # write auto gen config
            f.write(annotation_start)
            f.write(
                base_conf.format(fe_ip=fe_ip,
                                 rpc_port=rpc_port,
                                 query_port=query_port,
                                 http_port=http_port))
            if cluster.is_cloud:
                multi_cluster_bes = ",".join([
                    "{}:{}:{}:{}:{}".format(
                        be.get_ip(),
                        be.meta["ports"]["heartbeat_service_port"],
                        be.meta["ports"]["webserver_port"],
                        be.cloud_unique_id(), be.meta["ports"]["brpc_port"])
                    for be in cluster.get_all_nodes(CLUSTER.Node.TYPE_BE)
                ])
                master_fe = cluster.get_remote_fe_node(
                ) if cluster.is_host_network() else cluster.get_node(
                    CLUSTER.Node.TYPE_FE, 1)
                f.write(
                    cloud_conf.format(
                        fe_ip=fe_ip,
                        ms_endpoint=cluster.get_meta_server_addr(),
                        recycle_endpoint=cluster.get_recycle_addr(),
                        multi_cluster_bes=multi_cluster_bes,
                        fe_cloud_unique_id=master_fe.cloud_unique_id()))
            f.write(annotation_end + "\n\n")

            # write not-auto gen config
            in_annotation = False
            annotation_end_line_idx = -100
            for line_idx, line in enumerate(old_contents):
                line = line.rstrip()
                if line == annotation_start:
                    in_annotation = True
                elif line == annotation_end:
                    in_annotation = False
                    annotation_end_line_idx = line_idx
                elif not in_annotation:
                    if line or line_idx != annotation_end_line_idx + 1:
                        f.write(line + "\n")

        print("\nWrite succ: " + regression_conf_custom)


class ListCommand(Command):

    def add_parser(self, args_parsers):
        parser = args_parsers.add_parser(
            "ls", help="List running doris compose clusters.")
        parser.add_argument(
            "NAME",
            nargs="*",
            help=
            "Specify multiple clusters, if specific, show all their containers."
        )
        self._add_parser_common_args(parser)
        parser.add_argument("--detail",
                            default=False,
                            action=self._get_parser_bool_action(True),
                            help="Print more detail fields.")

    def _hint_cluster_bad(self, cluster_name):
        cluster_path = CLUSTER.get_cluster_path(cluster_name)
        if not os.path.exists(cluster_path):
            LOG.info(
                utils.render_yellow(
                    f"Not exits cluster directory in '{CLUSTER.LOCAL_DORIS_PATH}'"
                ))
        elif not os.path.exists(CLUSTER.Cluster._get_meta_file(cluster_name)):
            LOG.error(
                utils.render_red(
                    f"Not exits cluster meta file in '{cluster_path}'"))
        else:
            try:
                CLUSTER.Cluster.load(cluster_name)
            except:
                LOG.error(utils.render_red("meta file is bad or incompatible with current doris-compose.py. " \
                    "Run command `down --clean` to destroy it then recreate a new one"))

    def run(self, args):
        COMPOSE_MISSING = "(missing)"
        COMPOSE_BAD = "(bad)"
        COMPOSE_GOOD = ""

        SERVICE_DEAD = "dead"
        SERVICE_RUNNING = "running"

        class ComposeService(object):

            def __init__(self, name, ip, image):
                self.name = name
                self.ip = ip
                self.image = image

        def parse_cluster_compose_file(cluster_name):
            compose_file = CLUSTER.get_compose_file(cluster_name)
            try:
                cluster = CLUSTER.Cluster.load(cluster_name)
                ip_for_host_mode = cluster.local_network_ip if cluster.is_host_network(
                ) else ""
            except:
                ip_for_host_mode = ""
            if not os.path.exists(compose_file):
                return COMPOSE_MISSING, {}
            try:
                compose = utils.read_compose_file(compose_file)
                if not compose:
                    return COMPOSE_BAD, {}
                services = compose.get("services", {})
                if services is None:
                    return COMPOSE_BAD, {}
                return COMPOSE_GOOD, {
                    service: ComposeService(
                        service, ip_for_host_mode if ip_for_host_mode else
                        list(service_conf["networks"].values())[0]
                        ["ipv4_address"], service_conf["image"])
                    for service, service_conf in services.items()
                }
            except:
                return COMPOSE_BAD, {}

        clusters = {}
        search_names = []
        if args.NAME:
            search_names = args.NAME
        else:
            search_names = CLUSTER.get_all_cluster_names()

        for cluster_name in search_names:
            status, services = parse_cluster_compose_file(cluster_name)
            clusters[cluster_name] = {"status": status, "services": services}

        docker_clusters = utils.get_doris_containers(args.NAME)
        for cluster_name, containers in docker_clusters.items():
            cluster_info = clusters.get(cluster_name, None)
            if not cluster_info:
                cluster_info = {"status": COMPOSE_MISSING, "services": {}}
                clusters[cluster_name] = cluster_info
            for container in containers:
                #if container.status == "running" and cluster_info[
                #        "status"] == COMPOSE_GOOD and (
                #            container.name not in cluster_info["services"]):
                #    container.status = "orphans"
                cluster_info["services"][container.name] = container

        TYPE_COMPOSESERVICE = type(ComposeService("", "", ""))
        if not args.NAME:
            header = ("CLUSTER", "OWNER", "STATUS", "MASTER FE", "CLOUD",
                      "NETWORK MODE", "CONFIG FILES")
            rows = []
            for name in sorted(clusters.keys()):
                cluster_info = clusters[name]
                service_statuses = {}
                for _, container in cluster_info["services"].items():
                    status = SERVICE_DEAD if type(
                        container) == TYPE_COMPOSESERVICE else container.status
                    service_statuses[status] = service_statuses.get(status,
                                                                    0) + 1
                show_status = ",".join([
                    "{}({})".format(status, count)
                    for status, count in service_statuses.items()
                ])
                owner = utils.get_path_owner(CLUSTER.get_cluster_path(name))
                compose_file = CLUSTER.get_compose_file(name)

                is_cloud = ""
                network_mode = ""
                try:
                    cluster = CLUSTER.Cluster.load(name)
                    is_cloud = "true" if cluster.is_cloud else "false"
                    network_mode = "host" if cluster.is_host_network(
                    ) else "bridge"
                except:
                    pass

                rows.append(
                    (name, owner, show_status,
                     CLUSTER.get_master_fe_endpoint(name), is_cloud,
                     network_mode, "{}{}".format(compose_file,
                                                 cluster_info["status"])))
            return self._print_table(header, rows)

        header = [
            "CLUSTER", "NAME", "IP", "STATUS", "CONTAINER ID", "IMAGE",
            "CREATED", "alive", "is_master", "backend_id", "tablet_num",
            "last_heartbeat", "err_msg"
        ]
        if args.detail:
            header += [
                "query_port",
                "http_port",
                "path",
                "edit_log_port",
                "heartbeat_port",
            ]

        rows = []
        for cluster_name in sorted(clusters.keys()):
            fe_ids = {}
            be_ids = {}
            services = clusters[cluster_name]["services"]
            cluster = None
            try:
                cluster = CLUSTER.Cluster.load(cluster_name)
            except:
                pass

            db_mgr = database.get_db_mgr(
                cluster_name,
                cluster.get_all_node_net_infos() if cluster else [], False)
            nodes = []
            for service_name, container in services.items():
                _, node_type, id = utils.parse_service_name(container.name)
                node = ListNode()
                node.cluster_name = cluster_name
                node.node_type = node_type
                node.id = id
                node.update_db_info(cluster, db_mgr)
                nodes.append(node)

                if node_type == CLUSTER.Node.TYPE_FE:
                    fe_ids[id] = True
                elif node_type == CLUSTER.Node.TYPE_BE:
                    be_ids[id] = True

                if type(container) == TYPE_COMPOSESERVICE:
                    node.ip = container.ip
                    node.image = container.image
                    node.status = SERVICE_DEAD
                else:
                    node.created = dateutil.parser.parse(
                        container.attrs.get("Created")).astimezone().strftime(
                            "%Y-%m-%d %H:%M:%S")
                    if cluster and cluster.is_host_network():
                        node.ip = cluster.local_network_ip
                    else:
                        network_name = utils.get_network_name(cluster.name)
                        node.ip = container.attrs["NetworkSettings"]["Networks"][network_name] \
                                ["IPAMConfig"]["IPv4Address"]
                    node.image = container.attrs["Config"]["Image"]
                    if not node.image:
                        node.image = ",".join(container.image.tags)
                    node.container_id = container.short_id
                    node.status = container.status
                    if node.container_id and \
                        node_type in (CLUSTER.Node.TYPE_FDB,
                                     CLUSTER.Node.TYPE_MS,
                                     CLUSTER.Node.TYPE_RECYCLE):
                        node.alive = "true"

            for id, fe in db_mgr.fe_states.items():
                if fe_ids.get(id, False):
                    continue
                node = ListNode()
                node.cluster_name = cluster_name
                node.node_type = CLUSTER.Node.TYPE_FE
                node.id = id
                node.status = SERVICE_RUNNING if fe.alive else SERVICE_DEAD
                node.update_db_info(cluster, db_mgr)
                nodes.append(node)
            for id, be in db_mgr.be_states.items():
                if be_ids.get(id, False):
                    continue
                node = ListNode()
                node.cluster_name = cluster_name
                node.node_type = CLUSTER.Node.TYPE_BE
                node.id = id
                node.status = SERVICE_RUNNING if be.alive else SERVICE_DEAD
                node.update_db_info(cluster, db_mgr)
                nodes.append(node)

            def get_node_seq(node):
                return CLUSTER.get_node_seq(node.node_type, node.id)

            for node in sorted(nodes, key=get_node_seq):
                rows.append(node.info(args.detail))

        ret = self._print_table(header, rows)
        if len(args.NAME) == 1 and len(rows) == 0:
            self._hint_cluster_bad(args.NAME[0])

        return ret


class InfoCommand(Command):

    def add_parser(self, args_parsers):
        parser = args_parsers.add_parser(
            "info", help="Show info like cloud.ini, port, path, etc")
        self._add_parser_common_args(parser)

    def run(self, args):

        header = ["key", "value", "scope"]
        cloud_cfg_file_env = os.getenv("DORIS_CLOUD_CFG_FILE")
        cloud_cfg_file = cloud_cfg_file_env if cloud_cfg_file_env else "${LOCAL_DORIS_PATH}/cloud.ini"
        rows = [
            ("LOCAL_DORIS_PATH", CLUSTER.LOCAL_DORIS_PATH, "env variable"),
            ("DORIS_CLOUD_CFG_FILE", cloud_cfg_file, "env variable"),
            ("FE_QUERY_PORT", CLUSTER.FE_QUERY_PORT, "constant"),
            ("FE_HTTP_PORT", CLUSTER.FE_HTTP_PORT, "constant"),
            ("FE_EDITLOG_PORT", CLUSTER.FE_EDITLOG_PORT, "constant"),
            ("FE_JAVA_DBG_PORT", CLUSTER.FE_JAVA_DBG_PORT, "constant"),
            ("BE_HEARTBEAT_PORT", CLUSTER.BE_HEARTBEAT_PORT, "constant"),
            ("BE_WEBSVR_PORT", CLUSTER.BE_WEBSVR_PORT, "constant"),
            ("MS_PORT", CLUSTER.MS_PORT, "constant"),
            ("RECYCLER_PORT", CLUSTER.MS_PORT, "constant"),
        ]

        if os.path.exists(CLUSTER.CLOUD_CFG_FILE):
            with open(CLUSTER.CLOUD_CFG_FILE, "r") as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith("#"):
                        key, value = line.split("=", 1)
                        rows.append((key.strip(), value.strip(), "cloud.ini"))

        return self._print_table(header, rows)


class AddRWPermCommand(Command):

    def add_parser(self, args_parsers):
        parser = args_parsers.add_parser(
            "add-rw-perm",
            help="Add read and write permissions to the cluster files")
        parser.add_argument("NAME", help="Specify cluster name.")
        self._add_parser_common_args(parser)

    def run(self, args):
        utils.enable_dir_with_rw_perm(CLUSTER.get_cluster_path(args.NAME))
        return ""


class RollbackCommand(Command):

    def add_parser(self, args_parsers):
        parser = args_parsers.add_parser(
            "rollback",
            help="Rollback cloud cluster to a snapshot. " \
                 "Stop ALL FE/BE, clean metadata/data, and restart with new cloud_unique_id and instance_id. " \
                 "Only available for cloud mode clusters."
        )
        parser.add_argument("NAME", help="Specify cluster name.")
        parser.add_argument(
            "--cluster-snapshot",
            type=str,
            required=True,
            help="Cluster snapshot JSON content for rollback. " \
                 "Example: '{\"instance_id\":\"instance_id_xxx\"}'"
        )
        parser.add_argument(
            "--instance-id",
            type=str,
            help="New instance ID for the cluster after rollback. " \
                 "If not specified, will generate a new one based on timestamp."
        )
        parser.add_argument(
            "--wait-timeout",
            type=int,
            default=0,
            help="Specify wait seconds for fe/be ready for service: 0 not wait (default), " \
                 "> 0 max wait seconds, -1 wait unlimited."
        )
        self._add_parser_common_args(parser)
        return parser

    def _update_config_cloud_unique_id(self, conf_path, new_cloud_unique_id):
        """Update cloud_unique_id in fe.conf or be.conf file.

        This method updates the cloud_unique_id value in the config file.
        If the line exists, it will be replaced; if not found, it will be added.
        """
        updated = False
        lines = []

        # Read existing config
        with open(conf_path, "r") as f:
            lines = f.readlines()

        # Try to update existing cloud_unique_id line
        for i, line in enumerate(lines):
            stripped = line.strip()
            # Match lines like: cloud_unique_id = xxx or cloud_unique_id=xxx
            if stripped.startswith("cloud_unique_id") and ("=" in stripped):
                lines[i] = f"cloud_unique_id = {new_cloud_unique_id}\n"
                updated = True
                break

        # If not found, add it to the doris-compose section
        if not updated:
            raise Exception("cloud_unique_id not found in config file: " + conf_path)

        # Write back to file
        with open(conf_path, "w") as f:
            f.writelines(lines)

    def run(self, args):
        cluster = CLUSTER.Cluster.load(args.NAME)

        # Validate: only cloud clusters support rollback
        if not cluster.is_cloud:
            raise Exception("Rollback is only supported for cloud clusters")

        # Rollback must include ALL FE/BE nodes
        fe_nodes = cluster.get_all_nodes(CLUSTER.Node.TYPE_FE)
        be_nodes = cluster.get_all_nodes(CLUSTER.Node.TYPE_BE)

        if not fe_nodes and not be_nodes:
            raise Exception("No FE or BE nodes to rollback")

        # Generate new instance_id and rollback timestamp
        rollback_ts = str(int(time.time()))
        new_instance_id = args.instance_id or f"instance_{cluster.name}_{rollback_ts}"

        LOG.info(f"Starting rollback with instance_id: {new_instance_id}, timestamp: {rollback_ts}")

        # Step 1: Stop FE/BE nodes
        LOG.info("Step 1/5: Stopping FE/BE nodes...")
        fe_ids = [node.id for node in fe_nodes]
        be_ids = [node.id for node in be_nodes]

        stop_nodes = fe_nodes + be_nodes
        utils.exec_docker_compose_command(
            cluster.get_compose_file(),
            "stop",
            options=["-t", "1"],
            nodes=stop_nodes
        )
        LOG.info(f"Stopped {len(fe_nodes)} FE and {len(be_nodes)} BE nodes")

        cluster.is_rollback = True  # Add ROLLBACK envs

        # Step 2: Clean metadata and data
        LOG.info("Step 2/5: Cleaning metadata and data directories...")

        for fe in fe_nodes:
            fe_meta_path = os.path.join(fe.get_path(), "doris-meta")
            if os.path.exists(fe_meta_path):
                utils.enable_dir_with_rw_perm(fe_meta_path)
                shutil.rmtree(fe_meta_path)
                os.makedirs(fe_meta_path, exist_ok=True)
                LOG.info(f"  Cleaned FE-{fe.id} doris-meta/")

        for be in be_nodes:
            be_storage_path = os.path.join(be.get_path(), "storage")
            if os.path.exists(be_storage_path):
                utils.enable_dir_with_rw_perm(be_storage_path)
                shutil.rmtree(be_storage_path)
                # Recreate storage directories based on disk configuration
                be.init_disk(cluster.be_disks)
                LOG.info(f"  Cleaned BE-{be.id} storage/")

        # Step 3: Update cluster configuration
        LOG.info("Step 3/5: Updating cluster configuration...")

        # Update instance_id
        old_instance_id = cluster.instance_id
        cluster.instance_id = new_instance_id
        LOG.info(f"  Updated instance_id: {old_instance_id} -> {new_instance_id}")

        # Update cluster_snapshot for FE-1
        cluster.cluster_snapshot = args.cluster_snapshot
        fe1 = cluster.get_node(CLUSTER.Node.TYPE_FE, 1)
        snapshot_file = os.path.join(fe1.get_path(), "conf", "cluster_snapshot.json")
        with open(snapshot_file, "w") as f:
            f.write(args.cluster_snapshot)
        LOG.info(f"  Written cluster_snapshot to {snapshot_file}")

        # Step 4: Update cloud_unique_id by setting rollback_timestamp in node meta
        LOG.info("Step 4/5: Generating new cloud_unique_id...")

        for fe in fe_nodes:
            fe.meta["rollback_timestamp"] = rollback_ts
            old_id = f"{cluster.name}_sql_server_{fe.id}"
            new_id = fe.cloud_unique_id()
            LOG.info(f"  FE-{fe.id}: {old_id} -> {new_id}")

            fe_conf_path = os.path.join(fe.get_path(), "conf", "fe.conf")
            if os.path.exists(fe_conf_path):
                self._update_config_cloud_unique_id(fe_conf_path, new_id)

        for be in be_nodes:
            be.meta["rollback_timestamp"] = rollback_ts
            old_id = f"{cluster.name}_compute_node_{be.id}"
            new_id = be.cloud_unique_id()
            LOG.info(f"  BE-{be.id}: {old_id} -> {new_id}")

            be_conf_path = os.path.join(be.get_path(), "conf", "be.conf")
            if os.path.exists(be_conf_path):
                self._update_config_cloud_unique_id(be_conf_path, new_id)

        # Save updated cluster configuration
        cluster.save()
        LOG.info("  Saved cluster configuration")

        # Step 5: Start FE/BE nodes
        LOG.info("Step 5/5: Starting FE/BE nodes with new configuration...")

        utils.exec_docker_compose_command(
            cluster.get_compose_file(),
            "up",
            options=["-d"],
            nodes=stop_nodes
        )

        # Wait for services to be ready
        if not cluster.is_host_network():
            wait_service(True, args.wait_timeout, cluster, fe_ids, be_ids)

        LOG.info("Rollback completed successfully.")

        return "Rollback completed successfully."


ALL_COMMANDS = [
    UpCommand("up"),
    DownCommand("down"),
    StartCommand("start"),
    StopCommand("stop"),
    RestartCommand("restart"),
    SimpleCommand("pause", "Pause the doris containers. "),
    SimpleCommand("unpause", "Unpause the doris containers. "),
    GenConfCommand("config"),
    InfoCommand("info"),
    ListCommand("ls"),
    AddRWPermCommand("add-rw-perm"),
    RollbackCommand("rollback"),
]