| # 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 cluster as CLUSTER |
| import os.path |
| import pymysql |
| import time |
| import utils |
| import uuid |
| |
| LOG = utils.get_logger() |
| |
| |
| class FEState(object): |
| |
| def __init__(self, id, ip, is_master, alive, last_heartbeat, err_msg, role, |
| query_port, rpc_port, http_port, edit_log_port): |
| self.id = id |
| self.ip = ip |
| self.is_master = is_master |
| self.alive = alive |
| self.last_heartbeat = last_heartbeat |
| self.err_msg = err_msg |
| self.role = role |
| self.query_port = query_port |
| self.rpc_port = rpc_port |
| self.http_port = http_port |
| self.edit_log_port = edit_log_port |
| |
| |
| class BEState(object): |
| |
| def __init__(self, id, ip, backend_id, decommissioned, alive, tablet_num, |
| last_heartbeat, err_msg, http_port, heartbeat_service_port): |
| self.id = id |
| self.ip = ip |
| self.backend_id = backend_id |
| self.decommissioned = decommissioned |
| self.alive = alive |
| self.tablet_num = tablet_num |
| self.last_heartbeat = last_heartbeat |
| self.err_msg = err_msg |
| self.http_port = http_port |
| self.heartbeat_service_port = heartbeat_service_port |
| |
| |
| class DBManager(object): |
| |
| def __init__(self, all_node_net_infos): |
| self.all_node_net_infos = all_node_net_infos |
| self.fe_states = {} |
| self.be_states = {} |
| self.conn = None |
| self.fe_ip = "" |
| self.fe_port = -1 |
| |
| def get_all_fe(self): |
| return self.fe_states |
| |
| def get_fe(self, id): |
| return self.fe_states.get(id, None) |
| |
| def get_be(self, id): |
| return self.be_states.get(id, None) |
| |
| def load_states(self): |
| self._load_fe_states() |
| self._load_be_states() |
| |
| def add_fe(self, fe_endpoint, role): |
| try: |
| sql = f"ALTER SYSTEM ADD {role} '{fe_endpoint}'" |
| self._exec_query(sql) |
| LOG.info(f"Added {role} FE {fe_endpoint} via SQL successfully.") |
| except Exception as e: |
| LOG.error(f"Failed to add FE {fe_endpoint} via SQL: {str(e)}") |
| raise |
| |
| def drop_fe(self, fe_endpoint): |
| id = CLUSTER.Node.get_id_from_ip(fe_endpoint[:fe_endpoint.find(":")]) |
| try: |
| role = self.get_fe(id).role if self.get_fe(id) else "FOLLOWER" |
| self._exec_query("ALTER SYSTEM DROP {} '{}'".format( |
| role, fe_endpoint)) |
| LOG.info("Drop fe {} with id {} from db succ.".format( |
| fe_endpoint, id)) |
| except Exception as e: |
| if str(e).find("frontend does not exist") >= 0: |
| LOG.info( |
| "Drop fe {} with id {} from db succ cause it does not exist in db." |
| .format(fe_endpoint, id)) |
| return |
| raise e |
| |
| def add_be(self, be_endpoint): |
| try: |
| sql = f"ALTER SYSTEM ADD BACKEND '{be_endpoint}'" |
| self._exec_query(sql) |
| LOG.info(f"Added BE {be_endpoint} via SQL successfully.") |
| except Exception as e: |
| LOG.error(f"Failed to add BE {be_endpoint} via SQL: {str(e)}") |
| raise |
| |
| def drop_be(self, be_endpoint): |
| id = CLUSTER.Node.get_id_from_ip(be_endpoint[:be_endpoint.find(":")]) |
| try: |
| self._exec_query( |
| "ALTER SYSTEM DROPP BACKEND '{}'".format(be_endpoint)) |
| LOG.info("Drop be {} with id {} from db succ.".format( |
| be_endpoint, id)) |
| except Exception as e: |
| if str(e).find("backend does not exists") >= 0: |
| LOG.info( |
| "Drop be {} with id {} from db succ cause it does not exist in db." |
| .format(be_endpoint, id)) |
| return |
| raise e |
| |
| def decommission_be(self, be_endpoint): |
| old_tablet_num = 0 |
| id = CLUSTER.Node.get_id_from_ip(be_endpoint[:be_endpoint.find(":")]) |
| start_ts = time.time() |
| if id not in self.be_states: |
| self._load_be_states() |
| if id in self.be_states: |
| be = self.be_states[id] |
| old_tablet_num = be.tablet_num |
| if not be.alive: |
| raise Exception("Decommission be {} with id {} fail " \ |
| "cause it's not alive, maybe you should specific --drop-force " \ |
| " to dropp it from db".format(be_endpoint, id)) |
| try: |
| self._exec_query( |
| "ALTER SYSTEM DECOMMISSION BACKEND '{}'".format(be_endpoint)) |
| LOG.info("Mark be {} with id {} as decommissioned, start migrate its tablets, " \ |
| "wait migrating job finish.".format(be_endpoint, id)) |
| except Exception as e: |
| if str(e).find("Backend does not exist") >= 0: |
| LOG.info("Decommission be {} with id {} from db succ " \ |
| "cause it does not exist in db.".format(be_endpoint, id)) |
| return |
| raise e |
| |
| while True: |
| self._load_be_states() |
| be = self.be_states.get(id, None) |
| if not be: |
| LOG.info("Decommission be {} succ, total migrate {} tablets, " \ |
| "has drop it from db.".format(be_endpoint, old_tablet_num)) |
| return |
| LOG.info( |
| "Decommission be {} status: alive {}, decommissioned {}. " \ |
| "It is migrating its tablets, left {}/{} tablets. Time elapse {} s." |
| .format(be_endpoint, be.alive, be.decommissioned, be.tablet_num, old_tablet_num, |
| int(time.time() - start_ts))) |
| |
| time.sleep(1) |
| |
| def create_default_storage_vault(self, cloud_store_config): |
| try: |
| # Create storage vault |
| create_vault_sql = f""" |
| CREATE STORAGE VAULT IF NOT EXISTS default_vault |
| PROPERTIES ( |
| "type" = "S3", |
| "s3.access_key" = "{cloud_store_config['DORIS_CLOUD_AK']}", |
| "s3.secret_key" = "{cloud_store_config['DORIS_CLOUD_SK']}", |
| "s3.endpoint" = "{cloud_store_config['DORIS_CLOUD_ENDPOINT']}", |
| "s3.bucket" = "{cloud_store_config['DORIS_CLOUD_BUCKET']}", |
| "s3.region" = "{cloud_store_config['DORIS_CLOUD_REGION']}", |
| "s3.root.path" = "{str(uuid.uuid4())}", |
| "provider" = "{cloud_store_config['DORIS_CLOUD_PROVIDER']}", |
| "use_path_style" = "false" |
| ); |
| """ |
| # create hk storage vault from beijing cost 14s |
| self._reset_conn(read_timeout=20) |
| self._exec_query(create_vault_sql) |
| LOG.info("Created storage vault 'default_vault'") |
| |
| # Set as default storage vault |
| set_default_vault_sql = "SET default_vault as DEFAULT STORAGE VAULT;" |
| self._exec_query(set_default_vault_sql) |
| LOG.info("Set 'default_vault' as the default storage vault") |
| |
| except Exception as e: |
| LOG.error(f"Failed to create default storage vault: {str(e)}") |
| raise |
| |
| def _load_fe_states(self): |
| fe_states = {} |
| alive_master_fe_ip = None |
| alive_master_fe_port = None |
| for record in self._exec_query("show frontends"): |
| name = record["Name"] |
| ip = record["Host"] |
| role = record["Role"] |
| is_master = utils.is_true(record["IsMaster"]) |
| alive = utils.is_true(record["Alive"]) |
| query_port = int(record["QueryPort"]) |
| http_port = int(record["HttpPort"]) |
| rpc_port = int(record["RpcPort"]) |
| edit_log_port = int(record["EditLogPort"]) |
| id = None |
| for net_info in self.all_node_net_infos: |
| if net_info.ip == ip and net_info.ports.get("query_port", |
| -1) == query_port: |
| id = net_info.id |
| break |
| if not id: |
| id = CLUSTER.Node.get_id_from_ip(ip) |
| last_heartbeat = utils.escape_null(record["LastHeartbeat"]) |
| err_msg = record["ErrMsg"] |
| fe = FEState(id, ip, is_master, alive, last_heartbeat, err_msg, |
| role, query_port, rpc_port, http_port, edit_log_port) |
| fe_states[id] = fe |
| if is_master and alive: |
| alive_master_fe_ip = ip |
| alive_master_fe_port = query_port |
| LOG.debug( |
| "record of show frontends, name {}, ip {}, port {}, alive {}, is_master {}, role {}" |
| .format(name, ip, query_port, alive, is_master, role)) |
| |
| self.fe_states = fe_states |
| if alive_master_fe_ip and (alive_master_fe_ip != self.fe_ip |
| or alive_master_fe_port != self.fe_port): |
| self.fe_ip = alive_master_fe_ip |
| self.fe_port = alive_master_fe_port |
| self._reset_conn() |
| |
| def _load_be_states(self): |
| be_states = {} |
| for record in self._exec_query("show backends"): |
| backend_id = int(record["BackendId"]) |
| alive = utils.is_true(record["Alive"]) |
| decommissioned = utils.is_true(record["SystemDecommissioned"]) |
| tablet_num = int(record["TabletNum"]) |
| ip = record["Host"] |
| heartbeat_service_port = int(record["HeartbeatPort"]) |
| http_port = int(record["HttpPort"]) |
| id = None |
| for net_info in self.all_node_net_infos: |
| if net_info.ip == ip and net_info.ports.get( |
| "heartbeat_service_port", |
| -1) == heartbeat_service_port: |
| id = net_info.id |
| break |
| if not id: |
| id = CLUSTER.Node.get_id_from_ip(ip) |
| last_heartbeat = utils.escape_null(record["LastHeartbeat"]) |
| err_msg = record["ErrMsg"] |
| be = BEState(id, ip, backend_id, decommissioned, alive, tablet_num, |
| last_heartbeat, err_msg, http_port, |
| heartbeat_service_port) |
| be_states[id] = be |
| self.be_states = be_states |
| |
| # return rows, and each row is a record map |
| def _exec_query(self, sql, retries=3): |
| self._prepare_conn() |
| for attempt in range(retries): |
| try: |
| with self.conn.cursor() as cursor: |
| cursor.execute(sql) |
| fields = [field_md[0] for field_md in cursor.description |
| ] if cursor.description else [] |
| return [ |
| dict(zip(fields, row)) for row in cursor.fetchall() |
| ] |
| except Exception as e: |
| LOG.warning( |
| f"Error occurred: fe {self.fe_ip}:{self.fe_port}, sql `{sql}`, err {e}" |
| ) |
| if "timed out" in str(e).lower() and attempt < retries - 1: |
| LOG.warning( |
| f"Query timed out, fe {self.fe_ip}:{self.fe_port}. Retrying {attempt + 1}/{retries}..." |
| ) |
| self._reset_conn() |
| else: |
| raise e |
| raise Exception("Max retries exceeded") |
| |
| def _prepare_conn(self): |
| if self.conn: |
| return |
| self._reset_conn() |
| |
| def _reset_conn(self, read_timeout=10, connect_timeout=3): |
| self.conn = pymysql.connect(user="root", |
| host=self.fe_ip, |
| read_timeout=read_timeout, |
| connect_timeout=connect_timeout, |
| port=self.fe_port) |
| |
| |
| def get_db_mgr(cluster_name, all_node_net_infos, required_load_succ=True): |
| assert cluster_name |
| db_mgr = DBManager(all_node_net_infos) |
| master_fe_query_addr_file = CLUSTER.get_master_fe_addr_path(cluster_name) |
| master_fe_query_addr = None |
| if os.path.exists(master_fe_query_addr_file): |
| with open(master_fe_query_addr_file, "r") as f: |
| master_fe_query_addr = f.read().strip() |
| |
| if not master_fe_query_addr: |
| return db_mgr |
| |
| pos = master_fe_query_addr.find(':') |
| master_fe_ip = master_fe_query_addr[:pos] |
| master_fe_port = int(master_fe_query_addr[pos + 1:]) |
| |
| chose_fe_ip = "" |
| chose_fe_port = -1 |
| alive_fe = None |
| cluster = CLUSTER.Cluster.load(cluster_name) |
| containers = utils.get_doris_containers(cluster_name).get(cluster_name, []) |
| for container in containers: |
| if utils.is_container_running(container): |
| _, node_type, id = utils.parse_service_name(container.name) |
| if node_type == CLUSTER.Node.TYPE_FE: |
| node = cluster.get_node(node_type, id) |
| if not alive_fe: |
| alive_fe = node |
| if node.get_ip() == master_fe_ip and node.meta["ports"][ |
| "query_port"] == master_fe_port: |
| alive_fe = node |
| break |
| if alive_fe: |
| chose_fe_ip = alive_fe.get_ip() |
| chose_fe_port = alive_fe.meta["ports"]["query_port"] |
| elif utils.is_socket_avail(master_fe_ip, master_fe_port): |
| chose_fe_ip = master_fe_ip |
| chose_fe_port = master_fe_port |
| else: |
| LOG.debug("no available alive fe") |
| return db_mgr |
| |
| db_mgr.fe_ip = chose_fe_ip |
| db_mgr.fe_port = chose_fe_port |
| try: |
| db_mgr.load_states() |
| except Exception as e: |
| if required_load_succ: |
| raise e |
| #LOG.exception(e) |
| |
| return db_mgr |
