yan
/
mdserver-web
mirror of https://github.com/midoks/mdserver-web.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #765 from midoks/dev 

mysql 9.0 兼容优化
This commit is contained in:
Mr Chen 2025-06-25 18:55:27 +08:00 committed by GitHub
parent b8f4ed2767 351d4a0ad6
commit 3595ae6066
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 87 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
plugins/mysql-community/index.py
View File

@ -406,6 +406,12 @@ def getShowLogFile():
def getMdb8Ver():
return ['8.0','8.1','8.2','8.3','8.4','9.0','9.1',"9.2"]
def getSlaveName():
mdb8 = getMdb8Ver()
if mw.inArray(mdb8, version):
return 'replica'
return 'slave'
def pGetDbUser():
if mw.isAppleSystem():
user = mw.execShell("who | sed -n '2, 1p' |awk '{print $1}'")[0].strip()
@ -2621,10 +2627,11 @@ def initSlaveStatusSyncUser(version=''):
if len(slave_data) < 1:
return mw.returnJson(False, '需要先添加同步用户配置!')
slave_name = getSlaveName()
# print(data)
pdb = pMysqlDb()
if len(slave_data) == 1:
dlist = pdb.query('show slave status')
dlist = pdb.query('show '+slave_name+' status')
if len(dlist) > 0:
return mw.returnJson(False, '已经初始化好了zz...')
@ -2660,8 +2667,8 @@ def initSlaveStatusSyncUser(version=''):
# pdb.query("start slave user='{}' password='{}';".format(
# u['user'], u['pass']))
pdb.query("start slave")
pdb.query("start all slaves")
pdb.query("start "+slave_name)
pdb.query("start all "+slave_name)
if msg == '':
msg = '初始化成功!'
@ -2669,8 +2676,9 @@ def initSlaveStatusSyncUser(version=''):
def initSlaveStatusSSH(version=''):
slave_name = getSlaveName()
db = pMysqlDb()
dlist = db.query('show slave status')
dlist = db.query('show '+slave_name+' status')
conn = pSqliteDb('slave_id_rsa')
ssh_list = conn.field('ip,port,id_rsa,db_user').select()
@ -2684,8 +2692,8 @@ def initSlaveStatusSSH(version=''):
paramiko.util.log_to_file('paramiko.log')
ssh = paramiko.SSHClient()
db.query('stop slave')
db.query('reset slave all')
db.query('stop '+slave_name)
db.query('reset '+slave_name+' all')
for data in ssh_list:
ip = data['ip']
SSH_PRIVATE_KEY = "/tmp/t_ssh_" + ip + ".txt"
@ -2738,7 +2746,7 @@ def initSlaveStatusSSH(version=''):
os.system("rm -rf " + SSH_PRIVATE_KEY)
except Exception as e:
return mw.returnJson(False, '[主][' + ip + ']:SSH认证配置连接失败!' + str(e))
db.query('start slave')
db.query('start '+slave_name)
return mw.returnJson(True, '初始化成功!')
@ -2749,7 +2757,10 @@ def setSlaveStatus(version=''):
mode = mw.readFile(mode_file)
pdb = pMysqlDb()
dlist = pdb.query('show slave status')
slave_name = getSlaveName()
cmd = 'show '+slave_name+' status'
dlist = pdb.query(cmd)
if len(dlist) == 0:
return mw.returnJson(False, '需要手动添加同步账户或者执行初始化!')
@ -2758,9 +2769,11 @@ def setSlaveStatus(version=''):
cmd = "slave"
if 'Channel_Name' in v:
ch_name = v['Channel_Name']
cmd = "slave for channel '{}'".format(ch_name)
cmd = slave_name + " for channel '{}'".format(ch_name)
if (v["Slave_IO_Running"] == 'Yes' or v["Slave_SQL_Running"] == 'Yes'):
if (( 'Slave_IO_Running' in v and v["Slave_IO_Running"] == 'Yes') or ('Slave_SQL_Running' in v and v["Slave_SQL_Running"] == 'Yes')):
pdb.query("stop {}".format(cmd))
elif (( 'Replica_IO_Running' in v and v["Replica_IO_Running"] == 'Yes') or ( 'Replica_SQL_Running' in v and v["Replica_SQL_Running"] == 'Yes') ):
pdb.query("stop {}".format(cmd))
else:
pdb.query("start {}".format(cmd))
@ -2771,13 +2784,17 @@ def setSlaveStatus(version=''):
def deleteSlave(version=''):
args = getArgs()
db = pMysqlDb()
slave_name = 'slave'
mdb8 = getMdb8Ver()
if mw.inArray(mdb8, version):
slave_name = 'replica'
if 'sign' in args:
sign = args['sign']
db.query("stop slave for channel '{}'".format(sign))
db.query("reset slave all for channel '{}'".format(sign))
db.query("stop {} for channel '{}'".format(slave_name,sign))
db.query("reset {} all for channel '{}'".format(slave_name, sign))
else:
db.query('stop slave')
db.query('reset slave all')
db.query('stop '+slave_name)
db.query('reset '+slave_name+' all')
return mw.returnJson(True, '删除成功!')
@ -3152,8 +3169,11 @@ def doFullSyncUserImportContentForChannel(file, channel_name):
# print(file, channel_name)
content = mw.readFile(file)
content = content.replace('STOP SLAVE;', "STOP SLAVE for channel '{}';".format(channel_name))
content = content.replace('START SLAVE;', "START SLAVE for channel '{}';".format(channel_name))
slave_name = getSlaveName()
slave_name = slave_name.upper()
content = content.replace('STOP '+slave_name+';', "STOP {} for channel '{}';".format(slave_name,channel_name))
content = content.replace('START '+slave_name+';', "START {} for channel '{}';".format(slave_name,channel_name))
find_head = "CHANGE MASTER TO "
find_re = find_head+"(.*?);"

