#!/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 keyring_file_path = None 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 idx == nodeidx: local_ip = val['ip'] server_port = val['port'] 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'] if 'keyring_file_path' in val: keyring_file_path = val['keyring_file_path'] idx = idx + 1 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 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) if keyring_file_path is not None: replace_items["place_holder_keyring_file_path"] = keyring_file_path 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 doinit = args.initialize 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() 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 \""]) if doinit == "yes": 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])) os.system("rm -f %s" % cnf_file_path) if os.path.exists('ca.pem'): os.system("cp -f ca.pem %s" % dirs[1]) if os.path.exists('server-cert.pem'): os.system("cp -f server-cert.pem %s" % dirs[1]) if os.path.exists('server-key.pem'): os.system("cp -f server-key.pem %s" % 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 ssl is set, turn on it. os.system("sed -e 's/^#ssl/ssl/' -i " + cnf_file_path) if 'place_holder_keyring_file_path' in replace_items: os.system("sed -e '/keyring_file/s/^#//' -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) if doinit == "yes": 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, CONNECTION_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;" 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) else: sys_cmd = " ".join([cmd0, install_path + '/bin/mysql', '--connect-expired-password', '-S' + prod_dir + '/mysql.sock', '-uroot', '-proot', '-e', '"select version();"']) for idx in range(100): result = subprocess.check_output(sys_cmd + '; exit 0', shell = True, stderr=subprocess.STDOUT).decode() if result.find('version') >= 0: break os.system('sleep 10\n') ret = os.system(sys_cmd) if ret != 0: raise Exception("Fail to execute command:" + sys_cmd) 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('--initialize', type=str, default='yes', choices=['yes','no']) 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)