#!/bin/python import os import os.path import sys import re import time import random import fcntl import struct import socket import subprocess import json import shlex import pwd import grp from distutils.util import strtobool import argparse def my_print(toprint): if sys.version_info.major == 2: sys.stdout.write(toprint + "\n") else: print(toprint) sys.stdout.flush() def param_replace(string, rep_dict): pattern = re.compile("|".join([re.escape(k) for k in rep_dict.keys()]), re.M) return pattern.sub(lambda x: rep_dict[x.group(0)], string) def get_root_init_pwd(logdir): install_inf = open(logdir + '/mysqld.err', 'r') lines = install_inf.readlines() for line in lines: if 'A temporary password is generated for root@localhost' in line: ret = line.split('root@localhost: ')[1][:-1] return ret def make_mgr_args(mgr_config_path, replace_items, target_node_index, ha_mode, dbuser): jsconf = open(mgr_config_path) jstr = jsconf.read() jsconf.close() jscfg = json.loads(jstr) nodeidx = target_node_index idx = 0 white_list = '' local_addr = '' local_ip = '' seeds = '' is_master_node = False weight = 50 mgr_port = 0 server_xport = 0 server_port = 0 innodb_buffer_pool_size = 0 rocksdb_block_cache_size = '16G' server_data_prefix = '' server_log_prefix = '' server_innolog_prefix = '' db_inst_user = '' log_arch = '' master_ip = '' master_port = 0 cpu_cores = 8 for val in jscfg['nodes']: if val['is_primary'] == True : if idx == nodeidx: is_master_node = True master_ip = val['ip'] master_port = val['port'] if ha_mode == 'mgr' : if white_list != '': white_list = white_list + ',' seeds = seeds + ',' white_list = white_list + val['ip'] addr=val['ip'] if addr.find(":") >= 0: addr='['+addr+']' seeds = seeds + addr + ':' + str(val['mgr_port']) if idx == nodeidx: local_ip = val['ip'] server_port = val['port'] if ha_mode == 'mgr': local_addr = addr + ':' + str(val['mgr_port']) mgr_port = val['mgr_port'] server_xport = val['xport'] weight = val['election_weight'] innodb_buffer_pool_size = val['innodb_buffer_pool_size'] server_data_prefix = val['data_dir_path'] server_log_prefix = val['log_dir_path'] db_inst_user = val.get('user', dbuser) if 'innodb_log_dir_path' in val: server_innolog_prefix = val['innodb_log_dir_path'] if 'cpu_cores' in val: cpu_cores = val['cpu_cores'] idx = idx + 1 if ha_mode == 'mgr': mgr_num_nodes = idx + 1 #if mgr_num_nodes < 2: # raise ValueError("Config error, need at least two nodes in the shard.") if local_addr == '' or nodeidx < 0 or nodeidx >= mgr_num_nodes: raise ValueError("Config error, target_node_index must be in [0, {}).".format(mgr_num_nodes)) if seeds == '': raise RuntimeError("Config error, no primary node specified.") if server_xport == mgr_port or server_port == mgr_port or server_port == server_xport : raise ValueError("Config error, MGR port(" + str(mgr_port) + "), client regular port(" + str(server_port) + ") and X protocol port(" + str(server_xport) + ") must be different.") data_path = server_data_prefix data_dir = data_path + "/data" innodb_dir = data_dir tmp_dir = data_path + "/temp" log_path = server_log_prefix log_dir = log_path prod_dir = log_path log_relay = log_path + "/relay" log_bin_arg = log_path + "/binlog" # seperate binlog/relaylogs from innodb redo logs, store them in two dirs so possibly two disks to do parallel IO. if server_innolog_prefix == '': log_arch = data_path + "/redolog" else: log_arch = server_innolog_prefix + "/redolog" replace_items["place_holder_ip"] = local_ip if ha_mode == 'mgr': replace_items["place_holder_mgr_recovery_retry_count"] = str(mgr_num_nodes*100) replace_items["place_holder_mgr_local_address"] = local_addr replace_items["place_holder_mgr_seeds"] = seeds replace_items["place_holder_mgr_whitelist"] = white_list replace_items["place_holder_mgr_member_weight"] = str(weight) replace_items["place_holder_mgr_group_name"] = jscfg['group_uuid'] replace_items["prod_dir"] = prod_dir replace_items["data_dir"] = data_dir replace_items["innodb_dir"] = innodb_dir replace_items["log_dir"] = log_dir replace_items["tmp_dir"] = tmp_dir replace_items["log_relay"] = log_relay replace_items["log_bin_arg"] = log_bin_arg replace_items["log_arch"] = log_arch replace_items["place_holder_innodb_buffer_pool_size"] = innodb_buffer_pool_size replace_items["place_holder_rocksdb_block_cache_size"] = rocksdb_block_cache_size replace_items["place_holder_x_port"] =str(server_xport) replace_items["place_holder_port"] = str(server_port) replace_items["place_holder_user"] = db_inst_user replace_items["place_holder_thread_concurrency"] = str(cpu_cores * 4) dirs=[prod_dir, data_dir, innodb_dir, log_dir, tmp_dir, log_relay, log_bin_arg, log_arch] return is_master_node, local_ip, server_port, data_path, log_path, log_arch, log_dir, db_inst_user, dirs, master_ip, master_port class MysqlConfig: def __init__(self, install_path, args): config_template_file = args.dbcfg server_id = args.server_id cluster_id = args.cluster_id shard_id = args.shard_id mgr_config_path = args.config target_node_index = args.target_node_index ha_mode = args.ha_mode dbuser = args.user metaseeds = args.meta_addrs replace_items = { #"place_holder_extra_port": str(int(server_port)+10000), mysql8.0 doesn't have 'extra_port' var. "base_dir": install_path, "place_holder_server_id": str(server_id), "place_holder_shard_id": "0", "place_holder_cluster_id": "0", } is_master, server_ip, server_port, data_path, log_path, log_arch, log_dir, user, dirs, master_ip, master_port = make_mgr_args(mgr_config_path, replace_items, target_node_index, ha_mode, dbuser) prod_dir = dirs[0] config_template = open(config_template_file, 'r').read() conf = param_replace(config_template, replace_items) group = grp.getgrgid(pwd.getpwnam(user).pw_gid).gr_name etc_path = install_path + "/etc" if not os.path.exists(etc_path): os.mkdir(etc_path) if euser == 'root': subprocess.call(["chown", "-R", user+":"+group, etc_path]) conf_list_file = etc_path+"/instances_list.txt" portstr = str(server_port) ret = os.system("grep '^" + portstr + "==>' " + conf_list_file + " >/dev/null 2>/dev/null") if ret == 0: raise Exception("Invalid port:" + portstr + ", The port is in use!") for d in dirs: if not os.path.exists(d): os.makedirs(d) euser = pwd.getpwuid(os.geteuid()).pw_name if euser == 'root': subprocess.call(["chown", "-R", user+":"+group, log_path]) subprocess.call(["chown", "-R", user+":"+group, log_arch]) subprocess.call(["chown", "-R", user+":"+group, data_path]) cnf_file_path = etc_path + "/" + portstr +".cnf" cnf_file = open(cnf_file_path, 'w') cnf_file.write(conf) cnf_file.close() if ha_mode != 'rbr' or args.fullsync == 0: os.system("sed -e 's/enable_fullsync.*=.*true/enable_fullsync = false/' -i " + cnf_file_path) start_mgr_sql = '' if ha_mode == 'mgr': if is_master: start_mgr_sql = 'SET GLOBAL group_replication_bootstrap_group=ON; START GROUP_REPLICATION; SET GLOBAL group_replication_bootstrap_group=OFF; ' #select group_replication_set_as_primary(@@server_uuid); this can't be done now, it requires a quorum. else: start_mgr_sql = 'START GROUP_REPLICATION;' elif ha_mode == 'rbr': if is_master: start_mgr_sql = '' else: start_mgr_sql = 'change master to MASTER_AUTO_POSITION = 1, MASTER_HOST=\'' + master_ip + '\' , MASTER_PORT=' + str(master_port) + ', MASTER_USER=\'repl\' , MASTER_PASSWORD=\'repl_pwd\' for channel \'kunlun_repl\'; start slave' cmd0 = "export LD_LIBRARY_PATH=" + install_path + "/lib:$LD_LIBRARY_PATH;" if euser == user: cmdstr = " ".join([cmd0, install_path+"/bin/mysqld", "--defaults-file="+cnf_file_path, "--user="+user, "--initialize"]) else: cmdstr = " ".join(["su", user, "-c", "\"", cmd0, install_path+"/bin/mysqld", "--defaults-file="+cnf_file_path, "--user="+user, "--initialize \""]) os.system(cmdstr) root_init_password = get_root_init_pwd(log_dir) assert(root_init_password != None) # dirs[1] is data_dir os.system("cp -f %s %s" % (cnf_file_path, dirs[1])) cnf_file_path = dirs[1] + "/" + portstr +".cnf" # Enable the mgr options, which have to be commented at initialization because plugins are not loaded when mysqld is started with --initialize. if ha_mode == 'mgr': os.system("sed -e 's/^#group_replication_/group_replication_/' -i " + cnf_file_path) os.system("sed -e 's/^#clone_/clone_/' -i " + cnf_file_path) os.system("sed -e '/group_replication.so/s/^#plugin_load_add/plugin_load_add/' -i " + cnf_file_path) os.system("sed -e '/group_replication.so/!s/^#plugin_load/plugin_load/' -i " + cnf_file_path) os.system("sed -e 's/^#rocksdb_/rocksdb_/' -i " + cnf_file_path) if euser == user: os.system(" ".join([cmd0, "./bootmysql.sh", install_path, cnf_file_path, user])) else: os.system(" ".join(["su", user, "-c", "\"", cmd0, "./bootmysql.sh", install_path, cnf_file_path, user+"\""])) os.system("echo \"" + str(server_port) + "==>" + cnf_file_path + "\" >> " + conf_list_file) os.system("sed -e 's/#skip_name_resolve=on/skip_name_resolve=on/' -i " + cnf_file_path) change_pwd_sql = "set sql_log_bin=0;ALTER USER 'root'@'localhost' IDENTIFIED BY 'root';" #TODO: use md5 signature for each below password, don't use plain text password; AND delete 'root' user. # the python mysql client lib cuts long sql text to multiple sections, so do 'set sql_log_bin=0; ' at beginning of each query. init_sql = "set sql_log_bin=0; create user clustmgr identified by 'clustmgr_pwd'; GRANT ROLE_ADMIN, SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, RELOAD, PROCESS, REFERENCES, INDEX, ALTER, SHOW DATABASES, CREATE TEMPORARY TABLES, LOCK TABLES, EXECUTE, REPLICATION SLAVE, REPLICATION CLIENT, CREATE VIEW, SHOW VIEW, CREATE ROUTINE, ALTER ROUTINE, EVENT, TRIGGER, BINLOG_ADMIN, GROUP_REPLICATION_ADMIN, REPLICATION_SLAVE_ADMIN, PERSIST_RO_VARIABLES_ADMIN, SYSTEM_VARIABLES_ADMIN, XA_RECOVER_ADMIN on *.* to clustmgr@'%'; flush privileges;" \ + "set sql_log_bin=0; create user repl identified by 'repl_pwd'; grant replication slave,replication client, BACKUP_ADMIN, CLONE_ADMIN on *.* to 'repl'@'%' ; flush privileges;" \ + "set sql_log_bin=0; create user agent@localhost identified by 'agent_pwd'; grant all on *.* to 'agent'@'localhost' with grant option;flush privileges;select version();" init_sql2 = "set sql_log_bin=0; create user pgx identified by 'pgx_pwd' ; grant CREATE USER,ROLE_ADMIN,Select,Insert,Update,Delete,Create,Drop,Process,References,Index,Alter,SHOW DATABASES,CREATE TEMPORARY TABLES,LOCK TABLES,Execute,CREATE VIEW,SHOW VIEW,CREATE ROUTINE,ALTER ROUTINE,Event,Trigger, reload, REPLICATION SLAVE, SYSTEM_VARIABLES_ADMIN, PERSIST_RO_VARIABLES_ADMIN, BINLOG_ADMIN, GROUP_REPLICATION_ADMIN, REPLICATION_SLAVE_ADMIN, XA_RECOVER_ADMIN on *.* to 'pgx'@'%'; flush privileges;" \ + "set sql_log_bin=0;delete from mysql.db where Db='test\_%' and Host='%' ;delete from mysql.db where Db='test' and Host='%';flush privileges;" if ha_mode == 'mgr': init_sql2 = init_sql2 + '''CHANGE MASTER TO MASTER_USER='repl', MASTER_PASSWORD='repl_pwd' FOR CHANNEL 'group_replication_recovery';''' if ha_mode != 'no_rep': init_sql2 = init_sql2 + start_mgr_sql init_sql3 = "".join(["set sql_log_bin=0; insert into kunlun_sysdb.cluster_info (cluster_name,shard_name) values ('",cluster_id,"','",shard_id,"')"]) sys_cmd = " ".join([cmd0, install_path + '/bin/mysql', '--connect-expired-password', '-S' + prod_dir + '/mysql.sock', '-uroot', '-p'+"'"+root_init_password+"'", '-e', '"' + change_pwd_sql + init_sql + '"', '; exit 0']) add_proc_cmd = " ".join([cmd0, install_path + '/bin/mysql', '--connect-expired-password', '-S' + prod_dir + '/mysql.sock', '-uroot', '-proot <' , install_path+'/dba_tools/seq_reserve_vals.sql' ]) add_proc_sysdb = " ".join([cmd0, install_path + '/bin/mysql', '--connect-expired-password', '-S' + prod_dir + '/mysql.sock', '-uroot', '-proot <' , install_path+'/dba_tools/sys_db_table.sql' ]) initcmd2 = " ".join([cmd0, install_path + "/bin/mysql", "--connect-expired-password", '-S' + prod_dir + '/mysql.sock', '-uroot -proot', '-e', '"' + init_sql2 + '"\n']) initcmd3 = " ".join([cmd0, install_path + "/bin/mysql", "--connect-expired-password", '-S' + prod_dir + '/mysql.sock', '-uroot -proot', '-e', '"' + init_sql3 + '"\n']) for idx in range(100): result = subprocess.check_output(sys_cmd, shell = True, stderr=subprocess.STDOUT).decode() if result.find('version') >= 0: break os.system('sleep 10\n') # we have disabled sql_log_bin in the script so that replicas can execute it too. ret = os.system(add_proc_sysdb) if ret != 0: raise Exception("Fail to execute command:" + add_proc_sysdb) ret = os.system(initcmd3) if ret != 0: raise Exception("Fail to execute command:" + initcmd3) ret = os.system(add_proc_cmd) if ret != 0: raise Exception("Fail to execute command:" + add_proc_cmd) if metaseeds is not None: init_sql4 = "set sql_log_bin=0; insert into kunlun_sysdb.metadb_info (conn_addr) values ('%s')" % metaseeds initcmd4 = " ".join([cmd0, install_path + "/bin/mysql", "--connect-expired-password", '-S' + prod_dir + '/mysql.sock', '-uroot -proot', '-e', '"' + init_sql4 + '"\n']) ret = os.system(initcmd4) if ret != 0: raise Exception("Fail to execute command:" + initcmd4) ret = os.system(initcmd2) if ret != 0: raise Exception("Fail to execute command:" + initcmd2) readonlysql = "set global super_read_only = ON;" readonlycmd = " ".join([cmd0, install_path + "/bin/mysql", "--connect-expired-password", '-S' + prod_dir + '/mysql.sock', '-uroot -proot', '-e', '"' + readonlysql + '"\n']) if ha_mode == 'rbr' and (not is_master): ret = os.system(readonlycmd) if ret != 0: raise Exception("Fail to execute command:" + readonlycmd) if ha_mode == 'mgr' or ha_mode == 'rbr': os.system("sed -e 's/#super_read_only=OFF/super_read_only=ON/' -i " + cnf_file_path) # uuid_cmd_str = install_path + "/bin/mysql --silent --skip-column-names --connect-expired-password -S" + prod_dir + '/mysql.sock -uroot -proot -e "set @uuid_str=uuid(); set global group_replication_group_name=@uuid_str; ' + start_mgr_sql+ ' select @uuid_str;"\n' # uuid_cmd=shlex.split(uuid_cmd_str) # popen_ret = subprocess.Popen(uuid_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=install_path) # uuid_str, errmsg = popen_ret.communicate() # uuid_str = uuid_str[:-1] # chop off the trailing \n # subprocess.call(shlex.split("sed -e 's/place_holder_mgr_group_name/" + uuid_str + "/' -i " + cnf_file_path)) # append the new instance's port to datadir mapping into instance_list.txt def print_usage(): my_print('Usage: install-mysql.py --config /path/of/mgr/config/file --target_node_index idx --cluster_id ID --shard_id N [--dbcfg /db/config/template/path/template.cnf] [--user db_init_user] [--server_id N] [--ha_mode mgr|no_rep|rbr] [--meta_addrs]') if __name__ == "__main__": random.seed(str(sys.argv[1:])) parser = argparse.ArgumentParser(description='Install the storage node.') parser.add_argument('--config', type=str, help="The config path", required=True) parser.add_argument('--target_node_index', type=int, help = "target node", required=True) parser.add_argument('--cluster_id', type=str, help = "the id for the cluster", required=True) parser.add_argument('--shard_id', type=str, help = "the id for the shard",required=True) parser.add_argument('--dbcfg', type=str, help = "target node", default='./template.cnf') parser.add_argument('--user', type=str, help = "user_used_to_initialize", default=pwd.getpwuid(os.getuid()).pw_name) parser.add_argument('--server_id', type=int, help = "the id for the server", default=random.randint(1,65535)) parser.add_argument('--ha_mode', type=str, default='mgr', choices=['mgr','no_rep', 'rbr']) parser.add_argument('--meta_addrs', type=str, help="The metadata seeds", default=None) parser.add_argument('--fullsync', type=int, help="whether to enable fullsync for rbr mode", default=1) args = parser.parse_args() try: if not os.path.exists(args.dbcfg): raise ValueError("DB config template file {} doesn't exist!".format(args.dbcfg)) install_path = os.path.dirname(os.getcwd()) my_print("Installing mysql instance, please wait...") # undocument option, used for internal testing. MysqlConfig(install_path, args) except KeyError as e: print_usage() my_print(e)