79
plugins/mysql/index.py
View File

@ -437,6 +437,13 @@ def getShowLogFile():
def getMdb8Ver():
return ['8.0','8.1','8.2','8.3','8.4','9.0',"9.1","9.2"]
def getSlaveName():
mdb8 = getMdb8Ver()
if mw.inArray(mdb8, version):
return 'replica'
return 'slave'
def pGetDbUser():
if mw.isAppleSystem():
user = mw.execShell(
@ -2855,12 +2862,11 @@ def initSlaveStatusSyncUser(version=''):
if len(slave_data) < 1:
return mw.returnJson(False, '需要先添加同步用户配置!')
slave_name = getSlaveName()
# print(data)
pdb = pMysqlDb()
if len(slave_data) == 1:
cmd_slave = 'show slave status'
if pk_version.parse(version) < pk_version.parse("8.0"):
cmd_slave = 'SHOW REPLICA STATUS'
cmd_slave = 'show '+slave_name+' status'
dlist = pdb.query(cmd_slave)
if dlist and len(dlist) > 0:
return mw.returnJson(False, '已经初始化好了zz...')
@ -2899,8 +2905,8 @@ def initSlaveStatusSyncUser(version=''):
# pdb.query("start slave user='{}' password='{}';".format(
# u['user'], u['pass']))
pdb.query("start slave")
pdb.query("start all slaves")
pdb.query("start "+slave_name)
pdb.query("start all "+slave_name)
if msg == '':
msg = '初始化成功!'
@ -2908,8 +2914,9 @@ def initSlaveStatusSyncUser(version=''):
def initSlaveStatusSSH(version=''):
slave_name = getSlaveName()
db = pMysqlDb()
dlist = db.query('show slave status')
dlist = db.query('show '+slave_name+' status')
conn = pSqliteDb('slave_id_rsa')
ssh_list = conn.field('ip,port,id_rsa,db_user').select()
@ -2923,8 +2930,8 @@ def initSlaveStatusSSH(version=''):
paramiko.util.log_to_file('paramiko.log')
ssh = paramiko.SSHClient()
db.query('stop slave')
db.query('reset slave all')
db.query('stop '+slave_name)
db.query('reset '+slave_name+' all')
for data in ssh_list:
ip = data['ip']
SSH_PRIVATE_KEY = "/tmp/t_ssh_" + ip + ".txt"
@ -2977,7 +2984,7 @@ def initSlaveStatusSSH(version=''):
os.system("rm -rf " + SSH_PRIVATE_KEY)
except Exception as e:
return mw.returnJson(False, '[主][' + ip + ']:SSH认证配置连接失败!' + str(e))
db.query('start slave')
db.query('start '+slave_name)
return mw.returnJson(True, '初始化成功!')
@ -2988,18 +2995,21 @@ def setSlaveStatus(version=''):
mode = mw.readFile(mode_file)
pdb = pMysqlDb()
dlist = pdb.query('show slave status')
slave_name = getSlaveName()
dlist = pdb.query(cmd)
if len(dlist) == 0:
return mw.returnJson(False, '需要手动添加同步账户或者执行初始化!')
for v in dlist:
connection_name = ''
cmd = "slave"
if 'Channel_Name' in v:
ch_name = v['Channel_Name']
cmd = "slave for channel '{}'".format(ch_name)
cmd = slave_name + " for channel '{}'".format(ch_name)
if (v["Slave_IO_Running"] == 'Yes' or v["Slave_SQL_Running"] == 'Yes'):
if (( 'Slave_IO_Running' in v and v["Slave_IO_Running"] == 'Yes') or ('Slave_SQL_Running' in v and v["Slave_SQL_Running"] == 'Yes')):
pdb.query("stop {}".format(cmd))
elif (( 'Replica_IO_Running' in v and v["Replica_IO_Running"] == 'Yes') or ( 'Replica_SQL_Running' in v and v["Replica_SQL_Running"] == 'Yes') ):
pdb.query("stop {}".format(cmd))
else:
pdb.query("start {}".format(cmd))
@ -3008,12 +3018,15 @@ def setSlaveStatus(version=''):
def deleteSlaveFunc(sign = ''):
db = pMysqlDb()
slave_name = getSlaveName()
if sign != '':
db.query("stop slave for channel '{}'".format(sign))
db.query("reset slave all for channel '{}'".format(sign))
sign = args['sign']
db.query("stop {} for channel '{}'".format(slave_name,sign))
db.query("reset {} all for channel '{}'".format(slave_name, sign))
else:
db.query('stop slave')
db.query('reset slave all')
db.query('stop '+slave_name)
db.query('reset '+slave_name+' all')
def deleteSlave(version=''):
args = getArgs()
@ -3180,11 +3193,15 @@ def syncDatabaseRepair(version=''):
sync_count_data = sync_db.query(cmd_count_sql)
if local_count_data != sync_count_data:
print("all data compare: ",local_count_data, sync_count_data)
inconsistent_table.append(table_name)
diff = sync_count_data[0]['num'] - local_count_data[0]['num']
print(table_name+', need sync. diff,'+str(diff))
mw.writeFile(tmp_log, table_name+', need sync. diff,'+str(diff)+'\n','a+')
if sync_count_data == None:
print("sync:"+table_name+" is not exists!!!")
mw.writeFile(tmp_log, "sync:"+table_name+" is not exists!!!"+'\n','a+')
else:
print("all data compare: ",local_count_data, sync_count_data)
inconsistent_table.append(table_name)
diff = sync_count_data[0]['num'] - local_count_data[0]['num']
print(table_name+', need sync. diff,'+str(diff))
mw.writeFile(tmp_log, table_name+', need sync. diff,'+str(diff)+'\n','a+')
else:
print(table_name+' check ok.')
mw.writeFile(tmp_log, table_name+' check ok.\n','a+')
@ -3400,10 +3417,13 @@ def doFullSyncUserImportContentForChannel(file, channel_name):
# print(file, channel_name)
content = mw.readFile(file)
content = content.replace('STOP SLAVE;', "STOP SLAVE for channel '{}';".format(channel_name))
content = content.replace('START SLAVE;', "START SLAVE for channel '{}';".format(channel_name))
slave_name = getSlaveName()
slave_name = slave_name.upper()
find_head = r"CHANGE MASTER TO "
content = content.replace('STOP '+slave_name+';', "STOP {} for channel '{}';".format(slave_name,channel_name))
content = content.replace('START '+slave_name+';', "START {} for channel '{}';".format(slave_name,channel_name))
find_head = "CHANGE MASTER TO "
find_re = find_head+"(.*?);"
find_r = re.search(find_re, content, re.I|re.M)
if find_r:
@ -3418,6 +3438,7 @@ def doFullSyncUserImportContentForChannel(file, channel_name):
def doFullSyncUser(version=''):
slave_name = getSlaveName()
which_pv = mw.execShell('which pv')
is_exist_pv = False
if os.path.exists(which_pv[0]):
@ -3479,9 +3500,9 @@ def doFullSyncUser(version=''):
mdb8 = getMdb8Ver()
if mw.inArray(mdb8,version):
db.query("stop slave user='{}' password='{}';".format(user, apass))
db.query("stop {} user='{}' password='{}';".format(slave_name,user, apass))
else:
db.query("stop slave")
db.query("stop "+slave_name)
time.sleep(1)
@ -3577,9 +3598,9 @@ def doFullSyncUser(version=''):
if mw.inArray(mdb8,version):
db.query("start replica user='{}' password='{}';".format(user, apass))
else:
db.query("start slave")
db.query("start "+slave_name)
db.query("start all slaves")
db.query("start all "+slave_name)
time_all_e = time.time()
cos = time_all_e - time_all_s
writeDbSyncStatus({'code': 6, 'msg': '总耗时:'+str(int(cos))+'秒,从库重启完成...', 'progress': 100})

1
plugins/mysql/install.sh
View File

@ -11,6 +11,7 @@ export PATH
# cd /www/server/mdserver-web && source bin/activate && python3 plugins/mysql/index.py sync_database_repair {"db":"xxx","sign":""}
# cd /www/server/mdserver-web && source bin/activate && python3 plugins/mysql/index.py init_slave_status
# cd /www/server/mdserver-web && source bin/activate && python3 plugins/mysql/index.py install_pre_inspection
# cd /www/server/mdserver-web && source bin/activate && python3 plugins/mysql/index.py set_slave_status {"close":"change"}
curPath=`pwd`
rootPath=$(dirname "$curPath")
rootPath=$(dirname "$rootPath")