| # 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") | |
| parser.add_argument( | |
| "--tde-aws-ak", | |
| type=str, | |
| default="", | |
| help="tde aws ak") | |
| parser.add_argument( | |
| "--tde-aws-sk", | |
| type=str, | |
| default="", | |
| help="tde aws sk") | |
| parser.add_argument( | |
| "--tde-aliyun-ak", | |
| type=str, | |
| default="", | |
| help="tde aliyun ak") | |
| parser.add_argument( | |
| "--tde-aliyun-sk", | |
| type=str, | |
| default="", | |
| help="tde aliyun 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', '') | |
| env = getattr(args, 'env', None) | |
| enable_storage_vault = CLUSTER.is_true( | |
| CLUSTER.get_env_value(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, 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, | |
| args.tde_aws_ak, args.tde_aws_sk, args.tde_aliyun_ak, args.tde_aliyun_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"), | |
| ] |